refactor: format tmqSim.c

utils/test/c/tmqSim.c

@@ -14,13 +14,13 @@
  */
 
 #include <assert.h>
+#include <math.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <time.h>
-#include <math.h>
 
 #include "taos.h"
 #include "taosdef.h"
@@ -45,11 +45,7 @@ typedef enum {
   NOTIFY_CMD_ID_BUTT,
 } NOTIFY_CMD_ID;
 
-typedef enum enumQUERY_TYPE {
-    NO_INSERT_TYPE,
-    INSERT_TYPE,
-    QUERY_TYPE_BUT
-} QUERY_TYPE;
+typedef enum enumQUERY_TYPE { NO_INSERT_TYPE, INSERT_TYPE, QUERY_TYPE_BUT } QUERY_TYPE;
 
 typedef struct {
   TdThread thread;
@@ -153,7 +149,6 @@ static void printHelp() {
   printf("%s%s\n", indent, "-x");
   printf("%s%s%s\n", indent, indent, "consume thread number, default is 1");
 
-
   printf("%s%s\n", indent, "-l");
   printf("%s%s%s%d\n", indent, indent, "run duration unit is minutes, default is ", g_stConfInfo.runDurationMinutes);
   printf("%s%s\n", indent, "-p");
@@ -165,7 +160,6 @@ static void printHelp() {
   printf("%s%s\n", indent, "-n");
   printf("%s%s%s\n", indent, indent, "payload len unit is byte, default is 1000");
 
- 
   exit(EXIT_SUCCESS);
 }
 
@@ -294,7 +288,7 @@ void parseArgument(int32_t argc, char* argv[]) {
       g_stConfInfo.producerRate = atol(argv[++i]);
     } else if (strcmp(argv[i], "-n") == 0) {
       g_stConfInfo.payloadLen = atol(argv[++i]);
-      if(g_stConfInfo.payloadLen <= 0 || g_stConfInfo.payloadLen > 1024 * 1024 * 1024){
+      if (g_stConfInfo.payloadLen <= 0 || g_stConfInfo.payloadLen > 1024 * 1024 * 1024) {
         pError("%s calloc size is too large: %s %s", GREEN, argv[++i], NC);
         exit(-1);
       }
@@ -379,7 +373,6 @@ int queryDB(TAOS* taos, char* command) {
   return -1;
 }
 
-
 void addRowsToVgroupId(SThreadInfo* pInfo, int32_t vgroupId, int32_t rows) {
   int32_t i;
   for (i = 0; i < pInfo->numOfVgroups; i++) {
@@ -418,7 +411,6 @@ TAOS* createNewTaosConnect() {
   return NULL;
 }
 
-
 int32_t saveConsumeContentToTbl(SThreadInfo* pInfo, char* buf) {
   char sqlStr[1100] = {0};
 
@@ -548,8 +540,7 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char* val, TAOS_FIELD* f
       break;
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
-    case TSDB_DATA_TYPE_JSON:
-      {
+    case TSDB_DATA_TYPE_JSON: {
       int32_t bufIndex = 0;
       for (int32_t i = 0; i < length; i++) {
         buf[bufIndex] = val[i];
@@ -562,8 +553,7 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char* val, TAOS_FIELD* f
       buf[bufIndex] = 0;
 
       taosFprintfFile(pFile, "%s%s%s", quotationStr, buf, quotationStr);
-      }
-      break;
+    } break;
     case TSDB_DATA_TYPE_TIMESTAMP:
       shellFormatTimestamp(buf, *(int64_t*)val, precision);
       taosFprintfFile(pFile, "%s%s%s", quotationStr, buf, quotationStr);
@@ -635,7 +625,6 @@ static int32_t data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
   return totalRows;
 }
 
-
 static int32_t meta_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex) {
   char    buf[1024];
   int32_t totalRows = 0;
@@ -652,21 +641,21 @@ static int32_t meta_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
     tmq_raw_data raw = {0};
     int32_t      code = tmq_get_raw(msg, &raw);
 
-    if(code == TSDB_CODE_SUCCESS){
-//	  int retCode = queryDB(pInfo->taos, "use metadb");
-//	  if (retCode != 0) {
-//		taosFprintfFile(g_fp, "error when use metadb\n");
-//		taosCloseFile(&g_fp);
-//		exit(-1);
-//	  }
-//	  taosFprintfFile(g_fp, "raw:%p\n", &raw);
-//
-//      tmq_write_raw(pInfo->taos, raw);
+    if (code == TSDB_CODE_SUCCESS) {
+      //	  int retCode = queryDB(pInfo->taos, "use metadb");
+      //	  if (retCode != 0) {
+      //		taosFprintfFile(g_fp, "error when use metadb\n");
+      //		taosCloseFile(&g_fp);
+      //		exit(-1);
+      //	  }
+      //	  taosFprintfFile(g_fp, "raw:%p\n", &raw);
+      //
+      //      tmq_write_raw(pInfo->taos, raw);
     }
 
     char* result = tmq_get_json_meta(msg);
-    if(result && strcmp(result, "") != 0){
-  	  //printf("meta result: %s\n", result);
+    if (result && strcmp(result, "") != 0) {
+      // printf("meta result: %s\n", result);
       taosFprintfFile(pInfo->pConsumeMetaFile, "%s\n", result);
     }
     tmq_free_json_meta(result);
@@ -683,8 +672,8 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
   char sqlStr[1024] = {0};
 
   // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
-  sprintf(sqlStr, "insert into %s.notifyinfo values (%" PRId64 ", %d, %d)", g_stConfInfo.cdbName, atomic_fetch_add_64(&g_stConfInfo.nowTime, 1), cmdId,
-          pInfo->consumerId);
+  sprintf(sqlStr, "insert into %s.notifyinfo values (%" PRId64 ", %d, %d)", g_stConfInfo.cdbName,
+          atomic_fetch_add_64(&g_stConfInfo.nowTime, 1), cmdId, pInfo->consumerId);
 
   taos_query_a(pInfo->taos, sqlStr, appNothing, NULL);
 
@@ -818,9 +807,9 @@ void loop_consume(SThreadInfo* pInfo) {
         tmq_res_t msgType = tmq_get_res_type(tmqMsg);
         if (msgType == TMQ_RES_TABLE_META) {
           totalRows += meta_msg_process(tmqMsg, pInfo, totalMsgs);
-		} else if (msgType == TMQ_RES_DATA){
+        } else if (msgType == TMQ_RES_DATA) {
           totalRows += data_msg_process(tmqMsg, pInfo, totalMsgs);
-                } else if (msgType == TMQ_RES_METADATA){
+        } else if (msgType == TMQ_RES_METADATA) {
           meta_msg_process(tmqMsg, pInfo, totalMsgs);
           totalRows += data_msg_process(tmqMsg, pInfo, totalMsgs);
         }
@@ -947,7 +936,8 @@ void parseConsumeInfo() {
     token = strtok(g_stConfInfo.stThreads[i].topicString, delim);
     while (token != NULL) {
       // printf("%s\n", token );
-      tstrncpy(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic], token, sizeof(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic]));
+      tstrncpy(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic], token,
+               sizeof(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic]));
       ltrim(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic]);
       // printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]);
       g_stConfInfo.stThreads[i].numOfTopic++;
@@ -963,7 +953,8 @@ void parseConsumeInfo() {
         ltrim(pstr);
         char* ret = strchr(pstr, ch);
         memcpy(g_stConfInfo.stThreads[i].key[g_stConfInfo.stThreads[i].numOfKey], pstr, ret - pstr);
-        tstrncpy(g_stConfInfo.stThreads[i].value[g_stConfInfo.stThreads[i].numOfKey], ret + 1, sizeof(g_stConfInfo.stThreads[i].value[g_stConfInfo.stThreads[i].numOfKey]));
+        tstrncpy(g_stConfInfo.stThreads[i].value[g_stConfInfo.stThreads[i].numOfKey], ret + 1,
+                 sizeof(g_stConfInfo.stThreads[i].value[g_stConfInfo.stThreads[i].numOfKey]));
         // printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey],
         // g_stConfInfo.value[g_stConfInfo.numOfKey]);
         g_stConfInfo.stThreads[i].numOfKey++;
@@ -984,7 +975,7 @@ int32_t getConsumeInfo() {
   }
 
   sprintf(sqlStr, "select * from %s.consumeinfo", g_stConfInfo.cdbName);
-  TAOS_RES *pRes = taos_query(pConn, sqlStr);
+  TAOS_RES* pRes = taos_query(pConn, sqlStr);
   if (taos_errno(pRes) != 0) {
     taosFprintfFile(g_fp, "error in get consumeinfo for %s\n", taos_errstr(pRes));
     taosCloseFile(&g_fp);
@@ -1040,18 +1031,17 @@ int32_t getConsumeInfo() {
   return 0;
 }
 
-
 static int32_t omb_data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex, int64_t* lenOfRows) {
-  char    buf[16*1024];
+  char    buf[16 * 1024];
   int32_t totalRows = 0;
   int32_t totalLen = 0;
 
   // printf("topic: %s\n", tmq_get_topic_name(msg));
-  //int32_t     vgroupId = tmq_get_vgroup_id(msg);
-  //const char* dbName = tmq_get_db_name(msg);
+  // int32_t     vgroupId = tmq_get_vgroup_id(msg);
+  // const char* dbName = tmq_get_db_name(msg);
 
-  //taosFprintfFile(g_fp, "consumerId: %d, msg index:%" PRId64 "\n", pInfo->consumerId, msgIndex);
-  //taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table",
+  // taosFprintfFile(g_fp, "consumerId: %d, msg index:%" PRId64 "\n", pInfo->consumerId, msgIndex);
+  // taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table",
   //                 tmq_get_topic_name(msg), vgroupId);
 
   while (1) {
@@ -1061,9 +1051,9 @@ static int32_t omb_data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t m
 
     TAOS_FIELD* fields = taos_fetch_fields(msg);
     int32_t     numOfFields = taos_field_count(msg);
-    //int32_t*    length = taos_fetch_lengths(msg);
-    //int32_t     precision = taos_result_precision(msg);
-    //const char* tbName = tmq_get_table_name(msg);
+    // int32_t*    length = taos_fetch_lengths(msg);
+    // int32_t     precision = taos_result_precision(msg);
+    // const char* tbName = tmq_get_table_name(msg);
 
     taos_print_row(buf, row, fields, numOfFields);
     totalLen += strlen(buf);
@@ -1085,8 +1075,7 @@ void omb_loop_consume(SThreadInfo* pInfo) {
   char tmpString[128];
   taosFprintfFile(g_fp, "%s consumer id %d start to loop pull msg\n", getCurrentTimeString(tmpString),
                   pInfo->consumerId);
-  printf("%s consumer id %d start to loop pull msg\n", getCurrentTimeString(tmpString),
-				  pInfo->consumerId);
+  printf("%s consumer id %d start to loop pull msg\n", getCurrentTimeString(tmpString), pInfo->consumerId);
 
   pInfo->ts = taosGetTimestampMs();
 
@@ -1109,14 +1098,16 @@ void omb_loop_consume(SThreadInfo* pInfo) {
       if (currentPrintTime - lastPrintTime > 10 * 1000) {
         int64_t currentLenOfMsg = totalLenOfMsg - lastTotalLenOfMsg;
         int64_t deltaTime = currentPrintTime - lastPrintTime;
-	    printf("consumer id %d has currently cons total rows: %" PRId64 ", msgs: %" PRId64 ", rate: %.3f msgs/s, %.1f MB/s\n",
-		        pInfo->consumerId, totalRows, totalMsgs, 
-		        (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime, 
-		        currentLenOfMsg*1000.0/(1024*1024)/deltaTime);
+        printf("consumer id %d has currently cons total rows: %" PRId64 ", msgs: %" PRId64
+               ", rate: %.3f msgs/s, %.1f MB/s\n",
+               pInfo->consumerId, totalRows, totalMsgs, (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime,
+               currentLenOfMsg * 1000.0 / (1024 * 1024) / deltaTime);
 
-        taosFprintfFile(
-            g_fp, "consumer id %d has currently poll total msgs: %" PRId64 ", period cons rate: %.3f msgs/s, %.1f MB/s\n",
-            pInfo->consumerId, totalMsgs, (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime, currentLenOfMsg*1000.0/deltaTime);
+        taosFprintfFile(g_fp,
+                        "consumer id %d has currently poll total msgs: %" PRId64
+                        ", period cons rate: %.3f msgs/s, %.1f MB/s\n",
+                        pInfo->consumerId, totalMsgs, (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime,
+                        currentLenOfMsg * 1000.0 / deltaTime);
         lastPrintTime = currentPrintTime;
         lastTotalMsgs = totalMsgs;
         lastTotalLenOfMsg = totalLenOfMsg;
@@ -1128,10 +1119,10 @@ void omb_loop_consume(SThreadInfo* pInfo) {
       int64_t currentPrintTime = taosGetTimestampMs();
       int64_t currentLenOfMsg = totalLenOfMsg - lastTotalLenOfMsg;
       int64_t deltaTime = currentPrintTime - lastPrintTime;
-	  printf("consumer id %d has currently cons total rows: %" PRId64 ", msgs: %" PRId64 ", rate: %.3f msgs/s, %.1f MB/s\n",
-		      pInfo->consumerId, totalRows, totalMsgs, 
-		      (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime, 
-		      currentLenOfMsg*1000.0/(1024*1024)/deltaTime);
+      printf("consumer id %d has currently cons total rows: %" PRId64 ", msgs: %" PRId64
+             ", rate: %.3f msgs/s, %.1f MB/s\n",
+             pInfo->consumerId, totalRows, totalMsgs, (totalMsgs - lastTotalMsgs) * 1000.0 / deltaTime,
+             currentLenOfMsg * 1000.0 / (1024 * 1024) / deltaTime);
       break;
     }
   }
@@ -1139,10 +1130,8 @@ void omb_loop_consume(SThreadInfo* pInfo) {
   pInfo->consumeMsgCnt = totalMsgs;
   pInfo->consumeRowCnt = totalRows;
   pInfo->consumeLen = totalLenOfMsg;
-
 }
 
-
 void* ombConsumeThreadFunc(void* param) {
   SThreadInfo* pInfo = (SThreadInfo*)param;
 
@@ -1209,10 +1198,8 @@ void* ombConsumeThreadFunc(void* param) {
   return NULL;
 }
 
-
-
-static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type) {
-    TAOS_RES *res = taos_query(taos, command);
+static int queryDbExec(TAOS* taos, char* command, QUERY_TYPE type) {
+  TAOS_RES* res = taos_query(taos, command);
   int32_t   code = taos_errno(res);
 
   if (code != 0) {
@@ -1243,7 +1230,8 @@ void* ombProduceThreadFunc(void* param) {
   int64_t affectedRowsTotal = 0;
   int64_t sendMsgs = 0;
 
-  uint32_t  totalSendLoopTimes = g_stConfInfo.runDurationMinutes * 60 * 1000 / SEND_TIME_UNIT;  // send some msgs per 10ms 
+  uint32_t totalSendLoopTimes =
+      g_stConfInfo.runDurationMinutes * 60 * 1000 / SEND_TIME_UNIT;  // send some msgs per 10ms
   uint32_t batchPerTblTimes = pInfo->producerRate / 100 / g_stConfInfo.batchSize;
   uint32_t remainder = (pInfo->producerRate / 100) % g_stConfInfo.batchSize;
   if (remainder) {
@@ -1258,14 +1246,15 @@ void* ombProduceThreadFunc(void* param) {
     return NULL;
   }
 
-  printf("Produce Info: totalSendLoopTimes: %d, batchPerTblTimes: %d, producerRate: %d\n", totalSendLoopTimes, batchPerTblTimes, pInfo->producerRate);
+  printf("Produce Info: totalSendLoopTimes: %d, batchPerTblTimes: %d, producerRate: %d\n", totalSendLoopTimes,
+         batchPerTblTimes, pInfo->producerRate);
 
   char ctbName[128] = {0};
   sprintf(ctbName, "%s.ctb%d", g_stConfInfo.dbName, pInfo->consumerId);
 
   int64_t lastPrintTime = taosGetTimestampUs();
   int64_t totalMsgLen = 0;
-  //int64_t timeStamp = taosGetTimestampUs();
+  // int64_t timeStamp = taosGetTimestampUs();
   while (totalSendLoopTimes) {
     int64_t startTs = taosGetTimestampUs();
     for (int i = 0; i < batchPerTblTimes; ++i) {
@@ -1274,10 +1263,10 @@ void* ombProduceThreadFunc(void* param) {
         msgsOfSql = remainder;
       }
       int len = 0;
-	  len += snprintf(sqlBuf+len, MAX_SQL_LEN - len, "insert into %s values ", ctbName);
+      len += snprintf(sqlBuf + len, MAX_SQL_LEN - len, "insert into %s values ", ctbName);
       for (int j = 0; j < msgsOfSql; j++) {
         int64_t timeStamp = taosGetTimestampNs();
-            len += snprintf(sqlBuf+len, MAX_SQL_LEN - len, "(%" PRId64 ", \"%s\")", timeStamp, g_payload);
+        len += snprintf(sqlBuf + len, MAX_SQL_LEN - len, "(%" PRId64 ", \"%s\")", timeStamp, g_payload);
         sendMsgs++;
         pInfo->totalProduceMsgs++;
       }
@@ -1295,7 +1284,7 @@ void* ombProduceThreadFunc(void* param) {
 
       affectedRowsTotal += affectedRows;
 
-	  //printf("Produce Info: affectedRows: %" PRId64 "\n", affectedRows);
+      // printf("Produce Info: affectedRows: %" PRId64 "\n", affectedRows);
     }
     totalSendLoopTimes -= 1;
 
@@ -1304,7 +1293,7 @@ void* ombProduceThreadFunc(void* param) {
     int64_t delta = currentTs - startTs;
     if (delta < SEND_TIME_UNIT * 1000) {
       int64_t sleepLen = (int32_t)(SEND_TIME_UNIT * 1000 - delta);
-	  //printf("sleep %" PRId64 " us, use time: %" PRId64 " us\n", sleepLen, delta);
+      // printf("sleep %" PRId64 " us, use time: %" PRId64 " us\n", sleepLen, delta);
       taosUsleep((int32_t)sleepLen);
     }
 
@@ -1313,24 +1302,21 @@ void* ombProduceThreadFunc(void* param) {
     if (delta > 10 * 1000 * 1000) {
       printf("producer[%d] info: %" PRId64 " msgs, %" PRId64 " Byte, %" PRId64 " us, totalSendLoopTimes: %d\n",
              pInfo->consumerId, sendMsgs, totalMsgLen, delta, totalSendLoopTimes);
-		printf("producer[%d] rate: %1.f msgs/s, %1.f KB/s\n",
-				  pInfo->consumerId, 
-				  sendMsgs * 1000.0 * 1000 / delta, 
-				  (totalMsgLen / 1024.0)  / (delta / (1000*1000)));
+      printf("producer[%d] rate: %1.f msgs/s, %1.f KB/s\n", pInfo->consumerId, sendMsgs * 1000.0 * 1000 / delta,
+             (totalMsgLen / 1024.0) / (delta / (1000 * 1000)));
       lastPrintTime = currentTs;
       sendMsgs = 0;
       totalMsgLen = 0;
     }
   }
 
-  printf("affectedRowsTotal: %"PRId64"\n", affectedRowsTotal);
+  printf("affectedRowsTotal: %" PRId64 "\n", affectedRowsTotal);
   taos_close(pInfo->taos);
   pInfo->taos = NULL;
   taosMemoryFree(sqlBuf);
   return NULL;
 }
 
-
 void printProduceInfo(int64_t start) {
   int64_t totalMsgs = 0;
   int64_t totalLenOfMsgs = 0;
@@ -1347,15 +1333,12 @@ void printProduceInfo(int64_t start) {
   double tInMs = (double)t / 1000000.0;
   printf("Spent %.3f seconds to prod %" PRIu64 " msgs, %" PRIu64 " Byte\n\n", tInMs, totalMsgs, totalLenOfMsgs);
 
-  
   printf("Spent %.3f seconds to prod %" PRIu64 " msgs with %d producer(s), throughput: %.3f msgs/s, %.1f MB/s\n\n",
-		  tInMs, totalMsgs, g_stConfInfo.producers, 
-		  (double)totalMsgs / tInMs, 
-		  (double)totalLenOfMsgs/(1024.0*1024)/tInMs);
+         tInMs, totalMsgs, g_stConfInfo.producers, (double)totalMsgs / tInMs,
+         (double)totalLenOfMsgs / (1024.0 * 1024) / tInMs);
   return;
 }
 
-
 void startOmbConsume() {
   TdThreadAttr thattr;
   taosThreadAttrInit(&thattr);
@@ -1365,7 +1348,7 @@ void startOmbConsume() {
     TAOS* taos = createNewTaosConnect();
     if (taos == NULL) {
       taosFprintfFile(g_fp, "taos_connect() fail, can not create db, stbl, ctbl, topic!\n");
-      return ;
+      return;
     }
 
     char stbName[16] = "stb";
@@ -1376,16 +1359,19 @@ void startOmbConsume() {
     printf("SQL: %s\n", sql);
     queryDbExec(taos, sql, NO_INSERT_TYPE);
 
-    sprintf(sql, "create database if not exists %s precision 'ns' vgroups %d", g_stConfInfo.dbName, g_stConfInfo.producers);
+    sprintf(sql, "create database if not exists %s precision 'ns' vgroups %d", g_stConfInfo.dbName,
+            g_stConfInfo.producers);
     printf("SQL: %s\n", sql);
     queryDbExec(taos, sql, NO_INSERT_TYPE);
 
-    sprintf(sql, "create stable %s.%s (ts timestamp, payload binary(%d)) tags (t bigint) ", g_stConfInfo.dbName, stbName, g_stConfInfo.payloadLen);
+    sprintf(sql, "create stable %s.%s (ts timestamp, payload binary(%d)) tags (t bigint) ", g_stConfInfo.dbName,
+            stbName, g_stConfInfo.payloadLen);
     printf("SQL: %s\n", sql);
     queryDbExec(taos, sql, NO_INSERT_TYPE);
 
     for (int i = 0; i < g_stConfInfo.producers; i++) {
-	  sprintf(sql, "create table %s.%s%d using %s.stb tags(%d) ", g_stConfInfo.dbName, ctbPrefix, i, g_stConfInfo.dbName, i);
+      sprintf(sql, "create table %s.%s%d using %s.stb tags(%d) ", g_stConfInfo.dbName, ctbPrefix, i,
+              g_stConfInfo.dbName, i);
       printf("SQL: %s\n", sql);
       queryDbExec(taos, sql, NO_INSERT_TYPE);
     }
@@ -1395,7 +1381,6 @@ void startOmbConsume() {
     printf("SQL: %s\n", sql);
     queryDbExec(taos, sql, NO_INSERT_TYPE);
 
-
     int32_t producerRate = ceil(g_stConfInfo.producerRate / g_stConfInfo.producers);
 
     printf("==== create %d produce thread ====\n", g_stConfInfo.producers);
@@ -1454,16 +1439,15 @@ void startOmbConsume() {
   if (0 == t) t = 1;
 
   double tInMs = (double)t / 1000000.0;
-  taosFprintfFile(g_fp,
-                  "Spent %.3f seconds to poll msgs: %" PRIu64 " with %d thread(s), throughput: %.3f msgs/s, %.1f MB/s\n\n",
-                  tInMs, totalMsgs, g_stConfInfo.numOfThread, 
-                  (double)(totalMsgs / tInMs), 
-                  (double)totalLenOfMsgs/(1024*1024)/tInMs);
+  taosFprintfFile(
+      g_fp, "Spent %.3f seconds to poll msgs: %" PRIu64 " with %d thread(s), throughput: %.3f msgs/s, %.1f MB/s\n\n",
+      tInMs, totalMsgs, g_stConfInfo.numOfThread, (double)(totalMsgs / tInMs),
+      (double)totalLenOfMsgs / (1024 * 1024) / tInMs);
 
-  printf("Spent %.3f seconds to cons rows: %" PRIu64 " msgs: %" PRIu64 " with %d thread(s), throughput: %.3f msgs/s, %.1f MB/s\n\n",
-		  tInMs, totalRows, totalMsgs, g_stConfInfo.numOfThread, 
-		  (double)(totalMsgs / tInMs), 
-		  (double)totalLenOfMsgs/(1024*1024)/tInMs);
+  printf("Spent %.3f seconds to cons rows: %" PRIu64 " msgs: %" PRIu64
+         " with %d thread(s), throughput: %.3f msgs/s, %.1f MB/s\n\n",
+         tInMs, totalRows, totalMsgs, g_stConfInfo.numOfThread, (double)(totalMsgs / tInMs),
+         (double)totalLenOfMsgs / (1024 * 1024) / tInMs);
 
   taosFprintfFile(g_fp, "==== close tmqlog ====\n");
   taosCloseFile(&g_fp);
@@ -1471,7 +1455,6 @@ void startOmbConsume() {
   return;
 }
 
-
 int main(int32_t argc, char* argv[]) {
   parseArgument(argc, argv);