Merge remote-tracking branch 'origin/3.0' into fix/dnode

include/common/tmsg.h

@@ -2510,6 +2510,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/libs/catalog/catalog.h

@@ -67,6 +67,7 @@ typedef struct SCatalogReq {
   SArray *pUdf;           // element is udf name
   SArray *pIndex;         // element is index name
   SArray *pUser;          // element is SUserAuthInfo
+  SArray *pTableIndex;    // element is SNAME
   bool    qNodeRequired;  // valid qnode
   bool    forceUpdate;    
 } SCatalogReq;
@@ -82,6 +83,7 @@ typedef struct SMetaData {
   SArray    *pDbInfo;     // pRes = SDbInfo*
   SArray    *pTableMeta;  // pRes = STableMeta*
   SArray    *pTableHash;  // pRes = SVgroupInfo*
+  SArray    *pTableIndex; // pRes = SArray<STableIndexInfo>*
   SArray    *pUdfList;    // pRes = SFuncInfo*
   SArray    *pIndex;      // pRes = SIndexInfo*
   SArray    *pUser;       // pRes = bool*
@@ -277,7 +279,7 @@ int32_t catalogGetDBCfg(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, cons
 
 int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* indexName, SIndexInfo* pInfo);
 
-int32_t catalogGetTableIndex(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const char* tbFName, SArray** pRes);
+int32_t catalogGetTableIndex(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pRes);
 
 int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo* pInfo);
 

include/libs/function/function.h

@@ -146,6 +146,7 @@ typedef struct SqlFunctionCtx {
   struct SDiskbasedBuf  *pBuf;
   struct SSDataBlock    *pSrcBlock;
   int32_t                curBufPage;
+  bool                   increase;
 
   char                   udfName[TSDB_FUNC_NAME_LEN];
 } SqlFunctionCtx;

include/libs/function/functionMgt.h

@@ -129,6 +129,8 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_SPREAD_MERGE,
   FUNCTION_TYPE_HISTOGRAM_PARTIAL,
   FUNCTION_TYPE_HISTOGRAM_MERGE,
+  FUNCTION_TYPE_HYPERLOGLOG_PARTIAL,
+  FUNCTION_TYPE_HYPERLOGLOG_MERGE,
 
   // user defined funcion
   FUNCTION_TYPE_UDF = 10000

include/libs/nodes/plannodes.h

@@ -304,6 +304,7 @@ typedef struct SDownstreamSourceNode {
 typedef struct SExchangePhysiNode {
   SPhysiNode node;
   int32_t    srcGroupId;     // group id of datasource suplans
+  bool       singleChannel;
   SNodeList* pSrcEndPoints;  // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
 } SExchangePhysiNode;
 

source/client/src/clientSml.c

@@ -303,7 +303,7 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
           uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
         }
         taos_free_result(res2);
-        taosMsleep(500);
+        taosMsleep(10);
       }
       break;
     }
@@ -327,7 +327,7 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
           uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
         }
         taos_free_result(res2);
-        taosMsleep(500);
+        taosMsleep(10);
       }
       break;
     }
@@ -350,7 +350,7 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
           uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
         }
         taos_free_result(res2);
-        taosMsleep(500);
+        taosMsleep(10);
       }
       break;
     }
@@ -373,7 +373,7 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
           uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
         }
         taos_free_result(res2);
-        taosMsleep(500);
+        taosMsleep(10);
       }
       break;
     }
@@ -424,7 +424,7 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
           uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
         }
         taos_free_result(res2);
-        taosMsleep(500);
+        taosMsleep(10);
       }
       break;
     }
@@ -461,18 +461,18 @@ static int32_t smlProcessSchemaAction(SSmlHandle* info, SSchema* schemaField, SH
 
 static int32_t smlModifyDBSchemas(SSmlHandle* info) {
   int32_t code = 0;
+  SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
+  SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
+  strcpy(pName.dbname, info->pRequest->pDb);
 
   SSmlSTableMeta** tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, NULL);
   while (tableMetaSml) {
     SSmlSTableMeta* sTableData = *tableMetaSml;
-
     STableMeta *pTableMeta = NULL;
-    SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
 
     size_t superTableLen = 0;
     void *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
-    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
-    strcpy(pName.dbname, info->pRequest->pDb);
+    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
     memcpy(pName.tname, superTable, superTableLen);
 
     code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
@@ -487,7 +487,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
       code = smlApplySchemaAction(info, &schemaAction);
       if (code != TSDB_CODE_SUCCESS) {
         uError("SML:0x%"PRIx64" smlApplySchemaAction failed. can not create %s", info->id, schemaAction.createSTable.sTableName);
-        return code;
+        goto end;
       }
       info->cost.numOfCreateSTables++;
     }else if (code == TSDB_CODE_SUCCESS) {
@@ -502,7 +502,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
       code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &schemaAction, true);
       if (code != TSDB_CODE_SUCCESS) {
         taosHashCleanup(hashTmp);
-        return code;
+        goto end;
       }
 
       taosHashClear(hashTmp);
@@ -512,29 +512,33 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
       code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &schemaAction, false);
       taosHashCleanup(hashTmp);
       if (code != TSDB_CODE_SUCCESS) {
-        return code;
+        goto end;
       }
 
-      code = catalogRefreshTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, -1);
+      code = catalogRefreshTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, 1);
       if (code != TSDB_CODE_SUCCESS) {
-        return code;
+        goto end;
       }
     } else {
       uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
-      return code;
+      goto end;
     }
     if(pTableMeta) taosMemoryFree(pTableMeta);
 
     code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
     if (code != TSDB_CODE_SUCCESS) {
       uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, (char*)superTable);
-      return code;
+      goto end;
     }
     sTableData->tableMeta = pTableMeta;
 
     tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, tableMetaSml);
   }
   return 0;
+
+end:
+  catalogRefreshTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, 1);
+  return code;
 }
 
 //=========================================================================
@@ -1544,7 +1548,7 @@ static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsV
   }
 
   size_t typeLen = strlen(type->valuestring);
-  if (typeLen == 1 && type->valuestring[0] == 's') {
+  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
     //seconds
     timeDouble = timeDouble * 1e9;
     if(smlDoubleToInt64OverFlow(timeDouble)){
@@ -1552,9 +1556,10 @@ static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsV
       return TSDB_CODE_TSC_INVALID_TIME_STAMP;
     }
     *tsVal = timeDouble;
-  } else if (typeLen == 2 && type->valuestring[1] == 's') {
+  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
     switch (type->valuestring[0]) {
       case 'm':
+      case 'M':
         //milliseconds
         timeDouble = timeDouble * 1e6;
         if(smlDoubleToInt64OverFlow(timeDouble)){
@@ -1564,6 +1569,7 @@ static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsV
         *tsVal = timeDouble;
         break;
       case 'u':
+      case 'U':
         //microseconds
         timeDouble = timeDouble * 1e3;
         if(smlDoubleToInt64OverFlow(timeDouble)){
@@ -1573,6 +1579,7 @@ static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsV
         *tsVal = timeDouble;
         break;
       case 'n':
+      case 'N':
         //nanoseconds
         *tsVal = timeDouble;
         break;
@@ -2285,6 +2292,8 @@ static int32_t smlParseLine(SSmlHandle *info, char* lines[], int numLines){
 
 static int smlProcess(SSmlHandle *info, char* lines[], int numLines) {
   int32_t code = TSDB_CODE_SUCCESS;
+  int32_t retryNum = 0;
+
   info->cost.parseTime = taosGetTimestampUs();
 
   code = smlParseLine(info, lines, numLines);
@@ -2298,7 +2307,12 @@ static int smlProcess(SSmlHandle *info, char* lines[], int numLines) {
   info->cost.numOfCTables = taosHashGetSize(info->childTables);
 
   info->cost.schemaTime = taosGetTimestampUs();
+
+  do{
     code = smlModifyDBSchemas(info);
+    if (code == 0) break;
+  } while (retryNum++ < taosHashGetSize(info->superTables));
+
   if (code != 0) {
     uError("SML:0x%"PRIx64" smlModifyDBSchemas error : %s", info->id, tstrerror(code));
     goto cleanup;
@@ -2409,6 +2423,7 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
   info->pRequest->code = smlProcess(info, lines, numLines);
 
 end:
+  info->taos->schemalessType = 0;
   uDebug("result:%s", info->msgBuf.buf);
   smlDestroyInfo(info);
   return (TAOS_RES*)request;

source/client/test/smlTest.cpp

@@ -1272,14 +1272,40 @@ TEST(testCase, sml_params_Test) {
   };
   TAOS_RES* res = taos_schemaless_insert(taos, (char**)sql, 1, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_MILLI_SECONDS);
   ASSERT_EQ(taos_errno(res), TSDB_CODE_PAR_DB_NOT_SPECIFIED);
-  taos_free_result(pRes);
+  taos_free_result(res);
 
   pRes = taos_query(taos, "use param");
-  taos_free_result(pRes);
+  taos_free_result(res);
 
   res = taos_schemaless_insert(taos, (char**)sql, 1, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_MILLI_SECONDS);
   ASSERT_EQ(taos_errno(res), TSDB_CODE_SML_INVALID_DB_CONF);
+  taos_free_result(res);
+}
+
+TEST(testCase, sml_16384_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, nullptr);
+
+  TAOS_RES* pRes = taos_query(taos, "create database if not exists d16384 schemaless 1");
+  taos_free_result(pRes);
+
+  const char *sql[] = {
+      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c0=t,c1=127i8 1626006833639000000",
+  };
+
+  pRes = taos_query(taos, "use d16384");
   taos_free_result(pRes);
+
+  TAOS_RES* res = taos_schemaless_insert(taos, (char**)sql, 1, TSDB_SML_LINE_PROTOCOL, 0);
+  ASSERT_EQ(taos_errno(res), 0);
+  taos_free_result(res);
+
+  const char *sql1[] = {
+      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c0=f,c1=127i8,c11=L\"ncharColValue\",c10=t 1626006833639000000",
+  };
+  TAOS_RES* res1 = taos_schemaless_insert(taos, (char**)sql1, 1, TSDB_SML_LINE_PROTOCOL, 0);
+  ASSERT_EQ(taos_errno(res1), 0);
+  taos_free_result(res1);
 }
 
 TEST(testCase, sml_oom_Test) {
@@ -1303,3 +1329,29 @@ TEST(testCase, sml_oom_Test) {
   ASSERT_EQ(taos_errno(res), 0);
   taos_free_result(pRes);
 }
+
+TEST(testCase, sml_16368_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, nullptr);
+
+  TAOS_RES* pRes = taos_query(taos, "create database if not exists d16368 schemaless 1");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use d16368");
+  taos_free_result(pRes);
+
+  const char *sql[] = {
+      "[{\"metric\": \"st123456\", \"timestamp\": {\"value\": 1626006833639000, \"type\": \"us\"}, \"value\": 1, \"tags\": {\"t1\": 3, \"t2\": {\"value\": 4, \"type\": \"double\"}, \"t3\": {\"value\": \"t3\", \"type\": \"binary\"}}},\n"
+      "{\"metric\": \"st123456\", \"timestamp\": {\"value\": 1626006833739000, \"type\": \"us\"}, \"value\": 2, \"tags\": {\"t1\": {\"value\": 4, \"type\": \"double\"}, \"t3\": {\"value\": \"t4\", \"type\": \"binary\"}, \"t2\": {\"value\": 5, \"type\": \"double\"}, \"t4\": {\"value\": 5, \"type\": \"double\"}}},\n"
+      "{\"metric\": \"stb_name\", \"timestamp\": {\"value\": 1626006833639100, \"type\": \"us\"}, \"value\": 3, \"tags\": {\"t2\": {\"value\": 5, \"type\": \"double\"}, \"t3\": {\"value\": \"ste\", \"type\": \"nchar\"}}},\n"
+      "{\"metric\": \"stf567890\", \"timestamp\": {\"value\": 1626006833639200, \"type\": \"us\"}, \"value\": 4, \"tags\": {\"t1\": {\"value\": 4, \"type\": \"bigint\"}, \"t3\": {\"value\": \"t4\", \"type\": \"binary\"}, \"t2\": {\"value\": 5, \"type\": \"double\"}, \"t4\": {\"value\": 5, \"type\": \"double\"}}},\n"
+      "{\"metric\": \"st123456\", \"timestamp\": {\"value\": 1626006833639300, \"type\": \"us\"}, \"value\": {\"value\": 5, \"type\": \"double\"}, \"tags\": {\"t1\": {\"value\": 4, \"type\": \"double\"}, \"t2\": 5.0, \"t3\": {\"value\": \"t4\", \"type\": \"binary\"}}},\n"
+      "{\"metric\": \"stb_name\", \"timestamp\": {\"value\": 1626006833639400, \"type\": \"us\"}, \"value\": {\"value\": 6, \"type\": \"double\"}, \"tags\": {\"t2\": 5.0, \"t3\": {\"value\": \"ste2\", \"type\": \"nchar\"}}},\n"
+      "{\"metric\": \"stb_name\", \"timestamp\": {\"value\": 1626006834639400, \"type\": \"us\"}, \"value\": {\"value\": 7, \"type\": \"double\"}, \"tags\": {\"t2\": {\"value\": 5.0, \"type\": \"double\"}, \"t3\": {\"value\": \"ste2\", \"type\": \"nchar\"}}},\n"
+      "{\"metric\": \"st123456\", \"timestamp\": {\"value\": 1626006833839006, \"type\": \"us\"}, \"value\": {\"value\": 8, \"type\": \"double\"}, \"tags\": {\"t1\": {\"value\": 4, \"type\": \"double\"}, \"t3\": {\"value\": \"t4\", \"type\": \"binary\"}, \"t2\": {\"value\": 5, \"type\": \"double\"}, \"t4\": {\"value\": 5, \"type\": \"double\"}}},\n"
+      "{\"metric\": \"st123456\", \"timestamp\": {\"value\": 1626006833939007, \"type\": \"us\"}, \"value\": {\"value\": 9, \"type\": \"double\"}, \"tags\": {\"t1\": 4, \"t3\": {\"value\": \"t4\", \"type\": \"binary\"}, \"t2\": {\"value\": 5, \"type\": \"double\"}, \"t4\": {\"value\": 5, \"type\": \"double\"}}}]"
+  };
+  pRes = taos_schemaless_insert(taos, (char**)sql, sizeof(sql)/sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_MICRO_SECONDS);
+  ASSERT_EQ(taos_errno(pRes), 0);
+  taos_free_result(pRes);
+}

source/common/src/tdatablock.c

@@ -1752,7 +1752,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
       taosArrayClear(tagArray);
       taosArrayPush(tagArray, &tagVal);
       tTagNew(tagArray, 1, false, &pTag);
-      if (!pTag) {
+      if (pTag == NULL) {
         tdDestroySVCreateTbReq(&createTbReq);
         taosArrayDestroy(tagArray);
         return NULL;
@@ -1763,9 +1763,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
       tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
 
       tdDestroySVCreateTbReq(&createTbReq);
-
       if (code < 0) {
-        tdDestroySVCreateTbReq(&createTbReq);
         taosArrayDestroy(tagArray);
         return NULL;
       }
@@ -1804,8 +1802,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
     int32_t schemaLen = 0;
     if (createTb) {
       SVCreateTbReq createTbReq = {0};
-      char*         cname = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
-      snprintf(cname, TSDB_TABLE_FNAME_LEN, "%s:%ld", stbFullName, pDataBlock->info.groupId);
+      char*         cname = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
       createTbReq.name = cname;
       createTbReq.flags = 0;
       createTbReq.type = TSDB_CHILD_TABLE;
@@ -1819,7 +1816,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
       taosArrayPush(tagArray, &tagVal);
       STag* pTag = NULL;
       tTagNew(tagArray, 1, false, &pTag);
-      if (!pTag) {
+      if (pTag == NULL) {
         tdDestroySVCreateTbReq(&createTbReq);
         taosArrayDestroy(tagArray);
         taosMemoryFreeClear(ret);
@@ -1945,7 +1942,6 @@ void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen
 
 const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t numOfRows, const char* pData) {
   blockDataEnsureCapacity(pBlock, numOfRows);
-  pBlock->info.rows = numOfRows;
 
   const char* pStart = pData;
 
@@ -2019,6 +2015,7 @@ const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t
     pStart += colLen[i];
   }
 
+  pBlock->info.rows = numOfRows;
   ASSERT(pStart - pData == dataLen);
   return pStart;
 }

source/common/src/tdataformat.c

@@ -244,7 +244,7 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
     }
   }
 
-  // ASSERT(flags); // only 1 column(ts)
+  ASSERT(flags);
 
   // decide
   uint32_t nData = 0;
@@ -268,8 +268,8 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
         nDataT = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntv;
         break;
       default:
-        break; // only ts column
-        // ASSERT(0);
+        break;
+        ASSERT(0);
     }
 
     uint8_t tflags = 0;
@@ -374,7 +374,7 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
           ptv = pf + pTSchema->flen;
           break;
         default:
-          // ASSERT(0);
+          ASSERT(0);
           break;
       }
     } else {
@@ -421,12 +421,26 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
     _set_none:
       if ((flags & 0xf0) == 0) {
         setBitMap(pb, 0, iColumn - 1, flags);
+        if (flags & TSROW_HAS_VAL) { // set 0
+          if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+            *(VarDataOffsetT *)(pf + pTColumn->offset) = 0;
+          } else {
+            tPutValue(pf + pTColumn->offset, &((SValue){0}), pTColumn->type);
+          }
+        }
       }
       continue;
 
     _set_null:
       if ((flags & 0xf0) == 0) {
         setBitMap(pb, 1, iColumn - 1, flags);
+        if (flags & TSROW_HAS_VAL) { // set 0
+          if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+            *(VarDataOffsetT *)(pf + pTColumn->offset) = 0;
+          } else {
+            tPutValue(pf + pTColumn->offset, &((SValue){0}), pTColumn->type);
+          }
+        }
       } else {
         SET_IDX(pidx, pTSKVRow->nCols, nkv, flags);
         pTSKVRow->nCols++;
@@ -497,7 +511,7 @@ void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal
   SValue    value;
 
   ASSERT(iCol < pTSchema->numOfCols);
-  // ASSERT(flags); // only 1 ts column
+  ASSERT(flags);
   ASSERT(pRow->sver == pTSchema->version);
 
   if (iCol == 0) {

source/common/src/tmsg.c

@@ -2491,6 +2491,15 @@ int32_t tDeserializeSTableIndexRsp(void *buf, int32_t bufLen, STableIndexRsp *pR
   return 0;
 }
 
+void tFreeSTableIndexInfo(void* info) {
+  if (NULL == info) {
+    return;
+  }
+
+  STableIndexInfo *pInfo = (STableIndexInfo*)info;
+
+  taosMemoryFree(pInfo->expr);
+}
 
 int32_t tSerializeSShowReq(void *buf, int32_t bufLen, SShowReq *pReq) {
   SEncoder encoder = {0};

source/common/test/dataformatTest.cpp

@@ -52,61 +52,61 @@ STSchema *genSTSchema(int16_t nCols) {
 
     switch (i) {
       case 0: {
-        pSchema[0].type = TSDB_DATA_TYPE_TIMESTAMP;
-        pSchema[0].bytes = TYPE_BYTES[pSchema[0].type];
+        pSchema[i].type = TSDB_DATA_TYPE_TIMESTAMP;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 1: {
-        pSchema[1].type = TSDB_DATA_TYPE_INT;
-        pSchema[1].bytes = TYPE_BYTES[pSchema[1].type];
+        pSchema[i].type = TSDB_DATA_TYPE_INT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
         ;
       } break;
       case 2: {
-        pSchema[2].type = TSDB_DATA_TYPE_BIGINT;
-        pSchema[2].bytes = TYPE_BYTES[pSchema[2].type];
+        pSchema[i].type = TSDB_DATA_TYPE_BIGINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 3: {
-        pSchema[3].type = TSDB_DATA_TYPE_FLOAT;
-        pSchema[3].bytes = TYPE_BYTES[pSchema[3].type];
+        pSchema[i].type = TSDB_DATA_TYPE_FLOAT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 4: {
-        pSchema[4].type = TSDB_DATA_TYPE_DOUBLE;
-        pSchema[4].bytes = TYPE_BYTES[pSchema[4].type];
+        pSchema[i].type = TSDB_DATA_TYPE_DOUBLE;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 5: {
-        pSchema[5].type = TSDB_DATA_TYPE_BINARY;
-        pSchema[5].bytes = 12;
+        pSchema[i].type = TSDB_DATA_TYPE_BINARY;
+        pSchema[i].bytes = 12;
       } break;
       case 6: {
-        pSchema[6].type = TSDB_DATA_TYPE_NCHAR;
-        pSchema[6].bytes = 42;
+        pSchema[i].type = TSDB_DATA_TYPE_NCHAR;
+        pSchema[i].bytes = 42;
       } break;
       case 7: {
-        pSchema[7].type = TSDB_DATA_TYPE_TINYINT;
-        pSchema[7].bytes = TYPE_BYTES[pSchema[7].type];
+        pSchema[i].type = TSDB_DATA_TYPE_TINYINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 8: {
-        pSchema[8].type = TSDB_DATA_TYPE_SMALLINT;
-        pSchema[8].bytes = TYPE_BYTES[pSchema[8].type];
+        pSchema[i].type = TSDB_DATA_TYPE_SMALLINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 9: {
-        pSchema[9].type = TSDB_DATA_TYPE_BOOL;
-        pSchema[9].bytes = TYPE_BYTES[pSchema[9].type];
+        pSchema[i].type = TSDB_DATA_TYPE_BOOL;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 10: {
-        pSchema[10].type = TSDB_DATA_TYPE_UTINYINT;
-        pSchema[10].bytes = TYPE_BYTES[pSchema[10].type];
+        pSchema[i].type = TSDB_DATA_TYPE_UTINYINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 11: {
-        pSchema[11].type = TSDB_DATA_TYPE_USMALLINT;
-        pSchema[11].bytes = TYPE_BYTES[pSchema[11].type];
+        pSchema[i].type = TSDB_DATA_TYPE_USMALLINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 12: {
-        pSchema[12].type = TSDB_DATA_TYPE_UINT;
-        pSchema[12].bytes = TYPE_BYTES[pSchema[12].type];
+        pSchema[i].type = TSDB_DATA_TYPE_UINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
       case 13: {
-        pSchema[13].type = TSDB_DATA_TYPE_UBIGINT;
-        pSchema[13].bytes = TYPE_BYTES[pSchema[13].type];
+        pSchema[i].type = TSDB_DATA_TYPE_UBIGINT;
+        pSchema[i].bytes = TYPE_BYTES[pSchema[i].type];
       } break;
 
       default:
@@ -146,9 +146,9 @@ static int32_t genTestData(const char **data, int16_t nCols, SArray **pArray) {
       case 0:
         sscanf(data[i], "%" PRIi64, &colVal.value.ts);
         break;
-      case 1: {
+      case 1:
         sscanf(data[i], "%" PRIi32, &colVal.value.i32);
-      } break;
+        break;
       case 2:
         sscanf(data[i], "%" PRIi64, &colVal.value.i64);
         break;
@@ -274,9 +274,6 @@ int32_t debugPrintSColVal(SColVal *cv, int8_t type) {
     case TSDB_DATA_TYPE_MEDIUMBLOB:
       printf("MedBLOB ");
       break;
-    // case TSDB_DATA_TYPE_BINARY:
-    //   printf("BINARY ");
-    //   break;
     case TSDB_DATA_TYPE_MAX:
       printf("UNDEF ");
       break;
@@ -404,7 +401,8 @@ static void checkTSRow(const char **data, STSRow2 *row, STSchema *pTSchema) {
 }
 
 TEST(testCase, AllNormTest) {
-  int16_t  nCols = 1;
+  int16_t       nCols = 14;
+  STSRowBuilder rb = {0};
   STSRow2      *row = nullptr;
   SArray       *pArray = taosArrayInit(nCols, sizeof(SColVal));
   EXPECT_NE(pArray, nullptr);
@@ -414,15 +412,16 @@ TEST(testCase, AllNormTest) {
 
   // ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 binary(10), c6 nchar(10), c7 tinyint, c8 smallint,
   // c9 bool
-  char *data[10] = {"1653694220000", "10", "20", "10.1", "10.1", "binary10", "nchar10", "10", "10", "1"};
+  char *data[14] = {"1653694220000", "no", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "no", "no", "no", "no", "no"};
 
   genTestData((const char **)&data, nCols, &pArray);
 
-  tTSRowNew(NULL, pArray, pTSchema, &row);
+  tTSRowNew(&rb, pArray, pTSchema, &row);
 
   debugPrintTSRow(row, pTSchema, __func__, __LINE__);
   checkTSRow((const char **)&data, row, pTSchema);
 
+  tsRowBuilderClear(&rb);
   taosArrayDestroy(pArray);
   taosMemoryFree(pTSchema);
 }
@@ -443,12 +442,12 @@ TEST(testCase, NoneTest) {
   const char *data[nRows][nCols] = {
       {"1653694220000", "no", "20", "10.1", "10.1", "binary10", "no", "10", "10", "nu", "10", "20", "30", "40"},
       {"1653694220001", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no"},
-      {"1653694220002", "no", "no", "no", "no", "no", "nu", "no", "no", "no", "no", "no", "no", "nu"},
-      {"1653694220003", "nu", "no", "no", "no", "no", "nu", "no", "no", "no", "no", "no", "no", "no"},
+      {"1653694220002", "10", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no"},
+      {"1653694220003", "10", "10", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no", "no"},
       {"1653694220004", "no", "20", "no", "no", "no", "nchar10", "no", "no", "no", "no", "no", "no", "no"},
       {"1653694220005", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu"},
       {"1653694220006", "no", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu"},
-      {"1653694220007", "no", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "no"},
+      {"1653694220007", "no", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "no", "no", "no", "no", "no"},
       {"1653694220008", "no", "nu", "nu", "nu", "binary10", "nu", "nu", "nu", "nu", "nu", "nu", "nu", "no"},
       {"1653694220009", "no", "nu", "nu", "nu", "binary10", "nu", "nu", "10", "no", "nu", "nu", "nu", "100"},
       {"1653694220010", "-1", "-1", "-1", "-1", "binary10", "nu", "-1", "0", "0", "0", "0", "0", "0"},
@@ -465,13 +464,12 @@ TEST(testCase, NoneTest) {
       {"1653694220019", "no", "9223372036854775807", "nu", "nu", "bin10", "nu", "nu", "10", "no", "254", "nu", "nu",
        "no"}};
 
-
   for (int r = 0; r < nRows; ++r) {
     genTestData((const char **)&data[r], nCols, &pArray);
     tTSRowNew(NULL, pArray, pTSchema, &row);
     debugPrintTSRow(row, pTSchema, __func__, __LINE__);  // debug print
     checkTSRow((const char **)&data[r], row, pTSchema);  // check
-    taosMemoryFreeClear(row);
+    tTSRowFree(row);
     taosArrayClear(pArray);
   }
 

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

@@ -186,7 +186,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p
     SVgObj* pVgroup;
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
     if (pIter == NULL) break;
-    if (pVgroup->dbUid != pStream->dbUid) {
+    if (strcmp(pVgroup->dbName, pStream->targetDb) != 0) {
       sdbRelease(pSdb, pVgroup);
       continue;
     }
@@ -286,7 +286,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
   pStream->tasks = taosArrayInit(totLevel, sizeof(void*));
 
   bool hasExtraSink = false;
-  if (totLevel == 2 || strcmp(pStream->sourceDb, pStream->targetDb) != 0) {
+  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
@@ -405,7 +406,6 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         if (pStream->fixedSinkVgId == 0) {
           pTask->dispatchType = TASK_DISPATCH__SHUFFLE;
 
-          /*pTask->dispatchMsgType = TDMT_VND_TASK_WRITE_EXEC;*/
           pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
           SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
           ASSERT(pDb);
@@ -426,10 +426,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
             SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);
             for (int32_t j = 0; j < sinkLvSize; j++) {
               SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
-              /*printf("vgid %d node id %d\n", pVgInfo->vgId, pTask->nodeId);*/
               if (pLastLevelTask->nodeId == pVgInfo->vgId) {
                 pVgInfo->taskId = pLastLevelTask->taskId;
-                /*printf("taskid %d set to %d\n", pVgInfo->taskId, pTask->taskId);*/
                 break;
               }
             }

source/libs/catalog/inc/catalogInt.h

@@ -67,6 +67,7 @@ typedef enum {
   CTG_TASK_GET_DB_INFO,
   CTG_TASK_GET_TB_META,
   CTG_TASK_GET_TB_HASH,
+  CTG_TASK_GET_TB_INDEX,
   CTG_TASK_GET_INDEX,
   CTG_TASK_GET_UDF,
   CTG_TASK_GET_USER,
@@ -93,6 +94,10 @@ typedef struct SCtgTbMetaCtx {
   int32_t flag;
 } SCtgTbMetaCtx;
 
+typedef struct SCtgTbIndexCtx {
+  SName* pName;
+} SCtgTbIndexCtx;
+
 typedef struct SCtgDbVgCtx {
   char dbFName[TSDB_DB_FNAME_LEN];
 } SCtgDbVgCtx;
@@ -189,6 +194,7 @@ typedef struct SCtgJob {
   int32_t          indexNum;
   int32_t          userNum;
   int32_t          dbInfoNum;
+  int32_t          tbIndexNum;
 } SCtgJob;
 
 typedef struct SCtgMsgCtx {
@@ -490,7 +496,7 @@ int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutpu
 int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask);
 int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
 int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *out, SCtgTask* pTask);
-int32_t ctgGetTbIndexFromMnode(CTG_PARAMS, const char *tbFName, SArray** out, SCtgTask* pTask);
+int32_t ctgGetTbIndexFromMnode(CTG_PARAMS, SName* name, SArray** out, SCtgTask* pTask);
 int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
 int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
 int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);

source/libs/catalog/src/catalog.c

@@ -1136,14 +1136,14 @@ int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps
   CTG_API_LEAVE(ctgGetIndexInfoFromMnode(CTG_PARAMS_LIST(), indexName, pInfo, NULL));
 }
 
-int32_t catalogGetTableIndex(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const char* tbFName, SArray** pRes) {
+int32_t catalogGetTableIndex(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pRes) {
   CTG_API_ENTER();
   
-  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == tbFName || NULL == pRes) {
+  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName || NULL == pRes) {
     CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
-  CTG_API_LEAVE(ctgGetTbIndexFromMnode(CTG_PARAMS_LIST(), tbFName, pRes, NULL));
+  CTG_API_LEAVE(ctgGetTbIndexFromMnode(CTG_PARAMS_LIST(), (SName*)pTableName, pRes, NULL));
 }
 
 

source/libs/catalog/src/ctgAsync.c

@@ -44,7 +44,7 @@ int32_t ctgInitGetTbMetaTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:0x%" PRIx64 " the %d task type %s initialized, tableName:%s", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type), name->tname);
+  qDebug("QID:0x%" PRIx64 " the %d task type %s initialized, tbName:%s", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type), name->tname);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -232,6 +232,35 @@ int32_t ctgInitGetUserTask(SCtgJob *pJob, int32_t taskIdx, SUserAuthInfo *user)
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgInitGetTbIndexTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
+  SCtgTask task = {0};
+
+  task.type = CTG_TASK_GET_TB_INDEX;
+  task.taskId = taskIdx;
+  task.pJob = pJob;
+
+  task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgTbIndexCtx));
+  if (NULL == task.taskCtx) {
+    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+  }
+
+  SCtgTbIndexCtx* ctx = task.taskCtx;
+  ctx->pName = taosMemoryMalloc(sizeof(*name));
+  if (NULL == ctx->pName) {
+    taosMemoryFree(task.taskCtx);
+    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+  }
+  
+  memcpy(ctx->pName, name, sizeof(*name));
+
+  taosArrayPush(pJob->pTasks, &task);
+
+  qDebug("QID:%" PRIx64 " the %d task type %s initialized, tbName:%s", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type), name->tname);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t ctgHandleForceUpdate(SCatalog* pCtg, SCtgJob *pJob, const SCatalogReq* pReq) {
   int32_t dbNum = pJob->dbCfgNum + pJob->dbVgNum + pJob->dbInfoNum;
   if (dbNum > 0) {
@@ -329,8 +358,9 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
   int32_t indexNum = (int32_t)taosArrayGetSize(pReq->pIndex);
   int32_t userNum = (int32_t)taosArrayGetSize(pReq->pUser);
   int32_t dbInfoNum = (int32_t)taosArrayGetSize(pReq->pDbInfo);
+  int32_t tbIndexNum = (int32_t)taosArrayGetSize(pReq->pTableIndex);
 
-  *taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dbCfgNum + indexNum + userNum + dbInfoNum;
+  *taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dbCfgNum + indexNum + userNum + dbInfoNum + tbIndexNum;
   if (*taskNum <= 0) {
     ctgDebug("Empty input for job, no need to retrieve meta, reqId:0x%" PRIx64, reqId);
     return TSDB_CODE_SUCCESS;
@@ -360,6 +390,7 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
   pJob->indexNum = indexNum;
   pJob->userNum = userNum;
   pJob->dbInfoNum = dbInfoNum;
+  pJob->tbIndexNum = tbIndexNum;
 
   pJob->pTasks = taosArrayInit(*taskNum, sizeof(SCtgTask));
 
@@ -398,6 +429,11 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
     CTG_ERR_JRET(ctgInitGetTbHashTask(pJob, taskIdx++, name));
   }
 
+  for (int32_t i = 0; i < tbIndexNum; ++i) {
+    SName* name = taosArrayGet(pReq->pTableIndex, i);
+    CTG_ERR_JRET(ctgInitGetTbIndexTask(pJob, taskIdx++, name));
+  }
+
   for (int32_t i = 0; i < indexNum; ++i) {
     char* indexName = taosArrayGet(pReq->pIndex, i);
     CTG_ERR_JRET(ctgInitGetIndexTask(pJob, taskIdx++, indexName));
@@ -479,6 +515,21 @@ int32_t ctgDumpTbHashRes(SCtgTask* pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgDumpTbIndexRes(SCtgTask* pTask) {
+  SCtgJob* pJob = pTask->pJob;
+  if (NULL == pJob->jobRes.pTableIndex) {
+    pJob->jobRes.pTableIndex = taosArrayInit(pJob->tbIndexNum, sizeof(SMetaRes));
+    if (NULL == pJob->jobRes.pTableIndex) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+  }
+
+  SMetaRes res = {.code = pTask->code, .pRes = pTask->res};
+  taosArrayPush(pJob->jobRes.pTableHash, &res);
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t ctgDumpIndexRes(SCtgTask* pTask) {
   SCtgJob* pJob = pTask->pJob;
   if (NULL == pJob->jobRes.pIndex) {
@@ -817,6 +868,20 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgHandleGetTbIndexRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) {
+  int32_t code = 0;
+  CTG_ERR_JRET(ctgProcessRspMsg(pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
+
+  TSWAP(pTask->res, pTask->msgCtx.out);
+  
+_return:
+
+  ctgHandleTaskEnd(pTask, code);
+
+  CTG_RET(code);
+}
+
+
 int32_t ctgHandleGetDbCfgRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) {
   int32_t code = 0;
   CTG_ERR_JRET(ctgProcessRspMsg(pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
@@ -1056,13 +1121,24 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgLaunchGetTbIndexTask(SCtgTask *pTask) {
+  int32_t code = 0;
+  SCatalog* pCtg = pTask->pJob->pCtg; 
+  void *pTrans = pTask->pJob->pTrans; 
+  const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
+  SCtgTbIndexCtx* pCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
+  
+  CTG_ERR_RET(ctgGetTbIndexFromMnode(CTG_PARAMS_LIST(), pCtx->pName, NULL, pTask));
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t ctgLaunchGetQnodeTask(SCtgTask *pTask) {
   SCatalog* pCtg = pTask->pJob->pCtg; 
   void *pTrans = pTask->pJob->pTrans; 
   const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
 
   CTG_ERR_RET(ctgGetQnodeListFromMnode(CTG_PARAMS_LIST(), NULL, pTask));
-
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1174,6 +1250,7 @@ SCtgAsyncFps gCtgAsyncFps[] = {
   {ctgLaunchGetDbInfoTask,  ctgHandleGetDbInfoRsp,  ctgDumpDbInfoRes},
   {ctgLaunchGetTbMetaTask,  ctgHandleGetTbMetaRsp,  ctgDumpTbMetaRes},
   {ctgLaunchGetTbHashTask,  ctgHandleGetTbHashRsp,  ctgDumpTbHashRes},
+  {ctgLaunchGetTbIndexTask, ctgHandleGetTbIndexRsp, ctgDumpTbIndexRes},
   {ctgLaunchGetIndexTask,   ctgHandleGetIndexRsp,   ctgDumpIndexRes},
   {ctgLaunchGetUdfTask,     ctgHandleGetUdfRsp,     ctgDumpUdfRes},
   {ctgLaunchGetUserTask,    ctgHandleGetUserRsp,    ctgDumpUserRes},

source/libs/catalog/src/ctgRemote.c

@@ -427,11 +427,13 @@ int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t ctgGetTbIndexFromMnode(CTG_PARAMS, const char *tbFName, SArray** out, SCtgTask* pTask) {
+int32_t ctgGetTbIndexFromMnode(CTG_PARAMS, SName *name, SArray** out, SCtgTask* pTask) {
   char *msg = NULL;
   int32_t msgLen = 0;
   int32_t reqType = TDMT_MND_GET_TABLE_INDEX;
   void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
+  char tbFName[TSDB_TABLE_FNAME_LEN];
+  tNameExtractFullName(name, tbFName);
 
   ctgDebug("try to get tb index from mnode, tbFName:%s", tbFName);
 

source/libs/catalog/src/ctgUtil.c

@@ -60,6 +60,9 @@ void ctgFreeSMetaData(SMetaData* pData) {
   taosArrayDestroy(pData->pTableHash);
   pData->pTableHash = NULL;
 
+  taosArrayDestroy(pData->pTableIndex);
+  pData->pTableIndex = NULL;
+  
   taosArrayDestroy(pData->pUdfList);
   pData->pUdfList = NULL;
 
@@ -248,6 +251,14 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
       taosMemoryFreeClear(pCtx->out);
       break;
     }
+    case TDMT_MND_GET_TABLE_INDEX: {
+      SArray** pOut = (SArray**)pCtx->out;
+      if (pOut) {
+        taosArrayDestroyEx(*pOut, tFreeSTableIndexInfo);
+        taosMemoryFreeClear(pCtx->out);
+      }
+      break;
+    }
     case TDMT_MND_RETRIEVE_FUNC: {
       SFuncInfo* pOut = (SFuncInfo*)pCtx->out;
       taosMemoryFree(pOut->pCode);
@@ -344,6 +355,13 @@ void ctgFreeTask(SCtgTask* pTask) {
       taosMemoryFreeClear(pTask->res);
       break;
     }
+    case CTG_TASK_GET_TB_INDEX: {
+      SCtgTbIndexCtx* taskCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
+      taosMemoryFreeClear(taskCtx->pName);
+      taosMemoryFreeClear(pTask->taskCtx);
+      taosArrayDestroyEx(pTask->res, tFreeSTableIndexInfo);
+      break;
+    }
     case CTG_TASK_GET_INDEX: {
       taosMemoryFreeClear(pTask->taskCtx);
       taosMemoryFreeClear(pTask->res);

source/libs/command/inc/commandInt.h

@@ -32,15 +32,17 @@ extern "C" {
 #define EXPLAIN_PROJECTION_FORMAT "Projection"
 #define EXPLAIN_JOIN_FORMAT "%s"
 #define EXPLAIN_AGG_FORMAT "Aggragate"
+#define EXPLAIN_INDEF_ROWS_FORMAT "Indefinite Rows Function"
 #define EXPLAIN_EXCHANGE_FORMAT "Data Exchange %d:1"
 #define EXPLAIN_SORT_FORMAT "Sort"
 #define EXPLAIN_INTERVAL_FORMAT "Interval on Column %s"
+#define EXPLAIN_FILL_FORMAT "Fill"
 #define EXPLAIN_SESSION_FORMAT "Session"
 #define EXPLAIN_STATE_WINDOW_FORMAT "StateWindow on Column %s"
 #define EXPLAIN_PARITION_FORMAT "Partition on Column %s"
 #define EXPLAIN_ORDER_FORMAT "Order: %s"
 #define EXPLAIN_FILTER_FORMAT "Filter: "
-#define EXPLAIN_FILL_FORMAT "Fill: %s"
+#define EXPLAIN_FILL_VALUE_FORMAT "Fill Values: "
 #define EXPLAIN_ON_CONDITIONS_FORMAT "Join Cond: "
 #define EXPLAIN_TIMERANGE_FORMAT "Time Range: [%" PRId64 ", %" PRId64 "]"
 #define EXPLAIN_OUTPUT_FORMAT "Output: "
@@ -66,6 +68,8 @@ extern "C" {
 #define EXPLAIN_WIDTH_FORMAT "width=%d"
 #define EXPLAIN_FUNCTIONS_FORMAT "functions=%d"
 #define EXPLAIN_EXECINFO_FORMAT "cost=%.3f..%.3f rows=%" PRIu64
+#define EXPLAIN_MODE_FORMAT "mode=%s"
+#define EXPLAIN_STRING_TYPE_FORMAT "%s"
 
 typedef struct SExplainGroup {
   int32_t   nodeNum;

source/libs/command/src/explain.c

@@ -179,6 +179,21 @@ int32_t qExplainGenerateResChildren(SPhysiNode *pNode, SExplainGroup *group, SNo
       pPhysiChildren = mergePhysiNode->node.pChildren;
       break;
     }
+    case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: {
+      SIndefRowsFuncPhysiNode *indefPhysiNode = (SIndefRowsFuncPhysiNode *)pNode;
+      pPhysiChildren = indefPhysiNode->node.pChildren;
+      break;
+    }
+    case QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL: {
+      SMergeIntervalPhysiNode *intPhysiNode = (SMergeIntervalPhysiNode *)pNode;
+      pPhysiChildren = intPhysiNode->window.node.pChildren;
+      break;
+    }
+    case QUERY_NODE_PHYSICAL_PLAN_FILL: {
+      SFillPhysiNode *fillPhysiNode = (SFillPhysiNode *)pNode;
+      pPhysiChildren = fillPhysiNode->node.pChildren;
+      break;
+    }
     default:
       qError("not supported physical node type %d", pNode->type);
       QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
@@ -212,12 +227,15 @@ int32_t qExplainGenerateResNodeExecInfo(SArray **pExecInfo, SExplainGroup *group
   SExplainRsp *rsp = NULL;
   for (int32_t i = 0; i < group->nodeNum; ++i) {
     rsp = taosArrayGet(group->nodeExecInfo, i);
+/*
     if (group->physiPlanExecIdx >= rsp->numOfPlans) {
       qError("physiPlanIdx %d exceed plan num %d", group->physiPlanExecIdx, rsp->numOfPlans);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
     taosArrayPush(*pExecInfo, rsp->subplanInfo + group->physiPlanExecIdx);
+*/
+    taosArrayPush(*pExecInfo, rsp->subplanInfo);
   }
 
   ++group->physiPlanExecIdx;
@@ -599,6 +617,42 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
       }
       break;
     }
+    case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC: {
+      SIndefRowsFuncPhysiNode *pIndefNode = (SIndefRowsFuncPhysiNode *)pNode;
+      EXPLAIN_ROW_NEW(level, EXPLAIN_INDEF_ROWS_FORMAT);
+      EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
+      if (pResNode->pExecInfo) {
+        QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      }
+      if (pIndefNode->pVectorFuncs) {
+        EXPLAIN_ROW_APPEND(EXPLAIN_FUNCTIONS_FORMAT, pIndefNode->pVectorFuncs->length);
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      }
+      EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIndefNode->node.pOutputDataBlockDesc->totalRowSize);
+      EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
+      EXPLAIN_ROW_END();
+      QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
+
+      if (verbose) {
+        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_OUTPUT_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT,
+                           nodesGetOutputNumFromSlotList(pIndefNode->node.pOutputDataBlockDesc->pSlots));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIndefNode->node.pOutputDataBlockDesc->outputRowSize);
+        EXPLAIN_ROW_END();
+        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+
+        if (pIndefNode->node.pConditions) {
+          EXPLAIN_ROW_NEW(level + 1, EXPLAIN_FILTER_FORMAT);
+          QRY_ERR_RET(nodesNodeToSQL(pIndefNode->node.pConditions, tbuf + VARSTR_HEADER_SIZE,
+                                     TSDB_EXPLAIN_RESULT_ROW_SIZE, &tlen));
+          EXPLAIN_ROW_END();
+          QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        }
+      }
+      break;
+    }
     case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE: {
       SExchangePhysiNode *pExchNode = (SExchangePhysiNode *)pNode;
       SExplainGroup      *group = taosHashGet(ctx->groupHash, &pExchNode->srcGroupId, sizeof(pExchNode->srcGroupId));
@@ -607,7 +661,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
       }
 
-      EXPLAIN_ROW_NEW(level, EXPLAIN_EXCHANGE_FORMAT, group->nodeNum);
+      EXPLAIN_ROW_NEW(level, EXPLAIN_EXCHANGE_FORMAT, pExchNode->singleChannel ? 1 : group->nodeNum);
       EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
       if (pResNode->pExecInfo) {
         QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
@@ -750,6 +804,106 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
       }
       break;
     }
+    case QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL: {
+      SMergeIntervalPhysiNode *pIntNode = (SMergeIntervalPhysiNode *)pNode;
+      EXPLAIN_ROW_NEW(level, EXPLAIN_INTERVAL_FORMAT, nodesGetNameFromColumnNode(pIntNode->window.pTspk));
+      EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
+      if (pResNode->pExecInfo) {
+        QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      }
+      EXPLAIN_ROW_APPEND(EXPLAIN_FUNCTIONS_FORMAT, pIntNode->window.pFuncs->length);
+      EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIntNode->window.node.pOutputDataBlockDesc->totalRowSize);
+      EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
+      EXPLAIN_ROW_END();
+      QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
+
+      if (verbose) {
+        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_OUTPUT_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT,
+                           nodesGetOutputNumFromSlotList(pIntNode->window.node.pOutputDataBlockDesc->pSlots));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIntNode->window.node.pOutputDataBlockDesc->outputRowSize);
+        EXPLAIN_ROW_END();
+        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        uint8_t precision = getIntervalPrecision(pIntNode);
+        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_TIME_WINDOWS_FORMAT,
+                        INVERAL_TIME_FROM_PRECISION_TO_UNIT(pIntNode->interval, pIntNode->intervalUnit, precision),
+                        pIntNode->intervalUnit, pIntNode->offset, getPrecisionUnit(precision),
+                        INVERAL_TIME_FROM_PRECISION_TO_UNIT(pIntNode->sliding, pIntNode->slidingUnit, precision),
+                        pIntNode->slidingUnit);
+        EXPLAIN_ROW_END();
+        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+
+        if (pIntNode->window.node.pConditions) {
+          EXPLAIN_ROW_NEW(level + 1, EXPLAIN_FILTER_FORMAT);
+          QRY_ERR_RET(nodesNodeToSQL(pIntNode->window.node.pConditions, tbuf + VARSTR_HEADER_SIZE,
+                                     TSDB_EXPLAIN_RESULT_ROW_SIZE, &tlen));
+          EXPLAIN_ROW_END();
+          QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        }
+      }
+      break;
+    }
+    case QUERY_NODE_PHYSICAL_PLAN_FILL: {
+      SFillPhysiNode *pFillNode = (SFillPhysiNode *)pNode;
+      EXPLAIN_ROW_NEW(level, EXPLAIN_FILL_FORMAT);
+      EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
+      if (pResNode->pExecInfo) {
+        QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      }
+      EXPLAIN_ROW_APPEND(EXPLAIN_MODE_FORMAT, nodesGetFillModeString(pFillNode->mode));
+      EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+      EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pFillNode->node.pOutputDataBlockDesc->totalRowSize);
+      EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
+      EXPLAIN_ROW_END();
+      QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
+
+      if (verbose) {
+        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_OUTPUT_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT,
+                           nodesGetOutputNumFromSlotList(pFillNode->node.pOutputDataBlockDesc->pSlots));
+        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+        EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pFillNode->node.pOutputDataBlockDesc->outputRowSize);
+        EXPLAIN_ROW_END();
+        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        if (pFillNode->pValues) {
+          SNodeListNode *pValues = (SNodeListNode*)pFillNode->pValues;
+          EXPLAIN_ROW_NEW(level + 1, EXPLAIN_FILL_VALUE_FORMAT);
+          SNode* tNode = NULL;
+          int32_t i = 0;
+          FOREACH(tNode, pValues->pNodeList) {
+            if (i) {
+              EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
+            }
+            SValueNode* tValue = (SValueNode*)tNode;
+            char *value = nodesGetStrValueFromNode(tValue);
+            EXPLAIN_ROW_APPEND(EXPLAIN_STRING_TYPE_FORMAT, value);
+            taosMemoryFree(value);
+            ++i;
+          }
+
+          EXPLAIN_ROW_END();
+          QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        }
+
+        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_TIMERANGE_FORMAT, pFillNode->timeRange.skey,
+                        pFillNode->timeRange.ekey);
+        EXPLAIN_ROW_END();
+        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+
+        if (pFillNode->node.pConditions) {
+          EXPLAIN_ROW_NEW(level + 1, EXPLAIN_FILTER_FORMAT);
+          QRY_ERR_RET(nodesNodeToSQL(pFillNode->node.pConditions, tbuf + VARSTR_HEADER_SIZE,
+                                     TSDB_EXPLAIN_RESULT_ROW_SIZE, &tlen));
+          EXPLAIN_ROW_END();
+          QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
+        }
+      }
+      break;
+    }
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: {
       SSessionWinodwPhysiNode *pSessNode = (SSessionWinodwPhysiNode *)pNode;
       EXPLAIN_ROW_NEW(level, EXPLAIN_SESSION_FORMAT);

source/libs/executor/inc/executorimpl.h

@@ -804,7 +804,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
 
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
-                                          STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo);
+                                          STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo, bool isStream);
 SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
     SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);
 SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
@@ -892,8 +892,8 @@ STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowI
 int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
     __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
 int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
-int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey);
-int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey);
+int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey,
+    SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t size);
 SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize);
 SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex);
 int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows,

source/libs/executor/src/executorimpl.c

@@ -1130,6 +1130,7 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
     pCtx->start.key = INT64_MIN;
     pCtx->end.key = INT64_MIN;
     pCtx->numOfParams = pExpr->base.numOfParams;
+    pCtx->increase = false;
 
     pCtx->param = pFunct->pParam;
     //    for (int32_t j = 0; j < pCtx->numOfParams; ++j) {
@@ -2008,11 +2009,19 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprI
         // the _wstartts needs to copy to 20 following rows, since the results of top-k expands to 20 different rows.
         SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
         char*            in = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
+        if (pCtx[j].increase) {
+          int64_t ts = *(int64_t*) in;
+          for (int32_t k = 0; k < pRow->numOfRows; ++k) {
+            colDataAppend(pColInfoData, pBlock->info.rows + k, (const char *)&ts, pCtx[j].resultInfo->isNullRes);
+            ts++;
+          }
+        } else {
           for (int32_t k = 0; k < pRow->numOfRows; ++k) {
             colDataAppend(pColInfoData, pBlock->info.rows + k, in, pCtx[j].resultInfo->isNullRes);
           }
         }
       }
+    }
 
     releaseBufPage(pBuf, page);
     pBlock->info.rows += pRow->numOfRows;
@@ -4676,7 +4685,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     }
 
     int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
-    pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pTaskInfo);
+    bool isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
+    pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pTaskInfo, isStream);
 
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL == type) {
     int32_t children = 8;
@@ -5339,7 +5349,8 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, size_t size) {
+int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t size) {
+  pSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
   pSup->keySize = sizeof(int64_t) + sizeof(TSKEY);
   pSup->pKeyBuf = taosMemoryCalloc(1, pSup->keySize);
   pSup->pResultRows = taosArrayInit(1024, size);
@@ -5358,15 +5369,11 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, size
   if (bufSize <= pageSize) {
     bufSize = pageSize * 4;
   }
-  return createDiskbasedBuf(&pSup->pResultBuf, pageSize, bufSize, pKey, TD_TMP_DIR_PATH);
-}
-
-int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey) {
-  return initStreamAggSupporter(pSup, pKey, sizeof(SResultWindowInfo));
-}
-
-int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey) {
-  return initStreamAggSupporter(pSup, pKey, sizeof(SStateWindowInfo));
+  int32_t code = createDiskbasedBuf(&pSup->pResultBuf, pageSize, bufSize, pKey, TD_TMP_DIR_PATH);
+  for(int32_t i = 0; i < numOfOutput; ++i) {
+    pCtx[i].pBuf = pSup->pResultBuf;
+  }
+  return code;
 }
 
 int64_t getSmaWaterMark(int64_t interval, double filesFactor) {

source/libs/executor/src/scanoperator.c

@@ -1756,6 +1756,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
   ;
   pInfo->readHandle = *pReadHandle;
   pInfo->curPos = 0;
+  pInfo->pFilterNode = pPhyNode->node.pConditions;
   pOperator->name = "TagScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
   pOperator->blocking = false;

source/libs/executor/src/timewindowoperator.c

@@ -1438,9 +1438,15 @@ static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SInt
   return needed;
 }
 
+void increaseTs(SqlFunctionCtx* pCtx) {
+  if (pCtx[0].pExpr->pExpr->_function.pFunctNode->funcType == FUNCTION_TYPE_WSTARTTS) {
+    pCtx[0].increase = true;
+  }
+}
+
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
-                                          STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo) {
+                                          STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo, bool isStream) {
   SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
   SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -1461,6 +1467,11 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
   int32_t code =
       initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str);
   
+  if (isStream) {
+    ASSERT(numOfCols > 0);
+    increaseTs(pInfo->binfo.pCtx);
+  }
+
   initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
 
   pInfo->invertible = allInvertible(pInfo->binfo.pCtx, numOfCols);
@@ -2128,6 +2139,8 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
   SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
   int32_t code = initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols,
       pResBlock, keyBufSize, pTaskInfo->id.str);
+  ASSERT(numOfCols > 0);
+  increaseTs(pInfo->binfo.pCtx);
   initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
@@ -2212,6 +2225,8 @@ int32_t initBiasicInfo(SOptrBasicInfo* pBasicInfo, SExprInfo* pExprInfo,
   for (int32_t i = 0; i < numOfCols; ++i) {
     pBasicInfo->pCtx[i].pBuf = NULL;
   }
+  ASSERT(numOfCols > 0);
+  increaseTs(pBasicInfo->pCtx);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2228,6 +2243,10 @@ void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int
   pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
 }
 
+int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
+  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SResultWindowInfo));
+}
+
 SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                                   SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId,
                                                   STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo) {
@@ -2244,8 +2263,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SEx
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
-  pInfo->streamAggSup.resultRowSize = getResultRowSize(pInfo->binfo.pCtx, numOfCols);
-  code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo");
+  
+  code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pInfo->binfo.pCtx, numOfCols);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -3097,6 +3116,10 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
   return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
 }
 
+int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
+  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SStateWindowInfo));
+}
+
 SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                 SExecTaskInfo* pTaskInfo) {
   SStreamStateWinodwPhysiNode* pStateNode = (SStreamStateWinodwPhysiNode*)pPhyNode;
@@ -3130,8 +3153,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
     goto _error;
   }
 
-  pInfo->streamAggSup.resultRowSize = getResultRowSize(pInfo->binfo.pCtx, numOfCols);
-  code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo");
+  code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pInfo->binfo.pCtx, numOfCols);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }

source/libs/function/inc/builtinsimpl.h

@@ -102,6 +102,8 @@ bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
 int32_t topFunction(SqlFunctionCtx *pCtx);
 int32_t bottomFunction(SqlFunctionCtx *pCtx);
 int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t topCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
+int32_t bottomCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
 
 bool getSpreadFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool spreadFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
@@ -126,7 +128,10 @@ int32_t getHistogramInfoSize();
 
 bool getHLLFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t hllFunction(SqlFunctionCtx* pCtx);
+int32_t hllFunctionMerge(SqlFunctionCtx* pCtx);
 int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t hllPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t getHLLInfoSize();
 
 bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);

source/libs/function/src/builtins.c

@@ -313,6 +313,7 @@ static int32_t translateApercentileImpl(SFunctionNode* pFunc, char* pErrBuf, int
 static int32_t translateApercentilePartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateApercentileImpl(pFunc, pErrBuf, len, true);
 }
+
 static int32_t translateApercentileMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateApercentileImpl(pFunc, pErrBuf, len, false);
 }
@@ -401,6 +402,7 @@ static int32_t translateSpreadImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
 static int32_t translateSpreadPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateSpreadImpl(pFunc, pErrBuf, len, true);
 }
+
 static int32_t translateSpreadMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateSpreadImpl(pFunc, pErrBuf, len, false);
 }
@@ -551,6 +553,7 @@ static int32_t translateHistogramImpl(SFunctionNode* pFunc, char* pErrBuf, int32
 static int32_t translateHistogramPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateHistogramImpl(pFunc, pErrBuf, len, true);
 }
+
 static int32_t translateHistogramMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return translateHistogramImpl(pFunc, pErrBuf, len, false);
 }
@@ -564,6 +567,28 @@ static int32_t translateHLL(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateHLLImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) {
+  if (1 != LIST_LENGTH(pFunc->pParameterList)) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  if (isPartial) {
+    pFunc->node.resType = (SDataType){.bytes = getHistogramInfoSize() + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
+  } else {
+    pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT};
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateHLLPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateHLLImpl(pFunc, pErrBuf, len, true);
+}
+
+static int32_t translateHLLMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateHLLImpl(pFunc, pErrBuf, len, false);
+}
+
 static bool validateStateOper(const SValueNode* pVal) {
   if (TSDB_DATA_TYPE_BINARY != pVal->node.resType.type) {
     return false;
@@ -1339,6 +1364,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .initFunc     = functionSetup,
     .processFunc  = topFunction,
     .finalizeFunc = topBotFinalize,
+    .combineFunc  = topCombine,
   },
   {
     .name = "bottom",
@@ -1348,7 +1374,8 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = functionSetup,
     .processFunc  = bottomFunction,
-    .finalizeFunc = topBotFinalize
+    .finalizeFunc = topBotFinalize,
+    .combineFunc  = bottomCombine,
   },
   {
     .name = "spread",
@@ -1478,6 +1505,28 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .getEnvFunc   = getHLLFuncEnv,
     .initFunc     = functionSetup,
     .processFunc  = hllFunction,
+    .finalizeFunc = hllFinalize,
+    .pPartialFunc = "_hyperloglog_partial",
+    .pMergeFunc   = "_hyperloglog_merge"
+  },
+  {
+    .name = "_hyperloglog_partial",
+    .type = FUNCTION_TYPE_HYPERLOGLOG_PARTIAL,
+    .classification = FUNC_MGT_AGG_FUNC,
+    .translateFunc = translateHLLPartial,
+    .getEnvFunc   = getHLLFuncEnv,
+    .initFunc     = functionSetup,
+    .processFunc  = hllFunction,
+    .finalizeFunc = hllPartialFinalize
+  },
+  {
+    .name = "_hyperloglog_merge",
+    .type = FUNCTION_TYPE_HYPERLOGLOG_MERGE,
+    .classification = FUNC_MGT_AGG_FUNC,
+    .translateFunc = translateHLLMerge,
+    .getEnvFunc   = getHLLFuncEnv,
+    .initFunc     = functionSetup,
+    .processFunc  = hllFunctionMerge,
     .finalizeFunc = hllFinalize
   },
   {

source/libs/function/src/builtinsimpl.c

@@ -1389,6 +1389,18 @@ void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuple
   }
 }
 
+void releaseSource(STuplePos* pPos) {
+  if (pPos->pageId == -1) {
+    return ;
+  }
+  // Todo(liuyao) relase row
+}
+
+void replaceTupleData(STuplePos* pDestPos, STuplePos* pSourcePos) {
+  releaseSource(pDestPos);
+  *pDestPos = *pSourcePos;
+}
+
 int32_t minMaxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t isMinFunc) {
   SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
   SMinmaxResInfo*      pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
@@ -1400,10 +1412,12 @@ int32_t minMaxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int3
     if (pSBuf->assign && 
         ( (((*(double*)&pDBuf->v) < (*(double*)&pSBuf->v)) ^ isMinFunc) || !pDBuf->assign ) ) {
       *(double*) &pDBuf->v = *(double*) &pSBuf->v;
+      replaceTupleData(&pDBuf->tuplePos, &pSBuf->tuplePos);
     }
   } else {
     if ( pSBuf->assign && ( ((pDBuf->v < pSBuf->v) ^ isMinFunc) || !pDBuf->assign ) ) {
       pDBuf->v = pSBuf->v;
+      replaceTupleData(&pDBuf->tuplePos, &pSBuf->tuplePos);
     }
   }
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
@@ -2856,7 +2870,6 @@ void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
 int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
   STopBotRes*          pRes = GET_ROWCELL_INTERBUF(pEntryInfo);
-  pEntryInfo->complete = true;
 
   int32_t type = pCtx->input.pData[0]->info.type;
   int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
@@ -2881,6 +2894,67 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   return pEntryInfo->numOfRes;
 }
 
+void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type,
+    bool isTopQuery) {
+  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
+  STopBotRes* pRes = getTopBotOutputInfo(pCtx);
+  int32_t     maxSize = pCtx->param[1].param.i;
+  STopBotResItem* pItems = pRes->pItems;
+  assert(pItems != NULL);
+
+  // not full yet
+  if (pEntryInfo->numOfRes < maxSize) {
+    STopBotResItem* pItem = &pItems[pEntryInfo->numOfRes];
+    pItem->v = pSourceItem->v;
+    pItem->uid = pSourceItem->uid;
+    pItem->tuplePos.pageId = -1;
+    replaceTupleData(&pItem->tuplePos, &pSourceItem->tuplePos);
+    pEntryInfo->numOfRes++;
+    taosheapsort((void*)pItems, sizeof(STopBotResItem), pEntryInfo->numOfRes, (const void*)&type, topBotResComparFn,
+                 !isTopQuery);
+  } else {  // replace the minimum value in the result
+    if ((isTopQuery && (
+        (IS_SIGNED_NUMERIC_TYPE(type) && pSourceItem->v.i > pItems[0].v.i) ||
+        (IS_UNSIGNED_NUMERIC_TYPE(type) && pSourceItem->v.u > pItems[0].v.u) ||
+        (IS_FLOAT_TYPE(type) && pSourceItem->v.d > pItems[0].v.d)))
+        || (!isTopQuery && (
+        (IS_SIGNED_NUMERIC_TYPE(type) && pSourceItem->v.i < pItems[0].v.i) ||
+        (IS_UNSIGNED_NUMERIC_TYPE(type) && pSourceItem->v.u < pItems[0].v.u) ||
+        (IS_FLOAT_TYPE(type) && pSourceItem->v.d < pItems[0].v.d))
+        )) {
+      // replace the old data and the coresponding tuple data
+      STopBotResItem* pItem = &pItems[0];
+      pItem->v = pSourceItem->v;
+      pItem->uid = pSourceItem->uid;
+
+      // save the data of this tuple by over writing the old data
+      replaceTupleData(&pItem->tuplePos, &pSourceItem->tuplePos);
+      taosheapadjust((void*)pItems, sizeof(STopBotResItem), 0, pEntryInfo->numOfRes - 1, (const void*)&type,
+                     topBotResComparFn, NULL, !isTopQuery);
+    }
+  }
+}
+
+int32_t topCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
+  int32_t type = pDestCtx->input.pData[0]->info.type;
+  SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
+  STopBotRes*          pSBuf = getTopBotOutputInfo(pSourceCtx);
+  for (int32_t i = 0; i < pSResInfo->numOfRes; i++) {
+    addResult(pDestCtx, pSBuf->pItems + i, type, true);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t bottomCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
+  int32_t type = pDestCtx->input.pData[0]->info.type;
+  SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
+  STopBotRes*          pSBuf = getTopBotOutputInfo(pSourceCtx);
+  for (int32_t i = 0; i < pSResInfo->numOfRes; i++) {
+    addResult(pDestCtx, pSBuf->pItems + i, type, false);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 bool getSpreadFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SSpreadInfo);
   return true;
@@ -3411,7 +3485,6 @@ int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 }
 
 int32_t histogramPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
-  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
   SHistoFuncInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
   int32_t resultBytes = getHistogramInfoSize();
   char *res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
@@ -3428,6 +3501,10 @@ int32_t histogramPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   return 1;
 }
 
+int32_t getHLLInfoSize() {
+  return (int32_t)sizeof(SHLLInfo);
+}
+
 bool getHLLFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SHLLInfo);
   return true;
@@ -3553,6 +3630,27 @@ int32_t hllFunction(SqlFunctionCtx *pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t hllFunctionMerge(SqlFunctionCtx *pCtx) {
+  SInputColumnInfoData* pInput = &pCtx->input;
+  SColumnInfoData* pCol = pInput->pData[0];
+  ASSERT(pCol->info.type == TSDB_DATA_TYPE_BINARY);
+
+  SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+
+  int32_t start = pInput->startRowIndex;
+  char* data = colDataGetData(pCol, start);
+  SHLLInfo* pInputInfo = (SHLLInfo *)varDataVal(data);
+
+  for (int32_t k = 0; k < HLL_BUCKETS; ++k) {
+    if (pInfo->buckets[k] < pInputInfo->buckets[k]) {
+      pInfo->buckets[k] = pInputInfo->buckets[k];
+    }
+  }
+
+  SET_VAL(GET_RES_INFO(pCtx), 1, 1);
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SResultRowEntryInfo *pInfo = GET_RES_INFO(pCtx);
 
@@ -3565,6 +3663,24 @@ int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   return functionFinalize(pCtx, pBlock);
 }
 
+int32_t hllPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+  int32_t resultBytes = getHLLInfoSize();
+  char *res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
+
+  memcpy(varDataVal(res), pInfo, resultBytes);
+  varDataSetLen(res, resultBytes);
+
+  int32_t          slotId = pCtx->pExpr->base.resSchema.slotId;
+  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+
+  colDataAppend(pCol, pBlock->info.rows, res, false);
+
+  taosMemoryFree(res);
+  return pResInfo->numOfRes;
+}
+
 bool getStateFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SStateInfo);
   return true;

source/libs/planner/src/planPhysiCreater.c

@@ -1186,6 +1186,7 @@ static int32_t createExchangePhysiNodeByMerge(SMergePhysiNode* pMerge) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   pExchange->srcGroupId = pMerge->srcGroupId;
+  pExchange->singleChannel = true;
   pExchange->node.pParent = (SPhysiNode*)pMerge;
   pExchange->node.pOutputDataBlockDesc = nodesCloneNode(pMerge->node.pOutputDataBlockDesc);
   if (NULL == pExchange->node.pOutputDataBlockDesc) {

source/libs/scalar/src/sclvector.c

@@ -348,7 +348,7 @@ static FORCE_INLINE void varToNchar(char* buf, SScalarParam* pOut, int32_t rowIn
   int32_t outputMaxLen = (inputLen + 1) * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
 
   char* t = taosMemoryCalloc(1, outputMaxLen);
-  /*int32_t resLen = */taosMbsToUcs4(varDataVal(buf), inputLen, (TdUcs4*) varDataVal(t), outputMaxLen, &len);
+  /*int32_t resLen = */taosMbsToUcs4(varDataVal(buf), inputLen, (TdUcs4*) varDataVal(t), outputMaxLen - VARSTR_HEADER_SIZE, &len);
   varDataSetLen(t, len);
 
   colDataAppend(pOut->columnData, rowIndex, t, false);

source/libs/wal/src/walWrite.c

@@ -253,6 +253,8 @@ int walRoll(SWal *pWal) {
 
 static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
   SWalIdxEntry entry = {.ver = ver, .offset = offset};
+  /*int64_t      idxOffset = taosLSeekFile(pWal->pWriteIdxTFile, 0, SEEK_CUR);*/
+  /*wDebug("write index: ver: %ld, offset: %ld, at %ld", ver, offset, idxOffset);*/
   int size = taosWriteFile(pWal->pWriteIdxTFile, &entry, sizeof(SWalIdxEntry));
   if (size != sizeof(SWalIdxEntry)) {
     terrno = TAOS_SYSTEM_ERROR(errno);

source/os/src/osFile.c

@@ -316,12 +316,12 @@ int64_t taosCloseFile(TdFilePtr *ppFile) {
     (*ppFile)->fp = NULL;
   }
   if ((*ppFile)->fd >= 0) {
-  #ifdef WINDOWS
+#ifdef WINDOWS
     HANDLE h = (HANDLE)_get_osfhandle((*ppFile)->fd);
     !FlushFileBuffers(h);
-  #else
+#else
     fsync((*ppFile)->fd);
-  #endif
+#endif
     close((*ppFile)->fd);
     (*ppFile)->fd = -1;
   }
@@ -345,11 +345,11 @@ int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
   char   *tbuf = (char *)buf;
 
   while (leftbytes > 0) {
-  #ifdef WINDOWS
+#ifdef WINDOWS
     readbytes = _read(pFile->fd, (void *)tbuf, (uint32_t)leftbytes);
-  #else
+#else
     readbytes = read(pFile->fd, (void *)tbuf, (uint32_t)leftbytes);
-  #endif
+#endif
     if (readbytes < 0) {
       if (errno == EINTR) {
         continue;
@@ -433,9 +433,6 @@ int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
 }
 
 int64_t taosLSeekFile(TdFilePtr pFile, int64_t offset, int32_t whence) {
-  if (pFile == NULL) {
-    return 0;
-  }
 #if FILE_WITH_LOCK
   taosThreadRwlockRdlock(&(pFile->rwlock));
 #endif
@@ -565,12 +562,12 @@ int32_t taosFsyncFile(TdFilePtr pFile) {
 
   if (pFile->fp != NULL) return fflush(pFile->fp);
   if (pFile->fd >= 0) {
-  #ifdef WINDOWS
+#ifdef WINDOWS
     HANDLE h = (HANDLE)_get_osfhandle(pFile->fd);
     return !FlushFileBuffers(h);
-  #else
+#else
     return fsync(pFile->fd);
-  #endif
+#endif
   }
   return 0;
 }

source/os/src/osTime.c

@@ -28,29 +28,29 @@
 
 #ifdef WINDOWS
 
-#include <time.h>  
 #include <stdlib.h>
 #include <string.h>
+#include <time.h>
 //#define TM_YEAR_BASE 1970 //origin
-#define TM_YEAR_BASE 1900 //slguan
+#define TM_YEAR_BASE 1900  // slguan
 /*
-* We do not implement alternate representations. However, we always
-* check whether a given modifier is allowed for a certain conversion.
-*/
+ * We do not implement alternate representations. However, we always
+ * check whether a given modifier is allowed for a certain conversion.
+ */
 #define ALT_E 0x01
 #define ALT_O 0x02
-#define	LEGAL_ALT(x)		{ if (alt_format & ~(x)) return (0); }
-
+#define LEGAL_ALT(x)                   \
+  {                                    \
+    if (alt_format & ~(x)) return (0); \
+  }
 
-static int conv_num(const char **buf, int *dest, int llim, int ulim)
-{
+static int conv_num(const char **buf, int *dest, int llim, int ulim) {
   int result = 0;
 
   /* The limit also determines the number of valid digits. */
   int rulim = ulim;
 
-    if (**buf < '0' || **buf > '9')
-        return (0);
+  if (**buf < '0' || **buf > '9') return (0);
 
   do {
     result *= 10;
@@ -58,62 +58,18 @@ static int conv_num(const char **buf, int *dest, int llim, int ulim)
     rulim /= 10;
   } while ((result * 10 <= ulim) && rulim && **buf >= '0' && **buf <= '9');
 
-    if (result < llim || result > ulim)
-        return (0);
+  if (result < llim || result > ulim) return (0);
 
   *dest = result;
   return (1);
 }
 
-static const char *day[7] = {
-    "Sunday", "Monday", "Tuesday", "Wednesday", "Thursday",
-    "Friday", "Saturday"
-};
-static const char *abday[7] = {
-    "Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"
-};
-static const char *mon[12] = {
-    "January", "February", "March", "April", "May", "June", "July",
-    "August", "September", "October", "November", "December"
-};
-static const char *abmon[12] = {
-    "Jan", "Feb", "Mar", "Apr", "May", "Jun",
-    "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"
-};
-static const char *am_pm[2] = {
-    "AM", "PM"
-};
-
-#define BILLION                             (1E9)
-
-static BOOL g_first_time = 1;
-static LARGE_INTEGER g_counts_per_sec;
-
-int clock_gettime(int dummy, struct timespec *ct)
-{
-    LARGE_INTEGER count;
-
-    if (g_first_time)
-    {
-        g_first_time = 0;
-
-        if (0 == QueryPerformanceFrequency(&g_counts_per_sec))
-        {
-            g_counts_per_sec.QuadPart = 0;
-        }
-    }
-
-    if ((NULL == ct) || (g_counts_per_sec.QuadPart <= 0) ||
-            (0 == QueryPerformanceCounter(&count)))
-    {
-        return -1;
-    }
-
-    ct->tv_sec = count.QuadPart / g_counts_per_sec.QuadPart;
-    ct->tv_nsec = ((count.QuadPart % g_counts_per_sec.QuadPart) * BILLION) / g_counts_per_sec.QuadPart;
-
-    return 0;
-}
+static const char *day[7] = {"Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"};
+static const char *abday[7] = {"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"};
+static const char *mon[12] = {"January", "February", "March",     "April",   "May",      "June",
+                              "July",    "August",   "September", "October", "November", "December"};
+static const char *abmon[12] = {"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};
+static const char *am_pm[2] = {"AM", "PM"};
 
 #else
 #include <sys/time.h>
@@ -134,22 +90,19 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
 
     /* Eat up white-space. */
     if (isspace(c)) {
-            while (isspace(*bp))
-                bp++;
+      while (isspace(*bp)) bp++;
 
       fmt++;
       continue;
     }
 
-        if ((c = *fmt++) != '%')
-            goto literal;
-
+    if ((c = *fmt++) != '%') goto literal;
 
-    again:		switch (c = *fmt++) {
+  again:
+    switch (c = *fmt++) {
       case '%': /* "%%" is converted to "%". */
-        literal :
-            if (c != *bp++)
-                return (0);
+      literal:
+        if (c != *bp++) return (0);
         break;
 
         /*
@@ -171,44 +124,37 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
          */
       case 'c': /* Date and time, using the locale's format. */
         LEGAL_ALT(ALT_E);
-        if (!(bp = taosStrpTime(bp, "%x %X", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%x %X", tm))) return (0);
         break;
 
       case 'D': /* The date as "%m/%d/%y". */
         LEGAL_ALT(0);
-        if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm))) return (0);
         break;
 
       case 'R': /* The time as "%H:%M". */
         LEGAL_ALT(0);
-        if (!(bp = taosStrpTime(bp, "%H:%M", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%H:%M", tm))) return (0);
         break;
 
       case 'r': /* The time in 12-hour clock representation. */
         LEGAL_ALT(0);
-        if (!(bp = taosStrpTime(bp, "%I:%M:%S %p", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%I:%M:%S %p", tm))) return (0);
         break;
 
       case 'T': /* The time as "%H:%M:%S". */
         LEGAL_ALT(0);
-        if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm))) return (0);
         break;
 
       case 'X': /* The time, using the locale's format. */
         LEGAL_ALT(ALT_E);
-        if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm))) return (0);
         break;
 
       case 'x': /* The date, using the locale's format. */
         LEGAL_ALT(ALT_E);
-        if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
-            return (0);
+        if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm))) return (0);
         break;
 
         /*
@@ -220,18 +166,15 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
         for (i = 0; i < 7; i++) {
           /* Full name. */
           len = strlen(day[i]);
-            if (strncmp(day[i], bp, len) == 0)
-                break;
+          if (strncmp(day[i], bp, len) == 0) break;
 
           /* Abbreviated name. */
           len = strlen(abday[i]);
-            if (strncmp(abday[i], bp, len) == 0)
-                break;
+          if (strncmp(abday[i], bp, len) == 0) break;
         }
 
         /* Nothing matched. */
-        if (i == 7)
-            return (0);
+        if (i == 7) return (0);
 
         tm->tm_wday = i;
         bp += len;
@@ -244,18 +187,15 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
         for (i = 0; i < 12; i++) {
           /* Full name. */
           len = strlen(mon[i]);
-            if (strncmp(mon[i], bp, len) == 0)
-                break;
+          if (strncmp(mon[i], bp, len) == 0) break;
 
           /* Abbreviated name. */
           len = strlen(abmon[i]);
-            if (strncmp(abmon[i], bp, len) == 0)
-                break;
+          if (strncmp(abmon[i], bp, len) == 0) break;
         }
 
         /* Nothing matched. */
-        if (i == 12)
-            return (0);
+        if (i == 12) return (0);
 
         tm->tm_mon = i;
         bp += len;
@@ -263,13 +203,11 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
 
       case 'C': /* The century number. */
         LEGAL_ALT(ALT_E);
-        if (!(conv_num(&bp, &i, 0, 99)))
-            return (0);
+        if (!(conv_num(&bp, &i, 0, 99))) return (0);
 
         if (split_year) {
           tm->tm_year = (tm->tm_year % 100) + (i * 100);
-        }
-        else {
+        } else {
           tm->tm_year = i * 100;
           split_year = 1;
         }
@@ -278,8 +216,7 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
       case 'd': /* The day of month. */
       case 'e':
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_mday, 1, 31)))
-            return (0);
+        if (!(conv_num(&bp, &tm->tm_mday, 1, 31))) return (0);
         break;
 
       case 'k': /* The hour (24-hour clock representation). */
@@ -287,8 +224,7 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
         /* FALLTHROUGH */
       case 'H':
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_hour, 0, 23)))
-            return (0);
+        if (!(conv_num(&bp, &tm->tm_hour, 0, 23))) return (0);
         break;
 
       case 'l': /* The hour (12-hour clock representation). */
@@ -296,29 +232,24 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
         /* FALLTHROUGH */
       case 'I':
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_hour, 1, 12)))
-            return (0);
-        if (tm->tm_hour == 12)
-            tm->tm_hour = 0;
+        if (!(conv_num(&bp, &tm->tm_hour, 1, 12))) return (0);
+        if (tm->tm_hour == 12) tm->tm_hour = 0;
         break;
 
       case 'j': /* The day of year. */
         LEGAL_ALT(0);
-        if (!(conv_num(&bp, &i, 1, 366)))
-            return (0);
+        if (!(conv_num(&bp, &i, 1, 366))) return (0);
         tm->tm_yday = i - 1;
         break;
 
       case 'M': /* The minute. */
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_min, 0, 59)))
-            return (0);
+        if (!(conv_num(&bp, &tm->tm_min, 0, 59))) return (0);
         break;
 
       case 'm': /* The month. */
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &i, 1, 12)))
-            return (0);
+        if (!(conv_num(&bp, &i, 1, 12))) return (0);
         tm->tm_mon = i - 1;
         break;
 
@@ -326,16 +257,14 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
         LEGAL_ALT(0);
         /* AM? */
         if (strcmp(am_pm[0], bp) == 0) {
-            if (tm->tm_hour > 11)
-                return (0);
+          if (tm->tm_hour > 11) return (0);
 
           bp += strlen(am_pm[0]);
           break;
         }
         /* PM? */
         else if (strcmp(am_pm[1], bp) == 0) {
-            if (tm->tm_hour > 11)
-                return (0);
+          if (tm->tm_hour > 11) return (0);
 
           tm->tm_hour += 12;
           bp += strlen(am_pm[1]);
@@ -347,8 +276,7 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
 
       case 'S': /* The seconds. */
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_sec, 0, 61)))
-            return (0);
+        if (!(conv_num(&bp, &tm->tm_sec, 0, 61))) return (0);
         break;
 
       case 'U': /* The week of year, beginning on sunday. */
@@ -360,28 +288,24 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
          * point to calculate a real value, so just check the
          * range for now.
          */
-        if (!(conv_num(&bp, &i, 0, 53)))
-            return (0);
+        if (!(conv_num(&bp, &i, 0, 53))) return (0);
         break;
 
       case 'w': /* The day of week, beginning on sunday. */
         LEGAL_ALT(ALT_O);
-        if (!(conv_num(&bp, &tm->tm_wday, 0, 6)))
-            return (0);
+        if (!(conv_num(&bp, &tm->tm_wday, 0, 6))) return (0);
         break;
 
       case 'Y': /* The year. */
         LEGAL_ALT(ALT_E);
-        if (!(conv_num(&bp, &i, 0, 9999)))
-            return (0);
+        if (!(conv_num(&bp, &i, 0, 9999))) return (0);
 
         tm->tm_year = i - TM_YEAR_BASE;
         break;
 
       case 'y': /* The year within 100 years of the epoch. */
         LEGAL_ALT(ALT_E | ALT_O);
-        if (!(conv_num(&bp, &i, 0, 99)))
-            return (0);
+        if (!(conv_num(&bp, &i, 0, 99))) return (0);
 
         if (split_year) {
           tm->tm_year = ((tm->tm_year / 100) * 100) + i;
@@ -400,16 +324,12 @@ char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
       case 'n': /* Any kind of white-space. */
       case 't':
         LEGAL_ALT(0);
-        while (isspace(*bp))
-            bp++;
+        while (isspace(*bp)) bp++;
         break;
 
-
       default: /* Unknown/unsupported conversion. */
         return (0);
     }
-
-
   }
 
   /* LINTED functional specification */
@@ -435,13 +355,9 @@ FORCE_INLINE int32_t taosGetTimeOfDay(struct timeval *tv) {
 #endif
 }
 
-time_t taosTime(time_t *t) {
-  return time(t);
-}
+time_t taosTime(time_t *t) { return time(t); }
 
-time_t taosMktime(struct tm *timep) {
-  return mktime(timep);
-}
+time_t taosMktime(struct tm *timep) { return mktime(timep); }
 
 struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
   if (result == NULL) {
@@ -456,5 +372,36 @@ struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
 }
 
 int32_t taosGetTimestampSec() { return (int32_t)time(NULL); }
-
-int32_t taosClockGetTime(int clock_id, struct timespec *pTS) { return clock_gettime(clock_id, pTS); }
+int32_t taosClockGetTime(int clock_id, struct timespec *pTS) {
+#ifdef WINDOWS
\ No newline at end of file
+  LARGE_INTEGER        t;
+  FILETIME             f;
+  static FILETIME      ff;
+  static SYSTEMTIME    ss;
+  static LARGE_INTEGER offset;
+
+  ss.wYear = 1970;
+  ss.wMonth = 1;
+  ss.wDay = 1;
+  ss.wHour = 0;
+  ss.wMinute = 0;
+  ss.wSecond = 0;
+  ss.wMilliseconds = 0;
+  SystemTimeToFileTime(&ss, &ff);
+  offset.QuadPart = ff.dwHighDateTime;
+  offset.QuadPart <<= 32;
+  offset.QuadPart |= ff.dwLowDateTime;
+
+  GetSystemTimeAsFileTime(&f);
+  t.QuadPart = f.dwHighDateTime;
+  t.QuadPart <<= 32;
+  t.QuadPart |= f.dwLowDateTime;
+
+  t.QuadPart -= offset.QuadPart;
+  pTS->tv_sec = t.QuadPart / 10000000;
+  pTS->tv_nsec = (t.QuadPart % 10000000)*100;
+  return (0);
+#else
+  return clock_gettime(clock_id, pTS);
+#endif
+}
\ No newline at end of file

tests/pytest/util/dnodes.py

@@ -574,6 +574,9 @@ class TDDnodes:
 
     def stopAll(self):
         tdLog.info("stop all dnodes")
+        if (not self.dnodes[0].remoteIP == ""):
+            self.dnodes[0].remoteExec(self.dnodes[0].cfgDict, "for i in range(len(tdDnodes.dnodes)):\n    tdDnodes.dnodes[i].running=1\ntdDnodes.stopAll()")
+            return
         for i in range(len(self.dnodes)):
             self.dnodes[i].stop()
 

tests/script/tsim/stream/session0.sim

@@ -17,7 +17,7 @@ sql use test
 
 
 sql create table t1(ts timestamp, a int, b int , c int, d double,id int);
-sql create stream streams2 trigger at_once  into streamt as select  _wstartts, count(*) c1,  sum(a), max(a), min(d), stddev(a), last(a), first(d), max(id) s from t1 session(ts,10s);
+sql create stream streams1 trigger at_once  into streamt as select  _wstartts, count(*) c1,  sum(a), max(a), min(d), stddev(a), last(a), first(d), max(id) s from t1 session(ts,10s);
 sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL,1);
 sql insert into t1 values(1648791223001,10,2,3,1.1,2);
 sql insert into t1 values(1648791233002,3,2,3,2.1,3);
@@ -176,4 +176,107 @@ if $data08 != 13 then
   return -1
 endi
 
+sql create database test2 vgroups 1;
+sql use test2;
+sql create table t2(ts timestamp, a int, b int , c int, d double, id int);
+sql create stream streams2 trigger at_once  watermark 1d into streamt2 as select  _wstartts,apercentile(a,30) c1, apercentile(a,70), apercentile(a,20,"t-digest") c2, apercentile(a,60,"t-digest") c3, max(id) c4 from t2 session(ts,10s);
+sql insert into t2 values(1648791213001,1,1,3,1.0,1);
+sql insert into t2 values(1648791213002,2,2,6,3.4,2);
+sql insert into t2 values(1648791213003,4,9,3,4.8,3);
+sql insert into t2 values(1648791233003,3,4,3,2.1,4);
+sql insert into t2 values(1648791233004,3,5,3,3.4,5);
+sql insert into t2 values(1648791233005,3,6,3,7.6,6);
+
+#
+sql insert into t2 values(1648791223003,20,7,3,10.1,7);
+
+$loop_count = 0
+loop2:
+sleep 300
+sql select * from streamt2 where c4=7;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 1 then
+  print ======$rows
+  goto loop2
+endi
+
+# row 0
+if $data01 != 2.091607978 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 3.274823935 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+if $data03 != 1.800000000 then
+  print ======$data03
+  return -1
+endi
+
+if $data04 != 3.350000000 then
+  print ======$data04
+  return -1
+endi
+
+sql create database test3 vgroups 1;
+sql use test3;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams3 trigger at_once  watermark 1d into streamt3 as select  _wstartts, min(b), a,c from t1 session(ts,10s);
+sql create stream streams4 trigger at_once  watermark 1d into streamt4 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
+sql create stream streams5 trigger at_once  watermark 1d into streamt5 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
+sql create stream streams6 trigger at_once  watermark 1d into streamt6 as select  _wstartts, max(b), a,c from t1 session(ts,10s);
+sql insert into t1 values(1648791213001,1,1,1,1.0);
+sql insert into t1 values(1648791213002,2,3,2,3.4);
+sql insert into t1 values(1648791213003,4,9,3,4.8);
+sql insert into t1 values(1648791213004,4,5,4,4.8);
+
+sql insert into t1 values(1648791233004,3,4,0,2.1);
+sql insert into t1 values(1648791233005,3,0,6,3.4);
+sql insert into t1 values(1648791233006,3,6,7,7.6);
+sql insert into t1 values(1648791233007,3,13,8,7.6);
+
+
+sql insert into t1 values(1648791223004,20,7,9,10.1);
+
+$loop_count = 0
+loop3:
+sleep 300
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sql select * from streamt3;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
+sql select * from streamt4;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
+sql select * from streamt5;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
+sql select * from streamt6;
+if $rows == 0 then
+  print ======$rows
+  goto loop3
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/system-test/0-others/taosdlog.py

@@ -23,7 +23,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/0-others/udf_cluster.py

@@ -37,7 +37,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/0-others/udf_create.py

@@ -9,8 +9,9 @@ from util.sql import *
 from util.cases import *
 from util.dnodes import *
 import subprocess
-# import win32gui
-# import threading
+if (platform.system().lower() == 'windows'):
+    import win32gui
+import threading
 
 class TDTestCase:
 
@@ -535,17 +536,18 @@ class TDTestCase:
         return udf1_sqls ,udf2_sqls
 
 
-    # def checkRunTimeError(self):
-    #     while 1:
-    #         time.sleep(1)
-    #         hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
-    #         if hwnd:
-    #             os.system("TASKKILL /F /IM udfd.exe")
+    def checkRunTimeError(self):
+        if (platform.system().lower() == 'windows' and tdDnodes.dnodes[0].remoteIP == ""):
+            while 1:
+                time.sleep(1)
+                hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
+                if hwnd:
+                    os.system("TASKKILL /F /IM udfd.exe")
 
     def unexpected_create(self):
-        # if (platform.system().lower() == 'windows' and tdDnodes.dnodes[0].remoteIP == ""):
-        #     checkErrorThread = threading.Thread(target=self.checkRunTimeError,daemon=True)
-        #     checkErrorThread.start()
+        if (platform.system().lower() == 'windows' and tdDnodes.dnodes[0].remoteIP == ""):
+            checkErrorThread = threading.Thread(target=self.checkRunTimeError,daemon=True)
+            checkErrorThread.start()
 
         tdLog.info(" create function with out bufsize ")
         tdSql.query("drop function udf1 ")

tests/system-test/0-others/user_control.py

@@ -3,6 +3,7 @@ import taos
 import time
 import inspect
 import traceback
+import socket
 from dataclasses  import dataclass
 
 from util.log import *
@@ -102,7 +103,7 @@ class TDconnect:
 
 
 def taos_connect(
-    host    = "127.0.0.1",
+    host    = socket.gethostname(),
     port    = 6030,
     user    = "root",
     passwd  = "taosdata",

tests/system-test/1-insert/insertWithMoreVgroup.py

@@ -54,7 +54,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root)-len("/build/bin")]

tests/system-test/1-insert/mutipythonnodebugtaosd.py

@@ -52,7 +52,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root)-len("/build/bin")]

tests/system-test/1-insert/test_stmt_insert_query_ex.py

@@ -49,7 +49,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root)-len("/build/bin")]

tests/system-test/1-insert/test_stmt_muti_insert_query.py

@@ -49,7 +49,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root)-len("/build/bin")]

tests/system-test/1-insert/test_stmt_set_tbname_tag.py

@@ -49,7 +49,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root)-len("/build/bin")]

tests/system-test/2-query/histogram.py

@@ -400,7 +400,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/2-query/timezone.py

@@ -3,7 +3,10 @@ from util.log import *
 from util.sql import *
 from util.cases import *
 
+import platform
 import os
+if platform.system().lower() == 'windows':
+    import tzlocal
 
 
 class TDTestCase:
@@ -15,8 +18,12 @@ class TDTestCase:
     def run(self):  # sourcery skip: extract-duplicate-method
         tdSql.prepare()
         # get system timezone
-        time_zone_arr = os.popen('timedatectl | grep zone').read(
-        ).strip().split(':')
+        if platform.system().lower() == 'windows':
+            time_zone_1 = tzlocal.get_localzone_name()
+            time_zone_2 = time.strftime('(UTC, %z)')
+            time_zone = time_zone_1 + " " + time_zone_2
+        else:
+            time_zone_arr = os.popen('timedatectl | grep zone').read().strip().split(':')
             if len(time_zone_arr) > 1:
                 time_zone = time_zone_arr[1].lstrip()
             else:

tests/system-test/7-tmq/basic5.py

@@ -13,6 +13,12 @@ from util.dnodes import *
 
 class TDTestCase:
     hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP)
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
     #rpcDebugFlagVal = '143'
     #clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
     #clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
@@ -34,7 +40,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]
@@ -192,6 +198,9 @@ class TDTestCase:
         
         shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
         shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        if (platform.system().lower() == 'windows'):
+            shellCmd += "> nul 2>&1 &"   
+        else:
             shellCmd += "> /dev/null 2>&1 &"  
         tdLog.info(shellCmd)
         os.system(shellCmd)        
@@ -306,6 +315,9 @@ class TDTestCase:
         
         shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
         shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        if (platform.system().lower() == 'windows'):
+            shellCmd += "> nul 2>&1 &"   
+        else:
             shellCmd += "> /dev/null 2>&1 &"  
         tdLog.info(shellCmd)
         os.system(shellCmd)        
@@ -438,6 +450,9 @@ class TDTestCase:
         
         shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
         shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        if (platform.system().lower() == 'windows'):
+            shellCmd += "> nul 2>&1 &"   
+        else:
             shellCmd += "> /dev/null 2>&1 &"        
         tdLog.info(shellCmd)
         os.system(shellCmd)        

tests/system-test/7-tmq/db.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/schema.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeDb.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeDb0.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeDb1.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb0.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb1.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb2.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb3.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/subscribeStb4.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/tmqDnode.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/7-tmq/tmqModule.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/99-TDcase/TD-15517.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/99-TDcase/TD-15554.py

@@ -33,7 +33,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/99-TDcase/TD-15557.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/99-TDcase/TD-15563.py

@@ -34,7 +34,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/99-TDcase/TD-16025.py

@@ -41,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]

tests/system-test/fulltest.bat

@@ -3,7 +3,7 @@ python3 .\test.py -f 0-others\taosShell.py
 python3 .\test.py -f 0-others\taosShellError.py
 python3 .\test.py -f 0-others\taosShellNetChk.py
 python3 .\test.py -f 0-others\telemetry.py
-python3 .\test.py -f 0-others\taosdMonitor.py
+@REM python3 .\test.py -f 0-others\taosdMonitor.py
 python3 .\test.py -f 0-others\udfTest.py
 python3 .\test.py -f 0-others\udf_create.py
 python3 .\test.py -f 0-others\udf_restart_taosd.py

tests/system-test/test.py

@@ -20,6 +20,7 @@ import time
 import base64
 import json
 import platform
+import socket
 from distutils.log import warn as printf
 from fabric2 import Connection
 sys.path.append("../pytest")
@@ -149,7 +150,7 @@ if __name__ == "__main__":
         tdLog.info('stop All dnodes')
     
     if masterIp == "":
-        host = '127.0.0.1'
+        host = socket.gethostname()
     else:
         try:
             config = eval(masterIp)
@@ -170,8 +171,8 @@ if __name__ == "__main__":
         try:
             if key_word in open(fileName, encoding='UTF-8').read():
                 is_test_framework = 1
-        except:
-            pass
+        except Exception as r:
+            print(r)
         updateCfgDictStr = ''
         if is_test_framework:
             moduleName = fileName.replace(".py", "").replace(os.sep, ".")
@@ -181,8 +182,8 @@ if __name__ == "__main__":
                 if ((json.dumps(updateCfgDict) == '{}') and (ucase.updatecfgDict is not None)):
                     updateCfgDict = ucase.updatecfgDict
                     updateCfgDictStr = "-d %s"%base64.b64encode(json.dumps(updateCfgDict).encode()).decode()
-            except :
-                pass
+            except Exception as r:
+                print(r)
         else:
             pass
         tdDnodes.deploy(1,updateCfgDict)