Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/alter_table

include/libs/function/tudf.h

@@ -87,6 +87,7 @@ typedef struct SUdfInterBuf {
 } SUdfInterBuf;
 typedef void *UdfcFuncHandle;
 
+//low level APIs
 /**
  * setup udf
  * @param udf, in
@@ -115,6 +116,9 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t
  */
 int32_t doTeardownUdf(UdfcFuncHandle handle);
 
+void freeUdfInterBuf(SUdfInterBuf *buf);
+
+//high level APIs
 bool udfAggGetEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
 int32_t udfAggProcess(struct SqlFunctionCtx *pCtx);

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

@@ -3055,7 +3055,8 @@ static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) {
 bool tsdbNextDataBlock(tsdbReaderT pHandle) {
   STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
 
-  for (int32_t i = 0; i < taosArrayGetSize(pTsdbReadHandle->pColumns); ++i) {
+  size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
+  for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
     colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
   }

source/libs/executor/inc/executil.h

@@ -139,7 +139,7 @@ typedef struct {
   int32_t colId;
 } SStddevInterResult;
 
-void    initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult);
+void    initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order);
 void    initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);
 
 void    cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);

source/libs/executor/src/executil.c

@@ -184,7 +184,7 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
   pGroupResInfo->index     = 0;
 }
 
-static int32_t resultrowCompar1(const void* p1, const void* p2) {
+static int32_t resultrowComparAsc(const void* p1, const void* p2) {
   SResKeyPos* pp1 = *(SResKeyPos**) p1;
   SResKeyPos* pp2 = *(SResKeyPos**) p2;
 
@@ -202,7 +202,11 @@ static int32_t resultrowCompar1(const void* p1, const void* p2) {
   }
 }
 
-void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult) {
+static int32_t resultrowComparDesc(const void* p1, const void* p2) {
+  return resultrowComparAsc(p2, p1);
+}
+
+void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) {
   if (pGroupResInfo->pRows != NULL) {
     taosArrayDestroy(pGroupResInfo->pRows);
   }
@@ -224,8 +228,9 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, boo
     taosArrayPush(pGroupResInfo->pRows, &p);
   }
 
-  if (sortGroupResult) {
-    qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, resultrowCompar1);
+  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
+    __compar_fn_t fn = (order == TSDB_ORDER_ASC)? resultrowComparAsc:resultrowComparDesc;
+    qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn);
   }
 
   pGroupResInfo->index = 0;

source/libs/executor/src/executorimpl.c

@@ -3707,7 +3707,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
   finalizeMultiTupleQueryResult(pOperator->numOfExprs, pAggInfo->aggSup.pResultBuf, &pAggInfo->binfo.resultRowInfo,
                                 pAggInfo->binfo.rowCellInfoOffset);
 
-  initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, false);
+  initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0);
   OPTR_SET_OPENED(pOperator);
   return TSDB_CODE_SUCCESS;
 }

source/libs/executor/src/groupoperator.c

@@ -314,7 +314,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
   //  }
 
   blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
-  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, false);
+  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, 0);
 
   while(1) {
     doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);

source/libs/executor/src/scanoperator.c

@@ -102,7 +102,7 @@ static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t ord
   tw->ekey -= 1;
 }
 
-static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo) {
+static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {
   STimeWindow w = {0};
 
   // 0 by default, which means it is not a interval operator of the upstream operator.
@@ -110,13 +110,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
     return false;
   }
 
-  // todo handle the time range case
-  TSKEY sk = INT64_MIN;
-  TSKEY ek = INT64_MAX;
-  //  TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey);
-  //  TSKEY ek = MAX(pQueryAttr->window.skey, pQueryAttr->window.ekey);
-
-  if (true) {
+  if (order == TSDB_ORDER_ASC) {
     getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w);
     assert(w.ekey >= pBlockInfo->window.skey);
 
@@ -124,8 +118,8 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
       return true;
     }
 
-    while (1) {  // todo handle the desc order scan case
-      getNextTimeWindow(pInterval, &w, TSDB_ORDER_ASC);
+    while (1) {
+      getNextTimeWindow(pInterval, &w, order);
       if (w.skey > pBlockInfo->window.ekey) {
         break;
       }
@@ -136,24 +130,24 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
       }
     }
   } else {
-    //    getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w);
-    //    assert(w.skey <= pBlockInfo->window.ekey);
-    //
-    //    if (w.skey > pBlockInfo->window.skey) {
-    //      return true;
-    //    }
-    //
-    //    while(1) {
-    //      getNextTimeWindow(pQueryAttr, &w);
-    //      if (w.ekey < pBlockInfo->window.skey) {
-    //        break;
-    //      }
-    //
-    //      assert(w.skey < pBlockInfo->window.skey);
-    //      if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
-    //        return true;
-    //      }
-    //    }
+    getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey, &w);
+    assert(w.skey <= pBlockInfo->window.ekey);
+
+    if (w.skey > pBlockInfo->window.skey) {
+      return true;
+    }
+
+    while(1) {
+      getNextTimeWindow(pInterval, &w, order);
+      if (w.ekey < pBlockInfo->window.skey) {
+        break;
+      }
+
+      assert(w.skey < pBlockInfo->window.skey);
+      if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
+        return true;
+      }
+    }
   }
 
   return false;
@@ -172,7 +166,8 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
   pCost->totalRows += pBlock->info.rows;
 
   *status = pInfo->dataBlockLoadFlag;
-  if (pTableScanInfo->pFilterNode != NULL || overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info)) {
+  if (pTableScanInfo->pFilterNode != NULL ||
+      overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info, pTableScanInfo->cond.order)) {
     (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;
   }
 
@@ -188,6 +183,13 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
     qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
            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 < pBlockInfo->numOfCols; ++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;
@@ -466,6 +468,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   }
 
   pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
+//  pInfo->scanInfo = (SScanInfo){.numOfAsc = 0, .numOfDesc = 1}; // for debug purpose
 
   pInfo->readHandle        = *readHandle;
   pInfo->interval          = extractIntervalInfo(pTableScanNode);

source/libs/executor/src/timewindowoperator.c

@@ -54,8 +54,8 @@ static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols, int32_t rows,
   if (tsCols == NULL) {
     ts = ascQuery ? win->skey : win->ekey;
   } else {
-    int32_t offset = ascQuery ? 0 : rows - 1;
-    ts = tsCols[offset];
+//    int32_t offset = ascQuery ? 0 : rows - 1;
+    ts = tsCols[0];
   }
 
   return ts;
@@ -172,14 +172,22 @@ static FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_se
       }
     }
   } else {
-    int32_t end = searchFn((char*)pData, pos + 1, ekey, order);
+    int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order);
     if (end >= 0) {
-      forwardStep = pos - end;
+      forwardStep = end;
 
-      if (pData[end] == ekey) {
+      if (pData[end + pos] == ekey) {
         forwardStep += 1;
       }
     }
+//    int32_t end = searchFn((char*)pData, pos + 1, ekey, order);
+//    if (end >= 0) {
+//      forwardStep = pos - end;
+//
+//      if (pData[end] == ekey) {
+//        forwardStep += 1;
+//      }
+//    }
   }
 
   assert(forwardStep >= 0);
@@ -203,17 +211,25 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
   if (order == TSDB_ORDER_DESC) {
     // find the first position which is smaller than the key
     while (1) {
-      if (key >= keyList[lastPos]) return lastPos;
-      if (key == keyList[firstPos]) return firstPos;
-      if (key < keyList[firstPos]) return firstPos - 1;
+      if (key >= keyList[firstPos]) return firstPos;
+      if (key == keyList[lastPos]) return lastPos;
+
+      if (key < keyList[lastPos]) {
+        lastPos += 1;
+        if (lastPos >= num) {
+          return -1;
+        } else {
+          return lastPos;
+        }
+      }
 
       numOfRows = lastPos - firstPos + 1;
       midPos = (numOfRows >> 1) + firstPos;
 
       if (key < keyList[midPos]) {
-        lastPos = midPos - 1;
-      } else if (key > keyList[midPos]) {
         firstPos = midPos + 1;
+      } else if (key > keyList[midPos]) {
+        lastPos = midPos - 1;
       } else {
         break;
       }
@@ -273,12 +289,12 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary
     if (ekey > pDataBlockInfo->window.skey && pPrimaryColumn) {
       num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn);
       if (item != NULL) {
-        item->lastKey = pPrimaryColumn[startPos - (num - 1)] + step;
+        item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
       }
     } else {
-      num = startPos + 1;
+      num = pDataBlockInfo->rows - startPos;
       if (item != NULL) {
-        item->lastKey = pDataBlockInfo->window.skey + step;
+        item->lastKey = pDataBlockInfo->window.ekey + step;
       }
     }
   }
@@ -470,20 +486,17 @@ static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext,
     return -1;
   }
 
-  TSKEY   startKey = ascQuery ? pNext->skey : pNext->ekey;
+  TSKEY   skey = ascQuery ? pNext->skey : pNext->ekey;
   int32_t startPos = 0;
 
   // tumbling time window query, a special case of sliding time window query
   if (pInterval->sliding == pInterval->interval && prevPosition != -1) {
-    int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
-    startPos = prevPosition + factor;
+    startPos = prevPosition + 1;
   } else {
-    if (startKey <= pDataBlockInfo->window.skey && ascQuery) {
+    if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) {
       startPos = 0;
-    } else if (startKey >= pDataBlockInfo->window.ekey && !ascQuery) {
-      startPos = pDataBlockInfo->rows - 1;
     } else {
-      startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, startKey, order);
+      startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order);
     }
   }
 
@@ -608,7 +621,7 @@ static void saveDataBlockLastRow(char** pRow, SArray* pDataBlock, int32_t rowInd
 }
 
 static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pSDataBlock,
-                               int32_t tableGroupId) {
+                               uint64_t tableGroupId) {
   SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
 
   SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
@@ -620,7 +633,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
   }
 
   int32_t step = 1;
-  bool    ascScan = true;
+  bool    ascScan = (pInfo->order == TSDB_ORDER_ASC);
 
   //  int32_t prevIndex = pResultRowInfo->curPos;
 
@@ -630,7 +643,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
     tsCols = (int64_t*)pColDataInfo->pData;
   }
 
-  int32_t startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
+  int32_t startPos = 0;
   TSKEY   ts = getStartTsKey(&pSDataBlock->info.window, tsCols, pSDataBlock->info.rows, ascScan);
 
   STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
@@ -654,9 +667,10 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
   }
 
   int32_t forwardStep = 0;
-  TSKEY   ekey = win.ekey;
+  TSKEY   ekey = ascScan? win.ekey:win.skey;
   forwardStep =
-      getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
+      getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
+  ASSERT(forwardStep > 0);
 
   // prev time window not interpolation yet.
   //  int32_t curIndex = pResultRowInfo->curPos;
@@ -731,9 +745,9 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
       taosArrayPush(pUpdated, &pos);
     }
 
-    ekey = nextWin.ekey;  // reviseWindowEkey(pQueryAttr, &nextWin);
+    ekey = ascScan? nextWin.ekey:nextWin.skey;
     forwardStep =
-        getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
+        getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
 
     // window start(end) key interpolation
     doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, forwardStep,
@@ -761,7 +775,8 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
   SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
   SIntervalAggOperatorInfo* pInfo = pOperator->info;
 
-  int32_t        order = TSDB_ORDER_ASC;
+  int32_t scanFlag = MAIN_SCAN;
+
   SOperatorInfo* downstream = pOperator->pDownstream[0];
 
   while (1) {
@@ -773,8 +788,10 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
       break;
     }
 
+    getTableScanInfo(pOperator, &pInfo->order, &scanFlag);
+
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, scanFlag, true);
     STableQueryInfo* pTableQueryInfo = pInfo->pCurrent;
 
     setIntervalQueryRange(pTableQueryInfo, pBlock->info.window.skey, &pTaskInfo->window);
@@ -800,7 +817,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
   finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pInfo->binfo.resultRowInfo,
                                 pInfo->binfo.rowCellInfoOffset);
 
-  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true);
+  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->order);
   OPTR_SET_OPENED(pOperator);
   return TSDB_CODE_SUCCESS;
 }
@@ -945,7 +962,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
   finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo,
                                 pBInfo->rowCellInfoOffset);
 
-  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true);
+  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
   doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
   if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
@@ -1070,6 +1087,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       doClearWindows(pInfo, pOperator->numOfExprs, pBlock);
       continue;
     }
+    pInfo->order = TSDB_ORDER_ASC;
     pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
   }
 
@@ -1119,7 +1137,6 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
 
   pInfo->order = TSDB_ORDER_ASC;
   pInfo->interval = *pInterval;
-  //  pInfo->execModel      = OPTR_EXEC_MODEL_STREAM;
   pInfo->execModel = pTaskInfo->execModel;
   pInfo->win = pTaskInfo->window;
   pInfo->twAggSup = *pTwAggSupp;
@@ -1338,7 +1355,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
   finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo,
                                 pBInfo->rowCellInfoOffset);
 
-  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true);
+  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
   doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
   if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {

source/libs/function/src/tudf.c

@@ -795,7 +795,6 @@ int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SS
   }
   output->info.hasVarCol = hasVarCol;
 
-  //TODO: free the array output->pDataBlock
   output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
   for (int32_t i = 0; i < numOfCols; ++i) {
     taosArrayPush(output->pDataBlock, (input + i)->columnData);
@@ -809,8 +808,12 @@ int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) {
     return -1;
   }
   output->numOfRows = input->info.rows;
-  //TODO: memory
-  output->columnData = taosArrayGet(input->pDataBlock, 0);
+
+  output->columnData = taosMemoryMalloc(sizeof(SColumnInfoData));
+  memcpy(output->columnData,
+         taosArrayGet(input->pDataBlock, 0),
+         sizeof(SColumnInfoData));
+
   return 0;
 }
 
@@ -833,7 +836,7 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *
   fnDebug("udfc get uv task result. task: %p, uvTask: %p", task, uvTask);
   if (uvTask->type == UV_TASK_REQ_RSP) {
     if (uvTask->rspBuf.base != NULL) {
-      SUdfResponse rsp;
+      SUdfResponse rsp = {0};
       void* buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp);
       assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base));
       task->errCode = rsp.code;
@@ -1427,7 +1430,10 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t
   int32_t err = callUdf(handle, callType, &inputBlock, NULL, NULL, &resultBlock, NULL);
   if (err == 0) {
     convertDataBlockToScalarParm(&resultBlock, output);
+    taosArrayDestroy(resultBlock.pDataBlock);
   }
+
+  taosArrayDestroy(inputBlock.pDataBlock);
   return err;
 }
 
@@ -1508,16 +1514,15 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) {
 
   udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);
 
-  SUdfTeardownResponse *rsp = &task->_teardown.rsp;
   int32_t err = task->errCode;
 
   udfcRunUdfUvTask(task, UV_TASK_DISCONNECT);
 
+  fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
+
   taosMemoryFree(task->session);
   taosMemoryFree(task);
 
-  fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
-
   return err;
 }
 
@@ -1564,6 +1569,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
   }
   udfRes->interResNum = buf.numOfResult;
   memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
+  freeUdfInterBuf(&buf);
   return true;
 }
 
@@ -1621,7 +1627,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
   blockDataDestroy(inputBlock);
   taosArrayDestroy(tempBlock.pDataBlock);
 
-  taosMemoryFree(newState.buf);
+  freeUdfInterBuf(&newState);
   return udfCode;
 }
 
@@ -1650,6 +1656,8 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock) {
     GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum;
   }
 
+  freeUdfInterBuf(&resultBuf);
+
   int32_t numOfResults = functionFinalizeWithResultBuf(pCtx, pBlock, udfRes->finalResBuf);
   releaseUdfFuncHandle(pCtx->udfName);
   return udfCallCode == 0 ? numOfResults : udfCallCode;

source/libs/function/src/udfd.c

@@ -96,10 +96,14 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf);
 
 int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
   strcpy(udf->name, udfName);
+  int32_t err = 0;
+  err = udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf);
+  if (err != 0) {
+    fnError("can not retrieve udf from mnode. udf name %s", udfName);
+    return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
+  }
 
-  udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf);
-  //strcpy(udf->path, "/home/slzhou/TDengine/debug/build/lib/libudf1.so");
-  int err = uv_dlopen(udf->path, &udf->lib);
+  err = uv_dlopen(udf->path, &udf->lib);
   if (err != 0) {
     fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
     return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
@@ -142,7 +146,7 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
 
 void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
     // TODO: tracable id from client. connect, setup, call, teardown
-    fnInfo("%" PRId64 " setup request. udf name: %s", request->seqNum, request->setup.udfName);
+    fnInfo( "setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName);
     SUdfSetupRequest *setup = &request->setup;
     int32_t code = TSDB_CODE_SUCCESS;
     SUdf *udf = NULL;
@@ -276,7 +280,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
 
 void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
   SUdfTeardownRequest *teardown = &request->teardown;
-  fnInfo("teardown. %" PRId64 "handle:%" PRIx64, request->seqNum, teardown->udfHandle);
+  fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle);
   SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle);
   SUdf *udf = handle->udf;
   bool  unloadUdf = false;
@@ -800,17 +804,11 @@ static int32_t udfdRun() {
   global.udfsHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
   uv_mutex_init(&global.udfsMutex);
 
-  if (udfdUvInit() != 0) {
-    fnError("uv init failure");
-    return -2;
-  }
-
   fnInfo("start the udfd");
   int code = uv_run(global.loop, UV_RUN_DEFAULT);
   fnInfo("udfd stopped. result: %s, code: %d", uv_err_name(code), code);
   int codeClose = uv_loop_close(global.loop);
   fnDebug("uv loop close. result: %s", uv_err_name(codeClose));
-  removeListeningPipe();
   uv_mutex_destroy(&global.udfsMutex);
   taosHashCleanup(global.udfsHash);
   return 0;
@@ -853,8 +851,14 @@ int main(int argc, char *argv[]) {
     return -4;
   }
 
+  if (udfdUvInit() != 0) {
+    fnError("uv init failure");
+    return -5;
+  }
+
   udfdRun();
 
-  udfdCloseClientRpc();
+  removeListeningPipe();
 
+  udfdCloseClientRpc();
 }

source/libs/function/test/runUdf.c

@@ -34,20 +34,13 @@ static int32_t initLog() {
   return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0);
 }
 
-int main(int argc, char *argv[]) {
-  parseArgs(argc, argv);
-  initLog();
-  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
-    fnError("failed to start since read config error");
-    return -1;
-  }
-
-  udfcOpen();
-  uv_sleep(1000);
-
+int scalarFuncTest() {
   UdfcFuncHandle handle;
 
-  doSetupUdf("udf1", &handle);
+  if (doSetupUdf("udf1", &handle) != 0) {
+    fnError("setup udf failure");
+    return -1;
+  }
 
   SSDataBlock  block = {0};
   SSDataBlock *pBlock = █
@@ -74,11 +67,78 @@ int main(int argc, char *argv[]) {
   input.columnData = taosArrayGet(pBlock->pDataBlock, 0);
   SScalarParam output = {0};
   doCallUdfScalarFunc(handle, &input, 1, &output);
-
+  taosArrayDestroy(pBlock->pDataBlock);
   SColumnInfoData *col = output.columnData;
   for (int32_t i = 0; i < output.numOfRows; ++i) {
     fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t)));
   }
+
+  colDataDestroy(output.columnData);
+  taosMemoryFree(output.columnData);
+
   doTeardownUdf(handle);
+
+  return 0;
+}
+
+int aggregateFuncTest() {
+  UdfcFuncHandle handle;
+
+  if (doSetupUdf("udf2", &handle) != 0) {
+    fnError("setup udf failure");
+    return -1;
+  }
+
+  SSDataBlock  block = {0};
+  SSDataBlock *pBlock = &block;
+  pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
+  pBlock->info.numOfCols = 1;
+  pBlock->info.rows = 4;
+  char data[16] = {0};
+  char bitmap[4] = {0};
+  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
+    SColumnInfoData colInfo = {0};
+    colInfo.info.type = TSDB_DATA_TYPE_INT;
+    colInfo.info.bytes = sizeof(int32_t);
+    colInfo.info.colId = 1;
+    colInfo.pData = data;
+    colInfo.nullbitmap = bitmap;
+    for (int32_t j = 0; j < pBlock->info.rows; ++j) {
+      colDataAppendInt32(&colInfo, j, &j);
+    }
+    taosArrayPush(pBlock->pDataBlock, &colInfo);
+  }
+
+  SUdfInterBuf buf = {0};
+  SUdfInterBuf newBuf = {0};
+  SUdfInterBuf resultBuf = {0};
+  doCallUdfAggInit(handle, &buf);
+  doCallUdfAggProcess(handle, pBlock, &buf, &newBuf);
+  taosArrayDestroy(pBlock->pDataBlock);
+
+  doCallUdfAggFinalize(handle, &newBuf, &resultBuf);
+  fprintf(stderr, "agg result: %f\n", *(double*)resultBuf.buf);
+
+  freeUdfInterBuf(&buf);
+  freeUdfInterBuf(&newBuf);
+  freeUdfInterBuf(&resultBuf);
+  doTeardownUdf(handle);
+
+  return 0;
+}
+
+int main(int argc, char *argv[]) {
+  parseArgs(argc, argv);
+  initLog();
+  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
+    fnError("failed to start since read config error");
+    return -1;
+  }
+
+  udfcOpen();
+  uv_sleep(1000);
+
+  scalarFuncTest();
+  aggregateFuncTest();
   udfcClose();
 }

tests/system-test/7-tmq/subscribeDb.py

@@ -251,7 +251,7 @@ class TDTestCase:
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
         topicList    = topicName1
         ifcheckdata  = 0
-        ifManualCommit = 0
+        ifManualCommit = 1
         keyList      = 'group.id:cgrp1,\
                         enable.auto.commit:false,\
                         auto.commit.interval.ms:6000,\
@@ -410,7 +410,7 @@ class TDTestCase:
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
         topicList    = topicName1
         ifcheckdata  = 0
-        ifManualCommit = 0
+        ifManualCommit = 1
         keyList      = 'group.id:cgrp1,\
                         enable.auto.commit:false,\
                         auto.commit.interval.ms:6000,\
@@ -487,7 +487,7 @@ class TDTestCase:
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
         topicList    = topicName1
         ifcheckdata  = 0
-        ifManualCommit = 0
+        ifManualCommit = 1
         keyList      = 'group.id:cgrp1,\
                         enable.auto.commit:false,\
                         auto.commit.interval.ms:6000,\
@@ -659,7 +659,7 @@ class TDTestCase:
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
         topicList    = topicName1 + ',' + topicName2
         ifcheckdata  = 0
-        ifManualCommit = 0
+        ifManualCommit = 1
         keyList      = 'group.id:cgrp1,\
                         enable.auto.commit:false,\
                         auto.commit.interval.ms:6000,\
@@ -708,13 +708,13 @@ class TDTestCase:
         cfgPath = buildPath + "/../sim/psim/cfg"
         tdLog.info("cfgPath: %s" % cfgPath)
 
-        #self.tmqCase1(cfgPath, buildPath)
-        #self.tmqCase2(cfgPath, buildPath) 
-        #self.tmqCase3(cfgPath, buildPath)
+        self.tmqCase1(cfgPath, buildPath)
+        self.tmqCase2(cfgPath, buildPath) 
+        self.tmqCase3(cfgPath, buildPath)
         self.tmqCase4(cfgPath, buildPath)
-        #self.tmqCase5(cfgPath, buildPath)
+        self.tmqCase5(cfgPath, buildPath)
         self.tmqCase6(cfgPath, buildPath)
-        #self.tmqCase7(cfgPath, buildPath)
+        self.tmqCase7(cfgPath, buildPath)
 
 
     def stop(self):