Merge pull request #21222 from taosdata/fix/TD-24001

enh: optimize timeline function validation

include/common/tcommon.h

@@ -231,6 +231,7 @@ typedef struct SColumnInfoData {
   };
   SColumnInfo info;     // column info
   bool        hasNull;  // if current column data has null value.
+  bool        reassigned; // if current column data is reassigned.
 } SColumnInfoData;
 
 typedef struct SQueryTableDataCond {

include/common/tdatablock.h

@@ -178,6 +178,7 @@ int32_t getJsonValueLen(const char* data);
 
 int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
+int32_t colDataReassignVal(SColumnInfoData* pColumnInfoData, uint32_t dstRowIdx, uint32_t srcRowIdx, const char* pData);
 int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows, bool trimValue);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
                         const SColumnInfoData* pSource, int32_t numOfRow2);

include/libs/nodes/nodes.h

@@ -328,6 +328,8 @@ void       nodesListInsertList(SNodeList* pTarget, SListCell* pPos, SNodeList* p
 SNode*     nodesListGetNode(SNodeList* pList, int32_t index);
 SListCell* nodesListGetCell(SNodeList* pList, int32_t index);
 void       nodesDestroyList(SNodeList* pList);
+bool       nodesListMatch(const SNodeList* pList, const SNodeList* pSubList);
+
 // Only clear the linked list structure, without releasing the elements inside
 void nodesClearList(SNodeList* pList);
 
@@ -346,6 +348,7 @@ void nodesRewriteExprPostOrder(SNode** pNode, FNodeRewriter rewriter, void* pCon
 void nodesRewriteExprsPostOrder(SNodeList* pList, FNodeRewriter rewriter, void* pContext);
 
 bool nodesEqualNode(const SNode* a, const SNode* b);
+bool nodesMatchNode(const SNode* pSub, const SNode* pNode);
 
 SNode*     nodesCloneNode(const SNode* pNode);
 SNodeList* nodesCloneList(const SNodeList* pList);

include/libs/nodes/querynodes.h

@@ -241,6 +241,12 @@ typedef enum EFillMode {
   FILL_MODE_NEXT
 } EFillMode;
 
+typedef enum ETimeLineMode {
+  TIME_LINE_NONE = 1,
+  TIME_LINE_MULTI,
+  TIME_LINE_GLOBAL,
+} ETimeLineMode;
+
 typedef struct SFillNode {
   ENodeType   type;  // QUERY_NODE_FILL
   EFillMode   mode;
@@ -263,50 +269,50 @@ typedef struct SCaseWhenNode {
 } SCaseWhenNode;
 
 typedef struct SSelectStmt {
-  ENodeType   type;  // QUERY_NODE_SELECT_STMT
-  bool        isDistinct;
-  SNodeList*  pProjectionList;
-  SNode*      pFromTable;
-  SNode*      pWhere;
-  SNodeList*  pPartitionByList;
-  SNodeList*  pTags;      // for create stream
-  SNode*      pSubtable;  // for create stream
-  SNode*      pWindow;
-  SNodeList*  pGroupByList;  // SGroupingSetNode
-  SNode*      pHaving;
-  SNode*      pRange;
-  SNode*      pEvery;
-  SNode*      pFill;
-  SNodeList*  pOrderByList;  // SOrderByExprNode
-  SLimitNode* pLimit;
-  SLimitNode* pSlimit;
-  STimeWindow timeRange;
-  char        stmtName[TSDB_TABLE_NAME_LEN];
-  uint8_t     precision;
-  int32_t     selectFuncNum;
-  int32_t     returnRows;  // EFuncReturnRows
-  bool        isEmptyResult;
-  bool        isTimeLineResult;
-  bool        isSubquery;
-  bool        hasAggFuncs;
-  bool        hasRepeatScanFuncs;
-  bool        hasIndefiniteRowsFunc;
-  bool        hasMultiRowsFunc;
-  bool        hasSelectFunc;
-  bool        hasSelectValFunc;
-  bool        hasOtherVectorFunc;
-  bool        hasUniqueFunc;
-  bool        hasTailFunc;
-  bool        hasInterpFunc;
-  bool        hasInterpPseudoColFunc;
-  bool        hasLastRowFunc;
-  bool        hasLastFunc;
-  bool        hasTimeLineFunc;
-  bool        hasUdaf;
-  bool        hasStateKey;
-  bool        onlyHasKeepOrderFunc;
-  bool        groupSort;
-  bool        tagScan;
+  ENodeType     type;  // QUERY_NODE_SELECT_STMT
+  bool          isDistinct;
+  SNodeList*    pProjectionList;
+  SNode*        pFromTable;
+  SNode*        pWhere;
+  SNodeList*    pPartitionByList;
+  SNodeList*    pTags;      // for create stream
+  SNode*        pSubtable;  // for create stream
+  SNode*        pWindow;
+  SNodeList*    pGroupByList;  // SGroupingSetNode
+  SNode*        pHaving;
+  SNode*        pRange;
+  SNode*        pEvery;
+  SNode*        pFill;
+  SNodeList*    pOrderByList;  // SOrderByExprNode
+  SLimitNode*   pLimit;
+  SLimitNode*   pSlimit;
+  STimeWindow   timeRange;
+  char          stmtName[TSDB_TABLE_NAME_LEN];
+  uint8_t       precision;
+  int32_t       selectFuncNum;
+  int32_t       returnRows;  // EFuncReturnRows
+  ETimeLineMode timeLineResMode;
+  bool          isEmptyResult;
+  bool          isSubquery;
+  bool          hasAggFuncs;
+  bool          hasRepeatScanFuncs;
+  bool          hasIndefiniteRowsFunc;
+  bool          hasMultiRowsFunc;
+  bool          hasSelectFunc;
+  bool          hasSelectValFunc;
+  bool          hasOtherVectorFunc;
+  bool          hasUniqueFunc;
+  bool          hasTailFunc;
+  bool          hasInterpFunc;
+  bool          hasInterpPseudoColFunc;
+  bool          hasLastRowFunc;
+  bool          hasLastFunc;
+  bool          hasTimeLineFunc;
+  bool          hasUdaf;
+  bool          hasStateKey;
+  bool          onlyHasKeepOrderFunc;
+  bool          groupSort;
+  bool          tagScan;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;
@@ -321,6 +327,7 @@ typedef struct SSetOperator {
   SNode*           pLimit;
   char             stmtName[TSDB_TABLE_NAME_LEN];
   uint8_t          precision;
+  ETimeLineMode    timeLineResMode;  
 } SSetOperator;
 
 typedef enum ESqlClause {

source/common/src/tdatablock.c

@@ -23,6 +23,20 @@
 
 int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows) {
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    if (pColumnInfoData->reassigned) {
+      int32_t totalSize = 0;
+      for (int32_t row = 0; row < numOfRows; ++row) {
+        char* pColData = pColumnInfoData->pData + pColumnInfoData->varmeta.offset[row];
+        int32_t colSize = 0;
+        if (pColumnInfoData->info.type == TSDB_DATA_TYPE_JSON) {
+          colSize = getJsonValueLen(pColData);
+        } else {
+          colSize = varDataTLen(pColData);
+        }
+        totalSize += colSize;
+      }
+      return totalSize;
+    }
     return pColumnInfoData->varmeta.length;
   } else {
     if (pColumnInfoData->info.type == TSDB_DATA_TYPE_NULL) {
@@ -126,6 +140,29 @@ int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const
   return 0;
 }
 
+int32_t colDataReassignVal(SColumnInfoData* pColumnInfoData, uint32_t dstRowIdx, uint32_t srcRowIdx, const char* pData) {
+  int32_t type = pColumnInfoData->info.type;
+  if (IS_VAR_DATA_TYPE(type)) {
+    int32_t dataLen = 0;
+    if (type == TSDB_DATA_TYPE_JSON) {
+      dataLen = getJsonValueLen(pData);
+    } else {
+      dataLen = varDataTLen(pData);
+    }
+
+    SVarColAttr* pAttr = &pColumnInfoData->varmeta;
+
+    pColumnInfoData->varmeta.offset[dstRowIdx] = pColumnInfoData->varmeta.offset[srcRowIdx];
+    pColumnInfoData->reassigned = true;
+  } else {
+    memcpy(pColumnInfoData->pData + pColumnInfoData->info.bytes * dstRowIdx, pData, pColumnInfoData->info.bytes);
+    colDataClearNull_f(pColumnInfoData->nullbitmap, dstRowIdx);
+  }
+
+  return 0;
+}
+
+
 int32_t colDataReserve(SColumnInfoData* pColumnInfoData, size_t newSize) {
   if (!IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
     return TSDB_CODE_SUCCESS;
@@ -580,8 +617,22 @@ int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock) {
     *(int32_t*)pStart = dataSize;
     pStart += sizeof(int32_t);
 
-    memcpy(pStart, pCol->pData, dataSize);
-    pStart += dataSize;
+    if (pCol->reassigned && IS_VAR_DATA_TYPE(pCol->info.type)) {
+      for (int32_t row = 0; row < numOfRows; ++row) {
+        char* pColData = pCol->pData + pCol->varmeta.offset[row];
+        int32_t colSize = 0;
+        if (pCol->info.type == TSDB_DATA_TYPE_JSON) {
+          colSize = getJsonValueLen(pColData);
+        } else {
+          colSize = varDataTLen(pColData);
+        }
+        memcpy(pStart, pColData, colSize);
+        pStart += colSize;
+      }
+    } else {
+      memcpy(pStart, pCol->pData, dataSize);
+      pStart += dataSize;
+    }    
   }
 
   return 0;
@@ -1741,7 +1792,20 @@ int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
     int32_t len = colDataGetLength(pColData, rows);
     tlen += taosEncodeFixedI32(buf, len);
 
-    tlen += taosEncodeBinary(buf, pColData->pData, len);
+    if (pColData->reassigned && IS_VAR_DATA_TYPE(pColData->info.type)) {
+      for (int32_t row = 0; row < rows; ++row) {
+        char* pData = pColData->pData + pColData->varmeta.offset[row];
+        int32_t colSize = 0;
+        if (pColData->info.type == TSDB_DATA_TYPE_JSON) {
+          colSize = getJsonValueLen(pData);
+        } else {
+          colSize = varDataTLen(pData);
+        }
+        tlen += taosEncodeBinary(buf, pData, colSize);
+      }
+    } else {
+      tlen += taosEncodeBinary(buf, pColData->pData, len);
+    }
   }
   return tlen;
 }
@@ -2502,12 +2566,29 @@ int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
     data += metaSize;
     dataLen += metaSize;
 
-    colSizes[col] = colDataGetLength(pColRes, numOfRows);
-    dataLen += colSizes[col];
-    if (pColRes->pData != NULL) {
-      memmove(data, pColRes->pData, colSizes[col]);
+     if (pColRes->reassigned && IS_VAR_DATA_TYPE(pColRes->info.type)) {
+        colSizes[col] = 0;
+        for (int32_t row = 0; row < numOfRows; ++row) {
+          char* pColData = pColRes->pData + pColRes->varmeta.offset[row];
+          int32_t colSize = 0;
+          if (pColRes->info.type == TSDB_DATA_TYPE_JSON) {
+            colSize = getJsonValueLen(pColData);
+          } else {
+            colSize = varDataTLen(pColData);
+          }
+          colSizes[col] += colSize;
+          dataLen += colSize;
+          memmove(data, pColData, colSize);
+          data += colSize;
+        }
+    } else {
+      colSizes[col] = colDataGetLength(pColRes, numOfRows);
+      dataLen += colSizes[col];
+      if (pColRes->pData != NULL) {
+        memmove(data, pColRes->pData, colSizes[col]);
+      }
+      data += colSizes[col];
     }
-    data += colSizes[col];
 
     colSizes[col] = htonl(colSizes[col]);
 //    uError("blockEncode col bytes:%d, type:%d, size:%d, htonl size:%d", pColRes->info.bytes, pColRes->info.type, htonl(colSizes[col]), colSizes[col]);

source/libs/executor/src/executorInt.c

@@ -562,7 +562,6 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoD
       int32_t numOfRows = 0;
       if (IS_VAR_DATA_TYPE(pDst->info.type)) {
         int32_t j = 0;
-        pDst->varmeta.length = 0;
 
         while (j < totalRows) {
           if (pIndicator[j] == 0) {
@@ -573,18 +572,8 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoD
           if (colDataIsNull_var(pDst, j)) {
             colDataSetNull_var(pDst, numOfRows);
           } else {
-            // fix address sanitizer error. p1 may point to memory that will change during realloc of colDataSetVal, first copy it to p2
             char* p1 = colDataGetVarData(pDst, j);
-            int32_t len = 0;
-            if (pDst->info.type == TSDB_DATA_TYPE_JSON) {
-              len = getJsonValueLen(p1);
-            } else {
-              len = varDataTLen(p1);
-            }
-            char* p2 = taosMemoryMalloc(len);
-            memcpy(p2, p1, len);
-            colDataSetVal(pDst, numOfRows, p2, false);
-            taosMemoryFree(p2);
+            colDataReassignVal(pDst, numOfRows, j, p1);
           }
           numOfRows += 1;
           j += 1;

source/libs/executor/src/sortoperator.c

@@ -153,6 +153,7 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
       colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
     }
 
+    pDataBlock->info.dataLoad = 1;
     pDataBlock->info.rows = p->info.rows;
   }
 

source/libs/function/src/tudf.c

@@ -791,7 +791,21 @@ int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlo
       memcpy(udfCol->colData.varLenCol.varOffsets, col->varmeta.offset, udfCol->colData.varLenCol.varOffsetsLen);
       udfCol->colData.varLenCol.payloadLen = colDataGetLength(col, udfBlock->numOfRows);
       udfCol->colData.varLenCol.payload = taosMemoryMalloc(udfCol->colData.varLenCol.payloadLen);
-      memcpy(udfCol->colData.varLenCol.payload, col->pData, udfCol->colData.varLenCol.payloadLen);
+      if (col->reassigned) {
+        for (int32_t row = 0; row < udfCol->colData.numOfRows; ++row) {
+          char* pColData = col->pData + col->varmeta.offset[row];
+          int32_t colSize = 0;
+          if (col->info.type == TSDB_DATA_TYPE_JSON) {
+            colSize = getJsonValueLen(pColData);
+          } else {
+            colSize = varDataTLen(pColData);
+          }
+          memcpy(udfCol->colData.varLenCol.payload, pColData, colSize);
+          udfCol->colData.varLenCol.payload += colSize;
+        }
+      } else {
+        memcpy(udfCol->colData.varLenCol.payload, col->pData, udfCol->colData.varLenCol.payloadLen);
+      }
     } else {
       udfCol->colData.fixLenCol.nullBitmapLen = BitmapLen(udfCol->colData.numOfRows);
       int32_t bitmapLen = udfCol->colData.fixLenCol.nullBitmapLen;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -669,7 +669,7 @@ static int32_t selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
   COPY_CHAR_ARRAY_FIELD(stmtName);
   COPY_SCALAR_FIELD(precision);
   COPY_SCALAR_FIELD(isEmptyResult);
-  COPY_SCALAR_FIELD(isTimeLineResult);
+  COPY_SCALAR_FIELD(timeLineResMode);
   COPY_SCALAR_FIELD(hasAggFuncs);
   COPY_SCALAR_FIELD(hasRepeatScanFuncs);
   return TSDB_CODE_SUCCESS;

source/libs/nodes/src/nodesMatchFuncs.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "querynodes.h"
+
+#define MATCH_SCALAR_FIELD(fldname)           \
+  do {                                          \
+    if (p->fldname != pSub->fldname) return false; \
+  } while (0)
+
+#define MATCH_STRING(a, b) (((a) != NULL && (b) != NULL) ? (strcmp((a), (b)) == 0) : (a) == (b))
+
+#define MATCH_VARDATA(a, b)                                                                                   \
+  (((a) != NULL && (b) != NULL)                                                                                 \
+       ? (varDataLen((a)) == varDataLen((b)) && memcmp(varDataVal((a)), varDataVal((b)), varDataLen((a))) == 0) \
+       : (a) == (b))
+
+#define MATCH_STRING_FIELD(fldname)                          \
+  do {                                                         \
+    if (!MATCH_STRING(p->fldname, pSub->fldname)) return false; \
+  } while (0)
+
+#define MATCH_VARDATA_FIELD(fldname)                          \
+  do {                                                          \
+    if (!MATCH_VARDATA(p->fldname, pSub->fldname)) return false; \
+  } while (0)
+
+#define MATCH_OBJECT_FIELD(fldname, matchFunc)          \
+  do {                                                    \
+    if (!matchFunc(p->fldname, pSub->fldname)) return false; \
+  } while (0)
+
+#define MATCH_NODE_FIELD(fldname)                            \
+  do {                                                         \
+    if (!nodesMatchNode(pSub->fldname, p->fldname)) return false; \
+  } while (0)
+
+#define MATCH_NODE_LIST_FIELD(fldname)                          \
+  do {                                                            \
+    if (!nodesListMatch(p->fldname, pSub->fldname)) return false; \
+  } while (0)
+
+    
+bool nodesListMatchExists(const SNodeList* pList, const SNode* pTarget) {
+  if (NULL == pList || NULL == pTarget) {
+    return false;
+  }
+  SNode* node = NULL;
+  bool exists = false;
+  FOREACH(node, pList) {
+    if (nodesMatchNode(node, pTarget)) {
+      exists = true;
+      break;
+    }
+  }
+
+  return exists;
+}
+
+bool nodesListMatch(const SNodeList* pList, const SNodeList* pSubList) {
+  if (pList == pSubList) {
+    return true;
+  }
+
+  if (NULL == pList || NULL == pSubList) {
+    return false;
+  }
+
+  if (pList->length != pSubList->length) {
+    return false;
+  }
+
+  SNode* node = NULL;
+  bool match = true;
+  FOREACH(node, pList) {
+    if (!nodesListMatchExists(pSubList, node)) {
+      match = false;
+      break;
+    }
+  }
+  return match;
+}
+
+static bool columnNodeMatch(const SColumnNode* pSub, const SColumnNode* p) {
+  if (0 == strcmp(p->colName, pSub->node.aliasName)) {
+    return true;
+  }
+  return false;
+}
+
+static bool valueNodeMatch(const SValueNode* pSub, const SValueNode* p) {
+  return nodesEqualNode((SNode*)pSub, (SNode*)p);
+}
+
+static bool operatorNodeMatch(const SOperatorNode* pSub, const SOperatorNode* p) {
+  MATCH_SCALAR_FIELD(opType);
+  MATCH_NODE_FIELD(pLeft);
+  MATCH_NODE_FIELD(pRight);
+  return true;
+}
+
+static bool logicConditionNodeMatch(const SLogicConditionNode* pSub, const SLogicConditionNode* p) {
+  MATCH_SCALAR_FIELD(condType);
+  MATCH_NODE_LIST_FIELD(pParameterList);
+  return true;
+}
+
+static bool functionNodeMatch(const SFunctionNode* pSub, const SFunctionNode* p) {
+  MATCH_SCALAR_FIELD(funcId);
+  MATCH_STRING_FIELD(functionName);
+  MATCH_NODE_LIST_FIELD(pParameterList);
+  return true;
+}
+
+static bool whenThenNodeMatch(const SWhenThenNode* pSub, const SWhenThenNode* p) {
+  MATCH_NODE_FIELD(pWhen);
+  MATCH_NODE_FIELD(pThen);
+  return true;
+}
+
+static bool caseWhenNodeMatch(const SCaseWhenNode* pSub, const SCaseWhenNode* p) {
+  MATCH_NODE_FIELD(pCase);
+  MATCH_NODE_FIELD(pElse);
+  MATCH_NODE_LIST_FIELD(pWhenThenList);
+  return true;
+}
+
+bool nodesMatchNode(const SNode* pSub, const SNode* p) {
+  if (pSub == p) {
+    return true;
+  }
+
+  if (NULL == pSub || NULL == p) {
+    return false;
+  }
+
+  if (nodeType(pSub) != nodeType(p)) {
+    return false;
+  }
+
+  switch (nodeType(p)) {
+    case QUERY_NODE_COLUMN:
+      return columnNodeMatch((const SColumnNode*)pSub, (const SColumnNode*)p);
+    case QUERY_NODE_VALUE:
+      return valueNodeMatch((const SValueNode*)pSub, (const SValueNode*)p);
+    case QUERY_NODE_OPERATOR:
+      return operatorNodeMatch((const SOperatorNode*)pSub, (const SOperatorNode*)p);
+    case QUERY_NODE_LOGIC_CONDITION:
+      return logicConditionNodeMatch((const SLogicConditionNode*)pSub, (const SLogicConditionNode*)p);
+    case QUERY_NODE_FUNCTION:
+      return functionNodeMatch((const SFunctionNode*)pSub, (const SFunctionNode*)p);
+    case QUERY_NODE_WHEN_THEN:
+      return whenThenNodeMatch((const SWhenThenNode*)pSub, (const SWhenThenNode*)p);
+    case QUERY_NODE_CASE_WHEN:
+      return caseWhenNodeMatch((const SCaseWhenNode*)pSub, (const SCaseWhenNode*)p);
+    case QUERY_NODE_REAL_TABLE:
+    case QUERY_NODE_TEMP_TABLE:
+    case QUERY_NODE_JOIN_TABLE:
+    case QUERY_NODE_GROUPING_SET:
+    case QUERY_NODE_ORDER_BY_EXPR:
+    case QUERY_NODE_LIMIT:
+      return false;
+    default:
+      break;
+  }
+
+  return false;
+}

source/libs/parser/inc/parUtil.h

@@ -84,6 +84,7 @@ int32_t       getNumOfColumns(const STableMeta* pTableMeta);
 int32_t       getNumOfTags(const STableMeta* pTableMeta);
 STableComInfo getTableInfo(const STableMeta* pTableMeta);
 STableMeta*   tableMetaDup(const STableMeta* pTableMeta);
+int32_t getTableTypeFromTableNode(SNode *pTable);
 
 int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen);
 int32_t getVnodeSysTableTargetName(int32_t acctId, SNode* pWhere, SName* pName);

source/libs/parser/src/parTranslater.c

@@ -758,18 +758,40 @@ static SNodeList* getProjectList(const SNode* pNode) {
 
 static bool isTimeLineQuery(SNode* pStmt) {
   if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
-    return ((SSelectStmt*)pStmt)->isTimeLineResult;
+    return (TIME_LINE_MULTI == ((SSelectStmt*)pStmt)->timeLineResMode) || (TIME_LINE_GLOBAL == ((SSelectStmt*)pStmt)->timeLineResMode);
+  } else if (QUERY_NODE_SET_OPERATOR == nodeType(pStmt)) {
+    return TIME_LINE_GLOBAL == ((SSetOperator*)pStmt)->timeLineResMode;
   } else {
     return false;
   }
 }
 
 static bool isGlobalTimeLineQuery(SNode* pStmt) {
-  if (!isTimeLineQuery(pStmt)) {
+  if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
+    return TIME_LINE_GLOBAL == ((SSelectStmt*)pStmt)->timeLineResMode;
+  } else if (QUERY_NODE_SET_OPERATOR == nodeType(pStmt)) {
+    return TIME_LINE_GLOBAL == ((SSetOperator*)pStmt)->timeLineResMode;
+  } else {
     return false;
   }
-  SSelectStmt* pSelect = (SSelectStmt*)pStmt;
-  return NULL == pSelect->pPartitionByList || NULL != pSelect->pOrderByList;
+}
+
+static bool isTimeLineAlignedQuery(SNode* pStmt) {
+  SSelectStmt *pSelect = (SSelectStmt *)pStmt;
+  if (isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
+    return true;
+  }
+  if (!isTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
+    return false;
+  }
+  if (QUERY_NODE_SELECT_STMT != nodeType(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
+    return false;
+  }
+  SSelectStmt *pSub = (SSelectStmt *)((STempTableNode*)pSelect->pFromTable)->pSubquery;
+  if (nodesListMatch(pSelect->pPartitionByList, pSub->pPartitionByList)) {
+    return true;
+  }
+  return false;
 }
 
 static bool isPrimaryKeyImpl(SNode* pExpr) {
@@ -1575,7 +1597,7 @@ static int32_t translateTimelineFunc(STranslateContext* pCxt, SFunctionNode* pFu
   }
   SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt;
   if (NULL != pSelect->pFromTable && QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) &&
-      !isTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery)) {
+      !isGlobalTimeLineQuery(((STempTableNode*)pSelect->pFromTable)->pSubquery) && !isTimeLineAlignedQuery(pCxt->pCurrStmt)) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_ALLOWED_FUNC,
                                    "%s function requires valid time series input", pFunc->functionName);
   }
@@ -2289,7 +2311,7 @@ static int32_t checkAggColCoexist(STranslateContext* pCxt, SSelectStmt* pSelect)
     return TSDB_CODE_SUCCESS;
   }
   if (!pSelect->onlyHasKeepOrderFunc) {
-    pSelect->isTimeLineResult = false;
+    pSelect->timeLineResMode = TIME_LINE_NONE;
   }
   CheckAggColCoexistCxt cxt = {.pTranslateCxt = pCxt, .existCol = false};
   nodesRewriteExprs(pSelect->pProjectionList, doCheckAggColCoexist, &cxt);
@@ -2636,9 +2658,9 @@ static int32_t replaceTbName(STranslateContext* pCxt, SSelectStmt* pSelect) {
 
 static int32_t checkJoinTable(STranslateContext* pCxt, SJoinTableNode* pJoinTable) {
   if ((QUERY_NODE_TEMP_TABLE == nodeType(pJoinTable->pLeft) &&
-       !isTimeLineQuery(((STempTableNode*)pJoinTable->pLeft)->pSubquery)) ||
+       !isGlobalTimeLineQuery(((STempTableNode*)pJoinTable->pLeft)->pSubquery)) ||
       (QUERY_NODE_TEMP_TABLE == nodeType(pJoinTable->pRight) &&
-       !isTimeLineQuery(((STempTableNode*)pJoinTable->pRight)->pSubquery))) {
+       !isGlobalTimeLineQuery(((STempTableNode*)pJoinTable->pRight)->pSubquery))) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_SUPPORT_JOIN,
                                    "Join requires valid time series input");
   }
@@ -2673,7 +2695,7 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
         }
         if (TSDB_SYSTEM_TABLE == pRealTable->pMeta->tableType) {
           if (isSelectStmt(pCxt->pCurrStmt)) {
-            ((SSelectStmt*)pCxt->pCurrStmt)->isTimeLineResult = false;
+            ((SSelectStmt*)pCxt->pCurrStmt)->timeLineResMode = TIME_LINE_NONE;
           } else if (isDeleteStmt(pCxt->pCurrStmt)) {
             code = TSDB_CODE_TSC_INVALID_OPERATION;
             break;
@@ -3152,7 +3174,7 @@ static int32_t translateGroupBy(STranslateContext* pCxt, SSelectStmt* pSelect) {
     return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_GROUPBY_WINDOW_COEXIST);
   }
   pCxt->currClause = SQL_CLAUSE_GROUP_BY;
-  pSelect->isTimeLineResult = false;
+  pSelect->timeLineResMode = TIME_LINE_NONE;
   return translateExprList(pCxt, pSelect->pGroupByList);
 }
 
@@ -3552,7 +3574,18 @@ static int32_t translateInterp(STranslateContext* pCxt, SSelectStmt* pSelect) {
 
 static int32_t translatePartitionBy(STranslateContext* pCxt, SSelectStmt* pSelect) {
   pCxt->currClause = SQL_CLAUSE_PARTITION_BY;
-  int32_t code = translateExprList(pCxt, pSelect->pPartitionByList);
+  int32_t code = TSDB_CODE_SUCCESS;
+  
+  if (pSelect->pPartitionByList) {
+    int8_t typeType = getTableTypeFromTableNode(pSelect->pFromTable);
+    SNode* pPar = nodesListGetNode(pSelect->pPartitionByList, 0);
+    if (!((TSDB_NORMAL_TABLE == typeType || TSDB_CHILD_TABLE == typeType) && 
+      1 == pSelect->pPartitionByList->length && (QUERY_NODE_FUNCTION == nodeType(pPar) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pPar)->funcType))) {
+      pSelect->timeLineResMode = TIME_LINE_MULTI;
+    }
+    
+    code = translateExprList(pCxt, pSelect->pPartitionByList);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = translateExprList(pCxt, pSelect->pTags);
   }
@@ -3681,9 +3714,9 @@ static void resetResultTimeline(SSelectStmt* pSelect) {
   if ((QUERY_NODE_TEMP_TABLE == nodeType(pSelect->pFromTable) &&
        isPrimaryKey((STempTableNode*)pSelect->pFromTable, pOrder)) ||
       (QUERY_NODE_TEMP_TABLE != nodeType(pSelect->pFromTable) && isPrimaryKeyImpl(pOrder))) {
-    pSelect->isTimeLineResult = true;
+    pSelect->timeLineResMode = TIME_LINE_GLOBAL;
   } else {
-    pSelect->isTimeLineResult = false;
+    pSelect->timeLineResMode = TIME_LINE_NONE;
   }
 }
 
@@ -3813,8 +3846,13 @@ static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pS
       pLeftExpr = pLeftFuncExpr;
     }
     snprintf(pRightExpr->aliasName, sizeof(pRightExpr->aliasName), "%s", pLeftExpr->aliasName);
-    if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList,
-                                                       createSetOperProject(pSetOperator->stmtName, pLeft))) {
+    SNode* pProj = createSetOperProject(pSetOperator->stmtName, pLeft);
+    if (QUERY_NODE_COLUMN == nodeType(pLeft) && QUERY_NODE_COLUMN == nodeType(pRight) 
+      && ((SColumnNode*)pLeft)->colId == PRIMARYKEY_TIMESTAMP_COL_ID
+      && ((SColumnNode*)pRight)->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
+      ((SColumnNode*)pProj)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
+    }
+    if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList, pProj)) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
   }
@@ -3826,6 +3864,10 @@ static uint8_t calcSetOperatorPrecision(SSetOperator* pSetOperator) {
 }
 
 static int32_t translateSetOperOrderBy(STranslateContext* pCxt, SSetOperator* pSetOperator) {
+  if (NULL == pSetOperator->pOrderByList || pSetOperator->pOrderByList->length <= 0) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   bool    other;
   int32_t code = translateOrderByPosition(pCxt, pSetOperator->pProjectionList, pSetOperator->pOrderByList, &other);
   if (TSDB_CODE_SUCCESS == code) {
@@ -3838,6 +3880,14 @@ static int32_t translateSetOperOrderBy(STranslateContext* pCxt, SSetOperator* pS
   if (TSDB_CODE_SUCCESS == code) {
     code = replaceOrderByAlias(pCxt, pSetOperator->pProjectionList, pSetOperator->pOrderByList);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    SNode* pOrder = ((SOrderByExprNode*)nodesListGetNode(pSetOperator->pOrderByList, 0))->pExpr;
+    if (isPrimaryKeyImpl(pOrder)) {
+      pSetOperator->timeLineResMode = TIME_LINE_GLOBAL;
+    } else {
+      pSetOperator->timeLineResMode = TIME_LINE_NONE;
+    }
+  }
   return code;
 }
 
@@ -6324,7 +6374,7 @@ static int32_t subtableExprHasColumnOrPseudoColumn(SNode* pNode) {
 static int32_t checkStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt) {
   SSelectStmt* pSelect = (SSelectStmt*)pStmt->pQuery;
   if (TSDB_DATA_TYPE_TIMESTAMP != ((SExprNode*)nodesListGetNode(pSelect->pProjectionList, 0))->resType.type ||
-      !pSelect->isTimeLineResult || crossTableWithoutAggOper(pSelect) || NULL != pSelect->pOrderByList ||
+      !isTimeLineQuery(pStmt->pQuery) || crossTableWithoutAggOper(pSelect) || NULL != pSelect->pOrderByList ||
       crossTableWithUdaf(pSelect) || isEventWindowQuery(pSelect) || hasJsonTypeProjection(pSelect)) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported stream query");
   }

source/libs/parser/src/parUtil.c

@@ -249,6 +249,17 @@ int32_t getNumOfTags(const STableMeta* pTableMeta) { return getTableInfo(pTableM
 
 STableComInfo getTableInfo(const STableMeta* pTableMeta) { return pTableMeta->tableInfo; }
 
+int32_t getTableTypeFromTableNode(SNode *pTable) {
+  if (NULL == pTable) {
+    return -1;
+  }
+  if (QUERY_NODE_REAL_TABLE != nodeType(pTable)) {
+    return -1;
+  }
+  return ((SRealTableNode *)pTable)->pMeta->tableType;
+}
+
+
 STableMeta* tableMetaDup(const STableMeta* pTableMeta) {
   int32_t numOfFields = TABLE_TOTAL_COL_NUM(pTableMeta);
   if (numOfFields > TSDB_MAX_COLUMNS || numOfFields < TSDB_MIN_COLUMNS) {
@@ -684,7 +695,7 @@ SNode* createSelectStmtImpl(bool isDistinct, SNodeList* pProjectionList, SNode*
   select->pProjectionList = pProjectionList;
   select->pFromTable = pTable;
   sprintf(select->stmtName, "%p", select);
-  select->isTimeLineResult = true;
+  select->timeLineResMode = select->isDistinct ? TIME_LINE_NONE : TIME_LINE_GLOBAL;
   select->onlyHasKeepOrderFunc = true;
   select->timeRange = TSWINDOW_INITIALIZER;
   return (SNode*)select;

source/libs/parser/test/parSelectTest.cpp

@@ -117,6 +117,15 @@ TEST_F(ParserSelectTest, timelineFunc) {
   run("SELECT LAST(*), FIRST(*) FROM t1 INTERVAL(10s)");
 
   run("SELECT diff(c1) FROM t1");
+
+  run("select diff(ts) from (select _wstart as ts, count(*) from st1 partition by tbname interval(1d))", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
+
+  run("select diff(ts) from (select _wstart as ts, count(*) from st1 partition by tbname interval(1d) order by ts)");
+
+  run("select t1.* from st1s1 t1, (select _wstart as ts, count(*) from st1s2 partition by tbname interval(1d)) WHERE t1.ts = t2.ts", TSDB_CODE_PAR_NOT_SUPPORT_JOIN);
+
+  run("select t1.* from st1s1 t1, (select _wstart as ts, count(*) from st1s2 partition by tbname interval(1d) order by ts) t2 WHERE t1.ts = t2.ts");
+
 }
 
 TEST_F(ParserSelectTest, selectFunc) {
@@ -325,6 +334,10 @@ TEST_F(ParserSelectTest, subquery) {
   run("SELECT SUM(a) FROM (SELECT MAX(c1) a, _wstart FROM st1s1 PARTITION BY TBNAME INTERVAL(1m) ORDER BY _WSTART) "
       "INTERVAL(1n)");
 
+  run("SELECT diff(a) FROM (SELECT _wstart, tag1, tag2, MAX(c1) a FROM st1 PARTITION BY tag1 INTERVAL(1m)) PARTITION BY tag1");
+
+  run("SELECT diff(a) FROM (SELECT _wstart, tag1, tag2, MAX(c1) a FROM st1 PARTITION BY tag1 INTERVAL(1m)) PARTITION BY tag2", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
+
   run("SELECT _C0 FROM (SELECT _ROWTS, ts FROM st1s1)");
 
   run("SELECT ts FROM (SELECT t1.ts FROM st1s1 t1)");

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

@@ -4028,13 +4028,13 @@ class TDTestCase:
         tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1}) partition by tbname range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
 
         tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union select * from {dbname}.{ctbname2}) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
-        tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
+        tdSql.query(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
 
         tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2}) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
-        tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
+        tdSql.query(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
 
         tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2}) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
-        tdSql.error(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
+        tdSql.query(f"select _irowts, _isfilled, interp(c0) from (select * from {dbname}.{ctbname1} union all select * from {dbname}.{ctbname2} order by ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")
 
         tdSql.query(f"select _irowts, _isfilled, interp(c0) from (select {ctbname1}.ts,{ctbname1}.c0 from {dbname}.{ctbname1}, {dbname}.{ctbname2} where {ctbname1}.ts = {ctbname2}.ts) range('2020-02-01 00:00:00', '2020-02-01 00:00:14') every(1s) fill(null)")