feat: fill physical plan

include/libs/nodes/nodes.h

@@ -185,6 +185,7 @@ typedef enum ENodeType {
   QUERY_NODE_LOGIC_PLAN_VNODE_MODIF,
   QUERY_NODE_LOGIC_PLAN_EXCHANGE,
   QUERY_NODE_LOGIC_PLAN_WINDOW,
+  QUERY_NODE_LOGIC_PLAN_FILL,
   QUERY_NODE_LOGIC_PLAN_SORT,
   QUERY_NODE_LOGIC_PLAN_PARTITION,
   QUERY_NODE_LOGIC_SUBPLAN,
@@ -202,6 +203,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
   QUERY_NODE_PHYSICAL_PLAN_SORT,
   QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
+  QUERY_NODE_PHYSICAL_PLAN_FILL,
   QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_PARTITION,

include/libs/nodes/plannodes.h

@@ -102,7 +102,6 @@ typedef struct SWindowLogicNode {
   int64_t     sliding;
   int8_t      intervalUnit;
   int8_t      slidingUnit;
-  SFillNode*  pFill;
   int64_t     sessionGap;
   SNode*      pTspk;
   SNode*      pStateExpr;
@@ -110,6 +109,13 @@ typedef struct SWindowLogicNode {
   int64_t     watermark;
 } SWindowLogicNode;
 
+typedef struct SFillLogicNode {
+  SLogicNode node;
+  EFillMode  mode;
+  SNode*     pWStartTs;
+  SNode*     pValues;  // SNodeListNode
+} SFillLogicNode;
+
 typedef struct SSortLogicNode {
   SLogicNode node;
   SNodeList* pSortKeys;
@@ -223,10 +229,12 @@ typedef struct SProjectPhysiNode {
 typedef struct SJoinPhysiNode {
   SPhysiNode node;
   EJoinType  joinType;
-  SNode*     pOnConditions;  // in or out tuple ?
+  SNode*     pOnConditions;
   SNodeList* pTargets;
 } SJoinPhysiNode;
 
+typedef SJoinPhysiNode SSortMergeJoinPhysiNode;
+
 typedef struct SAggPhysiNode {
   SPhysiNode node;
   SNodeList* pExprs;  // these are expression list of group_by_clause and parameter expression of aggregate function
@@ -263,9 +271,16 @@ typedef struct SIntervalPhysiNode {
   int64_t          sliding;
   int8_t           intervalUnit;
   int8_t           slidingUnit;
-  SFillNode*       pFill;
 } SIntervalPhysiNode;
 
+typedef struct SFillPhysiNode {
+  SPhysiNode node;
+  EFillMode  mode;
+  SNode*     pWStartTs;  // SColumnNode
+  SNode*     pValues;    // SNodeListNode
+  SNodeList* pTargets;
+} SFillPhysiNode;
+
 typedef struct SMultiTableIntervalPhysiNode {
   SIntervalPhysiNode interval;
   SNodeList*         pPartitionKeys;
@@ -340,7 +355,7 @@ typedef struct SQueryPlan {
   int32_t      numOfSubplans;
   SNodeList*   pSubplans;  // Element is SNodeListNode. The execution level of subplan, starting from 0.
   SExplainInfo explainInfo;
-  SArray* pPlaceholderValues;
+  SArray*      pPlaceholderValues;
 } SQueryPlan;
 
 void nodesWalkPhysiPlan(SNode* pNode, FNodeWalker walker, void* pContext);

include/libs/nodes/querynodes.h

@@ -89,7 +89,7 @@ typedef struct SValueNode {
     char*    p;
   } datum;
   int64_t typeData;
-  char unit;
+  char    unit;
 } SValueNode;
 
 typedef struct SOperatorNode {
@@ -211,7 +211,8 @@ typedef enum EFillMode {
 typedef struct SFillNode {
   ENodeType type;  // QUERY_NODE_FILL
   EFillMode mode;
-  SNode*    pValues;  // SNodeListNode
+  SNode*    pValues;    // SNodeListNode
+  SNode*    pWStartTs;  // _wstartts pseudo column
 } SFillNode;
 
 typedef struct SSelectStmt {
@@ -300,7 +301,7 @@ int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char*
                             SNodeList** pCols);
 
 typedef bool (*FFuncClassifier)(int32_t funcId);
-int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNodeList** pFuncs);
+int32_t nodesCollectFuncs(SSelectStmt* pSelect, ESqlClause clause, FFuncClassifier classifier, SNodeList** pFuncs);
 
 int32_t nodesCollectSpecialNodes(SSelectStmt* pSelect, ESqlClause clause, ENodeType type, SNodeList** pNodes);
 
@@ -314,11 +315,11 @@ bool nodesIsJsonOp(const SOperatorNode* pOp);
 bool nodesIsTimeorderQuery(const SNode* pQuery);
 bool nodesIsTimelineQuery(const SNode* pQuery);
 
-void* nodesGetValueFromNode(SValueNode* pNode);
-int32_t nodesSetValueNodeValue(SValueNode* pNode, void *value);
-char* nodesGetStrValueFromNode(SValueNode* pNode);
-char* getFillModeString(EFillMode mode);
-void  valueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
+void*   nodesGetValueFromNode(SValueNode* pNode);
+int32_t nodesSetValueNodeValue(SValueNode* pNode, void* value);
+char*   nodesGetStrValueFromNode(SValueNode* pNode);
+char*   getFillModeString(EFillMode mode);
+void    valueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
 
 #ifdef __cplusplus
 }

source/libs/executor/src/executorimpl.c

@@ -1179,7 +1179,7 @@ static void setPseudoOutputColInfo(SSDataBlock* pResult, SqlFunctionCtx* pCtx, S
 }
 
 int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
-                           int32_t numOfOutput, SArray* pPseudoList) {
+                              int32_t numOfOutput, SArray* pPseudoList) {
   setPseudoOutputColInfo(pResult, pCtx, pPseudoList);
   pResult->info.groupId = pSrcBlock->info.groupId;
 
@@ -1258,7 +1258,7 @@ int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBloc
         SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
 
         SScalarParam dest = {.columnData = &idata};
-        int32_t code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
+        int32_t      code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
         if (code != TSDB_CODE_SUCCESS) {
           taosArrayDestroy(pBlockList);
           return code;
@@ -3959,9 +3959,8 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
       }
 
       SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp;
-      code =
-          setSDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
-                                    pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL);
+      code = setSDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
+                                       pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL);
       if (code != 0) {
         goto _error;
       }
@@ -4781,8 +4780,8 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
 
     // there is an scalar expression that needs to be calculated before apply the group aggregation.
     if (pAggInfo->pScalarExprInfo != NULL) {
-      int32_t code = projectApplyFunctions(pAggInfo->pScalarExprInfo, pBlock, pBlock, pAggInfo->pScalarCtx, pAggInfo->numOfScalarExpr,
-                            NULL);
+      int32_t code = projectApplyFunctions(pAggInfo->pScalarExprInfo, pBlock, pBlock, pAggInfo->pScalarCtx,
+                                           pAggInfo->numOfScalarExpr, NULL);
       if (code != TSDB_CODE_SUCCESS) {
         pTaskInfo->code = code;
         longjmp(pTaskInfo->env, pTaskInfo->code);
@@ -6540,8 +6539,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
       int32_t numOfCols = 0;
       SArray* pCols = extractColMatchInfo(pScanPhyNode->pScanCols, pScanPhyNode->node.pOutputDataBlockDesc, &numOfCols);
-      SOperatorInfo* pOperator =
-          createStreamScanOperatorInfo(pHandle->reader, pResBlock, pCols, tableIdList, pTaskInfo, pScanPhyNode->node.pConditions);
+      SOperatorInfo* pOperator = createStreamScanOperatorInfo(pHandle->reader, pResBlock, pCols, tableIdList, pTaskInfo,
+                                                              pScanPhyNode->node.pConditions);
       taosArrayDestroy(tableIdList);
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
@@ -6622,10 +6621,11 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, primaryTsSlotId, &as,
                                        pTableGroupInfo, pTaskInfo);
 
-    if (pIntervalPhyNode->pFill != NULL) {
-      pOptr = createFillOperatorInfo(pOptr, pExprInfo, num, &interval, pResBlock, pIntervalPhyNode->pFill->mode, NULL,
-                                     false, pTaskInfo);
-    }
+    // if (pIntervalPhyNode->pFill != NULL) {
+    //   pOptr = createFillOperatorInfo(pOptr, pExprInfo, num, &interval, pResBlock, pIntervalPhyNode->pFill->mode,
+    //   NULL,
+    //                                  false, pTaskInfo);
+    // }
 
   } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
     SSortPhysiNode* pSortPhyNode = (SSortPhysiNode*)pPhyNode;
@@ -6927,7 +6927,7 @@ tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle*
   code = initQueryTableDataCond(&cond, pTableScanNode);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
-   }
+  }
 #if 0
   return tsdbQueryTables(pHandle->reader, &cond, pTableGroupInfo, queryId, taskId);
 #endif

source/libs/nodes/src/nodesCloneFuncs.c

@@ -191,6 +191,7 @@ static SNode* nodeListNodeCopy(const SNodeListNode* pSrc, SNodeListNode* pDst) {
 static SNode* fillNodeCopy(const SFillNode* pSrc, SFillNode* pDst) {
   COPY_SCALAR_FIELD(mode);
   CLONE_NODE_FIELD(pValues);
+  CLONE_NODE_FIELD(pWStartTs);
   return (SNode*)pDst;
 }
 
@@ -269,11 +270,18 @@ static SNode* logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* pD
   COPY_ALL_SCALAR_FIELDS;
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   CLONE_NODE_LIST_FIELD(pFuncs);
-  CLONE_NODE_FIELD(pFill);
   CLONE_NODE_FIELD(pTspk);
   return (SNode*)pDst;
 }
 
+static SNode* logicFillCopy(const SFillLogicNode* pSrc, SFillLogicNode* pDst) {
+  COPY_ALL_SCALAR_FIELDS;
+  COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
+  CLONE_NODE_FIELD(pWStartTs);
+  CLONE_NODE_FIELD(pValues);
+  return (SNode*)pDst;
+}
+
 static SNode* logicSortCopy(const SSortLogicNode* pSrc, SSortLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   CLONE_NODE_LIST_FIELD(pSortKeys);
@@ -370,6 +378,8 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
       return logicExchangeCopy((const SExchangeLogicNode*)pNode, (SExchangeLogicNode*)pDst);
     case QUERY_NODE_LOGIC_PLAN_WINDOW:
       return logicWindowCopy((const SWindowLogicNode*)pNode, (SWindowLogicNode*)pDst);
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return logicFillCopy((const SFillLogicNode*)pNode, (SFillLogicNode*)pDst);
     case QUERY_NODE_LOGIC_PLAN_SORT:
       return logicSortCopy((const SSortLogicNode*)pNode, (SSortLogicNode*)pDst);
     case QUERY_NODE_LOGIC_PLAN_PARTITION:

source/libs/nodes/src/nodesCodeFuncs.c

@@ -192,6 +192,8 @@ const char* nodesNodeName(ENodeType type) {
       return "LogicExchange";
     case QUERY_NODE_LOGIC_PLAN_WINDOW:
       return "LogicWindow";
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return "LogicFill";
     case QUERY_NODE_LOGIC_PLAN_SORT:
       return "LogicSort";
     case QUERY_NODE_LOGIC_PLAN_PARTITION:
@@ -222,6 +224,8 @@ const char* nodesNodeName(ENodeType type) {
       return "PhysiSort";
     case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
       return "PhysiInterval";
+    case QUERY_NODE_PHYSICAL_PLAN_FILL:
+      return "PhysiFill";
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       return "PhysiSessionWindow";
     case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
@@ -564,6 +568,44 @@ static int32_t jsonToLogicProjectNode(const SJson* pJson, void* pObj) {
   return code;
 }
 
+static const char* jkFillLogicPlanMode = "Mode";
+static const char* jkFillLogicPlanWStartTs = "WStartTs";
+static const char* jkFillLogicPlanValues = "Values";
+
+static int32_t logicFillNodeToJson(const void* pObj, SJson* pJson) {
+  const SFillLogicNode* pNode = (const SFillLogicNode*)pObj;
+
+  int32_t code = logicPlanNodeToJson(pObj, pJson);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkFillLogicPlanMode, pNode->mode);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkFillLogicPlanWStartTs, nodeToJson, pNode->pWStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkFillLogicPlanValues, nodeToJson, pNode->pValues);
+  }
+
+  return code;
+}
+
+static int32_t jsonToLogicFillNode(const SJson* pJson, void* pObj) {
+  SFillLogicNode* pNode = (SFillLogicNode*)pObj;
+
+  int32_t code = jsonToLogicPlanNode(pJson, pObj);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetNumberValue(pJson, jkFillLogicPlanMode, pNode->mode);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkFillLogicPlanWStartTs, &pNode->pWStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkFillLogicPlanValues, &pNode->pValues);
+  }
+
+  return code;
+}
+
 static const char* jkSortLogicPlanSortKeys = "SortKeys";
 
 static int32_t logicSortNodeToJson(const void* pObj, SJson* pJson) {
@@ -1382,7 +1424,6 @@ static const char* jkIntervalPhysiPlanOffset = "Offset";
 static const char* jkIntervalPhysiPlanSliding = "Sliding";
 static const char* jkIntervalPhysiPlanIntervalUnit = "intervalUnit";
 static const char* jkIntervalPhysiPlanSlidingUnit = "slidingUnit";
-static const char* jkIntervalPhysiPlanFill = "Fill";
 
 static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
   const SIntervalPhysiNode* pNode = (const SIntervalPhysiNode*)pObj;
@@ -1403,9 +1444,6 @@ static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddIntegerToObject(pJson, jkIntervalPhysiPlanSlidingUnit, pNode->slidingUnit);
   }
-  if (TSDB_CODE_SUCCESS == code) {
-    code = tjsonAddObject(pJson, jkIntervalPhysiPlanFill, nodeToJson, pNode->pFill);
-  }
 
   return code;
 }
@@ -1429,8 +1467,50 @@ static int32_t jsonToPhysiIntervalNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetTinyIntValue(pJson, jkIntervalPhysiPlanSlidingUnit, &pNode->slidingUnit);
   }
+
+  return code;
+}
+
+static const char* jkFillPhysiPlanMode = "Mode";
+static const char* jkFillPhysiPlanWStartTs = "WStartTs";
+static const char* jkFillPhysiPlanValues = "Values";
+static const char* jkFillPhysiPlanTargets = "Targets";
+
+static int32_t physiFillNodeToJson(const void* pObj, SJson* pJson) {
+  const SFillPhysiNode* pNode = (const SFillPhysiNode*)pObj;
+
+  int32_t code = physicPlanNodeToJson(pObj, pJson);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkFillPhysiPlanMode, pNode->mode);
+  }
   if (TSDB_CODE_SUCCESS == code) {
-    code = jsonToNodeObject(pJson, jkIntervalPhysiPlanFill, (SNode**)&pNode->pFill);
+    code = tjsonAddObject(pJson, jkFillPhysiPlanWStartTs, nodeToJson, pNode->pWStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkFillPhysiPlanValues, nodeToJson, pNode->pValues);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodeListToJson(pJson, jkFillPhysiPlanTargets, pNode->pTargets);
+  }
+
+  return code;
+}
+
+static int32_t jsonToPhysiFillNode(const SJson* pJson, void* pObj) {
+  SFillPhysiNode* pNode = (SFillPhysiNode*)pObj;
+
+  int32_t code = jsonToPhysiWindowNode(pJson, pObj);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetNumberValue(pJson, jkFillPhysiPlanMode, pNode->mode);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkFillPhysiPlanWStartTs, &pNode->pWStartTs);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkFillPhysiPlanValues, &pNode->pValues);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeList(pJson, jkFillPhysiPlanTargets, &pNode->pTargets);
   }
 
   return code;
@@ -2328,6 +2408,7 @@ static int32_t jsonToNodeListNode(const SJson* pJson, void* pObj) {
 
 static const char* jkFillMode = "Mode";
 static const char* jkFillValues = "Values";
+static const char* jkFillWStartTs = "WStartTs";
 
 static int32_t fillNodeToJson(const void* pObj, SJson* pJson) {
   const SFillNode* pNode = (const SFillNode*)pObj;
@@ -2336,6 +2417,9 @@ static int32_t fillNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkFillValues, nodeToJson, pNode->pValues);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkFillWStartTs, nodeToJson, pNode->pWStartTs);
+  }
 
   return code;
 }
@@ -2347,6 +2431,9 @@ static int32_t jsonToFillNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkFillValues, &pNode->pValues);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkFillWStartTs, &pNode->pWStartTs);
+  }
 
   return code;
 }
@@ -2707,6 +2794,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
       return logicProjectNodeToJson(pObj, pJson);
     case QUERY_NODE_LOGIC_PLAN_VNODE_MODIF:
       break;
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return logicFillNodeToJson(pObj, pJson);
     case QUERY_NODE_LOGIC_PLAN_SORT:
       return logicSortNodeToJson(pObj, pJson);
     case QUERY_NODE_LOGIC_PLAN_PARTITION:
@@ -2735,6 +2824,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
       return physiSortNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
       return physiIntervalNodeToJson(pObj, pJson);
+    case QUERY_NODE_PHYSICAL_PLAN_FILL:
+      return physiFillNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       return physiSessionWindowNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
@@ -2795,6 +2886,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
       return jsonToLogicScanNode(pJson, pObj);
     case QUERY_NODE_LOGIC_PLAN_PROJECT:
       return jsonToLogicProjectNode(pJson, pObj);
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return jsonToLogicFillNode(pJson, pObj);
     case QUERY_NODE_LOGIC_PLAN_SORT:
       return jsonToLogicSortNode(pJson, pObj);
     case QUERY_NODE_LOGIC_PLAN_PARTITION:
@@ -2821,6 +2914,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
       return jsonToPhysiSortNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
       return jsonToPhysiIntervalNode(pJson, pObj);
+    case QUERY_NODE_PHYSICAL_PLAN_FILL:
+      return jsonToPhysiFillNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       return jsonToPhysiSessionWindowNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -132,9 +132,14 @@ static EDealRes dispatchExpr(SNode* pNode, ETraversalOrder order, FNodeWalker wa
     case QUERY_NODE_NODE_LIST:
       res = walkExprs(((SNodeListNode*)pNode)->pNodeList, order, walker, pContext);
       break;
-    case QUERY_NODE_FILL:
-      res = walkExpr(((SFillNode*)pNode)->pValues, order, walker, pContext);
+    case QUERY_NODE_FILL: {
+      SFillNode* pFill = (SFillNode*)pNode;
+      res = walkExpr(pFill->pValues, order, walker, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = walkExpr(pFill->pWStartTs, order, walker, pContext);
+      }
       break;
+    }
     case QUERY_NODE_RAW_EXPR:
       res = walkExpr(((SRawExprNode*)pNode)->pNode, order, walker, pContext);
       break;
@@ -272,9 +277,14 @@ static EDealRes rewriteExpr(SNode** pRawNode, ETraversalOrder order, FNodeRewrit
     case QUERY_NODE_NODE_LIST:
       res = rewriteExprs(((SNodeListNode*)pNode)->pNodeList, order, rewriter, pContext);
       break;
-    case QUERY_NODE_FILL:
-      res = rewriteExpr(&(((SFillNode*)pNode)->pValues), order, rewriter, pContext);
+    case QUERY_NODE_FILL: {
+      SFillNode* pFill = (SFillNode*)pNode;
+      res = rewriteExpr(&pFill->pValues, order, rewriter, pContext);
+      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
+        res = rewriteExpr(&(pFill->pWStartTs), order, rewriter, pContext);
+      }
       break;
+    }
     case QUERY_NODE_RAW_EXPR:
       res = rewriteExpr(&(((SRawExprNode*)pNode)->pNode), order, rewriter, pContext);
       break;
@@ -333,6 +343,9 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa
     case SQL_CLAUSE_PARTITION_BY:
       nodesWalkExpr(pSelect->pWindow, walker, pContext);
     case SQL_CLAUSE_WINDOW:
+      if (NULL != pSelect->pWindow && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) {
+        nodesWalkExpr(((SIntervalWindowNode*)pSelect->pWindow)->pFill, walker, pContext);
+      }
       nodesWalkExprs(pSelect->pGroupByList, walker, pContext);
     case SQL_CLAUSE_GROUP_BY:
       nodesWalkExpr(pSelect->pHaving, walker, pContext);
@@ -362,6 +375,9 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit
     case SQL_CLAUSE_PARTITION_BY:
       nodesRewriteExpr(&(pSelect->pWindow), rewriter, pContext);
     case SQL_CLAUSE_WINDOW:
+      if (NULL != pSelect->pWindow && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) {
+        nodesRewriteExpr(&(((SIntervalWindowNode*)pSelect->pWindow)->pFill), rewriter, pContext);
+      }
       nodesRewriteExprs(pSelect->pGroupByList, rewriter, pContext);
     case SQL_CLAUSE_GROUP_BY:
       nodesRewriteExpr(&(pSelect->pHaving), rewriter, pContext);
@@ -496,14 +512,9 @@ static EDealRes dispatchPhysiPlan(SNode* pNode, ETraversalOrder order, FNodeWalk
       }
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL: {
-      SIntervalPhysiNode* pInterval = (SIntervalPhysiNode*)pNode;
+    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
       res = walkWindowPhysi((SWinodwPhysiNode*)pNode, order, walker, pContext);
-      if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {
-        res = walkPhysiPlan((SNode*)pInterval->pFill, order, walker, pContext);
-      }
       break;
-    }
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       res = walkWindowPhysi((SWinodwPhysiNode*)pNode, order, walker, pContext);
       break;

source/libs/nodes/src/nodesUtilFuncs.c

@@ -213,6 +213,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SExchangeLogicNode));
     case QUERY_NODE_LOGIC_PLAN_WINDOW:
       return makeNode(type, sizeof(SWindowLogicNode));
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return makeNode(type, sizeof(SFillLogicNode));
     case QUERY_NODE_LOGIC_PLAN_SORT:
       return makeNode(type, sizeof(SSortLogicNode));
     case QUERY_NODE_LOGIC_PLAN_PARTITION:
@@ -243,6 +245,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SSortPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
       return makeNode(type, sizeof(SIntervalPhysiNode));
+    case QUERY_NODE_PHYSICAL_PLAN_FILL:
+      return makeNode(type, sizeof(SFillPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       return makeNode(type, sizeof(SSessionWinodwPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
@@ -373,9 +377,12 @@ void nodesDestroyNode(SNodeptr pNode) {
     case QUERY_NODE_NODE_LIST:
       nodesDestroyList(((SNodeListNode*)pNode)->pNodeList);
       break;
-    case QUERY_NODE_FILL:
-      nodesDestroyNode(((SFillNode*)pNode)->pValues);
+    case QUERY_NODE_FILL: {
+      SFillNode* pFill = (SFillNode*)pNode;
+      nodesDestroyNode(pFill->pValues);
+      nodesDestroyNode(pFill->pWStartTs);
       break;
+    }
     case QUERY_NODE_RAW_EXPR:
       nodesDestroyNode(((SRawExprNode*)pNode)->pNode);
       break;
@@ -554,7 +561,6 @@ void nodesDestroyNode(SNodeptr pNode) {
       SWindowLogicNode* pLogicNode = (SWindowLogicNode*)pNode;
       destroyLogicNode((SLogicNode*)pLogicNode);
       nodesDestroyList(pLogicNode->pFuncs);
-      nodesDestroyNode(pLogicNode->pFill);
       nodesDestroyNode(pLogicNode->pTspk);
       break;
     }
@@ -630,12 +636,9 @@ void nodesDestroyNode(SNodeptr pNode) {
       nodesDestroyNode(pPhyNode->pSortKeys);
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL: {
-      SIntervalPhysiNode* pPhyNode = (SIntervalPhysiNode*)pNode;
-      destroyWinodwPhysiNode((SWinodwPhysiNode*)pPhyNode);
-      nodesDestroyNode(pPhyNode->pFill);
+    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
+      destroyWinodwPhysiNode((SWinodwPhysiNode*)pNode);
       break;
-    }
     case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
       destroyWinodwPhysiNode((SWinodwPhysiNode*)pNode);
       break;
@@ -894,7 +897,7 @@ void* nodesGetValueFromNode(SValueNode* pNode) {
   return NULL;
 }
 
-int32_t nodesSetValueNodeValue(SValueNode* pNode, void *value) {
+int32_t nodesSetValueNodeValue(SValueNode* pNode, void* value) {
   switch (pNode->node.resType.type) {
     case TSDB_DATA_TYPE_BOOL:
       pNode->datum.b = *(bool*)value;
@@ -1192,7 +1195,7 @@ static EDealRes collectFuncs(SNode* pNode, void* pContext) {
   return DEAL_RES_CONTINUE;
 }
 
-int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNodeList** pFuncs) {
+int32_t nodesCollectFuncs(SSelectStmt* pSelect, ESqlClause clause, FFuncClassifier classifier, SNodeList** pFuncs) {
   if (NULL == pSelect || NULL == pFuncs) {
     return TSDB_CODE_FAILED;
   }
@@ -1203,7 +1206,7 @@ int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNod
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   *pFuncs = NULL;
-  nodesWalkSelectStmt(pSelect, SQL_CLAUSE_GROUP_BY, collectFuncs, &cxt);
+  nodesWalkSelectStmt(pSelect, clause, collectFuncs, &cxt);
   if (TSDB_CODE_SUCCESS != cxt.errCode) {
     nodesDestroyList(cxt.pFuncs);
     return cxt.errCode;

source/libs/parser/src/parAstCreater.c

@@ -501,6 +501,12 @@ SNode* createFillNode(SAstCreateContext* pCxt, EFillMode mode, SNode* pValues) {
   CHECK_OUT_OF_MEM(fill);
   fill->mode = mode;
   fill->pValues = pValues;
+  fill->pWStartTs = nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == fill->pWStartTs) {
+    nodesDestroyNode(fill);
+    CHECK_OUT_OF_MEM(fill->pWStartTs);
+  }
+  strcpy(((SFunctionNode*)fill->pWStartTs)->functionName, "_wstartts");
   return (SNode*)fill;
 }
 

source/libs/parser/src/parTokenizer.c

@@ -26,193 +26,192 @@ typedef struct SKeyword {
   uint8_t     len;   // length
 } SKeyword;
 
+// clang-format off
 // keywords in sql string
 static SKeyword keywordTable[] = {
-    {"ACCOUNT", TK_ACCOUNT},
-    {"ACCOUNTS", TK_ACCOUNTS},
-    {"ADD", TK_ADD},
-    {"AGGREGATE", TK_AGGREGATE},
-    {"ALL", TK_ALL},
-    {"ALTER", TK_ALTER},
-    {"ANALYZE", TK_ANALYZE},
-    {"AND", TK_AND},
-    {"APPS", TK_APPS},
-    {"AS", TK_AS},
-    {"ASC", TK_ASC},
-    {"AT_ONCE", TK_AT_ONCE},
-    {"BETWEEN", TK_BETWEEN},
-    {"BINARY", TK_BINARY},
-    {"BIGINT", TK_BIGINT},
-    // {"BLOCKS", TK_BLOCKS},
-    {"BNODE", TK_BNODE},
-    {"BNODES", TK_BNODES},
-    {"BOOL", TK_BOOL},
-    {"BUFFER", TK_BUFFER},
-    {"BUFSIZE", TK_BUFSIZE},
-    {"BY", TK_BY},
-    {"CACHE", TK_CACHE},
-    {"CACHELAST", TK_CACHELAST},
-    {"CAST", TK_CAST},
-    {"CLUSTER", TK_CLUSTER},
-    {"COLUMN", TK_COLUMN},
-    {"COMMENT", TK_COMMENT},
-    {"COMP", TK_COMP},
-    {"COMPACT", TK_COMPACT},
-    {"CONNS", TK_CONNS},
-    {"CONNECTION", TK_CONNECTION},
-    {"CONNECTIONS", TK_CONNECTIONS},
-    {"COUNT", TK_COUNT},
-    {"CREATE", TK_CREATE},
-    {"DATABASE", TK_DATABASE},
-    {"DATABASES", TK_DATABASES},
-    {"DAYS", TK_DAYS},
-    {"DBS", TK_DBS},
-    {"DELAY", TK_DELAY},
-    {"DESC", TK_DESC},
-    {"DESCRIBE", TK_DESCRIBE},
-    {"DISTINCT", TK_DISTINCT},
-    {"DNODE", TK_DNODE},
-    {"DNODES", TK_DNODES},
-    {"DOUBLE", TK_DOUBLE},
-    {"DROP", TK_DROP},
-    {"EXISTS", TK_EXISTS},
-    {"EXPLAIN", TK_EXPLAIN},
-    {"FILE_FACTOR", TK_FILE_FACTOR},
-    {"FILL", TK_FILL},
-    {"FIRST", TK_FIRST},
-    {"FLOAT", TK_FLOAT},
-    {"FROM", TK_FROM},
-    {"FSYNC", TK_FSYNC},
-    {"FUNCTION", TK_FUNCTION},
-    {"FUNCTIONS", TK_FUNCTIONS},
-    {"GRANTS", TK_GRANTS},
-    {"GROUP", TK_GROUP},
-    {"HAVING", TK_HAVING},
-    {"IF", TK_IF},
-    {"IMPORT", TK_IMPORT},
-    {"IN", TK_IN},
-    {"INDEX", TK_INDEX},
-    {"INDEXES", TK_INDEXES},
-    {"INNER", TK_INNER},
-    {"INT", TK_INT},
-    {"INSERT", TK_INSERT},
-    {"INTEGER", TK_INTEGER},
-    {"INTERVAL", TK_INTERVAL},
-    {"INTO", TK_INTO},
-    {"IS", TK_IS},
-    {"JOIN", TK_JOIN},
-    {"JSON", TK_JSON},
-    {"KEEP", TK_KEEP},
-    {"KILL", TK_KILL},
-    {"LAST", TK_LAST},
-    {"LAST_ROW", TK_LAST_ROW},
-    {"LICENCE", TK_LICENCE},
-    {"LIKE", TK_LIKE},
-    {"LIMIT", TK_LIMIT},
-    {"LINEAR", TK_LINEAR},
-    {"LOCAL", TK_LOCAL},
-    {"MATCH", TK_MATCH},
-    {"MAXROWS", TK_MAXROWS},
-    {"MINROWS", TK_MINROWS},
-    {"MINUS", TK_MINUS},
-    {"MNODE", TK_MNODE},
-    {"MNODES", TK_MNODES},
-    {"MODIFY", TK_MODIFY},
-    {"MODULES", TK_MODULES},
-    {"NCHAR", TK_NCHAR},
-    {"NMATCH", TK_NMATCH},
-    {"NONE", TK_NONE},
-    {"NOT", TK_NOT},
-    {"NOW", TK_NOW},
-    {"NULL", TK_NULL},
-    {"NULLS", TK_NULLS},
-    {"OFFSET", TK_OFFSET},
-    {"ON", TK_ON},
-    {"OR", TK_OR},
-    {"ORDER", TK_ORDER},
-    {"OUTPUTTYPE", TK_OUTPUTTYPE},
-    {"PARTITION", TK_PARTITION},
-    {"PASS", TK_PASS},
-    {"PAGES", TK_PAGES},
-    {"PAGESIZE", TK_PAGESIZE},
-    {"PORT", TK_PORT},
-    {"PPS", TK_PPS},
-    {"PRECISION", TK_PRECISION},
-    {"PRIVILEGE", TK_PRIVILEGE},
-    {"PREV", TK_PREV},
-    {"QNODE", TK_QNODE},
-    {"QNODES", TK_QNODES},
-    {"QTIME", TK_QTIME},
-    {"QUERIES", TK_QUERIES},
-    {"QUERY", TK_QUERY},
-    // {"QUORUM", TK_QUORUM},
-    {"RATIO", TK_RATIO},
-    {"REPLICA", TK_REPLICA},
-    {"RESET", TK_RESET},
-    {"RETENTIONS", TK_RETENTIONS},
-    {"ROLLUP", TK_ROLLUP},
-    {"SCHEMA", TK_SCHEMA},
-    {"SCORES", TK_SCORES},
-    {"SELECT", TK_SELECT},
-    {"SESSION", TK_SESSION},
-    {"SET", TK_SET},
-    {"SHOW", TK_SHOW},
+    {"ACCOUNT",       TK_ACCOUNT},
+    {"ACCOUNTS",      TK_ACCOUNTS},
+    {"ADD",           TK_ADD},
+    {"AGGREGATE",     TK_AGGREGATE},
+    {"ALL",           TK_ALL},
+    {"ALTER",         TK_ALTER},
+    {"ANALYZE",       TK_ANALYZE},
+    {"AND",           TK_AND},
+    {"APPS",          TK_APPS},
+    {"AS",            TK_AS},
+    {"ASC",           TK_ASC},
+    {"AT_ONCE",       TK_AT_ONCE},
+    {"BETWEEN",       TK_BETWEEN},
+    {"BINARY",        TK_BINARY},
+    {"BIGINT",        TK_BIGINT},
+    {"BNODE",         TK_BNODE},
+    {"BNODES",        TK_BNODES},
+    {"BOOL",          TK_BOOL},
+    {"BUFFER",        TK_BUFFER},
+    {"BUFSIZE",       TK_BUFSIZE},
+    {"BY",            TK_BY},
+    {"CACHE",         TK_CACHE},
+    {"CACHELAST",     TK_CACHELAST},
+    {"CAST",          TK_CAST},
+    {"CLUSTER",       TK_CLUSTER},
+    {"COLUMN",        TK_COLUMN},
+    {"COMMENT",       TK_COMMENT},
+    {"COMP",          TK_COMP},
+    {"COMPACT",       TK_COMPACT},
+    {"CONNS",         TK_CONNS},
+    {"CONNECTION",    TK_CONNECTION},
+    {"CONNECTIONS",   TK_CONNECTIONS},
+    {"COUNT",         TK_COUNT},
+    {"CREATE",        TK_CREATE},
+    {"DATABASE",      TK_DATABASE},
+    {"DATABASES",     TK_DATABASES},
+    {"DAYS",          TK_DAYS},
+    {"DBS",           TK_DBS},
+    {"DELAY",         TK_DELAY},
+    {"DESC",          TK_DESC},
+    {"DESCRIBE",      TK_DESCRIBE},
+    {"DISTINCT",      TK_DISTINCT},
+    {"DNODE",         TK_DNODE},
+    {"DNODES",        TK_DNODES},
+    {"DOUBLE",        TK_DOUBLE},
+    {"DROP",          TK_DROP},
+    {"EXISTS",        TK_EXISTS},
+    {"EXPLAIN",       TK_EXPLAIN},
+    {"FILE_FACTOR",   TK_FILE_FACTOR},
+    {"FILL",          TK_FILL},
+    {"FIRST",         TK_FIRST},
+    {"FLOAT",         TK_FLOAT},
+    {"FROM",          TK_FROM},
+    {"FSYNC",         TK_FSYNC},
+    {"FUNCTION",      TK_FUNCTION},
+    {"FUNCTIONS",     TK_FUNCTIONS},
+    {"GRANTS",        TK_GRANTS},
+    {"GROUP",         TK_GROUP},
+    {"HAVING",        TK_HAVING},
+    {"IF",            TK_IF},
+    {"IMPORT",        TK_IMPORT},
+    {"IN",            TK_IN},
+    {"INDEX",         TK_INDEX},
+    {"INDEXES",       TK_INDEXES},
+    {"INNER",         TK_INNER},
+    {"INT",           TK_INT},
+    {"INSERT",        TK_INSERT},
+    {"INTEGER",       TK_INTEGER},
+    {"INTERVAL",      TK_INTERVAL},
+    {"INTO",          TK_INTO},
+    {"IS",            TK_IS},
+    {"JOIN",          TK_JOIN},
+    {"JSON",          TK_JSON},
+    {"KEEP",          TK_KEEP},
+    {"KILL",          TK_KILL},
+    {"LAST",          TK_LAST},
+    {"LAST_ROW",      TK_LAST_ROW},
+    {"LICENCE",       TK_LICENCE},
+    {"LIKE",          TK_LIKE},
+    {"LIMIT",         TK_LIMIT},
+    {"LINEAR",        TK_LINEAR},
+    {"LOCAL",         TK_LOCAL},
+    {"MATCH",         TK_MATCH},
+    {"MAXROWS",       TK_MAXROWS},
+    {"MINROWS",       TK_MINROWS},
+    {"MINUS",         TK_MINUS},
+    {"MNODE",         TK_MNODE},
+    {"MNODES",        TK_MNODES},
+    {"MODIFY",        TK_MODIFY},
+    {"MODULES",       TK_MODULES},
+    {"NCHAR",         TK_NCHAR},
+    {"NEXT",          TK_NEXT},
+    {"NMATCH",        TK_NMATCH},
+    {"NONE",          TK_NONE},
+    {"NOT",           TK_NOT},
+    {"NOW",           TK_NOW},
+    {"NULL",          TK_NULL},
+    {"NULLS",         TK_NULLS},
+    {"OFFSET",        TK_OFFSET},
+    {"ON",            TK_ON},
+    {"OR",            TK_OR},
+    {"ORDER",         TK_ORDER},
+    {"OUTPUTTYPE",    TK_OUTPUTTYPE},
+    {"PARTITION",     TK_PARTITION},
+    {"PASS",          TK_PASS},
+    {"PAGES",         TK_PAGES},
+    {"PAGESIZE",      TK_PAGESIZE},
+    {"PORT",          TK_PORT},
+    {"PPS",           TK_PPS},
+    {"PRECISION",     TK_PRECISION},
+    {"PRIVILEGE",     TK_PRIVILEGE},
+    {"PREV",          TK_PREV},
+    {"QNODE",         TK_QNODE},
+    {"QNODES",        TK_QNODES},
+    {"QTIME",         TK_QTIME},
+    {"QUERIES",       TK_QUERIES},
+    {"QUERY",         TK_QUERY},
+    {"RATIO",         TK_RATIO},
+    {"REPLICA",       TK_REPLICA},
+    {"RESET",         TK_RESET},
+    {"RETENTIONS",    TK_RETENTIONS},
+    {"ROLLUP",        TK_ROLLUP},
+    {"SCHEMA",        TK_SCHEMA},
+    {"SCORES",        TK_SCORES},
+    {"SELECT",        TK_SELECT},
+    {"SESSION",       TK_SESSION},
+    {"SET",           TK_SET},
+    {"SHOW",          TK_SHOW},
     {"SINGLE_STABLE", TK_SINGLE_STABLE},
-    {"SLIDING", TK_SLIDING},
-    {"SLIMIT", TK_SLIMIT},
-    {"SMA", TK_SMA},
-    {"SMALLINT", TK_SMALLINT},
-    {"SNODE", TK_SNODE},
-    {"SNODES", TK_SNODES},
-    {"SOFFSET", TK_SOFFSET},
-    {"STABLE", TK_STABLE},
-    {"STABLES", TK_STABLES},
-    {"STATE", TK_STATE},
-    {"STATE_WINDOW", TK_STATE_WINDOW},
-    {"STORAGE", TK_STORAGE},
-    {"STREAM", TK_STREAM},
-    {"STREAMS", TK_STREAMS},
-    // {"STREAM_MODE", TK_STREAM_MODE},
-    {"STRICT", TK_STRICT},
-    {"SYNCDB", TK_SYNCDB},
-    {"TABLE", TK_TABLE},
-    {"TABLES", TK_TABLES},
-    {"TAG", TK_TAG},
-    {"TAGS", TK_TAGS},
-    {"TBNAME", TK_TBNAME},
-    {"TIMESTAMP", TK_TIMESTAMP},
-    {"TIMEZONE", TK_TIMEZONE},
-    {"TINYINT", TK_TINYINT},
-    {"TODAY", TK_TODAY},
-    {"TOPIC", TK_TOPIC},
-    {"TOPICS", TK_TOPICS},
-    {"TRIGGER", TK_TRIGGER},
-    {"TSERIES", TK_TSERIES},
-    {"TTL", TK_TTL},
-    {"UNION", TK_UNION},
-    {"UNSIGNED", TK_UNSIGNED},
-    {"USE", TK_USE},
-    {"USER", TK_USER},
-    {"USERS", TK_USERS},
-    {"USING", TK_USING},
-    {"VALUE", TK_VALUE},
-    {"VALUES", TK_VALUES},
-    {"VARCHAR", TK_VARCHAR},
-    {"VARIABLES", TK_VARIABLES},
-    {"VERBOSE", TK_VERBOSE},
-    {"VGROUPS", TK_VGROUPS},
-    {"VNODES", TK_VNODES},
-    {"WAL", TK_WAL},
-    {"WATERMARK", TK_WATERMARK},
-    {"WHERE", TK_WHERE},
-    {"WINDOW_CLOSE", TK_WINDOW_CLOSE},
-    {"WITH", TK_WITH},
-    {"_QENDTS", TK_QENDTS},
-    {"_QSTARTTS", TK_QSTARTTS},
-    {"_ROWTS", TK_ROWTS},
-    {"_WDURATION", TK_WDURATION},
-    {"_WENDTS", TK_WENDTS},
-    {"_WSTARTTS", TK_WSTARTTS},
+    {"SLIDING",       TK_SLIDING},
+    {"SLIMIT",        TK_SLIMIT},
+    {"SMA",           TK_SMA},
+    {"SMALLINT",      TK_SMALLINT},
+    {"SNODE",         TK_SNODE},
+    {"SNODES",        TK_SNODES},
+    {"SOFFSET",       TK_SOFFSET},
+    {"STABLE",        TK_STABLE},
+    {"STABLES",       TK_STABLES},
+    {"STATE",         TK_STATE},
+    {"STATE_WINDOW",  TK_STATE_WINDOW},
+    {"STORAGE",       TK_STORAGE},
+    {"STREAM",        TK_STREAM},
+    {"STREAMS",       TK_STREAMS},
+    {"STRICT",        TK_STRICT},
+    {"SYNCDB",        TK_SYNCDB},
+    {"TABLE",         TK_TABLE},
+    {"TABLES",        TK_TABLES},
+    {"TAG",           TK_TAG},
+    {"TAGS",          TK_TAGS},
+    {"TBNAME",        TK_TBNAME},
+    {"TIMESTAMP",     TK_TIMESTAMP},
+    {"TIMEZONE",      TK_TIMEZONE},
+    {"TINYINT",       TK_TINYINT},
+    {"TODAY",         TK_TODAY},
+    {"TOPIC",         TK_TOPIC},
+    {"TOPICS",        TK_TOPICS},
+    {"TRIGGER",       TK_TRIGGER},
+    {"TSERIES",       TK_TSERIES},
+    {"TTL",           TK_TTL},
+    {"UNION",         TK_UNION},
+    {"UNSIGNED",      TK_UNSIGNED},
+    {"USE",           TK_USE},
+    {"USER",          TK_USER},
+    {"USERS",         TK_USERS},
+    {"USING",         TK_USING},
+    {"VALUE",         TK_VALUE},
+    {"VALUES",        TK_VALUES},
+    {"VARCHAR",       TK_VARCHAR},
+    {"VARIABLES",     TK_VARIABLES},
+    {"VERBOSE",       TK_VERBOSE},
+    {"VGROUPS",       TK_VGROUPS},
+    {"VNODES",        TK_VNODES},
+    {"WAL",           TK_WAL},
+    {"WATERMARK",     TK_WATERMARK},
+    {"WHERE",         TK_WHERE},
+    {"WINDOW_CLOSE",  TK_WINDOW_CLOSE},
+    {"WITH",          TK_WITH},
+    {"_QENDTS",       TK_QENDTS},
+    {"_QSTARTTS",     TK_QSTARTTS},
+    {"_ROWTS",        TK_ROWTS},
+    {"_WDURATION",    TK_WDURATION},
+    {"_WENDTS",       TK_WENDTS},
+    {"_WSTARTTS",     TK_WSTARTTS},
     // {"ID",           TK_ID},
     // {"STRING",       TK_STRING},
     // {"EQ",           TK_EQ},
@@ -279,6 +278,7 @@ static SKeyword keywordTable[] = {
     // {"PARTITIONS",   TK_PARTITIONS},
     // {"MODE",         TK_MODE},
 };
+// clang-format on
 
 static const char isIdChar[] = {
     /* x0 x1 x2 x3 x4 x5 x6 x7 x8 x9 xA xB xC xD xE xF */

source/libs/parser/test/parSelectTest.cpp

@@ -24,170 +24,179 @@ class ParserSelectTest : public ParserTestBase {};
 TEST_F(ParserSelectTest, basic) {
   useDb("root", "test");
 
-  run("select * from t1");
+  run("SELECT * FROM t1");
 
-  run("select * from test.t1");
+  run("SELECT * FROM test.t1");
 
-  run("select ts, c1 from t1");
+  run("SELECT ts, c1 FROM t1");
 
-  run("select ts, t.c1 from (select * from t1) t");
+  run("SELECT ts, t.c1 FROM (SELECT * FROM t1) t");
 
-  run("select * from t1 tt1, t1 tt2 where tt1.c1 = tt2.c1");
+  run("SELECT * FROM t1 tt1, t1 tt2 where tt1.c1 = tt2.c1");
 }
 
 TEST_F(ParserSelectTest, constant) {
   useDb("root", "test");
 
-  run("select 123, 20.4, 'abc', \"wxy\", timestamp '2022-02-09 17:30:20', true, false, 10s from t1");
+  run("SELECT 123, 20.4, 'abc', \"wxy\", timestamp '2022-02-09 17:30:20', true, false, 10s FROM t1");
 
-  run("select 1234567890123456789012345678901234567890, 20.1234567890123456789012345678901234567890, 'abc', \"wxy\", "
-      "timestamp '2022-02-09 17:30:20', true, false, 15s from t1");
+  run("SELECT 1234567890123456789012345678901234567890, 20.1234567890123456789012345678901234567890, 'abc', \"wxy\", "
+      "timestamp '2022-02-09 17:30:20', true, false, 15s FROM t1");
 
-  run("select 123 + 45 from t1 where 2 - 1");
+  run("SELECT 123 + 45 FROM t1 where 2 - 1");
 }
 
 TEST_F(ParserSelectTest, expression) {
   useDb("root", "test");
 
-  run("select ts + 10s, c1 + 10, concat(c2, 'abc') from t1");
+  run("SELECT ts + 10s, c1 + 10, concat(c2, 'abc') FROM t1");
 
-  run("select ts > 0, c1 < 20 and c2 = 'qaz' from t1");
+  run("SELECT ts > 0, c1 < 20 and c2 = 'qaz' FROM t1");
 
-  run("select ts > 0, c1 between 10 and 20 and c2 = 'qaz' from t1");
+  run("SELECT ts > 0, c1 between 10 and 20 and c2 = 'qaz' FROM t1");
 }
 
 TEST_F(ParserSelectTest, condition) {
   useDb("root", "test");
 
-  run("select c1 from t1 where ts in (true, false)");
+  run("SELECT c1 FROM t1 where ts in (true, false)");
 
-  run("select * from t1 where c1 > 10 and c1 is not null");
+  run("SELECT * FROM t1 where c1 > 10 and c1 is not null");
 }
 
 TEST_F(ParserSelectTest, pseudoColumn) {
   useDb("root", "test");
 
-  run("select _wstartts, _wendts, count(*) from t1 interval(10s)");
+  run("SELECT _wstartts, _wendts, COUNT(*) FROM t1 INTERVAL(10s)");
 }
 
 TEST_F(ParserSelectTest, multiResFunc) {
   useDb("root", "test");
 
-  run("select last(*), first(*), last_row(*) from t1");
+  run("SELECT last(*), first(*), last_row(*) FROM t1");
 
-  run("select last(c1, c2), first(t1.*), last_row(c3) from t1");
+  run("SELECT last(c1, c2), first(t1.*), last_row(c3) FROM t1");
 
-  run("select last(t2.*), first(t1.c1, t2.*), last_row(t1.*, t2.*) from st1s1 t1, st1s2 t2 where t1.ts = t2.ts");
+  run("SELECT last(t2.*), first(t1.c1, t2.*), last_row(t1.*, t2.*) FROM st1s1 t1, st1s2 t2 where t1.ts = t2.ts");
 }
 
 TEST_F(ParserSelectTest, timelineFunc) {
   useDb("root", "test");
 
-  run("select last(*), first(*) from t1");
+  run("SELECT last(*), first(*) FROM t1");
 
-  run("select last(*), first(*) from t1 group by c1");
+  run("SELECT last(*), first(*) FROM t1 group by c1");
 
-  run("select last(*), first(*) from t1 interval(10s)");
+  run("SELECT last(*), first(*) FROM t1 INTERVAL(10s)");
 
-  run("select diff(c1) from t1");
+  run("SELECT diff(c1) FROM t1");
 }
 
 TEST_F(ParserSelectTest, clause) {
   useDb("root", "test");
 
   // group by clause
-  run("select count(*) cnt from t1 where c1 > 0");
+  run("SELECT COUNT(*) cnt FROM t1 where c1 > 0");
 
-  run("select count(*), c2 cnt from t1 where c1 > 0 group by c2");
+  run("SELECT COUNT(*), c2 cnt FROM t1 where c1 > 0 group by c2");
 
-  run("select count(*) cnt from t1 where c1 > 0 group by c2 having count(c1) > 10");
+  run("SELECT COUNT(*) cnt FROM t1 where c1 > 0 group by c2 having COUNT(c1) > 10");
 
-  run("select count(*), c1, c2 + 10, c1 + c2 cnt from t1 where c1 > 0 group by c2, c1");
+  run("SELECT COUNT(*), c1, c2 + 10, c1 + c2 cnt FROM t1 where c1 > 0 group by c2, c1");
 
-  run("select count(*), c1 + 10, c2 cnt from t1 where c1 > 0 group by c1 + 10, c2");
+  run("SELECT COUNT(*), c1 + 10, c2 cnt FROM t1 where c1 > 0 group by c1 + 10, c2");
 
   // order by clause
-  run("select count(*) cnt from t1 where c1 > 0 group by c2 order by cnt");
+  run("SELECT COUNT(*) cnt FROM t1 where c1 > 0 group by c2 order by cnt");
 
-  run("select count(*) cnt from t1 where c1 > 0 group by c2 order by 1");
+  run("SELECT COUNT(*) cnt FROM t1 where c1 > 0 group by c2 order by 1");
 
   // distinct clause
-  // run("select distinct c1, c2 from t1 where c1 > 0 order by c1");
+  // run("SELECT distinct c1, c2 FROM t1 where c1 > 0 order by c1");
 
-  // run("select distinct c1 + 10, c2 from t1 where c1 > 0 order by c1 + 10, c2");
+  // run("SELECT distinct c1 + 10, c2 FROM t1 where c1 > 0 order by c1 + 10, c2");
 
-  // run("select distinct c1 + 10 cc1, c2 cc2 from t1 where c1 > 0 order by cc1, c2");
+  // run("SELECT distinct c1 + 10 cc1, c2 cc2 FROM t1 where c1 > 0 order by cc1, c2");
 
-  // run("select distinct count(c2) from t1 where c1 > 0 group by c1 order by count(c2)");
+  // run("SELECT distinct COUNT(c2) FROM t1 where c1 > 0 group by c1 order by COUNT(c2)");
 }
 
-TEST_F(ParserSelectTest, window) {
+// INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)] [FILL(fill_mod_and_val)]
+// fill_mod_and_val = { NONE | PREV | NULL | LINEAR | NEXT | value_mod }
+// value_mod = VALUE , val ...
+TEST_F(ParserSelectTest, interval) {
   useDb("root", "test");
-
-  run("select count(*) from t1 interval(10s)");
+  // INTERVAL(interval_val)
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s)");
+  // INTERVAL(interval_val, interval_offset)
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s, 5s)");
+  // INTERVAL(interval_val, interval_offset) SLIDING (sliding_val)
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s, 5s) SLIDING(7s)");
+  // INTERVAL(interval_val) FILL(NONE)
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s) FILL(NONE)");
 }
 
 TEST_F(ParserSelectTest, semanticError) {
   useDb("root", "test");
 
   // TSDB_CODE_PAR_INVALID_COLUMN
-  run("select c1, cc1 from t1", TSDB_CODE_PAR_INVALID_COLUMN, PARSER_STAGE_TRANSLATE);
+  run("SELECT c1, cc1 FROM t1", TSDB_CODE_PAR_INVALID_COLUMN, PARSER_STAGE_TRANSLATE);
 
-  run("select t1.c1, t1.cc1 from t1", TSDB_CODE_PAR_INVALID_COLUMN, PARSER_STAGE_TRANSLATE);
+  run("SELECT t1.c1, t1.cc1 FROM t1", TSDB_CODE_PAR_INVALID_COLUMN, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_TABLE_NOT_EXIST
-  run("select * from t10", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
+  run("SELECT * FROM t10", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
 
-  run("select * from test.t10", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
+  run("SELECT * FROM test.t10", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
 
-  run("select t2.c1 from t1", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
+  run("SELECT t2.c1 FROM t1", TSDB_CODE_PAR_TABLE_NOT_EXIST, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_AMBIGUOUS_COLUMN
-  run("select c2 from t1 tt1, t1 tt2 where tt1.c1 = tt2.c1", TSDB_CODE_PAR_AMBIGUOUS_COLUMN, PARSER_STAGE_TRANSLATE);
+  run("SELECT c2 FROM t1 tt1, t1 tt2 where tt1.c1 = tt2.c1", TSDB_CODE_PAR_AMBIGUOUS_COLUMN, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_WRONG_VALUE_TYPE
-  run("select timestamp '2010' from t1", TSDB_CODE_PAR_WRONG_VALUE_TYPE, PARSER_STAGE_TRANSLATE);
+  run("SELECT timestamp '2010' FROM t1", TSDB_CODE_PAR_WRONG_VALUE_TYPE, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION
-  run("select c2 from t1 tt1 join t1 tt2 on count(*) > 0", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION,
+  run("SELECT c2 FROM t1 tt1 join t1 tt2 on COUNT(*) > 0", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION,
       PARSER_STAGE_TRANSLATE);
 
-  run("select c2 from t1 where count(*) > 0", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION, PARSER_STAGE_TRANSLATE);
+  run("SELECT c2 FROM t1 where COUNT(*) > 0", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION, PARSER_STAGE_TRANSLATE);
 
-  run("select c2 from t1 group by count(*)", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION, PARSER_STAGE_TRANSLATE);
+  run("SELECT c2 FROM t1 group by COUNT(*)", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT
-  run("select c2 from t1 order by 0", TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT, PARSER_STAGE_TRANSLATE);
+  run("SELECT c2 FROM t1 order by 0", TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT, PARSER_STAGE_TRANSLATE);
 
-  run("select c2 from t1 order by 2", TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT, PARSER_STAGE_TRANSLATE);
+  run("SELECT c2 FROM t1 order by 2", TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION
-  run("select count(*) cnt from t1 having c1 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION, PARSER_STAGE_TRANSLATE);
+  run("SELECT COUNT(*) cnt FROM t1 having c1 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION, PARSER_STAGE_TRANSLATE);
 
-  run("select count(*) cnt from t1 group by c2 having c1 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
+  run("SELECT COUNT(*) cnt FROM t1 group by c2 having c1 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 
-  run("select count(*), c1 cnt from t1 group by c2 having c2 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
+  run("SELECT COUNT(*), c1 cnt FROM t1 group by c2 having c2 > 0", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 
-  run("select count(*) cnt from t1 group by c2 having c2 > 0 order by c1", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
+  run("SELECT COUNT(*) cnt FROM t1 group by c2 having c2 > 0 order by c1", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_NOT_SINGLE_GROUP
-  run("select count(*), c1 from t1", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
+  run("SELECT COUNT(*), c1 FROM t1", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
 
-  run("select count(*) from t1 order by c1", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
+  run("SELECT COUNT(*) FROM t1 order by c1", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
 
-  run("select c1 from t1 order by count(*)", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
+  run("SELECT c1 FROM t1 order by COUNT(*)", TSDB_CODE_PAR_NOT_SINGLE_GROUP, PARSER_STAGE_TRANSLATE);
 
   // TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION
-  run("select distinct c1, c2 from t1 where c1 > 0 order by ts", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
+  run("SELECT distinct c1, c2 FROM t1 where c1 > 0 order by ts", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 
-  run("select distinct c1 from t1 where c1 > 0 order by count(c2)", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
+  run("SELECT distinct c1 FROM t1 where c1 > 0 order by COUNT(c2)", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 
-  run("select distinct c2 from t1 where c1 > 0 order by count(c2)", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
+  run("SELECT distinct c2 FROM t1 where c1 > 0 order by COUNT(c2)", TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,
       PARSER_STAGE_TRANSLATE);
 }
 

source/libs/planner/src/planLogicCreater.c

@@ -274,7 +274,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
   }
 
   if (TSDB_CODE_SUCCESS == code) {
-    code = nodesCollectFuncs(pSelect, fmIsScanPseudoColumnFunc, &pScan->pScanPseudoCols);
+    code = nodesCollectFuncs(pSelect, SQL_CLAUSE_FROM, fmIsScanPseudoColumnFunc, &pScan->pScanPseudoCols);
   }
 
   pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->pMeta);
@@ -440,7 +440,7 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
   }
 
   if (TSDB_CODE_SUCCESS == code && pSelect->hasAggFuncs) {
-    code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pAgg->pAggFuncs);
+    code = nodesCollectFuncs(pSelect, SQL_CLAUSE_GROUP_BY, fmIsAggFunc, &pAgg->pAggFuncs);
   }
 
   // rewrite the expression in subsequent clauses
@@ -474,7 +474,7 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
 
 static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SWindowLogicNode* pWindow,
                                              SLogicNode** pLogicNode) {
-  int32_t code = nodesCollectFuncs(pSelect, fmIsWindowClauseFunc, &pWindow->pFuncs);
+  int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_WINDOW, fmIsWindowClauseFunc, &pWindow->pFuncs);
 
   if (pCxt->pPlanCxt->streamQuery) {
     pWindow->triggerType = pCxt->pPlanCxt->triggerType;
@@ -559,14 +559,6 @@ static int32_t createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SInterva
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  if (NULL != pInterval->pFill) {
-    pWindow->pFill = nodesCloneNode(pInterval->pFill);
-    if (NULL == pWindow->pFill) {
-      nodesDestroyNode(pWindow);
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-  }
-
   return createWindowLogicNodeFinalize(pCxt, pSelect, pWindow, pLogicNode);
 }
 
@@ -589,6 +581,37 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
   return TSDB_CODE_FAILED;
 }
 
+static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
+  if (NULL == pSelect->pWindow || QUERY_NODE_INTERVAL_WINDOW != nodeType(pSelect->pWindow) ||
+      NULL == ((SIntervalWindowNode*)pSelect->pWindow)->pFill) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SFillNode* pFillNode = (SFillNode*)(((SIntervalWindowNode*)pSelect->pWindow)->pFill);
+
+  SFillLogicNode* pFill = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_FILL);
+  if (NULL == pFill) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_WINDOW, NULL, COLLECT_COL_TYPE_ALL, &pFill->node.pTargets);
+
+  pFill->mode = pFillNode->mode;
+  pFill->pValues = nodesCloneNode(pFillNode->pValues);
+  pFill->pWStartTs = nodesCloneNode(pFillNode->pWStartTs);
+  if ((NULL != pFillNode->pValues && NULL == pFill->pValues) || NULL == pFill->pWStartTs) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pLogicNode = (SLogicNode*)pFill;
+  } else {
+    nodesDestroyNode(pFill);
+  }
+
+  return code;
+}
+
 static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
   if (NULL == pSelect->pOrderByList) {
     return TSDB_CODE_SUCCESS;
@@ -753,6 +776,9 @@ static int32_t createSelectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
   if (TSDB_CODE_SUCCESS == code) {
     code = createChildLogicNode(pCxt, pSelect, createWindowLogicNode, &pRoot);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = createChildLogicNode(pCxt, pSelect, createFillLogicNode, &pRoot);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = createChildLogicNode(pCxt, pSelect, createAggLogicNode, &pRoot);
   }

source/libs/planner/src/planPhysiCreater.c

@@ -17,8 +17,8 @@
 
 #include "catalog.h"
 #include "functionMgt.h"
-#include "tglobal.h"
 #include "systable.h"
+#include "tglobal.h"
 
 typedef struct SSlotIdInfo {
   int16_t slotId;
@@ -880,12 +880,6 @@ static int32_t createIntervalPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChil
   pInterval->intervalUnit = pWindowLogicNode->intervalUnit;
   pInterval->slidingUnit = pWindowLogicNode->slidingUnit;
 
-  pInterval->pFill = nodesCloneNode(pWindowLogicNode->pFill);
-  if (NULL != pWindowLogicNode->pFill && NULL == pInterval->pFill) {
-    nodesDestroyNode(pInterval);
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
   return createWindowPhysiNodeFinalize(pCxt, pChildren, &pInterval->window, pWindowLogicNode, pPhyNode);
 }
 
@@ -1035,6 +1029,43 @@ static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChi
   return code;
 }
 
+static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SFillLogicNode* pFillNode,
+                                   SPhysiNode** pPhyNode) {
+  SFillPhysiNode* pFill = (SFillPhysiNode*)makePhysiNode(pCxt, getPrecision(pChildren), (SLogicNode*)pFillNode,
+                                                         QUERY_NODE_PHYSICAL_PLAN_FILL);
+  if (NULL == pFill) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  SDataBlockDescNode* pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
+  int32_t code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pFillNode->node.pTargets, &pFill->pTargets);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = addDataBlockSlots(pCxt, pFill->pTargets, pFill->node.pOutputDataBlockDesc);
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    pFill->pWStartTs = nodesCloneNode(pFillNode->pWStartTs);
+    if (NULL == pFill->pWStartTs) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code && NULL != pFillNode->pValues) {
+    pFill->pValues = nodesCloneNode(pFillNode->pValues);
+    if (NULL == pFill->pValues) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pPhyNode = (SPhysiNode*)pFill;
+  } else {
+    nodesDestroyNode(pFill);
+  }
+
+  return code;
+}
+
 static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode, SSubplan* pSubplan,
                                  SNodeList* pChildren, SPhysiNode** pPhyNode) {
   switch (nodeType(pLogicNode)) {
@@ -1054,6 +1085,8 @@ static int32_t doCreatePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode
       return createSortPhysiNode(pCxt, pChildren, (SSortLogicNode*)pLogicNode, pPhyNode);
     case QUERY_NODE_LOGIC_PLAN_PARTITION:
       return createPartitionPhysiNode(pCxt, pChildren, (SPartitionLogicNode*)pLogicNode, pPhyNode);
+    case QUERY_NODE_LOGIC_PLAN_FILL:
+      return createFillPhysiNode(pCxt, pChildren, (SFillLogicNode*)pLogicNode, pPhyNode);
     default:
       break;
   }

source/libs/planner/test/planIntervalTest.cpp

@@ -23,19 +23,19 @@ class PlanIntervalTest : public PlannerTestBase {};
 TEST_F(PlanIntervalTest, basic) {
   useDb("root", "test");
 
-  run("select count(*) from t1 interval(10s)");
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s)");
 }
 
 TEST_F(PlanIntervalTest, pseudoCol) {
   useDb("root", "test");
 
-  run("select _wstartts, _wduration, _wendts, count(*) from t1 interval(10s)");
+  run("SELECT _WSTARTTS, _WDURATION, _WENDTS, COUNT(*) FROM t1 INTERVAL(10s)");
 }
 
 TEST_F(PlanIntervalTest, fill) {
   useDb("root", "test");
 
-  run("select count(*) from t1 interval(10s) fill(linear)");
+  run("SELECT COUNT(*) FROM t1 INTERVAL(10s) FILL(LINEAR)");
 
-  run("select count(*), sum(c1) from t1 interval(10s) fill(value, 10, 20)");
+  run("SELECT COUNT(*), sum(c1) FROM t1 INTERVAL(10s) FILL(VALUE, 10, 20)");
 }