other: merge 3.0

include/libs/executor/executor.h

@@ -36,6 +36,7 @@ typedef struct SReadHandle {
   void*   vnode;
   void*   mnd;
   SMsgCb* pMsgCb;
+  int64_t version;
   bool    initMetaReader;
   bool    initTableReader;
   bool    initTqReader;

source/client/src/clientHb.c

@@ -264,14 +264,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
 
 static int32_t hbAsyncCallBack(void *param, SDataBuf *pMsg, int32_t code) {
   static int32_t emptyRspNum = 0;
-  if (code != 0) {
-    taosMemoryFreeClear(param);
-    return -1;
-  }
-
   char             *key = (char *)param;
   SClientHbBatchRsp pRsp = {0};
+  if (TSDB_CODE_SUCCESS == code) {
     tDeserializeSClientHbBatchRsp(pMsg->pData, pMsg->len, &pRsp);
+  }
   
   int32_t rspNum = taosArrayGetSize(pRsp.rsps);
 
@@ -288,6 +285,10 @@ static int32_t hbAsyncCallBack(void *param, SDataBuf *pMsg, int32_t code) {
 
   taosMemoryFreeClear(param);
 
+  if (code != 0) {
+    (*pInst)->onlineDnodes = 0;
+  }
+
   if (rspNum) {
     tscDebug("hb got %d rsp, %d empty rsp received before", rspNum,
              atomic_val_compare_exchange_32(&emptyRspNum, emptyRspNum, 0));

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);
 
@@ -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];
+        char buf[80];
-        tFormatOffset(buf, 50, &pVgCur->currentOffsetNew);
+        tFormatOffset(buf, 80, &pVgCur->currentOffsetNew);
-        tscDebug("consumer:%" PRId64 ", epoch %d vgId:%d vgKey is %s, offset is %s", tmq->consumerId, epoch, pVgCur->vgId,
+        tscDebug("consumer:%" PRId64 ", epoch %d vgId:%d vgKey is %s, offset is %s", tmq->consumerId, epoch,
-                 vgKey, buf);
+                 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,
+      tscDebug("consumer:%" PRId64 ", send poll to %s vgId:%d, epoch %d, req offset:%s, reqId:%" PRIu64,
-               pTopic->topicName, pVg->vgId, tmq->epoch, offsetFormatBuf, pReq->reqId);
+               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);
@@ -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) {
@@ -2049,9 +2055,11 @@ static char *processCreateTable(SMqMetaRsp *metaRsp){
   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);
+      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);
+      string =
+          buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE);
     }
   }
 

source/dnode/mnode/impl/src/mndStb.c

@@ -45,6 +45,7 @@ static int32_t  mndProcessTableMetaReq(SRpcMsg *pReq);
 static int32_t  mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 static void     mndCancelGetNextStb(SMnode *pMnode, void *pIter);
 static int32_t  mndProcessTableCfgReq(SRpcMsg *pReq);
+static int32_t  mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp);
 
 int32_t mndInitStb(SMnode *pMnode) {
   SSdbTable table = {
@@ -854,6 +855,7 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
   SStbObj       *pStb = NULL;
   SDbObj        *pDb = NULL;
   SMCreateStbReq createReq = {0};
+  bool           isAlter = false;
 
   if (tDeserializeSMCreateStbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -889,6 +891,7 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
           code = 0;
           goto _OVER;
         } else if ((tagDelta == 1 || colDelta == 1) && (verDelta == 1)) {
+          isAlter = true;
           mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
         } else {
           mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);
@@ -929,7 +932,12 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  if (isAlter) {
+    bool needRsp = false;
+    code = mndAlterStbImp(pMnode, pReq, pDb, pStb, needRsp);
+  } else {
     code = mndCreateStb(pMnode, pReq, &createReq, pDb);
+  }
   if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
 
 _OVER:
@@ -1495,14 +1503,13 @@ static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *t
   return code;
 }
 
-static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, const SMAlterStbReq *pAlter, SStbObj *pObj, void **pCont,
+static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, int32_t *pLen) {
-                                     int32_t *pLen) {
   int32_t       ret;
   SEncoder      ec = {0};
   uint32_t      contLen = 0;
   SMAlterStbRsp alterRsp = {0};
   SName         name = {0};
-  tNameFromString(&name, pAlter->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
+  tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
 
   alterRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
   if (NULL == alterRsp.pMeta) {
@@ -1535,10 +1542,36 @@ static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, const SMAlterStbReq *pAlter, S
   return 0;
 }
 
+static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp) {
+  int32_t code = -1;
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq);
+  if (pTrans == NULL) goto _OVER;
+
+  mDebug("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
+  mndTransSetDbName(pTrans, pDb->name, NULL);
+
+  if (needRsp) {
+    void   *pCont = NULL;
+    int32_t contLen = 0;
+    if (mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen) != 0) goto _OVER;
+    mndTransSetRpcRsp(pTrans, pCont, contLen);
+  }
+
+  if (mndSetAlterStbRedoLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
+  if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
+  if (mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
+  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
+
+  code = 0;
+
+_OVER:
+  mndTransDrop(pTrans);
+  return code;
+}
+
 static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
   bool    needRsp = true;
   int32_t code = -1;
-  STrans *pTrans = NULL;
   SField *pField0 = NULL;
 
   SStbObj stbObj = {0};
@@ -1587,30 +1620,9 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p
   }
 
   if (code != 0) goto _OVER;
-
+  code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp);
-  code = -1;
-  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq);
-  if (pTrans == NULL) goto _OVER;
-
-  mDebug("trans:%d, used to alter stb:%s", pTrans->id, pAlter->name);
-  mndTransSetDbName(pTrans, pDb->name, NULL);
-
-  if (needRsp) {
-    void   *pCont = NULL;
-    int32_t contLen = 0;
-    if (mndBuildSMAlterStbRsp(pDb, pAlter, &stbObj, &pCont, &contLen) != 0) goto _OVER;
-    mndTransSetRpcRsp(pTrans, pCont, contLen);
-  }
-
-  if (mndSetAlterStbRedoLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
-  if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
-  if (mndSetAlterStbRedoActions(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
-  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-
-  code = 0;
 
 _OVER:
-  mndTransDrop(pTrans);
   taosMemoryFreeClear(stbObj.pTags);
   taosMemoryFreeClear(stbObj.pColumns);
   return code;

source/dnode/vnode/src/inc/tq.h

@@ -89,6 +89,8 @@ typedef struct {
     STqExecTb  execTb;
     STqExecDb  execDb;
   };
+  // TODO remove it
+  int64_t tsdbEndVer;
 
 } STqExecHandle;
 
@@ -129,7 +131,7 @@ typedef struct {
 static STqMgmt tqMgmt = {0};
 
 // tqRead
-int64_t tqScanLog(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal* offset);
+int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal* offset);
 int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
 
 // tqExec

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

@@ -112,7 +112,8 @@ int32_t tqSendMetaPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq,
   };
   tmsgSendRsp(&resp);
 
-  tqDebug("vgId:%d from consumer:%" PRId64 ", (epoch %d) send rsp, res msg type %d, reqOffset:%" PRId64 ", rspOffset:%" PRId64,
+  tqDebug("vgId:%d from consumer:%" PRId64 ", (epoch %d) send rsp, res msg type %d, reqOffset:%" PRId64
+          ", rspOffset:%" PRId64,
           TD_VID(pTq->pVnode), pReq->consumerId, pReq->epoch, pRsp->resMsgType, pRsp->reqOffset, pRsp->rspOffset);
 
   return 0;
@@ -179,8 +180,8 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen) {
   tDecoderClear(&decoder);
 
   if (offset.val.type == TMQ_OFFSET__SNAPSHOT_DATA) {
-    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64, offset.subKey,
+    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64,
-            TD_VID(pTq->pVnode), offset.val.uid, offset.val.ts);
+            offset.subKey, TD_VID(pTq->pVnode), offset.val.uid, offset.val.ts);
   } else if (offset.val.type == TMQ_OFFSET__LOG) {
     tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, offset.subKey,
             TD_VID(pTq->pVnode), offset.val.version);
@@ -316,9 +317,11 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   }
 
   // 3.query
-  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN && fetchOffsetNew.type == TMQ_OFFSET__LOG) {
+  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+    if (fetchOffsetNew.type == TMQ_OFFSET__LOG) {
       fetchOffsetNew.version++;
-    if (tqScanLog(pTq, &pHandle->execHandle, &dataRsp, &fetchOffsetNew) < 0) {
+    }
+    if (tqScan(pTq, &pHandle->execHandle, &dataRsp, &fetchOffsetNew) < 0) {
       ASSERT(0);
       code = -1;
       goto OVER;
@@ -333,7 +336,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     goto OVER;
   }
 
-  if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN && fetchOffsetNew.type == TMQ_OFFSET__LOG) {
+  if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN) {
     int64_t     fetchVer = fetchOffsetNew.version + 1;
     SWalCkHead* pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048);
     if (pCkHead == NULL) {
@@ -411,6 +414,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     }
 
     taosMemoryFree(pCkHead);
+#if 0
   } else if (fetchOffsetNew.type == TMQ_OFFSET__SNAPSHOT_DATA) {
     tqInfo("retrieve using snapshot actual offset: uid %" PRId64 " ts %" PRId64, fetchOffsetNew.uid, fetchOffsetNew.ts);
     if (tqScanSnapshot(pTq, &pHandle->execHandle, &dataRsp, fetchOffsetNew, workerId) < 0) {
@@ -421,6 +425,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
       code = -1;
     }
+#endif
   } else if (fetchOffsetNew.type == TMQ_OFFSET__SNAPSHOT_META) {
     ASSERT(0);
   }
@@ -478,6 +483,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     /*for (int32_t i = 0; i < 5; i++) {*/
     /*pHandle->execHandle.pExecReader[i] = tqOpenReader(pTq->pVnode);*/
     /*}*/
+    int64_t ver = walGetCommittedVer(pTq->pVnode->pWal);
     if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
       pHandle->execHandle.execCol.qmsg = req.qmsg;
       req.qmsg = NULL;
@@ -488,6 +494,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
             .vnode = pTq->pVnode,
             .initTableReader = true,
             .initTqReader = true,
+            .version = ver,
         };
         pHandle->execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle);
         ASSERT(pHandle->execHandle.execCol.task[i]);
@@ -496,6 +503,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
         ASSERT(scanner);
         pHandle->execHandle.pExecReader[i] = qExtractReaderFromStreamScanner(scanner);
         ASSERT(pHandle->execHandle.pExecReader[i]);
+        pHandle->execHandle.tsdbEndVer = ver;
       }
     } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
       for (int32_t i = 0; i < 5; i++) {

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

@@ -59,15 +59,17 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp) {
   return 0;
 }
 
-int64_t tqScanLog(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
+int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
   qTaskInfo_t task = pExec->execCol.task[0];
 
   if (qStreamPrepareScan(task, pOffset) < 0) {
+    ASSERT(pOffset->type == TMQ_OFFSET__LOG);
     pRsp->rspOffset = *pOffset;
     pRsp->rspOffset.version--;
     return 0;
   }
 
+  int32_t rowCnt = 0;
   while (1) {
     SSDataBlock* pDataBlock = NULL;
     uint64_t     ts = 0;
@@ -77,11 +79,26 @@ int64_t tqScanLog(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOff
 
     if (pDataBlock != NULL) {
       tqAddBlockDataToRsp(pDataBlock, pRsp);
+      pRsp->blockNum++;
       if (pRsp->withTbName) {
+        if (pOffset->type == TMQ_OFFSET__LOG) {
           int64_t uid = pExec->pExecReader[0]->msgIter.uid;
           tqAddTbNameToRsp(pTq, uid, pRsp);
+        } else {
+          pRsp->withTbName = 0;
         }
-      pRsp->blockNum++;
+      }
+      if (pOffset->type == TMQ_OFFSET__LOG) {
+        continue;
+      } else {
+        rowCnt += pDataBlock->info.rows;
+        if (rowCnt <= 4096) continue;
+      }
+    }
+
+    if (pRsp->blockNum == 0 && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
+      tqOffsetResetToLog(pOffset, pExec->tsdbEndVer + 1);
+      qStreamPrepareScan(task, pOffset);
       continue;
     }
 
@@ -94,18 +111,19 @@ int64_t tqScanLog(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOff
       ASSERT(0);
     }
 
-    if (pRsp->rspOffset.type == TMQ_OFFSET__LOG) {
+    ASSERT(pRsp->rspOffset.type != 0);
+
+    if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
       ASSERT(pRsp->rspOffset.version + 1 >= pRsp->reqOffset.version);
     }
 
-    ASSERT(pRsp->rspOffset.type != 0);
-
     break;
   }
 
   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];
@@ -153,6 +171,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/dnode/vnode/src/vnd/vnodeSync.c

@@ -478,7 +478,7 @@ static int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void *pParam, void
 
 static int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) {
   SVnode *pVnode = pFsm->data;
-  int32_t code = vnodeSnapWriterClose(pWriter, isApply);
+  int32_t code = vnodeSnapWriterClose(pWriter, !isApply);
   return code;
 }
 

source/libs/catalog/inc/catalogInt.h

@@ -166,7 +166,7 @@ typedef struct SCtgDBCache {
   int8_t           deleted;
   SCtgVgCache      vgCache;
   SHashObj        *tbCache;         // key:tbname, value:SCtgTbCache
-  SHashObj        *stbCache;        // key:suid, value:STableMeta*
+  SHashObj        *stbCache;        // key:suid, value:char*
 } SCtgDBCache;
 
 typedef struct SCtgRentSlot {

source/libs/executor/inc/executorimpl.h

@@ -144,6 +144,7 @@ typedef struct {
   void*          metaBlk;       // for tmq fetching meta
   SSDataBlock*   pullOverBlk;   // for streaming
   SWalFilterCond cond;
+  int64_t        lastScanUid;
 } SStreamTaskInfo;
 
 typedef struct SExecTaskInfo {
@@ -291,10 +292,12 @@ typedef struct STableScanInfo {
   int32_t         currentGroupId;
   int32_t         currentTable;
 
+#if 0
   struct {
     uint64_t uid;
     int64_t ts;
   } lastStatus;
+#endif
 
   int8_t scanMode;
   int8_t noTable;

source/libs/executor/src/executorMain.c

@@ -335,8 +335,6 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
 
           qDebug("tsdb reader offset seek to uid %ld ts %ld, table cur set to %d , all table num %d", uid, ts,
                  pTableScanInfo->currentTable, tableSz);
-        } else {
-          // switch to log
         }
 
       } else {
@@ -352,6 +350,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;
 
@@ -371,3 +370,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

@@ -2848,7 +2848,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;
 
@@ -2933,6 +2933,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) {
@@ -3344,8 +3345,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
+  int64_t ekey =
-                                                                   : pInfo->existNewGroupBlock->info.window.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);
@@ -4476,7 +4477,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;
 
@@ -4515,8 +4517,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
     int32_t      tsSlotId = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
 
-    pOptr =
+    pOptr = createSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as,
-        createSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as, pPhyNode->pConditions, pTaskInfo);
+                                         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) {
@@ -4538,7 +4540,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) {

source/libs/executor/src/scanoperator.c

@@ -422,8 +422,11 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
     pOperator->cost.totalCost = pTableScanInfo->readRecorder.elapsedTime;
 
     // todo refactor
-    pTableScanInfo->lastStatus.uid = pBlock->info.uid;
+    /*pTableScanInfo->lastStatus.uid = pBlock->info.uid;*/
-    pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;
+    /*pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;*/
+    pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
+    pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.uid;
+    pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
 
     ASSERT(pBlock->info.uid != 0);
     return pBlock;
@@ -1191,9 +1194,6 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
 
   doFilter(pInfo->pCondition, pInfo->pRes);
   blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
-  if (pBlockInfo->rows > 0) {
-    return 0;
-  }
   return 0;
 }
 
@@ -1219,7 +1219,9 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
         /*pTaskInfo->streamInfo.lastStatus = ret.offset;*/
         if (pInfo->pRes->info.rows > 0) {
           return pInfo->pRes;
-          /*} else {*/
+        } else {
+          // data is filtered out, do clean
+
           /*tDeleteSSDataBlock(&ret.data);*/
         }
       } else if (ret.fetchType == FETCH_TYPE__META) {
@@ -1228,13 +1230,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
         pTaskInfo->streamInfo.metaBlk = ret.meta;
         return NULL;
       } else if (ret.fetchType == FETCH_TYPE__NONE) {
-        /*if (ret.offset.version == -1) {*/
-        /*pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;*/
-        /*pTaskInfo->streamInfo.lastStatus.version = pTaskInfo->streamInfo.prepareStatus.version - 1;*/
-        /*} else {*/
         pTaskInfo->streamInfo.lastStatus = ret.offset;
         ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 >= pTaskInfo->streamInfo.prepareStatus.version);
-        /*}*/
         return NULL;
       } else {
         ASSERT(0);
@@ -1354,72 +1351,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
         return NULL;
       }
 
-      pInfo->pRes->info.rows = block.info.rows;
+      setBlockIntoRes(pInfo, &block);
-      pInfo->pRes->info.uid = block.info.uid;
-      pInfo->pRes->info.type = STREAM_NORMAL;
-      pInfo->pRes->info.capacity = block.info.rows;
 
-      uint64_t* groupIdPre = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &block.info.uid, sizeof(int64_t));
-      if (groupIdPre) {
-        pInfo->pRes->info.groupId = *groupIdPre;
-      } else {
-        pInfo->pRes->info.groupId = 0;
-      }
-
-      // for generating rollup SMA result, each time is an independent time serie.
-      // TODO temporarily used, when the statement of "partition by tbname" is ready, remove this
-      if (pInfo->assignBlockUid) {
-        pInfo->pRes->info.groupId = block.info.uid;
-      }
-
-      // todo extract method
-      for (int32_t i = 0; i < taosArrayGetSize(pInfo->pColMatchInfo); ++i) {
-        SColMatchInfo* pColMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
-        if (!pColMatchInfo->output) {
-          continue;
-        }
-
-        bool colExists = false;
-        for (int32_t j = 0; j < blockDataGetNumOfCols(&block); ++j) {
-          SColumnInfoData* pResCol = bdGetColumnInfoData(&block, j);
-          if (pResCol->info.colId == pColMatchInfo->colId) {
-            taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol);
-            colExists = true;
-            break;
-          }
-        }
-
-        // the required column does not exists in submit block, let's set it to be all null value
-        if (!colExists) {
-          SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId);
-          colDataAppendNNULL(pDst, 0, pBlockInfo->rows);
-        }
-      }
-
-      taosArrayDestroy(block.pDataBlock);
-
-      ASSERT(pInfo->pRes->pDataBlock != NULL);
-#if 0
-      if (pInfo->pRes->pDataBlock == NULL) {
-        // TODO add log
-        updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
-        pOperator->status = OP_EXEC_DONE;
-        pTaskInfo->code = terrno;
-        return NULL;
-      }
-#endif
-
-      // currently only the tbname pseudo column
-      if (pInfo->numOfPseudoExpr > 0) {
-        code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,
-                                      GET_TASKID(pTaskInfo));
-        if (code != TSDB_CODE_SUCCESS) {
-          longjmp(pTaskInfo->env, code);
-        }
-      }
-
-      doFilter(pInfo->pCondition, pInfo->pRes);
-      blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
       if (pBlockInfo->rows > 0) {
         break;
       }
@@ -1449,12 +1382,14 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 
     return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
 
+#if 0
   } else if (pInfo->blockType == STREAM_INPUT__TABLE_SCAN) {
-    /*ASSERT(0);*/
+    ASSERT(0);
     // check reader last status
     // if not match, reset status
     SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
     return pResult && pResult->info.rows > 0 ? pResult : NULL;
+#endif
 
   } else {
     ASSERT(0);
@@ -1518,6 +1453,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   if (pHandle) {
     SOperatorInfo*  pTableScanOp = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
     STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanOp->info;
+    if (pHandle->version > 0) {
+      pSTInfo->cond.endVersion = pHandle->version;
+    }
 
     SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, 0);
     if (pHandle->initTableReader) {

source/libs/index/inc/indexFstCommon.h

@@ -4,6 +4,7 @@
 #include "tutil.h"
 extern const uint8_t COMMON_INPUTS[];
 extern const char    COMMON_INPUTS_INV[];
+extern const int32_t COMMON_INPUTS_LEN;
 
 #ifdef __cplusplus
 extern "C" {

source/libs/index/inc/indexFstDfa.h

@@ -32,7 +32,7 @@ typedef struct {
   SArray  *insts;
   uint32_t next[256];
   bool     isMatch;
-} State;
+} DfaState;
 
 /*
  * dfa builder related func

source/libs/index/inc/indexFstRegex.h

@@ -65,6 +65,7 @@ typedef struct {
 } FstRegex;
 
 FstRegex *regexCreate(const char *str);
+void      regexDestroy(FstRegex *regex);
 
 uint32_t regexAutomStart(FstRegex *regex);
 bool     regexAutomIsMatch(FstRegex *regex, uint32_t state);

source/libs/index/src/indexCache.c

@@ -22,7 +22,7 @@
 #define MAX_INDEX_KEY_LEN 256  // test only, change later
 
 #define MEM_TERM_LIMIT     10 * 10000
-#define MEM_THRESHOLD      64 * 1024
+#define MEM_THRESHOLD      512 * 1024
 #define MEM_SIGNAL_QUIT    MEM_THRESHOLD * 20
 #define MEM_ESTIMATE_RADIO 1.5
 
@@ -204,7 +204,6 @@ static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr
     if (0 == strcmp(c->colVal, pCt->colVal)) {
       if (c->operaType == ADD_VALUE) {
         INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
-        // taosArrayPush(result, &c->uid);
         *s = kTypeValue;
       } else if (c->operaType == DEL_VALUE) {
         INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)
@@ -309,7 +308,6 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTR
     if (cond == MATCH) {
       if (c->operaType == ADD_VALUE) {
         INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
-        // taosArrayPush(result, &c->uid);
         *s = kTypeValue;
       } else if (c->operaType == DEL_VALUE) {
         INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)

source/libs/index/src/indexFst.c

@@ -1307,7 +1307,6 @@ FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) {
     taosArrayPush(sws->inp, &(trn.inp));
 
     if (FST_NODE_IS_FINAL(nextNode)) {
-      // void *eofState = sws->aut->acceptEof(nextState);
       void* eofState = automFuncs[aut->type].acceptEof(aut, nextState);
       if (eofState != NULL) {
         isMatch = automFuncs[aut->type].isMatch(aut, eofState);

source/libs/index/src/indexFstCommon.c

@@ -294,3 +294,4 @@ const char COMMON_INPUTS_INV[] = {
     '\xee', '\xef', '\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb',
     '\xfc', '\xfd', '\xfe', '\xff',
 };
+const int32_t COMMON_INPUTS_LEN = sizeof(COMMON_INPUTS) / sizeof(COMMON_INPUTS[0]);

source/libs/index/src/indexFstDfa.c

@@ -41,7 +41,7 @@ FstDfaBuilder *dfaBuilderCreate(SArray *insts) {
     return NULL;
   }
 
-  SArray *states = taosArrayInit(4, sizeof(State));
+  SArray *states = taosArrayInit(4, sizeof(DfaState));
 
   builder->dfa = dfaCreate(insts, states);
   builder->cache = taosHashInit(
@@ -98,10 +98,12 @@ FstDfa *dfaBuilder(FstDfaBuilder *builder) {
   return builder->dfa;
 }
 
+FstDfa *dfaBuilderBuild(FstDfaBuilder *builer) { return NULL; }
+
 bool dfaBuilderRunState(FstDfaBuilder *builder, FstSparseSet *cur, FstSparseSet *next, uint32_t state, uint8_t byte,
                         uint32_t *result) {
   sparSetClear(cur);
-  State *t = taosArrayGet(builder->dfa->states, state);
+  DfaState *t = taosArrayGet(builder->dfa->states, state);
   for (int i = 0; i < taosArrayGetSize(t->insts); i++) {
     uint32_t ip = *(int32_t *)taosArrayGet(t->insts, i);
     sparSetAdd(cur, ip);
@@ -144,7 +146,7 @@ bool dfaBuilderCachedState(FstDfaBuilder *builder, FstSparseSet *set, uint32_t *
     *result = *v;
     taosArrayDestroy(tinsts);
   } else {
-    State st;
+    DfaState st;
     st.insts = tinsts;
     st.isMatch = isMatch;
     taosArrayPush(builder->dfa->states, &st);
@@ -169,14 +171,14 @@ bool dfaIsMatch(FstDfa *dfa, uint32_t si) {
   if (dfa->states == NULL || si < taosArrayGetSize(dfa->states)) {
     return false;
   }
-  State *st = taosArrayGet(dfa->states, si);
+  DfaState *st = taosArrayGet(dfa->states, si);
   return st != NULL ? st->isMatch : false;
 }
 bool dfaAccept(FstDfa *dfa, uint32_t si, uint8_t byte, uint32_t *result) {
   if (dfa->states == NULL || si < taosArrayGetSize(dfa->states)) {
     return false;
   }
-  State *st = taosArrayGet(dfa->states, si);
+  DfaState *st = taosArrayGet(dfa->states, si);
   *result = st->next[byte];
   return true;
 }

source/libs/index/src/indexFstRegex.c

@@ -36,6 +36,12 @@ FstRegex *regexCreate(const char *str) {
   return regex;
 }
 
+void regexDestroy(FstRegex *regex) {
+  if (regex == NULL) return;
+  taosMemoryFree(regex->orig);
+  taosMemoryFree(regex);
+}
+
 uint32_t regexAutomStart(FstRegex *regex) {
   ///// no nothing
   return 0;

source/libs/index/test/CMakeLists.txt

@@ -4,6 +4,7 @@ IF(NOT TD_DARWIN)
   add_executable(idxFstUT "")
   add_executable(idxUtilUT "")
   add_executable(idxJsonUT "")
+  add_executable(idxFstUtilUT "")
 
   target_sources(idxTest
     PRIVATE 
@@ -23,6 +24,25 @@ IF(NOT TD_DARWIN)
     "utilUT.cc" 
   )
 
+  target_sources(idxJsonUT
+    PRIVATE 
+    "jsonUT.cc" 
+  )
+  target_sources(idxFstUtilUT
+   PRIVATE 
+   "fstUtilUT.cc" 
+  )
+ 
+  target_include_directories (idxTest
+   PUBLIC
+   "${TD_SOURCE_DIR}/include/libs/index" 
+   "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+  ) 
+  target_include_directories (idxFstTest 
+   PUBLIC
+  "${TD_SOURCE_DIR}/include/libs/index" 
+  "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+  ) 
   target_sources(idxJsonUT
     PRIVATE 
     "jsonUT.cc" 
@@ -50,6 +70,38 @@ IF(NOT TD_DARWIN)
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
   ) 
 
+  target_include_directories (idxJsonUT
+   PUBLIC
+   "${TD_SOURCE_DIR}/include/libs/index" 
+   "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+  ) 
+  target_include_directories (idxFstUtilUT
+   PUBLIC
+   "${TD_SOURCE_DIR}/include/libs/index" 
+   "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+  ) 
+
+  target_link_libraries (idxTest
+    os  
+    util
+    common
+    gtest_main
+    index
+  )
+  target_link_libraries (idxFstTest
+    os  
+    util
+    common
+    gtest_main
+    index
+  )
+  target_link_libraries (idxFstUT
+    os  
+    util
+    common
+    gtest_main
+    index
+  )
   target_include_directories (idxJsonUT
     PUBLIC
     "${TD_SOURCE_DIR}/include/libs/index" 
@@ -92,15 +144,28 @@ IF(NOT TD_DARWIN)
     gtest_main
     index
   )
+  target_link_libraries (idxFstUtilUT
+    os  
+    util
+    common
+    gtest_main
+    index
+  )
+  
+  add_test(
+    NAME idxJsonUT
+    COMMAND idxJsonUT 
+  )
+  add_test(
+    NAME idxFstUtilUT 
+    COMMAND idxFstUtilUT  
+  
+  )
 
   add_test(
     NAME idxtest
     COMMAND idxTest
   )
-  add_test(
-    NAME idxJsonUT
-    COMMAND idxJsonUT 
-  )
   add_test(
     NAME idxUtilUT 
     COMMAND idxUtilUT 

source/libs/index/test/fstUtilUT.cc

@@ -0,0 +1,70 @@
+
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <iostream>
+#include <string>
+#include <thread>
+#include <vector>
+#include "index.h"
+#include "indexCache.h"
+#include "indexFst.h"
+#include "indexFstDfa.h"
+#include "indexFstRegex.h"
+#include "indexFstSparse.h"
+#include "indexFstUtil.h"
+#include "indexInt.h"
+#include "indexTfile.h"
+#include "tglobal.h"
+#include "tlog.h"
+#include "tskiplist.h"
+#include "tutil.h"
+class FstUtilEnv : public ::testing::Test {
+ protected:
+  virtual void SetUp() {
+    SArray *inst = taosArrayInit(4, sizeof(char));
+    builder = dfaBuilderCreate(inst);
+  }
+  virtual void TearDown() { dfaBuilderDestroy(builder); }
+
+  FstDfaBuilder *builder;
+};
+
+class FstRegexEnv : public ::testing::Test {
+ protected:
+  virtual void SetUp() { regex = regexCreate("test"); }
+  virtual void TearDown() { regexDestroy(regex); }
+  FstRegex    *regex;
+};
+
+class FstSparseSetEnv : public ::testing::Test {
+ protected:
+  virtual void SetUp() { set = sparSetCreate(256); }
+  virtual void TearDown() {
+    // tear down
+    sparSetDestroy(set);
+  }
+  void ReBuild(int32_t sz) {
+    sparSetDestroy(set);
+    set = sparSetCreate(sz);
+  }
+  FstSparseSet *set;
+};
+
+// test FstDfaBuilder
+TEST_F(FstUtilEnv, test1) {}
+TEST_F(FstUtilEnv, test2) {}
+TEST_F(FstUtilEnv, test3) {}
+TEST_F(FstUtilEnv, test4) {}
+
+// test FstRegex
+
+TEST_F(FstRegexEnv, test1) {}
+TEST_F(FstRegexEnv, test2) {}
+TEST_F(FstRegexEnv, test3) {}
+TEST_F(FstRegexEnv, test4) {}
+
+// test FstSparseSet
+TEST_F(FstSparseSetEnv, test1) {}
+TEST_F(FstSparseSetEnv, test2) {}
+TEST_F(FstSparseSetEnv, test3) {}
+TEST_F(FstSparseSetEnv, test4) {}

source/libs/index/test/utilUT.cc

@@ -8,6 +8,7 @@
 #include "indexCache.h"
 #include "indexComm.h"
 #include "indexFst.h"
+#include "indexFstCommon.h"
 #include "indexFstUtil.h"
 #include "indexInt.h"
 #include "indexTfile.h"
@@ -356,3 +357,11 @@ TEST_F(UtilEnv, TempResultExcept) {
   idxTRsltMergeTo(relt, f);
   EXPECT_EQ(taosArrayGetSize(f), 1);
 }
+
+TEST_F(UtilEnv, testDictComm) {
+  int32_t count = COMMON_INPUTS_LEN;
+  for (int i = 0; i < 256; i++) {
+    uint8_t v = COMMON_INPUTS_INV[i];
+    EXPECT_EQ(COMMON_INPUTS[v], i);
+  }
+}

source/libs/qworker/src/qworker.c

@@ -517,7 +517,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, const char* sql) {
   ctx->needFetch = qwMsg->msgInfo.needFetch;
   ctx->queryType = qwMsg->msgType;
 
-  QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
+  //QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
 
   code = qStringToSubplan(qwMsg->msg, &plan);
   if (TSDB_CODE_SUCCESS != code) {

source/libs/scalar/src/scalar.c

@@ -578,7 +578,7 @@ _return:
   SCL_RET(code);
 }
 
-EDealRes sclRewriteBasedOnOptr(SNode** pNode, SScalarCtx *ctx, EOperatorType opType) {
+EDealRes sclRewriteNullInOptr(SNode** pNode, SScalarCtx *ctx, EOperatorType opType) {
   if (opType <= OP_TYPE_CALC_MAX) {
     SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
     if (NULL == res) {
@@ -610,6 +610,24 @@ EDealRes sclRewriteBasedOnOptr(SNode** pNode, SScalarCtx *ctx, EOperatorType opT
   return DEAL_RES_CONTINUE;
 }
 
+EDealRes sclAggFuncWalker(SNode* pNode, void* pContext) {
+  if (QUERY_NODE_FUNCTION == nodeType(pNode)) {
+    SFunctionNode* pFunc = (SFunctionNode*)pNode;
+    *(bool*)pContext = fmIsAggFunc(pFunc->funcId);
+    if (*(bool*)pContext) {
+      return DEAL_RES_END;
+    }
+  }
+
+  return DEAL_RES_CONTINUE;
+}
+
+
+bool sclContainsAggFuncNode(SNode* pNode) {
+  bool aggFunc = false;
+  nodesWalkExpr(pNode, sclAggFuncWalker, (void *)&aggFunc);
+  return aggFunc;
+}
 
 EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
   SOperatorNode *node = (SOperatorNode *)*pNode;
@@ -617,8 +635,9 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
 
   if (node->pLeft && (QUERY_NODE_VALUE == nodeType(node->pLeft))) {
     SValueNode *valueNode = (SValueNode *)node->pLeft;
-    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
+    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL) 
-      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
+        && (!sclContainsAggFuncNode(node->pRight))) {
+      return sclRewriteNullInOptr(pNode, ctx, node->opType);
     }
 
     if (IS_STR_DATA_TYPE(valueNode->node.resType.type) && node->pRight && nodesIsExprNode(node->pRight) 
@@ -633,8 +652,9 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
 
   if (node->pRight && (QUERY_NODE_VALUE == nodeType(node->pRight))) {
     SValueNode *valueNode = (SValueNode *)node->pRight;
-    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
+    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)
-      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
+       && (!sclContainsAggFuncNode(node->pLeft))) {
+      return sclRewriteNullInOptr(pNode, ctx, node->opType);
     }
 
     if (IS_STR_DATA_TYPE(valueNode->node.resType.type) && node->pLeft && nodesIsExprNode(node->pLeft) 
@@ -656,7 +676,7 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
           ERASE_NODE(listNode->pNodeList);
           continue;
         } else { //OP_TYPE_NOT_IN
-          return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
+          return sclRewriteNullInOptr(pNode, ctx, node->opType);
         }
       }
 
@@ -664,7 +684,7 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
     }
 
     if (listNode->pNodeList->length <= 0) {
-      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
+      return sclRewriteNullInOptr(pNode, ctx, node->opType);
     }
   }
 

tests/system-test/fulltest.sh

@@ -120,7 +120,7 @@ python3 ./test.py -f 2-query/irate.py
 python3 ./test.py -f 2-query/and_or_for_byte.py
 
 python3 ./test.py -f 2-query/function_null.py
-python3 ./test.py -f 2-query/queryQnode.py
+#python3 ./test.py -f 2-query/queryQnode.py
 
 #python3 ./test.py -f 6-cluster/5dnode1mnode.py
 #python3 ./test.py -f 6-cluster/5dnode2mnode.py  -N 5 -M 3