refactor(stream)

source/client/src/tmq.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "cJSON.h"
 #include "clientInt.h"
 #include "clientLog.h"
 #include "parser.h"
@@ -23,7 +24,6 @@
 #include "tqueue.h"
 #include "tref.h"
 #include "ttimer.h"
-#include "cJSON.h"
 
 int32_t tmqAskEp(tmq_t* tmq, bool async);
 
@@ -106,8 +106,8 @@ struct tmq_t {
   tsem_t rspSem;
 };
 
-struct tmq_raw_data{
-  void    *raw_meta;
+struct tmq_raw_data {
+  void*   raw_meta;
   int32_t raw_meta_len;
   int16_t raw_meta_type;
 };
@@ -953,6 +953,8 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
     goto FAIL;
   }
 
+  tscInfo("consumer %ld is setup, consumer group %s", pTmq->consumerId, pTmq->groupId);
+
   return pTmq;
 
 FAIL:
@@ -1194,10 +1196,10 @@ bool tmqUpdateEp2(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
       for (int32_t j = 0; j < vgNumCur; j++) {
         SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, j);
         sprintf(vgKey, "%s:%d", pTopicCur->topicName, pVgCur->vgId);
-        char buf[50];
-        tFormatOffset(buf, 50, &pVgCur->currentOffsetNew);
-        tscDebug("consumer:%" PRId64 ", epoch %d vgId:%d vgKey is %s, offset is %s", tmq->consumerId, epoch, pVgCur->vgId,
-                 vgKey, buf);
+        char buf[80];
+        tFormatOffset(buf, 80, &pVgCur->currentOffsetNew);
+        tscDebug("consumer:%" PRId64 ", epoch %d vgId:%d vgKey is %s, offset is %s", tmq->consumerId, epoch,
+                 pVgCur->vgId, vgKey, buf);
         taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffsetNew, sizeof(STqOffsetVal));
       }
     }
@@ -1564,7 +1566,8 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
       int32_t      vgStatus = atomic_val_compare_exchange_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE, TMQ_VG_STATUS__WAIT);
       if (vgStatus != TMQ_VG_STATUS__IDLE) {
         int32_t vgSkipCnt = atomic_add_fetch_32(&pVg->vgSkipCnt, 1);
-        tscTrace("consumer:%" PRId64 ", epoch %d skip vgId:%d skip cnt %d", tmq->consumerId, tmq->epoch, pVg->vgId, vgSkipCnt);
+        tscTrace("consumer:%" PRId64 ", epoch %d skip vgId:%d skip cnt %d", tmq->consumerId, tmq->epoch, pVg->vgId,
+                 vgSkipCnt);
         continue;
         /*if (vgSkipCnt < 10000) continue;*/
 #if 0
@@ -1620,8 +1623,8 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
 
       char offsetFormatBuf[80];
       tFormatOffset(offsetFormatBuf, 80, &pVg->currentOffsetNew);
-      tscDebug("consumer:%" PRId64 ", send poll to %s vgId:%d, epoch %d, req offset:%s, reqId:%" PRIu64, tmq->consumerId,
-               pTopic->topicName, pVg->vgId, tmq->epoch, offsetFormatBuf, pReq->reqId);
+      tscDebug("consumer:%" PRId64 ", send poll to %s vgId:%d, epoch %d, req offset:%s, reqId:%" PRIu64,
+               tmq->consumerId, pTopic->topicName, pVg->vgId, tmq->epoch, offsetFormatBuf, pReq->reqId);
       /*printf("send vgId:%d %" PRId64 "\n", pVg->vgId, pVg->currentOffset);*/
       asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);
       pVg->pollCnt++;
@@ -1669,7 +1672,8 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       int32_t consumerEpoch = atomic_load_32(&tmq->epoch);
       if (pollRspWrapper->dataRsp.head.epoch == consumerEpoch) {
         SMqClientVg* pVg = pollRspWrapper->vgHandle;
-        /*printf("vgId:%d offset %" PRId64 " up to %" PRId64 "\n", pVg->vgId, pVg->currentOffset, rspMsg->msg.rspOffset);*/
+        /*printf("vgId:%d offset %" PRId64 " up to %" PRId64 "\n", pVg->vgId, pVg->currentOffset,
+         * rspMsg->msg.rspOffset);*/
         pVg->currentOffsetNew = pollRspWrapper->dataRsp.rspOffset;
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
         if (pollRspWrapper->dataRsp.blockNum == 0) {
@@ -1691,7 +1695,8 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       int32_t            consumerEpoch = atomic_load_32(&tmq->epoch);
       if (pollRspWrapper->metaRsp.head.epoch == consumerEpoch) {
         SMqClientVg* pVg = pollRspWrapper->vgHandle;
-        /*printf("vgId:%d offset %" PRId64 " up to %" PRId64 "\n", pVg->vgId, pVg->currentOffset, rspMsg->msg.rspOffset);*/
+        /*printf("vgId:%d offset %" PRId64 " up to %" PRId64 "\n", pVg->vgId, pVg->currentOffset,
+         * rspMsg->msg.rspOffset);*/
         pVg->currentOffsetNew.version = pollRspWrapper->metaRsp.rspOffset;
         pVg->currentOffsetNew.type = TMQ_OFFSET__LOG;
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
@@ -1848,9 +1853,9 @@ const char* tmq_get_table_name(TAOS_RES* res) {
   return NULL;
 }
 
-tmq_raw_data *tmq_get_raw_meta(TAOS_RES* res) {
+tmq_raw_data* tmq_get_raw_meta(TAOS_RES* res) {
   if (TD_RES_TMQ_META(res)) {
-    tmq_raw_data *raw = taosMemoryCalloc(1, sizeof(tmq_raw_data));
+    tmq_raw_data*  raw = taosMemoryCalloc(1, sizeof(tmq_raw_data));
     SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
     raw->raw_meta = pMetaRspObj->metaRsp.metaRsp;
     raw->raw_meta_len = pMetaRspObj->metaRsp.metaRspLen;
@@ -1860,7 +1865,8 @@ tmq_raw_data *tmq_get_raw_meta(TAOS_RES* res) {
   return NULL;
 }
 
-static char *buildCreateTableJson(SSchemaWrapper *schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id, int8_t t){
+static char* buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id,
+                                  int8_t t) {
   char*  string = NULL;
   cJSON* json = cJSON_CreateObject();
   if (json == NULL) {
@@ -1870,31 +1876,31 @@ static char *buildCreateTableJson(SSchemaWrapper *schemaRow, SSchemaWrapper* sch
   cJSON_AddItemToObject(json, "type", type);
 
   char uid[32] = {0};
-  sprintf(uid, "%"PRIi64, id);
+  sprintf(uid, "%" PRIi64, id);
   cJSON* id_ = cJSON_CreateString(uid);
   cJSON_AddItemToObject(json, "id", id_);
   cJSON* tableName = cJSON_CreateString(name);
   cJSON_AddItemToObject(json, "tableName", tableName);
   cJSON* tableType = cJSON_CreateString(t == TSDB_NORMAL_TABLE ? "normal" : "super");
   cJSON_AddItemToObject(json, "tableType", tableType);
-//  cJSON* version = cJSON_CreateNumber(1);
-//  cJSON_AddItemToObject(json, "version", version);
+  //  cJSON* version = cJSON_CreateNumber(1);
+  //  cJSON_AddItemToObject(json, "version", version);
 
   cJSON* columns = cJSON_CreateArray();
-  for(int i = 0; i < schemaRow->nCols; i++){
-    cJSON* column = cJSON_CreateObject();
-    SSchema *s = schemaRow->pSchema + i;
-    cJSON* cname = cJSON_CreateString(s->name);
+  for (int i = 0; i < schemaRow->nCols; i++) {
+    cJSON*   column = cJSON_CreateObject();
+    SSchema* s = schemaRow->pSchema + i;
+    cJSON*   cname = cJSON_CreateString(s->name);
     cJSON_AddItemToObject(column, "name", cname);
     cJSON* ctype = cJSON_CreateNumber(s->type);
     cJSON_AddItemToObject(column, "type", ctype);
-    if(s->type == TSDB_DATA_TYPE_BINARY){
+    if (s->type == TSDB_DATA_TYPE_BINARY) {
       int32_t length = s->bytes - VARSTR_HEADER_SIZE;
-      cJSON* cbytes = cJSON_CreateNumber(length);
+      cJSON*  cbytes = cJSON_CreateNumber(length);
       cJSON_AddItemToObject(column, "length", cbytes);
-    }else if (s->type == TSDB_DATA_TYPE_NCHAR){
-      int32_t length = (s->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE;
-      cJSON* cbytes = cJSON_CreateNumber(length);
+    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {
+      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
+      cJSON*  cbytes = cJSON_CreateNumber(length);
       cJSON_AddItemToObject(column, "length", cbytes);
     }
     cJSON_AddItemToArray(columns, column);
@@ -1902,20 +1908,20 @@ static char *buildCreateTableJson(SSchemaWrapper *schemaRow, SSchemaWrapper* sch
   cJSON_AddItemToObject(json, "columns", columns);
 
   cJSON* tags = cJSON_CreateArray();
-  for(int i = 0; schemaTag && i < schemaTag->nCols; i++){
-    cJSON* tag = cJSON_CreateObject();
-    SSchema *s = schemaTag->pSchema + i;
-    cJSON* tname = cJSON_CreateString(s->name);
+  for (int i = 0; schemaTag && i < schemaTag->nCols; i++) {
+    cJSON*   tag = cJSON_CreateObject();
+    SSchema* s = schemaTag->pSchema + i;
+    cJSON*   tname = cJSON_CreateString(s->name);
     cJSON_AddItemToObject(tag, "name", tname);
     cJSON* ttype = cJSON_CreateNumber(s->type);
     cJSON_AddItemToObject(tag, "type", ttype);
-    if(s->type == TSDB_DATA_TYPE_BINARY){
+    if (s->type == TSDB_DATA_TYPE_BINARY) {
       int32_t length = s->bytes - VARSTR_HEADER_SIZE;
-      cJSON* cbytes = cJSON_CreateNumber(length);
+      cJSON*  cbytes = cJSON_CreateNumber(length);
       cJSON_AddItemToObject(tag, "length", cbytes);
-    }else if (s->type == TSDB_DATA_TYPE_NCHAR){
-      int32_t length = (s->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE;
-      cJSON* cbytes = cJSON_CreateNumber(length);
+    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {
+      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
+      cJSON*  cbytes = cJSON_CreateNumber(length);
       cJSON_AddItemToObject(tag, "length", cbytes);
     }
     cJSON_AddItemToArray(tags, tag);
@@ -1927,13 +1933,13 @@ static char *buildCreateTableJson(SSchemaWrapper *schemaRow, SSchemaWrapper* sch
   return string;
 }
 
-static char *processCreateStb(SMqMetaRsp *metaRsp){
+static char* processCreateStb(SMqMetaRsp* metaRsp) {
   SVCreateStbReq req = {0};
   SDecoder       coder;
-  char*  string = NULL;
+  char*          string = NULL;
 
   // decode and process req
-  void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
   int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
 
@@ -1949,7 +1955,7 @@ _err:
   return string;
 }
 
-static char *buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t id){
+static char* buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t id) {
   char*  string = NULL;
   cJSON* json = cJSON_CreateObject();
   if (json == NULL) {
@@ -1958,7 +1964,7 @@ static char *buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t
   cJSON* type = cJSON_CreateString("create");
   cJSON_AddItemToObject(json, "type", type);
   char cid[32] = {0};
-  sprintf(cid, "%"PRIi64, id);
+  sprintf(cid, "%" PRIi64, id);
   cJSON* cid_ = cJSON_CreateString(cid);
   cJSON_AddItemToObject(json, "id", cid_);
 
@@ -1968,19 +1974,19 @@ static char *buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t
   cJSON_AddItemToObject(json, "tableType", tableType);
 
   char sid_[32] = {0};
-  sprintf(sid_, "%"PRIi64, sid);
+  sprintf(sid_, "%" PRIi64, sid);
   cJSON* using = cJSON_CreateString(sid_);
   cJSON_AddItemToObject(json, "using", using);
-//  cJSON* version = cJSON_CreateNumber(1);
-//  cJSON_AddItemToObject(json, "version", version);
+  //  cJSON* version = cJSON_CreateNumber(1);
+  //  cJSON_AddItemToObject(json, "version", version);
 
   cJSON* tags = cJSON_CreateArray();
 
-  if (tTagIsJson(pTag)) {   // todo
+  if (tTagIsJson(pTag)) {  // todo
     char* pJson = parseTagDatatoJson(pTag);
 
     cJSON* tag = cJSON_CreateObject();
-    cJSON* tname = cJSON_CreateString("unknown");                 // todo
+    cJSON* tname = cJSON_CreateString("unknown");  // todo
     cJSON_AddItemToObject(tag, "name", tname);
     cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON);
     cJSON_AddItemToObject(tag, "type", ttype);
@@ -1999,12 +2005,12 @@ static char *buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t
     goto end;
   }
 
-  for(int i = 0; i < taosArrayGetSize(pTagVals); i++){
+  for (int i = 0; i < taosArrayGetSize(pTagVals); i++) {
     STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i);
 
     cJSON* tag = cJSON_CreateObject();
-//    cJSON* tname = cJSON_CreateNumber(pTagVal->cid);
-    cJSON* tname = cJSON_CreateString("unkonwn");    // todo
+    //    cJSON* tname = cJSON_CreateNumber(pTagVal->cid);
+    cJSON* tname = cJSON_CreateString("unkonwn");  // todo
     cJSON_AddItemToObject(tag, "name", tname);
     cJSON* ttype = cJSON_CreateNumber(pTagVal->type);
     cJSON_AddItemToObject(tag, "type", ttype);
@@ -2032,13 +2038,13 @@ end:
   return string;
 }
 
-static char *processCreateTable(SMqMetaRsp *metaRsp){
+static char* processCreateTable(SMqMetaRsp* metaRsp) {
   SDecoder           decoder = {0};
   SVCreateTbBatchReq req = {0};
-  SVCreateTbReq     *pCreateReq;
-  char              *string = NULL;
+  SVCreateTbReq*     pCreateReq;
+  char*              string = NULL;
   // decode
-  void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
   int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);
   tDecoderInit(&decoder, data, len);
   if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
@@ -2048,27 +2054,29 @@ static char *processCreateTable(SMqMetaRsp *metaRsp){
   // loop to create table
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
     pCreateReq = req.pReqs + iReq;
-    if(pCreateReq->type == TSDB_CHILD_TABLE){
-      string = buildCreateCTableJson((STag*)pCreateReq->ctb.pTag, pCreateReq->ctb.suid, pCreateReq->name, pCreateReq->uid);
-    }else if(pCreateReq->type == TSDB_NORMAL_TABLE){
-      string = buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE);
+    if (pCreateReq->type == TSDB_CHILD_TABLE) {
+      string =
+          buildCreateCTableJson((STag*)pCreateReq->ctb.pTag, pCreateReq->ctb.suid, pCreateReq->name, pCreateReq->uid);
+    } else if (pCreateReq->type == TSDB_NORMAL_TABLE) {
+      string =
+          buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE);
     }
   }
 
   tDecoderClear(&decoder);
 
-  _exit:
+_exit:
   tDecoderClear(&decoder);
   return string;
 }
 
-static char *processAlterTable(SMqMetaRsp *metaRsp){
-  SDecoder           decoder = {0};
-  SVAlterTbReq       vAlterTbReq = {0};
-  char              *string = NULL;
+static char* processAlterTable(SMqMetaRsp* metaRsp) {
+  SDecoder     decoder = {0};
+  SVAlterTbReq vAlterTbReq = {0};
+  char*        string = NULL;
 
   // decode
-  void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
   int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);
   tDecoderInit(&decoder, data, len);
   if (tDecodeSVAlterTbReq(&decoder, &vAlterTbReq) < 0) {
@@ -2081,8 +2089,8 @@ static char *processAlterTable(SMqMetaRsp *metaRsp){
   }
   cJSON* type = cJSON_CreateString("alter");
   cJSON_AddItemToObject(json, "type", type);
-//  cJSON* uid = cJSON_CreateNumber(id);
-//  cJSON_AddItemToObject(json, "uid", uid);
+  //  cJSON* uid = cJSON_CreateNumber(id);
+  //  cJSON_AddItemToObject(json, "uid", uid);
   cJSON* tableName = cJSON_CreateString(vAlterTbReq.tbName);
   cJSON_AddItemToObject(json, "tableName", tableName);
   cJSON* tableType = cJSON_CreateString(vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL ? "child" : "normal");
@@ -2097,43 +2105,43 @@ static char *processAlterTable(SMqMetaRsp *metaRsp){
       cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);
       cJSON_AddItemToObject(json, "colType", colType);
 
-      if(vAlterTbReq.type == TSDB_DATA_TYPE_BINARY){
+      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY) {
         int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;
-        cJSON* cbytes = cJSON_CreateNumber(length);
+        cJSON*  cbytes = cJSON_CreateNumber(length);
         cJSON_AddItemToObject(json, "colLength", cbytes);
-      }else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR){
-        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE;
-        cJSON* cbytes = cJSON_CreateNumber(length);
+      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {
+        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
+        cJSON*  cbytes = cJSON_CreateNumber(length);
         cJSON_AddItemToObject(json, "colLength", cbytes);
       }
       break;
     }
-    case TSDB_ALTER_TABLE_DROP_COLUMN:{
+    case TSDB_ALTER_TABLE_DROP_COLUMN: {
       cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_DROP_COLUMN);
       cJSON_AddItemToObject(json, "alterType", alterType);
       cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);
       cJSON_AddItemToObject(json, "colName", colName);
       break;
     }
-    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:{
+    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {
       cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES);
       cJSON_AddItemToObject(json, "alterType", alterType);
       cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);
       cJSON_AddItemToObject(json, "colName", colName);
       cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);
       cJSON_AddItemToObject(json, "colType", colType);
-      if(vAlterTbReq.type == TSDB_DATA_TYPE_BINARY){
+      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY) {
         int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;
-        cJSON* cbytes = cJSON_CreateNumber(length);
+        cJSON*  cbytes = cJSON_CreateNumber(length);
         cJSON_AddItemToObject(json, "colLength", cbytes);
-      }else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR){
-        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE;
-        cJSON* cbytes = cJSON_CreateNumber(length);
+      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {
+        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
+        cJSON*  cbytes = cJSON_CreateNumber(length);
         cJSON_AddItemToObject(json, "colLength", cbytes);
       }
       break;
     }
-    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:{
+    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {
       cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME);
       cJSON_AddItemToObject(json, "alterType", alterType);
       cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);
@@ -2142,12 +2150,12 @@ static char *processAlterTable(SMqMetaRsp *metaRsp){
       cJSON_AddItemToObject(json, "colNewName", colNewName);
       break;
     }
-    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL:{
+    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: {
       cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_TAG_VAL);
       cJSON_AddItemToObject(json, "alterType", alterType);
       cJSON* tagName = cJSON_CreateString(vAlterTbReq.tagName);
       cJSON_AddItemToObject(json, "colName", tagName);
-      cJSON* colValue = cJSON_CreateString("invalid, todo");   // todo
+      cJSON* colValue = cJSON_CreateString("invalid, todo");  // todo
       cJSON_AddItemToObject(json, "colValue", colValue);
       cJSON* isNull = cJSON_CreateBool(vAlterTbReq.isNull);
       cJSON_AddItemToObject(json, "colValueNull", isNull);
@@ -2158,18 +2166,18 @@ static char *processAlterTable(SMqMetaRsp *metaRsp){
   }
   string = cJSON_PrintUnformatted(json);
 
-  _exit:
+_exit:
   tDecoderClear(&decoder);
   return string;
 }
 
-static char *processDropSTable(SMqMetaRsp *metaRsp){
-  SDecoder           decoder = {0};
-  SVDropStbReq       req = {0};
-  char              *string = NULL;
+static char* processDropSTable(SMqMetaRsp* metaRsp) {
+  SDecoder     decoder = {0};
+  SVDropStbReq req = {0};
+  char*        string = NULL;
 
   // decode
-  void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
   int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);
   tDecoderInit(&decoder, data, len);
   if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
@@ -2183,7 +2191,7 @@ static char *processDropSTable(SMqMetaRsp *metaRsp){
   cJSON* type = cJSON_CreateString("drop");
   cJSON_AddItemToObject(json, "type", type);
   char uid[32] = {0};
-  sprintf(uid, "%"PRIi64, req.suid);
+  sprintf(uid, "%" PRIi64, req.suid);
   cJSON* id = cJSON_CreateString(uid);
   cJSON_AddItemToObject(json, "id", id);
   cJSON* tableName = cJSON_CreateString(req.name);
@@ -2193,18 +2201,18 @@ static char *processDropSTable(SMqMetaRsp *metaRsp){
 
   string = cJSON_PrintUnformatted(json);
 
-  _exit:
+_exit:
   tDecoderClear(&decoder);
   return string;
 }
 
-static char *processDropTable(SMqMetaRsp *metaRsp){
-  SDecoder           decoder = {0};
-  SVDropTbBatchReq   req = {0};
-  char              *string = NULL;
+static char* processDropTable(SMqMetaRsp* metaRsp) {
+  SDecoder         decoder = {0};
+  SVDropTbBatchReq req = {0};
+  char*            string = NULL;
 
   // decode
-  void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));
   int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);
   tDecoderInit(&decoder, data, len);
   if (tDecodeSVDropTbBatchReq(&decoder, &req) < 0) {
@@ -2217,57 +2225,57 @@ static char *processDropTable(SMqMetaRsp *metaRsp){
   }
   cJSON* type = cJSON_CreateString("drop");
   cJSON_AddItemToObject(json, "type", type);
-//  cJSON* uid = cJSON_CreateNumber(id);
-//  cJSON_AddItemToObject(json, "uid", uid);
-//  cJSON* tableType = cJSON_CreateString("normal");
-//  cJSON_AddItemToObject(json, "tableType", tableType);
+  //  cJSON* uid = cJSON_CreateNumber(id);
+  //  cJSON_AddItemToObject(json, "uid", uid);
+  //  cJSON* tableType = cJSON_CreateString("normal");
+  //  cJSON_AddItemToObject(json, "tableType", tableType);
 
   cJSON* tableNameList = cJSON_CreateArray();
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
     SVDropTbReq* pDropTbReq = req.pReqs + iReq;
 
-    cJSON* tableName = cJSON_CreateString(pDropTbReq->name);   // todo
+    cJSON* tableName = cJSON_CreateString(pDropTbReq->name);  // todo
     cJSON_AddItemToArray(tableNameList, tableName);
   }
   cJSON_AddItemToObject(json, "tableNameList", tableNameList);
 
   string = cJSON_PrintUnformatted(json);
 
-  _exit:
+_exit:
   tDecoderClear(&decoder);
   return string;
 }
 
-char *tmq_get_json_meta(TAOS_RES *res){
+char* tmq_get_json_meta(TAOS_RES* res) {
   if (!TD_RES_TMQ_META(res)) {
     return NULL;
   }
 
   SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res;
-  if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_STB){
+  if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_STB) {
     return processCreateStb(&pMetaRspObj->metaRsp);
-  }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_STB){
+  } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_STB) {
     return processCreateStb(&pMetaRspObj->metaRsp);
-  }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_STB){
+  } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_STB) {
     return processDropSTable(&pMetaRspObj->metaRsp);
-  }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_TABLE){
+  } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_TABLE) {
     return processCreateTable(&pMetaRspObj->metaRsp);
-  }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_TABLE){
+  } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_TABLE) {
     return processAlterTable(&pMetaRspObj->metaRsp);
-  }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_TABLE){
+  } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_TABLE) {
     return processDropTable(&pMetaRspObj->metaRsp);
   }
   return NULL;
 }
 
-static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
+static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) {
   SVCreateStbReq req = {0};
   SDecoder       coder;
   SMCreateStbReq pReq = {0};
-  int32_t code = TSDB_CODE_SUCCESS;
-  SRequestObj* pRequest = NULL;
+  int32_t        code = TSDB_CODE_SUCCESS;
+  SRequestObj*   pRequest = NULL;
 
-  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
+  STscObj* pTscObj = acquireTscObj(*(int64_t*)taos);
   if (NULL == pTscObj) {
     code = TSDB_CODE_TSC_DISCONNECTED;
     goto end;
@@ -2278,12 +2286,12 @@ static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  if(!pRequest->pDb){
+  if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
   }
   // decode and process req
-  void* data = POINTER_SHIFT(meta, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(meta, sizeof(SMsgHead));
   int32_t len = metaLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
   if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
@@ -2292,16 +2300,16 @@ static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
   }
   // build create stable
   pReq.pColumns = taosArrayInit(req.schemaRow.nCols, sizeof(SField));
-  for(int32_t i = 0; i < req.schemaRow.nCols; i++){
+  for (int32_t i = 0; i < req.schemaRow.nCols; i++) {
     SSchema* pSchema = req.schemaRow.pSchema + i;
-    SField   field   = {.type = pSchema->type, .bytes = pSchema->bytes};
+    SField   field = {.type = pSchema->type, .bytes = pSchema->bytes};
     strcpy(field.name, pSchema->name);
     taosArrayPush(pReq.pColumns, &field);
   }
   pReq.pTags = taosArrayInit(req.schemaTag.nCols, sizeof(SField));
-  for(int32_t i = 0; i < req.schemaTag.nCols; i++){
+  for (int32_t i = 0; i < req.schemaTag.nCols; i++) {
     SSchema* pSchema = req.schemaTag.pSchema + i;
-    SField   field   = {.type = pSchema->type, .bytes = pSchema->bytes};
+    SField   field = {.type = pSchema->type, .bytes = pSchema->bytes};
     strcpy(field.name, pSchema->name);
     taosArrayPush(pReq.pTags, &field);
   }
@@ -2327,7 +2335,7 @@ static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
   }
   tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq);
 
-  SQuery      pQuery = {0};
+  SQuery pQuery = {0};
   pQuery.execMode = QUERY_EXEC_MODE_RPC;
   pQuery.pCmdMsg = &pCmdMsg;
   pQuery.msgType = pQuery.pCmdMsg->msgType;
@@ -2337,21 +2345,21 @@ static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
   code = pRequest->code;
   taosMemoryFree(pCmdMsg.pMsg);
 
-  end:
+end:
   destroyRequest(pRequest);
   tFreeSMCreateStbReq(&pReq);
   tDecoderClear(&coder);
   return code;
 }
 
-static int32_t taosDropStb(TAOS *taos, void *meta, int32_t metaLen){
+static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) {
   SVDropStbReq req = {0};
   SDecoder     coder;
   SMDropStbReq pReq = {0};
-  int32_t code = TSDB_CODE_SUCCESS;
+  int32_t      code = TSDB_CODE_SUCCESS;
   SRequestObj* pRequest = NULL;
 
-  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
+  STscObj* pTscObj = acquireTscObj(*(int64_t*)taos);
   if (NULL == pTscObj) {
     code = TSDB_CODE_TSC_DISCONNECTED;
     goto end;
@@ -2362,12 +2370,12 @@ static int32_t taosDropStb(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  if(!pRequest->pDb){
+  if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
   }
   // decode and process req
-  void* data = POINTER_SHIFT(meta, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(meta, sizeof(SMsgHead));
   int32_t len = metaLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
   if (tDecodeSVDropStbReq(&coder, &req) < 0) {
@@ -2393,7 +2401,7 @@ static int32_t taosDropStb(TAOS *taos, void *meta, int32_t metaLen){
   }
   tSerializeSMDropStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq);
 
-  SQuery      pQuery = {0};
+  SQuery pQuery = {0};
   pQuery.execMode = QUERY_EXEC_MODE_RPC;
   pQuery.pCmdMsg = &pCmdMsg;
   pQuery.msgType = pQuery.pCmdMsg->msgType;
@@ -2403,7 +2411,7 @@ static int32_t taosDropStb(TAOS *taos, void *meta, int32_t metaLen){
   code = pRequest->code;
   taosMemoryFree(pCmdMsg.pMsg);
 
-  end:
+end:
   destroyRequest(pRequest);
   tDecoderClear(&coder);
   return code;
@@ -2416,18 +2424,18 @@ typedef struct SVgroupCreateTableBatch {
 } SVgroupCreateTableBatch;
 
 static void destroyCreateTbReqBatch(void* data) {
-  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*) data;
+  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data;
   taosArrayDestroy(pTbBatch->req.pArray);
 }
 
-static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
-  SVCreateTbBatchReq  req             = {0};
-  SDecoder            coder           = {0};
-  int32_t             code            = TSDB_CODE_SUCCESS;
-  SRequestObj        *pRequest        = NULL;
-  SQuery             *pQuery          = NULL;
-  SHashObj           *pVgroupHashmap  = NULL;
-  STscObj            *pTscObj         = acquireTscObj(*(int64_t *)taos);
+static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
+  SVCreateTbBatchReq req = {0};
+  SDecoder           coder = {0};
+  int32_t            code = TSDB_CODE_SUCCESS;
+  SRequestObj*       pRequest = NULL;
+  SQuery*            pQuery = NULL;
+  SHashObj*          pVgroupHashmap = NULL;
+  STscObj*           pTscObj = acquireTscObj(*(int64_t*)taos);
 
   if (NULL == pTscObj) {
     code = TSDB_CODE_TSC_DISCONNECTED;
@@ -2439,12 +2447,12 @@ static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  if(!pRequest->pDb){
+  if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
   }
   // decode and process req
-  void* data = POINTER_SHIFT(meta, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(meta, sizeof(SMsgHead));
   int32_t len = metaLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
   if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) {
@@ -2452,8 +2460,8 @@ static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  SVCreateTbReq     *pCreateReq = NULL;
-  SCatalog* pCatalog = NULL;
+  SVCreateTbReq* pCreateReq = NULL;
+  SCatalog*      pCatalog = NULL;
   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
   if (code != TSDB_CODE_SUCCESS) {
     goto end;
@@ -2467,15 +2475,15 @@ static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
   taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch);
 
   SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,
-      .requestId = pRequest->requestId,
-      .requestObjRefId = pRequest->self,
-      .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
+                           .requestId = pRequest->requestId,
+                           .requestObjRefId = pRequest->self,
+                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
   // loop to create table
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
     pCreateReq = req.pReqs + iReq;
 
     SVgroupInfo pInfo = {0};
-    SName pName;
+    SName       pName;
     toName(pTscObj->acctId, pRequest->pDb, pCreateReq->name, &pName);
     code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo);
     if (code != TSDB_CODE_SUCCESS) {
@@ -2507,7 +2515,7 @@ static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
   pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
   pQuery->msgType = TDMT_VND_CREATE_TABLE;
   pQuery->stableQuery = false;
-  pQuery->pRoot   = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT);
+  pQuery->pRoot = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT);
 
   code = rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
   if (code != TSDB_CODE_SUCCESS) {
@@ -2515,10 +2523,10 @@ static int32_t taosCreateTable(TAOS *taos, void *meta, int32_t metaLen){
   }
 
   launchQueryImpl(pRequest, pQuery, false, NULL);
-  pQuery  = NULL;          // no need to free in the end
-  code    = pRequest->code;
+  pQuery = NULL;  // no need to free in the end
+  code = pRequest->code;
 
-  end:
+end:
   taosHashCleanup(pVgroupHashmap);
   destroyRequest(pRequest);
   tDecoderClear(&coder);
@@ -2537,14 +2545,14 @@ static void destroyDropTbReqBatch(void* data) {
   taosArrayDestroy(pTbBatch->req.pArray);
 }
 
-static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
-  SVDropTbBatchReq    req             = {0};
-  SDecoder            coder           = {0};
-  int32_t             code            = TSDB_CODE_SUCCESS;
-  SRequestObj        *pRequest        = NULL;
-  SQuery             *pQuery          = NULL;
-  SHashObj           *pVgroupHashmap  = NULL;
-  STscObj            *pTscObj         = acquireTscObj(*(int64_t *)taos);
+static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) {
+  SVDropTbBatchReq req = {0};
+  SDecoder         coder = {0};
+  int32_t          code = TSDB_CODE_SUCCESS;
+  SRequestObj*     pRequest = NULL;
+  SQuery*          pQuery = NULL;
+  SHashObj*        pVgroupHashmap = NULL;
+  STscObj*         pTscObj = acquireTscObj(*(int64_t*)taos);
 
   if (NULL == pTscObj) {
     code = TSDB_CODE_TSC_DISCONNECTED;
@@ -2556,12 +2564,12 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  if(!pRequest->pDb){
+  if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
   }
   // decode and process req
-  void* data = POINTER_SHIFT(meta, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(meta, sizeof(SMsgHead));
   int32_t len = metaLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
   if (tDecodeSVDropTbBatchReq(&coder, &req) < 0) {
@@ -2569,8 +2577,8 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  SVDropTbReq     *pDropReq = NULL;
-  SCatalog        *pCatalog = NULL;
+  SVDropTbReq* pDropReq = NULL;
+  SCatalog*    pCatalog = NULL;
   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
   if (code != TSDB_CODE_SUCCESS) {
     goto end;
@@ -2584,15 +2592,15 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
   taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch);
 
   SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,
-      .requestId = pRequest->requestId,
-      .requestObjRefId = pRequest->self,
-      .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
+                           .requestId = pRequest->requestId,
+                           .requestObjRefId = pRequest->self,
+                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
   // loop to create table
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
     pDropReq = req.pReqs + iReq;
 
     SVgroupInfo pInfo = {0};
-    SName pName;
+    SName       pName;
     toName(pTscObj->acctId, pRequest->pDb, pDropReq->name, &pName);
     code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo);
     if (code != TSDB_CODE_SUCCESS) {
@@ -2622,7 +2630,7 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
   pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
   pQuery->msgType = TDMT_VND_DROP_TABLE;
   pQuery->stableQuery = false;
-  pQuery->pRoot   = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT);
+  pQuery->pRoot = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT);
 
   code = rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
   if (code != TSDB_CODE_SUCCESS) {
@@ -2630,10 +2638,10 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
   }
 
   launchQueryImpl(pRequest, pQuery, false, NULL);
-  pQuery  = NULL;          // no need to free in the end
-  code    = pRequest->code;
+  pQuery = NULL;  // no need to free in the end
+  code = pRequest->code;
 
-  end:
+end:
   taosHashCleanup(pVgroupHashmap);
   destroyRequest(pRequest);
   tDecoderClear(&coder);
@@ -2641,15 +2649,15 @@ static int32_t taosDropTable(TAOS *taos, void *meta, int32_t metaLen){
   return code;
 }
 
-static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){
-  SVAlterTbReq        req             = {0};
-  SDecoder            coder           = {0};
-  int32_t             code            = TSDB_CODE_SUCCESS;
-  SRequestObj        *pRequest        = NULL;
-  SQuery             *pQuery          = NULL;
-  SArray             *pArray          = NULL;
-  SVgDataBlocks      *pVgData         = NULL;
-  STscObj            *pTscObj         = acquireTscObj(*(int64_t *)taos);
+static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) {
+  SVAlterTbReq   req = {0};
+  SDecoder       coder = {0};
+  int32_t        code = TSDB_CODE_SUCCESS;
+  SRequestObj*   pRequest = NULL;
+  SQuery*        pQuery = NULL;
+  SArray*        pArray = NULL;
+  SVgDataBlocks* pVgData = NULL;
+  STscObj*       pTscObj = acquireTscObj(*(int64_t*)taos);
 
   if (NULL == pTscObj) {
     code = TSDB_CODE_TSC_DISCONNECTED;
@@ -2661,12 +2669,12 @@ static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){
     goto end;
   }
 
-  if(!pRequest->pDb){
+  if (!pRequest->pDb) {
     code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     goto end;
   }
   // decode and process req
-  void* data = POINTER_SHIFT(meta, sizeof(SMsgHead));
+  void*   data = POINTER_SHIFT(meta, sizeof(SMsgHead));
   int32_t len = metaLen - sizeof(SMsgHead);
   tDecoderInit(&coder, data, len);
   if (tDecodeSVAlterTbReq(&coder, &req) < 0) {
@@ -2675,23 +2683,23 @@ static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){
   }
 
   // do not deal TSDB_ALTER_TABLE_UPDATE_OPTIONS
-  if(req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS){
+  if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) {
     goto end;
   }
 
-  SCatalog        *pCatalog = NULL;
+  SCatalog* pCatalog = NULL;
   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
   if (code != TSDB_CODE_SUCCESS) {
     goto end;
   }
 
   SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,
-      .requestId = pRequest->requestId,
-      .requestObjRefId = pRequest->self,
-      .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
+                           .requestId = pRequest->requestId,
+                           .requestObjRefId = pRequest->self,
+                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
 
   SVgroupInfo pInfo = {0};
-  SName pName = {0};
+  SName       pName = {0};
   toName(pTscObj->acctId, pRequest->pDb, req.tbName, &pName);
   code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo);
   if (code != TSDB_CODE_SUCCESS) {
@@ -2725,7 +2733,7 @@ static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){
   pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
   pQuery->msgType = TDMT_VND_ALTER_TABLE;
   pQuery->stableQuery = false;
-  pQuery->pRoot   = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
+  pQuery->pRoot = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT);
 
   code = rewriteToVnodeModifyOpStmt(pQuery, pArray);
   if (code != TSDB_CODE_SUCCESS) {
@@ -2733,14 +2741,14 @@ static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){
   }
 
   launchQueryImpl(pRequest, pQuery, false, NULL);
-  pQuery  = NULL;          // no need to free in the end
+  pQuery = NULL;  // no need to free in the end
   pVgData = NULL;
-  pArray  = NULL;
-  code    = pRequest->code;
+  pArray = NULL;
+  code = pRequest->code;
 
 end:
   taosArrayDestroy(pArray);
-  if(pVgData) taosMemoryFreeClear(pVgData->pData);
+  if (pVgData) taosMemoryFreeClear(pVgData->pData);
   taosMemoryFreeClear(pVgData);
   destroyRequest(pRequest);
   tDecoderClear(&coder);
@@ -2748,22 +2756,22 @@ end:
   return code;
 }
 
-int32_t taos_write_raw_meta(TAOS *taos, tmq_raw_data *raw_meta){
+int32_t taos_write_raw_meta(TAOS* taos, tmq_raw_data* raw_meta) {
   if (!taos || !raw_meta) {
     return TSDB_CODE_INVALID_PARA;
   }
 
-  if(raw_meta->raw_meta_type == TDMT_VND_CREATE_STB) {
+  if (raw_meta->raw_meta_type == TDMT_VND_CREATE_STB) {
     return taosCreateStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
-  }else if(raw_meta->raw_meta_type == TDMT_VND_ALTER_STB){
+  } else if (raw_meta->raw_meta_type == TDMT_VND_ALTER_STB) {
     return taosCreateStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
-  }else if(raw_meta->raw_meta_type == TDMT_VND_DROP_STB){
+  } else if (raw_meta->raw_meta_type == TDMT_VND_DROP_STB) {
     return taosDropStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
-  }else if(raw_meta->raw_meta_type == TDMT_VND_CREATE_TABLE){
+  } else if (raw_meta->raw_meta_type == TDMT_VND_CREATE_TABLE) {
     return taosCreateTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
-  }else if(raw_meta->raw_meta_type == TDMT_VND_ALTER_TABLE){
+  } else if (raw_meta->raw_meta_type == TDMT_VND_ALTER_TABLE) {
     return taosAlterTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
-  }else if(raw_meta->raw_meta_type == TDMT_VND_DROP_TABLE){
+  } else if (raw_meta->raw_meta_type == TDMT_VND_DROP_TABLE) {
     return taosDropTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len);
   }
   return TSDB_CODE_INVALID_PARA;

source/dnode/vnode/src/tq/tqExec.c

@@ -89,8 +89,10 @@ int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffset
         }
       }
       if (pOffset->type == TMQ_OFFSET__LOG) {
+        continue;
+      } else {
         rowCnt += pDataBlock->info.rows;
-        if (rowCnt >= 4096) break;
+        if (rowCnt <= 4096) continue;
       }
       continue;
     }
@@ -116,6 +118,7 @@ int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffset
   return 0;
 }
 
+#if 0
 int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal offset, int32_t workerId) {
   ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
   qTaskInfo_t task = pExec->execCol.task[workerId];
@@ -163,6 +166,7 @@ int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, S
 
   return 0;
 }
+#endif
 
 int32_t tqLogScanExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataRsp* pRsp, int32_t workerId) {
   if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) {

source/libs/executor/inc/executorimpl.h

@@ -299,10 +299,12 @@ typedef struct STableScanInfo {
   uint64_t        queryId;   // todo remove it
   uint64_t        taskId;    // todo remove it
 
+#if 0
   struct {
     uint64_t uid;
     int64_t ts;
   } lastStatus;
+#endif
 
   int8_t scanMode;
   int8_t noTable;

source/libs/executor/src/executorMain.c

@@ -351,6 +351,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
   return 0;
 }
 
+#if 0
 int32_t qStreamPrepareTsdbScan(qTaskInfo_t tinfo, uint64_t uid, int64_t ts) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
 
@@ -370,3 +371,4 @@ int32_t qGetStreamScanStatus(qTaskInfo_t tinfo, uint64_t* uid, int64_t* ts) {
 
   return doGetScanStatus(pTaskInfo->pRoot, uid, ts);
 }
+#endif

source/libs/executor/src/executorimpl.c

@@ -2463,7 +2463,7 @@ static void destroySortedMergeOperatorInfo(void* param, int32_t numOfOutput) {
 
   blockDataDestroy(pInfo->binfo.pRes);
   cleanupAggSup(&pInfo->aggSup);
-  
+
   taosMemoryFreeClear(param);
 }
 
@@ -2844,7 +2844,7 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
     }
   }
 }
-
+#if 0
 int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts) {
   uint8_t type = pOperator->operatorType;
 
@@ -2930,6 +2930,7 @@ int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts) {
 
   return TSDB_CODE_SUCCESS;
 }
+#endif
 
 // this is a blocking operator
 static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
@@ -3341,8 +3342,8 @@ static void doHandleRemainBlockForNewGroupImpl(SFillOperatorInfo* pInfo, SResult
                                                SExecTaskInfo* pTaskInfo) {
   pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
 
-  int64_t ekey = Q_STATUS_EQUAL(pTaskInfo->status, TASK_COMPLETED) ? pInfo->win.ekey
-                                                                   : pInfo->existNewGroupBlock->info.window.ekey;
+  int64_t ekey =
+      Q_STATUS_EQUAL(pTaskInfo->status, TASK_COMPLETED) ? pInfo->win.ekey : pInfo->existNewGroupBlock->info.window.ekey;
   taosResetFillInfo(pInfo->pFillInfo, getFillInfoStart(pInfo->pFillInfo));
 
   taosFillSetStartInfo(pInfo->pFillInfo, pInfo->existNewGroupBlock->info.rows, ekey);
@@ -3678,14 +3679,14 @@ void cleanupBasicInfo(SOptrBasicInfo* pInfo) {
 void destroyBasicOperatorInfo(void* param, int32_t numOfOutput) {
   SOptrBasicInfo* pInfo = (SOptrBasicInfo*)param;
   cleanupBasicInfo(pInfo);
-  
+
   taosMemoryFreeClear(param);
 }
 
 void destroyAggOperatorInfo(void* param, int32_t numOfOutput) {
   SAggOperatorInfo* pInfo = (SAggOperatorInfo*)param;
-  cleanupBasicInfo(&pInfo->binfo);  
-  
+  cleanupBasicInfo(&pInfo->binfo);
+
   taosMemoryFreeClear(param);
 }
 
@@ -3694,7 +3695,7 @@ void destroySFillOperatorInfo(void* param, int32_t numOfOutput) {
   pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
   pInfo->pRes = blockDataDestroy(pInfo->pRes);
   taosMemoryFreeClear(pInfo->p);
-  
+
   taosMemoryFreeClear(param);
 }
 
@@ -3706,7 +3707,7 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
   cleanupBasicInfo(&pInfo->binfo);
   cleanupAggSup(&pInfo->aggSup);
   taosArrayDestroy(pInfo->pPseudoColInfo);
-  
+
   taosMemoryFreeClear(param);
 }
 
@@ -3724,7 +3725,7 @@ static void destroyIndefinitOperatorInfo(void* param, int32_t numOfOutput) {
   taosArrayDestroy(pInfo->pPseudoColInfo);
   cleanupAggSup(&pInfo->aggSup);
   cleanupExprSupp(&pInfo->scalarSup);
-  
+
   taosMemoryFreeClear(param);
 }
 
@@ -3743,7 +3744,7 @@ void doDestroyExchangeOperatorInfo(void* param) {
   }
 
   tsem_destroy(&pExInfo->ready);
-  
+
   taosMemoryFreeClear(param);
 }
 
@@ -3972,7 +3973,7 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
   pInfo->pFillInfo = taosCreateFillInfo(order, w.skey, 0, capacity, numOfCols, pInterval, fillType, pColInfo, id);
 
   pInfo->win = win;
-  pInfo->p   = taosMemoryCalloc(numOfCols, POINTER_BYTES);
+  pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
   if (pInfo->pFillInfo == NULL || pInfo->p == NULL) {
     taosMemoryFree(pInfo->pFillInfo);
     taosMemoryFree(pInfo->p);
@@ -4465,7 +4466,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
                           .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision};
 
     int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
-    pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, pPhyNode->pConditions, pTaskInfo);
+    pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId,
+                                                   pPhyNode->pConditions, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
     SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
 
@@ -4504,8 +4506,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
     int32_t      tsSlotId = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
 
-    pOptr =
-        createSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as, pPhyNode->pConditions, pTaskInfo);
+    pOptr = createSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as,
+                                         pPhyNode->pConditions, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION == type) {
     pOptr = createStreamSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION == type) {
@@ -4527,7 +4529,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
     SColumnNode* pColNode = (SColumnNode*)((STargetNode*)pStateNode->pStateKey)->pExpr;
     SColumn      col = extractColumnFromColumnNode(pColNode);
-    pOptr = createStatewindowOperatorInfo(ops[0], pExprInfo, num, pResBlock, &as, tsSlotId, &col, pPhyNode->pConditions, pTaskInfo);
+    pOptr = createStatewindowOperatorInfo(ops[0], pExprInfo, num, pResBlock, &as, tsSlotId, &col, pPhyNode->pConditions,
+                                          pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE == type) {
     pOptr = createStreamStateAggOperatorInfo(ops[0], pPhyNode, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
@@ -4798,7 +4801,7 @@ int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pT
         return TSDB_CODE_OUT_OF_MEMORY;
       }
       pInserterParam->readHandle = readHandle;
-      
+
       *pParam = pInserterParam;
       break;
     }