Merge branch '3.0' into enh/top_bot_split

docs-en/14-reference/03-connector/rust.mdx

@@ -45,15 +45,15 @@ Add the [libtaos][libtaos] dependency to the [Rust](https://rust-lang.org) proje
 <Tabs defaultValue="native">
 <TabItem value="native" label="native connection">
 
-Add [libtaos][libtaos] to the ``Cargo.toml`'' file.
+Add [libtaos][libtaos] to the `Cargo.toml` file.
 
-``toml
+```toml
 [dependencies]
 # use default feature
 libtaos = "*"
 ```
 
-</TabItem
+</TabItem>
 <TabItem value="rest" label="REST connection">
 
 Add [libtaos][libtaos] to the `Cargo.toml` file and enable the `rest` feature.

include/common/tcommon.h

@@ -50,6 +50,7 @@ typedef enum EStreamType {
   STREAM_INVERT,
   STREAM_REPROCESS,
   STREAM_INVALID,
+  STREAM_GET_ALL,
 } EStreamType;
 
 typedef struct {

include/common/tglobal.h

@@ -96,6 +96,7 @@ extern bool    tsDeadLockKillQuery;
 
 // query client
 extern int32_t tsQueryPolicy;
+extern int32_t tsQuerySmaOptimize;
 
 // client
 extern int32_t tsMinSlidingTime;

include/common/tmsg.h

@@ -1496,6 +1496,7 @@ typedef struct {
 
 #define STREAM_TRIGGER_AT_ONCE          1
 #define STREAM_TRIGGER_WINDOW_CLOSE     2
+#define STREAM_TRIGGER_MAX_DELAY        3
 
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
@@ -2343,15 +2344,17 @@ typedef struct {
   char           indexName[TSDB_INDEX_NAME_LEN];
   int32_t        exprLen;
   int32_t        tagsFilterLen;
-  int32_t   numOfVgroups;
   int64_t        indexUid;
   tb_uid_t       tableUid;  // super/child/common table uid
+  tb_uid_t       dstTbUid;  // for dstVgroup
   int64_t        interval;
   int64_t        offset;  // use unit by precision of DB
   int64_t        sliding;
+  char*          dstTbName;  // for dstVgroup
   char*          expr;       // sma expression
   char*          tagsFilter;
-  SVgEpSet* pVgEpSet;
+  SSchemaWrapper schemaRow;  // for dstVgroup
+  SSchemaWrapper schemaTag;  // for dstVgroup
 } STSma;                     // Time-range-wise SMA
 
 typedef STSma SVCreateTSmaReq;
@@ -2436,27 +2439,6 @@ static int32_t tDecodeTSmaWrapper(SDecoder* pDecoder, STSmaWrapper* pReq) {
   return 0;
 }
 
-typedef struct {
-  int64_t     indexUid;
-  STimeWindow queryWindow;
-} SVGetTsmaExpWndsReq;
-
-#define SMA_WNDS_EXPIRE_FLAG      (0x1)
-#define SMA_WNDS_IS_EXPIRE(flag)  (((flag)&SMA_WNDS_EXPIRE_FLAG) != 0)
-#define SMA_WNDS_SET_EXPIRE(flag) ((flag) |= SMA_WNDS_EXPIRE_FLAG)
-
-typedef struct {
-  int64_t indexUid;
-  int8_t  flags;  // 0x1 all window expired
-  int32_t numExpWnds;
-  TSKEY   wndSKeys[];
-} SVGetTsmaExpWndsRsp;
-
-int32_t tEncodeSVGetTSmaExpWndsReq(SEncoder* pCoder, const SVGetTsmaExpWndsReq* pReq);
-int32_t tDecodeSVGetTsmaExpWndsReq(SDecoder* pCoder, SVGetTsmaExpWndsReq* pReq);
-int32_t tEncodeSVGetTSmaExpWndsRsp(SEncoder* pCoder, const SVGetTsmaExpWndsRsp* pReq);
-int32_t tDecodeSVGetTsmaExpWndsRsp(SDecoder* pCoder, SVGetTsmaExpWndsRsp* pReq);
-
 typedef struct {
   int idx;
 } SMCreateFullTextReq;
@@ -2516,6 +2498,7 @@ typedef struct {
 
 int32_t tSerializeSTableIndexRsp(void* buf, int32_t bufLen, const STableIndexRsp* pRsp);
 int32_t tDeserializeSTableIndexRsp(void* buf, int32_t bufLen, STableIndexRsp* pRsp);
+
 void tFreeSTableIndexInfo(void* pInfo);
 
 typedef struct {

include/common/tmsgdef.h

@@ -190,7 +190,6 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_SMA, "vnode-drop-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT_RSMA, "vnode-submit-rsma", SSubmitReq, SSubmitRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_GET_TSMA_EXP_WNDS, "vnode-get-tsma-expired-windows", SVGetTsmaExpWndsReq, SVGetTsmaExpWndsRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "delete-data", SVDeleteReq, SVDeleteRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "alter-config", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_REPLICA, "alter-replica", NULL, NULL)
@@ -237,6 +236,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_SYNC_CONFIG_CHANGE, "sync-config-change", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_SEND, "sync-snapshot-send", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_RSP, "sync-snapshot-rsp", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_SYNC_LEADER_TRANSFER, "sync-leader-transfer", NULL, NULL)
   
 #if defined(TD_MSG_NUMBER_)
   TDMT_MAX

include/libs/function/functionMgt.h

@@ -172,6 +172,7 @@ bool fmIsMultiResFunc(int32_t funcId);
 bool fmIsRepeatScanFunc(int32_t funcId);
 bool fmIsUserDefinedFunc(int32_t funcId);
 bool fmIsDistExecFunc(int32_t funcId);
+bool fmIsForbidFillFunc(int32_t funcId);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
 

include/libs/nodes/cmdnodes.h

@@ -47,7 +47,7 @@ typedef struct SDatabaseOptions {
   int32_t     maxRowsPerBlock;
   int32_t     minRowsPerBlock;
   SNodeList*  pKeep;
-  int32_t     keep[3];
+  int64_t     keep[3];
   int32_t     pages;
   int32_t     pagesize;
   char        precisionStr[3];

include/libs/parser/parser.h

@@ -63,7 +63,7 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata
 void qDestroyQuery(SQuery* pQueryNode);
 
 int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
-int32_t qSetSTableIdForRSma(SNode* pStmt, int64_t uid);
+int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid);
 
 int32_t     qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
 int32_t     qResetStmtDataBlock(void* block, bool keepBuf);

include/libs/sync/sync.h

@@ -48,6 +48,7 @@ typedef enum {
   TAOS_SYNC_PROPOSE_SUCCESS = 0,
   TAOS_SYNC_PROPOSE_NOT_LEADER = 1,
   TAOS_SYNC_PROPOSE_OTHER_ERROR = 2,
+  TAOS_SYNC_ONLY_ONE_REPLICA = 3,
 } ESyncProposeCode;
 
 typedef enum {
@@ -200,6 +201,9 @@ int32_t     syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg);
 int32_t syncReconfigRaw(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg);
 
+int32_t syncLeaderTransfer(int64_t rid);
+int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader);
+
 // to be moved to static
 void syncStartNormal(int64_t rid);
 void syncStartStandBy(int64_t rid);

include/libs/sync/syncTools.h

@@ -398,6 +398,8 @@ typedef struct SyncSnapshotSend {
   SyncTerm  term;
   SyncIndex lastIndex;  // lastIndex of snapshot
   SyncTerm  lastTerm;   // lastTerm of snapshot
+  SyncIndex lastConfigIndex;
+  SSyncCfg  lastConfig;
   SyncTerm  privateTerm;
   int32_t   seq;
   uint32_t  dataLen;
@@ -456,6 +458,36 @@ void syncSnapshotRspPrint2(char* s, const SyncSnapshotRsp* pMsg);
 void syncSnapshotRspLog(const SyncSnapshotRsp* pMsg);
 void syncSnapshotRspLog2(char* s, const SyncSnapshotRsp* pMsg);
 
+// ---------------------------------------------
+typedef struct SyncLeaderTransfer {
+  uint32_t bytes;
+  int32_t  vgId;
+  uint32_t msgType;
+  /*
+   SRaftId  srcId;
+   SRaftId  destId;
+   */
+  SRaftId newLeaderId;
+} SyncLeaderTransfer;
+
+SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId);
+void                syncLeaderTransferDestroy(SyncLeaderTransfer* pMsg);
+void                syncLeaderTransferSerialize(const SyncLeaderTransfer* pMsg, char* buf, uint32_t bufLen);
+void                syncLeaderTransferDeserialize(const char* buf, uint32_t len, SyncLeaderTransfer* pMsg);
+char*               syncLeaderTransferSerialize2(const SyncLeaderTransfer* pMsg, uint32_t* len);
+SyncLeaderTransfer* syncLeaderTransferDeserialize2(const char* buf, uint32_t len);
+void                syncLeaderTransfer2RpcMsg(const SyncLeaderTransfer* pMsg, SRpcMsg* pRpcMsg);
+void                syncLeaderTransferFromRpcMsg(const SRpcMsg* pRpcMsg, SyncLeaderTransfer* pMsg);
+SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg);
+cJSON*              syncLeaderTransfer2Json(const SyncLeaderTransfer* pMsg);
+char*               syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg);
+
+// for debug ----------------------
+void syncLeaderTransferPrint(const SyncLeaderTransfer* pMsg);
+void syncLeaderTransferPrint2(char* s, const SyncLeaderTransfer* pMsg);
+void syncLeaderTransferLog(const SyncLeaderTransfer* pMsg);
+void syncLeaderTransferLog2(char* s, const SyncLeaderTransfer* pMsg);
+
 // on message ----------------------
 int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);

include/util/taoserror.h

@@ -352,9 +352,6 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TDB_NO_CACHE_LAST_ROW         TAOS_DEF_ERROR_CODE(0, 0x0619)
 #define TSDB_CODE_TDB_TABLE_RECREATED           TAOS_DEF_ERROR_CODE(0, 0x061A)
 #define TSDB_CODE_TDB_TDB_ENV_OPEN_ERROR        TAOS_DEF_ERROR_CODE(0, 0x061B)
-#define TSDB_CODE_TDB_NO_SMA_INDEX_IN_META      TAOS_DEF_ERROR_CODE(0, 0x061C)
-#define TSDB_CODE_TDB_INVALID_SMA_STAT          TAOS_DEF_ERROR_CODE(0, 0x061D)
-#define TSDB_CODE_TDB_TSMA_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x061E)
 
 // query
 #define TSDB_CODE_QRY_INVALID_QHANDLE           TAOS_DEF_ERROR_CODE(0, 0x0700)
@@ -655,6 +652,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x2654)
 #define TSDB_CODE_PAR_INVALID_DELETE_WHERE      TAOS_DEF_ERROR_CODE(0, 0x2655)
 #define TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG   TAOS_DEF_ERROR_CODE(0, 0x2656)
+#define TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC     TAOS_DEF_ERROR_CODE(0, 0x2657)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)
@@ -685,6 +683,19 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SML_INVALID_DATA              TAOS_DEF_ERROR_CODE(0, 0x3002)
 #define TSDB_CODE_SML_INVALID_DB_CONF           TAOS_DEF_ERROR_CODE(0, 0x3003)
 
+//tsma
+#define TSDB_CODE_TSMA_ALREADY_EXIST             TAOS_DEF_ERROR_CODE(0, 0x3100)
+#define TSDB_CODE_TSMA_NO_INDEX_IN_META          TAOS_DEF_ERROR_CODE(0, 0x3101)
+#define TSDB_CODE_TSMA_INVALID_ENV               TAOS_DEF_ERROR_CODE(0, 0x3102)
+#define TSDB_CODE_TSMA_INVALID_STAT              TAOS_DEF_ERROR_CODE(0, 0x3103)
+#define TSDB_CODE_TSMA_NO_INDEX_IN_CACHE         TAOS_DEF_ERROR_CODE(0, 0x3104)
+#define TSDB_CODE_TSMA_RM_SKEY_IN_HASH           TAOS_DEF_ERROR_CODE(0, 0x3105)
+
+//rsma
+#define TSDB_CODE_RSMA_INVALID_ENV               TAOS_DEF_ERROR_CODE(0, 0x3150)
+#define TSDB_CODE_RSMA_INVALID_STAT              TAOS_DEF_ERROR_CODE(0, 0x3151)
+
+
 #ifdef __cplusplus
 }
 #endif

source/client/src/clientImpl.c

@@ -689,16 +689,15 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery) {
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
 
       SAppInstInfo* pAppInfo = getAppInfo(pRequest);
-
       if (TSDB_CODE_SUCCESS == code) {
         code = qCreateQueryPlan(&cxt, &pRequest->body.pDag, pNodeList);
-        tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
       }
 
       if (TSDB_CODE_SUCCESS == code) {
         schedulerAsyncExecJob(pAppInfo->pTransporter, pNodeList, pRequest->body.pDag, &pRequest->body.queryJob,
                               pRequest->sqlstr, pRequest->metric.start, schedulerExecCb, pRequest);
       } else {
+        tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
         pRequest->body.queryFp(pRequest->body.param, pRequest, code);
       }
 

source/client/src/clientSml.c

@@ -72,7 +72,7 @@ for (int i = 1; i < keyLen; ++i) {      \
 #define NCHAR_ADD_LEN 3         // L"nchar"   3 means L" "
 
 #define MAX_RETRY_TIMES 5
-#define LINE_BATCH  20
+#define LINE_BATCH  20000
 //=================================================================================================
 typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
 
@@ -161,7 +161,6 @@ typedef struct {
 
 typedef struct{
   SRequestObj*    request;
-  SCatalog*       catalog;
   tsem_t          sem;
   TdThreadSpinlock lock;
 } Params;
@@ -1292,9 +1291,46 @@ static void smlDestroyTableInfo(SSmlHandle* info, SSmlTableInfo *tag){
   taosMemoryFree(tag);
 }
 
+static int32_t smlKvTimeArrayCompare(const void* key1, const void* key2) {
+  SArray *s1 = *(SArray **)key1;
+  SArray *s2 = *(SArray **)key2;
+  SSmlKv *kv1 = (SSmlKv *)taosArrayGetP(s1, 0);
+  SSmlKv *kv2 = (SSmlKv *)taosArrayGetP(s2, 0);
+  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
+  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
+  if (kv1->i < kv2->i) {
+    return -1;
+  } else if (kv1->i > kv2->i) {
+    return 1;
+  } else {
+    return 0;
+  }
+}
+
+static int32_t smlKvTimeHashCompare(const void* key1, const void* key2) {
+  SHashObj *s1 = *(SHashObj **)key1;
+  SHashObj *s2 = *(SHashObj **)key2;
+  SSmlKv *kv1 = (SSmlKv *)taosHashGet(s1, TS, TS_LEN);
+  SSmlKv *kv2 = (SSmlKv *)taosHashGet(s2, TS, TS_LEN);
+  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
+  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
+  if (kv1->i < kv2->i) {
+    return -1;
+  } else if (kv1->i > kv2->i) {
+    return 1;
+  } else {
+    return 0;
+  }
+}
+
 static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
   if(dataFormat){
+    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GE);
+    if(p == NULL){
       taosArrayPush(oneTable->cols, &cols);
+    }else{
+      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
+    }
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1307,8 +1343,13 @@ static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *col
     SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
     taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
   }
-  taosArrayPush(oneTable->cols, &kvHash);
 
+  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GE);
+  if(p == NULL){
+    taosArrayPush(oneTable->cols, &kvHash);
+  }else{
+    taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
+  }
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1419,6 +1460,11 @@ static SSmlHandle* smlBuildSmlInfo(TAOS* taos, SRequestObj* request, SMLProtocol
   ((SVnodeModifOpStmt*)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;
 
   info->taos        = (STscObj *)taos;
+  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
+  if(code != TSDB_CODE_SUCCESS){
+    uError("SML:0x%"PRIx64" get catalog error %d", info->id, code);
+    goto cleanup;
+  }
 
   info->precision   = precision;
   info->protocol    = protocol;
@@ -2196,6 +2242,7 @@ static int32_t smlInsertData(SSmlHandle* info) {
     code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
                        (*pMeta)->tableMeta, tableData->childTableName, info->msgBuf.buf, info->msgBuf.len);
     if(code != TSDB_CODE_SUCCESS){
+      uError("SML:0x%"PRIx64" smlBindData failed", info->id);
       return code;
     }
     oneTable = (SSmlTableInfo**)taosHashIterate(info->childTables, oneTable);
@@ -2261,7 +2308,7 @@ static int smlProcess(SSmlHandle *info, char* lines[], int numLines) {
   code = smlParseLine(info, lines, numLines);
   if (code != 0) {
     uError("SML:0x%"PRIx64" smlParseLine error : %s", info->id, tstrerror(code));
-    goto cleanup;
+    return code;
   }
 
   info->cost.lineNum = numLines;
@@ -2277,24 +2324,27 @@ static int smlProcess(SSmlHandle *info, char* lines[], int numLines) {
 
   if (code != 0) {
     uError("SML:0x%"PRIx64" smlModifyDBSchemas error : %s", info->id, tstrerror(code));
-    goto cleanup;
+    return code;
   }
 
   info->cost.insertBindTime = taosGetTimestampUs();
   code = smlInsertData(info);
   if (code != 0) {
     uError("SML:0x%"PRIx64" smlInsertData error : %s", info->id, tstrerror(code));
-    goto cleanup;
+    return code;
   }
-  info->cost.endTime = taosGetTimestampUs();
 
-cleanup:
-  info->cost.code = code;
-  smlPrintStatisticInfo(info);
   return code;
 }
 
-static int32_t isSchemalessDb(STscObj *taos, SCatalog *catalog){
+static int32_t isSchemalessDb(STscObj *taos){
+  SCatalog* catalog = NULL;
+  int32_t code = catalogGetHandle(((STscObj *)taos)->pAppInfo->clusterId, &catalog);
+  if(code != TSDB_CODE_SUCCESS){
+    uError("SML get catalog error %d", code);
+    return code;
+  }
+
   SName          name;
   tNameSetDbName(&name, taos->acctId, taos->db, strlen(taos->db));
   char dbFname[TSDB_DB_FNAME_LEN] = {0};
@@ -2302,7 +2352,7 @@ static int32_t isSchemalessDb(STscObj *taos, SCatalog *catalog){
   SDbCfgInfo pInfo = {0};
   SEpSet ep = getEpSet_s(&taos->pAppInfo->mgmtEp);
 
-  int32_t code = catalogGetDBCfg(catalog, taos->pAppInfo->pTransporter, &ep, dbFname, &pInfo);
+  code = catalogGetDBCfg(catalog, taos->pAppInfo->pTransporter, &ep, dbFname, &pInfo);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
@@ -2329,6 +2379,9 @@ static void smlInsertCallback(void* param, void* res, int32_t code) {
   printf("SML:0x%" PRIx64 " insert finished, code: %d, total: %d\n", info->id, code, info->affectedRows);
   Params *pParam = info->params;
   bool isLast = info->isLast;
+  info->cost.endTime = taosGetTimestampUs();
+  info->cost.code = code;
+  smlPrintStatisticInfo(info);
   smlDestroyInfo(info);
 
   if(isLast){
@@ -2373,20 +2426,13 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
   tsem_init(&params.sem, 0, 0);
   taosThreadSpinInit(&(params.lock), 0);
 
-  int32_t code = catalogGetHandle(((STscObj *)taos)->pAppInfo->clusterId, &params.catalog);
-  if(code != TSDB_CODE_SUCCESS){
-    uError("SML get catalog error %d", code);
-    request->code = code;
-    goto end;
-  }
-
   if(request->pDb == NULL){
     request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
     smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
     goto end;
   }
 
-  if(isSchemalessDb(((STscObj *)taos), params.catalog) != TSDB_CODE_SUCCESS){
+  if(isSchemalessDb(((STscObj *)taos)) != TSDB_CODE_SUCCESS){
     request->code = TSDB_CODE_SML_INVALID_DB_CONF;
     smlBuildInvalidDataMsg(&msg, "Cannot write data to a non schemaless database", NULL);
     goto end;
@@ -2436,11 +2482,10 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
     }
 
     info->params = &params;
-    info->pCatalog = params.catalog;
     info->affectedRows = perBatch;
     info->pRequest->body.queryFp = smlInsertCallback;
     info->pRequest->body.param   = info;
-    code = smlProcess(info, lines, perBatch);
+    int32_t code = smlProcess(info, lines, perBatch);
     lines += perBatch;
     if (code != TSDB_CODE_SUCCESS){
       info->pRequest->body.queryFp(info, req, code);

source/client/test/clientTests.cpp

@@ -778,9 +778,9 @@ TEST(testCase, async_api_test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(pConn, nullptr);
 
-  taos_query(pConn, "use test");
+  taos_query(pConn, "use nest");
 
-  TAOS_RES* pRes = taos_query(pConn, "desc abc1.tu");
+  TAOS_RES* pRes = taos_query(pConn, "select NOW() from (select * from regular_table_2 where tbname in ('regular_table_2_1') and  q_bigint <= 9223372036854775807 and  q_tinyint <= 127 and    q_bool in ( true , false) ) order by ts;");
   if (taos_errno(pRes) != 0) {
     printf("failed, reason:%s\n", taos_errstr(pRes));
   }

source/common/src/tdatablock.c

@@ -294,7 +294,7 @@ int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, in
 
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows) {
   ASSERT(pColumnInfoData != NULL && pSource != NULL && pColumnInfoData->info.type == pSource->info.type);
-  if (numOfRows == 0) {
+  if (numOfRows <= 0) {
     return numOfRows;
   }
 
@@ -1239,6 +1239,9 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
         return NULL;
       }
 
+      if (pSrc->pData == NULL) {
+        continue;
+      }
       colDataAssign(pDst, pSrc, pDataBlock->info.rows);
     }
 
@@ -1631,6 +1634,10 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
             break;
           default:
             if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
+              if (pCol->type == pColInfoData->info.type) {
+                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, var, true, offset,
+                                    k);
+              } else {
                 char tv[8] = {0};
                 if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
                   float v = 0;
@@ -1649,7 +1656,9 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
                   GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
                   SET_TYPED_DATA(&tv, pCol->type, v);
                 }
-              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset, k);
+                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset,
+                                    k);
+              }
             } else {
               uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
               TASSERT(0);

source/common/src/tglobal.c

@@ -86,6 +86,7 @@ bool tsSmlDataFormat =
 
 // query
 int32_t tsQueryPolicy = 1;
+int32_t tsQuerySmaOptimize = 1;
 
 /*
  * denote if the server needs to compress response message at the application layer to client, including query rsp,
@@ -331,6 +332,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "compressColData", tsCompressColData, -1, 100000000, 1) != 0) return -1;
   if (cfgAddBool(pCfg, "keepColumnName", tsKeepOriginalColumnName, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 3, 1) != 0) return -1;
+  if (cfgAddInt32(pCfg, "querySmaOptimize", tsQuerySmaOptimize, 0, 1, 1) != 0) return -1;
   if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1;
   if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1;
   if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1;
@@ -541,6 +543,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   tsKeepOriginalColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
   tsNumOfTaskQueueThreads = cfgGetItem(pCfg, "numOfTaskQueueThreads")->i32;
   tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32;
+  tsQuerySmaOptimize = cfgGetItem(pCfg, "querySmaOptimize")->i32;
   return 0;
 }
 

source/common/src/tmsg.c

@@ -3877,9 +3877,10 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
   if (tEncodeCStr(pCoder, pSma->indexName) < 0) return -1;
   if (tEncodeI32(pCoder, pSma->exprLen) < 0) return -1;
   if (tEncodeI32(pCoder, pSma->tagsFilterLen) < 0) return -1;
-  if (tEncodeI32(pCoder, pSma->numOfVgroups) < 0) return -1;
   if (tEncodeI64(pCoder, pSma->indexUid) < 0) return -1;
   if (tEncodeI64(pCoder, pSma->tableUid) < 0) return -1;
+  if (tEncodeI64(pCoder, pSma->dstTbUid) < 0) return -1;
+  if (tEncodeCStr(pCoder, pSma->dstTbName) < 0) return -1;
   if (tEncodeI64(pCoder, pSma->interval) < 0) return -1;
   if (tEncodeI64(pCoder, pSma->offset) < 0) return -1;
   if (tEncodeI64(pCoder, pSma->sliding) < 0) return -1;
@@ -3889,17 +3890,10 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
   if (pSma->tagsFilterLen > 0) {
     if (tEncodeCStr(pCoder, pSma->tagsFilter) < 0) return -1;
   }
-  for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
-    if (tEncodeI32(pCoder, pSma->pVgEpSet[v].vgId) < 0) return -1;
-    if (tEncodeI8(pCoder, pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
-    int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
-    if (tEncodeI8(pCoder, numOfEps) < 0) return -1;
-    for (int32_t n = 0; n < numOfEps; ++n) {
-      const SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
-      if (tEncodeCStr(pCoder, pEp->fqdn) < 0) return -1;
-      if (tEncodeU16(pCoder, pEp->port) < 0) return -1;
-    }
-  }
+
+  tEncodeSSchemaWrapper(pCoder, &pSma->schemaRow);
+  tEncodeSSchemaWrapper(pCoder, &pSma->schemaTag);
+
   return 0;
 }
 
@@ -3907,14 +3901,15 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
   if (tDecodeI8(pCoder, &pSma->version) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->intervalUnit) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->slidingUnit) < 0) return -1;
-  if (tDecodeI32(pCoder, &pSma->dstVgId) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->timezoneInt) < 0) return -1;
+  if (tDecodeI32(pCoder, &pSma->dstVgId) < 0) return -1;
   if (tDecodeCStrTo(pCoder, pSma->indexName) < 0) return -1;
   if (tDecodeI32(pCoder, &pSma->exprLen) < 0) return -1;
   if (tDecodeI32(pCoder, &pSma->tagsFilterLen) < 0) return -1;
-  if (tDecodeI32(pCoder, &pSma->numOfVgroups) < 0) return -1;
   if (tDecodeI64(pCoder, &pSma->indexUid) < 0) return -1;
   if (tDecodeI64(pCoder, &pSma->tableUid) < 0) return -1;
+  if (tDecodeI64(pCoder, &pSma->dstTbUid) < 0) return -1;
+  if (tDecodeCStr(pCoder, &pSma->dstTbName) < 0) return -1;
   if (tDecodeI64(pCoder, &pSma->interval) < 0) return -1;
   if (tDecodeI64(pCoder, &pSma->offset) < 0) return -1;
   if (tDecodeI64(pCoder, &pSma->sliding) < 0) return -1;
@@ -3928,27 +3923,9 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
   } else {
     pSma->tagsFilter = NULL;
   }
-  if (pSma->numOfVgroups > 0) {
-    pSma->pVgEpSet = (SVgEpSet *)tDecoderMalloc(pCoder, pSma->numOfVgroups * sizeof(SVgEpSet));
-    if (!pSma->pVgEpSet) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
-
-    memset(pSma->pVgEpSet, 0, pSma->numOfVgroups * sizeof(SVgEpSet));
-
-    for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
-      if (tDecodeI32(pCoder, &pSma->pVgEpSet[v].vgId) < 0) return -1;
-      if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
-      if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.numOfEps) < 0) return -1;
-      int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
-      for (int32_t n = 0; n < numOfEps; ++n) {
-        SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
-        if (tDecodeCStrTo(pCoder, pEp->fqdn) < 0) return -1;
-        if (tDecodeU16(pCoder, &pEp->port) < 0) return -1;
-      }
-    }
-  }
+  // only needed in dstVgroup
+  tDecodeSSchemaWrapperEx(pCoder, &pSma->schemaRow);
+  tDecodeSSchemaWrapperEx(pCoder, &pSma->schemaTag);
 
   return 0;
 }
@@ -3991,55 +3968,6 @@ int32_t tDecodeSVDropTSmaReq(SDecoder *pCoder, SVDropTSmaReq *pReq) {
   return 0;
 }
 
-int32_t tEncodeSVGetTSmaExpWndsReq(SEncoder *pCoder, const SVGetTsmaExpWndsReq *pReq) {
-  if (tStartEncode(pCoder) < 0) return -1;
-
-  if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1;
-  if (tEncodeI64(pCoder, pReq->queryWindow.skey) < 0) return -1;
-  if (tEncodeI64(pCoder, pReq->queryWindow.ekey) < 0) return -1;
-
-  tEndEncode(pCoder);
-  return 0;
-}
-
-int32_t tDecodeSVGetTsmaExpWndsReq(SDecoder *pCoder, SVGetTsmaExpWndsReq *pReq) {
-  if (tStartDecode(pCoder) < 0) return -1;
-
-  if (tDecodeI64(pCoder, &pReq->indexUid) < 0) return -1;
-  if (tDecodeI64(pCoder, &pReq->queryWindow.skey) < 0) return -1;
-  if (tDecodeI64(pCoder, &pReq->queryWindow.ekey) < 0) return -1;
-
-  tEndDecode(pCoder);
-  return 0;
-}
-
-int32_t tEncodeSVGetTSmaExpWndsRsp(SEncoder *pCoder, const SVGetTsmaExpWndsRsp *pReq) {
-  if (tStartEncode(pCoder) < 0) return -1;
-
-  if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1;
-  if (tEncodeI8(pCoder, pReq->flags) < 0) return -1;
-  if (tEncodeI32(pCoder, pReq->numExpWnds) < 0) return -1;
-  for (int32_t i = 0; i < pReq->numExpWnds; ++i) {
-    if (tEncodeI64(pCoder, pReq->wndSKeys[i]) < 0) return -1;
-  }
-  tEndEncode(pCoder);
-  return 0;
-}
-
-int32_t tDecodeSVGetTsmaExpWndsRsp(SDecoder *pCoder, SVGetTsmaExpWndsRsp *pReq) {
-  if (tStartDecode(pCoder) < 0) return -1;
-
-  if (tDecodeI64(pCoder, &pReq->indexUid) < 0) return -1;
-  if (tDecodeI8(pCoder, &pReq->flags) < 0) return -1;
-  if (tDecodeI32(pCoder, &pReq->numExpWnds) < 0) return -1;
-  for (int32_t i = 0; i < pReq->numExpWnds; ++i) {
-    if (tDecodeI64(pCoder, &pReq->wndSKeys[i]) < 0) return -1;
-  }
-
-  tEndDecode(pCoder);
-  return 0;
-}
-
 int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) {
   int32_t headLen = sizeof(SMsgHead);
   if (buf != NULL) {

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -38,10 +38,14 @@ static void dmProcessStatusRsp(SDnodeMgmt *pMgmt, SRpcMsg *pRsp) {
     SStatusRsp statusRsp = {0};
     if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
         tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
+      dTrace("status msg received from mnode, dnodeVer:%" PRId64 " saved:%" PRId64, statusRsp.dnodeVer,
+             pMgmt->pData->dnodeVer);
+      if (pMgmt->pData->dnodeVer != statusRsp.dnodeVer) {
         pMgmt->pData->dnodeVer = statusRsp.dnodeVer;
         dmUpdateDnodeCfg(pMgmt, &statusRsp.dnodeCfg);
         dmUpdateEps(pMgmt->pData, statusRsp.pDnodeEps);
       }
+    }
     rpcFreeCont(pRsp->pCont);
     tFreeSStatusRsp(&statusRsp);
   }
@@ -89,7 +93,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .info.ahandle = (void *)0x9527};
   SRpcMsg rpcRsp = {0};
 
-  dTrace("send status msg to mnode");
+  dTrace("send status msg to mnode, dnodeVer:%" PRId64, req.dnodeVer);
 
   SEpSet epSet = {0};
   dmGetMnodeEpSet(pMgmt->pData, &epSet);

source/dnode/mgmt/node_util/src/dmEps.c

@@ -81,6 +81,13 @@ int32_t dmReadEps(SDnodeData *pData) {
   }
   pData->dnodeId = dnodeId->valueint;
 
+  cJSON *dnodeVer = cJSON_GetObjectItem(root, "dnodeVer");
+  if (!dnodeVer || dnodeVer->type != cJSON_String) {
+    dError("failed to read %s since dnodeVer not found", file);
+    goto _OVER;
+  }
+  pData->dnodeVer = atoll(dnodeVer->valuestring);
+
   cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
   if (!clusterId || clusterId->type != cJSON_String) {
     dError("failed to read %s since clusterId not found", file);
@@ -193,6 +200,7 @@ int32_t dmWriteEps(SDnodeData *pData) {
 
   len += snprintf(content + len, maxLen - len, "{\n");
   len += snprintf(content + len, maxLen - len, "  \"dnodeId\": %d,\n", pData->dnodeId);
+  len += snprintf(content + len, maxLen - len, "  \"dnodeVer\": \"%" PRId64 "\",\n", pData->dnodeVer);
   len += snprintf(content + len, maxLen - len, "  \"clusterId\": \"%" PRId64 "\",\n", pData->clusterId);
   len += snprintf(content + len, maxLen - len, "  \"dropped\": %d,\n", pData->dropped);
   len += snprintf(content + len, maxLen - len, "  \"dnodes\": [{\n");
@@ -224,30 +232,15 @@ int32_t dmWriteEps(SDnodeData *pData) {
   }
 
   pData->updateTime = taosGetTimestampMs();
-  dDebug("successed to write %s", realfile);
+  dDebug("successed to write %s, dnodeVer:%" PRId64, realfile, pData->dnodeVer);
   return 0;
 }
 
 void dmUpdateEps(SDnodeData *pData, SArray *eps) {
-  int32_t numOfEps = taosArrayGetSize(eps);
-  if (numOfEps <= 0) return;
-
   taosThreadRwlockWrlock(&pData->lock);
-
-  int32_t numOfEpsOld = (int32_t)taosArrayGetSize(pData->dnodeEps);
-  if (numOfEps != numOfEpsOld) {
-    dDebug("new dnode list get from mnode");
+  dDebug("new dnode list get from mnode, dnodeVer:%" PRId64, pData->dnodeVer);
   dmResetEps(pData, eps);
   dmWriteEps(pData);
-  } else {
-    int32_t size = numOfEps * sizeof(SDnodeEp);
-    if (memcmp(pData->dnodeEps->pData, eps->pData, size) != 0) {
-      dDebug("new dnode list get from mnode");
-      dmResetEps(pData, eps);
-      dmWriteEps(pData);
-    }
-  }
-
   taosThreadRwlockUnlock(&pData->lock);
 }
 

source/dnode/mnode/impl/inc/mndDef.h

@@ -301,15 +301,16 @@ typedef struct {
   char           name[TSDB_TABLE_FNAME_LEN];
   char           stb[TSDB_TABLE_FNAME_LEN];
   char           db[TSDB_DB_FNAME_LEN];
+  char           dstTbName[TSDB_TABLE_FNAME_LEN];
   int64_t        createdTime;
   int64_t        uid;
   int64_t        stbUid;
   int64_t        dbUid;
+  int64_t        dstTbUid;
   int8_t         intervalUnit;
   int8_t         slidingUnit;
   int8_t         timezone;
   int32_t        dstVgId;  // for stream
-  int64_t   dstTbUid;
   int64_t        interval;
   int64_t        offset;
   int64_t        sliding;
@@ -317,12 +318,12 @@ typedef struct {
   int32_t        tagsFilterLen;
   int32_t        sqlLen;
   int32_t        astLen;
-  int32_t   numOfVgroups;
   char*          expr;
   char*          tagsFilter;
   char*          sql;
   char*          ast;
-  SVgEpSet* pVgEpSet;
+  SSchemaWrapper schemaRow;  // for dstVgroup
+  SSchemaWrapper schemaTag;  // for dstVgroup
 } SSmaObj;
 
 typedef struct {

source/dnode/mnode/impl/inc/mndScheduler.h

@@ -27,10 +27,10 @@ void    mndCleanupScheduler(SMnode* pMnode);
 
 int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub);
 
-int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
+int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
 
-int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr,
-                           int32_t* pLen, double filesFactor);
+int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType,
+                           int64_t watermark, double filesFactor);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/impl/inc/mndTrans.h

@@ -75,6 +75,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
 int32_t mndTransProcessRsp(SRpcMsg *pRsp);
 void    mndTransPullup(SMnode *pMnode);
 int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans);
+void    mndTransExecute(SMnode *pMnode, STrans *pTrans);
 
 #ifdef __cplusplus
 }

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

@@ -447,6 +447,8 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
 
   pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
   if (pConsumerOld == NULL) {
+    mInfo("receive subscribe request from new consumer: %ld", consumerId);
+
     pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup);
     tstrncpy(pConsumerNew->clientId, subscribe.clientId, 256);
     pConsumerNew->updateType = CONSUMER_UPDATE__MODIFY;
@@ -463,7 +465,12 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
 
   } else {
     /*taosRLockLatch(&pConsumerOld->lock);*/
+
     int32_t status = atomic_load_32(&pConsumerOld->status);
+
+    mInfo("receive subscribe request from old consumer: %ld, current status: %s", consumerId,
+          mndConsumerStatusName(status));
+
     if (status != MQ_CONSUMER_STATUS__READY) {
       terrno = TSDB_CODE_MND_CONSUMER_NOT_READY;
       goto SUBSCRIBE_OVER;

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

@@ -1424,10 +1424,10 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
     char    tmp[128] = {0};
     int32_t len = 0;
     if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) {
-      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%d,%d,%d", pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2,
+      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2,
                     pDb->cfg.daysToKeep0);
     } else {
-      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%d,%d,%d", pDb->cfg.daysToKeep0, pDb->cfg.daysToKeep1,
+      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep0, pDb->cfg.daysToKeep1,
                     pDb->cfg.daysToKeep2);
     }
 
@@ -1592,4 +1592,3 @@ static void mndCancelGetNextDb(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
-

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

@@ -385,7 +385,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
   int64_t dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE) + sdbGetTableVer(pMnode->pSdb, SDB_MNODE);
   int64_t curMs = taosGetTimestampMs();
   bool    online = mndIsDnodeOnline(pDnode, curMs);
-  bool    dnodeChanged = (statusReq.dnodeVer != dnodeVer);
+  bool    dnodeChanged = (statusReq.dnodeVer == 0) || (statusReq.dnodeVer != dnodeVer);
   bool    reboot = (pDnode->rebootTime != statusReq.rebootTime);
   bool    needCheck = !online || dnodeChanged || reboot;
 
@@ -427,7 +427,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
     if (!online) {
       mInfo("dnode:%d, from offline to online", pDnode->id);
     } else {
-      mDebug("dnode:%d, send dnode epset, online:%d dnode_ver:%" PRId64 ":%" PRId64 " reboot:%d", pDnode->id, online,
+      mDebug("dnode:%d, send dnode epset, online:%d dnodeVer:%" PRId64 ":%" PRId64 " reboot:%d", pDnode->id, online,
              statusReq.dnodeVer, dnodeVer, reboot);
     }
 

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

@@ -183,7 +183,10 @@ static int32_t mndProcessCommitOffsetReq(SRpcMsg *pMsg) {
 
   for (int32_t i = 0; i < commitOffsetReq.num; i++) {
     SMqOffset *pOffset = &commitOffsetReq.offsets[i];
+    mInfo("commit offset %ld to vg %d of consumer group %s on topic %s", pOffset->offset, pOffset->vgId,
+          pOffset->cgroup, pOffset->topicName);
     if (mndMakePartitionKey(key, pOffset->cgroup, pOffset->topicName, pOffset->vgId) < 0) {
+      mError("submit offset to topic %s failed", pOffset->topicName);
       return -1;
     }
     bool          create = false;

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

@@ -42,8 +42,8 @@ static int32_t mndAddTaskToTaskSet(SArray* pArray, SStreamTask* pTask) {
   return 0;
 }
 
-int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr,
-                           int32_t* pLen, double filesFactor) {
+int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType,
+                           int64_t watermark, double filesFactor) {
   SNode*      pAst = NULL;
   SQueryPlan* pPlan = NULL;
   terrno = TSDB_CODE_SUCCESS;
@@ -53,7 +53,7 @@ int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int
     goto END;
   }
 
-  if (qSetSTableIdForRSma(pAst, uid) < 0) {
+  if (qSetSTableIdForRsma(pAst, uid) < 0) {
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     goto END;
   }
@@ -86,7 +86,7 @@ int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int
   }
 
   SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
-  if (qSubPlanToString(plan, pStr, pLen) < 0) {
+  if (qSubPlanToString(plan, pDst, pDstLen) < 0) {
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     goto END;
   }
@@ -151,6 +151,9 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
     ASSERT(pDb);
 
     if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
+      ASSERT(0);
+      return -1;
+    }
     sdbRelease(pMnode->pSdb, pDb);
 
     SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
@@ -163,6 +166,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
         SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
         if (pLastLevelTask->nodeId == pVgInfo->vgId) {
           pVgInfo->taskId = pLastLevelTask->taskId;
+          ASSERT(pVgInfo->taskId != 0);
           break;
         }
       }
@@ -178,7 +182,6 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj*
     pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId;
     pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet;
   }
-  }
   return 0;
 }
 
@@ -335,7 +338,7 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj*
   return 0;
 }
 
-int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
+int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
   SSdb*       pSdb = pMnode->pSdb;
   SQueryPlan* pPlan = qStringToQueryPlan(pStream->physicalPlan);
   if (pPlan == NULL) {
@@ -361,6 +364,7 @@ int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream)
       mndAddFixedSinkTaskToStream(pMnode, pTrans, pStream);
     }
   }
+
   if (totLevel > 1) {
     SStreamTask* pFinalTask;
     // inner plan
@@ -378,7 +382,10 @@ int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream)
       pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
 
       // dispatch
-      mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask);
+      if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) {
+        qDestroyQueryPlan(pPlan);
+        return -1;
+      }
 
       // exec
       pFinalTask->execType = TASK_EXEC__PIPE;
@@ -472,229 +479,7 @@ int32_t mndScheduleStream1(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream)
       }
     }
   }
-  return 0;
-}
-
-int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
-  SSdb*       pSdb = pMnode->pSdb;
-  SQueryPlan* pPlan = qStringToQueryPlan(pStream->physicalPlan);
-  if (pPlan == NULL) {
-    terrno = TSDB_CODE_QRY_INVALID_INPUT;
-    return -1;
-  }
-  ASSERT(pStream->vgNum == 0);
-
-  int32_t totLevel = LIST_LENGTH(pPlan->pSubplans);
-  ASSERT(totLevel <= 2);
-  pStream->tasks = taosArrayInit(totLevel, sizeof(void*));
-
-  bool hasExtraSink = false;
-  bool externalTargetDB = strcmp(pStream->sourceDb, pStream->targetDb) != 0;
-  if (totLevel == 2 || externalTargetDB) {
-    SArray* taskOneLevel = taosArrayInit(0, sizeof(void*));
-    taosArrayPush(pStream->tasks, &taskOneLevel);
-    // add extra sink
-    hasExtraSink = true;
-    if (pStream->fixedSinkVgId == 0) {
-      mndAddShuffleSinkTasksToStream(pMnode, pTrans, pStream);
-    } else {
-      mndAddFixedSinkTaskToStream(pMnode, pTrans, pStream);
-    }
-  }
-
-  for (int32_t level = 0; level < totLevel; level++) {
-    SArray* taskOneLevel = taosArrayInit(0, sizeof(void*));
-    taosArrayPush(pStream->tasks, &taskOneLevel);
-    SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, level);
-    ASSERT(LIST_LENGTH(inner->pNodeList) == 1);
-
-    SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
-
-    // if (level == totLevel - 1 /* or no snode */) {
-    if (level == totLevel - 1) {
-      // last level, source, must assign to vnode
-      // must be scan type
-      ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
-
-      // replicate task to each vnode
-      void* pIter = NULL;
-      while (1) {
-        SVgObj* pVgroup;
-        pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
-        if (pIter == NULL) break;
-        if (pVgroup->dbUid != pStream->dbUid) {
-          sdbRelease(pSdb, pVgroup);
-          continue;
-        }
-        SStreamTask* pTask = tNewSStreamTask(pStream->uid);
-        mndAddTaskToTaskSet(taskOneLevel, pTask);
-        // source part
-        pTask->sourceType = TASK_SOURCE__SCAN;
-        pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
-
-        // sink part
-        if (level == 0) {
-          // only for inplace
-          pTask->sinkType = TASK_SINK__NONE;
-          if (!hasExtraSink) {
-#if 1
-            if (pStream->createdBy == STREAM_CREATED_BY__SMA) {
-              pTask->sinkType = TASK_SINK__SMA;
-              pTask->smaSink.smaId = pStream->smaId;
-            } else {
-              pTask->sinkType = TASK_SINK__TABLE;
-              pTask->tbSink.stbUid = pStream->targetStbUid;
-              memcpy(pTask->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN);
-              pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
-            }
-#endif
-          }
-        } else {
-          pTask->sinkType = TASK_SINK__NONE;
-        }
-
-        // dispatch part
-        if (level == 0 && !hasExtraSink) {
-          pTask->dispatchType = TASK_DISPATCH__NONE;
-        } else {
-          // add fixed ep dispatcher
-          int32_t lastLevel = level - 1;
-          if (hasExtraSink) lastLevel++;
-          ASSERT(lastLevel == 0);
-          SArray* pArray = taosArrayGetP(pStream->tasks, lastLevel);
-          // one merge only
-          ASSERT(taosArrayGetSize(pArray) == 1);
-          SStreamTask* lastLevelTask = taosArrayGetP(pArray, 0);
-          pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
-          pTask->dispatchType = TASK_DISPATCH__FIXED;
-
-          pTask->fixedEpDispatcher.taskId = lastLevelTask->taskId;
-          pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId;
-          pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet;
-        }
-
-        // exec part
-        pTask->execType = TASK_EXEC__PIPE;
-        if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
-          sdbRelease(pSdb, pVgroup);
-          qDestroyQueryPlan(pPlan);
-          return -1;
-        }
-        sdbRelease(pSdb, pVgroup);
-      }
-    } else {
-      // merge plan
-
-      // TODO if has snode, assign to snode
-
-      // else, assign to vnode
-      ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
-      SStreamTask* pTask = tNewSStreamTask(pStream->uid);
-      mndAddTaskToTaskSet(taskOneLevel, pTask);
-
-      // source part, currently only support multi source
-      pTask->sourceType = TASK_SOURCE__PIPE;
-      pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
-
-      // sink part
-      pTask->sinkType = TASK_SINK__NONE;
-
-      // dispatch part
-      ASSERT(hasExtraSink);
-      /*pTask->dispatchType = TASK_DISPATCH__NONE;*/
-#if 1
-
-      if (hasExtraSink) {
-        // add dispatcher
-        if (pStream->fixedSinkVgId == 0) {
-          pTask->dispatchType = TASK_DISPATCH__SHUFFLE;
-
-          pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
-          SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
-          ASSERT(pDb);
-          if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
-            sdbRelease(pSdb, pDb);
-            qDestroyQueryPlan(pPlan);
-            return -1;
-          }
-          sdbRelease(pSdb, pDb);
-
-          // put taskId to useDbRsp
-          // TODO: optimize
-          SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
-          int32_t sz = taosArrayGetSize(pVgs);
-          SArray* sinkLv = taosArrayGetP(pStream->tasks, 0);
-          int32_t sinkLvSize = taosArrayGetSize(sinkLv);
-          for (int32_t i = 0; i < sz; i++) {
-            SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);
-            for (int32_t j = 0; j < sinkLvSize; j++) {
-              SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
-              if (pLastLevelTask->nodeId == pVgInfo->vgId) {
-                pVgInfo->taskId = pLastLevelTask->taskId;
-                break;
-              }
-            }
-          }
-        } else {
-          pTask->dispatchType = TASK_DISPATCH__FIXED;
-          /*pTask->dispatchMsgType = TDMT_VND_TASK_WRITE_EXEC;*/
-          pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
-          SArray* pArray = taosArrayGetP(pStream->tasks, 0);
-          // one sink only
-          ASSERT(taosArrayGetSize(pArray) == 1);
-          SStreamTask* lastLevelTask = taosArrayGetP(pArray, 0);
-          pTask->fixedEpDispatcher.taskId = lastLevelTask->taskId;
-          pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId;
-          pTask->fixedEpDispatcher.epSet = lastLevelTask->epSet;
-        }
-      }
-#endif
-
-      // exec part
-      pTask->execType = TASK_EXEC__MERGE;
-      SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->dbUid);
-      ASSERT(pVgroup);
-      if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
-        sdbRelease(pSdb, pVgroup);
   qDestroyQueryPlan(pPlan);
-        return -1;
-      }
-      sdbRelease(pSdb, pVgroup);
-    }
-  }
-
-#if 0
-  if (totLevel == 2) {
-    void* pIter = NULL;
-    while (1) {
-      SVgObj* pVgroup;
-      pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
-      if (pIter == NULL) break;
-      if (pVgroup->dbUid != pStream->dbUid) {
-        sdbRelease(pSdb, pVgroup);
-        continue;
-      }
-      SStreamTask* pTask = tNewSStreamTask(pStream->uid);
-
-      // source part
-      pTask->sourceType = TASK_SOURCE__MERGE;
-      pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
-
-      // sink part
-      pTask->sinkType = TASK_SINK__NONE;
-
-      // dispatch part
-      pTask->dispatchType = TASK_DISPATCH__NONE;
-
-      // exec part
-      pTask->execType = TASK_EXEC__NONE;
-    }
-  }
-#endif
-
-  // free memory
-  qDestroyQueryPlan(pPlan);
-
   return 0;
 }
 

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

@@ -26,6 +26,7 @@
 #include "mndTrans.h"
 #include "mndUser.h"
 #include "mndVgroup.h"
+#include "parser.h"
 #include "tname.h"
 
 #define TSDB_SMA_VER_NUMBER   1
@@ -82,10 +83,12 @@ static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
   SDB_SET_BINARY(pRaw, dataPos, pSma->name, TSDB_TABLE_FNAME_LEN, _OVER)
   SDB_SET_BINARY(pRaw, dataPos, pSma->stb, TSDB_TABLE_FNAME_LEN, _OVER)
   SDB_SET_BINARY(pRaw, dataPos, pSma->db, TSDB_DB_FNAME_LEN, _OVER)
+  SDB_SET_BINARY(pRaw, dataPos, pSma->dstTbName, TSDB_DB_FNAME_LEN, _OVER)
   SDB_SET_INT64(pRaw, dataPos, pSma->createdTime, _OVER)
   SDB_SET_INT64(pRaw, dataPos, pSma->uid, _OVER)
   SDB_SET_INT64(pRaw, dataPos, pSma->stbUid, _OVER)
   SDB_SET_INT64(pRaw, dataPos, pSma->dbUid, _OVER)
+  SDB_SET_INT64(pRaw, dataPos, pSma->dstTbUid, _OVER)
   SDB_SET_INT8(pRaw, dataPos, pSma->intervalUnit, _OVER)
   SDB_SET_INT8(pRaw, dataPos, pSma->slidingUnit, _OVER)
   SDB_SET_INT8(pRaw, dataPos, pSma->timezone, _OVER)
@@ -147,10 +150,12 @@ static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw) {
   SDB_GET_BINARY(pRaw, dataPos, pSma->name, TSDB_TABLE_FNAME_LEN, _OVER)
   SDB_GET_BINARY(pRaw, dataPos, pSma->stb, TSDB_TABLE_FNAME_LEN, _OVER)
   SDB_GET_BINARY(pRaw, dataPos, pSma->db, TSDB_DB_FNAME_LEN, _OVER)
+  SDB_GET_BINARY(pRaw, dataPos, pSma->dstTbName, TSDB_DB_FNAME_LEN, _OVER)
   SDB_GET_INT64(pRaw, dataPos, &pSma->createdTime, _OVER)
   SDB_GET_INT64(pRaw, dataPos, &pSma->uid, _OVER)
   SDB_GET_INT64(pRaw, dataPos, &pSma->stbUid, _OVER)
   SDB_GET_INT64(pRaw, dataPos, &pSma->dbUid, _OVER)
+  SDB_GET_INT64(pRaw, dataPos, &pSma->dstTbUid, _OVER)
   SDB_GET_INT8(pRaw, dataPos, &pSma->intervalUnit, _OVER)
   SDB_GET_INT8(pRaw, dataPos, &pSma->slidingUnit, _OVER)
   SDB_GET_INT8(pRaw, dataPos, &pSma->timezone, _OVER)
@@ -260,13 +265,16 @@ static void *mndBuildVCreateSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSm
   req.tagsFilterLen = pSma->tagsFilterLen;
   req.indexUid = pSma->uid;
   req.tableUid = pSma->stbUid;
+  req.dstVgId = pSma->dstVgId;
+  req.dstTbUid = pSma->dstTbUid;
   req.interval = pSma->interval;
   req.offset = pSma->offset;
   req.sliding = pSma->sliding;
   req.expr = pSma->expr;
   req.tagsFilter = pSma->tagsFilter;
-  req.numOfVgroups = pSma->numOfVgroups;
-  req.pVgEpSet = pSma->pVgEpSet;
+  req.schemaRow = pSma->schemaRow;
+  req.schemaTag = pSma->schemaTag;
+  req.dstTbName = pSma->dstTbName;
 
   // get length
   int32_t ret = 0;
@@ -425,14 +433,30 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
   mndReleaseDnode(pMnode, pDnode);
 
   // todo add sma info here
-  SVgEpSet *pVgEpSet = NULL;
-  int32_t   numOfVgroups = 0;
-  if (mndSmaGetVgEpSet(pMnode, pDb, &pVgEpSet, &numOfVgroups) != 0) {
+  SNode *pAst = NULL;
+  if (nodesStringToNode(pSma->ast, &pAst) < 0) {
     return -1;
   }
+  if (qExtractResultSchema(pAst, &pSma->schemaRow.nCols, &pSma->schemaRow.pSchema) != 0) {
+    nodesDestroyNode(pAst);
+    return -1;
+  }
+  nodesDestroyNode(pAst);
+  pSma->schemaRow.version = 1;
+
+  // TODO: the schemaTag generated by qExtractResultXXX later.
+  pSma->schemaTag.nCols = 1;
+  pSma->schemaTag.version = 1;
+  pSma->schemaTag.pSchema = taosMemoryCalloc(1, sizeof(SSchema));
+  if (!pSma->schemaTag.pSchema) {
+    return -1;
+  }
+  pSma->schemaTag.pSchema[0].type = TSDB_DATA_TYPE_BIGINT;
+  pSma->schemaTag.pSchema[0].bytes = TYPE_BYTES[TSDB_DATA_TYPE_BIGINT];
+  pSma->schemaTag.pSchema[0].colId = pSma->schemaRow.nCols + PRIMARYKEY_TIMESTAMP_COL_ID;
+  pSma->schemaTag.pSchema[0].flags = 0;
+  snprintf(pSma->schemaTag.pSchema[0].name, TSDB_COL_NAME_LEN, "groupId");
 
-  pSma->pVgEpSet = pVgEpSet;
-  pSma->numOfVgroups = numOfVgroups;
 
   int32_t smaContLen = 0;
   void   *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
@@ -464,12 +488,15 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   memcpy(smaObj.db, pDb->name, TSDB_DB_FNAME_LEN);
   smaObj.createdTime = taosGetTimestampMs();
   smaObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
+  char resultTbName[TSDB_TABLE_FNAME_LEN + 16] = {0};
+  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "td.tsma.rst.tb.%s", pCreate->name);
+  memcpy(smaObj.dstTbName, resultTbName, TSDB_TABLE_FNAME_LEN);
+  smaObj.dstTbUid = mndGenerateUid(smaObj.dstTbName, TSDB_TABLE_FNAME_LEN);
   smaObj.stbUid = pStb->uid;
   smaObj.dbUid = pStb->dbUid;
   smaObj.intervalUnit = pCreate->intervalUnit;
   smaObj.slidingUnit = pCreate->slidingUnit;
   smaObj.timezone = pCreate->timezone;
-  smaObj.dstVgId = pCreate->dstVgId;
   smaObj.interval = pCreate->interval;
   smaObj.offset = pCreate->offset;
   smaObj.sliding = pCreate->sliding;
@@ -1088,52 +1115,3 @@ static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
\ No newline at end of file
-
-static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups) {
-  SSdb     *pSdb = pMnode->pSdb;
-  SVgObj   *pVgroup = NULL;
-  void     *pIter = NULL;
-  SVgEpSet *pVgEpSet = NULL;
-  int32_t   nAllocVgs = 16;
-  int32_t   nVgs = 0;
-
-  pVgEpSet = taosMemoryCalloc(nAllocVgs, sizeof(SVgEpSet));
-  if (!pVgEpSet) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-
-  while (1) {
-    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
-    if (pIter == NULL) break;
-    if (pVgroup->dbUid != pDb->uid) {
-      sdbRelease(pSdb, pVgroup);
-      continue;
-    }
-
-    if (nVgs >= nAllocVgs) {
-      void *p = taosMemoryRealloc(pVgEpSet, nAllocVgs * 2 * sizeof(SVgEpSet));
-      if (!p) {
-        taosMemoryFree(pVgEpSet);
-        sdbCancelFetch(pSdb, pIter);
-        sdbRelease(pSdb, pVgroup);
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        return -1;
-      }
-      pVgEpSet = (SVgEpSet *)p;
-      nAllocVgs *= 2;
-    }
-
-    (pVgEpSet + nVgs)->vgId = pVgroup->vgId;
-    (pVgEpSet + nVgs)->epSet = mndGetVgroupEpset(pMnode, pVgroup);
-
-    ++nVgs;
-
-    sdbRelease(pSdb, pVgroup);
-  }
-
-  *ppVgEpSet = pVgEpSet;
-  *numOfVgroups = nVgs;
-
-  return 0;
-}

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

@@ -395,13 +395,13 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
     req.pRSmaParam.xFilesFactor = pStb->xFilesFactor;
     req.pRSmaParam.delay = pStb->delay;
     if (pStb->ast1Len > 0) {
-      if (mndConvertRSmaTask(pStb->pAst1, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0, &req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len,
+      if (mndConvertRsmaTask(&req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len, pStb->pAst1, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0,
                              req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
         return NULL;
       }
     }
     if (pStb->ast2Len > 0) {
-      if (mndConvertRSmaTask(pStb->pAst2, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0, &req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len,
+      if (mndConvertRsmaTask(&req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len, pStb->pAst2, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0,
                              req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
         return NULL;
       }

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

@@ -252,8 +252,12 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
   }
 
   if (qExtractResultSchema(pAst, (int32_t *)&pStream->outputSchema.nCols, &pStream->outputSchema.pSchema) != 0) {
+    nodesDestroyNode(pAst);
     return -1;
   }
+  // free
+  nodesDestroyNode(pAst);
+
 
 #if 0
   printf("|");
@@ -269,7 +273,7 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
     return -1;
   }
 
-  if (mndScheduleStream1(pMnode, pTrans, pStream) < 0) {
+  if (mndScheduleStream(pMnode, pTrans, pStream) < 0) {
     mError("stream:%ld, schedule stream since %s", pStream->uid, terrstr());
     return -1;
   }

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

@@ -61,6 +61,12 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
     }
     tsem_post(&pMgmt->syncSem);
   } else {
+    STrans *pTrans = mndAcquireTrans(pMnode, transId);
+    if (pTrans != NULL) {
+      mndTransExecute(pMnode, pTrans);
+      mndReleaseTrans(pMnode, pTrans);
+    }
+
     if (cbMeta.index - sdbGetApplyIndex(pMnode->pSdb) > 100) {
       SSnapshotMeta sMeta = {0};
       if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) {
@@ -200,14 +206,14 @@ int32_t mndInitSync(SMnode *pMnode) {
     return -1;
   }
 
-  mDebug("mnode sync is opened, id:%" PRId64, pMgmt->sync);
+  mDebug("mnode-sync is opened, id:%" PRId64, pMgmt->sync);
   return 0;
 }
 
 void mndCleanupSync(SMnode *pMnode) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
   syncStop(pMgmt->sync);
-  mDebug("mnode sync is stopped, id:%" PRId64, pMgmt->sync);
+  mDebug("mnode-sync is stopped, id:%" PRId64, pMgmt->sync);
 
   tsem_destroy(&pMgmt->syncSem);
   memset(pMgmt, 0, sizeof(SSyncMgmt));

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

@@ -52,8 +52,8 @@ static bool    mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans);
 static bool    mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans);
 static bool    mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans);
 static bool    mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans);
+static bool    mndCantExecuteTransAction(SMnode *pMnode) { return !pMnode->deploy && !mndIsMaster(pMnode); }
 
-static void    mndTransExecute(SMnode *pMnode, STrans *pTrans);
 static void    mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
 static int32_t mndProcessTransReq(SRpcMsg *pReq);
 static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
@@ -297,7 +297,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
       SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
       action.pRaw = taosMemoryMalloc(dataLen);
       if (action.pRaw == NULL) goto _OVER;
-      mTrace("raw:%p, is created", pData);
+      // mTrace("raw:%p, is created", pData);
       SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
       if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
       action.pRaw = NULL;
@@ -330,7 +330,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
       SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
       action.pRaw = taosMemoryMalloc(dataLen);
       if (action.pRaw == NULL) goto _OVER;
-      mTrace("raw:%p, is created", pData);
+      // mTrace("raw:%p, is created", action.pRaw);
       SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
       if (taosArrayPush(pTrans->undoActions, &action) == NULL) goto _OVER;
       action.pRaw = NULL;
@@ -363,7 +363,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
       SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
       action.pRaw = taosMemoryMalloc(dataLen);
       if (action.pRaw == NULL) goto _OVER;
-      mTrace("raw:%p, is created", action.pRaw);
+      // mTrace("raw:%p, is created", action.pRaw);
       SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
       if (taosArrayPush(pTrans->commitActions, &action) == NULL) goto _OVER;
       action.pRaw = NULL;
@@ -517,12 +517,12 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
 
   if (pOld->stage == TRN_STAGE_COMMIT) {
     pOld->stage = TRN_STAGE_COMMIT_ACTION;
-    mTrace("trans:%d, stage from commit to commitAction", pNew->id);
+    mTrace("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
   }
 
   if (pOld->stage == TRN_STAGE_ROLLBACK) {
     pOld->stage = TRN_STAGE_FINISHED;
-    mTrace("trans:%d, stage from rollback to finished", pNew->id);
+    mTrace("trans:%d, stage from rollback to finished since perform update action", pNew->id);
   }
   return 0;
 }
@@ -914,7 +914,7 @@ static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransActi
 
 static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
   if (pAction->msgSent) return 0;
-  if (!pMnode->deploy && !mndIsMaster(pMnode)) return -1;
+  if (mndCantExecuteTransAction(pMnode)) return -1;
 
   int64_t signature = pTrans->id;
   signature = (signature << 32);
@@ -1114,9 +1114,9 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
       pTrans->lastEpset = pAction->epSet;
     }
 
-    if (code == 0) {
-      if (!pMnode->deploy && !mndIsMaster(pMnode)) break;
+    if (mndCantExecuteTransAction(pMnode)) break;
 
+    if (code == 0) {
       pTrans->code = 0;
       pTrans->redoActionPos++;
       mDebug("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
@@ -1160,6 +1160,8 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
     code = mndTransExecuteRedoActions(pMnode, pTrans);
   }
 
+  if (mndCantExecuteTransAction(pMnode)) return false;
+
   if (code == 0) {
     pTrans->code = 0;
     pTrans->stage = TRN_STAGE_COMMIT;
@@ -1185,6 +1187,8 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
 }
 
 static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans) {
+  if (mndCantExecuteTransAction(pMnode)) return false;
+
   bool    continueExec = true;
   int32_t code = mndTransCommit(pMnode, pTrans);
 
@@ -1233,6 +1237,8 @@ static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
   bool    continueExec = true;
   int32_t code = mndTransExecuteUndoActions(pMnode, pTrans);
 
+  if (mndCantExecuteTransAction(pMnode)) return false;
+
   if (code == 0) {
     pTrans->stage = TRN_STAGE_ROLLBACK;
     mDebug("trans:%d, stage from undoAction to rollback", pTrans->id);
@@ -1250,6 +1256,8 @@ static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
 }
 
 static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
+  if (mndCantExecuteTransAction(pMnode)) return false;
+
   bool    continueExec = true;
   int32_t code = mndTransRollback(pMnode, pTrans);
 
@@ -1284,10 +1292,11 @@ static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans) {
   return continueExec;
 }
 
-static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
+void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
   bool continueExec = true;
 
   while (continueExec) {
+    mDebug("trans:%d, continue to execute, stage:%s", pTrans->id, mndTransStr(pTrans->stage));
     pTrans->lastExecTime = taosGetTimestampMs();
     switch (pTrans->stage) {
       case TRN_STAGE_PREPARE:

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

@@ -163,7 +163,12 @@ void sdbSetApplyIndex(SSdb *pSdb, int64_t index) { pSdb->curVer = index; }
 
 void sdbSetApplyTerm(SSdb *pSdb, int64_t term) { pSdb->curTerm = term; }
 
-void sdbSetCurConfig(SSdb *pSdb, int64_t config) { pSdb->curConfig = config; }
+void sdbSetCurConfig(SSdb *pSdb, int64_t config) {
+  if (pSdb->curConfig != config) {
+    mDebug("mnode sync config set from %" PRId64 " to %" PRId64, pSdb->curConfig, config);
+    pSdb->curConfig = config;
+  }
+}
 
 int64_t sdbGetApplyIndex(SSdb *pSdb) { return pSdb->curVer; }
 

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

@@ -432,8 +432,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
   } else {
     pSdb->lastCommitVer = pSdb->curVer;
     pSdb->lastCommitTerm = pSdb->curTerm;
-    mDebug("write sdb file successfully, ver:%" PRId64 " term:%" PRId64 " file:%s", pSdb->lastCommitVer,
-           pSdb->lastCommitTerm, curfile);
+    mDebug("write sdb file successfully, index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s",
+           pSdb->lastCommitVer, pSdb->lastCommitTerm, pSdb->curConfig, curfile);
   }
 
   terrno = code;

source/dnode/vnode/CMakeLists.txt

@@ -28,7 +28,6 @@ target_sources(
 
     # sma
     "src/sma/sma.c"
-    "src/sma/smaTDBImpl.c"
     "src/sma/smaEnv.c"
     "src/sma/smaOpen.c"
     "src/sma/smaRollup.c"

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

@@ -43,34 +43,16 @@ typedef struct SRSmaInfo    SRSmaInfo;
 struct SSmaEnv {
   TdThreadRwlock lock;
   int8_t         type;
-  TXN            txn;
-  void          *pPool;  // SPoolMem
-  SDiskID        did;
-  TDB           *dbEnv;  // TODO: If it's better to put it in smaIndex level?
-  char          *path;   // relative path
   SSmaStat      *pStat;
 };
 
 #define SMA_ENV_LOCK(env)       ((env)->lock)
 #define SMA_ENV_TYPE(env)       ((env)->type)
-#define SMA_ENV_DID(env)        ((env)->did)
-#define SMA_ENV_ENV(env)        ((env)->dbEnv)
-#define SMA_ENV_PATH(env)       ((env)->path)
 #define SMA_ENV_STAT(env)       ((env)->pStat)
 #define SMA_ENV_STAT_ITEMS(env) ((env)->pStat->smaStatItems)
 
 struct SSmaStatItem {
-  /**
-   * @brief The field 'state' is here to demonstrate if one smaIndex is ready to provide service.
-   *    - TSDB_SMA_STAT_OK: 1) The sma calculation of history data is finished; 2) Or recevied information from
-   * Streaming Module or TSDB local persistence.
-   *    - TSDB_SMA_STAT_EXPIRED: 1) If sma calculation of history TS data is not finished; 2) Or if the TSDB is open,
-   * without information about its previous state.
-   *    - TSDB_SMA_STAT_DROPPED: 1)sma dropped
-   * N.B. only applicable to tsma
-   */
   int8_t state;  // ETsdbSmaStat
-  SHashObj *expiredWindows;  // key: skey of time window, value: version
   STSma *pTSma;  // cache schema
 };
 
@@ -84,29 +66,6 @@ struct SSmaStat {
 #define SMA_STAT_ITEMS(s)     ((s)->smaStatItems)
 #define SMA_STAT_INFO_HASH(s) ((s)->rsmaInfoHash)
 
-struct SSmaKey {
-  TSKEY   skey;
-  int64_t groupId;
-};
-
-typedef struct SDBFile SDBFile;
-
-struct SDBFile {
-  int32_t fid;
-  TTB    *pDB;
-  char   *path;
-};
-
-int32_t tdSmaBeginCommit(SSmaEnv *pEnv);
-int32_t tdSmaEndCommit(SSmaEnv *pEnv);
-
-int32_t smaOpenDBEnv(TDB **ppEnv, const char *path);
-int32_t smaCloseDBEnv(TDB *pEnv);
-int32_t smaOpenDBF(TDB *pEnv, SDBFile *pDBF);
-int32_t smaCloseDBF(SDBFile *pDBF);
-int32_t smaSaveSmaToDB(SDBFile *pDBF, void *pKey, int32_t keyLen, void *pVal, int32_t valLen, TXN *txn);
-void   *smaGetSmaDataByKey(SDBFile *pDBF, const void *pKey, int32_t keyLen, int32_t *valLen);
-
 void  tdDestroySmaEnv(SSmaEnv *pSmaEnv);
 void *tdFreeSmaEnv(SSmaEnv *pSmaEnv);
 #if 0
@@ -114,13 +73,6 @@ int32_t tbGetTSmaStatus(SSma *pSma, STSma *param, void *result);
 int32_t tbRemoveTSmaData(SSma *pSma, STSma *param, STimeWindow *pWin);
 #endif
 
-static FORCE_INLINE int32_t tdEncodeTSmaKey(int64_t groupId, TSKEY tsKey, void **pData) {
-  int32_t len = 0;
-  len += taosEncodeFixedI64(pData, tsKey);
-  len += taosEncodeFixedI64(pData, groupId);
-  return len;
-}
-
 int32_t tdInitSma(SSma *pSma);
 int32_t tdDropTSma(SSma *pSma, char *pMsg);
 int32_t tdDropTSmaData(SSma *pSma, int64_t indexUid);
@@ -128,13 +80,11 @@ int32_t tdInsertRSmaData(SSma *pSma, char *msg);
 
 int32_t tdRefSmaStat(SSma *pSma, SSmaStat *pStat);
 int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat);
-int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
+int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType, bool onlyCheck);
 
 int32_t tdLockSma(SSma *pSma);
 int32_t tdUnLockSma(SSma *pSma);
 
-int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
-
 static FORCE_INLINE int16_t tdTSmaAdd(SSma *pSma, int16_t n) { return atomic_add_fetch_16(&SMA_TSMA_NUM(pSma), n); }
 static FORCE_INLINE int16_t tdTSmaSub(SSma *pSma, int16_t n) { return atomic_sub_fetch_16(&SMA_TSMA_NUM(pSma), n); }
 
@@ -219,11 +169,8 @@ static int32_t  tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDisk
 void *tdFreeRSmaInfo(SRSmaInfo *pInfo);
 
 int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
-int32_t tdUpdateExpiredWindowImpl(SSma *pSma, const SSubmitReq *pMsg, int64_t version);
-// TODO: This is the basic params, and should wrap the params to a queryHandle.
-int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
-
-int32_t tdGetTSmaDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
+int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
+int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
 
 #ifdef __cplusplus
 }

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

@@ -150,7 +150,6 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t smaOpen(SVnode* pVnode);
 int32_t smaClose(SSma* pSma);
 
-int32_t tdUpdateExpireWindow(SSma* pSma, const SSubmitReq* pMsg, int64_t version);
 int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
 
@@ -227,7 +226,7 @@ struct SVnode {
   SQHandle*  pQuery;
 };
 
-#define TD_VID(PVNODE) (PVNODE)->config.vgId
+#define TD_VID(PVNODE) ((PVNODE)->config.vgId)
 
 #define VND_TSDB(vnd)       ((vnd)->pTsdb)
 #define VND_RSMA0(vnd)      ((vnd)->pTsdb)

source/dnode/vnode/src/meta/metaQuery.c

@@ -86,11 +86,15 @@ tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) {
   int      nData = 0;
   tb_uid_t uid = 0;
 
+  metaRLock(pMeta);
+
   if (tdbTbGet(pMeta->pNameIdx, name, strlen(name) + 1, &pData, &nData) == 0) {
     uid = *(tb_uid_t *)pData;
     tdbFree(pData);
   }
 
+  metaULock(pMeta);
+
   return 0;
 }
 

source/dnode/vnode/src/meta/metaSma.c

@@ -34,13 +34,13 @@ int32_t metaCreateTSma(SMeta *pMeta, int64_t version, SSmaCfg *pCfg) {
   SMetaReader mr = {0};
 
   // validate req
+  // save smaIndex
   metaReaderInit(&mr, pMeta, 0);
   if (metaGetTableEntryByUid(&mr, pCfg->indexUid) == 0) {
-// TODO: just for pass case
 #if 1
-    terrno = TSDB_CODE_TDB_TSMA_ALREADY_EXIST;
+    terrno = TSDB_CODE_TSMA_ALREADY_EXIST;
     metaReaderClear(&mr);
-    return -1;
+    return -1;  // don't goto _err;
 #else
     metaReaderClear(&mr);
     return 0;

source/dnode/vnode/src/sma/sma.c

@@ -36,25 +36,9 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg) {
   return code;
 }
 
-int32_t tdUpdateExpireWindow(SSma* pSma, const SSubmitReq* pMsg, int64_t version) {
-  int32_t code = TSDB_CODE_SUCCESS;
-  if ((code = tdUpdateExpiredWindowImpl(pSma, pMsg, version)) < 0) {
-    smaWarn("vgId:%d, update expired sma window failed since %s", SMA_VID(pSma), tstrerror(terrno));
-  }
-  return code;
-}
-
-int32_t tdGetTSmaData(SSma* pSma, char* pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) {
-  int32_t code = TSDB_CODE_SUCCESS;
-  if ((code = tdGetTSmaDataImpl(pSma, pData, indexUid, querySKey, nMaxResult)) < 0) {
-    smaWarn("vgId:%d, get tsma data failed since %s", SMA_VID(pSma), tstrerror(terrno));
-  }
-  return code;
-}
-
 int32_t smaGetTSmaDays(SVnodeCfg* pCfg, void* pCont, uint32_t contLen, int32_t* days) {
   int32_t code = TSDB_CODE_SUCCESS;
-  if ((code = tdGetTSmaDaysImpl(pCfg, pCont, contLen, days)) < 0) {
+  if ((code = tdProcessTSmaGetDaysImpl(pCfg, pCont, contLen, days)) < 0) {
     smaWarn("vgId:%d, get tsma days failed since %s", pCfg->vgId, tstrerror(terrno));
   }
   smaDebug("vgId:%d, get tsma days %d", pCfg->vgId, *days);

source/dnode/vnode/src/sma/smaEnv.c

@@ -151,31 +151,11 @@ static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path,
     return NULL;
   }
 
-  ASSERT(path && (strlen(path) > 0));
-  SMA_ENV_PATH(pEnv) = strdup(path);
-  if (!SMA_ENV_PATH(pEnv)) {
-    tdFreeSmaEnv(pEnv);
-    return NULL;
-  }
-
-  SMA_ENV_DID(pEnv) = did;
-
   if (tdInitSmaStat(&SMA_ENV_STAT(pEnv), smaType) != TSDB_CODE_SUCCESS) {
     tdFreeSmaEnv(pEnv);
     return NULL;
   }
 
-  char aname[TSDB_FILENAME_LEN] = {0};
-  tfsAbsoluteName(SMA_TFS(pSma), did, path, aname);
-  if (smaOpenDBEnv(&pEnv->dbEnv, aname) != TSDB_CODE_SUCCESS) {
-    tdFreeSmaEnv(pEnv);
-    return NULL;
-  }
-
-  if (!(pEnv->pPool = openPool())) {
-    tdFreeSmaEnv(pEnv);
-    return NULL;
-  }
 
   return pEnv;
 }
@@ -205,10 +185,7 @@ void tdDestroySmaEnv(SSmaEnv *pSmaEnv) {
   if (pSmaEnv) {
     tdDestroySmaState(pSmaEnv->pStat, SMA_ENV_TYPE(pSmaEnv));
     taosMemoryFreeClear(pSmaEnv->pStat);
-    taosMemoryFreeClear(pSmaEnv->path);
     taosThreadRwlockDestroy(&(pSmaEnv->lock));
-    smaCloseDBEnv(pSmaEnv->dbEnv);
-    closePool(pSmaEnv->pPool);
   }
 }
 
@@ -242,7 +219,7 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
   }
 
   /**
-   *  1. Lazy mode utilized when init SSmaStat to update expired window(or hungry mode when tdNew).
+   *  1. Lazy mode utilized when init SSmaStat to update expire window(or hungry mode when tdNew).
    *  2. Currently, there is mutex lock when init SSmaEnv, thus no need add lock on SSmaStat, and please add lock if
    * tdInitSmaStat invoked in other multithread environment later.
    */
@@ -280,7 +257,6 @@ void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
   if (pSmaStatItem) {
     tDestroyTSma(pSmaStatItem->pTSma);
     taosMemoryFreeClear(pSmaStatItem->pTSma);
-    taosHashCleanup(pSmaStatItem->expiredWindows);
     taosMemoryFreeClear(pSmaStatItem);
   }
   return NULL;
@@ -341,7 +317,7 @@ int32_t tdUnLockSma(SSma *pSma) {
   return 0;
 }
 
-int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
+int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType, bool onlyCheck) {
   SSmaEnv *pEnv = NULL;
 
   // return if already init
@@ -399,63 +375,3 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
 
   return TSDB_CODE_SUCCESS;
 };
-
-int32_t tdSmaBeginCommit(SSmaEnv *pEnv) {
-  TXN *pTxn = &pEnv->txn;
-  // start a new txn
-  tdbTxnOpen(pTxn, 0, poolMalloc, poolFree, pEnv->pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
-  if (tdbBegin(pEnv->dbEnv, pTxn) != 0) {
-    smaWarn("tdSma tdb begin commit fail");
-    return -1;
-  }
-  return 0;
-}
-
-int32_t tdSmaEndCommit(SSmaEnv *pEnv) {
-  TXN *pTxn = &pEnv->txn;
-
-  // Commit current txn
-  if (tdbCommit(pEnv->dbEnv, pTxn) != 0) {
-    smaWarn("tdSma tdb end commit fail");
-    return -1;
-  }
-  tdbTxnClose(pTxn);
-  clearPool(pEnv->pPool);
-  return 0;
-}
-
-#if 0
-/**
- * @brief Get the start TS key of the last data block of one interval/sliding.
- *
- * @param pSma
- * @param param
- * @param result
- * @return int32_t
- *         1) Return 0 and fill the result if the check procedure is normal;
- *         2) Return -1 if error occurs during the check procedure.
- */
-int32_t tdGetTSmaStatus(SSma *pSma, void *smaIndex, void *result) {
-  const char *procedure = "";
-  if (strncmp(procedure, "get the start TS key of the last data block", 100) != 0) {
-    return -1;
-  }
-  // fill the result
-  return TSDB_CODE_SUCCESS;
-}
-
-/**
- * @brief Remove the tSma data files related to param between pWin.
- *
- * @param pSma
- * @param param
- * @param pWin
- * @return int32_t
- */
-int32_t tdRemoveTSmaData(SSma *pSma, void *smaIndex, STimeWindow *pWin) {
-  // for ("tSmaFiles of param-interval-sliding between pWin") {
-  //   // remove the tSmaFile
-  // }
-  return TSDB_CODE_SUCCESS;
-}
-#endif

source/dnode/vnode/src/sma/smaRollup.c

@@ -65,7 +65,7 @@ static FORCE_INLINE int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SA
   pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), suid, sizeof(tb_uid_t));
   if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
     smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64, SMA_VID(pSma), *suid);
-    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
+    terrno = TSDB_CODE_RSMA_INVALID_STAT;
     return TSDB_CODE_FAILED;
   }
 
@@ -132,7 +132,7 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
   SSmaStat *pStat = SMA_ENV_STAT(pEnv);
   SHashObj *infoHash = NULL;
   if (!pStat || !(infoHash = SMA_STAT_INFO_HASH(pStat))) {
-    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
+    terrno = TSDB_CODE_RSMA_INVALID_STAT;
     return TSDB_CODE_FAILED;
   }
 
@@ -167,13 +167,13 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
  */
 int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
   SSma *pSma = pVnode->pSma;
-  SMeta *pMeta = pVnode->pMeta;
-  SMsgCb *pMsgCb = &pVnode->msgCb;
   if (!pReq->rollup) {
     smaTrace("vgId:%d, return directly since no rollup for stable %s %" PRIi64, SMA_VID(pSma), pReq->name, pReq->suid);
     return TSDB_CODE_SUCCESS;
   }
 
+  SMeta *pMeta = pVnode->pMeta;
+  SMsgCb *pMsgCb = &pVnode->msgCb;
   SRSmaParam *param = &pReq->pRSmaParam;
 
   if ((param->qmsg1Len == 0) && (param->qmsg2Len == 0)) {
@@ -181,7 +181,7 @@ int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
     return TSDB_CODE_SUCCESS;
   }
 
-  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
+  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP, false) != TSDB_CODE_SUCCESS) {
     terrno = TSDB_CODE_TDB_INIT_FAILED;
     return TSDB_CODE_FAILED;
   }

source/dnode/vnode/src/sma/smaTDBImpl.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define ALLOW_FORBID_FUNC
-
-#include "sma.h"
-
-int32_t smaOpenDBEnv(TDB **ppEnv, const char *path) {
-  int ret = 0;
-
-  if (path == NULL) return -1;
-
-  ret = tdbOpen(path, 4096, 256, ppEnv);  // use as param
-
-  if (ret != 0) {
-    smaError("failed to create tsdb db env, ret = %d", ret);
-    return -1;
-  }
-
-  return 0;
-}
-
-int32_t smaCloseDBEnv(TDB *pEnv) { return tdbClose(pEnv); }
-
-static inline int tdSmaKeyCmpr(const void *arg1, int len1, const void *arg2, int len2) {
-  const SSmaKey *pKey1 = (const SSmaKey *)arg1;
-  const SSmaKey *pKey2 = (const SSmaKey *)arg2;
-
-  ASSERT(len1 == len2 && len1 == sizeof(SSmaKey));
-
-  if (pKey1->skey < pKey2->skey) {
-    return -1;
-  } else if (pKey1->skey > pKey2->skey) {
-    return 1;
-  }
-  if (pKey1->groupId < pKey2->groupId) {
-    return -1;
-  } else if (pKey1->groupId > pKey2->groupId) {
-    return 1;
-  }
-
-  return 0;
-}
-
-static int32_t smaOpenDBDb(TTB **ppDB, TDB *pEnv, const char *pFName) {
-  tdb_cmpr_fn_t compFunc;
-
-  // Create a database
-  compFunc = tdSmaKeyCmpr;
-  if (tdbTbOpen(pFName, -1, -1, compFunc, pEnv, ppDB) < 0) {
-    return -1;
-  }
-
-  return 0;
-}
-
-static int32_t smaCloseDBDb(TTB *pDB) { return tdbTbClose(pDB); }
-
-int32_t smaOpenDBF(TDB *pEnv, SDBFile *pDBF) {
-  // TEnv is shared by a group of SDBFile
-  if (!pEnv || !pDBF) {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return -1;
-  }
-
-  // Open DBF
-  if (smaOpenDBDb(&(pDBF->pDB), pEnv, pDBF->path) < 0) {
-    smaError("failed to open DBF: %s", pDBF->path);
-    smaCloseDBDb(pDBF->pDB);
-    return -1;
-  }
-
-  return 0;
-}
-
-int32_t smaCloseDBF(SDBFile *pDBF) {
-  int32_t ret = 0;
-  if (pDBF->pDB) {
-    ret = smaCloseDBDb(pDBF->pDB);
-    pDBF->pDB = NULL;
-  }
-  taosMemoryFreeClear(pDBF->path);
-  return ret;
-}
-
-int32_t smaSaveSmaToDB(SDBFile *pDBF, void *pKey, int32_t keyLen, void *pVal, int32_t valLen, TXN *txn) {
-  int32_t ret;
-
-  printf("save tsma data into %s, keyLen:%d valLen:%d txn:%p\n", pDBF->path, keyLen, valLen, txn);
-  ret = tdbTbUpsert(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
-  if (ret < 0) {
-    smaError("failed to upsert tsma data into db, ret = %d", ret);
-    return -1;
-  }
-
-  return 0;
-}
-
-void *smaGetSmaDataByKey(SDBFile *pDBF, const void *pKey, int32_t keyLen, int32_t *valLen) {
-  void *pVal = NULL;
-  int   ret;
-
-  ret = tdbTbGet(pDBF->pDB, pKey, keyLen, &pVal, valLen);
-
-  if (ret < 0) {
-    smaError("failed to get tsma data from db, ret = %d", ret);
-    return NULL;
-  }
-
-  ASSERT(*valLen >= 0);
-
-  // TODO: lock?
-  // TODO: Would the key/value be destoryed during return the data?
-  // TODO: How about the key is updated while value length is changed? The original value buffer would be freed
-  // automatically?
-
-  return pVal;
-}
\ No newline at end of file

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

@@ -130,7 +130,17 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           TD_VID(pTq->pVnode), pReq->currentOffset, fetchOffset);
 
   STqHandle* pHandle = taosHashGet(pTq->handles, pReq->subKey, strlen(pReq->subKey));
-  ASSERT(pHandle);
+  /*ASSERT(pHandle);*/
+  if (pHandle == NULL) {
+    tqError("tmq poll: no consumer handle for consumer %ld in vg %d, subkey %s", consumerId, pTq->pVnode->config.vgId,
+            pReq->subKey);
+    return -1;
+  }
+  if (pHandle->consumerId != consumerId) {
+    tqError("tmq poll: consumer handle mismatch for consumer %ld in vg %d, subkey %s, handle consumer id %ld",
+            consumerId, pTq->pVnode->config.vgId, pReq->subKey, pHandle->consumerId);
+    return -1;
+  }
 
   int32_t consumerEpoch = atomic_load_32(&pHandle->epoch);
   while (consumerEpoch < reqEpoch) {

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

@@ -238,9 +238,6 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
   if (msgType == TDMT_VND_SUBMIT) {
     if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0;
 
-    if (tdUpdateExpireWindow(pTq->pVnode->pSma, msg, ver) != 0) {
-      // TODO handle sma error
-    }
     void* data = taosMemoryMalloc(msgLen);
     if (data == NULL) {
       return -1;

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -150,7 +150,7 @@ int32_t tsdbCommit(STsdb *pTsdb) {
   return code;
 
 _err:
-  tsdbError("vgId:%d failed to commit since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
+  tsdbError("vgId:%d, failed to commit since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
   return code;
 }
 

source/dnode/vnode/src/tsdb/tsdbMemTable.c

@@ -176,13 +176,13 @@ int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid
 
   pMemTable->nDelOp++;
 
-  tsdbError("vgId:%d delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
+  tsdbError("vgId:%d, delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
             " since %s",
             TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
   return code;
 
 _err:
-  tsdbError("vgId:%d failed to delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
+  tsdbError("vgId:%d, failed to delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
             " since %s",
             TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
   return code;

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

@@ -225,6 +225,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
   vTrace("message in fetch queue is processing");
   char   *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
+
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
       return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);
@@ -236,13 +237,10 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
       return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);
     case TDMT_VND_QUERY_HEARTBEAT:
       return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
-
     case TDMT_VND_TABLE_META:
       return vnodeGetTableMeta(pVnode, pMsg);
-
     case TDMT_VND_CONSUME:
       return tqProcessPollReq(pVnode->pTq, pMsg, pInfo->workerId);
-
     case TDMT_STREAM_TASK_RUN:
       return tqProcessTaskRunReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_TASK_DISPATCH:
@@ -279,7 +277,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, int64_t version, SRpcMsg *pMsg, SRp
 void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
   // TODO
 
-  // blockDebugShowData(data);
+  // blockDebugShowData(data, __func__);
   tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
 }
 
@@ -795,7 +793,9 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
         msgIter.suid = 0;
       }
 
+#ifdef TD_DEBUG_PRINT_ROW
       vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
+#endif
       tDecoderClear(&decoder);
     } else {
       submitBlkRsp.tblFName = taosMemoryMalloc(TSDB_TABLE_FNAME_LEN);

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -373,7 +373,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
   pTsdb->pTfs = tfsOpen(&pDisks, numOfDisks);
   EXPECT_NE(pTsdb->pTfs, nullptr);
 
-  // generate SSubmitReq msg and update expired window
+  // generate SSubmitReq msg and update expire window
   int16_t  schemaVer = 0;
   uint32_t mockRowLen = sizeof(STSRow);
   uint32_t mockRowNum = 2;

source/libs/executor/inc/executorimpl.h

@@ -759,7 +759,7 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scan
 int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
 
 void    doSetOperatorCompleted(SOperatorInfo* pOperator);
-void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, bool needFree);
+void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock);
 SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset);
 void    relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols);
 void    initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow);

source/libs/executor/src/executorimpl.c

@@ -1819,9 +1819,9 @@ void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numO
   }
 }
 
-static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep, bool needFree);
+static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep);
 
-void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, bool needFree) {
+void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock) {
   if (pFilterNode == NULL) {
     return;
   }
@@ -1840,30 +1840,29 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, bool needFree) {
   bool keep = filterExecute(filter, pBlock, &rowRes, NULL, param1.numOfCols);
   filterFreeInfo(filter);
 
-  extractQualifiedTupleByFilterResult(pBlock, rowRes, keep, needFree);
+  extractQualifiedTupleByFilterResult(pBlock, rowRes, keep);
   blockDataUpdateTsWindow(pBlock, 0);
 }
 
-void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep, bool needFree) {
+void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep) {
   if (keep) {
     return;
   }
 
   if (rowRes != NULL) {
-    SSDataBlock* px = createOneDataBlock(pBlock, false);
-    blockDataEnsureCapacity(px, pBlock->info.rows);
+    SSDataBlock* px = createOneDataBlock(pBlock, true);
 
     int32_t totalRows = pBlock->info.rows;
-
     for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
-      SColumnInfoData* pDst = taosArrayGet(px->pDataBlock, i);
-      SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, i);
-
+      SColumnInfoData* pSrc = taosArrayGet(px->pDataBlock, i);
+      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
       // it is a reserved column for scalar function, and no data in this column yet.
-      if (pSrc->pData == NULL) {
+      if (pDst->pData == NULL) {
         continue;
       }
 
+      colInfoDataCleanup(pDst, pBlock->info.rows);
+
       int32_t numOfRows = 0;
       for (int32_t j = 0; j < totalRows; ++j) {
         if (rowRes[j] == 0) {
@@ -1883,20 +1882,8 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR
       } else {
         ASSERT(pBlock->info.rows == numOfRows);
       }
-
-      SColumnInfoData tmp = *pSrc;
-      *pSrc = *pDst;
-      *pDst = tmp;
-
-      if (!needFree) {
-        if (IS_VAR_DATA_TYPE(pDst->info.type)) {  // this elements do not need free
-          pDst->varmeta.offset = NULL;
-        } else {
-          pDst->nullbitmap = NULL;
-        }
-        pDst->pData = NULL;
-      }
     }
+
     blockDataDestroy(px);  // fix memory leak
   } else {
     // do nothing
@@ -2137,11 +2124,6 @@ static int32_t compressQueryColData(SColumnInfoData* pColRes, int32_t numOfRows,
 }
 
 int32_t doFillTimeIntervalGapsInResults(struct SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t capacity) {
-  //  for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
-  //    SColumnInfoData* pColInfoData = taosArrayGet(pOutput->pDataBlock, i);
-  //    p[i] = pColInfoData->pData + (pColInfoData->info.bytes * pOutput->info.rows);
-  //  }
-
   int32_t numOfRows = (int32_t)taosFillResultDataBlock(pFillInfo, pBlock, capacity - pBlock->info.rows);
   pBlock->info.rows += numOfRows;
 
@@ -2896,7 +2878,6 @@ static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) {
     return seqLoadRemoteData(pOperator);
   } else {
     return concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
-    //    return concurrentlyLoadRemoteData(pOperator);
   }
 }
 
@@ -2922,9 +2903,14 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo) {
+static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo, const char* id) {
   size_t numOfSources = LIST_LENGTH(pExNode->pSrcEndPoints);
 
+  if (numOfSources == 0) {
+    qError("%s invalid number: %d of sources in exchange operator", id, (int32_t) numOfSources);
+    return TSDB_CODE_INVALID_PARA;
+  }
+
   pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode));
   pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo));
   if (pInfo->pSourceDataInfo == NULL || pInfo->pSources == NULL) {
@@ -2946,7 +2932,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
     goto _error;
   }
 
-  int32_t code = initExchangeOperator(pExNode, pInfo);
+  int32_t code = initExchangeOperator(pExNode, pInfo, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -2975,7 +2961,7 @@ _error:
 
   taosMemoryFreeClear(pInfo);
   taosMemoryFreeClear(pOperator);
-  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
+  pTaskInfo->code = code;
   return NULL;
 }
 
@@ -3744,7 +3730,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
       longjmp(pTaskInfo->env, code);
     }
 
-    doFilter(pProjectInfo->pFilterNode, pBlock, true);
+    doFilter(pProjectInfo->pFilterNode, pBlock);
 
     setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, false);
     blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
@@ -5263,7 +5249,7 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
   (*pTaskInfo)->pRoot = createOperatorTree(pPlan->pNode, *pTaskInfo, pHandle, queryId, taskId,
                                            &(*pTaskInfo)->tableqinfoList, pPlan->pTagCond);
   if (NULL == (*pTaskInfo)->pRoot) {
-    code = terrno;
+    code = (*pTaskInfo)->code;
     goto _complete;
   }
 
@@ -5276,7 +5262,6 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
 
 _complete:
   taosMemoryFreeClear(*pTaskInfo);
-
   terrno = code;
   return code;
 }

source/libs/executor/src/groupoperator.c

@@ -359,7 +359,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
 
   while(1) {
     doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
-    doFilter(pInfo->pCondition, pRes, true);
+    doFilter(pInfo->pCondition, pRes);
 
     bool hasRemain = hashRemainDataInGroupInfo(&pInfo->groupResInfo);
     if (!hasRemain) {

source/libs/executor/src/scanoperator.c

@@ -267,7 +267,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
   }
 
   int64_t st = taosGetTimestampMs();
-  doFilter(pTableScanInfo->pFilterNode, pBlock, false);
+  doFilter(pTableScanInfo->pFilterNode, pBlock);
 
   int64_t et = taosGetTimestampMs();
   pTableScanInfo->readRecorder.filterTime += (et - st);
@@ -948,7 +948,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
         addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes);
       }
 
-      doFilter(pInfo->pCondition, pInfo->pRes, false);
+      doFilter(pInfo->pCondition, pInfo->pRes);
       blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
       break;
     }
@@ -1716,7 +1716,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
   }
 
   pRes->info.rows = count;
-  doFilter(pInfo->pFilterNode, pRes, true);
+  doFilter(pInfo->pFilterNode, pRes);
 
   pOperator->resultInfo.totalRows += pRes->info.rows;
 

source/libs/executor/src/sortoperator.c

@@ -296,7 +296,10 @@ int32_t doOpenMultiwaySortMergeOperator(SOperatorInfo* pOperator) {
 }
 
 SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity,
-                                        SArray* pColMatchInfo, SMultiwaySortMergeOperatorInfo* pInfo) {
+                                        SArray* pColMatchInfo, SOperatorInfo* pOperator) {
+  SMultiwaySortMergeOperatorInfo* pInfo = pOperator->info;
+  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
+
   blockDataCleanup(pDataBlock);
 
   SSDataBlock* p = tsortGetSortedDataBlock(pHandle);
@@ -354,6 +357,8 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
   }
 
   blockDataDestroy(p);
+
+  qDebug("%s get sorted row blocks, rows:%d", GET_TASKID(pTaskInfo), pDataBlock->info.rows);
   return (pDataBlock->info.rows > 0) ? pDataBlock : NULL;
 }
 
@@ -371,7 +376,7 @@ SSDataBlock* doMultiwaySortMerge(SOperatorInfo* pOperator) {
   }
 
   SSDataBlock* pBlock = getMultiwaySortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes,
-                                                   pOperator->resultInfo.capacity, pInfo->pColMatchInfo, pInfo);
+                                                   pOperator->resultInfo.capacity, pInfo->pColMatchInfo, pOperator);
 
   if (pBlock != NULL) {
     pOperator->resultInfo.totalRows += pBlock->info.rows;

source/libs/executor/src/tsort.c

@@ -227,6 +227,8 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
     for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
       SSortSource* pSource = cmpParam->pSources[i];
       pSource->src.pBlock = pHandle->fetchfp(pSource->param);
+
+      // set current source id done
       if (pSource->src.pBlock == NULL) {
         pSource->src.rowIndex = -1;
         ++pHandle->numOfCompletedSources;
@@ -426,8 +428,16 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
   double sortPass = floorl(log2(numOfSources) / log2(pHandle->numOfPages));
 
   pHandle->totalElapsed = taosGetTimestampUs() - pHandle->startTs;
-  qDebug("%s %d rounds mergesort required to complete the sort, first-round sorted data size:%"PRIzu", sort elapsed:%"PRId64", total elapsed:%"PRId64,
-         pHandle->idStr, (int32_t) (sortPass + 1), pHandle->pBuf ? getTotalBufSize(pHandle->pBuf) : 0, pHandle->sortElapsed, pHandle->totalElapsed);
+
+  if (sortPass > 0) {
+    size_t s = pHandle->pBuf ? getTotalBufSize(pHandle->pBuf) : 0;
+    qDebug("%s %d rounds mergesort required to complete the sort, first-round sorted data size:%" PRIzu
+           ", sort elapsed:%" PRId64 ", total elapsed:%" PRId64,
+           pHandle->idStr, (int32_t)(sortPass + 1), s, pHandle->sortElapsed, pHandle->totalElapsed);
+  } else {
+    qDebug("%s ordered source:%"PRIzu", available buf:%d, no need internal sort", pHandle->idStr, numOfSources,
+        pHandle->numOfPages);
+  }
 
   int32_t numOfRows = blockDataGetCapacityInRow(pHandle->pDataBlock, pHandle->pageSize);
   blockDataEnsureCapacity(pHandle->pDataBlock, numOfRows);

source/libs/function/inc/functionMgtInt.h

@@ -41,6 +41,7 @@ extern "C" {
 #define FUNC_MGT_SCAN_PC_FUNC           FUNC_MGT_FUNC_CLASSIFICATION_MASK(12)
 #define FUNC_MGT_SELECT_FUNC            FUNC_MGT_FUNC_CLASSIFICATION_MASK(13)
 #define FUNC_MGT_REPEAT_SCAN_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(14)
+#define FUNC_MGT_FORBID_FILL_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(15)
 
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 

source/libs/function/src/builtinsimpl.c

@@ -2895,6 +2895,7 @@ void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
   for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
     SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
     if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) {
+      offset += pCol->info.bytes;
       continue;
     }