Merge pull request #11706 from taosdata/feature/tq

refactor(tmq): rewrite tq read function

include/libs/planner/planner.h

@@ -30,6 +30,7 @@ typedef struct SPlanContext {
   SNode* pAstRoot;
   bool topicQuery;
   bool streamQuery;
+  bool rSmaQuery;
   bool showRewrite;
   int8_t triggerType;
   int64_t watermark;

source/dnode/mnode/impl/inc/mndScheduler.h

@@ -29,6 +29,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
 
 int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
 
+int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/impl/src/mndScheduler.c

@@ -34,6 +34,54 @@
 
 extern bool tsStreamSchedV;
 
+int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen) {
+  SNode*      pAst = NULL;
+  SQueryPlan* pPlan = NULL;
+  terrno = TSDB_CODE_SUCCESS;
+
+  if (nodesStringToNode(ast, &pAst) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  SPlanContext cxt = {
+      .pAstRoot = pAst,
+      .topicQuery = false,
+      .streamQuery = true,
+      .rSmaQuery = true,
+      .triggerType = triggerType,
+      .watermark = watermark,
+  };
+  if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
+  if (levelNum != 1) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+  SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+
+  int32_t opNum = LIST_LENGTH(inner->pNodeList);
+  if (opNum != 1) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+  SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
+  if (qSubPlanToString(plan, pStr, pLen) < 0) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    goto END;
+  }
+
+END:
+  if (pAst) nodesDestroyNode(pAst);
+  if (pPlan) nodesDestroyNode(pPlan);
+  return terrno;
+}
+
 int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) {
   SCoder encoder;
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);

source/dnode/vnode/inc/vnode.h

@@ -109,8 +109,7 @@ int     tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList
 int     tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
 bool    tqNextDataBlock(STqReadHandle *pHandle);
-int     tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo);
+int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, int32_t *pNumOfRows);
-SArray *tqRetrieveDataBlock(STqReadHandle *pHandle);
 
 // need to reposition
 

source/dnode/vnode/src/tq/tqRead.c

@@ -82,16 +82,7 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
   return false;
 }
 
-int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) {
+int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, int32_t* pNumOfRows) {
-  // currently only rows are used
-
-  pBlockInfo->numOfCols = taosArrayGetSize(pHandle->pColIdList);
-  pBlockInfo->rows = pHandle->pBlock->numOfRows;
-  //  pBlockInfo->uid = pHandle->pBlock->uid; // the uid can not be assigned to pBlockData.
-  return 0;
-}
-
-SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
   /*int32_t         sversion = pHandle->pBlock->sversion;*/
   // TODO set to real sversion
   int32_t sversion = 0;
@@ -112,7 +103,7 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
   STSchema*       pTschema = pHandle->pSchema;
   SSchemaWrapper* pSchemaWrapper = pHandle->pSchemaWrapper;
 
-  int32_t numOfRows = pHandle->pBlock->numOfRows;
+  *pNumOfRows = pHandle->pBlock->numOfRows;
   /*int32_t numOfCols = pHandle->pSchema->numOfCols;*/
   int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
 
@@ -120,10 +111,11 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
     colNumNeed = pSchemaWrapper->nCols;
   }
 
-  SArray* pArray = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
+  *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
-  if (pArray == NULL) {
+  if (*ppCols == NULL) {
-    return NULL;
+    return -1;
   }
+
   int32_t colMeta = 0;
   int32_t colNeed = 0;
   while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
@@ -136,21 +128,24 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
       colNeed++;
     } else {
       SColumnInfoData colInfo = {0};
-      /*int             sz = numOfRows * pColSchema->bytes;*/
       colInfo.info.bytes = pColSchema->bytes;
       colInfo.info.colId = pColSchema->colId;
       colInfo.info.type = pColSchema->type;
 
-      if (colInfoDataEnsureCapacity(&colInfo, 0, numOfRows) < 0) {
+      if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
-        taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock);
+        goto FAIL;
-        return NULL;
       }
-      taosArrayPush(pArray, &colInfo);
+      taosArrayPush(*ppCols, &colInfo);
       colMeta++;
       colNeed++;
     }
   }
 
+  int32_t colActual = taosArrayGetSize(*ppCols);
+
+  // TODO in stream shuffle case, fetch groupId
+  *pGroupId = 0;
+
   STSRowIter iter = {0};
   tdSTSRowIterInit(&iter, pTschema);
   STSRow* row;
@@ -159,22 +154,22 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
   while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
     tdSTSRowIterReset(&iter, row);
     // get all wanted col of that block
-    int32_t colTot = taosArrayGetSize(pArray);
+    for (int32_t i = 0; i < colActual; i++) {
-    for (int32_t i = 0; i < colTot; i++) {
+      SColumnInfoData* pColData = taosArrayGet(*ppCols, i);
-      SColumnInfoData* pColData = taosArrayGet(pArray, i);
       SCellVal         sVal = {0};
       if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
         break;
       }
-      /*if (colDataAppend(pColData, curRow, sVal.val, false) < 0) {*/
       if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
-        taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock);
+        goto FAIL;
-        return NULL;
       }
     }
     curRow++;
   }
-  return pArray;
+  return 0;
+FAIL:
+  taosArrayDestroy(*ppCols);
+  return -1;
 }
 
 void tqReadHandleSetColIdList(STqReadHandle* pReadHandle, SArray* pColIdList) { pReadHandle->pColIdList = pColIdList; }

source/libs/executor/src/scanoperator.c

@@ -30,13 +30,12 @@
 #include "query.h"
 #include "tcompare.h"
 #include "thash.h"
-#include "vnode.h"
 #include "ttypes.h"
+#include "vnode.h"
 
 #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
 #define SWITCH_ORDER(n)              (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
 
-
 void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
   for (int32_t i = 0; i < numOfOutput; ++i) {
     SWITCH_ORDER(pCtx[i].order);
@@ -189,13 +188,13 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
   taosMemoryFreeClear(pBlock->pBlockAgg);
 
   if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {
-    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey,
+    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
-           pBlockInfo->window.ekey, pBlockInfo->rows);
+           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
     pCost->filterOutBlocks += 1;
     return TSDB_CODE_SUCCESS;
   } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
-    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey,
+    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
-           pBlockInfo->window.ekey, pBlockInfo->rows);
+           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
     pCost->skipBlocks += 1;
     return TSDB_CODE_SUCCESS;
   } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
@@ -249,8 +248,8 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
   doFilter(pTableScanInfo->pFilterNode, pBlock);
   if (pBlock->info.rows == 0) {
     pCost->filterOutBlocks += 1;
-    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey,
+    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
-           pBlockInfo->window.ekey, pBlockInfo->rows);
+           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
   }
 
   return TSDB_CODE_SUCCESS;
@@ -376,9 +375,10 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator, bool* newgroup) {
   return p;
 }
 
-SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t dataLoadFlag,
+SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput,
-                                           int32_t repeatTime, int32_t reverseTime, SArray* pColMatchInfo, SSDataBlock* pResBlock,
+                                           int32_t dataLoadFlag, int32_t repeatTime, int32_t reverseTime,
-                                           SNode* pCondition, SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo) {
+                                           SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition,
+                                           SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo) {
   assert(repeatTime > 0);
 
   STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
@@ -558,29 +558,42 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
     blockDataCleanup(pInfo->pRes);
 
     while (tqNextDataBlock(pInfo->readerHandle)) {
-      pTaskInfo->code = tqRetrieveDataBlockInfo(pInfo->readerHandle, pBlockInfo);
+      SArray*  pCols = NULL;
-      if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
+      uint64_t groupId;
-        terrno = pTaskInfo->code;
+      int32_t  numOfRows;
-        pOperator->status = OP_EXEC_DONE;
+      int32_t  code = tqRetrieveDataBlock(&pCols, pInfo->readerHandle, &groupId, &numOfRows);
+
+      if (code != TSDB_CODE_SUCCESS || numOfRows == 0) {
+        pTaskInfo->code = code;
         return NULL;
       }
 
-      if (pBlockInfo->rows == 0) {
+      pInfo->pRes->info.groupId = groupId;
-        break;
+      pInfo->pRes->info.rows = numOfRows;
-      }
-
-      SArray* pCols = tqRetrieveDataBlock(pInfo->readerHandle);
 
       int32_t numOfCols = pInfo->pRes->info.numOfCols;
       for (int32_t i = 0; i < numOfCols; ++i) {
-        SColumnInfoData* p = taosArrayGet(pCols, i);
         SColMatchInfo* pColMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
         if (!pColMatchInfo->output) {
           continue;
         }
 
-        ASSERT(pColMatchInfo->colId == p->info.colId);
+        bool colExists = false;
-        taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, p);
+        for (int32_t j = 0; j < taosArrayGetSize(pCols); ++j) {
+          SColumnInfoData* pResCol = taosArrayGet(pCols, j);
+          if (pResCol->info.colId == pColMatchInfo->colId) {
+            taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol);
+            colExists = true;
+            break;
+          }
+        }
+
+        // the required column does not exists in submit block, let's set it to be all null value
+        if (!colExists) {
+          SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId);
+          colInfoDataEnsureCapacity(pDst, 0, pBlockInfo->rows);
+          colDataAppendNNULL(pDst, 0, pBlockInfo->rows);
+        }
       }
 
       if (pInfo->pRes->pDataBlock == NULL) {
@@ -605,7 +618,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
   }
 }
 
-SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
+                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo) {
   SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {