Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/hzcheng_3.0

include/libs/nodes/nodes.h

@@ -208,6 +208,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_SORT,
   QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL,
+  QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_FILL,
   QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,

include/libs/nodes/querynodes.h

@@ -248,6 +248,7 @@ typedef struct SSetOperator {
   SNode*           pRight;
   SNodeList*       pOrderByList;  // SOrderByExprNode
   SNode*           pLimit;
+  char             stmtName[TSDB_TABLE_NAME_LEN];
 } SSetOperator;
 
 typedef enum ESqlClause {

include/util/taoserror.h

@@ -646,6 +646,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_ALTER_TABLE       TAOS_DEF_ERROR_CODE(0, 0x2649)
 #define TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY   TAOS_DEF_ERROR_CODE(0, 0x264A)
 #define TSDB_CODE_PAR_INVALID_MODIFY_COL        TAOS_DEF_ERROR_CODE(0, 0x264B)
+#define TSDB_CODE_PAR_INVALID_TBNAME            TAOS_DEF_ERROR_CODE(0, 0x264C)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/libs/executor/inc/executorimpl.h

@@ -378,6 +378,13 @@ typedef enum EStreamScanMode {
   STREAM_SCAN_FROM_DATAREADER,
 } EStreamScanMode;
 
+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 SStreamBlockScanInfo {
   SArray*      pBlockLists;      // multiple SSDatablock.
   SSDataBlock* pRes;             // result SSDataBlock
@@ -400,6 +407,8 @@ typedef struct SStreamBlockScanInfo {
   EStreamScanMode scanMode;
   SOperatorInfo* pOperatorDumy;
   SInterval      interval;     // if the upstream is an interval operator, the interval info is also kept here.
+  SCatchSupporter childAggSup;
+  SArray* childIds;
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -460,6 +469,16 @@ typedef struct SIntervalAggOperatorInfo {
   bool               invertible;
 } SIntervalAggOperatorInfo;
 
+typedef struct SStreamFinalIntervalOperatorInfo {
+  SOptrBasicInfo     binfo;              // basic info
+  SGroupResInfo      groupResInfo;       // multiple results build supporter
+  SInterval          interval;           // interval info
+  int32_t            primaryTsIndex;     // primary time stamp slot id from result of downstream operator.
+  SAggSupporter      aggSup;             // aggregate supporter
+  int32_t            order;              // current SSDataBlock scan order
+  STimeWindowAggSupp twAggSup;
+} SStreamFinalIntervalOperatorInfo;
+
 typedef struct SAggOperatorInfo {
   SOptrBasicInfo     binfo;
   SAggSupporter      aggSup;
@@ -696,6 +715,9 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                           STimeWindowAggSupp *pTwAggSupp, const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
+                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
+                                          STimeWindowAggSupp *pTwAggSupp, const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
 
                                                 SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
@@ -771,9 +793,8 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary
                                         TSKEY ekey, __block_search_fn_t searchFn, STableQueryInfo* item,
                                         int32_t order);
 int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
-
-void doClearWindow(SIntervalAggOperatorInfo* pInfo, char* pData, int16_t bytes,
-    uint64_t groupId, int32_t numOfOutput);
+int32_t initCatchSupporter(SCatchSupporter* pCatchSup, size_t rowSize, size_t keyBufSize,
+                         const char* pKey, const char* pDir);
 
 #ifdef __cplusplus
 }

source/libs/executor/src/executorimpl.c

@@ -342,28 +342,6 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId,
   return pResultRow;
 }
 
-void doClearWindow(SIntervalAggOperatorInfo* pInfo, char* pData, int16_t bytes,
-    uint64_t groupId, int32_t numOfOutput) {
-  SAggSupporter* pSup = &pInfo->aggSup;
-  SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId);
-  SResultRowPosition* p1 =
-      (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf,
-          GET_RES_WINDOW_KEY_LEN(bytes));
-  SResultRow* pResult = getResultRowByPos(pSup->pResultBuf, p1);
-  SqlFunctionCtx* pCtx = pInfo->binfo.pCtx;
-  for (int32_t i = 0; i < numOfOutput; ++i) {
-    pCtx[i].resultInfo = getResultCell(pResult, i, pInfo->binfo.rowCellInfoOffset);
-    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
-    if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) {
-      continue;
-    }
-    pResInfo->initialized = false;
-    if (pCtx[i].functionId != -1) {
-      pCtx[i].fpSet.init(&pCtx[i], pResInfo);
-    }
-  }
-}
-
 /**
  * the struct of key in hash table
  * +----------+---------------+
@@ -5321,3 +5299,16 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
 
   return TSDB_CODE_SUCCESS;
 }
+
+int32_t initCatchSupporter(SCatchSupporter* pCatchSup, size_t rowSize, size_t keyBufSize,
+                         const char* pKey, const char* pDir) {
+  pCatchSup->keySize = sizeof(int64_t) + sizeof(int64_t) + sizeof(TSKEY);
+  pCatchSup->pKeyBuf = taosMemoryCalloc(1, pCatchSup->keySize);
+  int32_t pageSize = rowSize * 32;
+  int32_t bufSize = pageSize * 4096;
+  createDiskbasedBuf(&pCatchSup->pDataBuf, pageSize, bufSize, pKey, pDir);
+  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
+  pCatchSup->pWindowHashTable = taosHashInit(10000, hashFn, true, HASH_NO_LOCK);;
+  return TSDB_CODE_SUCCESS;
+}
+                        
\ No newline at end of file

source/libs/executor/src/scanoperator.c

@@ -36,6 +36,11 @@
 #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
 #define SWITCH_ORDER(n)              (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
 
+typedef struct SWindowPosition {
+  int32_t pageId;
+  int32_t rowId;
+} SWindowPosition;
+
 static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity);
 static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                                      const char* dbName);
@@ -675,6 +680,96 @@ static SSDataBlock* getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool inverti
   return NULL;
 }
 
+void static setSupKeyBuf(SCatchSupporter* pSup, int64_t groupId, int64_t childId, TSKEY ts) {
+  int64_t* pKey = (int64_t*)pSup->pKeyBuf;
+  pKey[0] = groupId;
+  pKey[1] = childId;
+  pKey[2] = ts;
+}
+
+static int32_t catchWidonwInfo(SSDataBlock* pDataBlock, SCatchSupporter* pSup,
+    int32_t pageId, int32_t tsIndex, int64_t childId) {
+  SColumnInfoData* pColDataInfo = taosArrayGet(pDataBlock->pDataBlock, tsIndex);
+  TSKEY* tsCols = (int64_t*)pColDataInfo->pData;
+  for (int32_t i = 0; i < pDataBlock->info.rows; i++) {
+    setSupKeyBuf(pSup, pDataBlock->info.groupId, childId, tsCols[i]);
+    SWindowPosition* p1 = (SWindowPosition*)taosHashGet(pSup->pWindowHashTable,
+        pSup->pKeyBuf, pSup->keySize);
+    if (p1 == NULL) {
+      SWindowPosition pos = {.pageId = pageId, .rowId = i};
+      int32_t code = taosHashPut(pSup->pWindowHashTable, pSup->pKeyBuf, pSup->keySize, &pos,
+          sizeof(SWindowPosition));
+      if (code != TSDB_CODE_SUCCESS ) {
+        return code;
+      }
+    } else {
+      p1->pageId = pageId;
+      p1->rowId = i;
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t catchDatablock(SSDataBlock* pDataBlock, SCatchSupporter* pSup,
+    int32_t tsIndex, int64_t childId) {
+  int32_t start = 0;
+  int32_t stop = 0;
+  int32_t pageSize = getBufPageSize(pSup->pDataBuf);
+  while(start < pDataBlock->info.rows) {
+    blockDataSplitRows(pDataBlock, pDataBlock->info.hasVarCol, start, &stop, pageSize);
+    SSDataBlock* pDB = blockDataExtractBlock(pDataBlock, start, stop - start + 1);
+    if (pDB == NULL) {
+      return terrno;
+    }
+    int32_t pageId = -1;
+    void* pPage = getNewBufPage(pSup->pDataBuf, pDataBlock->info.groupId, &pageId);
+    if (pPage == NULL) {
+      blockDataDestroy(pDB);
+      return terrno;
+    }
+    int32_t size = blockDataGetSize(pDB) + sizeof(int32_t)  + pDB->info.numOfCols * sizeof(int32_t);
+    assert(size <= pageSize);
+    blockDataToBuf(pPage, pDB);
+    setBufPageDirty(pPage, true);
+    releaseBufPage(pSup->pDataBuf, pPage);
+    blockDataDestroy(pDB);
+    start = stop + 1;
+    int32_t code = catchWidonwInfo(pDB, pSup, pageId, tsIndex, childId);
+    if (code != TSDB_CODE_SUCCESS ) {
+      return code;
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static SSDataBlock* getDataFromCatch(SStreamBlockScanInfo* pInfo) {
+  SSDataBlock* pBlock = pInfo->pUpdateRes;
+  if (pInfo->updateResIndex < pBlock->info.rows) {
+    blockDataCleanup(pInfo->pRes);
+    SCatchSupporter* pCSup = &pInfo->childAggSup;
+    SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, 0);
+    TSKEY *tsCols = (TSKEY*)pColDataInfo->pData;
+    int32_t size = taosArrayGetSize(pInfo->childIds);
+    for (int32_t i = 0; i < size; i++) {
+      int64_t id = *(int64_t *)taosArrayGet(pInfo->childIds, i);
+      setSupKeyBuf(pCSup, pBlock->info.groupId, id,
+          tsCols[pInfo->updateResIndex]);
+      SWindowPosition* pos = (SWindowPosition*)taosHashGet(pCSup->pWindowHashTable,
+          pCSup->pKeyBuf, pCSup->keySize);
+      void* buf = getBufPage(pCSup->pDataBuf, pos->pageId);
+      SSDataBlock* pDB = createOneDataBlock(pInfo->pRes, false);
+      blockDataFromBuf(pDB, buf);
+      SSDataBlock* pSub = blockDataExtractBlock(pDB, pos->rowId, 1);
+      blockDataMerge(pInfo->pRes, pSub, NULL);
+      blockDataDestroy(pDB);
+      blockDataDestroy(pSub);
+    }
+    pInfo->updateResIndex++;
+    return pInfo->pRes;
+  }
+  return NULL;
+}
+
 static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;
@@ -688,6 +783,15 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
 
   size_t total = taosArrayGetSize(pInfo->pBlockLists);
   if (pInfo->blockType == STREAM_DATA_TYPE_SSDATA_BLOCK) {
+    if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
+      SSDataBlock* pDB = getDataFromCatch(pInfo);
+      if (pDB != NULL) {
+        return pDB;
+      } else {
+       pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+      }
+    }
+
     if (pInfo->validBlockIndex >= total) {
       doClearBufferedBlocks(pInfo);
       pOperator->status = OP_EXEC_DONE;
@@ -695,7 +799,17 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     }
 
     int32_t current = pInfo->validBlockIndex++;
-    return taosArrayGetP(pInfo->pBlockLists, current);
+    SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
+    if (pBlock->info.type == STREAM_REPROCESS) {
+      pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
+    } else {
+      int32_t code = catchDatablock(pBlock, &pInfo->childAggSup, pInfo->primaryTsIndex, 0);
+      if (code != TDB_CODE_SUCCESS) {
+        pTaskInfo->code = code;
+        longjmp(pTaskInfo->env, code);
+      }
+    }
+    return pBlock;
   } else {
     if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
       blockDataDestroy(pInfo->pUpdateRes);
@@ -857,6 +971,10 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataR
   pInfo->pOperatorDumy  = pOperatorDumy;
   pInfo->interval       = pSTInfo->interval;
 
+  size_t childKeyBufSize = sizeof(int64_t) + sizeof(int64_t) + sizeof(TSKEY);
+  initCatchSupporter(&pInfo->childAggSup, 1024, childKeyBufSize,
+      "StreamFinalInterval", "/tmp/"); // TODO(liuyao) get row size from phy plan
+
   pOperator->name       = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
   pOperator->blocking   = false;

source/libs/executor/src/timewindowoperator.c

@@ -8,6 +8,8 @@ typedef enum SResultTsInterpType {
   RESULT_ROW_END_INTERP = 2,
 } SResultTsInterpType;
 
+static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator);
+
 /*
  * There are two cases to handle:
  *
@@ -473,8 +475,7 @@ static bool setTimeWindowInterpolationEndTs(SOperatorInfo* pOperatorInfo, SqlFun
 }
 
 static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo,
-                                      TSKEY* primaryKeys, int32_t prevPosition, SIntervalAggOperatorInfo* pInfo) {
-  int32_t order = pInfo->order;
+                                      TSKEY* primaryKeys, int32_t prevPosition, int32_t order) {
   bool    ascQuery = (order == TSDB_ORDER_ASC);
 
   int32_t precision = pInterval->precision;
@@ -723,7 +724,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
   STimeWindow nextWin = win;
   while (1) {
     int32_t prevEndPos = (forwardStep - 1) * step + startPos;
-    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, pInfo);
+    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, pInfo->order);
     if (startPos < 0) {
       break;
     }
@@ -1031,18 +1032,41 @@ static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type
     }
   }
 }
-static void doClearWindows(SIntervalAggOperatorInfo* pInfo, int32_t numOfOutput, SSDataBlock* pBlock) {
-  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
+
+void doClearWindow(SAggSupporter* pSup, SOptrBasicInfo* pBinfo, char* pData,
+    int16_t bytes, uint64_t groupId, int32_t numOfOutput) {
+  SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId);
+  SResultRowPosition* p1 =
+      (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf,
+          GET_RES_WINDOW_KEY_LEN(bytes));
+  SResultRow* pResult = getResultRowByPos(pSup->pResultBuf, p1);
+  SqlFunctionCtx* pCtx = pBinfo->pCtx;
+  for (int32_t i = 0; i < numOfOutput; ++i) {
+    pCtx[i].resultInfo = getResultCell(pResult, i, pBinfo->rowCellInfoOffset);
+    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
+    if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) {
+      continue;
+    }
+    pResInfo->initialized = false;
+    if (pCtx[i].functionId != -1) {
+      pCtx[i].fpSet.init(&pCtx[i], pResInfo);
+    }
+  }
+}
+
+static void doClearWindows(SAggSupporter* pSup, SOptrBasicInfo* pBinfo,
+    SInterval* pIntrerval, int32_t tsIndex, int32_t numOfOutput, SSDataBlock* pBlock) {
+  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
   TSKEY *tsCols = (TSKEY*)pColDataInfo->pData;
   int32_t step = 0;
   for (int32_t i = 0; i < pBlock->info.rows; i += step) {
     SResultRowInfo dumyInfo;
     dumyInfo.cur.pageId = -1;
-    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], &pInfo->interval,
-                                        pInfo->interval.precision, NULL);
+    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], pIntrerval,
+                                        pIntrerval->precision, NULL);
     step = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, i,
         win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
-    doClearWindow(pInfo, (char*)&win.skey, sizeof(TKEY), pBlock->info.groupId, numOfOutput);
+    doClearWindow(pSup, pBinfo, (char*)&win.skey, sizeof(TKEY), pBlock->info.groupId, numOfOutput);
   }
 }
 
@@ -1084,7 +1108,8 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type);
     }
     if (pBlock->info.type == STREAM_REPROCESS) {
-      doClearWindows(pInfo, pOperator->numOfExprs, pBlock);
+      doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval,
+          pInfo->primaryTsIndex, pOperator->numOfExprs, pBlock);
       continue;
     }
     pInfo->order = TSDB_ORDER_ASC;
@@ -1097,8 +1122,6 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
   doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
 
-  // TODO: remove for stream
-  /*ASSERT(pInfo->binfo.pRes->info.rows > 0);*/
   pOperator->status = OP_RES_TO_RETURN;
 
   return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
@@ -1116,6 +1139,12 @@ void destroyIntervalOperatorInfo(void* param, int32_t numOfOutput) {
   cleanupAggSup(&pInfo->aggSup);
 }
 
+void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
+  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo *)param;
+  doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
+  cleanupAggSup(&pInfo->aggSup);
+}
+
 bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
   for (int32_t i = 0; i < numOfCols; i++) {
     if (!fmIsInvertible(pFCtx[i].functionId)) {
@@ -1185,6 +1214,63 @@ _error:
   return NULL;
 }
 
+SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
+                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
+                                          STimeWindowAggSupp* pTwAggSupp, const STableGroupInfo* pTableGroupInfo,
+                                          SExecTaskInfo* pTaskInfo) {
+  SStreamFinalIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamFinalIntervalOperatorInfo));
+  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
+  if (pInfo == NULL || pOperator == NULL) {
+    goto _error;
+  }
+
+  pInfo->order = TSDB_ORDER_ASC;
+  pInfo->interval = *pInterval;
+  pInfo->twAggSup = *pTwAggSupp;
+  pInfo->primaryTsIndex = primaryTsSlotId;
+
+  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
+  initResultSizeInfo(pOperator, 4096);
+
+  int32_t code =
+      initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock,
+          keyBufSize, pTaskInfo->id.str);
+
+  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
+
+  initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)1);
+
+  pOperator->name = "StreamFinalIntervalOperator";
+  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;
+  pOperator->blocking = true;
+  pOperator->status = OP_NOT_OPENED;
+  pOperator->pExpr = pExprInfo;
+  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->numOfExprs = numOfCols;
+  pOperator->info = pInfo;
+
+  pOperator->fpSet = createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL,
+      destroyStreamFinalIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow,
+      NULL);
+
+  code = appendDownstream(pOperator, &downstream, 1);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
+
+  return pOperator;
+
+_error:
+  destroyStreamFinalIntervalOperatorInfo(pInfo, numOfCols);
+  taosMemoryFreeClear(pInfo);
+  taosMemoryFreeClear(pOperator);
+  pTaskInfo->code = code;
+  return NULL;
+}
+
 SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                                 SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                                 STimeWindowAggSupp* pTwAggSupp, const STableGroupInfo* pTableGroupInfo,
@@ -1548,3 +1634,91 @@ _error:
   pTaskInfo->code = code;
   return NULL;
 }
+
+static SArray* doHashInterval(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pSDataBlock,
+                               int32_t tableGroupId) {
+  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info;
+  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
+  int32_t numOfOutput = pOperatorInfo->numOfExprs;
+  SArray* pUpdated = taosArrayInit(4, POINTER_BYTES);
+  int32_t step = 1;
+  bool    ascScan = true;
+  TSKEY* tsCols = NULL;
+  SResultRow* pResult = NULL;
+  int32_t forwardStep = 0;
+
+  if (pSDataBlock->pDataBlock != NULL) {
+    SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
+    tsCols = (int64_t*)pColDataInfo->pData;
+  }
+  int32_t startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
+  TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols, pSDataBlock->info.rows, ascScan);
+  STimeWindow nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts,
+      &pInfo->interval, pInfo->interval.precision, NULL);
+  while (1) {
+    int32_t code =
+        setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pInfo->binfo.pCtx,
+                               numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
+    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
+      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
+    }
+    SResKeyPos* pos = taosMemoryMalloc(sizeof(SResKeyPos) + sizeof(uint64_t));
+    pos->groupId = tableGroupId;
+    pos->pos = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
+    *(int64_t*)pos->key = pResult->win.skey;
+    taosArrayPush(pUpdated, &pos);
+    forwardStep =
+        getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, nextWin.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
+    // window start(end) key interpolation
+    doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, forwardStep,
+                                pInfo->order, false);
+    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
+    doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardStep, tsCols,
+                     pSDataBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
+    int32_t prevEndPos = (forwardStep - 1) * step + startPos;
+    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, pInfo->order);
+    if (startPos < 0) {
+      break;
+    }
+  }
+  return pUpdated;
+}
+
+static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
+  SStreamFinalIntervalOperatorInfo* pInfo = pOperator->info;
+  SOperatorInfo* downstream = pOperator->pDownstream[0];
+  SArray* pUpdated = NULL;
+
+  if (pOperator->status == OP_EXEC_DONE) {
+    return NULL;
+  } else if (pOperator->status == OP_RES_TO_RETURN) {
+    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
+    if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
+      pOperator->status = OP_EXEC_DONE;
+    }
+    return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
+  }
+
+  while (1) {
+    publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
+    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
+    publishOperatorProfEvent(downstream, QUERY_PROF_AFTER_OPERATOR_EXEC);
+    if (pBlock == NULL) {
+      break;
+    }
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
+    if (pBlock->info.type == STREAM_REPROCESS) {
+      doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval,
+          pInfo->primaryTsIndex, pOperator->numOfExprs, pBlock);
+      continue;
+    }
+    pUpdated = doHashInterval(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
+  }
+
+  finalizeUpdatedResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pInfo->binfo.rowCellInfoOffset);
+  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
+  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
+  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
+  pOperator->status = OP_RES_TO_RETURN;
+  return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
+}

source/libs/parser/inc/sql.y

@@ -621,6 +621,7 @@ column_reference(A) ::= table_name(B) NK_DOT column_name(C).
 
 pseudo_column(A) ::= ROWTS(B).                                                    { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
 pseudo_column(A) ::= TBNAME(B).                                                   { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
+pseudo_column(A) ::= table_name(B) NK_DOT TBNAME(C).                              { A = createRawExprNodeExt(pCxt, &B, &C, createFunctionNode(pCxt, &C, createNodeList(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &B)))); }
 pseudo_column(A) ::= QSTARTTS(B).                                                 { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
 pseudo_column(A) ::= QENDTS(B).                                                   { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
 pseudo_column(A) ::= WSTARTTS(B).                                                 { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }

source/libs/parser/src/parAstCreater.c

@@ -155,9 +155,9 @@ static bool checkPort(SAstCreateContext* pCxt, const SToken* pPortToken, int32_t
   return TSDB_CODE_SUCCESS == pCxt->errCode;
 }
 
-static bool checkDbName(SAstCreateContext* pCxt, SToken* pDbName, bool query) {
+static bool checkDbName(SAstCreateContext* pCxt, SToken* pDbName, bool demandDb) {
   if (NULL == pDbName) {
-    if (query && NULL == pCxt->pQueryCxt->db) {
+    if (demandDb && NULL == pCxt->pQueryCxt->db) {
       pCxt->errCode = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_DB_NOT_SPECIFIED);
     }
   } else {
@@ -457,6 +457,8 @@ SNode* createTempTableNode(SAstCreateContext* pCxt, SNode* pSubquery, const STok
   }
   if (QUERY_NODE_SELECT_STMT == nodeType(pSubquery)) {
     strcpy(((SSelectStmt*)pSubquery)->stmtName, tempTable->table.tableAlias);
+  } else if (QUERY_NODE_SET_OPERATOR == nodeType(pSubquery)) {
+    strcpy(((SSetOperator*)pSubquery)->stmtName, tempTable->table.tableAlias);
   }
   return (SNode*)tempTable;
 }
@@ -637,6 +639,7 @@ SNode* createSetOperator(SAstCreateContext* pCxt, ESetOperatorType type, SNode*
   setOp->opType = type;
   setOp->pLeft = pLeft;
   setOp->pRight = pRight;
+  sprintf(setOp->stmtName, "%p", setOp);
   return (SNode*)setOp;
 }
 
@@ -1140,7 +1143,7 @@ SNode* createAlterDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode, const
 
 SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, bool ignoreExists, SToken* pIndexName,
                              SToken* pTableName, SNodeList* pCols, SNode* pOptions) {
-  if (!checkIndexName(pCxt, pIndexName) || !checkTableName(pCxt, pTableName)) {
+  if (!checkIndexName(pCxt, pIndexName) || !checkTableName(pCxt, pTableName) || !checkDbName(pCxt, NULL, true)) {
     return NULL;
   }
   SCreateIndexStmt* pStmt = nodesMakeNode(QUERY_NODE_CREATE_INDEX_STMT);

source/libs/parser/src/parTranslater.c

@@ -766,6 +766,20 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
       pCxt->pCurrStmt->hasRepeatScanFuncs = true;
     }
   }
+  if (TSDB_CODE_SUCCESS == pCxt->errCode && fmIsScanPseudoColumnFunc(pFunc->funcId)) {
+    if (0 == LIST_LENGTH(pFunc->pParameterList)) {
+      if (QUERY_NODE_REAL_TABLE != nodeType(pCxt->pCurrStmt->pFromTable)) {
+        return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_TBNAME);
+      }
+    } else {
+      SValueNode* pVal = nodesListGetNode(pFunc->pParameterList, 0);
+      STableNode* pTable = NULL;
+      pCxt->errCode = findTable(pCxt, pVal->literal, &pTable);
+      if (TSDB_CODE_SUCCESS == pCxt->errCode && (NULL == pTable || QUERY_NODE_REAL_TABLE != nodeType(pTable))) {
+        return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_TBNAME);
+      }
+    }
+  }
   return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR;
 }
 
@@ -1758,12 +1772,13 @@ static int32_t translateSelect(STranslateContext* pCxt, SSelectStmt* pSelect) {
   return code;
 }
 
-static SNode* createSetOperProject(SNode* pNode) {
+static SNode* createSetOperProject(const char* pTableAlias, SNode* pNode) {
   SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
   if (NULL == pCol) {
     return NULL;
   }
   pCol->node.resType = ((SExprNode*)pNode)->resType;
+  strcpy(pCol->tableAlias, pTableAlias);
   strcpy(pCol->colName, ((SExprNode*)pNode)->aliasName);
   strcpy(pCol->node.aliasName, pCol->colName);
   return (SNode*)pCol;
@@ -1817,7 +1832,8 @@ static int32_t translateSetOperatorImpl(STranslateContext* pCxt, SSetOperator* p
     }
     strcpy(pRightExpr->aliasName, pLeftExpr->aliasName);
     pRightExpr->aliasName[strlen(pLeftExpr->aliasName)] = '\0';
-    if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList, createSetOperProject(pLeft))) {
+    if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList,
+                                                       createSetOperProject(pSetOperator->stmtName, pLeft))) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
   }

source/libs/parser/src/parUtil.c

@@ -158,6 +158,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Primary timestamp column cannot be dropped";
     case TSDB_CODE_PAR_INVALID_MODIFY_COL:
       return "Only binary/nchar column length could be modified";
+    case TSDB_CODE_PAR_INVALID_TBNAME:
+      return "Invalid tbname pseudo column";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/parser/test/parSelectTest.cpp

@@ -70,6 +70,12 @@ TEST_F(ParserSelectTest, pseudoColumn) {
   run("SELECT _WSTARTTS, _WENDTS, COUNT(*) FROM t1 INTERVAL(10s)");
 }
 
+TEST_F(ParserSelectTest, pseudoColumnSemanticCheck) {
+  useDb("root", "test");
+
+  run("SELECT TBNAME FROM (SELECT * FROM st1s1)", TSDB_CODE_PAR_INVALID_TBNAME, PARSER_STAGE_TRANSLATE);
+}
+
 TEST_F(ParserSelectTest, multiResFunc) {
   useDb("root", "test");
 

source/libs/planner/inc/planInt.h

@@ -27,12 +27,13 @@ extern "C" {
 #define QUERY_POLICY_HYBRID 2
 #define QUERY_POLICY_QNODE  3
 
-#define planFatal(param, ...) qFatal("PLAN: " param, __VA_ARGS__)
-#define planError(param, ...) qError("PLAN: " param, __VA_ARGS__)
-#define planWarn(param, ...)  qWarn("PLAN: " param, __VA_ARGS__)
-#define planInfo(param, ...)  qInfo("PLAN: " param, __VA_ARGS__)
-#define planDebug(param, ...) qDebug("PLAN: " param, __VA_ARGS__)
-#define planTrace(param, ...) qTrace("PLAN: " param, __VA_ARGS__)
+#define planFatal(param, ...)  qFatal("PLAN: " param, __VA_ARGS__)
+#define planError(param, ...)  qError("PLAN: " param, __VA_ARGS__)
+#define planWarn(param, ...)   qWarn("PLAN: " param, __VA_ARGS__)
+#define planInfo(param, ...)   qInfo("PLAN: " param, __VA_ARGS__)
+#define planDebug(param, ...)  qDebug("PLAN: " param, __VA_ARGS__)
+#define planDebugL(param, ...) qDebugL("PLAN: " param, __VA_ARGS__)
+#define planTrace(param, ...)  qTrace("PLAN: " param, __VA_ARGS__)
 
 int32_t generateUsageErrMsg(char* pBuf, int32_t len, int32_t errCode, ...);
 

source/libs/planner/src/planLogicCreater.c

@@ -310,12 +310,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
 
 static int32_t createSubqueryLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable,
                                        SLogicNode** pLogicNode) {
-  int32_t code = createQueryLogicNode(pCxt, pTable->pSubquery, pLogicNode);
-  if (TSDB_CODE_SUCCESS == code) {
-    SNode* pNode;
-    FOREACH(pNode, (*pLogicNode)->pTargets) { strcpy(((SColumnNode*)pNode)->tableAlias, pTable->table.tableAlias); }
-  }
-  return code;
+  return createQueryLogicNode(pCxt, pTable->pSubquery, pLogicNode);
 }
 
 static int32_t createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SJoinTableNode* pJoinTable,
@@ -879,7 +874,8 @@ static int32_t createSetOpProjectLogicNode(SLogicPlanContext* pCxt, SSetOperator
   }
 
   if (TSDB_CODE_SUCCESS == code) {
-    code = createColumnByProjections(pCxt, NULL, pSetOperator->pProjectionList, &pProject->node.pTargets);
+    code = createColumnByProjections(pCxt, pSetOperator->stmtName, pSetOperator->pProjectionList,
+                                     &pProject->node.pTargets);
   }
 
   if (TSDB_CODE_SUCCESS == code) {
@@ -933,7 +929,7 @@ static int32_t createSetOpLogicNode(SLogicPlanContext* pCxt, SSetOperator* pSetO
       code = createSetOpAggLogicNode(pCxt, pSetOperator, &pSetOp);
       break;
     default:
-      code = -1;
+      code = TSDB_CODE_FAILED;
       break;
   }
 

source/libs/planner/src/planOptimizer.c

@@ -598,39 +598,63 @@ static bool cpdIsPrimaryKeyEqualCond(SJoinLogicNode* pJoin, SNode* pCond) {
   return false;
 }
 
-static int32_t cpdCheckOpCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SNode* pOnCond) {
-  if (!cpdIsPrimaryKeyEqualCond(pJoin, pOnCond)) {
-    return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-  }
-  return TSDB_CODE_SUCCESS;
-}
-
-static int32_t cpdCheckLogicCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SLogicConditionNode* pOnCond) {
-  if (LOGIC_COND_TYPE_AND != pOnCond->condType) {
-    return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-  }
-  bool   hasPrimaryKeyEqualCond = false;
-  SNode* pCond = NULL;
-  FOREACH(pCond, pOnCond->pParameterList) {
-    if (cpdIsPrimaryKeyEqualCond(pJoin, pCond)) {
-      hasPrimaryKeyEqualCond = true;
+static bool cpdContainPrimaryKeyEqualCond(SJoinLogicNode* pJoin, SNode* pCond) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pCond)) {
+    SLogicConditionNode* pLogicCond = (SLogicConditionNode*)pCond;
+    if (LOGIC_COND_TYPE_AND != pLogicCond->condType) {
+      return false;
     }
-  }
-  if (!hasPrimaryKeyEqualCond) {
-    return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-  }
-  return TSDB_CODE_SUCCESS;
-}
+    bool   hasPrimaryKeyEqualCond = false;
+    SNode* pCond = NULL;
+    FOREACH(pCond, pLogicCond->pParameterList) {
+      if (cpdContainPrimaryKeyEqualCond(pJoin, pCond)) {
+        hasPrimaryKeyEqualCond = true;
+        break;
+      }
+    }
+    return hasPrimaryKeyEqualCond;
+  } else {
+    return cpdIsPrimaryKeyEqualCond(pJoin, pCond);
+  }
+}
+
+// static int32_t cpdCheckOpCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SNode* pOnCond) {
+//   if (!cpdIsPrimaryKeyEqualCond(pJoin, pOnCond)) {
+//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
+//   }
+//   return TSDB_CODE_SUCCESS;
+// }
+
+// static int32_t cpdCheckLogicCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SLogicConditionNode* pOnCond) {
+//   if (LOGIC_COND_TYPE_AND != pOnCond->condType) {
+//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
+//   }
+//   bool   hasPrimaryKeyEqualCond = false;
+//   SNode* pCond = NULL;
+//   FOREACH(pCond, pOnCond->pParameterList) {
+//     if (cpdIsPrimaryKeyEqualCond(pJoin, pCond)) {
+//       hasPrimaryKeyEqualCond = true;
+//     }
+//   }
+//   if (!hasPrimaryKeyEqualCond) {
+//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
+//   }
+//   return TSDB_CODE_SUCCESS;
+// }
 
 static int32_t cpdCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {
   if (NULL == pJoin->pOnConditions) {
     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_NOT_SUPPORT_CROSS_JOIN);
   }
-  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pJoin->pOnConditions)) {
-    return cpdCheckLogicCond(pCxt, pJoin, (SLogicConditionNode*)pJoin->pOnConditions);
-  } else {
-    return cpdCheckOpCond(pCxt, pJoin, pJoin->pOnConditions);
+  if (!cpdContainPrimaryKeyEqualCond(pJoin, pJoin->pOnConditions)) {
+    return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
   }
+  return TSDB_CODE_SUCCESS;
+  // if (QUERY_NODE_LOGIC_CONDITION == nodeType(pJoin->pOnConditions)) {
+  //   return cpdCheckLogicCond(pCxt, pJoin, (SLogicConditionNode*)pJoin->pOnConditions);
+  // } else {
+  //   return cpdCheckOpCond(pCxt, pJoin, pJoin->pOnConditions);
+  // }
 }
 
 static int32_t cpdPushJoinCondition(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {

source/libs/planner/src/planSpliter.c

@@ -204,6 +204,75 @@ static int32_t ctjSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
   return code;
 }
 
+static bool unionIsChildSubplan(SLogicNode* pLogicNode, int32_t groupId) {
+  if (QUERY_NODE_LOGIC_PLAN_EXCHANGE == nodeType(pLogicNode)) {
+    return ((SExchangeLogicNode*)pLogicNode)->srcGroupId == groupId;
+  }
+
+  SNode* pChild;
+  FOREACH(pChild, pLogicNode->pChildren) {
+    bool isChild = unionIsChildSubplan((SLogicNode*)pChild, groupId);
+    if (isChild) {
+      return isChild;
+    }
+  }
+  return false;
+}
+
+static int32_t unionMountSubplan(SLogicSubplan* pParent, SNodeList* pChildren) {
+  SNode* pChild = NULL;
+  WHERE_EACH(pChild, pChildren) {
+    if (unionIsChildSubplan(pParent->pNode, ((SLogicSubplan*)pChild)->id.groupId)) {
+      int32_t code = nodesListMakeAppend(&pParent->pChildren, pChild);
+      if (TSDB_CODE_SUCCESS == code) {
+        REPLACE_NODE(NULL);
+        ERASE_NODE(pChildren);
+        continue;
+      } else {
+        return code;
+      }
+    }
+    WHERE_NEXT;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static SLogicSubplan* unionCreateSubplan(SSplitContext* pCxt, SLogicNode* pNode) {
+  SLogicSubplan* pSubplan = nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
+  if (NULL == pSubplan) {
+    return NULL;
+  }
+  pSubplan->id.groupId = pCxt->groupId;
+  pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
+  pSubplan->pNode = pNode;
+  return pSubplan;
+}
+
+static int32_t unionSplitSubplan(SSplitContext* pCxt, SLogicSubplan* pUnionSubplan, SLogicNode* pSplitNode) {
+  SNodeList* pSubplanChildren = pUnionSubplan->pChildren;
+  pUnionSubplan->pChildren = NULL;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+
+  SNode* pChild = NULL;
+  FOREACH(pChild, pSplitNode->pChildren) {
+    SLogicSubplan* pNewSubplan = unionCreateSubplan(pCxt, (SLogicNode*)pChild);
+    code = nodesListMakeStrictAppend(&pUnionSubplan->pChildren, pNewSubplan);
+    if (TSDB_CODE_SUCCESS == code) {
+      REPLACE_NODE(NULL);
+      code = unionMountSubplan(pNewSubplan, pSubplanChildren);
+    }
+    if (TSDB_CODE_SUCCESS != code) {
+      break;
+    }
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    nodesDestroyList(pSubplanChildren);
+    DESTORY_LIST(pSplitNode->pChildren);
+  }
+  return code;
+}
+
 static SLogicNode* uaMatchByNode(SLogicNode* pNode) {
   if (QUERY_NODE_LOGIC_PLAN_PROJECT == nodeType(pNode) && LIST_LENGTH(pNode->pChildren) > 1) {
     return pNode;
@@ -227,17 +296,6 @@ static bool uaFindSplitNode(SLogicSubplan* pSubplan, SUaInfo* pInfo) {
   return NULL != pSplitNode;
 }
 
-static SLogicSubplan* uaCreateSubplan(SSplitContext* pCxt, SLogicNode* pNode) {
-  SLogicSubplan* pSubplan = nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
-  if (NULL == pSubplan) {
-    return NULL;
-  }
-  pSubplan->id.groupId = pCxt->groupId;
-  pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
-  pSubplan->pNode = pNode;
-  return pSubplan;
-}
-
 static int32_t uaCreateExchangeNode(SSplitContext* pCxt, SLogicSubplan* pSubplan, SProjectLogicNode* pProject) {
   SExchangeLogicNode* pExchange = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_EXCHANGE);
   if (NULL == pExchange) {
@@ -276,20 +334,8 @@ static int32_t uaSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t code = TSDB_CODE_SUCCESS;
-
-  SNode* pChild = NULL;
-  FOREACH(pChild, info.pProject->node.pChildren) {
-    code = nodesListMakeStrictAppend(&info.pSubplan->pChildren, uaCreateSubplan(pCxt, (SLogicNode*)pChild));
-    if (TSDB_CODE_SUCCESS == code) {
-      REPLACE_NODE(NULL);
-    } else {
-      break;
-    }
-  }
+  int32_t code = unionSplitSubplan(pCxt, info.pSubplan, (SLogicNode*)info.pProject);
   if (TSDB_CODE_SUCCESS == code) {
-    nodesClearList(info.pProject->node.pChildren);
-    info.pProject->node.pChildren = NULL;
     code = uaCreateExchangeNode(pCxt, info.pSubplan, info.pProject);
   }
   ++(pCxt->groupId);
@@ -343,20 +389,8 @@ static int32_t unSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t code = TSDB_CODE_SUCCESS;
-
-  SNode* pChild = NULL;
-  FOREACH(pChild, info.pAgg->node.pChildren) {
-    code = nodesListMakeStrictAppend(&info.pSubplan->pChildren, uaCreateSubplan(pCxt, (SLogicNode*)pChild));
-    if (TSDB_CODE_SUCCESS == code) {
-      REPLACE_NODE(NULL);
-    } else {
-      break;
-    }
-  }
+  int32_t code = unionSplitSubplan(pCxt, info.pSubplan, (SLogicNode*)info.pAgg);
   if (TSDB_CODE_SUCCESS == code) {
-    nodesClearList(info.pAgg->node.pChildren);
-    info.pAgg->node.pChildren = NULL;
     code = unCreateExchangeNode(pCxt, info.pSubplan, info.pAgg);
   }
   ++(pCxt->groupId);

source/libs/planner/test/planJoinTest.cpp

@@ -30,6 +30,14 @@ TEST_F(PlanJoinTest, basic) {
   run("SELECT t1.c1, t2.c1 FROM st1s1 t1 JOIN st1s2 t2 ON t1.ts = t2.ts");
 }
 
+TEST_F(PlanJoinTest, complex) {
+  useDb("root", "test");
+
+  run("SELECT t1.c1, t2.c2 FROM st1s1 t1, st1s2 t2 "
+      "WHERE t1.ts = t2.ts AND t1.c1 BETWEEN -10 AND 10 AND t2.c1 BETWEEN -100 AND 100 AND "
+      "(t1.c2 LIKE 'nchar%' OR t1.c1 = 0 OR t2.c2 LIKE 'nchar%' OR t2.c1 = 0)");
+}
+
 TEST_F(PlanJoinTest, withWhere) {
   useDb("root", "test");
 

source/libs/planner/test/planSetOpTest.cpp

@@ -23,11 +23,33 @@ class PlanSetOpTest : public PlannerTestBase {};
 TEST_F(PlanSetOpTest, unionAll) {
   useDb("root", "test");
 
-  run("select c1, c2 from t1 where c1 > 10 union all select c1, c2 from t1 where c1 > 20");
+  run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION ALL SELECT c1, c2 FROM t1 WHERE c1 > 20");
+}
+
+TEST_F(PlanSetOpTest, unionAllSubquery) {
+  useDb("root", "test");
+
+  run("SELECT * FROM (SELECT c1, c2 FROM t1 UNION ALL SELECT c1, c2 FROM t1)");
 }
 
 TEST_F(PlanSetOpTest, union) {
   useDb("root", "test");
 
-  run("select c1, c2 from t1 where c1 > 10 union select c1, c2 from t1 where c1 > 20");
+  run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION SELECT c1, c2 FROM t1 WHERE c1 > 20");
+}
+
+TEST_F(PlanSetOpTest, unionContainJoin) {
+  useDb("root", "test");
+
+  run("SELECT t1.c1 FROM st1s1 t1 join st1s2 t2 on t1.ts = t2.ts "
+      "WHERE t1.c1 IS NOT NULL GROUP BY t1.c1 HAVING t1.c1 IS NOT NULL "
+      "UNION "
+      "SELECT t1.c1 FROM st1s1 t1 join st1s2 t2 on t1.ts = t2.ts "
+      "WHERE t1.c1 IS NOT NULL GROUP BY t1.c1 HAVING t1.c1 IS NOT NULL");
+}
+
+TEST_F(PlanSetOpTest, unionSubquery) {
+  useDb("root", "test");
+
+  run("SELECT * FROM (SELECT c1, c2 FROM t1 UNION SELECT c1, c2 FROM t1)");
 }