Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/row_optimize

docs/en/07-develop/09-udf.md

@@ -205,13 +205,13 @@ Additional functions are defined in `taosudf.h` to make it easier to work with t
 
 To use your user-defined function in TDengine, first compile it to a dynamically linked library (DLL).
 
-For example, the sample UDF `add_one.c` can be compiled into a DLL as follows:
+For example, the sample UDF `bit_and.c` can be compiled into a DLL as follows:
 
 ```bash
-gcc -g -O0 -fPIC -shared add_one.c -o add_one.so
+gcc -g -O0 -fPIC -shared bit_and.c -o libbitand.so
 ```
 
-The generated DLL file `add_one.so` can now be used to implement your function. Note: GCC 7.5 or later is required.
+The generated DLL file `libbitand.so` can now be used to implement your function. Note: GCC 7.5 or later is required.
 
 ## Manage and Use User-Defined Functions
 After compiling your function into a DLL, you add it to TDengine. For more information, see [User-Defined Functions](../12-taos-sql/26-udf.md).

docs/en/12-taos-sql/26-udf.md

@@ -62,7 +62,7 @@ SHOW FUNCTIONS;
 
 The function name specified when creating UDF can be used directly in SQL statements, just like builtin functions. For example:
 ```sql
-SELECT X(c1,c2) FROM table/stable;
+SELECT bit_and(c1,c2) FROM table;
 ```
 
-The above SQL statement invokes function X for column c1 and c2. You can use query keywords like WHERE with user-defined functions.
+The above SQL statement invokes function X for column c1 and c2 on table. You can use query keywords like WHERE with user-defined functions.

docs/examples/rust/nativeexample/Cargo.toml

@@ -10,4 +10,4 @@ chrono = "0.4"
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] }
 
-taos = { version = "0.*" }
+taos = { version = "0.4.8" }

docs/examples/rust/nativeexample/examples/stmt_example.rs

@@ -12,7 +12,10 @@ async fn main() -> anyhow::Result<()> {
     // bind table name and tags
     stmt.set_tbname_tags(
         "d1001",
-        &[Value::VarChar("California.SanFransico".into()), Value::Int(2)],
+        &[
+            Value::VarChar("California.SanFransico".into()),
+            Value::Int(2),
+        ],
     )?;
     // bind values.
     let values = vec![

docs/examples/rust/nativeexample/examples/subscribe_demo.rs

@@ -50,7 +50,7 @@ async fn main() -> anyhow::Result<()> {
         // create super table
         format!("CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(24))"),
         // create topic for subscription
-        format!("CREATE TOPIC tmq_meters with META AS DATABASE {db}")
+        format!("CREATE TOPIC tmq_meters AS SELECT * FROM `meters`")
     ])
     .await?;
 
@@ -64,13 +64,9 @@ async fn main() -> anyhow::Result<()> {
     let mut consumer = tmq.build()?;
     consumer.subscribe(["tmq_meters"]).await?;
 
-    {
-        let mut stream = consumer.stream();
-
-        while let Some((offset, message)) = stream.try_next().await? {
-            // get information from offset
-
-            // the topic
+    consumer
+        .stream()
+        .try_for_each(|(offset, message)| async {
             let topic = offset.topic();
             // the vgroup id, like partition id in kafka.
             let vgroup_id = offset.vgroup_id();
@@ -78,20 +74,14 @@ async fn main() -> anyhow::Result<()> {
 
             if let Some(data) = message.into_data() {
                 while let Some(block) = data.fetch_raw_block().await? {
-                    // one block for one table, get table name if needed
-                    let name = block.table_name();
                     let records: Vec<Record> = block.deserialize().try_collect()?;
-                    println!(
-                        "** table: {}, got {} records: {:#?}\n",
-                        name.unwrap(),
-                        records.len(),
-                        records
-                    );
+                    println!("** read {} records: {:#?}\n", records.len(), records);
                 }
             }
             consumer.commit(offset).await?;
-        }
-    }
+            Ok(())
+        })
+        .await?;
 
     consumer.unsubscribe().await;
 

docs/examples/rust/restexample/examples/insert_example.rs

@@ -5,7 +5,6 @@ async fn main() -> anyhow::Result<()> {
     let dsn = "ws://";
     let taos = TaosBuilder::from_dsn(dsn)?.build()?;
 
-
     taos.exec_many([
         "DROP DATABASE IF EXISTS power",
         "CREATE DATABASE power",

docs/zh/07-develop/09-udf.md

@@ -205,13 +205,13 @@ typedef struct SUdfInterBuf {
 
 用户定义函数的 C 语言源代码无法直接被 TDengine 系统使用，而是需要先编译为 动态链接库，之后才能载入 TDengine 系统。
 
-例如，按照上一章节描述的规则准备好了用户定义函数的源代码 add_one.c，以 Linux 为例可以执行如下指令编译得到动态链接库文件：
+例如，按照上一章节描述的规则准备好了用户定义函数的源代码 bit_and.c，以 Linux 为例可以执行如下指令编译得到动态链接库文件：
 
 ```bash
-gcc -g -O0 -fPIC -shared add_one.c -o add_one.so
+gcc -g -O0 -fPIC -shared bit_and.c -o libbitand.so
 ```
 
-这样就准备好了动态链接库 add_one.so 文件，可以供后文创建 UDF 时使用了。为了保证可靠的系统运行，编译器 GCC 推荐使用 7.5 及以上版本。
+这样就准备好了动态链接库 libbitand.so 文件，可以供后文创建 UDF 时使用了。为了保证可靠的系统运行，编译器 GCC 推荐使用 7.5 及以上版本。
 
 ## 管理和使用UDF
 编译好的UDF，还需要将其加入到系统才能被正常的SQL调用。关于如何管理和使用UDF，参见[UDF使用说明](../12-taos-sql/26-udf.md)

docs/zh/12-taos-sql/26-udf.md

@@ -63,7 +63,7 @@ SHOW FUNCTIONS;
 
 在 SQL 指令中，可以直接以在系统中创建 UDF 时赋予的函数名来调用用户定义函数。例如：
 ```sql
-SELECT X(c1,c2) FROM table/stable;
+SELECT bit_and(c1,c2) FROM table;
 ```
 
-表示对名为 c1, c2 的数据列调用名为 X 的用户定义函数。SQL 指令中用户定义函数可以配合 WHERE 等查询特性来使用。
+表示对表 table 上名为 c1, c2 的数据列调用名为 bit_and 的用户定义函数。SQL 指令中用户定义函数可以配合 WHERE 等查询特性来使用。

include/libs/stream/streamState.h

@@ -35,6 +35,7 @@ typedef struct {
   TTB*         pFuncStateDb;
   TTB*         pFillStateDb;  // todo refactor
   TTB*         pSessionStateDb;
+  TTB*         pParNameDb;
   TXN          txn;
   int32_t      number;
 } SStreamState;
@@ -99,6 +100,9 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
 
+int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char* tbname);
+int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal);
+
 #if 0
 char* streamStateSessionDump(SStreamState* pState);
 #endif

include/util/taoserror.h

@@ -352,6 +352,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TDB_STB_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x061A)
 #define TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER  TAOS_DEF_ERROR_CODE(0, 0x061B)
 #define TSDB_CODE_TDB_TDB_ENV_OPEN_ERROR        TAOS_DEF_ERROR_CODE(0, 0x061C)
+#define TSDB_CODE_TDB_TABLE_IN_OTHER_STABLE     TAOS_DEF_ERROR_CODE(0, 0x061D)
 
 // query
 #define TSDB_CODE_QRY_INVALID_QHANDLE           TAOS_DEF_ERROR_CODE(0, 0x0700)

packaging/tools/tdengine.iss

@@ -63,7 +63,7 @@ Source: {#MyAppSourceDir}{#MyAppTaosdemoExeName}; DestDir: "{app}"; Flags: igNor
 
 [run]
 Filename: {sys}\sc.exe; Parameters: "create taosd start= DEMAND binPath= ""C:\\TDengine\\taosd.exe --win_service""" ; Flags: runhidden
-Filename: {sys}\sc.exe; Parameters: "create taosadapter start= DEMAND binPath= ""C:\\TDengine\\taosadapter.exe --win_service""" ; Flags: runhidden
+Filename: {sys}\sc.exe; Parameters: "create taosadapter start= DEMAND binPath= ""C:\\TDengine\\taosadapter.exe""" ; Flags: runhidden
 
 [UninstallRun]
 RunOnceId: "stoptaosd"; Filename: {sys}\sc.exe; Parameters: "stop taosd" ; Flags: runhidden

source/dnode/mgmt/mgmt_mnode/src/mmWorker.c

@@ -56,6 +56,9 @@ static void mmProcessRpcMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   if (IsReq(pMsg) && pMsg->info.handle != NULL && code != TSDB_CODE_ACTION_IN_PROGRESS) {
     if (code != 0 && terrno != 0) code = terrno;
     mmSendRsp(pMsg, code);
+  } else {
+    rpcFreeCont(pMsg->info.rsp);
+    pMsg->info.rsp = NULL;
   }
 
   if (code == TSDB_CODE_RPC_REDIRECT) {

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

@@ -577,9 +577,9 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {
 
     int32_t contLen = tSerializeSEpSet(NULL, 0, &epSet);
     pMsg->info.rsp = rpcMallocCont(contLen);
-    pMsg->info.hasEpSet = 1;
     if (pMsg->info.rsp != NULL) {
       tSerializeSEpSet(pMsg->info.rsp, contLen, &epSet);
+      pMsg->info.hasEpSet = 1;
       pMsg->info.rspLen = contLen;
       terrno = TSDB_CODE_RPC_REDIRECT;
     } else {

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

@@ -403,6 +403,11 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq, STableMe
   // validate req
   metaReaderInit(&mr, pMeta, 0);
   if (metaGetTableEntryByName(&mr, pReq->name) == 0) {
+    if (pReq->type == TSDB_CHILD_TABLE && pReq->ctb.suid != mr.me.ctbEntry.suid) {
+      terrno = TSDB_CODE_TDB_TABLE_IN_OTHER_STABLE;
+      metaReaderClear(&mr);
+      return -1;
+    }
     pReq->uid = mr.me.uid;
     if (pReq->type == TSDB_CHILD_TABLE) {
       pReq->ctb.suid = mr.me.ctbEntry.suid;

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

@@ -398,6 +398,35 @@ bool tqNextDataBlock(STqReader* pReader) {
   return false;
 }
 
+int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrapper* pSrc, char* mask) {
+  int32_t code;
+
+  int32_t cnt = 0;
+  for (int32_t i = 0; i < pSrc->nCols; i++) {
+    cnt += mask[i];
+  }
+
+  pDst->nCols = cnt;
+  pDst->pSchema = taosMemoryCalloc(cnt, sizeof(SSchema));
+  if (pDst->pSchema == NULL) {
+    return -1;
+  }
+
+  int32_t j = 0;
+  for (int32_t i = 0; i < pSrc->nCols; i++) {
+    if (mask[i]) {
+      pDst->pSchema[j++] = pSrc->pSchema[i];
+      SColumnInfoData colInfo =
+          createColumnInfoData(pSrc->pSchema[i].type, pSrc->pSchema[i].bytes, pSrc->pSchema[i].colId);
+      code = blockDataAppendColInfo(pBlock, &colInfo);
+      if (code != 0) {
+        return -1;
+      }
+    }
+  }
+  return 0;
+}
+
 bool tqNextDataBlockFilterOut(STqReader* pHandle, SHashObj* filterOutUids) {
   while (1) {
     if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
@@ -527,6 +556,119 @@ FAIL:
   return -1;
 }
 
+int32_t tqSplitRetrieveDataBlock(STqReader* pReader, SArray* blocks, SArray* schemas) {
+  int32_t sversion = htonl(pReader->pBlock->sversion);
+
+  if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion ||
+      pReader->cachedSchemaSuid != pReader->msgIter.suid) {
+    if (pReader->pSchema) taosMemoryFree(pReader->pSchema);
+    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
+    if (pReader->pSchema == NULL) {
+      tqWarn("cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64 "), version %d, possibly dropped table",
+             pReader->msgIter.uid, pReader->msgIter.suid, pReader->cachedSchemaVer);
+      /*ASSERT(0);*/
+      pReader->cachedSchemaSuid = 0;
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
+      return -1;
+    }
+
+    if (pReader->pSchemaWrapper) tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
+    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
+    if (pReader->pSchemaWrapper == NULL) {
+      tqWarn("cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
+             pReader->msgIter.uid, pReader->cachedSchemaVer);
+      /*ASSERT(0);*/
+      pReader->cachedSchemaSuid = 0;
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
+      return -1;
+    }
+    pReader->cachedSchemaVer = sversion;
+    pReader->cachedSchemaSuid = pReader->msgIter.suid;
+  }
+
+  STSchema*       pTschema = pReader->pSchema;
+  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
+
+  int32_t colAtMost = pSchemaWrapper->nCols;
+
+  int32_t curRow = 0;
+
+  char* assigned = taosMemoryCalloc(1, pSchemaWrapper->nCols);
+  if (assigned) return -1;
+
+  tInitSubmitBlkIter(&pReader->msgIter, pReader->pBlock, &pReader->blkIter);
+  STSRowIter iter = {0};
+  tdSTSRowIterInit(&iter, pTschema);
+  STSRow* row;
+
+  while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
+    bool buildNew = false;
+    tdSTSRowIterReset(&iter, row);
+
+    for (int32_t i = 0; i < colAtMost; i++) {
+      SCellVal sVal = {0};
+      if (!tdSTSRowIterFetch(&iter, pSchemaWrapper->pSchema[i].colId, pSchemaWrapper->pSchema[i].type, &sVal)) {
+        break;
+      }
+      if (curRow == 0) {
+        assigned[i] = sVal.valType != TD_VTYPE_NONE;
+        buildNew = true;
+      } else {
+        bool currentRowAssigned = sVal.valType != TD_VTYPE_NONE;
+        if (currentRowAssigned != assigned[i]) {
+          assigned[i] = currentRowAssigned;
+          buildNew = true;
+        }
+      }
+    }
+
+    if (buildNew) {
+      SSDataBlock    block;
+      SSchemaWrapper sw;
+      if (tqMaskBlock(&sw, &block, pSchemaWrapper, assigned) < 0) {
+        goto FAIL;
+      }
+
+      taosArrayPush(blocks, &block);
+      taosArrayPush(schemas, &sw);
+    }
+
+    SSDataBlock* pBlock = taosArrayGetLast(blocks);
+    pBlock->info.uid = pReader->msgIter.uid;
+    pBlock->info.rows = pReader->msgIter.numOfRows;
+    pBlock->info.version = pReader->pMsg->version;
+
+    if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows - curRow) < 0) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      goto FAIL;
+    }
+
+    tdSTSRowIterInit(&iter, pTschema);
+    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); i++) {
+      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
+      SCellVal         sVal = {0};
+
+      if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
+        break;
+      }
+
+      ASSERT(sVal.valType != TD_VTYPE_NONE);
+
+      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
+        goto FAIL;
+      }
+    }
+    curRow++;
+  }
+
+  taosMemoryFree(assigned);
+  return 0;
+
+FAIL:
+  taosMemoryFree(assigned);
+  return -1;
+}
+
 void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) { pReader->pColIdList = pColIdList; }
 
 int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) {

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

@@ -49,7 +49,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
 
   pTsdb->path = (char *)&pTsdb[1];
   snprintf(pTsdb->path, TD_PATH_MAX, "%s%s%s", pVnode->path, TD_DIRSEP, dir);
-  taosRealPath(pTsdb->path, NULL, slen);
+  // taosRealPath(pTsdb->path, NULL, slen);
   pTsdb->pVnode = pVnode;
   taosThreadRwlockInit(&pTsdb->rwLock, NULL);
   if (!pKeepCfg) {

source/libs/executor/inc/executorimpl.h

@@ -163,7 +163,7 @@ typedef struct {
   SArray*  pStopInfo;
 } STaskStopInfo;
 
-typedef struct SExecTaskInfo {
+struct SExecTaskInfo {
   STaskIdInfo   id;
   uint32_t      status;
   STimeWindow   window;
@@ -182,7 +182,7 @@ typedef struct SExecTaskInfo {
   struct SOperatorInfo* pRoot;
   SLocalFetch           localFetch;
   STaskStopInfo         stopInfo;
-} SExecTaskInfo;
+};
 
 enum {
   OP_NOT_OPENED = 0x0,
@@ -315,37 +315,39 @@ typedef struct STableMetaCacheInfo {
   uint64_t   cacheHit;
 } STableMetaCacheInfo;
 
-typedef struct STableScanInfo {
+typedef struct STableScanBase {
   STsdbReader*           dataReader;
-  SReadHandle            readHandle;
-  SLimitInfo             limitInfo;
   SFileBlockLoadRecorder readRecorder;
-  SScanInfo              scanInfo;
-  int32_t                scanTimes;
-  SSDataBlock*           pResBlock;
+  SQueryTableDataCond    cond;
+  SAggOptrPushDownInfo   pdInfo;
   SColMatchInfo          matchInfo;
+  SReadHandle            readHandle;
   SExprSupp              pseudoSup;
-  SQueryTableDataCond    cond;
+  STableMetaCacheInfo    metaCache;
   int32_t                scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
   int32_t                dataBlockLoadFlag;
+  SLimitInfo             limitInfo;
+} STableScanBase;
+
+typedef struct STableScanInfo {
+  STableScanBase         base;
+  SScanInfo              scanInfo;
+  int32_t                scanTimes;
+  SSDataBlock*           pResBlock;
   SSampleExecInfo        sample;  // sample execution info
   int32_t                currentGroupId;
   int32_t                currentTable;
   int8_t                 scanMode;
-  SAggOptrPushDownInfo   pdInfo;
   int8_t                 assignBlockUid;
-  STableMetaCacheInfo    metaCache;
 } STableScanInfo;
 
 typedef struct STableMergeScanInfo {
-  STableListInfo*        tableListInfo;
   int32_t                tableStartIndex;
   int32_t                tableEndIndex;
   bool                   hasGroupId;
   uint64_t               groupId;
   SArray*                queryConds;  // array of queryTableDataCond
-  STsdbReader*           pReader;
-  SReadHandle            readHandle;
+  STableScanBase         base;
   int32_t                bufPageSize;
   uint32_t               sortBufSize;  // max buffer size for in-memory sort
   SArray*                pSortInfo;
@@ -354,25 +356,10 @@ typedef struct STableMergeScanInfo {
   int64_t                startTs;  // sort start time
   SArray*                sortSourceParams;
   SLimitInfo             limitInfo;
-  SFileBlockLoadRecorder readRecorder;
   int64_t                numOfRows;
   SScanInfo              scanInfo;
   int32_t                scanTimes;
-  SqlFunctionCtx*        pCtx;  // which belongs to the direct upstream operator operator query context
-  SResultRowInfo*        pResultRowInfo;
-  int32_t*               rowEntryInfoOffset;
-  SExprInfo*             pExpr;
   SSDataBlock*           pResBlock;
-  SColMatchInfo          matchInfo;
-  int32_t                numOfOutput;
-  SExprSupp              pseudoSup;
-  SQueryTableDataCond    cond;
-  int32_t                scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
-  int32_t                dataBlockLoadFlag;
-
-  // if the upstream is an interval operator, the interval info is also kept here to get the time
-  // window to check if current data block needs to be loaded.
-  SInterval       interval;
   SSampleExecInfo        sample;  // sample execution info
   SSortExecInfo          sortExecInfo;
 } STableMergeScanInfo;
@@ -414,13 +401,6 @@ enum {
   PROJECT_RETRIEVE_DONE = 0x2,
 };
 
-typedef struct SCatchSupporter {
-  SHashObj*      pWindowHashTable;  // quick locate the window object for each window
-  SDiskbasedBuf* pDataBuf;          // buffer based on blocked-wised disk file
-  int32_t        keySize;
-  int64_t*       pKeyBuf;
-} SCatchSupporter;
-
 typedef struct SStreamAggSupporter {
   int32_t         resultRowSize;  // the result buffer size for each result row, with the meta data size for each row
   SSDataBlock*    pScanBlock;
@@ -504,7 +484,6 @@ typedef struct SStreamScanInfo {
   STimeWindow           updateWin;
   STimeWindowAggSupp    twAggSup;
   SSDataBlock*          pUpdateDataRes;
-  SHashObj*             pGroupIdTbNameMap;
   // status for tmq
   SNodeList* pGroupTags;
   SNode*     pTagCond;
@@ -612,7 +591,6 @@ typedef struct SStreamIntervalOperatorInfo {
   SArray*            pChildren;
   SStreamState*      pState;
   SWinKey            delKey;
-  SHashObj*          pGroupIdTbNameMap;  // uint64_t -> char[TSDB_TABLE_NAME_LEN]
 } SStreamIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -745,7 +723,6 @@ typedef struct SStreamSessionAggOperatorInfo {
   SPhysiNode*         pPhyNode;   // create new child
   bool                isFinal;
   bool                ignoreExpiredData;
-  SHashObj*           pGroupIdTbNameMap;
 } SStreamSessionAggOperatorInfo;
 
 typedef struct SStreamStateAggOperatorInfo {
@@ -761,7 +738,6 @@ typedef struct SStreamStateAggOperatorInfo {
   void*               pDelIterator;
   SArray*             pChildren;  // cache for children's result;
   bool                ignoreExpiredData;
-  SHashObj*           pGroupIdTbNameMap;
 } SStreamStateAggOperatorInfo;
 
 typedef struct SStreamPartitionOperatorInfo {
@@ -1046,8 +1022,8 @@ int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPos
 
 SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
                                            SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
-                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
+                                                SExecTaskInfo* pTaskInfo);
 
 void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
 

source/libs/executor/src/executor.c

@@ -1026,8 +1026,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
     SStreamScanInfo* pInfo = pOperator->info;
     if (pOffset->type == TMQ_OFFSET__LOG) {
       STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
-      tsdbReaderClose(pTSInfo->dataReader);
-      pTSInfo->dataReader = NULL;
+      tsdbReaderClose(pTSInfo->base.dataReader);
+      pTSInfo->base.dataReader = NULL;
 #if 0
       if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus) &&
           pInfo->tqReader->pWalReader->curVersion != pOffset->version) {
@@ -1079,23 +1079,23 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
       // TODO after dropping table, table may not found
       ASSERT(found);
 
-      if (pTableScanInfo->dataReader == NULL) {
+      if (pTableScanInfo->base.dataReader == NULL) {
         STableKeyInfo* pList = tableListGetInfo(pTaskInfo->pTableInfoList, 0);
         int32_t        num = tableListGetSize(pTaskInfo->pTableInfoList);
 
-        if (tsdbReaderOpen(pTableScanInfo->readHandle.vnode, &pTableScanInfo->cond, pList, num,
-                           &pTableScanInfo->dataReader, NULL) < 0 ||
-            pTableScanInfo->dataReader == NULL) {
+        if (tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &pTableScanInfo->base.cond, pList, num,
+                           &pTableScanInfo->base.dataReader, NULL) < 0 ||
+            pTableScanInfo->base.dataReader == NULL) {
           ASSERT(0);
         }
       }
 
       STableKeyInfo tki = {.uid = uid};
-      tsdbSetTableList(pTableScanInfo->dataReader, &tki, 1);
-      int64_t oldSkey = pTableScanInfo->cond.twindows.skey;
-      pTableScanInfo->cond.twindows.skey = ts + 1;
-      tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
-      pTableScanInfo->cond.twindows.skey = oldSkey;
+      tsdbSetTableList(pTableScanInfo->base.dataReader, &tki, 1);
+      int64_t oldSkey = pTableScanInfo->base.cond.twindows.skey;
+      pTableScanInfo->base.cond.twindows.skey = ts + 1;
+      tsdbReaderReset(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
+      pTableScanInfo->base.cond.twindows.skey = oldSkey;
       pTableScanInfo->scanTimes = 0;
 
       qDebug("tsdb reader offset seek to uid %" PRId64 " ts %" PRId64 ", table cur set to %d , all table num %d", uid,

source/libs/executor/src/executorimpl.c

@@ -1000,12 +1000,6 @@ int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableSc
   return TSDB_CODE_SUCCESS;
 }
 
-static void updateTableQueryInfoForReverseScan(STableQueryInfo* pTableQueryInfo) {
-  if (pTableQueryInfo == NULL) {
-    return;
-  }
-}
-
 void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status) {
   if (status == TASK_NOT_COMPLETED) {
     pTaskInfo->status = status;
@@ -1336,27 +1330,14 @@ void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGr
   pBlock->info.groupId = 0;
   ASSERT(!pbInfo->mergeResultBlock);
   doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
-  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
-    SStreamStateAggOperatorInfo* pInfo = pOperator->info;
 
-    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
-    if (tbname != NULL) {
-      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
-    } else {
+  void* tbname = NULL;
+  if (streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.groupId, &tbname) < 0) {
     pBlock->info.parTbName[0] = 0;
-    }
-  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
-             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
-             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
-    SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
-
-    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
-    if (tbname != NULL) {
-      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
   } else {
-      pBlock->info.parTbName[0] = 0;
-    }
+    memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
   }
+  tdbFree(tbname);
 }
 
 void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
@@ -1665,55 +1646,6 @@ int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t
 static int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t keyBufSize,
                                 const char* pKey);
 
-static bool needToMerge(SSDataBlock* pBlock, SArray* groupInfo, char** buf, int32_t rowIndex) {
-  size_t size = taosArrayGetSize(groupInfo);
-  if (size == 0) {
-    return true;
-  }
-
-  for (int32_t i = 0; i < size; ++i) {
-    int32_t* index = taosArrayGet(groupInfo, i);
-
-    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, *index);
-    bool             isNull = colDataIsNull(pColInfo, rowIndex, pBlock->info.rows, NULL);
-
-    if ((isNull && buf[i] != NULL) || (!isNull && buf[i] == NULL)) {
-      return false;
-    }
-
-    char* pCell = colDataGetData(pColInfo, rowIndex);
-    if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
-      if (varDataLen(pCell) != varDataLen(buf[i])) {
-        return false;
-      } else {
-        if (memcmp(varDataVal(pCell), varDataVal(buf[i]), varDataLen(pCell)) != 0) {
-          return false;
-        }
-      }
-    } else {
-      if (memcmp(pCell, buf[i], pColInfo->info.bytes) != 0) {
-        return false;
-      }
-    }
-  }
-
-  return 0;
-}
-
-static bool saveCurrentTuple(char** rowColData, SArray* pColumnList, SSDataBlock* pBlock, int32_t rowIndex) {
-  int32_t size = (int32_t)taosArrayGetSize(pColumnList);
-
-  for (int32_t i = 0; i < size; ++i) {
-    int32_t*         index = taosArrayGet(pColumnList, i);
-    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, *index);
-
-    char* data = colDataGetData(pColInfo, rowIndex);
-    memcpy(rowColData[i], data, colDataGetLength(pColInfo, rowIndex));
-  }
-
-  return true;
-}
-
 int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag) {
   // todo add more information about exchange operation
   int32_t type = pOperator->operatorType;
@@ -1725,13 +1657,13 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
     return TSDB_CODE_SUCCESS;
   } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
     STableScanInfo* pTableScanInfo = pOperator->info;
-    *order = pTableScanInfo->cond.order;
-    *scanFlag = pTableScanInfo->scanFlag;
+    *order = pTableScanInfo->base.cond.order;
+    *scanFlag = pTableScanInfo->base.scanFlag;
     return TSDB_CODE_SUCCESS;
   } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN) {
     STableMergeScanInfo* pTableScanInfo = pOperator->info;
-    *order = pTableScanInfo->cond.order;
-    *scanFlag = pTableScanInfo->scanFlag;
+    *order = pTableScanInfo->base.cond.order;
+    *scanFlag = pTableScanInfo->base.scanFlag;
     return TSDB_CODE_SUCCESS;
   } else {
     if (pOperator->pDownstream == NULL || pOperator->pDownstream[0] == NULL) {
@@ -2145,7 +2077,7 @@ static SSDataBlock* doFill(SOperatorInfo* pOperator) {
       break;
     }
 
-    doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo );
+    doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo);
     if (fillResult->info.rows > 0) {
       break;
     }
@@ -2372,14 +2304,14 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiN
   pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
   pInfo->groupId = UINT64_MAX;
 
-  setOperatorInfo(pOperator, "TableAggregate", QUERY_NODE_PHYSICAL_PLAN_HASH_AGG, true, OP_NOT_OPENED, pInfo, pTaskInfo);
-  pOperator->fpSet =
-      createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, destroyAggOperatorInfo, NULL);
+  setOperatorInfo(pOperator, "TableAggregate", QUERY_NODE_PHYSICAL_PLAN_HASH_AGG, true, OP_NOT_OPENED, pInfo,
+                  pTaskInfo);
+  pOperator->fpSet = createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, destroyAggOperatorInfo, NULL);
 
   if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
     STableScanInfo* pTableScanInfo = downstream->info;
-    pTableScanInfo->pdInfo.pExprSup = &pOperator->exprSupp;
-    pTableScanInfo->pdInfo.pAggSup = &pInfo->aggSup;
+    pTableScanInfo->base.pdInfo.pExprSup = &pOperator->exprSupp;
+    pTableScanInfo->base.pdInfo.pAggSup = &pInfo->aggSup;
   }
 
   code = appendDownstream(pOperator, &downstream, 1);
@@ -2744,7 +2676,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       }
 
       STableScanInfo* pScanInfo = pOperator->info;
-      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
+      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
     } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
       STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
 
@@ -2762,14 +2694,14 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
         return NULL;
       }
 
-      pOperator = createTableMergeScanOperatorInfo(pTableScanNode, pTableListInfo, pHandle, pTaskInfo);
+      pOperator = createTableMergeScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
       if (NULL == pOperator) {
         pTaskInfo->code = terrno;
         return NULL;
       }
 
       STableScanInfo* pScanInfo = pOperator->info;
-      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
+      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
     } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
       pOperator = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
                                              pTaskInfo);
@@ -3351,13 +3283,13 @@ int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pStat
 
     if (pBlock->info.groupId == 0) {
       pBlock->info.groupId = pPos->groupId;
-      SStreamIntervalOperatorInfo* pInfo = pOperator->info;
-      char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
-      if (tbname != NULL) {
-        memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
-      } else {
+      void* tbname = NULL;
+      if (streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.groupId, &tbname) < 0) {
         pBlock->info.parTbName[0] = 0;
+      } else {
+        memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
       }
+      tdbFree(tbname);
     } else {
       // current value belongs to different group, it can't be packed into one datablock
       if (pBlock->info.groupId != pPos->groupId) {
@@ -3443,30 +3375,13 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pSta
     if (pBlock->info.groupId == 0) {
       pBlock->info.groupId = pKey->groupId;
 
-      if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
-        SStreamStateAggOperatorInfo* pInfo = pOperator->info;
-
-        char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
-        if (tbname != NULL) {
-          memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
-        } else {
+      void* tbname = NULL;
+      if (streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.groupId, &tbname) < 0) {
         pBlock->info.parTbName[0] = 0;
-        }
-      } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
-                 pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
-                 pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
-        SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
-
-        char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
-        if (tbname != NULL) {
-          memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
-        } else {
-          pBlock->info.parTbName[0] = 0;
-        }
       } else {
-        ASSERT(0);
+        memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
       }
-
+      tdbFree(tbname);
     } else {
       // current value belongs to different group, it can't be packed into one datablock
       if (pBlock->info.groupId != pKey->groupId) {

source/libs/executor/src/timewindowoperator.c

@@ -1450,7 +1450,6 @@ static int32_t closeStreamIntervalWindow(SSHashObj* pHashMap, STimeWindowAggSupp
         }
       }
       tSimpleHashIterateRemove(pHashMap, pWinKey, sizeof(SWinKey), &pIte, &iter);
-      /*taosHashRemove(pInfo->pGroupIdTbNameMap, &pWinKey->groupId, sizeof(int64_t));*/
     }
   }
   return TSDB_CODE_SUCCESS;
@@ -1547,7 +1546,8 @@ static void doBuildDeleteResult(SStreamIntervalOperatorInfo* pInfo, SArray* pWin
   uint64_t uid = 0;
   for (int32_t i = *index; i < size; i++) {
     SWinKey* pWin = taosArrayGet(pWins, i);
-    char*    tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pWin->groupId, sizeof(int64_t));
+    void*    tbname = NULL;
+    streamStateGetParName(pInfo->pState, pWin->groupId, &tbname);
     if (tbname == NULL) {
       appendOneRowToStreamSpecialBlock(pBlock, &pWin->ts, &pWin->ts, &uid, &pWin->groupId, NULL);
     } else {
@@ -1555,6 +1555,7 @@ static void doBuildDeleteResult(SStreamIntervalOperatorInfo* pInfo, SArray* pWin
       STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
       appendOneRowToStreamSpecialBlock(pBlock, &pWin->ts, &pWin->ts, &uid, &pWin->groupId, parTbName);
     }
+    tdbFree(tbname);
     (*index)++;
   }
 }
@@ -1617,7 +1618,6 @@ void destroyStreamFinalIntervalOperatorInfo(void* param) {
   nodesDestroyNode((SNode*)pInfo->pPhyNode);
   colDataDestroy(&pInfo->twAggSup.timeWindowData);
   cleanupGroupResInfo(&pInfo->groupResInfo);
-  taosHashCleanup(pInfo->pGroupIdTbNameMap);
 
   taosMemoryFreeClear(param);
 }
@@ -2545,9 +2545,11 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode
   pInfo->interval.interval = pInterpPhyNode->interval;
   pInfo->current = pInfo->win.skey;
 
+  if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
     STableScanInfo* pScanInfo = (STableScanInfo*)downstream->info;
-  pScanInfo->cond.twindows = pInfo->win;
-  pScanInfo->cond.type = TIMEWINDOW_RANGE_EXTERNAL;
+    pScanInfo->base.cond.twindows = pInfo->win;
+    pScanInfo->base.cond.type = TIMEWINDOW_RANGE_EXTERNAL;
+  }
 
   setOperatorInfo(pOperator, "TimeSliceOperator", QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC, false, OP_NOT_OPENED, pInfo,
                   pTaskInfo);
@@ -3153,11 +3155,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval final recv" : "interval semi recv");
 
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-    }
-
     ASSERT(pBlock->info.type != STREAM_INVERT);
     if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PULL_DATA) {
       pInfo->binfo.pRes->info.type = pBlock->info.type;
@@ -3372,9 +3369,6 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
   pInfo->delKey.ts = INT64_MAX;
   pInfo->delKey.groupId = 0;
 
-  pInfo->pGroupIdTbNameMap =
-      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
-
   pOperator->operatorType = pPhyNode->type;
   pOperator->blocking = true;
   pOperator->status = OP_NOT_OPENED;
@@ -3425,7 +3419,6 @@ void destroyStreamSessionAggOperatorInfo(void* param) {
   blockDataDestroy(pInfo->pWinBlock);
   blockDataDestroy(pInfo->pUpdateRes);
   tSimpleHashCleanup(pInfo->pStDeleted);
-  taosHashCleanup(pInfo->pGroupIdTbNameMap);
 
   taosMemoryFreeClear(param);
 }
@@ -3857,30 +3850,18 @@ void doBuildDeleteDataBlock(SOperatorInfo* pOp, SSHashObj* pStDeleted, SSDataBlo
     SColumnInfoData* pCalEdCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
     colDataAppendNULL(pCalEdCol, pBlock->info.rows);
 
-    SHashObj* pGroupIdTbNameMap = NULL;
-    if (pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
-        pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
-        pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
-      SStreamSessionAggOperatorInfo* pInfo = pOp->info;
-      pGroupIdTbNameMap = pInfo->pGroupIdTbNameMap;
-    } else if (pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
-      SStreamStateAggOperatorInfo* pInfo = pOp->info;
-      pGroupIdTbNameMap = pInfo->pGroupIdTbNameMap;
-    } else {
-      ASSERT(0);
-    }
-
-    char*            tbname = taosHashGet(pGroupIdTbNameMap, &res->groupId, sizeof(int64_t));
     SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
+
+    void* tbname = NULL;
+    streamStateGetParName(pOp->pTaskInfo->streamInfo.pState, res->groupId, &tbname);
     if (tbname == NULL) {
-      /*printf("\n\n no tbname for group id %" PRId64 "%p %p\n\n", res->groupId, pOp->info, pGroupIdTbNameMap);*/
       colDataAppendNULL(pTableCol, pBlock->info.rows);
     } else {
       char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
       STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
       colDataAppend(pTableCol, pBlock->info.rows, (const char*)parTbName, false);
-      /*printf("\n\n get tbname %s group id %" PRId64 "\n\n", tbname, res->groupId);*/
     }
+    tdbFree(tbname);
     pBlock->info.rows += 1;
   }
   if ((*Ite) == NULL) {
@@ -4053,19 +4034,6 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "final session recv" : "single session recv");
 
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-      /*printf("\n\n put tbname %s group id %" PRId64 "\n\n into %p %p", pBlock->info.parTbName, pBlock->info.groupId,*/
-      /*pInfo, pInfo->pGroupIdTbNameMap);*/
-    }
-
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
-    }
-
     if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
         pBlock->info.type == STREAM_CLEAR) {
       SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));
@@ -4209,8 +4177,6 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
   pInfo->isFinal = false;
   pInfo->pPhyNode = pPhyNode;
   pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
-  pInfo->pGroupIdTbNameMap =
-      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
 
   setOperatorInfo(pOperator, "StreamSessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION, true,
                   OP_NOT_OPENED, pInfo, pTaskInfo);
@@ -4285,12 +4251,6 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, "semi session recv");
 
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
-    }
-
     if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
         pBlock->info.type == STREAM_CLEAR) {
       // gap must be 0
@@ -4418,7 +4378,6 @@ void destroyStreamStateOperatorInfo(void* param) {
   colDataDestroy(&pInfo->twAggSup.timeWindowData);
   blockDataDestroy(pInfo->pDelRes);
   tSimpleHashCleanup(pInfo->pSeDeleted);
-  taosHashCleanup(pInfo->pGroupIdTbNameMap);
   taosMemoryFreeClear(param);
 }
 
@@ -4612,12 +4571,6 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, "single state recv");
 
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
-    }
-
     if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
         pBlock->info.type == STREAM_CLEAR) {
       SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));
@@ -4731,9 +4684,6 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
   pInfo->pChildren = NULL;
   pInfo->ignoreExpiredData = pStateNode->window.igExpired;
 
-  pInfo->pGroupIdTbNameMap =
-      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
-
   setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED,
                   pInfo, pTaskInfo);
   pOperator->fpSet =
@@ -5384,12 +5334,6 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, "single interval recv");
 
-    if (pBlock->info.parTbName[0]) {
-      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
-                  TSDB_TABLE_NAME_LEN);
-      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
-    }
-
     if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
         pBlock->info.type == STREAM_CLEAR) {
       doDeleteWindows(pOperator, &pInfo->interval, pBlock, pInfo->pDelWins, pUpdatedMap);
@@ -5547,9 +5491,6 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
   pInfo->delKey.ts = INT64_MAX;
   pInfo->delKey.groupId = 0;
 
-  pInfo->pGroupIdTbNameMap =
-      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
-
   setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED,
                   pInfo, pTaskInfo);
   pOperator->fpSet =

source/libs/scalar/src/filter.c

@@ -1704,7 +1704,8 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
       }
 
       qDebug("GROUP Num:%u", info->groupNum);
-      for (uint32_t i = 0; i < info->groupNum; ++i) {
+      uint32_t maxDbgGrpNum = TMIN(info->groupNum, 1000);
+      for (uint32_t i = 0; i < maxDbgGrpNum; ++i) {
         SFilterGroup *group = &info->groups[i];
         qDebug("Group%d : unit num[%u]", i, group->unitNum);
 
@@ -3920,6 +3921,10 @@ EDealRes fltReviseRewriter(SNode **pNode, void *pContext) {
       } else {
         SColumnNode   *refNode = (SColumnNode *)node->pLeft;
         SNodeListNode *listNode = (SNodeListNode *)node->pRight;
+        if (LIST_LENGTH(listNode->pNodeList) > 10) {
+          stat->scalarMode = true;
+          return DEAL_RES_CONTINUE;
+        }
         int32_t        type = vectorGetConvertType(refNode->node.resType.type, listNode->dataType.type);
         if (0 != type && type != refNode->node.resType.type) {
           stat->scalarMode = true;

source/libs/stream/src/streamState.c

@@ -16,6 +16,7 @@
 #include "executor.h"
 #include "streamInc.h"
 #include "tcommon.h"
+#include "tcompare.h"
 #include "ttimer.h"
 
 // todo refactor
@@ -144,6 +145,11 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int
     goto _err;
   }
 
+  if (tdbTbOpen("parname.state.db", sizeof(int64_t), TSDB_TABLE_NAME_LEN, NULL, pState->db, &pState->pParNameDb, 0) <
+      0) {
+    goto _err;
+  }
+
   if (streamStateBegin(pState) < 0) {
     goto _err;
   }
@@ -157,6 +163,7 @@ _err:
   tdbTbClose(pState->pFuncStateDb);
   tdbTbClose(pState->pFillStateDb);
   tdbTbClose(pState->pSessionStateDb);
+  tdbTbClose(pState->pParNameDb);
   tdbClose(pState->db);
   taosMemoryFree(pState);
   return NULL;
@@ -169,6 +176,7 @@ void streamStateClose(SStreamState* pState) {
   tdbTbClose(pState->pFuncStateDb);
   tdbTbClose(pState->pFillStateDb);
   tdbTbClose(pState->pSessionStateDb);
+  tdbTbClose(pState->pParNameDb);
   tdbClose(pState->db);
 
   taosMemoryFree(pState);
@@ -812,6 +820,16 @@ _end:
   return res;
 }
 
+int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char tbname[TSDB_TABLE_NAME_LEN]) {
+  tdbTbUpsert(pState->pParNameDb, &groupId, sizeof(int64_t), tbname, TSDB_TABLE_NAME_LEN, &pState->txn);
+  return 0;
+}
+
+int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal) {
+  int32_t len;
+  return tdbTbGet(pState->pParNameDb, &groupId, sizeof(int64_t), pVal, &len);
+}
+
 #if 0
 char* streamStateSessionDump(SStreamState* pState) {
   SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));

source/libs/sync/src/syncMain.c

@@ -436,8 +436,15 @@ int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) {
   }
 
   int32_t ret = 0;
-  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
+  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER && pSyncNode->replicaNum > 1) {
     SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
+    if (pSyncNode->peersNum == 2) {
+      SyncIndex matchIndex0 = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[0]));
+      SyncIndex matchIndex1 = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[1]));
+      if (matchIndex1 > matchIndex0) {
+        newLeader = (pSyncNode->peersNodeInfo)[1];
+      }
+    }
     ret = syncNodeLeaderTransferTo(pSyncNode, newLeader);
   }
 
@@ -1396,7 +1403,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
       // reset sender
       bool reset = false;
       for (int32_t j = 0; j < TSDB_MAX_REPLICA; ++j) {
-        if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j])) {
+        if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j]) && oldSenders[j] != NULL) {
           char     host[128];
           uint16_t port;
           syncUtilU642Addr((pSyncNode->replicasId)[i].addr, host, sizeof(host), &port);
@@ -1413,6 +1420,8 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
 
           sNTrace(pSyncNode, "snapshot sender udpate replicaIndex from %d to %d, %s:%d, %p, reset:%d", oldreplicaIndex,
                   i, host, port, (pSyncNode->senders)[i], reset);
+
+          break;
         }
       }
     }

source/libs/transport/src/tmsgcb.c

@@ -44,7 +44,13 @@ int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) {
   return code;
 }
 
-void tmsgSendRsp(SRpcMsg* pMsg) { return (*defaultMsgCb.sendRspFp)(pMsg); }
+void tmsgSendRsp(SRpcMsg* pMsg) {
+#if 1
+  rpcSendResponse(pMsg);
+#else
+  return (*defaultMsgCb.sendRspFp)(pMsg);
+#endif
+}
 
 void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { (*defaultMsgCb.sendRedirectRspFp)(pMsg, pNewEpSet); }
 

source/libs/transport/src/transSvr.c

@@ -462,8 +462,6 @@ static void uvStartSendResp(SSvrMsg* smsg) {
   if (pConn->broken == true) {
     // persist by
     destroySmsg(smsg);
-    // transFreeMsg(smsg->msg.pCont);
-    // taosMemoryFree(smsg);
     transUnrefSrvHandle(pConn);
     return;
   }
@@ -1234,7 +1232,9 @@ int transReleaseSrvHandle(void* handle) {
   m->type = Release;
 
   tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
-  transAsyncSend(pThrd->asyncPool, &m->q);
+  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
+    destroySmsg(m);
+  }
 
   transReleaseExHandle(transGetRefMgt(), refId);
   return 0;
@@ -1269,7 +1269,9 @@ int transSendResponse(const STransMsg* msg) {
 
   STraceId* trace = (STraceId*)&msg->info.traceId;
   tGTrace("conn %p start to send resp (1/2)", exh->handle);
-  transAsyncSend(pThrd->asyncPool, &m->q);
+  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
+    destroySmsg(m);
+  }
 
   transReleaseExHandle(transGetRefMgt(), refId);
   return 0;
@@ -1303,7 +1305,9 @@ int transRegisterMsg(const STransMsg* msg) {
 
   STrans* pTransInst = pThrd->pTransInst;
   tTrace("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
-  transAsyncSend(pThrd->asyncPool, &m->q);
+  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
+    destroySmsg(m);
+  }
 
   transReleaseExHandle(transGetRefMgt(), refId);
   return 0;

source/os/src/osDir.c

@@ -336,6 +336,7 @@ int32_t taosRealPath(char *dirname, char *realPath, int32_t maxlen) {
 #else
   if (realpath(dirname, tmp) != NULL) {
 #endif
+    if (strlen(tmp) < maxlen) {
       if (realPath == NULL) {
         strncpy(dirname, tmp, maxlen);
       } else {
@@ -343,6 +344,7 @@ int32_t taosRealPath(char *dirname, char *realPath, int32_t maxlen) {
       }
       return 0;
     }
+  }
 
   return -1;
 }

source/util/src/terror.c

@@ -344,6 +344,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TDB_STB_ALREADY_EXIST,        "Stable already exists
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_STB_NOT_EXIST,            "Stable not exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER, "Table schema is old")
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TDB_ENV_OPEN_ERROR,       "TDB env open error")
+TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TABLE_IN_OTHER_STABLE,    "Table already exists in other stables")
 
 // query
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_QHANDLE,          "Invalid handle")

tests/parallel_test/cases.task

@@ -11,7 +11,7 @@
 ,,y,script,./test.sh -f tsim/user/privilege_db.sim
 ,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim
 ,,y,script,./test.sh -f tsim/db/alter_option.sim
-#,,y,script,./test.sh -f tsim/db/alter_replica_13.sim
+,,y,script,./test.sh -f tsim/db/alter_replica_13.sim
 ,,y,script,./test.sh -f tsim/db/alter_replica_31.sim
 ,,y,script,./test.sh -f tsim/db/basic1.sim
 ,,y,script,./test.sh -f tsim/db/basic2.sim
@@ -48,7 +48,7 @@
 ,,y,script,./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica3.sim
 ,,y,script,./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica1.sim
 ,,y,script,./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica3.sim
-,,,script,./test.sh -f tsim/dnode/drop_dnode_force.sim
+,,y,script,./test.sh -f tsim/dnode/drop_dnode_force.sim
 ,,y,script,./test.sh -f tsim/dnode/offline_reason.sim
 ,,y,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica1.sim
 ,,y,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim
@@ -56,7 +56,7 @@
 ,,y,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v2.sim
 ,,y,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v3.sim
 ,,y,script,./test.sh -f tsim/dnode/vnode_clean.sim
-,,,script,./test.sh -f tsim/dnode/use_dropped_dnode.sim 
+,,y,script,./test.sh -f tsim/dnode/use_dropped_dnode.sim 
 ,,y,script,./test.sh -f tsim/dnode/split_vgroup_replica1.sim 
 ,,y,script,./test.sh -f tsim/dnode/split_vgroup_replica3.sim 
 ,,y,script,./test.sh -f tsim/import/basic.sim
@@ -113,7 +113,7 @@
 ,,y,script,./test.sh -f tsim/parser/first_last.sim
 ,,y,script,./test.sh -f tsim/parser/fill_stb.sim
 ,,y,script,./test.sh -f tsim/parser/interp.sim
-#,,y,script,./test.sh -f tsim/parser/limit2.sim
+,,y,script,./test.sh -f tsim/parser/limit2.sim
 ,,y,script,./test.sh -f tsim/parser/fourArithmetic-basic.sim
 ,,y,script,./test.sh -f tsim/parser/function.sim
 ,,y,script,./test.sh -f tsim/parser/groupby-basic.sim
@@ -136,8 +136,8 @@
 ,,y,script,./test.sh -f tsim/parser/lastrow.sim
 ,,y,script,./test.sh -f tsim/parser/lastrow2.sim
 ,,y,script,./test.sh -f tsim/parser/like.sim
-,,,script,./test.sh -f tsim/parser/limit.sim
-,,,script,./test.sh -f tsim/parser/limit1.sim
+,,y,script,./test.sh -f tsim/parser/limit.sim
+,,y,script,./test.sh -f tsim/parser/limit1.sim
 ,,y,script,./test.sh -f tsim/parser/mixed_blocks.sim
 ,,y,script,./test.sh -f tsim/parser/nchar.sim
 ,,y,script,./test.sh -f tsim/parser/nestquery.sim
@@ -163,7 +163,7 @@
 ,,y,script,./test.sh -f tsim/parser/timestamp.sim
 ,,y,script,./test.sh -f tsim/parser/top_groupby.sim
 ,,y,script,./test.sh -f tsim/parser/topbot.sim
-,,,script,./test.sh -f tsim/parser/union.sim
+,,y,script,./test.sh -f tsim/parser/union.sim
 ,,y,script,./test.sh -f tsim/parser/union_sysinfo.sim
 ,,y,script,./test.sh -f tsim/parser/where.sim
 ,,y,script,./test.sh -f tsim/query/charScalarFunction.sim
@@ -176,11 +176,11 @@
 ,,y,script,./test.sh -f tsim/query/udf.sim
 ,,y,script,./test.sh -f tsim/qnode/basic1.sim
 ,,y,script,./test.sh -f tsim/snode/basic1.sim
-,,,script,./test.sh -f tsim/mnode/basic1.sim
-,,,script,./test.sh -f tsim/mnode/basic2.sim
-,,,script,./test.sh -f tsim/mnode/basic3.sim
-,,,script,./test.sh -f tsim/mnode/basic4.sim
-,,,script,./test.sh -f tsim/mnode/basic5.sim
+,,y,script,./test.sh -f tsim/mnode/basic1.sim
+,,y,script,./test.sh -f tsim/mnode/basic2.sim
+,,y,script,./test.sh -f tsim/mnode/basic3.sim
+,,y,script,./test.sh -f tsim/mnode/basic4.sim
+,,y,script,./test.sh -f tsim/mnode/basic5.sim
 ,,y,script,./test.sh -f tsim/show/basic.sim
 ,,y,script,./test.sh -f tsim/table/autocreate.sim
 ,,y,script,./test.sh -f tsim/table/basic1.sim
@@ -278,7 +278,7 @@
 ,,y,script,./test.sh -f tsim/stable/values.sim
 ,,y,script,./test.sh -f tsim/stable/vnode3.sim
 ,,y,script,./test.sh -f tsim/stable/metrics_idx.sim
-,,,script,./test.sh -f tsim/sma/drop_sma.sim
+,,n,script,./test.sh -f tsim/sma/drop_sma.sim
 ,,y,script,./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
 ,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
 ,,y,script,./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
@@ -291,7 +291,7 @@
 ,,n,script,./test.sh -f tsim/valgrind/checkError7.sim
 ,,n,script,./test.sh -f tsim/valgrind/checkError8.sim
 ,,n,script,./test.sh -f tsim/valgrind/checkUdf.sim
-,,,script,./test.sh -f tsim/vnode/replica3_basic.sim
+,,y,script,./test.sh -f tsim/vnode/replica3_basic.sim
 ,,y,script,./test.sh -f tsim/vnode/replica3_repeat.sim
 ,,y,script,./test.sh -f tsim/vnode/replica3_vgroup.sim
 ,,y,script,./test.sh -f tsim/vnode/replica3_many.sim
@@ -424,6 +424,7 @@
 ,,,system-test,python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py
 ,,,system-test,python3 ./test.py -f 1-insert/alter_stable.py
 ,,,system-test,python3 ./test.py -f 1-insert/alter_table.py
+,,,system-test,python3 ./test.py -f 1-insert/boundary.py
 ,,,system-test,python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
 ,,,system-test,python3 ./test.py -f 1-insert/table_comment.py
 ,,,system-test,python3 ./test.py -f 1-insert/time_range_wise.py
@@ -607,6 +608,8 @@
 ,,,system-test,python3 ./test.py -f 2-query/upper.py -R
 ,,,system-test,python3 ./test.py -f 2-query/varchar.py
 ,,,system-test,python3 ./test.py -f 2-query/varchar.py -R
+,,,system-test,python3 ./test.py -f 2-query/case_when.py 
+,,,system-test,python3 ./test.py -f 2-query/case_when.py -R
 ,,,system-test,python3 ./test.py -f 1-insert/update_data.py
 ,,,system-test,python3 ./test.py -f 1-insert/tb_100w_data_order.py
 ,,,system-test,python3 ./test.py -f 1-insert/delete_stable.py
@@ -813,6 +816,7 @@
 ,,,system-test,python3 ./test.py -f 2-query/last_row.py -Q 2
 ,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 2
 ,,,system-test,python3 ./test.py -f 2-query/sml.py -Q 2
+,,,system-test,python3 ./test.py -f 2-query/case_when.py -Q 2
 ,,,system-test,python3 ./test.py -f 2-query/between.py -Q  3
 ,,,system-test,python3 ./test.py -f 2-query/distinct.py -Q  3
 ,,,system-test,python3 ./test.py -f 2-query/varchar.py -Q  3
@@ -906,6 +910,7 @@
 ,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 3
 ,,,system-test,python3 ./test.py -f 2-query/sml.py -Q 3
 ,,,system-test,python3 ./test.py -f 2-query/interp.py -Q 3
+,,,system-test,python3 ./test.py -f 2-query/case_when.py -Q 3
 ,,,system-test,python3 ./test.py -f 2-query/between.py -Q  4
 ,,,system-test,python3 ./test.py -f 2-query/distinct.py -Q  4
 ,,,system-test,python3 ./test.py -f 2-query/varchar.py -Q  4
@@ -998,6 +1003,7 @@
 ,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 4
 ,,,system-test,python3 ./test.py -f 2-query/sml.py -Q 4
 ,,,system-test,python3 ./test.py -f 2-query/interp.py -Q 4
+,,,system-test,python3 ./test.py -f 2-query/case_when.py -Q 4
 
 #develop test
 ,,,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py

tests/pytest/util/sql.py

@@ -73,8 +73,15 @@ class TDSql:
         expectErrNotOccured = True
         try:
             self.cursor.execute(sql)
-        except BaseException:
+        except BaseException as e:
             expectErrNotOccured = False
+            caller = inspect.getframeinfo(inspect.stack()[1][0])
+            self.error_info = repr(e)
+            # print(error_info)
+            # self.error_info = error_info[error_info.index('(')+1:-1].split(",")[0].replace("'","")
+            # self.error_info = (','.join(error_info.split(",")[:-1]).split("(",1)[1:][0]).replace("'","")
+            # print("!!!!!!!!!!!!!!",self.error_info)
+            
         if expectErrNotOccured:
             caller = inspect.getframeinfo(inspect.stack()[1][0])
             tdLog.exit("%s(%d) failed: sql:%s, expect error not occured" % (caller.filename, caller.lineno, sql))
@@ -83,6 +90,8 @@ class TDSql:
             self.queryCols = 0
             self.queryResult = None
             tdLog.info("sql:%s, expect error occured" % (sql))
+            return self.error_info
+            
 
     def query(self, sql, row_tag=None,queryTimes=10):
         self.sql = sql

tests/script/tsim/db/alter_replica_13.sim

@@ -138,10 +138,10 @@ while $i < 10
   if $data[0][4] != leader then
     return -1
   endi
-  if $data[0][6] != follower then
+  if $data[0][6] == leader then
     return -1
   endi
-  if $data[0][8] != follower then
+  if $data[0][8] == leader then
     return -1
   endi
 endw 

tests/script/tsim/parser/limit2_query.sim

@@ -321,55 +321,41 @@ endi
 
 ### [TBASE-350]
 ## stb + interval + fill + group by + limit offset
-sql select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from $stb where ts >= $ts0 and ts <= $tsu partition by t1 interval(5m) fill(value, -1, -2, -3, -4 ,-7 ,'-8', '-9') limit 2 offset 10
+sql select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from lm2_tb0 where ts >= 1537146000000 and ts <= 1543145400000 partition by t1 interval(5m) fill(value, -1, -2, -3, -4 ,-7 ,'-8', '-9') limit 2 offset 10
 if $rows != 2 then
   return -1
 endi
-#add one more test case
-sql select max(c1), last(c8) from lm2_db0.lm2_tb0 where ts >= 1537146000000 and ts <= 1543145400000 interval(5m) fill(linear) limit 10 offset 4089;"
-
 
 $limit = 5
 $offset = $rowNum * 2
 $offset = $offset - 2
-sql select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from $stb where ts >= $ts0 and ts <= $tsu partition by t1 interval(5m) fill(value, -1, -2, -3, -4 ,-7 ,'-8', '-9') order by t1 limit $limit offset $offset
-if $rows != $tbNum then
+sql select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from lm2_tb0 where ts >= 1537146000000 and ts <= 1543145400000  partition by t1 interval(5m) fill(value, -1, -2, -3, -4 ,-7 ,'-8', '-9') order by t1 limit $limit offset $offset
+if $rows != 1 then
   return -1
 endi
-if $data00 != @18-11-25 19:30:00.000@ then
+if $data00 != 9 then
   return -1
 endi
 if $data01 != 9 then
   return -1
 endi
-if $data12 != 9 then
+if $data02 != 9.000000000 then
   return -1
 endi
-if $data23 != 9.000000000 then
+if $data03 != 9.000000000 then
   return -1
 endi
-if $data34 != 9.000000000 then
+if $data04 != 1 then
   return -1
 endi
-if $data45 != 1 then
+if $data05 != binary9 then
   return -1
 endi
-if $data56 != binary9 then
-  return -1
-endi
-if $data68 != 6 then
-  return -1
-endi
-if $data72 != -2 then
-  return -1
-endi
-if $data84 != -2.000000000 then
-  return -1
-endi
-if $data98 != 9 then
+if $data06 != nchar9 then
   return -1
 endi
 
+
 #add one more test case
 sql select max(c1), last(c8) from lm2_db0.lm2_tb0 where ts >= 1537146000000 and ts <= 1543145400000 interval(5m) fill(linear) limit 10 offset 4089;"
 

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

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+
+import math
+from random import randint
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+from util.boundary import *
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+        self.boundary = DataBoundary()
+        self.dbname_length_boundary = self.boundary.DBNAME_MAX_LENGTH
+        self.tbname_length_boundary = self.boundary.TBNAME_MAX_LENGTH
+        self.stbname_length_boundary = self.boundary.STBNAME_MAX_LENGTH
+        self.colname_length_boundary = self.boundary.COL_KEY_MAX_LENGTH
+        self.tagname_length_boundary = self.boundary.TAG_KEY_MAX_LENGTH
+        self.username_length_boundary = 23
+        self.password_length_boundary = 128
+    def dbname_length_check(self):
+        dbname_length = randint(1,self.dbname_length_boundary-1)
+        for dbname in [tdCom.get_long_name(self.dbname_length_boundary),tdCom.get_long_name(dbname_length)]:
+            tdSql.execute(f'create database if not exists {dbname}')
+            tdSql.query(f'select name from information_schema.ins_databases where name = "{dbname}"')
+            tdSql.checkEqual(tdSql.queryResult[0][0],dbname)
+            tdSql.execute(f'drop database if exists {dbname}')
+        dbname = tdCom.get_long_name(self.dbname_length_boundary+1)
+        tdSql.error(f'create database if not exists {dbname}')
+        if "Invalid identifier name" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+
+    def tbname_length_check(self):
+        tdSql.prepare()
+        tdSql.execute('use db')
+        tbname_length = randint(1,self.tbname_length_boundary-1)
+        tdSql.execute(f'create table stb (ts timestamp,c0 int) tags(t0 int)')
+        for tbname in [tdCom.get_long_name(self.tbname_length_boundary),tdCom.get_long_name(tbname_length)]:
+            tdSql.execute(f'create table {tbname} using stb tags(1)')
+            tdSql.query(f'select table_name from information_schema.ins_tables where table_name = "{tbname}"')
+            tdSql.checkEqual(tdSql.queryResult[0][0],tbname)
+            tdSql.execute(f'drop table {tbname}')
+        tbname = tdCom.get_long_name(self.tbname_length_boundary+1)
+        tdSql.error(f'create table {tbname} using stb tags(1)')
+        if "Invalid identifier name" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+        stbname_length = randint(1,self.stbname_length_boundary-1)
+        for stbname in [tdCom.get_long_name(self.stbname_length_boundary),tdCom.get_long_name(stbname_length)]:
+            tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
+            tdSql.query(f'select stable_name from information_schema.ins_stables where stable_name = "{stbname}"')
+            tdSql.checkEqual(tdSql.queryResult[0][0],stbname)
+            tdSql.execute(f'drop table {stbname}')
+        stbname = tdCom.get_long_name(self.stbname_length_boundary+1)
+        tdSql.error(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
+        print(tdSql.error_info)
+        if "Invalid identifier name" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+        tdSql.execute('drop database db')
+
+    def colname_length_check(self):
+        tdSql.prepare()
+        tdSql.execute('use db')
+        column_name_length = randint(1,self.colname_length_boundary-1)
+        for colname in [tdCom.get_long_name(column_name_length),tdCom.get_long_name(self.colname_length_boundary)]:
+            stbname = tdCom.get_long_name(3)
+            ntbname = tdCom.get_long_name(4)
+            tdSql.execute(f'create table {stbname} (ts timestamp,{colname} int) tags(t0 int)')
+            tdSql.query(f'describe {stbname}')
+            tdSql.checkEqual(tdSql.queryResult[1][0],colname)
+            tdSql.execute(f'create table {ntbname} (ts timestamp,{colname} int)')
+            tdSql.query(f'describe {ntbname}')
+            tdSql.checkEqual(tdSql.queryResult[1][0],colname)
+        colname = tdCom.get_long_name(self.colname_length_boundary+1)
+        tdSql.error(f'create table stb (ts timestamp,{colname} int) tags(t0 int)')
+        if "Invalid identifier name" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+        tdSql.execute('drop database db')
+
+    def tagname_length_check(self):
+        tdSql.prepare()
+        tdSql.execute('use db')
+        tag_name_length = randint(1,self.tagname_length_boundary-1)
+        for tagname in (tdCom.get_long_name(tag_name_length),tdCom.get_long_name(self.tagname_length_boundary)):
+            stbname = tdCom.get_long_name(3)
+            tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags({tagname} int)')
+            tdSql.query(f'describe {stbname}')
+            tdSql.checkEqual(tdSql.queryResult[-1][0],tagname)
+        tagname = tdCom.get_long_name(self.tagname_length_boundary+1)
+        tdSql.error(f'create table {stbname} (ts timestamp,c0 int) tags({tagname} int)')
+        if "Invalid identifier name" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+        tdSql.execute('drop database db')
+
+    def username_length_check(self):
+        username_length = randint(1,self.username_length_boundary-1)
+        for username in [tdCom.get_long_name(username_length),tdCom.get_long_name(self.username_length_boundary)]:
+            tdSql.execute(f'create user {username} pass "123"')
+            tdSql.query('show users')
+            for user in tdSql.queryResult:
+                if user[0].lower() != 'root':
+                    tdSql.checkEqual(user[0],username)
+            tdSql.execute(f'drop user {username}')
+        username = tdCom.get_long_name(self.username_length_boundary+1)
+        tdSql.error(f'create user {username} pass "123"')
+        if "Name or password too long" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+    
+    def password_length_check(self):
+        password_length = randint(1,self.password_length_boundary-1)
+        for password in [tdCom.get_long_name(password_length),tdCom.get_long_name(self.password_length_boundary)]:
+            username = tdCom.get_long_name(3)
+            tdSql.execute(f'create user {username} pass "{password}"')
+        password = tdCom.get_long_name(self.password_length_boundary+1)
+        tdSql.error(f'create user {username} pass "{password}"')
+        if "Name or password too long" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+    def sql_length_check(self):
+        insert_rows = 1021
+        tdSql.prepare()
+        tdSql.execute('use db')
+        tdSql.execute('create table ntb (ts timestamp,c0 binary(1013))')
+        values_sql = ''
+        value = tdCom.get_long_name(1013)
+        for num in range(insert_rows):
+            values_sql += f' (now+{num}s,"{value}")'
+        value = tdCom.get_long_name(65)
+        values_sql += f"(now-1s,'{value}')"
+        tdSql.execute(f'insert into ntb values{values_sql}')
+        tdSql.query('select * from ntb')
+        tdSql.checkRows(insert_rows+1)
+        tdSql.execute('create table ntb1 (ts timestamp,c0 binary(1013))')
+        tdSql.error(f'insert into ntb1 values{values_sql};')
+        print(tdSql.error_info)
+        if "SQL statement too long" in tdSql.error_info:
+            tdLog.info("error info is true!")
+        else:
+            tdLog.exit("error info is not true")
+        tdSql.execute('drop database db')
+    def run(self):
+        self.dbname_length_check()
+        self.tbname_length_check()
+        self.colname_length_check()
+        self.tagname_length_check()
+        self.username_length_check()
+        self.password_length_check()
+        self.sql_length_check()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

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

@@ -23,7 +23,7 @@ class TDTestCase:
     def init(self, conn, logSql, replicaVar=1):
         self.replicaVar = int(replicaVar)
         tdLog.debug("start to execute %s" % __file__)
-        tdSql.init(conn.cursor(),logSql)
+        tdSql.init(conn.cursor())
         self.setsql = TDSetSql()
         self.dbname = 'db_test'
         self.ntbname = 'ntb'