Merge pull request #13902 from taosdata/feature/3_liaohj

fix(query): remove group info during sort.

source/libs/executor/inc/executorimpl.h

@@ -696,10 +696,6 @@ typedef struct SSortedMergeOperatorInfo {
   int32_t          numOfResPerPage;
   char**           groupVal;
   SArray          *groupInfo;
-
-  bool hasGroupId;
-  uint64_t groupId;
-  STupleHandle* prefetchedTuple;
 } SSortedMergeOperatorInfo;
 
 typedef struct SSortOperatorInfo {
@@ -712,10 +708,6 @@ typedef struct SSortOperatorInfo {
 
   int64_t      startTs;       // sort start time
   uint64_t     sortElapsed;   // sort elapsed time, time to flush to disk not included.
-
-  STupleHandle *prefetchedTuple;
-  bool  hasGroupId;
-  uint64_t groupId;
 } SSortOperatorInfo;
 
 typedef struct STagFilterOperatorInfo {

source/libs/executor/inc/tsort.h

@@ -130,12 +130,6 @@ bool tsortIsNullVal(STupleHandle* pVHandle, int32_t colId);
  */
 void* tsortGetValue(STupleHandle* pVHandle, int32_t colId);
 
-/**
- *
- * @param pVHandle
- * @return
- */
-uint64_t tsortGetGroupId(STupleHandle* pVHandle);
 /**
  *
  * @param pSortHandle

source/libs/executor/src/executorimpl.c

@@ -3031,31 +3031,12 @@ SSDataBlock* getSortedMergeBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo
   blockDataEnsureCapacity(p, capacity);
 
   while (1) {
-    STupleHandle* pTupleHandle = NULL;
-    if (pInfo->prefetchedTuple == NULL) {
-      pTupleHandle = tsortNextTuple(pHandle);
-    } else {
-      pTupleHandle = pInfo->prefetchedTuple;
-      pInfo->groupId = tsortGetGroupId(pTupleHandle);
-      pInfo->prefetchedTuple = NULL;
-    }
-
+    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
     if (pTupleHandle == NULL) {
       break;
     }
 
-    uint64_t tupleGroupId = tsortGetGroupId(pTupleHandle);
-    if (!pInfo->hasGroupId) {
-      pInfo->groupId = tupleGroupId;
-      pInfo->hasGroupId = true;
-      appendOneRowToDataBlock(p, pTupleHandle);
-    } else if (pInfo->groupId == tupleGroupId) {
     appendOneRowToDataBlock(p, pTupleHandle);
-    } else {
-      pInfo->prefetchedTuple = pTupleHandle;
-      break;
-    }
-
     if (p->info.rows >= capacity) {
       break;
     }
@@ -3074,7 +3055,6 @@ SSDataBlock* getSortedMergeBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo
 
     pDataBlock->info.rows = p->info.rows;
     pDataBlock->info.capacity = p->info.rows;
-    pDataBlock->info.groupId = pInfo->groupId;
   }
 
   blockDataDestroy(p);
@@ -3340,7 +3320,7 @@ static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
     doSetOperatorCompleted(pOperator);
   }
 
-  size_t rows = blockDataGetNumOfRows(pInfo->pRes);  // pInfo->pRes : NULL;
+  size_t rows = blockDataGetNumOfRows(pInfo->pRes);
   pOperator->resultInfo.totalRows += rows;
 
   return (rows == 0) ? NULL : pInfo->pRes;
@@ -4920,7 +4900,10 @@ SArray* extractColumnInfo(SNodeList* pNodeList) {
 }
 
 SArray* extractPartitionColInfo(SNodeList* pNodeList) {
-  if (!pNodeList) return NULL;
+  if(!pNodeList) {
+      return NULL;
+  }
+
   size_t  numOfCols = LIST_LENGTH(pNodeList);
   SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
   if (pList == NULL) {

source/libs/executor/src/scanoperator.c

@@ -2175,25 +2175,13 @@ SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, int32_t capa
       pTupleHandle = tsortNextTuple(pHandle);
     } else {
       pTupleHandle = pInfo->prefetchedTuple;
-      pInfo->groupId = tsortGetGroupId(pTupleHandle);
-      pInfo->prefetchedTuple = NULL;
     }
 
     if (pTupleHandle == NULL) {
       break;
     }
 
-    uint64_t tupleGroupId = tsortGetGroupId(pTupleHandle);
-    if (!pInfo->hasGroupId) {
-      pInfo->groupId = tupleGroupId;
-      pInfo->hasGroupId = true;
-      appendOneRowToDataBlock(p, pTupleHandle);
-    } else if (pInfo->groupId == tupleGroupId) {
     appendOneRowToDataBlock(p, pTupleHandle);
-    } else {
-      pInfo->prefetchedTuple = pTupleHandle;
-      break;
-    }
 
     if (p->info.rows >= capacity) {
       break;

source/libs/executor/src/sortoperator.c

@@ -42,8 +42,6 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR
 
   pInfo->pSortInfo        = pSortInfo;
   pInfo->pColMatchInfo    = pColMatchColInfo;
-  pInfo->hasGroupId = false;
-  pInfo->prefetchedTuple = NULL;
   pOperator->name         = "SortOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT;
   pOperator->blocking     = true;
@@ -97,31 +95,12 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
   blockDataEnsureCapacity(p, capacity);
 
   while (1) {
-    STupleHandle* pTupleHandle = NULL;
-    if (pInfo->prefetchedTuple == NULL) {
-      pTupleHandle = tsortNextTuple(pHandle);
-    } else {
-      pTupleHandle = pInfo->prefetchedTuple;
-      pInfo->groupId = tsortGetGroupId(pTupleHandle);
-      pInfo->prefetchedTuple = NULL;
-    }
-
+    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
     if (pTupleHandle == NULL) {
       break;
     }
 
-    uint64_t tupleGroupId = tsortGetGroupId(pTupleHandle);
-    if (!pInfo->hasGroupId) {
-      pInfo->groupId = tupleGroupId;
-      pInfo->hasGroupId = true;
     appendOneRowToDataBlock(p, pTupleHandle);
-    } else if (pInfo->groupId == tupleGroupId) {
-      appendOneRowToDataBlock(p, pTupleHandle);
-    } else {
-      pInfo->prefetchedTuple = pTupleHandle;
-      break;
-    }
-
     if (p->info.rows >= capacity) {
       break;
     }
@@ -140,7 +119,6 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
 
     pDataBlock->info.rows = p->info.rows;
     pDataBlock->info.capacity = p->info.rows;
-    pDataBlock->info.groupId = pInfo->groupId;
   }
 
   blockDataDestroy(p);
@@ -255,10 +233,7 @@ typedef struct SMultiwaySortMergeOperatorInfo {
 
   SSDataBlock* pInputBlock;
   int64_t      startTs;  // sort start time
-
-  bool          hasGroupId;
   uint64_t     groupId;
-  STupleHandle* prefetchedTuple;
 } SMultiwaySortMergeOperatorInfo;
 
 int32_t doOpenMultiwaySortMergeOperator(SOperatorInfo* pOperator) {
@@ -312,31 +287,12 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
   blockDataEnsureCapacity(p, capacity);
 
   while (1) {
-    STupleHandle* pTupleHandle = NULL;
-    if (pInfo->prefetchedTuple == NULL) {
-      pTupleHandle = tsortNextTuple(pHandle);
-    } else {
-      pTupleHandle = pInfo->prefetchedTuple;
-      pInfo->groupId = tsortGetGroupId(pTupleHandle);
-      pInfo->prefetchedTuple = NULL;
-    }
-
+    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
     if (pTupleHandle == NULL) {
       break;
     }
 
-    uint64_t tupleGroupId = tsortGetGroupId(pTupleHandle);
-    if (!pInfo->hasGroupId) {
-      pInfo->groupId = tupleGroupId;
-      pInfo->hasGroupId = true;
-      appendOneRowToDataBlock(p, pTupleHandle);
-    } else if (pInfo->groupId == tupleGroupId) {
     appendOneRowToDataBlock(p, pTupleHandle);
-    } else {
-      pInfo->prefetchedTuple = pTupleHandle;
-      break;
-    }
-
     if (p->info.rows >= capacity) {
       break;
     }
@@ -432,14 +388,12 @@ SOperatorInfo* createMultiwaySortMergeOperatorInfo(SOperatorInfo** downStreams,
   pOperator->blocking  = false;
   pOperator->status    = OP_NOT_OPENED;
   pOperator->info      = pInfo;
-  pInfo->hasGroupId    = false;
-  pInfo->prefetchedTuple = NULL;
   pOperator->pTaskInfo = pTaskInfo;
 
   pInfo->bufPageSize   = getProperSortPageSize(rowSize);
 
   uint32_t numOfSources = taosArrayGetSize(pSortInfo);
-  numOfSources = TMAX(2, numOfSources);
+  numOfSources = TMAX(4, numOfSources);
 
   pInfo->sortBufSize    = numOfSources * pInfo->bufPageSize;
 

source/libs/executor/src/tsort.c

@@ -557,36 +557,21 @@ static int32_t createInitialSources(SSortHandle* pHandle) {
     SSortSource* source = taosArrayGetP(pHandle->pOrderedSource, 0);
     taosArrayClear(pHandle->pOrderedSource);
 
-    bool hasGroupId = false;
-    SSDataBlock* prefetchedDataBlock = NULL;
-
     while (1) {
-      SSDataBlock* pBlock = NULL;
-      if (prefetchedDataBlock == NULL) {
-        pBlock = pHandle->fetchfp(source->param);
-      } else {
-        pBlock = prefetchedDataBlock;
-        prefetchedDataBlock = NULL;
-      }
-
+      SSDataBlock* pBlock = pHandle->fetchfp(source->param);
       if (pBlock == NULL) {
         break;
       }
 
-      if (!hasGroupId) {
-        // calculate the buffer pages according to the total available buffers.
+      if (pHandle->pDataBlock == NULL) {
         pHandle->pageSize = getProperSortPageSize(blockDataGetRowSize(pBlock));
 
         // todo, number of pages are set according to the total available sort buffer
         pHandle->numOfPages = 1024;
         sortBufSize = pHandle->numOfPages * pHandle->pageSize;
-
-        hasGroupId = true;
         pHandle->pDataBlock = createOneDataBlock(pBlock, false);
       }
 
-      if (pHandle->pDataBlock->info.groupId == pBlock->info.groupId) {
-        // perform the scalar function calculation before apply the sort
       if (pHandle->beforeFp != NULL) {
         pHandle->beforeFp(pBlock, pHandle->param);
       }
@@ -607,10 +592,6 @@ static int32_t createInitialSources(SSortHandle* pHandle) {
 
         doAddToBuf(pHandle->pDataBlock, pHandle);
       }
-      } else {
-        prefetchedDataBlock = pBlock;
-        pHandle->pDataBlock = createOneDataBlock(pBlock, false);
-      }
     }
 
     if (pHandle->pDataBlock != NULL && pHandle->pDataBlock->info.rows > 0) {
@@ -758,10 +739,6 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colIndex) {
   }
 }
 
-uint64_t tsortGetGroupId(STupleHandle* pVHandle) {
-  return pVHandle->pBlock->info.groupId;
-}
-
 SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) {
   SSortExecInfo info = {0};
 

taos-tools @ 3d5aa76f

-Subproject commit 0a81480420d6601bbdb57770ee64e40f24c4ea83
+Subproject commit 3d5aa76f8c718dcffa100b45e4cbf313d499c356