Merge pull request #19209 from taosdata/enh/3.0_planner_optimize

enh: when there is _wend and _wduration in the query, you can use tsma optimization

include/libs/nodes/cmdnodes.h

@@ -295,7 +295,7 @@ typedef struct SShowTableTagsStmt {
   SNodeList* pTags;
 } SShowTableTagsStmt;
 
-typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT } EIndexType;
+typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT, INDEX_TYPE_NORMAL } EIndexType;
 
 typedef struct SIndexOptions {
   ENodeType  type;

source/libs/parser/inc/parInsertUtil.h

@@ -35,128 +35,27 @@ struct SToken;
     }                                \
   } while (0)
 
-#define IS_DATA_COL_ORDERED(spd) ((spd->orderStatus) == (int8_t)ORDER_STATUS_ORDERED)
-
-typedef enum EOrderStatus {
-  ORDER_STATUS_UNKNOWN = 0,
-  ORDER_STATUS_ORDERED = 1,
-  ORDER_STATUS_DISORDERED = 2,
-} EOrderStatus;
-
-typedef enum EValStat {
-  VAL_STAT_HAS = 0x0,    // 0 means has val
-  VAL_STAT_NONE = 0x01,  // 1 means no val
-} EValStat;
-
-typedef struct SBoundColumn {
-  int32_t offset;   // all column offset value
-  int32_t toffset;  // first part offset for SDataRow TODO: get offset from STSchema on future
-  uint8_t valStat;  // EValStat. denote if current column bound or not(0 means has val, 1 means no val)
-} SBoundColumn;
-
-typedef struct {
-  col_id_t schemaColIdx;
-  col_id_t boundIdx;
-  col_id_t finalIdx;
-} SBoundIdxInfo;
-
-typedef struct SParsedDataColInfo {
-  col_id_t       numOfCols;
-  col_id_t       numOfBound;
-  uint16_t       flen;        // TODO: get from STSchema
-  uint16_t       allNullLen;  // TODO: get from STSchema(base on SDataRow)
-  uint16_t       extendedVarLen;
-  uint16_t       boundNullLen;  // bound column len with all NULL value(without VarDataOffsetT/SColIdx part)
-  col_id_t      *boundColumns;  // bound column idx according to schema
-  SBoundColumn  *cols;
-  SBoundIdxInfo *colIdxInfo;
-  int8_t         orderStatus;  // bound columns
-} SParsedDataColInfo;
-
-typedef struct SInsertParseBaseContext {
-  SParseContext *pComCxt;
-  char          *pSql;
-  SMsgBuf        msg;
-} SInsertParseBaseContext;
-
-typedef struct SInsertParseSyntaxCxt {
-  SParseContext   *pComCxt;
-  char            *pSql;
-  SMsgBuf          msg;
-  SParseMetaCache *pMetaCache;
-} SInsertParseSyntaxCxt;
-
-typedef struct SMemParam {
-  SRowBuilder *rb;
-  SSchema     *schema;
-  int32_t      toffset;
-  col_id_t     colIdx;
-} SMemParam;
-
-typedef struct {
-  uint8_t rowType;  // default is 0, that is SDataRow
-  int32_t rowSize;
-} SMemRowBuilder;
-
-typedef struct STableDataBlocks {
-  int8_t      tsSource;     // where does the UNIX timestamp come from, server or client
-  bool        ordered;      // if current rows are ordered or not
-  int32_t     vgId;         // virtual group id
-  int64_t     prevTS;       // previous timestamp, recorded to decide if the records array is ts ascending
-  int32_t     numOfTables;  // number of tables in current submit block
-  int32_t     rowSize;      // row size for current table
-  uint32_t    nAllocSize;
-  uint32_t    headerSize;  // header for table info (uid, tid, submit metadata)
-  uint32_t    size;
-  STableMeta *pTableMeta;  // the tableMeta of current table, the table meta will be used during submit, keep a ref to
-                           // avoid to be removed from cache
-  char              *pData;
-  bool               cloned;
-  int32_t            createTbReqLen;
-  SParsedDataColInfo boundColumnInfo;
-  SRowBuilder        rowBuilder;
-} STableDataBlocks;
-
-int32_t insGetExtendedRowSize(STableDataBlocks *pBlock);
-void insGetSTSRowAppendInfo(uint8_t rowType, SParsedDataColInfo *spd, col_id_t idx, int32_t *toffset, col_id_t *colIdx);
-int32_t insSetBlockInfo(SSubmitBlk *pBlocks, STableDataBlocks *dataBuf, int32_t numOfRows, SMsgBuf *pMsg);
-int32_t insSchemaIdxCompar(const void *lhs, const void *rhs);
-int32_t insBoundIdxCompar(const void *lhs, const void *rhs);
-void    insSetBoundColumnInfo(SParsedDataColInfo *pColList, SSchema *pSchema, col_id_t numOfCols);
-void    insDestroyBlockArrayList(SArray *pDataBlockList);
-void    insDestroyBlockHashmap(SHashObj *pDataBlockHash);
-int32_t insInitRowBuilder(SRowBuilder *pBuilder, int16_t schemaVer, SParsedDataColInfo *pColInfo);
-int32_t insGetDataBlockFromList(SHashObj *pHashList, void *id, int32_t idLen, int32_t size, int32_t startOffset,
-                                int32_t rowSize, STableMeta *pTableMeta, STableDataBlocks **dataBlocks,
-                                SArray *pBlockList, SVCreateTbReq *pCreateTbReq);
-int32_t insMergeTableDataBlocks(SHashObj *pHashObj, SArray **pVgDataBlocks);
-int32_t insBuildCreateTbMsg(STableDataBlocks *pBlocks, SVCreateTbReq *pCreateTbReq);
-int32_t insAllocateMemForSize(STableDataBlocks *pDataBlock, int32_t allSize);
-int32_t insCreateSName(SName *pName, struct SToken *pTableName, int32_t acctId, const char *dbName, SMsgBuf *pMsgBuf);
-int16_t insFindCol(struct SToken *pColname, int16_t start, int16_t end, SSchema *pSchema);
-void    insBuildCreateTbReq(SVCreateTbReq *pTbReq, const char *tname, STag *pTag, int64_t suid, const char *sname,
-                            SArray *tagName, uint8_t tagNum, int32_t ttl);
-int32_t insMemRowAppend(SMsgBuf *pMsgBuf, const void *value, int32_t len, void *param);
-int32_t insCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start);
-int32_t insBuildOutput(SHashObj *pVgroupsHashObj, SArray *pVgDataBlocks, SArray **pDataBlocks);
-void    insDestroyDataBlock(STableDataBlocks *pDataBlock);
-
 typedef struct SVgroupDataCxt {
   int32_t      vgId;
   SSubmitReq2 *pData;
 } SVgroupDataCxt;
 
+int32_t insCreateSName(SName *pName, struct SToken *pTableName, int32_t acctId, const char *dbName, SMsgBuf *pMsgBuf);
+int16_t insFindCol(struct SToken *pColname, int16_t start, int16_t end, SSchema *pSchema);
+void    insBuildCreateTbReq(SVCreateTbReq *pTbReq, const char *tname, STag *pTag, int64_t suid, const char *sname,
+                            SArray *tagName, uint8_t tagNum, int32_t ttl);
 int32_t insInitBoundColsInfo(int32_t numOfBound, SBoundColInfo *pInfo);
 void    insCheckTableDataOrder(STableDataCxt *pTableCxt, TSKEY tsKey);
 int32_t insGetTableDataCxt(SHashObj *pHash, void *id, int32_t idLen, STableMeta *pTableMeta,
                            SVCreateTbReq **pCreateTbReq, STableDataCxt **pTableCxt, bool colMode);
-int32_t initTableColSubmitData(STableDataCxt* pTableCxt);
+int32_t initTableColSubmitData(STableDataCxt *pTableCxt);
 int32_t insMergeTableDataCxt(SHashObj *pTableHash, SArray **pVgDataBlocks);
 int32_t insBuildVgDataBlocks(SHashObj *pVgroupsHashObj, SArray *pVgDataBlocks, SArray **pDataBlocks);
 void    insDestroyTableDataCxtHashMap(SHashObj *pTableCxtHash);
 void    insDestroyVgroupDataCxt(SVgroupDataCxt *pVgCxt);
 void    insDestroyVgroupDataCxtList(SArray *pVgCxtList);
 void    insDestroyVgroupDataCxtHashMap(SHashObj *pVgCxtHash);
-void    insDestroyTableDataCxt(STableDataCxt* pTableCxt);
-void    destroyBoundColInfo(SBoundColInfo* pInfo);
+void    insDestroyTableDataCxt(STableDataCxt *pTableCxt);
+void    insDestroyBoundColInfo(SBoundColInfo *pInfo);
+
 #endif  // TDENGINE_PAR_INSERT_UTIL_H

source/libs/parser/inc/sql.y

@@ -460,6 +460,8 @@ tag_item(A) ::= column_name(B) AS column_alias(C).
 /************************************************ create index ********************************************************/
 cmd ::= CREATE SMA INDEX not_exists_opt(D)
   full_table_name(A) ON full_table_name(B) index_options(C).                      { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_SMA, D, A, B, NULL, C); }
+cmd ::= CREATE INDEX not_exists_opt(D)
+  full_table_name(A) ON full_table_name(B) NK_LP col_name_list(C) NK_RP.          { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_NORMAL, D, A, B, C, NULL); }
 cmd ::= DROP INDEX exists_opt(B) full_table_name(A).                              { pCxt->pRootNode = createDropIndexStmt(pCxt, B, A); }
 
 index_options(A) ::= FUNCTION NK_LP func_list(B) NK_RP INTERVAL

source/libs/parser/src/parInsertSml.c

@@ -348,7 +348,7 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
   }
 
 end:
-  destroyBoundColInfo(&bindTags);
+  insDestroyBoundColInfo(&bindTags);
   taosMemoryFree(pCreateTblReq);
   taosArrayDestroy(tagName);
   return ret;

source/libs/parser/src/parInsertSql.c

@@ -1466,7 +1466,7 @@ static int32_t parseInsertTableClauseBottom(SInsertParseContext* pCxt, SVnodeMod
 }
 
 static void resetEnvPreTable(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt) {
-  destroyBoundColInfo(&pCxt->tags);
+  insDestroyBoundColInfo(&pCxt->tags);
   taosMemoryFreeClear(pStmt->pTableMeta);
   tdDestroySVCreateTbReq(pStmt->pCreateTblReq);
   taosMemoryFreeClear(pStmt->pCreateTblReq);
@@ -1527,7 +1527,7 @@ static int32_t checkTableClauseFirstToken(SInsertParseContext* pCxt, SVnodeModif
 }
 
 static int32_t setStmtInfo(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt) {
-  SParsedDataColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags));
+  SBoundColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags));
   if (NULL == tags) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -1937,6 +1937,6 @@ int32_t parseInsertSql(SParseContext* pCxt, SQuery** pQuery, SCatalogReq* pCatal
       QUERY_EXEC_STAGE_SCHEDULE == (*pQuery)->execStage) {
     code = setRefreshMate(*pQuery);
   }
-  destroyBoundColInfo(&context.tags);
+  insDestroyBoundColInfo(&context.tags);
   return code;
 }

source/libs/parser/src/parTranslater.c

@@ -4569,58 +4569,109 @@ typedef struct SSampleAstInfo {
   SNode*      pSliding;
   SNodeList*  pPartitionByList;
   STableMeta* pRollupTableMeta;
+  bool        createSmaIndex;
 } SSampleAstInfo;
 
-static int32_t buildSampleAst(STranslateContext* pCxt, SSampleAstInfo* pInfo, char** pAst, int32_t* pLen, char** pExpr,
-                              int32_t* pExprLen) {
-  SSelectStmt* pSelect = (SSelectStmt*)nodesMakeNode(QUERY_NODE_SELECT_STMT);
-  if (NULL == pSelect) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-  sprintf(pSelect->stmtName, "%p", pSelect);
-
+static int32_t buildTableForSampleAst(SSampleAstInfo* pInfo, SNode** pOutput) {
   SRealTableNode* pTable = (SRealTableNode*)nodesMakeNode(QUERY_NODE_REAL_TABLE);
   if (NULL == pTable) {
-    nodesDestroyNode((SNode*)pSelect);
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   snprintf(pTable->table.dbName, sizeof(pTable->table.dbName), "%s", pInfo->pDbName);
   snprintf(pTable->table.tableName, sizeof(pTable->table.tableName), "%s", pInfo->pTableName);
   TSWAP(pTable->pMeta, pInfo->pRollupTableMeta);
-  pSelect->pFromTable = (SNode*)pTable;
+  *pOutput = (SNode*)pTable;
+  return TSDB_CODE_SUCCESS;
+}
 
-  TSWAP(pSelect->pProjectionList, pInfo->pFuncs);
+static int32_t addWstartToSampleProjects(SNodeList* pProjectionList) {
   SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
-  if (NULL == pSelect->pProjectionList || NULL == pFunc) {
-    nodesDestroyNode((SNode*)pSelect);
+  if (NULL == pFunc) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   strcpy(pFunc->functionName, "_wstart");
-  nodesListPushFront(pSelect->pProjectionList, (SNode*)pFunc);
-  SNode* pProject = NULL;
-  FOREACH(pProject, pSelect->pProjectionList) { sprintf(((SExprNode*)pProject)->aliasName, "#%p", pProject); }
+  return nodesListPushFront(pProjectionList, (SNode*)pFunc);
+}
 
-  TSWAP(pSelect->pPartitionByList, pInfo->pPartitionByList);
+static int32_t addWendToSampleProjects(SNodeList* pProjectionList) {
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFunc) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  strcpy(pFunc->functionName, "_wend");
+  return nodesListAppend(pProjectionList, (SNode*)pFunc);
+}
+
+static int32_t addWdurationToSampleProjects(SNodeList* pProjectionList) {
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFunc) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  strcpy(pFunc->functionName, "_wduration");
+  return nodesListAppend(pProjectionList, (SNode*)pFunc);
+}
+
+static int32_t buildProjectsForSampleAst(SSampleAstInfo* pInfo, SNodeList** pList) {
+  SNodeList* pProjectionList = pInfo->pFuncs;
+  pInfo->pFuncs = NULL;
+
+  int32_t code = addWstartToSampleProjects(pProjectionList);
+  if (TSDB_CODE_SUCCESS == code && pInfo->createSmaIndex) {
+    code = addWendToSampleProjects(pProjectionList);
+    if (TSDB_CODE_SUCCESS == code) {
+      code = addWdurationToSampleProjects(pProjectionList);
+    }
+  }
 
+  if (TSDB_CODE_SUCCESS == code) {
+    SNode* pProject = NULL;
+    FOREACH(pProject, pProjectionList) { sprintf(((SExprNode*)pProject)->aliasName, "#%p", pProject); }
+    *pList = pProjectionList;
+  } else {
+    nodesDestroyList(pProjectionList);
+  }
+  return code;
+}
+
+static int32_t buildIntervalForSampleAst(SSampleAstInfo* pInfo, SNode** pOutput) {
   SIntervalWindowNode* pInterval = (SIntervalWindowNode*)nodesMakeNode(QUERY_NODE_INTERVAL_WINDOW);
   if (NULL == pInterval) {
-    nodesDestroyNode((SNode*)pSelect);
     return TSDB_CODE_OUT_OF_MEMORY;
   }
-  pSelect->pWindow = (SNode*)pInterval;
   TSWAP(pInterval->pInterval, pInfo->pInterval);
   TSWAP(pInterval->pOffset, pInfo->pOffset);
   TSWAP(pInterval->pSliding, pInfo->pSliding);
   pInterval->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
   if (NULL == pInterval->pCol) {
-    nodesDestroyNode((SNode*)pSelect);
+    nodesDestroyNode((SNode*)pInterval);
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   ((SColumnNode*)pInterval->pCol)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
   strcpy(((SColumnNode*)pInterval->pCol)->colName, ROWTS_PSEUDO_COLUMN_NAME);
+  *pOutput = (SNode*)pInterval;
+  return TSDB_CODE_SUCCESS;
+}
 
+static int32_t buildSampleAst(STranslateContext* pCxt, SSampleAstInfo* pInfo, char** pAst, int32_t* pLen, char** pExpr,
+                              int32_t* pExprLen) {
+  SSelectStmt* pSelect = (SSelectStmt*)nodesMakeNode(QUERY_NODE_SELECT_STMT);
+  if (NULL == pSelect) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  sprintf(pSelect->stmtName, "%p", pSelect);
+
+  int32_t code = buildTableForSampleAst(pInfo, &pSelect->pFromTable);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = buildProjectsForSampleAst(pInfo, &pSelect->pProjectionList);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    TSWAP(pSelect->pPartitionByList, pInfo->pPartitionByList);
+    code = buildIntervalForSampleAst(pInfo, &pSelect->pWindow);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
     pCxt->createStream = true;
-  int32_t code = translateQuery(pCxt, (SNode*)pSelect);
+    code = translateQuery(pCxt, (SNode*)pSelect);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = nodesNodeToString((SNode*)pSelect, false, pAst, pLen);
   }
@@ -5162,6 +5213,7 @@ static int32_t getSmaIndexSql(STranslateContext* pCxt, char** pSql, int32_t* pLe
 }
 
 static int32_t buildSampleAstInfoByIndex(STranslateContext* pCxt, SCreateIndexStmt* pStmt, SSampleAstInfo* pInfo) {
+  pInfo->createSmaIndex = true;
   pInfo->pDbName = pStmt->dbName;
   pInfo->pTableName = pStmt->tableName;
   pInfo->pFuncs = nodesCloneList(pStmt->pOptions->pFuncs);

source/libs/planner/src/planOptimizer.c

@@ -1284,18 +1284,14 @@ static int32_t smaIndexOptFindSmaFunc(SNode* pQueryFunc, SNodeList* pSmaFuncs) {
   return -1;
 }
 
-static int32_t smaIndexOptCreateSmaCols(SNodeList* pFuncs, uint64_t tableId, SNodeList* pSmaFuncs, SNodeList** pOutput,
-                                        int32_t* pWStrartIndex) {
+static int32_t smaIndexOptCreateSmaCols(SNodeList* pFuncs, uint64_t tableId, SNodeList* pSmaFuncs,
+                                        SNodeList** pOutput) {
   SNodeList* pCols = NULL;
   SNode*     pFunc = NULL;
   int32_t    code = TSDB_CODE_SUCCESS;
   int32_t    index = 0;
   int32_t    smaFuncIndex = -1;
-  *pWStrartIndex = -1;
   FOREACH(pFunc, pFuncs) {
-    if (FUNCTION_TYPE_WSTART == ((SFunctionNode*)pFunc)->funcType) {
-      *pWStrartIndex = index;
-    }
     smaFuncIndex = smaIndexOptFindSmaFunc(pFunc, pSmaFuncs);
     if (smaFuncIndex < 0) {
       break;
@@ -1317,8 +1313,7 @@ static int32_t smaIndexOptCreateSmaCols(SNodeList* pFuncs, uint64_t tableId, SNo
   return code;
 }
 
-static int32_t smaIndexOptCouldApplyIndex(SScanLogicNode* pScan, STableIndexInfo* pIndex, SNodeList** pCols,
-                                          int32_t* pWStrartIndex) {
+static int32_t smaIndexOptCouldApplyIndex(SScanLogicNode* pScan, STableIndexInfo* pIndex, SNodeList** pCols) {
   SWindowLogicNode* pWindow = (SWindowLogicNode*)pScan->node.pParent;
   if (!smaIndexOptEqualInterval(pScan, pWindow, pIndex)) {
     return TSDB_CODE_SUCCESS;
@@ -1326,14 +1321,14 @@ static int32_t smaIndexOptCouldApplyIndex(SScanLogicNode* pScan, STableIndexInfo
   SNodeList* pSmaFuncs = NULL;
   int32_t    code = nodesStringToList(pIndex->expr, &pSmaFuncs);
   if (TSDB_CODE_SUCCESS == code) {
-    code = smaIndexOptCreateSmaCols(pWindow->pFuncs, pIndex->dstTbUid, pSmaFuncs, pCols, pWStrartIndex);
+    code = smaIndexOptCreateSmaCols(pWindow->pFuncs, pIndex->dstTbUid, pSmaFuncs, pCols);
   }
   nodesDestroyList(pSmaFuncs);
   return code;
 }
 
 static int32_t smaIndexOptApplyIndex(SLogicSubplan* pLogicSubplan, SScanLogicNode* pScan, STableIndexInfo* pIndex,
-                                     SNodeList* pSmaCols, int32_t wstrartIndex) {
+                                     SNodeList* pSmaCols) {
   SLogicNode* pSmaScan = NULL;
   int32_t     code = smaIndexOptCreateSmaScan(pScan, pIndex, pSmaCols, &pSmaScan);
   if (TSDB_CODE_SUCCESS == code) {
@@ -1350,10 +1345,9 @@ static int32_t smaIndexOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogi
   for (int32_t i = 0; i < nindexes; ++i) {
     STableIndexInfo* pIndex = taosArrayGet(pScan->pSmaIndexes, i);
     SNodeList*       pSmaCols = NULL;
-    int32_t          wstrartIndex = -1;
-    code = smaIndexOptCouldApplyIndex(pScan, pIndex, &pSmaCols, &wstrartIndex);
+    code = smaIndexOptCouldApplyIndex(pScan, pIndex, &pSmaCols);
     if (TSDB_CODE_SUCCESS == code && NULL != pSmaCols) {
-      code = smaIndexOptApplyIndex(pLogicSubplan, pScan, pIndex, pSmaCols, wstrartIndex);
+      code = smaIndexOptApplyIndex(pLogicSubplan, pScan, pIndex, pSmaCols);
       taosArrayDestroyEx(pScan->pSmaIndexes, smaIndexOptDestroySmaIndex);
       pScan->pSmaIndexes = NULL;
       pCxt->optimized = true;