feat: the unique function and tail function are rewritten as the corresponding clauses

include/libs/function/functionMgt.h

@@ -185,6 +185,8 @@ bool fmIsUserDefinedFunc(int32_t funcId);
 bool fmIsDistExecFunc(int32_t funcId);
 bool fmIsForbidFillFunc(int32_t funcId);
 bool fmIsForbidStreamFunc(int32_t funcId);
+bool fmIsForbidWindowFunc(int32_t funcId);
+bool fmIsForbidGroupByFunc(int32_t funcId);
 bool fmIsIntervalInterpoFunc(int32_t funcId);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);

include/libs/nodes/querynodes.h

@@ -247,6 +247,8 @@ typedef struct SSelectStmt {
   bool        hasIndefiniteRowsFunc;
   bool        hasSelectFunc;
   bool        hasSelectValFunc;
+  bool        hasUniqueFunc;
+  bool        hasTailFunc;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

include/util/taoserror.h

@@ -658,6 +658,9 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG   TAOS_DEF_ERROR_CODE(0, 0x2656)
 #define TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC     TAOS_DEF_ERROR_CODE(0, 0x2657)
 #define TSDB_CODE_PAR_INVALID_WINDOW_PC         TAOS_DEF_ERROR_CODE(0, 0x2658)
+#define TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC   TAOS_DEF_ERROR_CODE(0, 0x2659)
+#define TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC   TAOS_DEF_ERROR_CODE(0, 0x265A)
+#define TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC TAOS_DEF_ERROR_CODE(0, 0x265B)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/libs/function/inc/functionMgtInt.h

@@ -44,6 +44,8 @@ extern "C" {
 #define FUNC_MGT_FORBID_FILL_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(15)
 #define FUNC_MGT_INTERVAL_INTERPO_FUNC  FUNC_MGT_FUNC_CLASSIFICATION_MASK(16)
 #define FUNC_MGT_FORBID_STREAM_FUNC     FUNC_MGT_FUNC_CLASSIFICATION_MASK(17)
+#define FUNC_MGT_FORBID_WINDOW_FUNC     FUNC_MGT_FUNC_CLASSIFICATION_MASK(18)
+#define FUNC_MGT_FORBID_GROUP_BY_FUNC   FUNC_MGT_FUNC_CLASSIFICATION_MASK(19)
 
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 

source/libs/function/src/builtins.c

@@ -1085,12 +1085,7 @@ static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
+  pFunc->node.resType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType;
-  if (QUERY_NODE_COLUMN != nodeType(pPara)) {
-    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, "The parameters of UNIQUE can only be columns");
-  }
-
-  pFunc->node.resType = ((SExprNode*)pPara)->resType;
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2114,7 +2109,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "tail",
     .type = FUNCTION_TYPE_TAIL,
-    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC,
     .translateFunc = translateTail,
     .getEnvFunc   = getTailFuncEnv,
     .initFunc     = tailFunctionSetup,
@@ -2124,7 +2119,8 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "unique",
     .type = FUNCTION_TYPE_UNIQUE,
-    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | 
+                      FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC,
     .translateFunc = translateUnique,
     .getEnvFunc   = getUniqueFuncEnv,
     .initFunc     = uniqueFunctionSetup,

source/libs/function/src/functionMgt.c

@@ -165,6 +165,10 @@ bool fmIsIntervalInterpoFunc(int32_t funcId) { return isSpecificClassifyFunc(fun
 
 bool fmIsForbidStreamFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_STREAM_FUNC); }
 
+bool fmIsForbidWindowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_WINDOW_FUNC); }
+
+bool fmIsForbidGroupByFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_GROUP_BY_FUNC); }
+
 void fmFuncMgtDestroy() {
   void* m = gFunMgtService.pFuncNameHashTable;
   if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {

source/libs/parser/src/parTranslater.c

@@ -1098,11 +1098,43 @@ static int32_t translateWindowPseudoColumnFunc(STranslateContext* pCxt, SFunctio
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateForbidWindowFunc(STranslateContext* pCxt, SFunctionNode* pFunc) {
+  if (!fmIsForbidWindowFunc(pFunc->funcId)) {
+    return TSDB_CODE_SUCCESS;
+  }
+  if (NULL != pCxt->pCurrSelectStmt->pWindow) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC, pFunc->functionName);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateForbidStreamFunc(STranslateContext* pCxt, SFunctionNode* pFunc) {
+  if (!fmIsForbidStreamFunc(pFunc->funcId)) {
+    return TSDB_CODE_SUCCESS;
+  }
+  if (pCxt->createStream) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC, pFunc->functionName);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateForbidGroupByFunc(STranslateContext* pCxt, SFunctionNode* pFunc) {
+  if (!fmIsForbidGroupByFunc(pFunc->funcId)) {
+    return TSDB_CODE_SUCCESS;
+  }
+  if (NULL != pCxt->pCurrSelectStmt->pGroupByList) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC, pFunc->functionName);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 static void setFuncClassification(SSelectStmt* pSelect, SFunctionNode* pFunc) {
   if (NULL != pSelect) {
     pSelect->hasAggFuncs = pSelect->hasAggFuncs ? true : fmIsAggFunc(pFunc->funcId);
     pSelect->hasRepeatScanFuncs = pSelect->hasRepeatScanFuncs ? true : fmIsRepeatScanFunc(pFunc->funcId);
     pSelect->hasIndefiniteRowsFunc = pSelect->hasIndefiniteRowsFunc ? true : fmIsIndefiniteRowsFunc(pFunc->funcId);
+    pSelect->hasUniqueFunc = pSelect->hasUniqueFunc ? true : (FUNCTION_TYPE_UNIQUE == pFunc->funcType);
+    pSelect->hasTailFunc = pSelect->hasTailFunc ? true : (FUNCTION_TYPE_TAIL == pFunc->funcType);
   }
 }
 
@@ -1130,6 +1162,15 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
   if (TSDB_CODE_SUCCESS == pCxt->errCode) {
     pCxt->errCode = translateWindowPseudoColumnFunc(pCxt, pFunc);
   }
+  if (TSDB_CODE_SUCCESS == pCxt->errCode) {
+    pCxt->errCode = translateForbidWindowFunc(pCxt, pFunc);
+  }
+  if (TSDB_CODE_SUCCESS == pCxt->errCode) {
+    pCxt->errCode = translateForbidStreamFunc(pCxt, pFunc);
+  }
+  if (TSDB_CODE_SUCCESS == pCxt->errCode) {
+    pCxt->errCode = translateForbidGroupByFunc(pCxt, pFunc);
+  }
   if (TSDB_CODE_SUCCESS == pCxt->errCode) {
     setFuncClassification(pCxt->pCurrSelectStmt, pFunc);
   }
@@ -2116,6 +2157,136 @@ static int32_t rewriteTimelineFunc(STranslateContext* pCxt, SSelectStmt* pSelect
   return pCxt->errCode;
 }
 
+typedef struct SRwriteUniqueCxt {
+  STranslateContext* pTranslateCxt;
+  SNode*             pExpr;
+} SRwriteUniqueCxt;
+
+static EDealRes rewriteSeletcValueFunc(STranslateContext* pCxt, SNode** pNode) {
+  SFunctionNode* pFirst = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFirst) {
+    pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
+    return DEAL_RES_ERROR;
+  }
+  strcpy(pFirst->functionName, "first");
+  TSWAP(pFirst->pParameterList, ((SFunctionNode*)*pNode)->pParameterList);
+  nodesDestroyNode(*pNode);
+  *pNode = (SNode*)pFirst;
+  pCxt->errCode = fmGetFuncInfo(pFirst, pCxt->msgBuf.buf, pCxt->msgBuf.len);
+  pCxt->pCurrSelectStmt->hasAggFuncs = true;
+  return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR;
+}
+
+static EDealRes rewriteUniqueFunc(SNode** pNode, void* pContext) {
+  SRwriteUniqueCxt* pCxt = pContext;
+  if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
+    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
+    if (FUNCTION_TYPE_UNIQUE == pFunc->funcType) {
+      SNode* pExpr = nodesListGetNode(pFunc->pParameterList, 0);
+      NODES_CLEAR_LIST(pFunc->pParameterList);
+      nodesDestroyNode(*pNode);
+      *pNode = pExpr;
+      pCxt->pExpr = pExpr;
+      return DEAL_RES_IGNORE_CHILD;
+    } else if (FUNCTION_TYPE_SELECT_VALUE == pFunc->funcType) {
+      return rewriteSeletcValueFunc(pCxt->pTranslateCxt, pNode);
+    }
+  }
+  return DEAL_RES_CONTINUE;
+}
+
+static SNode* createGroupingSet(SNode* pExpr) {
+  SGroupingSetNode* pGroupingSet = (SGroupingSetNode*)nodesMakeNode(QUERY_NODE_GROUPING_SET);
+  if (NULL == pGroupingSet) {
+    return NULL;
+  }
+  pGroupingSet->groupingSetType = GP_TYPE_NORMAL;
+  if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pGroupingSet->pParameterList, nodesCloneNode(pExpr))) {
+    nodesDestroyNode((SNode*)pGroupingSet);
+    return NULL;
+  }
+  return (SNode*)pGroupingSet;
+}
+
+static int32_t rewriteUniqueStmt(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  if (!pSelect->hasUniqueFunc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SRwriteUniqueCxt cxt = {.pTranslateCxt = pCxt, .pExpr = NULL};
+  nodesRewriteExprs(pSelect->pProjectionList, rewriteUniqueFunc, &cxt);
+  if (TSDB_CODE_SUCCESS == cxt.pTranslateCxt->errCode) {
+    cxt.pTranslateCxt->errCode = nodesListMakeStrictAppend(&pSelect->pGroupByList, createGroupingSet(cxt.pExpr));
+  }
+  pSelect->hasIndefiniteRowsFunc = false;
+  return cxt.pTranslateCxt->errCode;
+}
+
+typedef struct SRwriteTailCxt {
+  STranslateContext* pTranslateCxt;
+  int64_t            limit;
+  int64_t            offset;
+} SRwriteTailCxt;
+
+static EDealRes rewriteTailFunc(SNode** pNode, void* pContext) {
+  SRwriteTailCxt* pCxt = pContext;
+  if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
+    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
+    if (FUNCTION_TYPE_TAIL == pFunc->funcType) {
+      pCxt->limit = ((SValueNode*)nodesListGetNode(pFunc->pParameterList, 1))->datum.i;
+      if (3 == LIST_LENGTH(pFunc->pParameterList)) {
+        pCxt->offset = ((SValueNode*)nodesListGetNode(pFunc->pParameterList, 2))->datum.i;
+      }
+      SNode* pExpr = nodesListGetNode(pFunc->pParameterList, 0);
+      NODES_CLEAR_LIST(pFunc->pParameterList);
+      nodesDestroyNode(*pNode);
+      *pNode = pExpr;
+      return DEAL_RES_IGNORE_CHILD;
+    }
+  }
+  return DEAL_RES_CONTINUE;
+}
+
+static int32_t createLimieNode(SRwriteTailCxt* pCxt, SLimitNode** pOutput) {
+  *pOutput = (SLimitNode*)nodesMakeNode(QUERY_NODE_LIMIT);
+  if (NULL == *pOutput) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  (*pOutput)->limit = pCxt->limit;
+  (*pOutput)->offset = pCxt->offset;
+  return TSDB_CODE_SUCCESS;
+}
+
+static SNode* createOrderByExpr(STranslateContext* pCxt) {
+  SOrderByExprNode* pOrder = (SOrderByExprNode*)nodesMakeNode(QUERY_NODE_ORDER_BY_EXPR);
+  if (NULL == pOrder) {
+    return NULL;
+  }
+  pCxt->errCode = createPrimaryKeyCol(pCxt, &pOrder->pExpr);
+  if (TSDB_CODE_SUCCESS != pCxt->errCode) {
+    nodesDestroyNode((SNode*)pOrder);
+    return NULL;
+  }
+  pOrder->order = ORDER_DESC;
+  pOrder->nullOrder = NULL_ORDER_FIRST;
+  return (SNode*)pOrder;
+}
+
+static int32_t rewriteTailStmt(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  if (!pSelect->hasTailFunc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  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;
+  return code;
+}
+
 static int32_t translateSelect(STranslateContext* pCxt, SSelectStmt* pSelect) {
   pCxt->pCurrSelectStmt = pSelect;
   int32_t code = translateFrom(pCxt, pSelect->pFromTable);
@@ -2147,6 +2318,12 @@ static int32_t translateSelect(STranslateContext* pCxt, SSelectStmt* pSelect) {
   if (TSDB_CODE_SUCCESS == code) {
     code = checkLimit(pCxt, pSelect);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteUniqueStmt(pCxt, pSelect);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteTailStmt(pCxt, pSelect);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = rewriteTimelineFunc(pCxt, pSelect);
   }

source/libs/parser/src/parUtil.c

@@ -185,9 +185,15 @@ static char* getSyntaxErrFormat(int32_t errCode) {
     case TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG:
       return "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes";
     case TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC:
-      return "%s function not allowed in fill query";
+      return "%s function does not supportted in fill query";
     case TSDB_CODE_PAR_INVALID_WINDOW_PC:
-      return "_WSTARTTS, _WENDTS and _WDURATION can only be used in window queries";
+      return "_WSTARTTS, _WENDTS and _WDURATION can only be used in window query";
+    case TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC:
+      return "%s function does not supportted in time window query";
+    case TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC:
+      return "%s function does not supportted in stream query";
+    case TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC:
+      return "%s function does not supportted in group query";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/parser/test/parInitialCTest.cpp

@@ -528,6 +528,12 @@ TEST_F(ParserInitialCTest, createStream) {
   clearCreateStreamReq();
 }
 
+TEST_F(ParserInitialCTest, createStreamSemanticCheck) {
+  useDb("root", "test");
+
+  run("CREATE STREAM s1 AS SELECT PERCENTILE(c1, 30) FROM t1", TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC);
+}
+
 TEST_F(ParserInitialCTest, createTable) {
   useDb("root", "test");
 

source/libs/parser/test/parSelectTest.cpp

@@ -161,6 +161,40 @@ TEST_F(ParserSelectTest, useDefinedFunc) {
   run("SELECT udf2(c1) FROM t1 GROUP BY c2");
 }
 
+TEST_F(ParserSelectTest, uniqueFunc) {
+  useDb("root", "test");
+
+  run("SELECT UNIQUE(c1) FROM t1");
+
+  run("SELECT UNIQUE(c2 + 10) FROM t1 WHERE c1 > 10");
+
+  run("SELECT UNIQUE(c2 + 10), ts, c2 FROM t1 WHERE c1 > 10");
+}
+
+TEST_F(ParserSelectTest, uniqueFuncSemanticCheck) {
+  useDb("root", "test");
+
+  run("SELECT UNIQUE(c1) FROM t1 INTERVAL(10S)", TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC);
+
+  run("SELECT UNIQUE(c1) FROM t1 GROUP BY c2", TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC);
+}
+
+TEST_F(ParserSelectTest, tailFunc) {
+  useDb("root", "test");
+
+  run("SELECT TAIL(c1, 10) FROM t1");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10");
+}
+
+TEST_F(ParserSelectTest, tailFuncSemanticCheck) {
+  useDb("root", "test");
+
+  run("SELECT TAIL(c1, 10) FROM t1 INTERVAL(10S)", TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC);
+
+  run("SELECT TAIL(c1, 10) FROM t1 GROUP BY c2", TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC);
+}
+
 TEST_F(ParserSelectTest, groupBy) {
   useDb("root", "test");
 

source/libs/planner/src/planner.c

@@ -21,7 +21,7 @@
 static void dumpQueryPlan(SQueryPlan* pPlan) {
   char* pStr = NULL;
   nodesNodeToString((SNode*)pPlan, false, &pStr, NULL);
-  planDebugL("Query Plan: %s", pStr);
+  planDebugL("QID:0x%" PRIx64 " Query Plan: %s", pPlan->queryId, pStr);
   taosMemoryFree(pStr);
 }
 

source/libs/planner/test/planBasicTest.cpp

@@ -53,3 +53,21 @@ TEST_F(PlanBasicTest, func) {
 
   run("SELECT TOP(c1, 60) FROM t1");
 }
+
+TEST_F(PlanBasicTest, uniqueFunc) {
+  useDb("root", "test");
+
+  run("SELECT UNIQUE(c1) FROM t1");
+
+  run("SELECT UNIQUE(c2 + 10) FROM t1 WHERE c1 > 10");
+
+  run("SELECT UNIQUE(c2 + 10), ts, c2 FROM t1 WHERE c1 > 10");
+}
+
+TEST_F(PlanBasicTest, tailFunc) {
+  useDb("root", "test");
+
+  run("SELECT TAIL(c1, 10) FROM t1");
+
+  run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10");
+}