Merge branch '3.0' into fix/TD-17012

include/common/tmsg.h

@@ -55,8 +55,9 @@ extern int32_t tMsgDict[];
 
 #define TMSG_SEG_CODE(TYPE) (((TYPE)&0xff00) >> 8)
 #define TMSG_SEG_SEQ(TYPE)  ((TYPE)&0xff)
-#define TMSG_INFO(TYPE)     tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)]
-#define TMSG_INDEX(TYPE)    (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE))
+#define TMSG_INFO(TYPE) \
+  (((TYPE) >= 0 && (TYPE) < TDMT_MAX) ? tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)] : 0)
+#define TMSG_INDEX(TYPE) (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE))
 
 typedef uint16_t tmsg_t;
 

include/libs/nodes/plannodes.h

@@ -82,6 +82,7 @@ typedef struct SScanLogicNode {
 typedef struct SJoinLogicNode {
   SLogicNode node;
   EJoinType  joinType;
+  SNode*     pMergeCondition;
   SNode*     pOnConditions;
   bool       isSingleTableJoin;
 } SJoinLogicNode;
@@ -329,6 +330,7 @@ typedef struct SInterpFuncPhysiNode {
 typedef struct SJoinPhysiNode {
   SPhysiNode node;
   EJoinType  joinType;
+  SNode*     pMergeCondition;
   SNode*     pOnConditions;
   SNodeList* pTargets;
 } SJoinPhysiNode;

include/libs/nodes/querynodes.h

@@ -376,6 +376,7 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit
 typedef enum ECollectColType { COLLECT_COL_TYPE_COL = 1, COLLECT_COL_TYPE_TAG, COLLECT_COL_TYPE_ALL } ECollectColType;
 int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char* pTableAlias, ECollectColType type,
                             SNodeList** pCols);
+int32_t nodesCollectColumnsFromNode(SNode* node, const char* pTableAlias, ECollectColType type, SNodeList** pCols);
 
 typedef bool (*FFuncClassifier)(int32_t funcId);
 int32_t nodesCollectFuncs(SSelectStmt* pSelect, ESqlClause clause, FFuncClassifier classifier, SNodeList** pFuncs);

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

@@ -274,15 +274,14 @@ static void mndGetDnodeData(SMnode *pMnode, SArray *pDnodeEps) {
 
     SDnodeEp dnodeEp = {0};
     dnodeEp.id = pDnode->id;
-    dnodeEp.isMnode = 0;
     dnodeEp.ep.port = pDnode->port;
     memcpy(dnodeEp.ep.fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
+    sdbRelease(pSdb, pDnode);
 
+    dnodeEp.isMnode = 0;
     if (mndIsMnode(pMnode, pDnode->id)) {
       dnodeEp.isMnode = 1;
     }
-
-    sdbRelease(pSdb, pDnode);
     taosArrayPush(pDnodeEps, &dnodeEp);
   }
 }

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

@@ -519,6 +519,8 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
     }
   }
 
+  syncNodeRelease(pSyncNode);
+
   if (code != 0) {
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
   }

source/dnode/mnode/sdb/src/sdb.c

@@ -131,7 +131,7 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
     hashType = TSDB_DATA_TYPE_BINARY;
   }
 
-  SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_NO_LOCK);
+  SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_ENTRY_LOCK);
   if (hash == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;

source/dnode/vnode/src/vnd/vnodeCfg.c

@@ -79,7 +79,7 @@ int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
     SJson *pNodeRetentions = tjsonCreateArray();
     tjsonAddItemToObject(pJson, "retentions", pNodeRetentions);
     for (int32_t i = 0; i < nRetention; ++i) {
-      SJson      *pNodeRetention = tjsonCreateObject();
+      SJson *           pNodeRetention = tjsonCreateObject();
       const SRetention *pRetention = pCfg->tsdbCfg.retentions + i;
       tjsonAddIntegerToObject(pNodeRetention, "freq", pRetention->freq);
       tjsonAddIntegerToObject(pNodeRetention, "freqUnit", pRetention->freqUnit);
@@ -118,45 +118,45 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
 
   int32_t code;
   tjsonGetNumberValue(pJson, "vgId", pCfg->vgId, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   if (tjsonGetStringValue(pJson, "dbname", pCfg->dbname) < 0) return -1;
   tjsonGetNumberValue(pJson, "dbId", pCfg->dbId, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "szPage", pCfg->szPage, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "szCache", pCfg->szCache, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "szBuf", pCfg->szBuf, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "isHeap", pCfg->isHeap, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "isWeak", pCfg->isWeak, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "isTsma", pCfg->isTsma, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "isRsma", pCfg->isRsma, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "precision", pCfg->tsdbCfg.precision, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "update", pCfg->tsdbCfg.update, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "compression", pCfg->tsdbCfg.compression, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "slLevel", pCfg->tsdbCfg.slLevel, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "daysPerFile", pCfg->tsdbCfg.days, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "minRows", pCfg->tsdbCfg.minRows, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "maxRows", pCfg->tsdbCfg.maxRows, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "keep0", pCfg->tsdbCfg.keep0, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "keep1", pCfg->tsdbCfg.keep1, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "keep2", pCfg->tsdbCfg.keep2, code);
-  if(code < 0) return -1;
-  SJson *pNodeRetentions = tjsonGetObjectItem(pJson, "retentions");
+  if (code < 0) return -1;
+  SJson * pNodeRetentions = tjsonGetObjectItem(pJson, "retentions");
   int32_t nRetention = tjsonGetArraySize(pNodeRetentions);
   if (nRetention > TSDB_RETENTION_MAX) {
     nRetention = TSDB_RETENTION_MAX;
@@ -170,30 +170,30 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
     tjsonGetNumberValue(pNodeRetention, "keepUnit", (pCfg->tsdbCfg.retentions)[i].keepUnit, code);
   }
   tjsonGetNumberValue(pJson, "wal.vgId", pCfg->walCfg.vgId, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.fsyncPeriod", pCfg->walCfg.fsyncPeriod, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.retentionPeriod", pCfg->walCfg.retentionPeriod, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.rollPeriod", pCfg->walCfg.rollPeriod, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.retentionSize", pCfg->walCfg.retentionSize, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.segSize", pCfg->walCfg.segSize, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "wal.level", pCfg->walCfg.level, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "hashBegin", pCfg->hashBegin, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "hashEnd", pCfg->hashEnd, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "hashMethod", pCfg->hashMethod, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
 
   tjsonGetNumberValue(pJson, "syncCfg.replicaNum", pCfg->syncCfg.replicaNum, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "syncCfg.myIndex", pCfg->syncCfg.myIndex, code);
-  if(code < 0) return -1;
+  if (code < 0) return -1;
 
   SJson *pNodeInfoArr = tjsonGetObjectItem(pJson, "syncCfg.nodeInfo");
   int    arraySize = tjsonGetArraySize(pNodeInfoArr);

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -27,10 +27,10 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   SMetaReader    mer1 = {0};
   SMetaReader    mer2 = {0};
   char           tableFName[TSDB_TABLE_FNAME_LEN];
-  SRpcMsg        rpcMsg;
+  SRpcMsg        rpcMsg = {0};
   int32_t        code = 0;
   int32_t        rspLen = 0;
-  void          *pRsp = NULL;
+  void *         pRsp = NULL;
   SSchemaWrapper schema = {0};
   SSchemaWrapper schemaTag = {0};
 
@@ -111,6 +111,7 @@ _exit:
   rpcMsg.pCont = pRsp;
   rpcMsg.contLen = rspLen;
   rpcMsg.code = code;
+  rpcMsg.msgType = pMsg->msgType;
 
   if (code) {
     qError("get table %s meta failed cause of %s", infoReq.tbName, tstrerror(code));
@@ -130,10 +131,10 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg) {
   SMetaReader    mer1 = {0};
   SMetaReader    mer2 = {0};
   char           tableFName[TSDB_TABLE_FNAME_LEN];
-  SRpcMsg        rpcMsg;
+  SRpcMsg        rpcMsg = {0};
   int32_t        code = 0;
   int32_t        rspLen = 0;
-  void          *pRsp = NULL;
+  void *         pRsp = NULL;
   SSchemaWrapper schema = {0};
   SSchemaWrapper schemaTag = {0};
 
@@ -220,6 +221,7 @@ _exit:
   rpcMsg.pCont = pRsp;
   rpcMsg.contLen = rspLen;
   rpcMsg.code = code;
+  rpcMsg.msgType = pMsg->msgType;
 
   if (code) {
     qError("get table %s cfg failed cause of %s", cfgReq.tbName, tstrerror(code));
@@ -260,7 +262,7 @@ void vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId) {
 }
 
 // wrapper of tsdb read interface
-tsdbReaderT tsdbQueryCacheLast(SVnode *pVnode, SQueryTableDataCond *pCond, STableListInfo* tableList, uint64_t qId,
+tsdbReaderT tsdbQueryCacheLast(SVnode *pVnode, SQueryTableDataCond *pCond, STableListInfo *tableList, uint64_t qId,
                                void *pMemRef) {
 #if 0
   return tsdbQueryCacheLastT(pVnode->pTsdb, pCond, groupList, qId, pMemRef);

source/libs/executor/inc/executorimpl.h

@@ -51,6 +51,13 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
 
 #define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0)
 
+#define START_TS_COLUMN_INDEX 0
+#define END_TS_COLUMN_INDEX 1
+#define UID_COLUMN_INDEX 2
+#define GROUPID_COLUMN_INDEX UID_COLUMN_INDEX
+#define DELETE_GROUPID_COLUMN_INDEX 2
+
+
 enum {
   // when this task starts to execute, this status will set
   TASK_NOT_COMPLETED = 0x1u,
@@ -364,6 +371,8 @@ typedef struct SStreamBlockScanInfo {
   int32_t         scanWinIndex;   // for state operator
   int32_t         pullDataResIndex;
   SSDataBlock*    pPullDataRes;             // pull data SSDataBlock
+  SSDataBlock*    pDeleteDataRes;             // delete data SSDataBlock
+  int32_t         deleteDataIndex;
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -429,6 +438,10 @@ typedef struct SIntervalAggOperatorInfo {
   bool               invertible;
   SArray*            pPrevValues;        //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
   bool               ignoreExpiredData;
+  SArray*            pRecycledPages;
+  SArray*            pDelWins;           // SWinRes
+  int32_t            delIndex;
+  SSDataBlock*       pDelRes;
 } SIntervalAggOperatorInfo;
 
 typedef struct SStreamFinalIntervalOperatorInfo {
@@ -451,6 +464,10 @@ typedef struct SStreamFinalIntervalOperatorInfo {
   int32_t            pullIndex;
   SSDataBlock*       pPullDataRes;
   bool               ignoreExpiredData;
+  SArray*            pRecycledPages;
+  SArray*            pDelWins;           // SWinRes
+  int32_t            delIndex;
+  SSDataBlock*       pDelRes;
 } SStreamFinalIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -680,7 +697,7 @@ typedef struct SJoinOperatorInfo {
   SSDataBlock       *pRight;
   int32_t            rightPos;
   SColumnInfo        rightCol;
-  SNode             *pOnCondition;
+  SNode             *pCondAfterMerge;
 } SJoinOperatorInfo;
 
 #define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)

source/libs/executor/src/joinoperator.c

@@ -53,13 +53,28 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
   pOperator->info         = pInfo;
   pOperator->pTaskInfo    = pTaskInfo;
 
-  SNode* pOnCondition = pJoinNode->pOnConditions;
-  if (nodeType(pOnCondition) == QUERY_NODE_OPERATOR) {
-    SOperatorNode* pNode = (SOperatorNode*)pOnCondition;
+  SNode* pMergeCondition = pJoinNode->pMergeCondition;
+  if (nodeType(pMergeCondition) == QUERY_NODE_OPERATOR) {
+    SOperatorNode* pNode = (SOperatorNode*)pMergeCondition;
     setJoinColumnInfo(&pInfo->leftCol, (SColumnNode*)pNode->pLeft);
     setJoinColumnInfo(&pInfo->rightCol, (SColumnNode*)pNode->pRight);
-  } else if (nodeType(pOnCondition) == QUERY_NODE_LOGIC_CONDITION) {
-    extractTimeCondition(pInfo, (SLogicConditionNode*)pOnCondition);
+  } else {
+    ASSERT(false);
+  }
+
+  if (pJoinNode->pOnConditions != NULL && pJoinNode->node.pConditions != NULL) {
+    pInfo->pCondAfterMerge = nodesMakeNode(QUERY_NODE_LOGIC_CONDITION);
+    SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(pInfo->pCondAfterMerge);
+    pLogicCond->pParameterList = nodesMakeList();
+    nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->pOnConditions));
+    nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->node.pConditions));
+    pLogicCond->condType = LOGIC_COND_TYPE_AND;
+  } else if (pJoinNode->pOnConditions != NULL) {
+    pInfo->pCondAfterMerge = nodesCloneNode(pJoinNode->pOnConditions);
+  } else if (pJoinNode->node.pConditions != NULL) {
+    pInfo->pCondAfterMerge = nodesCloneNode(pJoinNode->node.pConditions);
+  } else {
+    pInfo->pCondAfterMerge = NULL;
   }
 
   pOperator->fpSet =
@@ -88,15 +103,12 @@ void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) {
 
 void destroyMergeJoinOperator(void* param, int32_t numOfOutput) {
   SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param;
+  nodesDestroyNode(pJoinOperator->pCondAfterMerge);
 }
 
-SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) {
+static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) {
   SJoinOperatorInfo* pJoinInfo = pOperator->info;
 
-  SSDataBlock* pRes = pJoinInfo->pRes;
-  blockDataCleanup(pRes);
-  blockDataEnsureCapacity(pRes, 4096);
-
   int32_t nrows = 0;
 
   while (1) {
@@ -181,7 +193,28 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) {
       break;
     }
   }
+}
+
+SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) {
+  SJoinOperatorInfo* pJoinInfo = pOperator->info;
 
+  SSDataBlock* pRes = pJoinInfo->pRes;
+  blockDataCleanup(pRes);
+  blockDataEnsureCapacity(pRes, 4096);
+  while (true) {
+    int32_t numOfRowsBefore = pRes->info.rows;
+    doMergeJoinImpl(pOperator, pRes);
+    int32_t numOfNewRows = pRes->info.rows - numOfRowsBefore;
+    if (numOfNewRows == 0) {
+      break;
+    }
+    if (pJoinInfo->pCondAfterMerge != NULL) {
+      doFilter(pJoinInfo->pCondAfterMerge, pRes);
+    }
+    if (pRes->info.rows >= pOperator->resultInfo.threshold) {
+      break;
+    }
+  }
   return (pRes->info.rows > 0) ? pRes : NULL;
 }
 

source/libs/executor/src/scanoperator.c

@@ -807,18 +807,38 @@ static bool isStateWindow(SStreamBlockScanInfo* pInfo) {
   return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
 }
 
+static uint64_t getGroupId(SOperatorInfo* pOperator, uint64_t uid) {
+  uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &uid, sizeof(int64_t));
+  if (groupId) {
+    return *groupId;
+  }
+  return 0;
+  /* Todo(liuyao) for partition by column
+  recordNewGroupKeys(pTableScanInfo->pGroupCols, pTableScanInfo->pGroupColVals, pBlock, rowId);
+  int32_t len = buildGroupKeys(pTableScanInfo->keyBuf, pTableScanInfo->pGroupColVals);
+  uint64_t resId = 0;
+  uint64_t* groupId = taosHashGet(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len);
+  if (groupId) {
+    return *groupId;
+  } else if (len != 0) {
+    resId = calcGroupId(pTableScanInfo->keyBuf, len);
+    taosHashPut(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len, &resId, sizeof(uint64_t));
+  }
+  return resId;
+  */
+}
+
 static void setGroupId(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, int32_t groupColIndex, int32_t rowIndex) {
   ASSERT(rowIndex < pBlock->info.rows);
   switch (pBlock->info.type)
   {
+  case STREAM_DELETE_DATA:
   case STREAM_RETRIEVE: {
     SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex);
     uint64_t* groupCol = (uint64_t*)pColInfo->pData;
     pInfo->groupId = groupCol[rowIndex];
   }
     break;
-  case STREAM_DELETE_DATA:
-    break;
   default:
     break;
   }
@@ -840,14 +860,14 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int3
       int64_t              gap = pInfo->sessionSup.gap;
       int32_t              winIndex = 0;
       SResultWindowInfo*   pCurWin =
-          getSessionTimeWindow(pAggSup, tsCols[(*pRowIndex)], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
+          getSessionTimeWindow(pAggSup, tsCols[*pRowIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
       win = pCurWin->win;
-      (*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, (*pRowIndex), gap, NULL);
+      (*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, *pRowIndex, gap, NULL);
     } else {
       win =
-          getActiveTimeWindow(NULL, &dumyInfo, tsCols[(*pRowIndex)], &pInfo->interval, pInfo->interval.precision, NULL);
-      setGroupId(pInfo, pSDB, 2, *pRowIndex);
-      (*pRowIndex) += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, (*pRowIndex), win.ekey, binarySearchForKey, NULL,
+          getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, pInfo->interval.precision, NULL);
+      setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex);
+      (*pRowIndex) += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, *pRowIndex, win.ekey, binarySearchForKey, NULL,
                                                TSDB_ORDER_ASC);
     }
     needRead = true;
@@ -891,27 +911,6 @@ static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRow
   dest->info.rows++;
 }
 
-static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t rowId) {
-  uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
-  if (groupId) {
-    return *groupId;
-  }
-  return 0;
-  /* Todo(liuyao) for partition by column
-  recordNewGroupKeys(pTableScanInfo->pGroupCols, pTableScanInfo->pGroupColVals, pBlock, rowId);
-  int32_t len = buildGroupKeys(pTableScanInfo->keyBuf, pTableScanInfo->pGroupColVals);
-  uint64_t resId = 0;
-  uint64_t* groupId = taosHashGet(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len);
-  if (groupId) {
-    return *groupId;
-  } else if (len != 0) {
-    resId = calcGroupId(pTableScanInfo->keyBuf, len);
-    taosHashPut(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len, &resId, sizeof(uint64_t));
-  }
-  return resId;
-  */
-}
-
 static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
   while (1) {
     SSDataBlock* pResult = NULL;
@@ -935,7 +934,7 @@ static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, i
     SSDataBlock* pBlock = createOneDataBlock(pResult, true);
     blockDataCleanup(pResult);
     for (int32_t i = 0; i < pBlock->info.rows; i++) {
-      uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, i);
+      uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock->info.uid);
       if (id == pInfo->groupId) {
         copyOneRow(pResult, pBlock, i);
       }
@@ -944,6 +943,40 @@ static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, i
   */
 }
 
+static void copyDeleteDataBlock(SStreamBlockScanInfo* pInfo, SSDataBlock* pDelBlock, SOperatorInfo* pOperator, SSDataBlock* pUpdateRes) {
+  if (pDelBlock->info.rows == 0) {
+    return;
+  }
+  blockDataCleanup(pUpdateRes);
+  blockDataEnsureCapacity(pUpdateRes, 64);
+  ASSERT(taosArrayGetSize(pDelBlock->pDataBlock) >= 3);
+  SColumnInfoData* pStartTsCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX);
+  TSKEY* startData = (TSKEY*)pStartTsCol->pData;
+  SColumnInfoData* pEndTsCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX);
+  TSKEY* endData = (TSKEY*)pEndTsCol->pData;
+  SColumnInfoData* pGpCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX);
+  uint64_t* uidCol = (uint64_t*)pGpCol->pData;
+
+  SColumnInfoData* pDestTsCol = taosArrayGet(pUpdateRes->pDataBlock, START_TS_COLUMN_INDEX);
+  SColumnInfoData* pDestGpCol = taosArrayGet(pUpdateRes->pDataBlock, DELETE_GROUPID_COLUMN_INDEX);
+  for (int32_t i = pInfo->deleteDataIndex ; i < pDelBlock->info.rows &&
+      i < pDelBlock->info.capacity - (endData[i] - startData[i])/pInfo->interval.interval - 1; i++) {
+    uint64_t groupId = getGroupId(pOperator, uidCol[i]);
+    for (TSKEY startTs = startData[i]; startTs <= endData[i]; ) {
+      colDataAppend(pDestTsCol, pUpdateRes->info.rows, (const char*)&startTs, false);
+      colDataAppend(pDestGpCol, pUpdateRes->info.rows, (const char*)&groupId, false);
+      pUpdateRes->info.rows++;
+      startTs = taosTimeAdd(startTs, pInfo->interval.interval, pInfo->interval.intervalUnit, pInfo->interval.precision);
+    }
+    pInfo->deleteDataIndex++;
+  }
+
+  if (pInfo->deleteDataIndex > 0 && pInfo->deleteDataIndex == pDelBlock->info.rows) {
+    blockDataCleanup(pDelBlock);
+    pInfo->deleteDataIndex = 0;
+  }
+}
+
 static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
   blockDataCleanup(pUpdateBlock);
   int32_t size = taosArrayGetSize(pInfo->tsArray);
@@ -953,11 +986,11 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
     blockDataEnsureCapacity(pUpdateBlock, size);
 
     int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
-    pInfo->groupId = getGroupId(pInfo->pSnapshotReadOp, pBlock, rowId);
+    pInfo->groupId = getGroupId(pInfo->pSnapshotReadOp, pBlock->info.uid);
     int32_t i = 0;
     for (; i < size; i++) {
       rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
-      uint64_t id = getGroupId(pInfo->pSnapshotReadOp, pBlock, rowId);
+      uint64_t id = getGroupId(pInfo->pSnapshotReadOp, pBlock->info.uid);
       if (pInfo->groupId != id) {
         break;
       }
@@ -974,28 +1007,32 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
   if (size > 0 && pInfo->tsArrayIndex == size) {
     taosArrayClear(pInfo->tsArray);
   }
+
+  if (size == 0) {
+    copyDeleteDataBlock(pInfo, pInfo->pDeleteDataRes, pInfo->pSnapshotReadOp, pUpdateBlock);
+  }
 }
 
-static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock,
-                               SSDataBlock* pUpdateBlock) {
+static void checkUpdateData(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock,
+    bool out) {
   SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
   ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
   TSKEY* ts = (TSKEY*)pColDataInfo->pData;
   for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
-    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[rowId])) {
+    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[rowId]) && out) {
       taosArrayPush(pInfo->tsArray, &rowId);
     }
   }
-  if (!pUpdateBlock) {
-    taosArrayClear(pInfo->tsArray);
-    return;
+}
+
+static void setBlockGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t uidColIndex) {
+  ASSERT(taosArrayGetSize(pBlock->pDataBlock) >= 3);
+  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, uidColIndex);
+  uint64_t*        uidCol = (uint64_t*)pColDataInfo->pData;
+  ASSERT(pBlock->info.rows > 0);
+  for (int32_t i = 0 ; i < pBlock->info.rows; i++) {
+    uidCol[i] = getGroupId(pOperator, uidCol[i]);
   }
-  setUpdateData(pInfo, pBlock, pUpdateBlock);
-  // Todo(liuyao) get from tsdb
-  //  SSDataBlock* p = createOneDataBlock(pBlock, true);
-  //  p->info.type = STREAM_INVERT;
-  //  taosArrayClear(pInfo->tsArray);
-  //  return p;
 }
 
 static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
@@ -1020,13 +1057,29 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     int32_t      current = pInfo->validBlockIndex++;
     SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
     blockDataUpdateTsWindow(pBlock, 0);
-    if (pBlock->info.type == STREAM_RETRIEVE) {
+    switch (pBlock->info.type) {
+    case STREAM_RETRIEVE:{
       pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
       pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
       copyDataBlock(pInfo->pPullDataRes, pBlock);
       pInfo->pullDataResIndex = 0;
-      prepareDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
+      prepareDataScan(pInfo, pInfo->pPullDataRes, START_TS_COLUMN_INDEX, &pInfo->pullDataResIndex);
       updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
+    }
+      break;
+    case STREAM_DELETE_DATA: {
+      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
+      copyDataBlock(pInfo->pDeleteDataRes, pBlock);
+      copyDeleteDataBlock(pInfo, pInfo->pDeleteDataRes, pInfo->pSnapshotReadOp, pInfo->pUpdateRes);
+      pInfo->updateResIndex = 0;
+      prepareDataScan(pInfo, pInfo->pUpdateRes, START_TS_COLUMN_INDEX, &pInfo->updateResIndex);
+      pInfo->pUpdateRes->info.type = STREAM_DELETE_DATA;
+      return pInfo->pUpdateRes;
+    }
+      break;
+    default:
+      break;
     }
     return pBlock;
   } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
@@ -1043,39 +1096,33 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) {
       SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
       if (pSDB != NULL) {
-        getUpdateDataBlock(pInfo, true, pSDB, NULL);
+        checkUpdateData(pInfo, true, pSDB, false);
         pSDB->info.type = STREAM_PULL_DATA;
         return pSDB;
       }
       pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
-    } else {
-      if (isStateWindow(pInfo)) {
-        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
-        pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
-        if (!prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex)) {
-          pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
-        }
+    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
+      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
+      if (pSDB) {
+        pSDB->info.type = STREAM_NORMAL;
+        checkUpdateData(pInfo, true, pSDB, false);
+        return pSDB;
       }
-      if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
-        SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
-        if (pSDB == NULL) {
-          setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
-          if (pInfo->pUpdateRes->info.rows > 0) {
-            if (!isStateWindow(pInfo)) {
-              // Todo(liuyao) mybe can delete this.
-              bool test = prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
-              ASSERT(test == false);
-            }
-            return pInfo->pUpdateRes;
-          } else {
-            pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
-          }
-        } else {
-          pSDB->info.type = STREAM_NORMAL;
-          getUpdateDataBlock(pInfo, true, pSDB, NULL);
-          return pSDB;
-        }
+      setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
+      if (pInfo->pUpdateRes->info.rows > 0) {
+        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
+        return pInfo->pUpdateRes;
       }
+      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+    } else if (isStateWindow(pInfo)) {
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
+      pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
+      if (prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex)) {
+        ASSERT(pInfo->pUpdateRes->info.rows == 0);
+        // return empty data blcok
+        return pInfo->pUpdateRes;
+      }
+      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
     }
 
     SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
@@ -1169,7 +1216,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       pOperator->status = OP_EXEC_DONE;
     } else if (pInfo->pUpdateInfo) {
       pInfo->tsArrayIndex = 0;
-      getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
+      checkUpdateData(pInfo, true, pInfo->pRes, true);
+      setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
       if (pInfo->pUpdateRes->info.rows > 0) {
         if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) {
           pInfo->updateResIndex = 0;
@@ -1180,9 +1228,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
         }
       }
     }
-
     return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
-
   } else if (pInfo->blockType == STREAM_INPUT__DATA_SCAN) {
     // check reader last status
     // if not match, reset status
@@ -1295,6 +1341,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   pInfo->groupId = 0;
   pInfo->pPullDataRes = createPullDataBlock();
   pInfo->pStreamScanOp = pOperator;
+  pInfo->deleteDataIndex = 0;
+  pInfo->pDeleteDataRes = createPullDataBlock();
 
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;

source/libs/function/src/builtins.c

@@ -15,10 +15,10 @@
 
 #include "builtins.h"
 #include "builtinsimpl.h"
+#include "cJSON.h"
 #include "querynodes.h"
 #include "scalar.h"
 #include "taoserror.h"
-#include "cJSON.h"
 
 static int32_t buildFuncErrMsg(char* pErrBuf, int32_t len, int32_t errCode, const char* pFormat, ...) {
   va_list vArgList;
@@ -40,7 +40,7 @@ static int32_t invaildFuncParaValueErrMsg(char* pErrBuf, int32_t len, const char
   return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_PARA_VALUE, "Invalid parameter value : %s", pFuncName);
 }
 
-#define TIME_UNIT_INVALID 1
+#define TIME_UNIT_INVALID   1
 #define TIME_UNIT_TOO_SMALL 2
 
 static int32_t validateTimeUnitParam(uint8_t dbPrec, const SValueNode* pVal) {
@@ -701,7 +701,7 @@ static int32_t translateElapsed(SFunctionNode* pFunc, char* pErrBuf, int32_t len
 
     uint8_t dbPrec = pFunc->node.resType.precision;
 
-    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 1));
+    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1));
     if (ret == TIME_UNIT_TOO_SMALL) {
       return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
                              "ELAPSED function time unit parameter should be greater than db precision");
@@ -820,13 +820,13 @@ static int8_t validateHistogramBinType(char* binTypeStr) {
 }
 
 static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* errMsg, int32_t msgLen) {
-  const char *msg1 = "HISTOGRAM function requires four parameters";
-  const char *msg3 = "HISTOGRAM function invalid format for binDesc parameter";
-  const char *msg4 = "HISTOGRAM function binDesc parameter \"count\" should be in range [1, 1000]";
-  const char *msg5 = "HISTOGRAM function bin/parameter should be in range [-DBL_MAX, DBL_MAX]";
-  const char *msg6 = "HISTOGRAM function binDesc parameter \"width\" cannot be 0";
-  const char *msg7 = "HISTOGRAM function binDesc parameter \"start\" cannot be 0 with \"log_bin\" type";
-  const char *msg8 = "HISTOGRAM function binDesc parameter \"factor\" cannot be negative or equal to 0/1";
+  const char* msg1 = "HISTOGRAM function requires four parameters";
+  const char* msg3 = "HISTOGRAM function invalid format for binDesc parameter";
+  const char* msg4 = "HISTOGRAM function binDesc parameter \"count\" should be in range [1, 1000]";
+  const char* msg5 = "HISTOGRAM function bin/parameter should be in range [-DBL_MAX, DBL_MAX]";
+  const char* msg6 = "HISTOGRAM function binDesc parameter \"width\" cannot be 0";
+  const char* msg7 = "HISTOGRAM function binDesc parameter \"start\" cannot be 0 with \"log_bin\" type";
+  const char* msg8 = "HISTOGRAM function binDesc parameter \"factor\" cannot be negative or equal to 0/1";
 
   cJSON*  binDesc = cJSON_Parse(binDescStr);
   int32_t numOfBins;
@@ -1009,8 +1009,8 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l
     }
 
     if (i == 3 && pValue->datum.i != 1 && pValue->datum.i != 0) {
-        return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
-                               "HISTOGRAM function normalized parameter should be 0/1");
+      return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
+                             "HISTOGRAM function normalized parameter should be 0/1");
     }
   }
 
@@ -1067,8 +1067,8 @@ static int32_t translateHistogramImpl(SFunctionNode* pFunc, char* pErrBuf, int32
       }
 
       if (i == 3 && pValue->datum.i != 1 && pValue->datum.i != 0) {
-          return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
-                                 "HISTOGRAM function normalized parameter should be 0/1");
+        return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
+                               "HISTOGRAM function normalized parameter should be 0/1");
       }
     }
 
@@ -1223,7 +1223,7 @@ static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32
   if (numOfParams == 4) {
     uint8_t dbPrec = pFunc->node.resType.precision;
 
-    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 3));
+    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode*)nodesListGetNode(pFunc->pParameterList, 3));
     if (ret == TIME_UNIT_TOO_SMALL) {
       return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
                              "STATEDURATION function time unit parameter should be greater than db precision");
@@ -1437,10 +1437,6 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
 
 static int32_t translateFirstLastImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) {
   // first(col_list) will be rewritten as first(col)
-  if (2 != LIST_LENGTH(pFunc->pParameterList)) {  // input has two params c0,ts, is this a bug?
-    return TSDB_CODE_SUCCESS;
-  }
-
   SNode*  pPara = nodesListGetNode(pFunc->pParameterList, 0);
   uint8_t paraType = ((SExprNode*)pPara)->resType.type;
   int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
@@ -1738,7 +1734,7 @@ static int32_t translateTimeTruncate(SFunctionNode* pFunc, char* pErrBuf, int32_
   // add database precision as param
   uint8_t dbPrec = pFunc->node.resType.precision;
 
-  int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 1));
+  int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1));
   if (ret == TIME_UNIT_TOO_SMALL) {
     return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
                            "TIMETRUNCATE function time unit parameter should be greater than db precision");
@@ -1777,7 +1773,7 @@ static int32_t translateTimeDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t le
   uint8_t dbPrec = pFunc->node.resType.precision;
 
   if (3 == numOfParams) {
-    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 2));
+    int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode*)nodesListGetNode(pFunc->pParameterList, 2));
     if (ret == TIME_UNIT_TOO_SMALL) {
       return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
                              "TIMEDIFF function time unit parameter should be greater than db precision");

source/libs/function/src/tudf.c

@@ -1565,6 +1565,10 @@ void constructUdfService(void *argsThread) {
   //TODO return value of uv_run
   uv_run(&udfc->uvLoop, UV_RUN_DEFAULT);
   uv_loop_close(&udfc->uvLoop);
+
+  uv_walk(&udfc->uvLoop, udfUdfdCloseWalkCb, NULL);
+  uv_run(&udfc->uvLoop, UV_RUN_DEFAULT);
+  uv_loop_close(&udfc->uvLoop);
 }
 
 int32_t udfcOpen() {

source/libs/nodes/src/nodesCloneFuncs.c

@@ -368,6 +368,7 @@ static int32_t logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) {
 static int32_t logicJoinCopy(const SJoinLogicNode* pSrc, SJoinLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   COPY_SCALAR_FIELD(joinType);
+  CLONE_NODE_FIELD(pMergeCondition);
   CLONE_NODE_FIELD(pOnConditions);
   COPY_SCALAR_FIELD(isSingleTableJoin);
   return TSDB_CODE_SUCCESS;

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1254,6 +1254,7 @@ static int32_t jsonToLogicPlan(const SJson* pJson, void* pObj) {
 
 static const char* jkJoinLogicPlanJoinType = "JoinType";
 static const char* jkJoinLogicPlanOnConditions = "OnConditions";
+static const char* jkJoinLogicPlanMergeCondition = "MergeConditions";
 
 static int32_t logicJoinNodeToJson(const void* pObj, SJson* pJson) {
   const SJoinLogicNode* pNode = (const SJoinLogicNode*)pObj;
@@ -1262,6 +1263,9 @@ static int32_t logicJoinNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddIntegerToObject(pJson, jkJoinLogicPlanJoinType, pNode->joinType);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkJoinLogicPlanMergeCondition, nodeToJson, pNode->pMergeCondition);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkJoinLogicPlanOnConditions, nodeToJson, pNode->pOnConditions);
   }
@@ -1617,6 +1621,7 @@ static int32_t jsonToPhysiProjectNode(const SJson* pJson, void* pObj) {
 }
 
 static const char* jkJoinPhysiPlanJoinType = "JoinType";
+static const char* jkJoinPhysiPlanMergeCondition = "MergeCondition";
 static const char* jkJoinPhysiPlanOnConditions = "OnConditions";
 static const char* jkJoinPhysiPlanTargets = "Targets";
 
@@ -1627,6 +1632,9 @@ static int32_t physiJoinNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanJoinType, pNode->joinType);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkJoinPhysiPlanMergeCondition, nodeToJson, pNode->pMergeCondition);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkJoinPhysiPlanOnConditions, nodeToJson, pNode->pOnConditions);
   }
@@ -1648,6 +1656,9 @@ static int32_t jsonToPhysiJoinNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkJoinPhysiPlanOnConditions, &pNode->pOnConditions);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkJoinPhysiPlanMergeCondition, &pNode->pMergeCondition);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeList(pJson, jkJoinPhysiPlanTargets, &pNode->pTargets);
   }

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -470,6 +470,9 @@ static EDealRes dispatchPhysiPlan(SNode* pNode, ETraversalOrder order, FNodeWalk
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN: {
       SJoinPhysiNode* pJoin = (SJoinPhysiNode*)pNode;
       res = walkPhysiNode((SPhysiNode*)pNode, order, walker, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = walkPhysiPlan(pJoin->pMergeCondition, order, walker, pContext);
+      }
       if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
         res = walkPhysiPlan(pJoin->pOnConditions, order, walker, pContext);
       }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -718,6 +718,7 @@ void nodesDestroyNode(SNode* pNode) {
     case QUERY_NODE_LOGIC_PLAN_JOIN: {
       SJoinLogicNode* pLogicNode = (SJoinLogicNode*)pNode;
       destroyLogicNode((SLogicNode*)pLogicNode);
+      nodesDestroyNode(pLogicNode->pMergeCondition);
       nodesDestroyNode(pLogicNode->pOnConditions);
       break;
     }
@@ -828,6 +829,7 @@ void nodesDestroyNode(SNode* pNode) {
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN: {
       SJoinPhysiNode* pPhyNode = (SJoinPhysiNode*)pNode;
       destroyPhysiNode((SPhysiNode*)pPhyNode);
+      nodesDestroyNode(pPhyNode->pMergeCondition);
       nodesDestroyNode(pPhyNode->pOnConditions);
       nodesDestroyList(pPhyNode->pTargets);
       break;
@@ -1493,6 +1495,38 @@ int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char*
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t nodesCollectColumnsFromNode(SNode* node, const char* pTableAlias, ECollectColType type, SNodeList** pCols) {
+  if (NULL == pCols) {
+    return TSDB_CODE_FAILED;
+  }
+  SCollectColumnsCxt cxt = {
+      .errCode = TSDB_CODE_SUCCESS,
+      .pTableAlias = pTableAlias,
+      .collectType = type,
+      .pCols = (NULL == *pCols ? nodesMakeList() : *pCols),
+      .pColHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK)};
+  if (NULL == cxt.pCols || NULL == cxt.pColHash) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  *pCols = NULL;
+
+  nodesWalkExpr(node, collectColumns, &cxt);
+
+  taosHashCleanup(cxt.pColHash);
+  if (TSDB_CODE_SUCCESS != cxt.errCode) {
+    nodesDestroyList(cxt.pCols);
+    return cxt.errCode;
+  }
+  if (LIST_LENGTH(cxt.pCols) > 0) {
+    *pCols = cxt.pCols;
+  } else {
+    nodesDestroyList(cxt.pCols);
+  }
+
+  return TSDB_CODE_SUCCESS;
+
+}
+
 typedef struct SCollectFuncsCxt {
   int32_t         errCode;
   FFuncClassifier classifier;

source/libs/parser/src/parAstCreater.c

@@ -385,6 +385,15 @@ SNode* createLogicConditionNode(SAstCreateContext* pCxt, ELogicConditionType typ
 
 SNode* createOperatorNode(SAstCreateContext* pCxt, EOperatorType type, SNode* pLeft, SNode* pRight) {
   CHECK_PARSER_STATUS(pCxt);
+  if (OP_TYPE_MINUS == type && QUERY_NODE_VALUE == nodeType(pLeft)) {
+    SValueNode* pVal = (SValueNode*)pLeft;
+    char*       pNewLiteral = taosMemoryCalloc(1, strlen(pVal->literal) + 2);
+    CHECK_OUT_OF_MEM(pNewLiteral);
+    sprintf(pNewLiteral, "-%s", pVal->literal);
+    taosMemoryFree(pVal->literal);
+    pVal->literal = pNewLiteral;
+    return pLeft;
+  }
   SOperatorNode* op = (SOperatorNode*)nodesMakeNode(QUERY_NODE_OPERATOR);
   CHECK_OUT_OF_MEM(op);
   op->opType = type;

source/libs/parser/src/parTranslater.c

@@ -558,11 +558,11 @@ static void setColumnInfoByExpr(STempTableNode* pTable, SExprNode* pExpr, SColum
   pCol->node.resType = pExpr->resType;
 }
 
-static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* pTable, SNodeList* pList) {
+static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* pTable, bool igTags, SNodeList* pList) {
   if (QUERY_NODE_REAL_TABLE == nodeType(pTable)) {
     const STableMeta* pMeta = ((SRealTableNode*)pTable)->pMeta;
-    int32_t           nums =
-        pMeta->tableInfo.numOfColumns + ((TSDB_SUPER_TABLE == pMeta->tableType) ? pMeta->tableInfo.numOfTags : 0);
+    int32_t           nums = pMeta->tableInfo.numOfColumns +
+                   (igTags ? 0 : ((TSDB_SUPER_TABLE == pMeta->tableType) ? pMeta->tableInfo.numOfTags : 0));
     for (int32_t i = 0; i < nums; ++i) {
       SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
       if (NULL == pCol) {
@@ -878,6 +878,9 @@ static EDealRes translateNormalValue(STranslateContext* pCxt, SValueNode* pVal,
     }
     case TSDB_DATA_TYPE_VARCHAR:
     case TSDB_DATA_TYPE_VARBINARY: {
+      if (strict && (pVal->node.resType.bytes > targetDt.bytes - VARSTR_HEADER_SIZE)) {
+        return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pVal->literal);
+      }
       pVal->datum.p = taosMemoryCalloc(1, targetDt.bytes + 1);
       if (NULL == pVal->datum.p) {
         return generateDealNodeErrMsg(pCxt, TSDB_CODE_OUT_OF_MEMORY);
@@ -1333,6 +1336,9 @@ static int32_t rewriteSystemInfoFuncImpl(STranslateContext* pCxt, char* pLiteral
     pVal->isNull = true;
   } else {
     pVal->literal = pLiteral;
+    if (IS_VAR_DATA_TYPE(pVal->node.resType.type)) {
+      pVal->node.resType.bytes = strlen(pLiteral);
+    }
   }
   if (DEAL_RES_ERROR != translateValue(pCxt, pVal)) {
     *pNode = (SNode*)pVal;
@@ -1928,7 +1934,7 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
   return code;
 }
 
-static int32_t createAllColumns(STranslateContext* pCxt, SNodeList** pCols) {
+static int32_t createAllColumns(STranslateContext* pCxt, bool igTags, SNodeList** pCols) {
   *pCols = nodesMakeList();
   if (NULL == *pCols) {
     return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_OUT_OF_MEMORY);
@@ -1937,7 +1943,7 @@ static int32_t createAllColumns(STranslateContext* pCxt, SNodeList** pCols) {
   size_t  nums = taosArrayGetSize(pTables);
   for (size_t i = 0; i < nums; ++i) {
     STableNode* pTable = taosArrayGetP(pTables, i);
-    int32_t     code = createColumnsByTable(pCxt, pTable, *pCols);
+    int32_t     code = createColumnsByTable(pCxt, pTable, igTags, *pCols);
     if (TSDB_CODE_SUCCESS != code) {
       return code;
     }
@@ -1974,7 +1980,7 @@ static SNode* createMultiResFunc(SFunctionNode* pSrcFunc, SExprNode* pExpr) {
   return (SNode*)pFunc;
 }
 
-static int32_t createTableAllCols(STranslateContext* pCxt, SColumnNode* pCol, SNodeList** pOutput) {
+static int32_t createTableAllCols(STranslateContext* pCxt, SColumnNode* pCol, bool igTags, SNodeList** pOutput) {
   STableNode* pTable = NULL;
   int32_t     code = findTable(pCxt, pCol->tableAlias, &pTable);
   if (TSDB_CODE_SUCCESS == code && NULL == *pOutput) {
@@ -1984,7 +1990,7 @@ static int32_t createTableAllCols(STranslateContext* pCxt, SColumnNode* pCol, SN
     }
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = createColumnsByTable(pCxt, pTable, *pOutput);
+    code = createColumnsByTable(pCxt, pTable, igTags, *pOutput);
   }
   return code;
 }
@@ -2006,11 +2012,9 @@ static int32_t createMultiResFuncsParas(STranslateContext* pCxt, SNodeList* pSrc
   SNode*     pPara = NULL;
   FOREACH(pPara, pSrcParas) {
     if (isStar(pPara)) {
-      code = createAllColumns(pCxt, &pExprs);
-      // The syntax definition ensures that * and other parameters do not appear at the same time
-      break;
+      code = createAllColumns(pCxt, true, &pExprs);
     } else if (isTableStar(pPara)) {
-      code = createTableAllCols(pCxt, (SColumnNode*)pPara, &pExprs);
+      code = createTableAllCols(pCxt, (SColumnNode*)pPara, true, &pExprs);
     } else {
       code = nodesListMakeStrictAppend(&pExprs, nodesCloneNode(pPara));
     }
@@ -2069,7 +2073,7 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
     int32_t code = TSDB_CODE_SUCCESS;
     if (isStar(pNode)) {
       SNodeList* pCols = NULL;
-      code = createAllColumns(pCxt, &pCols);
+      code = createAllColumns(pCxt, false, &pCols);
       if (TSDB_CODE_SUCCESS == code) {
         INSERT_LIST(pSelect->pProjectionList, pCols);
         ERASE_NODE(pSelect->pProjectionList);
@@ -2085,7 +2089,7 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
       }
     } else if (isTableStar(pNode)) {
       SNodeList* pCols = NULL;
-      code = createTableAllCols(pCxt, (SColumnNode*)pNode, &pCols);
+      code = createTableAllCols(pCxt, (SColumnNode*)pNode, false, &pCols);
       if (TSDB_CODE_SUCCESS == code) {
         INSERT_LIST(pSelect->pProjectionList, pCols);
         ERASE_NODE(pSelect->pProjectionList);
@@ -5826,10 +5830,6 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
     return pCxt->errCode;
   }
 
-  if (IS_VAR_DATA_TYPE(pSchema->type) && strlen(pStmt->pVal->literal) > pSchema->bytes) {
-    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pStmt->pVal->literal);
-  }
-
   pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
   if (targetDt.type == TSDB_DATA_TYPE_JSON) {
     pReq->isNull = 0;

source/libs/parser/src/parUtil.c

@@ -236,7 +236,6 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
 
   const char* prefix = "syntax error";
   if (sourceStr == NULL) {
-    assert(additionalInfo != NULL);
     snprintf(pBuf->buf, pBuf->len, msgFormat1, additionalInfo);
     return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
   }
@@ -254,40 +253,25 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
   return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
 }
 
-SSchema* getTableColumnSchema(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
-  return (SSchema*)pTableMeta->schema;
-}
+SSchema* getTableColumnSchema(const STableMeta* pTableMeta) { return (SSchema*)pTableMeta->schema; }
 
 static SSchema* getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex) {
-  assert(pTableMeta != NULL && pTableMeta->schema != NULL && colIndex >= 0 &&
-         colIndex < (getNumOfColumns(pTableMeta) + getNumOfTags(pTableMeta)));
-
   SSchema* pSchema = (SSchema*)pTableMeta->schema;
   return &pSchema[colIndex];
 }
 
 SSchema* getTableTagSchema(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL &&
-         (pTableMeta->tableType == TSDB_SUPER_TABLE || pTableMeta->tableType == TSDB_CHILD_TABLE));
   return getOneColumnSchema(pTableMeta, getTableInfo(pTableMeta).numOfColumns);
 }
 
 int32_t getNumOfColumns(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
   // table created according to super table, use data from super table
   return getTableInfo(pTableMeta).numOfColumns;
 }
 
-int32_t getNumOfTags(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
-  return getTableInfo(pTableMeta).numOfTags;
-}
+int32_t getNumOfTags(const STableMeta* pTableMeta) { return getTableInfo(pTableMeta).numOfTags; }
 
-STableComInfo getTableInfo(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
-  return pTableMeta->tableInfo;
-}
+STableComInfo getTableInfo(const STableMeta* pTableMeta) { return pTableMeta->tableInfo; }
 
 STableMeta* tableMetaDup(const STableMeta* pTableMeta) {
   size_t size = TABLE_META_SIZE(pTableMeta);

source/libs/planner/src/planOptimizer.c

@@ -75,6 +75,11 @@ static SLogicNode* optFindPossibleNode(SLogicNode* pNode, FMayBeOptimized func)
   return NULL;
 }
 
+static void optResetParent(SLogicNode* pNode) {
+  SNode* pChild = NULL;
+  FOREACH(pChild, pNode->pChildren) { ((SLogicNode*)pChild)->pParent = pNode; }
+}
+
 EDealRes scanPathOptHaveNormalColImpl(SNode* pNode, void* pContext) {
   if (QUERY_NODE_COLUMN == nodeType(pNode)) {
     // *((bool*)pContext) = (COLUMN_TYPE_TAG != ((SColumnNode*)pNode)->colType);
@@ -480,12 +485,18 @@ static int32_t pushDownCondOptPushCondToProject(SOptimizeContext* pCxt, SProject
   return pushDownCondOptAppendCond(&pProject->node.pConditions, pCond);
 }
 
+static int32_t pushDownCondOptPushCondToJoin(SOptimizeContext* pCxt, SJoinLogicNode * pJoin, SNode** pCond) {
+  return pushDownCondOptAppendCond(&pJoin->node.pConditions, pCond);
+}
+
 static int32_t pushDownCondOptPushCondToChild(SOptimizeContext* pCxt, SLogicNode* pChild, SNode** pCond) {
   switch (nodeType(pChild)) {
     case QUERY_NODE_LOGIC_PLAN_SCAN:
       return pushDownCondOptPushCondToScan(pCxt, (SScanLogicNode*)pChild, pCond);
     case QUERY_NODE_LOGIC_PLAN_PROJECT:
       return pushDownCondOptPushCondToProject(pCxt, (SProjectLogicNode*)pChild, pCond);
+    case QUERY_NODE_LOGIC_PLAN_JOIN:
+      return pushDownCondOptPushCondToJoin(pCxt, (SJoinLogicNode*)pChild, pCond);
     default:
       break;
   }
@@ -554,13 +565,83 @@ static int32_t pushDownCondOptCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogic
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t pushDownCondOptPartJoinOnCondLogicCond(SJoinLogicNode* pJoin, SNode** ppMergeCond, SNode** ppOnCond) {
+  SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(pJoin->pOnConditions);
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  SNodeList* pOnConds = NULL;
+  SNode* pCond = NULL;
+  FOREACH(pCond, pLogicCond->pParameterList) {
+    if (pushDownCondOptIsPriKeyEqualCond(pJoin, pCond)) {
+      *ppMergeCond = nodesCloneNode(pCond);
+    } else {
+      code = nodesListMakeAppend(&pOnConds, nodesCloneNode(pCond));
+    }
+  }
+
+  SNode* pTempOnCond = NULL;
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesMergeConds(&pTempOnCond, &pOnConds);
+  }
+
+  if (TSDB_CODE_SUCCESS == code && NULL != *ppMergeCond) {
+    *ppOnCond = pTempOnCond;
+    nodesDestroyNode(pJoin->pOnConditions);
+    pJoin->pOnConditions = NULL;
+    return TSDB_CODE_SUCCESS;
+  } else {
+    nodesDestroyList(pOnConds);
+    nodesDestroyNode(pTempOnCond);
+    return TSDB_CODE_PLAN_INTERNAL_ERROR;
+  }
+}
+
+static int32_t pushDownCondOptPartJoinOnCond(SJoinLogicNode* pJoin, SNode** ppMergeCond, SNode** ppOnCond) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pJoin->pOnConditions) &&
+      LOGIC_COND_TYPE_AND == ((SLogicConditionNode*)(pJoin->pOnConditions))->condType) {
+    return pushDownCondOptPartJoinOnCondLogicCond(pJoin, ppMergeCond, ppOnCond);
+  }
+
+  if (pushDownCondOptIsPriKeyEqualCond(pJoin, pJoin->pOnConditions)) {
+    *ppMergeCond = nodesCloneNode(pJoin->pOnConditions);
+    *ppOnCond = NULL;
+    nodesDestroyNode(pJoin->pOnConditions);
+    pJoin->pOnConditions = NULL;
+    return TSDB_CODE_SUCCESS;
+  } else {
+    return TSDB_CODE_PLAN_INTERNAL_ERROR;
+  }
+}
+
+static int32_t pushDownCondOptJoinExtractMergeCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {
+  int32_t code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin);
+  SNode* pJoinMergeCond = NULL;
+  SNode* pJoinOnCond = NULL;
+  if (TSDB_CODE_SUCCESS == code) {
+    code = pushDownCondOptPartJoinOnCond(pJoin, &pJoinMergeCond, &pJoinOnCond);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    pJoin->pMergeCondition = pJoinMergeCond;
+    pJoin->pOnConditions = pJoinOnCond;
+  } else {
+    nodesDestroyNode(pJoinMergeCond);
+    nodesDestroyNode(pJoinOnCond);
+  }
+  return code;
+}
+
 static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {
   if (OPTIMIZE_FLAG_TEST_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE)) {
     return TSDB_CODE_SUCCESS;
   }
 
   if (NULL == pJoin->node.pConditions) {
-    return pushDownCondOptCheckJoinOnCond(pCxt, pJoin);
+    int32_t code = pushDownCondOptJoinExtractMergeCond(pCxt, pJoin);
+    if (TSDB_CODE_SUCCESS == code) {
+      OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE);
+      pCxt->optimized = true;
+    }
+    return code;
   }
 
   SNode*  pOnCond = NULL;
@@ -579,10 +660,13 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p
         pushDownCondOptPushCondToChild(pCxt, (SLogicNode*)nodesListGetNode(pJoin->node.pChildren, 1), &pRightChildCond);
   }
 
+  if (TSDB_CODE_SUCCESS == code) {
+    code = pushDownCondOptJoinExtractMergeCond(pCxt, pJoin);
+  }
+
   if (TSDB_CODE_SUCCESS == code) {
     OPTIMIZE_FLAG_SET_MASK(pJoin->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE);
     pCxt->optimized = true;
-    code = pushDownCondOptCheckJoinOnCond(pCxt, pJoin);
   } else {
     nodesDestroyNode(pOnCond);
     nodesDestroyNode(pLeftChildCond);
@@ -720,7 +804,8 @@ static int32_t pushDownCondOptDealAgg(SOptimizeContext* pCxt, SAggLogicNode* pAg
   // TODO: remove it after full implementation of pushing down to child
   if (1 != LIST_LENGTH(pAgg->node.pChildren) ||
       QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) &&
-          QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) {
+      QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) &&
+      QUERY_NODE_LOGIC_PLAN_JOIN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) {
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1460,6 +1545,7 @@ static int32_t rewriteTailOptCreateSort(SIndefRowsFuncLogicNode* pIndef, SLogicN
 
   pSort->groupSort = rewriteTailOptNeedGroupSort(pIndef);
   TSWAP(pSort->node.pChildren, pIndef->node.pChildren);
+  optResetParent((SLogicNode*)pSort);
   pSort->node.precision = pIndef->node.precision;
 
   SFunctionNode* pTail = NULL;
@@ -1667,6 +1753,7 @@ static int32_t rewriteUniqueOptCreateAgg(SIndefRowsFuncLogicNode* pIndef, SLogic
   }
 
   TSWAP(pAgg->node.pChildren, pIndef->node.pChildren);
+  optResetParent((SLogicNode*)pAgg);
   pAgg->node.precision = pIndef->node.precision;
 
   int32_t code = TSDB_CODE_SUCCESS;

source/libs/planner/src/planPhysiCreater.c

@@ -612,10 +612,8 @@ static int32_t createJoinPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
   int32_t             code = TSDB_CODE_SUCCESS;
 
   pJoin->joinType = pJoinLogicNode->joinType;
-  if (NULL != pJoinLogicNode->pOnConditions) {
-    code = setNodeSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->pOnConditions,
-                         &pJoin->pOnConditions);
-  }
+  setNodeSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->pMergeCondition,
+                &pJoin->pMergeCondition);
   if (TSDB_CODE_SUCCESS == code) {
     code = setListSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->node.pTargets,
                          &pJoin->pTargets);
@@ -623,6 +621,21 @@ static int32_t createJoinPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
   if (TSDB_CODE_SUCCESS == code) {
     code = addDataBlockSlots(pCxt, pJoin->pTargets, pJoin->node.pOutputDataBlockDesc);
   }
+
+  SNodeList* condCols = nodesMakeList();
+  if (TSDB_CODE_SUCCESS == code && NULL != pJoinLogicNode->pOnConditions) {
+    code = nodesCollectColumnsFromNode(pJoinLogicNode->pOnConditions, NULL, COLLECT_COL_TYPE_ALL, &condCols);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = addDataBlockSlots(pCxt, condCols, pJoin->node.pOutputDataBlockDesc);
+    nodesDestroyList(condCols);
+  }
+
+  if (TSDB_CODE_SUCCESS == code && NULL != pJoinLogicNode->pOnConditions) {
+    code = setNodeSlotId(pCxt, ((SPhysiNode*)pJoin)->pOutputDataBlockDesc->dataBlockId, -1, pJoinLogicNode->pOnConditions,
+                         &pJoin->pOnConditions);
+  }
+
   if (TSDB_CODE_SUCCESS == code) {
     code = setConditionsSlotId(pCxt, (const SLogicNode*)pJoinLogicNode, (SPhysiNode*)pJoin);
   }

source/libs/planner/src/planSpliter.c

@@ -1203,7 +1203,8 @@ typedef struct SQnodeSplitInfo {
 
 static bool qndSplFindSplitNode(SSplitContext* pCxt, SLogicSubplan* pSubplan, SLogicNode* pNode,
                                 SQnodeSplitInfo* pInfo) {
-  if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode) && NULL != pNode->pParent) {
+  if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode) && NULL != pNode->pParent &&
+      ((SScanLogicNode*)pNode)->scanSeq[0] < 1 && ((SScanLogicNode*)pNode)->scanSeq[1] < 1) {
     pInfo->pSplitNode = pNode;
     pInfo->pSubplan = pSubplan;
     return true;

source/libs/planner/test/planBasicTest.cpp

@@ -63,6 +63,10 @@ TEST_F(PlanBasicTest, uniqueFunc) {
   run("SELECT UNIQUE(c2 + 10), ts, c2 FROM t1 WHERE c1 > 10");
 
   run("SELECT UNIQUE(c1) a FROM t1 ORDER BY a");
+
+  run("SELECT ts, UNIQUE(c1) FROM st1 PARTITION BY TBNAME");
+
+  run("SELECT TBNAME, UNIQUE(c1) FROM st1 PARTITION BY TBNAME");
 }
 
 TEST_F(PlanBasicTest, tailFunc) {
@@ -81,6 +85,8 @@ TEST_F(PlanBasicTest, tailFunc) {
   run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 PARTITION BY c1 LIMIT 5");
 
   run("SELECT TAIL(c1, 2, 1) FROM st1s1 UNION ALL SELECT c1 FROM st1s2");
+
+  run("SELECT TAIL(c1, 1) FROM st2 WHERE jtag->'tag1' > 10");
 }
 
 TEST_F(PlanBasicTest, interpFunc) {

source/libs/planner/test/planOtherTest.cpp

@@ -91,10 +91,3 @@ TEST_F(PlanOtherTest, delete) {
 
   run("DELETE FROM st1 WHERE ts > now - 2d and ts < now - 1d AND tag1 = 10");
 }
-
-TEST_F(PlanOtherTest, queryPolicy) {
-  useDb("root", "test");
-
-  tsQueryPolicy = QUERY_POLICY_QNODE;
-  run("SELECT COUNT(*) FROM st1");
-}

source/libs/planner/test/planTestMain.cpp

@@ -78,6 +78,7 @@ static void parseArg(int argc, char* argv[]) {
     {"skipSql", required_argument, NULL, 's'},
     {"limitSql", required_argument, NULL, 'i'},
     {"log", required_argument, NULL, 'l'},
+    {"queryPolicy", required_argument, NULL, 'q'},
     {0, 0, 0, 0}
   };
   // clang-format on
@@ -95,6 +96,9 @@ static void parseArg(int argc, char* argv[]) {
       case 'l':
         setLogLevel(optarg);
         break;
+      case 'q':
+        setQueryPolicy(optarg);
+        break;
       default:
         break;
     }

source/libs/planner/test/planTestUtil.cpp

@@ -24,6 +24,7 @@
 #include "mockCatalogService.h"
 #include "parser.h"
 #include "planInt.h"
+#include "tglobal.h"
 
 using namespace std;
 using namespace testing;
@@ -53,6 +54,7 @@ DumpModule g_dumpModule = DUMP_MODULE_NOTHING;
 int32_t    g_skipSql = 0;
 int32_t    g_limitSql = 0;
 int32_t    g_logLevel = 131;
+int32_t    g_queryPolicy = QUERY_POLICY_VNODE;
 
 void setDumpModule(const char* pModule) {
   if (NULL == pModule) {
@@ -79,6 +81,7 @@ void setDumpModule(const char* pModule) {
 void setSkipSqlNum(const char* pNum) { g_skipSql = stoi(pNum); }
 void setLimitSqlNum(const char* pNum) { g_limitSql = stoi(pNum); }
 void setLogLevel(const char* pLogLevel) { g_logLevel = stoi(pLogLevel); }
+void setQueryPolicy(const char* pQueryPolicy) { g_queryPolicy = stoi(pQueryPolicy); }
 
 int32_t getLogLevel() { return g_logLevel; }
 
@@ -105,7 +108,23 @@ class PlannerTestBaseImpl {
     }
     ++sqlNum_;
 
+    switch (g_queryPolicy) {
+      case QUERY_POLICY_VNODE:
+      case QUERY_POLICY_HYBRID:
+      case QUERY_POLICY_QNODE:
+        runImpl(sql, g_queryPolicy);
+        break;
+      default:
+        runImpl(sql, QUERY_POLICY_VNODE);
+        runImpl(sql, QUERY_POLICY_HYBRID);
+        runImpl(sql, QUERY_POLICY_QNODE);
+        break;
+    }
+  }
+
+  void runImpl(const string& sql, int32_t queryPolicy) {
     reset();
+    tsQueryPolicy = queryPolicy;
     try {
       SQuery* pQuery = nullptr;
       doParseSql(sql, &pQuery);

source/libs/planner/test/planTestUtil.h

@@ -45,6 +45,7 @@ extern void    setDumpModule(const char* pModule);
 extern void    setSkipSqlNum(const char* pNum);
 extern void    setLimitSqlNum(const char* pNum);
 extern void    setLogLevel(const char* pLogLevel);
+extern void    setQueryPolicy(const char* pQueryPolicy);
 extern int32_t getLogLevel();
 
 #endif  // PLAN_TEST_UTIL_H

source/libs/sync/inc/syncSnapshot.h

@@ -83,9 +83,9 @@ typedef struct SSyncSnapshotReceiver {
 
 SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId);
 void                   snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
-int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg);
-int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
-bool    snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
+int32_t                snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg);
+int32_t                snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
+bool                   snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
 
 cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
 char  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);

source/libs/sync/src/syncMain.c

@@ -158,13 +158,13 @@ int32_t syncSetStandby(int64_t rid) {
   }
 
   if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) {
-    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
     if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
       terrno = TSDB_CODE_SYN_IS_LEADER;
     } else {
       terrno = TSDB_CODE_SYN_STANDBY_NOT_READY;
     }
     sError("failed to set standby since it is not follower, state:%s rid:%" PRId64, syncStr(pSyncNode->state), rid);
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
     return -1;
   }
 
@@ -620,6 +620,7 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
 
   SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
+    taosReleaseRef(tsNodeRefId, rid);
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
     return -1;
   }

source/libs/sync/src/syncRespMgr.c

@@ -14,6 +14,7 @@
  */
 
 #include "syncRespMgr.h"
+#include "syncRaftEntry.h"
 #include "syncRaftStore.h"
 
 SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) {
@@ -116,4 +117,59 @@ void syncRespClean(SSyncRespMgr *pObj) {
   taosThreadMutexUnlock(&(pObj->mutex));
 }
 
-void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {}
+void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
+  SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL);
+  int        cnt = 0;
+  SSyncNode *pSyncNode = pObj->data;
+
+  SArray *delIndexArray = taosArrayInit(0, sizeof(SyncIndex));
+  ASSERT(delIndexArray != NULL);
+
+  while (pStub) {
+    size_t     len;
+    void      *key = taosHashGetKey(pStub, &len);
+    SyncIndex *pIndex = (SyncIndex *)key;
+
+    int64_t nowMS = taosGetTimestampMs();
+    if (nowMS - pStub->createTime > ttl) {
+      taosArrayPush(delIndexArray, pIndex);
+      cnt++;
+
+      SSyncRaftEntry *pEntry = NULL;
+      int32_t         code = 0;
+      if (pSyncNode->pLogStore != NULL) {
+        code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, *pIndex, &pEntry);
+        if (code == 0 && pEntry != NULL) {
+          SFsmCbMeta cbMeta = {0};
+          cbMeta.index = pEntry->index;
+          cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, cbMeta.index);
+          cbMeta.isWeak = pEntry->isWeak;
+          cbMeta.code = TSDB_CODE_SYN_TIMEOUT;
+          cbMeta.state = pSyncNode->state;
+          cbMeta.seqNum = pEntry->seqNum;
+          cbMeta.term = pEntry->term;
+          cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
+          cbMeta.flag = 0;
+
+          SRpcMsg rpcMsg = pStub->rpcMsg;
+          rpcMsg.pCont = rpcMallocCont(pEntry->dataLen);
+          memcpy(rpcMsg.pCont, pEntry->data, pEntry->dataLen);
+          pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta);
+
+          syncEntryDestory(pEntry);
+        }
+      }
+    }
+
+    pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub);
+  }
+
+  int32_t arraySize = taosArrayGetSize(delIndexArray);
+  sDebug("vgId:%d, resp clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize);
+
+  for (int32_t i = 0; i < arraySize; ++i) {
+    SyncIndex *pIndex = taosArrayGet(delIndexArray, i);
+    taosHashRemove(pObj->pRespHash, pIndex, sizeof(SyncIndex));
+  }
+  taosArrayDestroy(delIndexArray);
+}

source/libs/sync/src/syncSnapshot.c

@@ -22,9 +22,11 @@
 #include "wal.h"
 
 //----------------------------------
-static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
-                                    SyncSnapshotSend *pBeginMsg);
-static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg);
+static void    snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg);
+static void    snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg);
+static void    snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver);
+static void    snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg);
+static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg);
 
 //----------------------------------
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex) {
@@ -68,7 +70,9 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
     // close reader
     if (pSender->pReader != NULL) {
       int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader);
-      ASSERT(ret == 0);
+      if (ret != 0) {
+        syncNodeErrorLog(pSender->pSyncNode, "stop reader error");
+      }
       pSender->pReader = NULL;
     }
 
@@ -79,7 +83,12 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
 
 bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; }
 
-// begin send snapshot by snapshot, pReader
+// begin send snapshot by param, snapshot, pReader
+//
+// action:
+// 1. assert reader not start
+// 2. update state
+// 3. send first snapshot block
 int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapshotParam, SSnapshot snapshot,
                             void *pReader) {
   ASSERT(!snapshotSenderIsStart(pSender));
@@ -98,7 +107,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho
 
   // update term
   pSender->term = pSender->pSyncNode->pRaftStore->currentTerm;
-  ++(pSender->privateTerm);
+  ++(pSender->privateTerm);  // increase private term
 
   // update state
   pSender->finish = false;
@@ -114,9 +123,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho
 
     code = pSender->pSyncNode->pLogStore->syncLogGetEntry(pSender->pSyncNode->pLogStore,
                                                           pSender->snapshot.lastConfigIndex, &pEntry);
-    if (code == 0) {
-      ASSERT(pEntry != NULL);
-
+    if (code == 0 && pEntry != NULL) {
       SRpcMsg rpcMsg;
       syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
@@ -207,6 +214,8 @@ int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
   pSender->start = false;
   pSender->finish = finish;
 
+  // do not update term, maybe print
+
   // event log
   do {
     char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender stop");
@@ -243,6 +252,7 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
   pMsg->srcId = pSender->pSyncNode->myRaftId;
   pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex];
   pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm;
+  pMsg->beginIndex = pSender->snapshotParam.start;
   pMsg->lastIndex = pSender->snapshot.lastApplyIndex;
   pMsg->lastTerm = pSender->snapshot.lastApplyTerm;
   pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex;
@@ -281,11 +291,13 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) {
     pMsg->srcId = pSender->pSyncNode->myRaftId;
     pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex];
     pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm;
+    pMsg->beginIndex = pSender->snapshotParam.start;
     pMsg->lastIndex = pSender->snapshot.lastApplyIndex;
     pMsg->lastTerm = pSender->snapshot.lastApplyTerm;
     pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex;
     pMsg->lastConfig = pSender->lastConfig;
     pMsg->seq = pSender->seq;
+    pMsg->privateTerm = pSender->privateTerm;
     memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen);
 
     // send msg
@@ -305,6 +317,12 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) {
   return 0;
 }
 
+static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg) {
+  ASSERT(pMsg->ack == pSender->seq);
+  pSender->ack = pMsg->ack;
+  ++(pSender->seq);
+}
+
 cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
@@ -371,10 +389,11 @@ char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
   syncUtilU642Addr(destId.addr, host, sizeof(host), &port);
 
   snprintf(s, len,
-           "%s {%p laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu replica-index:%d %s:%d}", event,
-           pSender, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm,
-           pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack, pSender->finish, pSender->privateTerm,
-           pSender->replicaIndex, host, port);
+           "%s {%p s-param:%ld e-param:%ld laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu "
+           "replica-index:%d %s:%d}",
+           event, pSender, pSender->snapshotParam.start, pSender->snapshotParam.end, pSender->snapshot.lastApplyIndex,
+           pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack,
+           pSender->finish, pSender->privateTerm, pSender->replicaIndex, host, port);
 
   return s;
 }
@@ -429,11 +448,10 @@ bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceive
 // static do start by privateTerm, pBeginMsg
 // receive first snapshot data
 // write first block data
-static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
-                                    SyncSnapshotSend *pBeginMsg) {
+static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) {
   // update state
   pReceiver->term = pReceiver->pSyncNode->pRaftStore->currentTerm;
-  pReceiver->privateTerm = privateTerm;
+  pReceiver->privateTerm = pBeginMsg->privateTerm;
   pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN;
   pReceiver->fromId = pBeginMsg->srcId;
   pReceiver->start = true;
@@ -445,7 +463,7 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p
   pReceiver->snapshotParam.start = pBeginMsg->beginIndex;
   pReceiver->snapshotParam.end = pBeginMsg->lastIndex;
 
-  // write data
+  // start writer
   ASSERT(pReceiver->pWriter == NULL);
   int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStartWrite(pReceiver->pSyncNode->pFsm,
                                                                  &(pReceiver->snapshotParam), &(pReceiver->pWriter));
@@ -481,10 +499,10 @@ static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
 
 // if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
 // if already start, force close, start again
-int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) {
+int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) {
   if (!snapshotReceiverIsStart(pReceiver)) {
     // first start
-    snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
+    snapshotReceiverDoStart(pReceiver, pBeginMsg);
 
   } else {
     // already start
@@ -494,12 +512,14 @@ int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm private
     snapshotReceiverForceStop(pReceiver);
 
     // start again
-    snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
+    snapshotReceiverDoStart(pReceiver, pBeginMsg);
   }
 
   return 0;
 }
 
+// just set start = false
+// FpSnapshotStopWrite should not be called, assert writer == NULL
 int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
   if (pReceiver->pWriter != NULL) {
     int32_t ret =
@@ -522,6 +542,7 @@ int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
   return 0;
 }
 
+// when recv last snapshot block, apply data into snapshot
 static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
   ASSERT(pMsg->seq == SYNC_SNAPSHOT_SEQ_END);
 
@@ -550,7 +571,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
       pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
     }
 
-    // stop writer
+    // stop writer, apply data
     code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true);
     if (code != 0) {
       syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
@@ -579,15 +600,20 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
   return 0;
 }
 
+// apply data block
+// update progress
 static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
   ASSERT(pMsg->seq == pReceiver->ack + 1);
 
   if (pReceiver->pWriter != NULL) {
     if (pMsg->dataLen > 0) {
+      // apply data block
       int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
                                                                    pMsg->data, pMsg->dataLen);
       ASSERT(code == 0);
     }
+
+    // update progress
     pReceiver->ack = pMsg->seq;
 
     // event log
@@ -665,14 +691,23 @@ char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event)
   uint16_t port;
   syncUtilU642Addr(fromId.addr, host, sizeof(host), &port);
 
-  snprintf(s, len, "%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld}", event,
-           pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port,
-           pReceiver->snapshot.lastApplyIndex, pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex);
+  snprintf(s, len,
+           "%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d s-param:%ld e-param:%ld laindex:%ld laterm:%lu "
+           "lcindex:%ld}",
+           event, pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port,
+           pReceiver->snapshotParam.start, pReceiver->snapshotParam.end, pReceiver->snapshot.lastApplyIndex,
+           pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex);
 
   return s;
 }
 
-// receiver do something
+// receiver on message
+//
+// condition 1, recv SYNC_SNAPSHOT_SEQ_BEGIN, start receiver, update privateTerm
+// condition 2, recv SYNC_SNAPSHOT_SEQ_END, finish receiver(apply snapshot data, update commit index, maybe reconfig)
+// condition 3, recv SYNC_SNAPSHOT_SEQ_FORCE_CLOSE, force close
+// condition 4, got data, update ack
+//
 int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
   // get receiver
   SSyncSnapshotReceiver *pReceiver = pSyncNode->pNewNodeReceiver;
@@ -683,11 +718,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
   if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
       if (pMsg->seq == SYNC_SNAPSHOT_SEQ_BEGIN) {
+        // condition 1
         // begin, no data
-        snapshotReceiverStart(pReceiver, pMsg->privateTerm, pMsg);
+        snapshotReceiverStart(pReceiver, pMsg);
         needRsp = true;
 
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) {
+        // condition 2
         // end, finish FSM
         code = snapshotReceiverFinish(pReceiver, pMsg);
         if (code == 0) {
@@ -697,7 +734,6 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
 
         // maybe update lastconfig
         if (pMsg->lastConfigIndex >= SYNC_INDEX_BEGIN) {
-          // int32_t  oldReplicaNum = pSyncNode->replicaNum;
           SSyncCfg oldSyncCfg = pSyncNode->pRaftCfg->cfg;
 
           // update new config myIndex
@@ -709,11 +745,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
         }
 
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_FORCE_CLOSE) {
+        // condition 3
         // force close
         snapshotReceiverForceStop(pReceiver);
         needRsp = false;
 
       } else if (pMsg->seq > SYNC_SNAPSHOT_SEQ_BEGIN && pMsg->seq < SYNC_SNAPSHOT_SEQ_END) {
+        // condition 4
         // transfering
         if (pMsg->seq == pReceiver->ack + 1) {
           snapshotReceiverGotData(pReceiver, pMsg);
@@ -752,6 +790,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
         syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg);
         syncSnapshotRspDestroy(pRspMsg);
       }
+
     } else {
       // error log
       do {
@@ -759,6 +798,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
         syncNodeErrorLog(pSyncNode, eventLog);
         taosMemoryFree(eventLog);
       } while (0);
+
+      return -1;
     }
   } else {
     // error log
@@ -767,19 +808,19 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
       syncNodeErrorLog(pSyncNode, eventLog);
       taosMemoryFree(eventLog);
     } while (0);
+
+    return -1;
   }
 
   return 0;
 }
 
-static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg) {
-  ASSERT(pMsg->ack == pSender->seq);
-  pSender->ack = pMsg->ack;
-  ++(pSender->seq);
-}
-
-// sender receives ack, set seq = ack + 1, send msg from seq
-// if ack == SYNC_SNAPSHOT_SEQ_END, stop sender
+// sender on message
+//
+// condition 1 sender receives SYNC_SNAPSHOT_SEQ_END, close sender
+// condition 2 sender receives ack, set seq = ack + 1, send msg from seq
+// condition 3 sender receives error msg, just print error log
+//
 int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
   // if already drop replica, do not process
   if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
@@ -794,12 +835,14 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
   // state, term, seq/ack
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
-      // receiver ack is finish, close sender
+      // condition 1
+      // receive ack is finish, close sender
       if (pMsg->ack == SYNC_SNAPSHOT_SEQ_END) {
         snapshotSenderStop(pSender, true);
         return 0;
       }
 
+      // condition 2
       // send next msg
       if (pMsg->ack == pSender->seq) {
         // update sender ack
@@ -807,6 +850,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
         snapshotSend(pSender);
 
       } else if (pMsg->ack == pSender->seq - 1) {
+        // maybe resend
         snapshotReSend(pSender);
 
       } else {

source/libs/sync/test/syncIndexTest.cpp

@@ -8,7 +8,13 @@ void print(SHashObj *pNextIndex) {
   printf("----------------\n");
   uint64_t *p = (uint64_t *)taosHashIterate(pNextIndex, NULL);
   while (p) {
-    printf("%lu \n", *p);
+
+    size_t len;
+    void*  key = taosHashGetKey(p, &len);
+
+    SRaftId *pRaftId = (SRaftId*)key;
+
+    printf("key:<%lu, %d>, value:%lu \n", pRaftId->addr, pRaftId->vgId, *p);
     p = (uint64_t *)taosHashIterate(pNextIndex, p);
   }
 }

source/libs/sync/test/syncRespMgrTest.cpp

@@ -73,9 +73,15 @@ void syncRespMgrGetAndDelTest(uint64_t i) {
   }
 }
 
+SSyncNode *createSyncNode() {
+  SSyncNode *pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode));
+  memset(pSyncNode, 0, sizeof(SSyncNode));
+  return pSyncNode;
+}
+
 void test1() {
   printf("------- test1 ---------\n");
-  pMgr = syncRespMgrCreate(NULL, 0);
+  pMgr = syncRespMgrCreate(createSyncNode(), 0);
   assert(pMgr != NULL);
 
   syncRespMgrInsert(10);
@@ -100,7 +106,7 @@ void test1() {
 
 void test2() {
   printf("------- test2 ---------\n");
-  pMgr = syncRespMgrCreate(NULL, 0);
+  pMgr = syncRespMgrCreate(createSyncNode(), 0);
   assert(pMgr != NULL);
 
   syncRespMgrInsert(10);
@@ -117,7 +123,7 @@ void test2() {
 
 void test3() {
   printf("------- test3 ---------\n");
-  pMgr = syncRespMgrCreate(NULL, 0);
+  pMgr = syncRespMgrCreate(createSyncNode(), 0);
   assert(pMgr != NULL);
 
   syncRespMgrInsert(10);
@@ -132,13 +138,34 @@ void test3() {
   syncRespMgrDestroy(pMgr);
 }
 
+void test4() {
+  printf("------- test4 ---------\n");
+  pMgr = syncRespMgrCreate(createSyncNode(), 2);
+  assert(pMgr != NULL);
+
+  syncRespMgrInsert(5);
+  syncRespMgrPrint();
+
+  taosMsleep(3000);
+
+  syncRespMgrInsert(3);
+  syncRespMgrPrint();
+
+  printf("====== after clean ttl \n");
+  syncRespClean(pMgr);
+  syncRespMgrPrint();
+
+  syncRespMgrDestroy(pMgr);
+}
+
 int main() {
   tsAsyncLog = 0;
-  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  sDebugFlag = DEBUG_DEBUG + DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
   logTest();
   test1();
   test2();
   test3();
+  test4();
 
   return 0;
 }

source/libs/transport/src/transSvr.c

@@ -402,8 +402,9 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
     } else {
       pHead->msgType = pMsg->msgType;
       // set up resp msg type
-      if (pHead->msgType == 0 && transMsgLenFromCont(pMsg->contLen) == sizeof(STransMsgHead))
+      if (pHead->msgType == 0 && transMsgLenFromCont(pMsg->contLen) == sizeof(STransMsgHead)) {
         pHead->msgType = pConn->inType + 1;
+      }
     }
   }
 

source/util/src/tpagedbuf.c

@@ -637,6 +637,7 @@ void dBufSetBufPageRecycled(SDiskbasedBuf* pBuf, void* pPage) {
   SListNode* pNode = tdListPopNode(pBuf->lruList, ppi->pn);
   taosMemoryFreeClear(ppi->pData);
   taosMemoryFreeClear(pNode);
+  ppi->pn = NULL;
 
   tdListAppend(pBuf->freePgList, &ppi);
 }

tests/script/tsim/stable/tag_modify.sim

@@ -14,7 +14,7 @@ sql_error alter table db.stb MODIFY tag ts int
 sql_error alter table db.stb MODIFY tag t2 binary(3)
 sql_error alter table db.stb MODIFY tag t2 int
 sql_error alter table db.stb MODIFY tag t1 int
-sql create table db.ctb using db.stb tags(101, "12345")
+sql create table db.ctb using db.stb tags(101, "123")
 sql insert into db.ctb values(now, 1, "1234")
 
 sql select * from db.stb
@@ -32,7 +32,7 @@ endi
 if $data[0][3] != 101 then
   return -1
 endi
-if $data[0][4] != 1234 then
+if $data[0][4] != 123 then
   return -1
 endi
 

tests/script/tsim/stable/tag_rename.sim

@@ -14,7 +14,7 @@ sql_error alter table db.stb rename tag ts c3
 sql_error alter table db.stb rename tag t2 t1
 sql_error alter table db.stb rename tag t2 t2
 sql_error alter table db.stb rename tag t1 t2
-sql create table db.ctb using db.stb tags(101, "12345")
+sql create table db.ctb using db.stb tags(101, "123")
 sql insert into db.ctb values(now, 1, "1234")
 
 sql select * from db.stb
@@ -32,7 +32,7 @@ endi
 if $data[0][3] != 101 then
   return -1
 endi
-if $data[0][4] != 1234 then
+if $data[0][4] != 123 then
   return -1
 endi
 
@@ -56,7 +56,7 @@ endi
 if $data[0][3] != 101 then
   return -1
 endi
-if $data[0][4] != 1234 then
+if $data[0][4] != 123 then
   return -1
 endi
 

tests/script/tsim/valgrind/basic.sim

@@ -26,8 +26,8 @@ sql show users
 sql alter user u1 sysinfo 1
 sql alter user u1 enable 1
 sql alter user u1 pass 'taosdata'
-sql alter user u2 sysinfo 0
 sql drop user u1
+sql_error alter user u2 sysinfo 0
 
 _OVER:
 system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/valgrind/basic2.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start -v
+sql connect
+
+print =============== step1: create drop show dnodes
+$x = 0
+step1:
+	$x = $x + 1
+	sleep 1000
+	if $x == 10 then
+	  print ====> dnode not ready!
+		return -1
+	endi
+sql show dnodes
+print ===> $data00 $data01 $data02 $data03 $data04 $data05
+if $rows != 1 then
+  return -1
+endi
+
+print =============== step2: create db
+sql create database db vgroups 1
+
+_OVER:
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/system-test/fulltest.sh

@@ -21,7 +21,7 @@ python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
 python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
 python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py
 python3 ./test.py -f 1-insert/alter_stable.py
-python3 ./test.py -f 1-insert/alter_table.py
+#python3 ./test.py -f 1-insert/alter_table.py
 python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
 python3 ./test.py -f 1-insert/table_comment.py
 python3 ./test.py -f 1-insert/time_range_wise.py
@@ -118,7 +118,7 @@ python3 ./test.py -f 2-query/twa.py
 python3 ./test.py -f 2-query/irate.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
@@ -157,7 +157,7 @@ python3 ./test.py -f 7-tmq/tmqCheckData1.py
 python3 ./test.py -f 7-tmq/tmqUdf.py
 #python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 5
 python3 ./test.py -f 7-tmq/tmqConsumerGroup.py
-python3 ./test.py -f 7-tmq/tmqShow.py
+#python3 ./test.py -f 7-tmq/tmqShow.py
 python3 ./test.py -f 7-tmq/tmqAlterSchema.py
 python3 ./test.py -f 7-tmq/tmqConsFromTsdb.py
 python3 ./test.py -f 7-tmq/tmqConsFromTsdb1.py

tests/system-test/simpletest.bat

@@ -6,7 +6,7 @@ python3 .\test.py -f 0-others\telemetry.py
 python3 .\test.py -f 0-others\taosdMonitor.py
 python3 .\test.py -f 0-others\udfTest.py
 python3 .\test.py -f 0-others\udf_create.py
-python3 .\test.py -f 0-others\udf_restart_taosd.py
+@REM python3 .\test.py -f 0-others\udf_restart_taosd.py
 @REM python3 .\test.py -f 0-others\cachelast.py
 
 @REM python3 .\test.py -f 0-others\user_control.py