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

fix(query):fix memory leak.

source/common/src/tdatablock.c

@@ -1174,8 +1174,6 @@ int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows)
 
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
   int32_t code = 0;
-  // ASSERT(numOfRows > 0);
-
   if (numOfRows == 0) {
     return TSDB_CODE_SUCCESS;
   }

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -234,6 +234,7 @@ static void destroyBlockScanInfo(SHashObj* pTableMap) {
     }
 
     taosArrayDestroy(p->delSkyline);
+    taosArrayDestroy(p->pBlockList);
     p->delSkyline = NULL;
   }
 
@@ -302,9 +303,9 @@ static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) {
   STimeWindow win = {0};
 
   while (1) {
-//    if (pReader->pFileReader != NULL) {
+    if (pReader->pFileReader != NULL) {
-//      tsdbDataFReaderClose(&pReader->pFileReader);
+      tsdbDataFReaderClose(&pReader->pFileReader);
-//    }
+    }
 
     pReader->status.pCurrentFileset = (SDFileSet*)taosArrayGet(pIter->pFileList, pIter->index);
 
@@ -696,12 +697,14 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, uint32_
 
       void* p = taosArrayPush(pScanInfo->pBlockList, &block);
       if (p == NULL) {
+        tMapDataClear(&mapData);
         return TSDB_CODE_OUT_OF_MEMORY;
       }
 
       (*numOfBlocks) += 1;
     }
 
+    tMapDataClear(&mapData);
     if (pScanInfo->pBlockList != NULL && taosArrayGetSize(pScanInfo->pBlockList) > 0) {
       (*numOfValidTables) += 1;
     }
@@ -1308,6 +1311,8 @@ static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIte
               pReader->idStr);
 
     pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
+
+    cleanupBlockOrderSupporter(&sup);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1990,6 +1995,7 @@ static TSDBKEY getCurrentKeyInBuf(SDataBlockIter* pBlockIter, STsdbReader* pRead
 
 static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
   SReaderStatus* pStatus = &pReader->status;
+  SArray* pIndexList = taosArrayInit(4, sizeof(SBlockIdx));
 
   while (1) {
     bool hasNext = filesetIteratorNext(&pStatus->fileIter, pReader);
@@ -1997,9 +2003,10 @@ static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
       break;
     }
 
-    SArray* pIndexList = taosArrayInit(4, sizeof(SBlockIdx));
+    taosArrayClear(pIndexList);
     int32_t code = doLoadBlockIndex(pReader, pReader->pFileReader, pIndexList);
     if (code != TSDB_CODE_SUCCESS) {
+      taosArrayDestroy(pIndexList);
       return code;
     }
 
@@ -2007,6 +2014,7 @@ static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
       uint32_t numOfValidTable = 0;
       code = doLoadFileBlock(pReader, pIndexList, &numOfValidTable, numOfBlocks);
       if (code != TSDB_CODE_SUCCESS) {
+        taosArrayDestroy(pIndexList);
         return code;
       }
 
@@ -2014,10 +2022,10 @@ static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
         break;
       }
     }
-
     // no blocks in current file, try next files
   }
 
+  taosArrayDestroy(pIndexList);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3081,10 +3089,13 @@ int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTa
 
   SDataBlockIter* pBlockIter = &pStatus->blockIter;
   pTableBlockInfo->numOfFiles += pStatus->fileIter.numOfFiles;
-  pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
+
+  if (pBlockIter->numOfBlocks > 0) {
+    pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks;
+  }
 
   pTableBlockInfo->numOfTables = numOfTables;
-  bool hasNext = true;
+  bool hasNext = (pBlockIter->numOfBlocks > 0);
 
   while (true) {
     if (hasNext) {

source/libs/executor/src/executorimpl.c

@@ -3475,9 +3475,12 @@ static SSDataBlock* doFill(SOperatorInfo* pOperator) {
 static void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs) {
   for (int32_t i = 0; i < numOfExprs; ++i) {
     SExprInfo* pExprInfo = &pExpr[i];
-    if (pExprInfo->pExpr->nodeType == QUERY_NODE_COLUMN) {
+    for(int32_t j = 0; j < pExprInfo->base.numOfParams; ++j) {
-      taosMemoryFree(pExprInfo->base.pParam[0].pCol);
+      if (pExprInfo->base.pParam[j].type == FUNC_PARAM_TYPE_COLUMN) {
+        taosMemoryFreeClear(pExprInfo->base.pParam[j].pCol);
+      }
     }
+
     taosMemoryFree(pExprInfo->base.pParam);
     taosMemoryFree(pExprInfo->pExpr);
   }
@@ -3685,10 +3688,20 @@ void destroyBasicOperatorInfo(void* param, int32_t numOfOutput) {
   taosMemoryFreeClear(param);
 }
 
+
+static void freeItem(void* pItem) {
+  void** p = pItem;
+  if (*p != NULL) {
+    taosMemoryFreeClear(*p);
+  }
+}
+
 void destroyAggOperatorInfo(void* param, int32_t numOfOutput) {
   SAggOperatorInfo* pInfo = (SAggOperatorInfo*)param;
   cleanupBasicInfo(&pInfo->binfo);
 
+  cleanupAggSup(&pInfo->aggSup);
+  taosArrayDestroyEx(pInfo->groupResInfo.pRows, freeItem);
   taosMemoryFreeClear(param);
 }
 

source/libs/executor/src/scanoperator.c

@@ -196,12 +196,6 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
            pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
     pCost->skipBlocks += 1;
 
-    // clear all data in pBlock that are set when handing the previous block
-    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
-      SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
-      pcol->pData = NULL;
-    }
-
     return TSDB_CODE_SUCCESS;
   } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
     pCost->loadBlockStatis += 1;

source/libs/function/src/builtinsimpl.c

@@ -5559,6 +5559,10 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, 0);
 
+  if (pData->totalRows == 0) {
+    pData->minRows = 0;
+  }
+
   int32_t row = 0;
   char    st[256] = {0};
   double  totalRawSize = pData->totalRows * pData->rowSize;
@@ -5570,10 +5574,14 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   varDataSetLen(st, len);
   colDataAppend(pColInfo, row++, st, false);
 
+  int64_t avgRows = 0;
+  if (pData->numOfBlocks > 0) {
+    avgRows = pData->totalRows / pData->numOfBlocks;
+  }
+
   len = sprintf(st + VARSTR_HEADER_SIZE,
                 "Total_Rows=[%" PRId64 "] Inmem_Rows=[%d] MinRows=[%d] MaxRows=[%d] Average_Rows=[%" PRId64 "]",
-                pData->totalRows, pData->numOfInmemRows, pData->minRows, pData->maxRows,
+                pData->totalRows, pData->numOfInmemRows, pData->minRows, pData->maxRows, avgRows);
-                pData->totalRows / pData->numOfBlocks);
 
   varDataSetLen(st, len);
   colDataAppend(pColInfo, row++, st, false);