refactor(query): do some internal refactor.

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

@@ -312,7 +312,7 @@ static int32_t mndSaveQueryList(SConnObj *pConn, SQueryHbReqBasic *pBasic) {
   pConn->numOfQueries = pBasic->queryDesc ? taosArrayGetSize(pBasic->queryDesc) : 0;
   pBasic->queryDesc = NULL;
 
-  mDebug("queries updated in conn %d, num:%d", pConn->id, pConn->numOfQueries);
+  mDebug("queries updated in conn %u, num:%d", pConn->id, pConn->numOfQueries);
 
   taosWUnLockLatch(&pConn->queryLock);
 

source/libs/executor/inc/executil.h

@@ -82,8 +82,6 @@ size_t getResultRowSize(struct SqlFunctionCtx* pCtx, int32_t numOfOutput);
 void   initResultRowInfo(SResultRowInfo* pResultRowInfo);
 void   cleanupResultRowInfo(SResultRowInfo* pResultRowInfo);
 
-void closeAllResultRows(SResultRowInfo* pResultRowInfo);
-
 void initResultRow(SResultRow* pResultRow);
 void closeResultRow(SResultRow* pResultRow);
 bool isResultRowClosed(SResultRow* pResultRow);

source/libs/executor/inc/executorimpl.h

@@ -108,7 +108,6 @@ typedef struct STaskCostInfo {
   SFileBlockLoadRecorder* pRecoder;
   uint64_t                elapsedTime;
 
-  uint64_t firstStageMergeTime;
   uint64_t winInfoSize;
   uint64_t tableInfoSize;
   uint64_t hashSize;
@@ -549,6 +548,7 @@ typedef struct SProjectOperatorInfo {
   SLimitInfo         limitInfo;
   bool               mergeDataBlocks;
   SSDataBlock*       pFinalRes;
+  SNode*             pCondition;
 } SProjectOperatorInfo;
 
 typedef struct SIndefOperatorInfo {

source/libs/executor/src/executil.c

@@ -43,10 +43,6 @@ void cleanupResultRowInfo(SResultRowInfo* pResultRowInfo) {
   }
 }
 
-void closeAllResultRows(SResultRowInfo* pResultRowInfo) {
-  // do nothing
-}
-
 bool isResultRowClosed(SResultRow* pRow) { return (pRow->closed == true); }
 
 void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
@@ -160,11 +156,13 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
 
 SArray* createSortInfo(SNodeList* pNodeList) {
   size_t numOfCols = 0;
+
   if (pNodeList != NULL) {
     numOfCols = LIST_LENGTH(pNodeList);
   } else {
     numOfCols = 0;
   }
+
   SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
   if (pList == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -196,10 +194,6 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
-    /*if (!pDescNode->output) {  // todo disable it temporarily*/
-    /*continue;*/
-    /*}*/
-
     SColumnInfoData idata =
         createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
     idata.info.scale = pDescNode->dataType.scale;
@@ -701,9 +695,6 @@ static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutpu
     }
   }
 
-#ifdef BUF_PAGE_DEBUG
-  qDebug("page_setSelect num:%d", num);
-#endif
   if (p != NULL) {
     p->subsidiaries.pCtx = pValCtx;
     p->subsidiaries.num = num;
@@ -852,7 +843,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
   // TODO: get it from stable scan node
   pCond->twindows = pTableScanNode->scanRange;
   pCond->suid = pTableScanNode->scan.suid;
-  pCond->type = BLOCK_LOAD_OFFSET_ORDER;
+  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
   pCond->startVersion = -1;
   pCond->endVersion = -1;
   //  pCond->type = pTableScanNode->scanFlag;
@@ -947,6 +938,7 @@ STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInter
 }
 
 // get the correct time window according to the handled timestamp
+// todo refactor
 STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
                                 int32_t order) {
   STimeWindow w = {0};

source/libs/executor/src/executorimpl.c

@@ -1665,9 +1665,6 @@ void queryCostStatis(SExecTaskInfo* pTaskInfo) {
   //  hashSize += taosHashGetMemSize(pRuntimeEnv->tableqinfoGroupInfo.map);
   //  pSummary->hashSize = hashSize;
 
-  // add the merge time
-  pSummary->elapsedTime += pSummary->firstStageMergeTime;
-
   //  SResultRowPool* p = pTaskInfo->pool;
   //  if (p != NULL) {
   //    pSummary->winInfoSize = getResultRowPoolMemSize(p);
@@ -1676,17 +1673,16 @@ void queryCostStatis(SExecTaskInfo* pTaskInfo) {
   //    pSummary->winInfoSize = 0;
   //    pSummary->numOfTimeWindows = 0;
   //  }
-  //
-  //  calculateOperatorProfResults(pQInfo);
 
   SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
   if (pSummary->pRecoder != NULL) {
-    qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64
-           " us, total blocks:%d, "
-           "load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
-           GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pRecorder->totalBlocks,
-           pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows, pRecorder->totalCheckedRows);
+    qDebug(
+        "%s :cost summary: elapsed time:%.2f ms, total blocks:%d, load block SMA:%d, load data block:%d, total rows:%"
+        PRId64 ", check rows:%" PRId64, GET_TASKID(pTaskInfo), pSummary->elapsedTime / 1000.0,
+        pRecorder->totalBlocks, pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows,
+        pRecorder->totalCheckedRows);
   }
+
   // qDebug("QInfo:0x%"PRIx64" :cost summary: winResPool size:%.2f Kb, numOfWin:%"PRId64", tableInfoSize:%.2f Kb,
   // hashTable:%.2f Kb", pQInfo->qId, pSummary->winInfoSize/1024.0,
   //      pSummary->numOfTimeWindows, pSummary->tableInfoSize/1024.0, pSummary->hashSize/1024.0);
@@ -3031,7 +3027,6 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
     }
   }
 
-  closeAllResultRows(&pAggInfo->binfo.resultRowInfo);
   initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0);
   OPTR_SET_OPENED(pOperator);
 
@@ -3400,27 +3395,33 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
       break;
     }
 
-    // no results generated
-    if (pInfo->pRes->info.rows == 0 || (!pProjectInfo->mergeDataBlocks)) {
-      break;
-    }
-
     if (pProjectInfo->mergeDataBlocks) {
-      pFinalRes->info.groupId = pInfo->pRes->info.groupId;
-      pFinalRes->info.version = pInfo->pRes->info.version;
+      if (pRes->info.rows > 0) {
+        pFinalRes->info.groupId = pRes->info.groupId;
+        pFinalRes->info.version = pRes->info.version;
 
         // continue merge data, ignore the group id
-      blockDataMerge(pFinalRes, pInfo->pRes);
-
-      if (pFinalRes->info.rows + pInfo->pRes->info.rows <= pOperator->resultInfo.threshold) {
+        blockDataMerge(pFinalRes, pRes);
+        if (pFinalRes->info.rows + pRes->info.rows <= pOperator->resultInfo.threshold) {
           continue;
         }
       }
 
       // do apply filter
-    SSDataBlock* p = pProjectInfo->mergeDataBlocks ? pFinalRes : pRes;
-    doFilter(pProjectInfo->pFilterNode, p, NULL);
-    if (p->info.rows > 0) {
+      doFilter(pProjectInfo->pFilterNode, pFinalRes, NULL);
+      if (pFinalRes->info.rows > 0 || pRes->info.rows == 0) {
+        break;
+      }
+    } else {
+      // do apply filter
+      if (pRes->info.rows > 0) {
+        doFilter(pProjectInfo->pFilterNode, pRes, NULL);
+        if (pRes->info.rows == 0) {
+          continue;
+        }
+      }
+
+      // no results generated
       break;
     }
   }
@@ -3885,7 +3886,6 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
 
   pInfo->binfo.pRes = pResBlock;
   pInfo->pFinalRes  = createOneDataBlock(pResBlock, false);
-
   pInfo->pFilterNode = pProjPhyNode->node.pConditions;
   pInfo->mergeDataBlocks = pProjPhyNode->mergeDataBlock;
 
@@ -4416,7 +4416,7 @@ static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pC
 
   pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
   pCond->suid = uid;
-  pCond->type = BLOCK_LOAD_OFFSET_ORDER;
+  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
   pCond->startVersion = -1;
   pCond->endVersion = -1;
 

source/libs/executor/src/timewindowoperator.c

@@ -1057,7 +1057,6 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
     hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag, NULL);
   }
 
-  closeAllResultRows(&pInfo->binfo.resultRowInfo);
   initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->order);
   OPTR_SET_OPENED(pOperator);
 
@@ -1213,7 +1212,6 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
   pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
 
   pOperator->status = OP_RES_TO_RETURN;
-  closeAllResultRows(&pBInfo->resultRowInfo);
 
   initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
@@ -2008,7 +2006,6 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
 
   // restore the value
   pOperator->status = OP_RES_TO_RETURN;
-  closeAllResultRows(&pBInfo->resultRowInfo);
 
   initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
@@ -2172,8 +2169,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
   SSDataBlock*            pResBlock = pSliceInfo->pRes;
   SExprSupp*              pSup = &pOperator->exprSupp;
 
-  blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
-
   //  if (pOperator->status == OP_RES_TO_RETURN) {
   //    //    doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes);
   //    if (pResBlock->info.rows == 0 || !hasDataInGroupInfo(&pSliceInfo->groupResInfo)) {