Merge branch '3.0' into feature/TD-13041

include/common/ttypes.h

@@ -30,7 +30,7 @@ typedef uint64_t TDRowVerT;
 typedef int16_t  col_id_t;
 typedef int8_t   col_type_t;
 typedef int32_t  col_bytes_t;
-typedef uint16_t schema_ver_t;
+typedef int32_t schema_ver_t;
 typedef int32_t  func_id_t;
 
 #pragma pack(push, 1)

include/libs/function/functionMgt.h

@@ -146,7 +146,8 @@ bool fmIsBuiltinFunc(const char* pFunc);
 
 bool fmIsAggFunc(int32_t funcId);
 bool fmIsScalarFunc(int32_t funcId);
-bool fmIsNonstandardSQLFunc(int32_t funcId);
+bool fmIsVectorFunc(int32_t funcId);
+bool fmIsIndefiniteRowsFunc(int32_t funcId);
 bool fmIsStringFunc(int32_t funcId);
 bool fmIsDatetimeFunc(int32_t funcId);
 bool fmIsSelectFunc(int32_t funcId);

include/libs/nodes/plannodes.h

@@ -54,6 +54,7 @@ typedef struct SScanLogicNode {
   int64_t            sliding;
   int8_t             intervalUnit;
   int8_t             slidingUnit;
+  SNode*             pTagCond;
 } SScanLogicNode;
 
 typedef struct SJoinLogicNode {
@@ -343,6 +344,7 @@ typedef struct SSubplan {
   SNodeList*     pParents;      // the data destination subplan, get data from current subplan
   SPhysiNode*    pNode;         // physical plan of current subplan
   SDataSinkNode* pDataSink;     // data of the subplan flow into the datasink
+  SNode*         pTagCond;
 } SSubplan;
 
 typedef enum EExplainMode { EXPLAIN_MODE_DISABLE = 1, EXPLAIN_MODE_STATIC, EXPLAIN_MODE_ANALYZE } EExplainMode;

include/libs/nodes/querynodes.h

@@ -241,7 +241,7 @@ typedef struct SSelectStmt {
   bool        isTimeOrderQuery;
   bool        hasAggFuncs;
   bool        hasRepeatScanFuncs;
-  bool        hasNonstdSQLFunc;
+  bool        hasIndefiniteRowsFunc;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

source/common/src/tdatablock.c

@@ -1538,7 +1538,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
   int32_t     msgLen = sizeof(SSubmitReq);
   int32_t     numOfBlks = 0;
   SRowBuilder rb = {0};
-  tdSRowInit(&rb, 0);  // TODO: use the latest version
+  tdSRowInit(&rb, pTSchema->version);  // TODO: use the latest version
 
   for (int32_t i = 0; i < sz; ++i) {
     SSDataBlock* pDataBlock = taosArrayGet(pDataBlocks, i);

source/common/src/tmsg.c

@@ -3817,7 +3817,7 @@ int tDecodeSVCreateStbReq(SDecoder *pCoder, SVCreateStbReq *pReq) {
 
 STSchema *tdGetSTSChemaFromSSChema(SSchema **pSchema, int32_t nCols) {
   STSchemaBuilder schemaBuilder = {0};
-  if (tdInitTSchemaBuilder(&schemaBuilder, 0) < 0) {
+  if (tdInitTSchemaBuilder(&schemaBuilder, 1) < 0) {
     return NULL;
   }
 

source/common/src/trow.c

@@ -924,7 +924,7 @@ void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag) {
   STSRowIter iter = {0};
   tdSTSRowIterInit(&iter, pSchema);
   tdSTSRowIterReset(&iter, row);
-  printf("%s >>>", tag);
+  printf("%s >>>type:%d,sver:%d ", tag, (int32_t)TD_ROW_TYPE(row), (int32_t)TD_ROW_SVER(row));
   for (int i = 0; i < pSchema->numOfCols; ++i) {
     STColumn *stCol = pSchema->columns + i;
     SCellVal  sVal = {255, NULL};

source/dnode/vnode/src/meta/metaQuery.c

@@ -278,12 +278,13 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   pSW = metaGetTableSchema(pMeta, quid, sver, 0);
   if (!pSW) return NULL;
 
-  tdInitTSchemaBuilder(&sb, 0);
+  tdInitTSchemaBuilder(&sb, sver);
   for (int i = 0; i < pSW->nCols; i++) {
     pSchema = pSW->pSchema + i;
     tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
   }
   pTSchema = tdGetSchemaFromBuilder(&sb);
+
   tdDestroyTSchemaBuilder(&sb);
 
   taosMemoryFree(pSW->pSchema);

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -1638,9 +1638,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
   int32_t numOfColsOfRow1 = 0;
 
   if (pSchema1 == NULL) {
-    // pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
-    // TODO: use the real schemaVersion
-    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
+    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
   }
 
 #ifdef TD_DEBUG_PRINT_ROW
@@ -1657,9 +1655,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
   if (row2) {
     isRow2DataRow = TD_IS_TP_ROW(row2);
     if (pSchema2 == NULL) {
-      // pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
-      // TODO: use the real schemaVersion
-      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
+      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
     }
     if (isRow2DataRow) {
       numOfColsOfRow2 = schemaNCols(pSchema2);

source/libs/function/inc/functionMgtInt.h

@@ -28,7 +28,7 @@ extern "C" {
 
 #define FUNC_MGT_AGG_FUNC               FUNC_MGT_FUNC_CLASSIFICATION_MASK(0)
 #define FUNC_MGT_SCALAR_FUNC            FUNC_MGT_FUNC_CLASSIFICATION_MASK(1)
-#define FUNC_MGT_NONSTANDARD_SQL_FUNC   FUNC_MGT_FUNC_CLASSIFICATION_MASK(2)
+#define FUNC_MGT_INDEFINITE_ROWS_FUNC   FUNC_MGT_FUNC_CLASSIFICATION_MASK(2)
 #define FUNC_MGT_STRING_FUNC            FUNC_MGT_FUNC_CLASSIFICATION_MASK(3)
 #define FUNC_MGT_DATETIME_FUNC          FUNC_MGT_FUNC_CLASSIFICATION_MASK(4)
 #define FUNC_MGT_TIMELINE_FUNC          FUNC_MGT_FUNC_CLASSIFICATION_MASK(5)

source/libs/function/src/builtins.c

@@ -837,7 +837,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "top",
     .type = FUNCTION_TYPE_TOP,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC,
+    .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
     .translateFunc = translateTop,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = functionSetup,
@@ -847,7 +847,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "bottom",
     .type = FUNCTION_TYPE_BOTTOM,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC,
+    .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
     .translateFunc = translateBottom,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = functionSetup,
@@ -929,7 +929,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "diff",
     .type = FUNCTION_TYPE_DIFF,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateDiff,
     .getEnvFunc   = getDiffFuncEnv,
     .initFunc     = diffFunctionSetup,
@@ -939,7 +939,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "state_count",
     .type = FUNCTION_TYPE_STATE_COUNT,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC,
     .translateFunc = translateStateCount,
     .getEnvFunc   = getStateFuncEnv,
     .initFunc     = functionSetup,
@@ -949,7 +949,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "state_duration",
     .type = FUNCTION_TYPE_STATE_DURATION,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateStateDuration,
     .getEnvFunc   = getStateFuncEnv,
     .initFunc     = functionSetup,
@@ -959,7 +959,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "csum",
     .type = FUNCTION_TYPE_CSUM,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateCsum,
     .getEnvFunc   = getCsumFuncEnv,
     .initFunc     = functionSetup,
@@ -969,7 +969,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "mavg",
     .type = FUNCTION_TYPE_MAVG,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateMavg,
     .getEnvFunc   = getMavgFuncEnv,
     .initFunc     = mavgFunctionSetup,
@@ -979,7 +979,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "sample",
     .type = FUNCTION_TYPE_SAMPLE,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateSample,
     .getEnvFunc   = getSampleFuncEnv,
     .initFunc     = sampleFunctionSetup,
@@ -989,7 +989,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "tail",
     .type = FUNCTION_TYPE_TAIL,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateTail,
     .getEnvFunc   = getTailFuncEnv,
     .initFunc     = tailFunctionSetup,
@@ -999,7 +999,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "unique",
     .type = FUNCTION_TYPE_UNIQUE,
-    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateUnique,
     .getEnvFunc   = getUniqueFuncEnv,
     .initFunc     = uniqueFunctionSetup,

source/libs/function/src/functionMgt.c

@@ -145,6 +145,8 @@ bool fmIsAggFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MG
 
 bool fmIsScalarFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SCALAR_FUNC); }
 
+bool fmIsVectorFunc(int32_t funcId) { return !fmIsScalarFunc(funcId); }
+
 bool fmIsSelectFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_FUNC); }
 
 bool fmIsTimelineFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TIMELINE_FUNC); }
@@ -157,7 +159,7 @@ bool fmIsWindowPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(
 
 bool fmIsWindowClauseFunc(int32_t funcId) { return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }
 
-bool fmIsNonstandardSQLFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_NONSTANDARD_SQL_FUNC); }
+bool fmIsIndefiniteRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INDEFINITE_ROWS_FUNC); }
 
 bool fmIsSpecialDataRequiredFunc(int32_t funcId) {
   return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);

source/libs/function/src/udfd.c

@@ -138,6 +138,7 @@ static void udfdCtrlReadCb(uv_stream_t *q, ssize_t nread, const uv_buf_t *buf);
 static int32_t udfdUvInit();
 static void udfdCloseWalkCb(uv_handle_t *handle, void *arg);
 static int32_t udfdRun();
+static void udfdConnectMnodeThreadFunc(void* args);
 
 void udfdProcessRequest(uv_work_t *req) {
   SUvUdfWork *uvUdf = (SUvUdfWork *)(req->data);
@@ -870,6 +871,23 @@ static int32_t udfdRun() {
   return 0;
 }
 
+void udfdConnectMnodeThreadFunc(void* args) {
+  int32_t retryMnodeTimes = 0;
+  int32_t code = 0;
+  while (retryMnodeTimes++ <= TSDB_MAX_REPLICA) {
+    uv_sleep(100 * (1 << retryMnodeTimes));
+    code = udfdConnectToMnode();
+    if (code == 0) {
+      break;
+    }
+    fnError("udfd can not connect to mnode, code: %s. retry", tstrerror(code));
+  }
+
+  if (code != 0) {
+    fnError("udfd can not connect to mnode");
+  }
+}
+
 int main(int argc, char *argv[]) {
   if (!taosCheckSystemIsSmallEnd()) {
     printf("failed to start since on non-small-end machines\n");
@@ -902,30 +920,18 @@ int main(int argc, char *argv[]) {
     return -3;
   }
 
-  int32_t retryMnodeTimes = 0;
-  int32_t code = 0;
-  while (retryMnodeTimes++ <= TSDB_MAX_REPLICA) {
-    uv_sleep(100 * (1 << retryMnodeTimes));
-    code = udfdConnectToMnode();
-    if (code == 0) {
-      break;
-    }
-    fnError("can not connect to mnode, code: %s. retry", tstrerror(code));
-  }
-
-  if (code != 0) {
-    fnError("failed to start since can not connect to mnode");
-    return -4;
-  }
-
   if (udfdUvInit() != 0) {
     fnError("uv init failure");
     return -5;
   }
 
+  uv_thread_t mnodeConnectThread;
+  uv_thread_create(&mnodeConnectThread, udfdConnectMnodeThreadFunc, NULL);
+
   udfdRun();
 
   removeListeningPipe();
+  uv_thread_join(&mnodeConnectThread);
   udfdCloseClientRpc();
 
   return 0;

source/libs/nodes/src/nodesCodeFuncs.c

@@ -492,6 +492,7 @@ static const char* jkScanLogicPlanScanCols = "ScanCols";
 static const char* jkScanLogicPlanScanPseudoCols = "ScanPseudoCols";
 static const char* jkScanLogicPlanTableMetaSize = "TableMetaSize";
 static const char* jkScanLogicPlanTableMeta = "TableMeta";
+static const char* jkScanLogicPlanTagCond = "TagCond";
 
 static int32_t logicScanNodeToJson(const void* pObj, SJson* pJson) {
   const SScanLogicNode* pNode = (const SScanLogicNode*)pObj;
@@ -509,6 +510,9 @@ static int32_t logicScanNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkScanLogicPlanTableMeta, tableMetaToJson, pNode->pMeta);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkScanLogicPlanTagCond, nodeToJson, pNode->pTagCond);
+  }
 
   return code;
 }
@@ -530,6 +534,9 @@ static int32_t jsonToLogicScanNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonMakeObject(pJson, jkScanLogicPlanTableMeta, jsonToTableMeta, (void**)&pNode->pMeta, objSize);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkScanLogicPlanTagCond, &pNode->pTagCond);
+  }
 
   return code;
 }
@@ -1777,6 +1784,7 @@ static const char* jkSubplanDbFName = "DbFName";
 static const char* jkSubplanNodeAddr = "NodeAddr";
 static const char* jkSubplanRootNode = "RootNode";
 static const char* jkSubplanDataSink = "DataSink";
+static const char* jkSubplanTagCond = "TagCond";
 
 static int32_t subplanToJson(const void* pObj, SJson* pJson) {
   const SSubplan* pNode = (const SSubplan*)pObj;
@@ -1803,6 +1811,9 @@ static int32_t subplanToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkSubplanDataSink, nodeToJson, pNode->pDataSink);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkSubplanTagCond, nodeToJson, pNode->pTagCond);
+  }
 
   return code;
 }
@@ -1833,6 +1844,9 @@ static int32_t jsonToSubplan(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkSubplanDataSink, (SNode**)&pNode->pDataSink);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkSubplanTagCond, (SNode**)&pNode->pTagCond);
+  }
 
   return code;
 }

source/libs/parser/src/parAstCreater.c

@@ -342,25 +342,19 @@ SNode* createLogicConditionNode(SAstCreateContext* pCxt, ELogicConditionType typ
   CHECK_OUT_OF_MEM(cond);
   cond->condType = type;
   cond->pParameterList = nodesMakeList();
-  if ((QUERY_NODE_LOGIC_CONDITION == nodeType(pParam1) && type != ((SLogicConditionNode*)pParam1)->condType) ||
-      (QUERY_NODE_LOGIC_CONDITION == nodeType(pParam2) && type != ((SLogicConditionNode*)pParam2)->condType)) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pParam1) && type == ((SLogicConditionNode*)pParam1)->condType) {
+    nodesListAppendList(cond->pParameterList, ((SLogicConditionNode*)pParam1)->pParameterList);
+    ((SLogicConditionNode*)pParam1)->pParameterList = NULL;
+    nodesDestroyNode(pParam1);
+  } else {
     nodesListAppend(cond->pParameterList, pParam1);
-    nodesListAppend(cond->pParameterList, pParam2);
+  }
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pParam2) && type == ((SLogicConditionNode*)pParam2)->condType) {
+    nodesListAppendList(cond->pParameterList, ((SLogicConditionNode*)pParam2)->pParameterList);
+    ((SLogicConditionNode*)pParam2)->pParameterList = NULL;
+    nodesDestroyNode(pParam2);
   } else {
-    if (QUERY_NODE_LOGIC_CONDITION == nodeType(pParam1)) {
-      nodesListAppendList(cond->pParameterList, ((SLogicConditionNode*)pParam1)->pParameterList);
-      ((SLogicConditionNode*)pParam1)->pParameterList = NULL;
-      nodesDestroyNode(pParam1);
-    } else {
-      nodesListAppend(cond->pParameterList, pParam1);
-    }
-    if (QUERY_NODE_LOGIC_CONDITION == nodeType(pParam2)) {
-      nodesListAppendList(cond->pParameterList, ((SLogicConditionNode*)pParam2)->pParameterList);
-      ((SLogicConditionNode*)pParam2)->pParameterList = NULL;
-      nodesDestroyNode(pParam2);
-    } else {
-      nodesListAppend(cond->pParameterList, pParam2);
-    }
+    nodesListAppend(cond->pParameterList, pParam2);
   }
   return (SNode*)cond;
 }

source/libs/parser/src/parInsert.c

@@ -647,7 +647,7 @@ static FORCE_INLINE int32_t MemRowAppend(SMsgBuf* pMsgBuf, const void* value, in
   if (TSDB_DATA_TYPE_BINARY == pa->schema->type) {
     const char* rowEnd = tdRowEnd(rb->pBuf);
     STR_WITH_SIZE_TO_VARSTR(rowEnd, value, len);
-    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, true, pa->toffset, pa->colIdx);
+    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
   } else if (TSDB_DATA_TYPE_NCHAR == pa->schema->type) {
     // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
     int32_t     output = 0;

source/libs/planner/CMakeLists.txt

@@ -8,7 +8,7 @@ target_include_directories(
 
 target_link_libraries(
     planner
-    PRIVATE os util nodes catalog cjson parser function qcom scalar
+    PRIVATE os util nodes catalog cjson parser function qcom scalar index
     PUBLIC transport
 )
 

source/libs/planner/src/planLogicCreater.c

@@ -418,7 +418,7 @@ static SColumnNode* createColumnByExpr(const char* pStmtName, SExprNode* pExpr)
 }
 
 static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
-  if (!pSelect->hasAggFuncs && NULL == pSelect->pGroupByList) {
+  if (!pSelect->hasAggFuncs && !pSelect->hasIndefiniteRowsFunc && NULL == pSelect->pGroupByList) {
     return TSDB_CODE_SUCCESS;
   }
 
@@ -442,8 +442,8 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
     code = rewriteExprForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY);
   }
 
-  if (TSDB_CODE_SUCCESS == code && pSelect->hasAggFuncs) {
-    code = nodesCollectFuncs(pSelect, SQL_CLAUSE_GROUP_BY, fmIsAggFunc, &pAgg->pAggFuncs);
+  if (TSDB_CODE_SUCCESS == code && (pSelect->hasAggFuncs || pSelect->hasIndefiniteRowsFunc)) {
+    code = nodesCollectFuncs(pSelect, SQL_CLAUSE_GROUP_BY, fmIsVectorFunc, &pAgg->pAggFuncs);
   }
 
   // rewrite the expression in subsequent clauses

source/libs/planner/src/planOptimizer.c

@@ -15,6 +15,7 @@
 
 #include "filter.h"
 #include "functionMgt.h"
+#include "index.h"
 #include "planInt.h"
 
 #define OPTIMIZE_FLAG_MASK(n) (1 << n)
@@ -313,22 +314,53 @@ static EDealRes cpdIsPrimaryKeyCondImpl(SNode* pNode, void* pContext) {
 }
 
 static bool cpdIsPrimaryKeyCond(SNode* pNode) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode)) {
+    return false;
+  }
   bool isPrimaryKeyCond = false;
   nodesWalkExpr(pNode, cpdIsPrimaryKeyCondImpl, &isPrimaryKeyCond);
   return isPrimaryKeyCond;
 }
 
-static int32_t cpdPartitionScanLogicCond(SScanLogicNode* pScan, SNode** pPrimaryKeyCond, SNode** pOtherCond) {
+static EDealRes cpdIsTagCondImpl(SNode* pNode, void* pContext) {
+  if (QUERY_NODE_COLUMN == nodeType(pNode)) {
+    *((bool*)pContext) = ((COLUMN_TYPE_TAG == ((SColumnNode*)pNode)->colType) ? true : false);
+    return *((bool*)pContext) ? DEAL_RES_CONTINUE : DEAL_RES_END;
+  }
+  return DEAL_RES_CONTINUE;
+}
+
+static bool cpdIsTagCond(SNode* pNode) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode)) {
+    return false;
+  }
+  bool isTagCond = false;
+  nodesWalkExpr(pNode, cpdIsTagCondImpl, &isTagCond);
+  return isTagCond;
+}
+
+static int32_t cpdPartitionScanLogicCond(SScanLogicNode* pScan, SNode** pPrimaryKeyCond, SNode** pTagCond,
+                                         SNode** pOtherCond) {
   SLogicConditionNode* pLogicCond = (SLogicConditionNode*)pScan->node.pConditions;
 
+  if (LOGIC_COND_TYPE_AND != pLogicCond->condType) {
+    *pPrimaryKeyCond = NULL;
+    *pOtherCond = pScan->node.pConditions;
+    pScan->node.pConditions = NULL;
+    return TSDB_CODE_SUCCESS;
+  }
+
   int32_t code = TSDB_CODE_SUCCESS;
 
   SNodeList* pPrimaryKeyConds = NULL;
+  SNodeList* pTagConds = NULL;
   SNodeList* pOtherConds = NULL;
   SNode*     pCond = NULL;
   FOREACH(pCond, pLogicCond->pParameterList) {
     if (cpdIsPrimaryKeyCond(pCond)) {
       code = nodesListMakeAppend(&pPrimaryKeyConds, nodesCloneNode(pCond));
+    } else if (cpdIsTagCond(pScan->node.pConditions)) {
+      code = nodesListMakeAppend(&pTagConds, nodesCloneNode(pCond));
     } else {
       code = nodesListMakeAppend(&pOtherConds, nodesCloneNode(pCond));
     }
@@ -338,37 +370,46 @@ static int32_t cpdPartitionScanLogicCond(SScanLogicNode* pScan, SNode** pPrimary
   }
 
   SNode* pTempPrimaryKeyCond = NULL;
+  SNode* pTempTagCond = NULL;
   SNode* pTempOtherCond = NULL;
   if (TSDB_CODE_SUCCESS == code) {
     code = cpdMergeConds(&pTempPrimaryKeyCond, &pPrimaryKeyConds);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = cpdMergeConds(&pTempTagCond, &pTagConds);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = cpdMergeConds(&pTempOtherCond, &pOtherConds);
   }
 
   if (TSDB_CODE_SUCCESS == code) {
     *pPrimaryKeyCond = pTempPrimaryKeyCond;
+    *pTagCond = pTempTagCond;
     *pOtherCond = pTempOtherCond;
     nodesDestroyNode(pScan->node.pConditions);
     pScan->node.pConditions = NULL;
   } else {
     nodesDestroyList(pPrimaryKeyConds);
+    nodesDestroyList(pTagConds);
     nodesDestroyList(pOtherConds);
     nodesDestroyNode(pTempPrimaryKeyCond);
+    nodesDestroyNode(pTempTagCond);
     nodesDestroyNode(pTempOtherCond);
   }
 
   return code;
 }
 
-static int32_t cpdPartitionScanCond(SScanLogicNode* pScan, SNode** pPrimaryKeyCond, SNode** pOtherCond) {
-  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pScan->node.pConditions) &&
-      LOGIC_COND_TYPE_AND == ((SLogicConditionNode*)pScan->node.pConditions)->condType) {
-    return cpdPartitionScanLogicCond(pScan, pPrimaryKeyCond, pOtherCond);
+static int32_t cpdPartitionScanCond(SScanLogicNode* pScan, SNode** pPrimaryKeyCond, SNode** pTagCond,
+                                    SNode** pOtherCond) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pScan->node.pConditions)) {
+    return cpdPartitionScanLogicCond(pScan, pPrimaryKeyCond, pTagCond, pOtherCond);
   }
 
   if (cpdIsPrimaryKeyCond(pScan->node.pConditions)) {
     *pPrimaryKeyCond = pScan->node.pConditions;
+  } else if (cpdIsTagCond(pScan->node.pConditions)) {
+    *pTagCond = pScan->node.pConditions;
   } else {
     *pOtherCond = pScan->node.pConditions;
   }
@@ -391,6 +432,32 @@ static int32_t cpdCalcTimeRange(SScanLogicNode* pScan, SNode** pPrimaryKeyCond,
   return code;
 }
 
+static int32_t cpdApplyTagIndex(SScanLogicNode* pScan, SNode** pTagCond, SNode** pOtherCond) {
+  int32_t       code = TSDB_CODE_SUCCESS;
+  SIdxFltStatus idxStatus = idxGetFltStatus(*pTagCond);
+  switch (idxStatus) {
+    case SFLT_NOT_INDEX:
+      code = cpdCondAppend(pOtherCond, pTagCond);
+      break;
+    case SFLT_COARSE_INDEX:
+      pScan->pTagCond = nodesCloneNode(*pTagCond);
+      if (NULL == pScan->pTagCond) {
+        code = TSDB_CODE_OUT_OF_MEMORY;
+        break;
+      }
+      code = cpdCondAppend(pOtherCond, pTagCond);
+      break;
+    case SFLT_ACCURATE_INDEX:
+      pScan->pTagCond = *pTagCond;
+      *pTagCond = NULL;
+      break;
+    default:
+      code = TSDB_CODE_FAILED;
+      break;
+  }
+  return code;
+}
+
 static int32_t cpdOptimizeScanCondition(SOptimizeContext* pCxt, SScanLogicNode* pScan) {
   if (NULL == pScan->node.pConditions || OPTIMIZE_FLAG_TEST_MASK(pScan->node.optimizedFlag, OPTIMIZE_FLAG_CPD) ||
       TSDB_SYSTEM_TABLE == pScan->pMeta->tableType) {
@@ -398,11 +465,15 @@ static int32_t cpdOptimizeScanCondition(SOptimizeContext* pCxt, SScanLogicNode*
   }
 
   SNode*  pPrimaryKeyCond = NULL;
+  SNode*  pTagCond = NULL;
   SNode*  pOtherCond = NULL;
-  int32_t code = cpdPartitionScanCond(pScan, &pPrimaryKeyCond, &pOtherCond);
+  int32_t code = cpdPartitionScanCond(pScan, &pPrimaryKeyCond, &pTagCond, &pOtherCond);
   if (TSDB_CODE_SUCCESS == code && NULL != pPrimaryKeyCond) {
     code = cpdCalcTimeRange(pScan, &pPrimaryKeyCond, &pOtherCond);
   }
+  if (TSDB_CODE_SUCCESS == code && NULL != pTagCond) {
+    code = cpdApplyTagIndex(pScan, &pTagCond, &pOtherCond);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     pScan->node.pConditions = pOtherCond;
   }
@@ -618,30 +689,6 @@ static bool cpdContainPrimaryKeyEqualCond(SJoinLogicNode* pJoin, SNode* pCond) {
   }
 }
 
-// static int32_t cpdCheckOpCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SNode* pOnCond) {
-//   if (!cpdIsPrimaryKeyEqualCond(pJoin, pOnCond)) {
-//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-//   }
-//   return TSDB_CODE_SUCCESS;
-// }
-
-// static int32_t cpdCheckLogicCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin, SLogicConditionNode* pOnCond) {
-//   if (LOGIC_COND_TYPE_AND != pOnCond->condType) {
-//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-//   }
-//   bool   hasPrimaryKeyEqualCond = false;
-//   SNode* pCond = NULL;
-//   FOREACH(pCond, pOnCond->pParameterList) {
-//     if (cpdIsPrimaryKeyEqualCond(pJoin, pCond)) {
-//       hasPrimaryKeyEqualCond = true;
-//     }
-//   }
-//   if (!hasPrimaryKeyEqualCond) {
-//     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
-//   }
-//   return TSDB_CODE_SUCCESS;
-// }
-
 static int32_t cpdCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {
   if (NULL == pJoin->pOnConditions) {
     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_NOT_SUPPORT_CROSS_JOIN);
@@ -650,11 +697,6 @@ static int32_t cpdCheckJoinOnCond(SOptimizeContext* pCxt, SJoinLogicNode* pJoin)
     return generateUsageErrMsg(pCxt->pPlanCxt->pMsg, pCxt->pPlanCxt->msgLen, TSDB_CODE_PLAN_EXPECTED_TS_EQUAL);
   }
   return TSDB_CODE_SUCCESS;
-  // if (QUERY_NODE_LOGIC_CONDITION == nodeType(pJoin->pOnConditions)) {
-  //   return cpdCheckLogicCond(pCxt, pJoin, (SLogicConditionNode*)pJoin->pOnConditions);
-  // } else {
-  //   return cpdCheckOpCond(pCxt, pJoin, pJoin->pOnConditions);
-  // }
 }
 
 static int32_t cpdPushJoinCondition(SOptimizeContext* pCxt, SJoinLogicNode* pJoin) {

source/libs/planner/src/planPhysiCreater.c

@@ -411,7 +411,7 @@ static int32_t createScanCols(SPhysiPlanContext* pCxt, SScanPhysiNode* pScanPhys
   return sortScanCols(pScanPhysiNode->pScanCols);
 }
 
-static int32_t createScanPhysiNodeFinalize(SPhysiPlanContext* pCxt, SScanLogicNode* pScanLogicNode,
+static int32_t createScanPhysiNodeFinalize(SPhysiPlanContext* pCxt, SSubplan* pSubplan, SScanLogicNode* pScanLogicNode,
                                            SScanPhysiNode* pScanPhysiNode, SPhysiNode** pPhyNode) {
   int32_t code = createScanCols(pCxt, pScanPhysiNode, pScanLogicNode->pScanCols);
   if (TSDB_CODE_SUCCESS == code) {
@@ -438,6 +438,12 @@ static int32_t createScanPhysiNodeFinalize(SPhysiPlanContext* pCxt, SScanLogicNo
     pScanPhysiNode->uid = pScanLogicNode->pMeta->uid;
     pScanPhysiNode->tableType = pScanLogicNode->pMeta->tableType;
     memcpy(&pScanPhysiNode->tableName, &pScanLogicNode->tableName, sizeof(SName));
+    if (NULL != pScanLogicNode->pTagCond) {
+      pSubplan->pTagCond = nodesCloneNode(pScanLogicNode->pTagCond);
+      if (NULL == pSubplan->pTagCond) {
+        code = TSDB_CODE_OUT_OF_MEMORY;
+      }
+    }
   }
 
   if (TSDB_CODE_SUCCESS == code) {
@@ -463,7 +469,7 @@ static int32_t createTagScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubpla
   }
   vgroupInfoToNodeAddr(pScanLogicNode->pVgroupList->vgroups, &pSubplan->execNode);
   taosArrayPush(pCxt->pExecNodeList, &pSubplan->execNode);
-  return createScanPhysiNodeFinalize(pCxt, pScanLogicNode, (SScanPhysiNode*)pTagScan, pPhyNode);
+  return createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pTagScan, pPhyNode);
 }
 
 static int32_t createTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan, SScanLogicNode* pScanLogicNode,
@@ -498,7 +504,7 @@ static int32_t createTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubp
   pTableScan->intervalUnit = pScanLogicNode->intervalUnit;
   pTableScan->slidingUnit = pScanLogicNode->slidingUnit;
 
-  return createScanPhysiNodeFinalize(pCxt, pScanLogicNode, (SScanPhysiNode*)pTableScan, pPhyNode);
+  return createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pTableScan, pPhyNode);
 }
 
 static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan,
@@ -522,7 +528,7 @@ static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan*
   pScan->mgmtEpSet = pCxt->pPlanCxt->mgmtEpSet;
   tNameGetFullDbName(&pScanLogicNode->tableName, pSubplan->dbFName);
 
-  return createScanPhysiNodeFinalize(pCxt, pScanLogicNode, (SScanPhysiNode*)pScan, pPhyNode);
+  return createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pScan, pPhyNode);
 }
 
 static int32_t createStreamScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan, SScanLogicNode* pScanLogicNode,

source/libs/planner/src/planner.c

@@ -18,6 +18,13 @@
 #include "planInt.h"
 #include "scalar.h"
 
+static void dumpQueryPlan(SQueryPlan* pPlan) {
+  char* pStr = NULL;
+  nodesNodeToString(pPlan, false, &pStr, NULL);
+  planDebugL("Query Plan: %s", pStr);
+  taosMemoryFree(pStr);
+}
+
 int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {
   SLogicNode*      pLogicNode = NULL;
   SLogicSubplan*   pLogicSubplan = NULL;
@@ -36,6 +43,9 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
   if (TSDB_CODE_SUCCESS == code) {
     code = createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    dumpQueryPlan(*pPlan);
+  }
 
   nodesDestroyNode(pLogicNode);
   nodesDestroyNode(pLogicSubplan);

source/libs/planner/test/planOptimizeTest.cpp

@@ -32,6 +32,12 @@ TEST_F(PlanOptimizeTest, optimizeScanData) {
   run("SELECT PERCENTILE(c1, 40), COUNT(*) FROM t1");
 }
 
+TEST_F(PlanOptimizeTest, ConditionPushDown) {
+  useDb("root", "test");
+
+  run("SELECT ts, c1 FROM st1 WHERE tag1 > 4");
+}
+
 TEST_F(PlanOptimizeTest, orderByPrimaryKey) {
   useDb("root", "test");
 

source/libs/planner/test/planTestUtil.cpp

@@ -233,45 +233,45 @@ class PlannerTestBaseImpl {
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_PARSER == module) {
       if (res_.prepareAst_.empty()) {
-        cout << "syntax tree : " << endl;
+        cout << "+++++++++++++++++++++syntax tree : " << endl;
         cout << res_.ast_ << endl;
       } else {
-        cout << "prepare syntax tree : " << endl;
+        cout << "+++++++++++++++++++++prepare syntax tree : " << endl;
         cout << res_.prepareAst_ << endl;
-        cout << "bound syntax tree : " << endl;
+        cout << "+++++++++++++++++++++bound syntax tree : " << endl;
         cout << res_.boundAst_ << endl;
-        cout << "syntax tree : " << endl;
+        cout << "+++++++++++++++++++++syntax tree : " << endl;
         cout << res_.ast_ << endl;
       }
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_LOGIC == module) {
-      cout << "raw logic plan : " << endl;
+      cout << "+++++++++++++++++++++raw logic plan : " << endl;
       cout << res_.rawLogicPlan_ << endl;
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_OPTIMIZED == module) {
-      cout << "optimized logic plan : " << endl;
+      cout << "+++++++++++++++++++++optimized logic plan : " << endl;
       cout << res_.optimizedLogicPlan_ << endl;
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_SPLIT == module) {
-      cout << "split logic plan : " << endl;
+      cout << "+++++++++++++++++++++split logic plan : " << endl;
       cout << res_.splitLogicPlan_ << endl;
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_SCALED == module) {
-      cout << "scaled logic plan : " << endl;
+      cout << "+++++++++++++++++++++scaled logic plan : " << endl;
       cout << res_.scaledLogicPlan_ << endl;
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_PHYSICAL == module) {
-      cout << "physical plan : " << endl;
+      cout << "+++++++++++++++++++++physical plan : " << endl;
       cout << res_.physiPlan_ << endl;
     }
 
     if (DUMP_MODULE_ALL == module || DUMP_MODULE_SUBPLAN == module) {
-      cout << "physical subplan : " << endl;
+      cout << "+++++++++++++++++++++physical subplan : " << endl;
       for (const auto& subplan : res_.physiSubplans_) {
         cout << subplan << endl;
       }