From 2ac9a78d58bb1fbae41a75178e202eaa860ed021 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Thu, 30 Jun 2022 17:52:34 +0800 Subject: [PATCH] feat: support 'select *, expr from ...' syntax --- tests/test/c/tmqSim.c | 41 +++++++++++++++++++---------------------- 1 file changed, 19 insertions(+), 22 deletions(-) diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index 81fa72d15a..38ee543f10 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -124,15 +124,13 @@ char* getCurrentTimeString(char* timeString) { return timeString; } -static void tmqStop(int signum, void *info, void *ctx) { +static void tmqStop(int signum, void* info, void* ctx) { running = 0; char tmpString[128]; - taosFprintfFile(g_fp, "%s tmqStop() receive stop signal[%d]\n", getCurrentTimeString(tmpString), signum); + taosFprintfFile(g_fp, "%s tmqStop() receive stop signal[%d]\n", getCurrentTimeString(tmpString), signum); } -static void tmqSetSignalHandle() { - taosSetSignal(SIGINT, tmqStop); -} +static void tmqSetSignalHandle() { taosSetSignal(SIGINT, tmqStop); } void initLogFile() { char filename[256]; @@ -433,7 +431,7 @@ static void dumpToFileForCheck(TdFilePtr pFile, TAOS_ROW row, TAOS_FIELD* fields int32_t precision) { for (int32_t i = 0; i < num_fields; i++) { if (i > 0) { - taosFprintfFile(pFile, "\n"); + taosFprintfFile(pFile, ","); } shellDumpFieldToFile(pFile, (const char*)row[i], fields + i, length[i], precision); } @@ -463,16 +461,16 @@ static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex) int32_t precision = taos_result_precision(msg); const char* tbName = tmq_get_table_name(msg); - #if 0 +#if 0 // get schema //============================== stub =================================================// for (int32_t i = 0; i < numOfFields; i++) { taosFprintfFile(g_fp, "%02d: name: %s, type: %d, len: %d\n", i, fields[i].name, fields[i].type, fields[i].bytes); } //============================== stub =================================================// - #endif +#endif - dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision); + dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision); taos_print_row(buf, row, fields, numOfFields); @@ -523,15 +521,15 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) { static int32_t g_once_commit_flag = 0; static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { - pError("tmq_commit_cb_print() commit %d\n", code); + pError("tmq_commit_cb_print() commit %d\n", code); - if (0 == g_once_commit_flag) { - g_once_commit_flag = 1; - notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); + if (0 == g_once_commit_flag) { + g_once_commit_flag = 1; + notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); } - - char tmpString[128]; - taosFprintfFile(g_fp, "%s tmq_commit_cb_print() be called\n", getCurrentTimeString(tmpString)); + + char tmpString[128]; + taosFprintfFile(g_fp, "%s tmq_commit_cb_print() be called\n", getCurrentTimeString(tmpString)); } void build_consumer(SThreadInfo* pInfo) { @@ -683,13 +681,13 @@ void* consumeThreadFunc(void* param) { pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0); if (pInfo->taos == NULL) { taosFprintfFile(g_fp, "taos_connect() fail, can not notify and save consume result to main scripte\n"); - return NULL; + return NULL; } build_consumer(pInfo); build_topic_list(pInfo); if ((NULL == pInfo->tmq) || (NULL == pInfo->topicList)) { - taosFprintfFile(g_fp, "create consumer fail! tmq is null or topicList is null\n"); + taosFprintfFile(g_fp, "create consumer fail! tmq is null or topicList is null\n"); assert(0); return NULL; } @@ -697,7 +695,7 @@ void* consumeThreadFunc(void* param) { int32_t err = tmq_subscribe(pInfo->tmq, pInfo->topicList); if (err != 0) { pError("tmq_subscribe() fail, reason: %s\n", tmq_err2str(err)); - taosFprintfFile(g_fp, "tmq_subscribe() fail! reason: %s\n", tmq_err2str(err)); + taosFprintfFile(g_fp, "tmq_subscribe() fail! reason: %s\n", tmq_err2str(err)); assert(0); return NULL; } @@ -718,13 +716,13 @@ void* consumeThreadFunc(void* param) { err = tmq_unsubscribe(pInfo->tmq); if (err != 0) { pError("tmq_unsubscribe() fail, reason: %s\n", tmq_err2str(err)); - taosFprintfFile(g_fp, "tmq_unsubscribe()! reason: %s\n", tmq_err2str(err)); + taosFprintfFile(g_fp, "tmq_unsubscribe()! reason: %s\n", tmq_err2str(err)); } err = tmq_consumer_close(pInfo->tmq); if (err != 0) { pError("tmq_consumer_close() fail, reason: %s\n", tmq_err2str(err)); - taosFprintfFile(g_fp, "tmq_consumer_close()! reason: %s\n", tmq_err2str(err)); + taosFprintfFile(g_fp, "tmq_consumer_close()! reason: %s\n", tmq_err2str(err)); } pInfo->tmq = NULL; @@ -871,4 +869,3 @@ int main(int32_t argc, char* argv[]) { return 0; } -