refactor(query): do some internal refactor.

include/libs/function/function.h

@@ -92,6 +92,8 @@ struct SResultRowEntryInfo;
 //for selectivity query, the corresponding tag value is assigned if the data is qualified
 typedef struct SSubsidiaryResInfo {
   int16_t num;
+  int32_t rowLen;
+  char*   buf;         // serialize data buffer
   struct SqlFunctionCtx **pCtx;
 } SSubsidiaryResInfo;
 
@@ -118,6 +120,11 @@ typedef struct SInputColumnInfoData {
   uint64_t          uid;            // table uid, used to set the tag value when building the final query result for selectivity functions.
 } SInputColumnInfoData;
 
+typedef struct SSerializeDataHandle {
+  struct SDiskbasedBuf* pBuf;
+  int32_t               currentPage;
+} SSerializeDataHandle;
+
 // sql function runtime context
 typedef struct SqlFunctionCtx {
   SInputColumnInfoData   input;
@@ -137,10 +144,9 @@ typedef struct SqlFunctionCtx {
   SFuncExecFuncs         fpSet;
   SScalarFuncExecFuncs   sfp;
   struct SExprInfo      *pExpr;
-  struct SDiskbasedBuf  *pBuf;
   struct SSDataBlock    *pSrcBlock;
   struct SSDataBlock    *pDstBlock; // used by indefinite rows function to set selectivity
-  int32_t                curBufPage;
+  SSerializeDataHandle   saveHandle;
   bool                   isStream;
 
   char                   udfName[TSDB_FUNC_NAME_LEN];

include/util/tpagedbuf.h

@@ -58,11 +58,10 @@ int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMem
 /**
  *
  * @param pBuf
- * @param groupId
  * @param pageId
  * @return
  */
-void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t groupId, int32_t* pageId);
+void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId);
 
 /**
  *

source/libs/executor/src/executil.c

@@ -46,8 +46,8 @@ size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
     rowSize += pCtx[i].resDataInfo.interBufSize;
   }
 
-  rowSize +=
-      (numOfOutput * sizeof(bool));  // expand rowSize to mark if col is null for top/bottom result(doSaveTupleData)
+  rowSize += (numOfOutput * sizeof(bool));
+  // expand rowSize to mark if col is null for top/bottom result(saveTupleData)
   return rowSize;
 }
 
@@ -1175,7 +1175,6 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
     SqlFunctionCtx* pCtx = &pFuncCtx[i];
 
     pCtx->functionId = -1;
-    pCtx->curBufPage = -1;
     pCtx->pExpr = pExpr;
 
     if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
@@ -1219,6 +1218,7 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
     pCtx->isStream = false;
 
     pCtx->param = pFunct->pParam;
+    pCtx->saveHandle.currentPage = -1;
   }
 
   for (int32_t i = 1; i < numOfOutput; ++i) {

source/libs/executor/src/executorimpl.c

@@ -187,7 +187,7 @@ SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int
   SIDList list = getDataBufPagesIdList(pResultBuf);
 
   if (taosArrayGetSize(list) == 0) {
-    pData = getNewBufPage(pResultBuf, tableGroupId, &pageId);
+    pData = getNewBufPage(pResultBuf, &pageId);
     pData->num = sizeof(SFilePage);
   } else {
     SPageInfo* pi = getLastPageInfo(list);
@@ -198,7 +198,7 @@ SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int
       // release current page first, and prepare the next one
       releaseBufPageInfo(pResultBuf, pi);
 
-      pData = getNewBufPage(pResultBuf, tableGroupId, &pageId);
+      pData = getNewBufPage(pResultBuf, &pageId);
       if (pData != NULL) {
         pData->num = sizeof(SFilePage);
       }
@@ -302,7 +302,7 @@ static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pRes
   SIDList list = getDataBufPagesIdList(pResultBuf);
 
   if (taosArrayGetSize(list) == 0) {
-    pData = getNewBufPage(pResultBuf, tid, &pageId);
+    pData = getNewBufPage(pResultBuf, &pageId);
     pData->num = sizeof(SFilePage);
   } else {
     SPageInfo* pi = getLastPageInfo(list);
@@ -313,7 +313,7 @@ static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pRes
       // release current page first, and prepare the next one
       releaseBufPageInfo(pResultBuf, pi);
 
-      pData = getNewBufPage(pResultBuf, tid, &pageId);
+      pData = getNewBufPage(pResultBuf, &pageId);
       if (pData != NULL) {
         pData->num = sizeof(SFilePage);
       }
@@ -3488,7 +3488,7 @@ int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInf
   }
 
   for (int32_t i = 0; i < numOfCols; ++i) {
-    pSup->pCtx[i].pBuf = pAggSup->pResultBuf;
+    pSup->pCtx[i].saveHandle.pBuf = pAggSup->pResultBuf;
   }
 
   return TSDB_CODE_SUCCESS;
@@ -3520,6 +3520,7 @@ void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
     }
 
     taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx);
+    taosMemoryFreeClear(pCtx[i].subsidiaries.buf);
     taosMemoryFree(pCtx[i].input.pData);
     taosMemoryFree(pCtx[i].input.pColumnDataAgg);
   }
@@ -4704,7 +4705,8 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlF
   }
   int32_t code = createDiskbasedBuf(&pSup->pResultBuf, pageSize, bufSize, pKey, tsTempDir);
   for (int32_t i = 0; i < numOfOutput; ++i) {
-    pCtx[i].pBuf = pSup->pResultBuf;
+    pCtx[i].saveHandle.pBuf = pSup->pResultBuf;
   }
+
   return code;
 }

source/libs/executor/src/groupoperator.c

@@ -547,7 +547,7 @@ void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInf
     p = taosHashGet(pInfo->pGroupSet, pInfo->keyBuf, len);
 
     int32_t pageId = 0;
-    pPage = getNewBufPage(pInfo->pBuf, 0, &pageId);
+    pPage = getNewBufPage(pInfo->pBuf, &pageId);
     taosArrayPush(p->pPageList, &pageId);
 
     *(int32_t *) pPage = 0;
@@ -562,7 +562,7 @@ void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInf
 
       // add a new page for current group
       int32_t pageId = 0;
-      pPage = getNewBufPage(pInfo->pBuf, 0, &pageId);
+      pPage = getNewBufPage(pInfo->pBuf, &pageId);
       taosArrayPush(p->pPageList, &pageId);
       memset(pPage, 0, getBufPageSize(pInfo->pBuf));
     }

source/libs/executor/src/projectoperator.c

@@ -195,16 +195,6 @@ static int32_t doIngroupLimitOffset(SLimitInfo* pLimitInfo, uint64_t groupId, SS
   return PROJECT_RETRIEVE_DONE;
 }
 
-void printDataBlock1(SSDataBlock* pBlock, const char* flag) {
-  if (!pBlock || pBlock->info.rows == 0) {
-    qDebug("===stream===printDataBlock: Block is Null or Empty");
-    return;
-  }
-  char* pBuf = NULL;
-  qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
-  taosMemoryFreeClear(pBuf);
-}
-
 SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
   SProjectOperatorInfo* pProjectInfo = pOperator->info;
   SOptrBasicInfo*       pInfo = &pProjectInfo->binfo;

source/libs/executor/src/timewindowoperator.c

@@ -3529,7 +3529,7 @@ int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo*
   initBasicInfo(pBasicInfo, pResultBlock);
 
   for (int32_t i = 0; i < numOfCols; ++i) {
-    pSup->pCtx[i].pBuf = NULL;
+    pSup->pCtx[i].saveHandle.pBuf = NULL;
   }
 
   ASSERT(numOfCols > 0);

source/libs/executor/src/tlinearhash.c

@@ -97,7 +97,7 @@ static int32_t doAddToBucket(SLHashObj* pHashObj, SLHashBucket* pBucket, int32_t
 
     // allocate the overflow buffer page to hold this k/v.
     int32_t newPageId = -1;
-    SFilePage* pNewPage = getNewBufPage(pHashObj->pBuf, 0, &newPageId);
+    SFilePage* pNewPage = getNewBufPage(pHashObj->pBuf, &newPageId);
     if (pNewPage == NULL) {
       return terrno;
     }
@@ -227,7 +227,7 @@ static int32_t doAddNewBucket(SLHashObj* pHashObj) {
   }
 
   int32_t pageId = -1;
-  SFilePage* p = getNewBufPage(pHashObj->pBuf, 0, &pageId);
+  SFilePage* p = getNewBufPage(pHashObj->pBuf, &pageId);
   if (p == NULL) {
     return terrno;
   }

source/libs/executor/src/tsort.c

@@ -180,7 +180,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
     }
 
     int32_t pageId = -1;
-    void* pPage = getNewBufPage(pHandle->pBuf, pHandle->sourceId, &pageId);
+    void* pPage = getNewBufPage(pHandle->pBuf, &pageId);
     if (pPage == NULL) {
       blockDataDestroy(p);
       return terrno;
@@ -512,7 +512,7 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
         }
 
         int32_t pageId = -1;
-        void* pPage = getNewBufPage(pHandle->pBuf, pHandle->sourceId, &pageId);
+        void* pPage = getNewBufPage(pHandle->pBuf, &pageId);
         if (pPage == NULL) {
           return terrno;
         }

source/libs/function/src/tpercentile.c

@@ -372,7 +372,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
         pPageIdList =  pList;
       }
 
-      pSlot->info.data = getNewBufPage(pBucket->pBuffer, groupId, &pageId);
+      pSlot->info.data = getNewBufPage(pBucket->pBuffer, &pageId);
       pSlot->info.pageId = pageId;
       taosArrayPush(pPageIdList, &pageId);
     }

source/util/src/tpagedbuf.c

@@ -371,7 +371,7 @@ int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMem
   return TSDB_CODE_SUCCESS;
 }
 
-void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t groupId, int32_t* pageId) {
+void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
   pBuf->statis.getPages += 1;
 
   char* availablePage = NULL;

source/util/test/pageBufferTest.cpp

@@ -18,7 +18,7 @@ void simpleTest() {
   int32_t pageId = 0;
   int32_t groupId = 0;
   
-  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf, &pageId));
   ASSERT_TRUE(pBufPage != NULL);
   
   ASSERT_EQ(getTotalBufSize(pBuf), 1024);
@@ -29,26 +29,26 @@ void simpleTest() {
 
   releaseBufPage(pBuf, pBufPage);
 
-  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf, &pageId));
 
   SFilePage* t = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t == pBufPage1);
 
-  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf, &pageId));
   SFilePage* t1 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t1 == pBufPage2);
 
-  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t2 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t2 == pBufPage3);
 
-  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t3 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t3 == pBufPage4);
 
   releaseBufPage(pBuf, pBufPage2);
 
-  SFilePage* pBufPage5 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage5 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t4 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t4 == pBufPage5);
 
@@ -64,7 +64,7 @@ void writeDownTest() {
   int32_t groupId = 0;
   int32_t nx = 12345;
 
-  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   ASSERT_TRUE(pBufPage != NULL);
 
   *(int32_t*)(pBufPage->data) = nx;
@@ -73,22 +73,22 @@ void writeDownTest() {
   setBufPageDirty(pBufPage, true);
   releaseBufPage(pBuf, pBufPage);
 
-  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf, &pageId));
   SFilePage* t1 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t1 == pBufPage1);
   ASSERT_TRUE(pageId == 1);
 
-  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf, &pageId));
   SFilePage* t2 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t2 == pBufPage2);
   ASSERT_TRUE(pageId == 2);
 
-  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t3 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t3 == pBufPage3);
   ASSERT_TRUE(pageId == 3);
 
-  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t4 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t4 == pBufPage4);
   ASSERT_TRUE(pageId == 4);
@@ -113,32 +113,32 @@ void recyclePageTest() {
   int32_t groupId = 0;
   int32_t nx = 12345;
 
-  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   ASSERT_TRUE(pBufPage != NULL);
   releaseBufPage(pBuf, pBufPage);
 
-  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage1 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t1 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t1 == pBufPage1);
   ASSERT_TRUE(pageId == 1);
 
-  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage2 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t2 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t2 == pBufPage2);
   ASSERT_TRUE(pageId == 2);
 
-  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage3 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t3 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t3 == pBufPage3);
   ASSERT_TRUE(pageId == 3);
 
-  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage4 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t4 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t4 == pBufPage4);
   ASSERT_TRUE(pageId == 4);
   releaseBufPage(pBuf, t4);
 
-  SFilePage* pBufPage5 = static_cast<SFilePage*>(getNewBufPage(pBuf, groupId, &pageId));
+  SFilePage* pBufPage5 = static_cast<SFilePage*>(getNewBufPage(pBuf,  &pageId));
   SFilePage* t5 = static_cast<SFilePage*>(getBufPage(pBuf, pageId));
   ASSERT_TRUE(t5 == pBufPage5);
   ASSERT_TRUE(pageId == 5);