feat: tail function rewrite to statement

include/libs/nodes/plannodes.h

@@ -24,6 +24,8 @@ extern "C" {
 #include "querynodes.h"
 #include "tname.h"
 
+#define SLOT_NAME_LEN TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN
+
 typedef struct SLogicNode {
   ENodeType          type;
   SNodeList*         pTargets;  // SColumnNode
@@ -100,6 +102,7 @@ typedef struct SProjectLogicNode {
 typedef struct SIndefRowsFuncLogicNode {
   SLogicNode node;
   SNodeList* pFuncs;
+  bool       isTailFunc;
 } SIndefRowsFuncLogicNode;
 
 typedef struct SInterpFuncLogicNode {
@@ -184,6 +187,7 @@ typedef struct SFillLogicNode {
 typedef struct SSortLogicNode {
   SLogicNode node;
   SNodeList* pSortKeys;
+  bool       groupSort;
 } SSortLogicNode;
 
 typedef struct SPartitionLogicNode {
@@ -230,6 +234,7 @@ typedef struct SSlotDescNode {
   bool      reserve;
   bool      output;
   bool      tag;
+  char      name[SLOT_NAME_LEN];
 } SSlotDescNode;
 
 typedef struct SDataBlockDescNode {

include/libs/nodes/querynodes.h

@@ -259,6 +259,7 @@ typedef struct SSelectStmt {
   bool        hasTailFunc;
   bool        hasInterpFunc;
   bool        hasLastRowFunc;
+  bool        groupSort;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -436,6 +436,7 @@ static int32_t logicFillCopy(const SFillLogicNode* pSrc, SFillLogicNode* pDst) {
 static int32_t logicSortCopy(const SSortLogicNode* pSrc, SSortLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   CLONE_NODE_LIST_FIELD(pSortKeys);
+  COPY_SCALAR_FIELD(groupSort);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/nodes/src/nodesCodeFuncs.c

@@ -234,6 +234,8 @@ const char* nodesNodeName(ENodeType type) {
       return "PhysiMerge";
     case QUERY_NODE_PHYSICAL_PLAN_SORT:
       return "PhysiSort";
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT:
+      return "PhysiGroupSort";
     case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
       return "PhysiHashInterval";
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL:
@@ -3369,6 +3371,7 @@ static const char* jkSlotDescSlotId = "SlotId";
 static const char* jkSlotDescDataType = "DataType";
 static const char* jkSlotDescReserve = "Reserve";
 static const char* jkSlotDescOutput = "Output";
+static const char* jkSlotDescName = "Name";
 
 static int32_t slotDescNodeToJson(const void* pObj, SJson* pJson) {
   const SSlotDescNode* pNode = (const SSlotDescNode*)pObj;
@@ -3383,6 +3386,9 @@ static int32_t slotDescNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddBoolToObject(pJson, jkSlotDescOutput, pNode->output);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddStringToObject(pJson, jkSlotDescName, pNode->name);
+  }
 
   return code;
 }
@@ -3400,6 +3406,9 @@ static int32_t jsonToSlotDescNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBoolValue(pJson, jkSlotDescOutput, &pNode->output);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetStringValue(pJson, jkSlotDescName, pNode->name);
+  }
 
   return code;
 }
@@ -4137,6 +4146,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
     case QUERY_NODE_PHYSICAL_PLAN_MERGE:
       return physiMergeNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_SORT:
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT:
       return physiSortNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL:
@@ -4280,6 +4290,7 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
     case QUERY_NODE_PHYSICAL_PLAN_MERGE:
       return jsonToPhysiMergeNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_SORT:
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT:
       return jsonToPhysiSortNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL:

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -500,7 +500,8 @@ static EDealRes dispatchPhysiPlan(SNode* pNode, ETraversalOrder order, FNodeWalk
       }
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_SORT: {
+    case QUERY_NODE_PHYSICAL_PLAN_SORT:
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT: {
       SSortPhysiNode* pSort = (SSortPhysiNode*)pNode;
       res = walkPhysiNode((SPhysiNode*)pNode, order, walker, pContext);
       if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {

source/libs/nodes/src/nodesUtilFuncs.c

@@ -288,6 +288,8 @@ SNode* nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SMergePhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_SORT:
       return makeNode(type, sizeof(SSortPhysiNode));
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT:
+      return makeNode(type, sizeof(SGroupSortPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
       return makeNode(type, sizeof(SIntervalPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL:
@@ -850,7 +852,8 @@ void nodesDestroyNode(SNode* pNode) {
       nodesDestroyList(pPhyNode->pTargets);
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_SORT: {
+    case QUERY_NODE_PHYSICAL_PLAN_SORT:
+    case QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT: {
       SSortPhysiNode* pPhyNode = (SSortPhysiNode*)pNode;
       destroyPhysiNode((SPhysiNode*)pPhyNode);
       nodesDestroyList(pPhyNode->pExprs);

source/libs/parser/src/parTranslater.c

@@ -2507,6 +2507,28 @@ static SNode* createOrderByExpr(STranslateContext* pCxt) {
   return (SNode*)pOrder;
 }
 
+static int32_t rewriteTailStmtInplace(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  SRwriteTailCxt cxt = {.pTranslateCxt = pCxt, .limit = -1, .offset = -1};
+  nodesRewriteExprs(pSelect->pProjectionList, rewriteTailFunc, &cxt);
+  int32_t code = nodesListMakeStrictAppend(&pSelect->pOrderByList, createOrderByExpr(pCxt));
+  if (TSDB_CODE_SUCCESS == code) {
+    code = createLimieNode(&cxt, &pSelect->pLimit);
+  }
+  pSelect->hasIndefiniteRowsFunc = false;
+  pSelect->groupSort = (NULL != pSelect->pPartitionByList);
+  return code;
+}
+
+static int32_t rewriteTailStmtSubquery(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  SSelectStmt* pSubquery = (SSelectStmt*)nodesMakeNode(QUERY_NODE_SELECT_STMT);
+  if (NULL == pSubquery) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  TSWAP(pSubquery->pProjectionList, pSelect->pProjectionList);
+
+  return TSDB_CODE_PAR_INTERNAL_ERROR;
+}
+
 /* case 1:
  * in:  select tail(expr, k, f) from t where_clause
  * out: select expr from t where_clause order by _rowts desc limit k offset f
@@ -2523,21 +2545,20 @@ static SNode* createOrderByExpr(STranslateContext* pCxt) {
  *
  * case 4:
  * in:  select tail(expr, k, f) from t where_clause partition_by_clause limit_clause
- * out:
+ * out: select expr from (
+ *        select expr, part_key_list from t where_clause partition_by_clause sort by _rowts desc limit k offset f
+ *      ) partition_by_clause limit_clause
  */
 static int32_t rewriteTailStmt(STranslateContext* pCxt, SSelectStmt* pSelect) {
   if (!pSelect->hasTailFunc) {
     return TSDB_CODE_SUCCESS;
   }
 
-  SRwriteTailCxt cxt = {.pTranslateCxt = pCxt, .limit = -1, .offset = -1};
+  if (NULL == pSelect->pOrderByList && NULL == pSelect->pLimit && NULL == pSelect->pSlimit) {
-  nodesRewriteExprs(pSelect->pProjectionList, rewriteTailFunc, &cxt);
+    return rewriteTailStmtInplace(pCxt, pSelect);
-  int32_t code = nodesListMakeStrictAppend(&pSelect->pOrderByList, createOrderByExpr(pCxt));
+  } else {
-  if (TSDB_CODE_SUCCESS == code) {
+    return rewriteTailStmtSubquery(pCxt, pSelect);
-    code = createLimieNode(&cxt, &pSelect->pLimit);
   }
-  pSelect->hasIndefiniteRowsFunc = false;
-  return code;
 }
 
 typedef struct SReplaceOrderByAliasCxt {

source/libs/planner/src/planLogicCreater.c

@@ -507,6 +507,8 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
+  pIdfRowsFunc->isTailFunc = pSelect->hasTailFunc;
+
   // indefinite rows functions and _select_values functions
   int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
   if (TSDB_CODE_SUCCESS == code) {
@@ -733,6 +735,8 @@ static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
+  pSort->groupSort = pSelect->groupSort;
+
   int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_ORDER_BY, NULL, COLLECT_COL_TYPE_ALL, &pSort->node.pTargets);
   if (TSDB_CODE_SUCCESS == code && NULL == pSort->node.pTargets) {
     code = nodesListMakeStrictAppend(&pSort->node.pTargets,

source/libs/planner/src/planOptimizer.c

@@ -1183,6 +1183,143 @@ static int32_t eliminateProjOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLog
   return eliminateProjOptimizeImpl(pCxt, pLogicSubplan, pProjectNode);
 }
 
+static bool rewriteTailOptMayBeOptimized(SLogicNode* pNode) {
+  return QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC == nodeType(pNode) && ((SIndefRowsFuncLogicNode*)pNode)->isTailFunc;
+}
+
+static SNode* rewriteTailOptCreateOrderByExpr(SNode* pSortKey) {
+  SOrderByExprNode* pOrder = (SOrderByExprNode*)nodesMakeNode(QUERY_NODE_ORDER_BY_EXPR);
+  if (NULL == pOrder) {
+    return NULL;
+  }
+  pOrder->order = ORDER_DESC;
+  pOrder->pExpr = nodesCloneNode(pSortKey);
+  if (NULL == pOrder->pExpr) {
+    nodesDestroyNode((SNode*)pOrder);
+    return NULL;
+  }
+  return (SNode*)pOrder;
+}
+
+static int32_t rewriteTailOptCreateLimit(SNode* pLimit, SNode* pOffset, SNode** pOutput) {
+  SLimitNode* pLimitNode = (SLimitNode*)nodesMakeNode(QUERY_NODE_LIMIT);
+  if (NULL == pLimitNode) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  pLimitNode->limit = NULL == pLimit ? -1 : ((SValueNode*)pLimit)->datum.i;
+  pLimitNode->offset = NULL == pOffset ? -1 : ((SValueNode*)pOffset)->datum.i;
+  *pOutput = (SNode*)pLimitNode;
+  return TSDB_CODE_SUCCESS;
+}
+
+static bool rewriteTailOptNeedGroupSort(SIndefRowsFuncLogicNode* pIndef) {
+  return 1 == LIST_LENGTH(pIndef->node.pChildren) &&
+         QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(nodesListGetNode(pIndef->node.pChildren, 0));
+}
+
+static int32_t rewriteTailOptCreateSort(SIndefRowsFuncLogicNode* pIndef, SLogicNode** pOutput) {
+  SSortLogicNode* pSort = (SSortLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_SORT);
+  if (NULL == pSort) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  pSort->groupSort = rewriteTailOptNeedGroupSort(pIndef);
+  TSWAP(pSort->node.pChildren, pIndef->node.pChildren);
+  pSort->node.precision = pIndef->node.precision;
+
+  // tail(expr, [limit, offset,] _rowts)
+  SFunctionNode* pTail = (SFunctionNode*)nodesListGetNode(pIndef->pFuncs, 0);
+  int32_t        limitIndex = LIST_LENGTH(pTail->pParameterList) > 2 ? 1 : -1;
+  int32_t        offsetIndex = LIST_LENGTH(pTail->pParameterList) > 3 ? 2 : -1;
+  int32_t        rowtsIndex = LIST_LENGTH(pTail->pParameterList) - 1;
+
+  int32_t code = nodesListMakeStrictAppend(
+      &pSort->pSortKeys, rewriteTailOptCreateOrderByExpr(nodesListGetNode(pTail->pParameterList, rowtsIndex)));
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteTailOptCreateLimit(limitIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, limitIndex),
+                                     offsetIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, offsetIndex),
+                                     &pSort->node.pLimit);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    pSort->node.pTargets = nodesCloneList(((SLogicNode*)nodesListGetNode(pSort->node.pChildren, 0))->pTargets);
+    if (NULL == pSort->node.pTargets) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pOutput = (SLogicNode*)pSort;
+  } else {
+    nodesDestroyNode((SNode*)pSort);
+  }
+
+  return code;
+}
+
+static SNode* rewriteTailOptCreateProjectExpr(SFunctionNode* pTail) {
+  SNode* pExpr = nodesCloneNode(nodesListGetNode(pTail->pParameterList, 0));
+  if (NULL == pExpr) {
+    return NULL;
+  }
+  strcpy(((SExprNode*)pExpr)->aliasName, pTail->node.aliasName);
+  return pExpr;
+}
+
+static int32_t rewriteTailOptCreateProject(SIndefRowsFuncLogicNode* pIndef, SLogicNode** pOutput) {
+  SProjectLogicNode* pProject = (SProjectLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_PROJECT);
+  if (NULL == pProject) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  TSWAP(pProject->node.pTargets, pIndef->node.pTargets);
+  pProject->node.precision = pIndef->node.precision;
+
+  int32_t code = nodesListMakeStrictAppend(
+      &pProject->pProjections, rewriteTailOptCreateProjectExpr((SFunctionNode*)nodesListGetNode(pIndef->pFuncs, 0)));
+  if (TSDB_CODE_SUCCESS == code) {
+    *pOutput = (SLogicNode*)pProject;
+  } else {
+    nodesDestroyNode((SNode*)pProject);
+  }
+  return code;
+}
+
+static int32_t rewriteTailOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan,
+                                       SIndefRowsFuncLogicNode* pIndef) {
+  SLogicNode* pSort = NULL;
+  SLogicNode* pProject = NULL;
+  int32_t     code = rewriteTailOptCreateSort(pIndef, &pSort);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteTailOptCreateProject(pIndef, &pProject);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesListMakeAppend(&pProject->pChildren, (SNode*)pSort);
+    pSort->pParent = pProject;
+    pSort = NULL;
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pIndef, pProject);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    nodesDestroyNode((SNode*)pIndef);
+  } else {
+    nodesDestroyNode((SNode*)pSort);
+    nodesDestroyNode((SNode*)pProject);
+  }
+  return code;
+}
+
+static int32_t rewriteTailOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
+  SIndefRowsFuncLogicNode* pIndef =
+      (SIndefRowsFuncLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, rewriteTailOptMayBeOptimized);
+
+  if (NULL == pIndef) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  return rewriteTailOptimizeImpl(pCxt, pLogicSubplan, pIndef);
+}
+
 // clang-format off
 static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "OptimizeScanData",  .optimizeFunc = osdOptimize},
@@ -1190,7 +1327,8 @@ static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "OrderByPrimaryKey", .optimizeFunc = opkOptimize},
   {.pName = "SmaIndex",          .optimizeFunc = smaOptimize},
   {.pName = "PartitionTags",     .optimizeFunc = partTagsOptimize},
-  {.pName = "EliminateProject",  .optimizeFunc = eliminateProjOptimize}
+  {.pName = "EliminateProject",  .optimizeFunc = eliminateProjOptimize},
+  {.pName = "RewriteTail",       .optimizeFunc = rewriteTailOptimize}
 };
 // clang-format on
 

source/libs/planner/src/planPhysiCreater.c

@@ -41,8 +41,12 @@ typedef struct SPhysiPlanContext {
 static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
   if (QUERY_NODE_COLUMN == nodeType(pNode)) {
     SColumnNode* pCol = (SColumnNode*)pNode;
-    if (NULL != pStmtName && '\0' != pStmtName[0]) {
+    if (NULL != pStmtName) {
-      return sprintf(pKey, "%s.%s", pStmtName, pCol->node.aliasName);
+      if ('\0' != pStmtName[0]) {
+        return sprintf(pKey, "%s.%s", pStmtName, pCol->node.aliasName);
+      } else {
+        return sprintf(pKey, "%s", pCol->node.aliasName);
+      }
     }
     if ('\0' == pCol->tableAlias[0]) {
       return sprintf(pKey, "%s", pCol->colName);
@@ -56,11 +60,13 @@ static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
   return sprintf(pKey, "%s", ((SExprNode*)pNode)->aliasName);
 }
 
-static SNode* createSlotDesc(SPhysiPlanContext* pCxt, const SNode* pNode, int16_t slotId, bool output, bool reserve) {
+static SNode* createSlotDesc(SPhysiPlanContext* pCxt, const char* pName, const SNode* pNode, int16_t slotId,
+                             bool output, bool reserve) {
   SSlotDescNode* pSlot = (SSlotDescNode*)nodesMakeNode(QUERY_NODE_SLOT_DESC);
   if (NULL == pSlot) {
     return NULL;
   }
+  strcpy(pSlot->name, pName);
   pSlot->slotId = slotId;
   pSlot->dataType = ((SExprNode*)pNode)->resType;
   pSlot->reserve = reserve;
@@ -99,10 +105,8 @@ static int32_t putSlotToHashImpl(int16_t dataBlockId, int16_t slotId, const char
   return taosHashPut(pHash, pName, len, &index, sizeof(SSlotIndex));
 }
 
-static int32_t putSlotToHash(int16_t dataBlockId, int16_t slotId, SNode* pNode, SHashObj* pHash) {
+static int32_t putSlotToHash(const char* pName, int16_t dataBlockId, int16_t slotId, SNode* pNode, SHashObj* pHash) {
-  char    name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN];
+  return putSlotToHashImpl(dataBlockId, slotId, pName, strlen(pName), pHash);
-  int32_t len = getSlotKey(pNode, NULL, name);
-  return putSlotToHashImpl(dataBlockId, slotId, name, len, pHash);
 }
 
 static int32_t createDataBlockDescHash(SPhysiPlanContext* pCxt, int32_t capacity, int16_t dataBlockId,
@@ -131,9 +135,11 @@ static int32_t buildDataBlockSlots(SPhysiPlanContext* pCxt, SNodeList* pList, SD
   int16_t slotId = 0;
   SNode*  pNode = NULL;
   FOREACH(pNode, pList) {
-    code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, pNode, slotId, true, false));
+    char name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN];
+    getSlotKey(pNode, NULL, name);
+    code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, name, pNode, slotId, true, false));
     if (TSDB_CODE_SUCCESS == code) {
-      code = putSlotToHash(pDataBlockDesc->dataBlockId, slotId, pNode, pHash);
+      code = putSlotToHash(name, pDataBlockDesc->dataBlockId, slotId, pNode, pHash);
     }
     if (TSDB_CODE_SUCCESS == code) {
       pDataBlockDesc->totalRowSize += ((SExprNode*)pNode)->resType.bytes;
@@ -196,7 +202,8 @@ static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList,
     int32_t     len = getSlotKey(pExpr, pStmtName, name);
     SSlotIndex* pIndex = taosHashGet(pHash, name, len);
     if (NULL == pIndex) {
-      code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, pExpr, nextSlotId, output, reserve));
+      code =
+          nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, name, pExpr, nextSlotId, output, reserve));
       if (TSDB_CODE_SUCCESS == code) {
         code = putSlotToHashImpl(pDataBlockDesc->dataBlockId, nextSlotId, name, len, pHash);
       }
@@ -1170,8 +1177,9 @@ static int32_t createWindowPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildr
 
 static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SSortLogicNode* pSortLogicNode,
                                    SPhysiNode** pPhyNode) {
-  SSortPhysiNode* pSort =
+  SSortPhysiNode* pSort = (SSortPhysiNode*)makePhysiNode(
-      (SSortPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pSortLogicNode, QUERY_NODE_PHYSICAL_PLAN_SORT);
+      pCxt, (SLogicNode*)pSortLogicNode,
+      pSortLogicNode->groupSort ? QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT : QUERY_NODE_PHYSICAL_PLAN_SORT);
   if (NULL == pSort) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -1185,7 +1193,7 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
   if (TSDB_CODE_SUCCESS == code && NULL != pPrecalcExprs) {
     code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pPrecalcExprs, &pSort->pExprs);
     if (TSDB_CODE_SUCCESS == code) {
-      code = addDataBlockSlots(pCxt, pSort->pExprs, pChildTupe);
+      code = pushdownDataBlockSlots(pCxt, pSort->pExprs, pChildTupe);
     }
   }
 

source/libs/planner/test/planBasicTest.cpp

@@ -67,6 +67,14 @@ TEST_F(PlanBasicTest, tailFunc) {
   run("SELECT TAIL(c1, 10) FROM t1");
 
   run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 PARTITION BY c1");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 ORDER BY 1");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 LIMIT 5");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 PARTITION BY c1 LIMIT 5");
 }
 
 TEST_F(PlanBasicTest, interpFunc) {

source/libs/planner/test/planTestMain.cpp

@@ -76,6 +76,7 @@ static void parseArg(int argc, char* argv[]) {
   static struct option long_options[] = {
     {"dump", optional_argument, NULL, 'd'},
     {"skipSql", required_argument, NULL, 's'},
+    {"limitSql", required_argument, NULL, 'i'},
     {"log", required_argument, NULL, 'l'},
     {0, 0, 0, 0}
   };
@@ -88,6 +89,9 @@ static void parseArg(int argc, char* argv[]) {
       case 's':
         setSkipSqlNum(optarg);
         break;
+      case 'i':
+        setLimitSqlNum(optarg);
+        break;
       case 'l':
         setLogLevel(optarg);
         break;

source/libs/planner/test/planTestUtil.cpp

@@ -51,6 +51,7 @@ enum DumpModule {
 
 DumpModule g_dumpModule = DUMP_MODULE_NOTHING;
 int32_t    g_skipSql = 0;
+int32_t    g_limitSql = 0;
 int32_t    g_logLevel = 131;
 
 void setDumpModule(const char* pModule) {
@@ -76,28 +77,33 @@ void setDumpModule(const char* pModule) {
 }
 
 void setSkipSqlNum(const char* pNum) { g_skipSql = stoi(pNum); }
-
+void setLimitSqlNum(const char* pNum) { g_limitSql = stoi(pNum); }
 void setLogLevel(const char* pLogLevel) { g_logLevel = stoi(pLogLevel); }
 
 int32_t getLogLevel() { return g_logLevel; }
 
 class PlannerTestBaseImpl {
  public:
-  PlannerTestBaseImpl() : sqlNo_(0) {}
+  PlannerTestBaseImpl() : sqlNo_(0), sqlNum_(0) {}
 
   void useDb(const string& user, const string& db) {
     caseEnv_.acctId_ = 0;
     caseEnv_.user_ = user;
     caseEnv_.db_ = db;
-    caseEnv_.nsql_ = g_skipSql;
+    caseEnv_.numOfSkipSql_ = g_skipSql;
+    caseEnv_.numOfLimitSql_ = g_limitSql;
   }
 
   void run(const string& sql) {
     ++sqlNo_;
-    if (caseEnv_.nsql_ > 0) {
+    if (caseEnv_.numOfSkipSql_ > 0) {
-      --(caseEnv_.nsql_);
+      --(caseEnv_.numOfSkipSql_);
       return;
     }
+    if (caseEnv_.numOfLimitSql_ > 0 && caseEnv_.numOfLimitSql_ == sqlNum_) {
+      return;
+    }
+    ++sqlNum_;
 
     reset();
     try {
@@ -134,7 +140,7 @@ class PlannerTestBaseImpl {
   }
 
   void prepare(const string& sql) {
-    if (caseEnv_.nsql_ > 0) {
+    if (caseEnv_.numOfSkipSql_ > 0) {
       return;
     }
 
@@ -148,7 +154,7 @@ class PlannerTestBaseImpl {
   }
 
   void bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx) {
-    if (caseEnv_.nsql_ > 0) {
+    if (caseEnv_.numOfSkipSql_ > 0) {
       return;
     }
 
@@ -161,8 +167,8 @@ class PlannerTestBaseImpl {
   }
 
   void exec() {
-    if (caseEnv_.nsql_ > 0) {
+    if (caseEnv_.numOfSkipSql_ > 0) {
-      --(caseEnv_.nsql_);
+      --(caseEnv_.numOfSkipSql_);
       return;
     }
 
@@ -197,9 +203,10 @@ class PlannerTestBaseImpl {
     int32_t acctId_;
     string  user_;
     string  db_;
-    int32_t nsql_;
+    int32_t numOfSkipSql_;
+    int32_t numOfLimitSql_;
 
-    caseEnv() : nsql_(0) {}
+    caseEnv() : numOfSkipSql_(0) {}
   };
 
   struct stmtEnv {
@@ -401,6 +408,7 @@ class PlannerTestBaseImpl {
   stmtEnv stmtEnv_;
   stmtRes res_;
   int32_t sqlNo_;
+  int32_t sqlNum_;
 };
 
 PlannerTestBase::PlannerTestBase() : impl_(new PlannerTestBaseImpl()) {}

source/libs/planner/test/planTestUtil.h

@@ -43,6 +43,7 @@ class PlannerTestBase : public testing::Test {
 
 extern void    setDumpModule(const char* pModule);
 extern void    setSkipSqlNum(const char* pNum);
+extern void    setLimitSqlNum(const char* pNum);
 extern void    setLogLevel(const char* pLogLevel);
 extern int32_t getLogLevel();