Merge remote-tracking branch 'origin/3.0' into fix/mnode

docs/en/07-develop/03-insert-data/01-sql-writing.mdx

@@ -42,7 +42,7 @@ INSERT INTO d1001 VALUES (1538548684000, 10.2, 220, 0.23) (1538548696650, 10.3,
 
 ### Insert into Multiple Tables
 
-Data can be inserted into multiple tables in the same SQL statement. The example below inserts 2 rows into table "d1001" and 1 row into table "d1002".
+Data can be inserted into multiple tables in single SQL statement. The example below inserts 2 rows into table "d1001" and 1 row into table "d1002".
 
 ```sql
 INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6, 218, 0.33) d1002 VALUES (1538548696800, 12.3, 221, 0.31);
@@ -52,15 +52,15 @@ For more details about `INSERT` please refer to [INSERT](/taos-sql/insert).
 
 :::info
 
-- Inserting in batches can improve performance. Normally, the higher the batch size, the better the performance. Please note that a single row can't exceed 48K bytes and each SQL statement can't exceed 1MB.
-- Inserting with multiple threads can also improve performance. However, depending on the system resources on the application side and the server side, when the number of inserting threads grows beyond a specific point the performance may drop instead of improving. The proper number of threads needs to be tested in a specific environment to find the best number.
+- Inserting in batches can improve performance. Normally, the higher the batch size, the better the performance. Please note that a single row can't exceed 48 KB bytes and each SQL statement can't exceed 1 MB.
+- Inserting with multiple threads can also improve performance. However, depending on the system resources on the application side and the server side, when the number of inserting threads grows beyond a specific point the performance may drop instead of improving. The proper number of threads needs to be tested in a specific environment to find the best number. The proper number of threads may be impacted by the system resources on the server side, the system resources on the client side, the table schemas, etc. 
 
 :::
 
 :::warning
 
-- If the timestamp for the row to be inserted already exists in the table, the behavior depends on the value of parameter `UPDATE`. If it's set to 0 (the default value), the row will be discarded. If it's set to 1, the new values will override the old values for the same row.
-- The timestamp to be inserted must be newer than the timestamp of subtracting current time by the parameter `KEEP`. If `KEEP` is set to 3650 days, then the data older than 3650 days ago can't be inserted. The timestamp to be inserted can't be newer than the timestamp of current time plus parameter `DAYS`. If `DAYS` is set to 2, the data newer than 2 days later can't be inserted.
+- If the timestamp for the row to be inserted already exists in the table, the old data will be overritten by the new values for the columns for which new values are provided, columns for which no new values are provided are not impacted.
+- The timestamp to be inserted must be newer than the timestamp of subtracting current time by the parameter `KEEP`. If `KEEP` is set to 3650 days, then the data older than 3650 days ago can't be inserted. The timestamp to be inserted can't be newer than the timestamp of current time plus parameter `DURATION`. If `DAYS` is set to 2, the data newer than 2 days later can't be inserted.
 
 :::
 
@@ -101,7 +101,7 @@ For more details about `INSERT` please refer to [INSERT](/taos-sql/insert).
 
 ### Insert with Parameter Binding
 
-TDengine also provides API support for parameter binding. Similar to MySQL, only `?` can be used in these APIs to represent the parameters to bind. From version 2.1.1.0 and 2.1.2.0, parameter binding support for inserting data has improved significantly to improve the insert performance by avoiding the cost of parsing SQL statements.
+TDengine also provides API support for parameter binding. Similar to MySQL, only `?` can be used in these APIs to represent the parameters to bind. Parameter binding support for inserting data has improved significantly to improve the insert performance by avoiding the cost of parsing SQL statements.
 
 Parameter binding is available only with native connection.
 

docs/zh/07-develop/03-insert-data/01-sql-writing.mdx

@@ -52,15 +52,15 @@ INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6,
 
 :::info
 
-- 要提高写入效率，需要批量写入。一批写入的记录条数越多，插入效率就越高。但一条记录不能超过 48K，一条 SQL 语句总长度不能超过 1M 。
-- TDengine 支持多线程同时写入，要进一步提高写入速度，一个客户端需要打开 20 个以上的线程同时写。但线程数达到一定数量后，无法再提高，甚至还会下降，因为线程频繁切换，带来额外开销。
+- 要提高写入效率，需要批量写入。一般来说一批写入的记录条数越多，插入效率就越高。但一条记录不能超过 48K，一条 SQL 语句总长度不能超过 1M 。
+- TDengine 支持多线程同时写入，要进一步提高写入速度，一个客户端需要打开多个同时写。但线程数达到一定数量后，无法再提高，甚至还会下降，因为线程频繁切换，会带来额外开销，合适的线程数量与服务端的处理能力，服务端的具体配置，数据库的参数，数据定义的 Schema，写入数据的 Batch Size 等很多因素相关。一般来说，服务端和客户端处理能力越强，所能支持的并发写入的线程可以越多；数据库配置时的 vgroups 越多（但仍然要在服务端的处理能力以内）则所能支持的并发写入越多；数据定义的 Schema 越简单，所能支持的并发写入越多。
 
 :::
 
 :::warning
 
-- 对同一张表，如果新插入记录的时间戳已经存在，默认情形下（UPDATE=0）新记录将被直接抛弃，也就是说，在一张表里，时间戳必须是唯一的。如果应用自动生成记录，很有可能生成的时间戳是一样的，这样，成功插入的记录条数会小于应用插入的记录条数。如果在创建数据库时使用了 UPDATE 1 选项，插入相同时间戳的新记录将覆盖原有记录。
-- 写入的数据的时间戳必须大于当前时间减去配置参数 keep 的时间。如果 keep 配置为 3650 天，那么无法写入比 3650 天还早的数据。写入数据的时间戳也不能大于当前时间加配置参数 days。如果 days 为 2，那么无法写入比当前时间还晚 2 天的数据。
+- 对同一张表，如果新插入记录的时间戳已经存在，则指定了新值的列会用新值覆盖旧值，而没有指定新值的列则不受影响。
+- 写入的数据的时间戳必须大于当前时间减去配置参数 keep 的时间。如果 keep 配置为 3650 天，那么无法写入比 3650 天还早的数据。写入数据的时间戳也不能大于当前时间加配置参数 duration。如果 duration 为 2，那么无法写入比当前时间还晚 2 天的数据。
 
 :::
 
@@ -104,7 +104,7 @@ INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6,
 
 ### 参数绑定写入
 
-TDengine 也提供了支持参数绑定的 Prepare API，与 MySQL 类似，这些 API 目前也仅支持用问号 `?` 来代表待绑定的参数。从 2.1.1.0 和 2.1.2.0 版本开始，TDengine 大幅改进了参数绑定接口对数据写入（INSERT）场景的支持。这样在通过参数绑定接口写入数据时，就避免了 SQL 语法解析的资源消耗，从而在绝大多数情况下显著提升写入性能。
+TDengine 也提供了支持参数绑定的 Prepare API，与 MySQL 类似，这些 API 目前也仅支持用问号 `?` 来代表待绑定的参数。在通过参数绑定接口写入数据时，就避免了 SQL 语法解析的资源消耗，从而在绝大多数情况下显著提升写入性能。
 
 需要注意的是，只有使用原生连接的连接器，才能使用参数绑定功能。
 

source/client/src/clientEnv.c

@@ -189,12 +189,15 @@ void destroyTscObj(void *pObj) {
 
   SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
   hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
-  int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
+
   destroyAllRequests(pTscObj->pRequests);
+  taosHashCleanup(pTscObj->pRequests);
+  
   schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
   tscDebug("connObj 0x%" PRIx64 " p:%p destroyed, remain inst totalConn:%" PRId64, pTscObj->id, pTscObj,
            pTscObj->pAppInfo->numOfConns);
 
+  int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
   if (0 == connNum) {
     destroyAppInst(pTscObj->pAppInfo);
   }

source/client/src/clientHb.c

@@ -671,8 +671,7 @@ static void *hbThreadFunc(void *param) {
   }
 #endif
   while (1) {
-    int8_t threadStop = atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 1, 2);
-    if (1 == threadStop) {
+    if (1 == clientHbMgr.threadStop) {
       break;
     }
 
@@ -760,9 +759,7 @@ static void hbStopThread() {
     return;
   }
 
-  while (2 != atomic_load_8(&clientHbMgr.threadStop)) {
-    taosUsleep(10);
-  }
+  taosThreadJoin(clientHbMgr.thread, NULL);    
 
   tscDebug("hb thread stopped");
 }

source/common/src/systable.c

@@ -123,6 +123,9 @@ static const SSysDbTableSchema userStbsSchema[] = {
     {.name = "tags", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "last_update", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "watermark", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "max_delay", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "rollup", .bytes = 128 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
 };
 
 static const SSysDbTableSchema streamSchema[] = {
@@ -146,8 +149,8 @@ static const SSysDbTableSchema userTblsSchema[] = {
     {.name = "uid", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "ttl", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "type", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "type", .bytes = 21 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
 };
 
 static const SSysDbTableSchema userTblDistSchema[] = {

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

@@ -2109,7 +2109,7 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)&pStb->updateTime, false);  // number of tables
 
-    pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     if (pStb->commentLen > 0) {
       char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
       STR_TO_VARSTR(comment, pStb->comment);
@@ -2122,6 +2122,34 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
       colDataAppendNULL(pColInfo, numOfRows);
     }
 
+    char watermark[64 + VARSTR_HEADER_SIZE] = {0};
+    sprintf(varDataVal(watermark), "%" PRId64 "a,%" PRId64 "a", pStb->watermark[0], pStb->watermark[1]);
+    varDataSetLen(watermark, strlen(varDataVal(watermark)));
+
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)watermark, false);
+
+    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
+    sprintf(varDataVal(maxDelay), "%" PRId64 "a,%" PRId64 "a", pStb->maxdelay[0], pStb->maxdelay[1]);
+    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
+
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)maxDelay, false);
+
+    char rollup[128 + VARSTR_HEADER_SIZE] = {0};
+    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
+    for (int32_t i = 0; i < rollupNum; ++i) {
+      char *funcName = taosArrayGet(pStb->pFuncs, i);
+      if (i) {
+        strcat(varDataVal(rollup), ", ");        
+      }
+      strcat(varDataVal(rollup), funcName);
+    }
+    varDataSetLen(rollup, strlen(varDataVal(rollup)));
+
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)rollup, false);
+
     numOfRows++;
     sdbRelease(pSdb, pStb);
   }

source/dnode/vnode/src/sma/smaRollup.c

@@ -65,9 +65,12 @@ struct SRSmaInfo {
 
 static SRSmaInfo *tdGetRSmaInfoByItem(SRSmaInfoItem *pItem) {
   // adapt accordingly if definition of SRSmaInfo update
-  int32_t rsmaInfoHeadLen = sizeof(int64_t) + sizeof(STSchema *);
-  ASSERT(pItem->level == 0 || pItem->level == 1);
-  return (SRSmaInfo *)POINTER_SHIFT(pItem, -sizeof(SRSmaInfoItem) * pItem->level - rsmaInfoHeadLen);
+  SRSmaInfo *pResult = NULL;
+  int32_t    rsmaInfoHeadLen = sizeof(int64_t) + sizeof(STSchema *);
+  ASSERT(pItem->level == TSDB_RETENTION_L1 || pItem->level == TSDB_RETENTION_L2);
+  pResult = (SRSmaInfo *)POINTER_SHIFT(pItem, -(sizeof(SRSmaInfoItem) * (pItem->level - 1) + rsmaInfoHeadLen));
+  ASSERT(pResult->pTSchema->numOfCols > 1);
+  return pResult;
 }
 
 struct SRSmaQTaskInfoItem {
@@ -276,7 +279,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
     if (pItem->maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) {
       pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
     }
-    pItem->level = idx;
+    pItem->level = idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2;
     smaInfo("vgId:%d table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
             ", finally maxdelay:%" PRIi32,
             SMA_VID(pSma), pRSmaInfo->suid, idx + 1, param->maxdelay[idx], param->watermark[idx], pItem->maxDelay);
@@ -1236,8 +1239,6 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
 
   SRSmaInfo *pRSmaInfo = tdGetRSmaInfoByItem(pItem);
 
-  ASSERT(pRSmaInfo->items[pItem->level].level == pItem->level);
-
   int8_t fetchTriggerStat =
       atomic_val_compare_exchange_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE, TASK_TRIGGER_STAT_INACTIVE);
   switch (fetchTriggerStat) {

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

@@ -400,7 +400,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
   pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
   pReader->verRange = getQueryVerRange(pVnode, pCond, level);
   pReader->type = pCond->type;
-  pReader->window = updateQueryTimeWindow(pVnode->pTsdb, &pCond->twindows);
+  pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
 
   ASSERT(pCond->numOfCols > 0);
 
@@ -2203,15 +2203,15 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* ret
 
     if (level == TSDB_RETENTION_L0) {
       *pLevel = TSDB_RETENTION_L0;
-      tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L0, str);
+      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L0, str);
       return VND_RSMA0(pVnode);
     } else if (level == TSDB_RETENTION_L1) {
       *pLevel = TSDB_RETENTION_L1;
-      tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L1, str);
+      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L1, str);
       return VND_RSMA1(pVnode);
     } else {
       *pLevel = TSDB_RETENTION_L2;
-      tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L2, str);
+      tsdbDebug("vgId:%d, rsma level %d is selected to query %s", vgId, TSDB_RETENTION_L2, str);
       return VND_RSMA2(pVnode);
     }
   }

source/libs/catalog/src/catalog.c

@@ -1293,6 +1293,7 @@ void catalogDestroy(void) {
 
   if (!taosCheckCurrentInDll()) {
     ctgClearCacheEnqueue(NULL, true, true, true);
+    taosThreadJoin(gCtgMgmt.updateThread, NULL);    
   }
 
   taosHashCleanup(gCtgMgmt.pCluster);

source/libs/function/src/builtinsimpl.c

@@ -907,11 +907,13 @@ int32_t avgFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SAvgRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t  start = pInput->startRowIndex;
-  char*    data = colDataGetData(pCol, start);
-  SAvgRes* pInputInfo = (SAvgRes*)varDataVal(data);
+  int32_t start = pInput->startRowIndex;
 
-  avgTransferInfo(pInputInfo, pInfo);
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SAvgRes* pInputInfo = (SAvgRes*)varDataVal(data);
+    avgTransferInfo(pInputInfo, pInfo);
+  }
 
   SET_VAL(GET_RES_INFO(pCtx), 1, 1);
 
@@ -2512,11 +2514,13 @@ int32_t apercentileFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SAPercentileInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
 
-  int32_t           start = pInput->startRowIndex;
-  char*             data = colDataGetData(pCol, start);
-  SAPercentileInfo* pInputInfo = (SAPercentileInfo*)varDataVal(data);
+  int32_t start = pInput->startRowIndex;
 
-  apercentileTransferInfo(pInputInfo, pInfo);
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SAPercentileInfo* pInputInfo = (SAPercentileInfo*)varDataVal(data);
+    apercentileTransferInfo(pInputInfo, pInfo);
+  }
 
   SET_VAL(pResInfo, 1, 1);
   return TSDB_CODE_SUCCESS;
@@ -2877,13 +2881,17 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer
 
   SFirstLastRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t        start = pInput->startRowIndex;
-  char*          data = colDataGetData(pCol, start);
-  SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data);
-
-  firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery);
+  int32_t start = pInput->startRowIndex;
+  int32_t numOfElems = 0;
 
-  int32_t numOfElems = pInputInfo->hasResult ? 1 : 0;
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data);
+    firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery);
+    if (!numOfElems) {
+      numOfElems = pInputInfo->hasResult ? 1 : 0;
+    }
+  }
 
   SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
 
@@ -3703,11 +3711,13 @@ int32_t spreadFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SSpreadInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t      start = pInput->startRowIndex;
-  char*        data = colDataGetData(pCol, start);
-  SSpreadInfo* pInputInfo = (SSpreadInfo*)varDataVal(data);
+  int32_t start = pInput->startRowIndex;
 
-  spreadTransferInfo(pInputInfo, pInfo);
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SSpreadInfo* pInputInfo = (SSpreadInfo*)varDataVal(data);
+    spreadTransferInfo(pInputInfo, pInfo);
+  }
 
   SET_VAL(GET_RES_INFO(pCtx), 1, 1);
 
@@ -3873,11 +3883,13 @@ int32_t elapsedFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SElapsedInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t       start = pInput->startRowIndex;
-  char*         data = colDataGetData(pCol, start);
-  SElapsedInfo* pInputInfo = (SElapsedInfo*)varDataVal(data);
+  int32_t start = pInput->startRowIndex;
 
-  elapsedTransferInfo(pInputInfo, pInfo);
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SElapsedInfo* pInputInfo = (SElapsedInfo*)varDataVal(data);
+    elapsedTransferInfo(pInputInfo, pInfo);
+  }
 
   SET_VAL(GET_RES_INFO(pCtx), 1, 1);
   return TSDB_CODE_SUCCESS;
@@ -4164,10 +4176,10 @@ int32_t histogramFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SHistoFuncInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t         start = pInput->startRowIndex;
+  int32_t start = pInput->startRowIndex;
 
   for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
-    char*       data = colDataGetData(pCol, i);
+    char* data = colDataGetData(pCol, i);
     SHistoFuncInfo* pInputInfo = (SHistoFuncInfo*)varDataVal(data);
     histogramTransferInfo(pInputInfo, pInfo);
   }
@@ -4385,11 +4397,13 @@ int32_t hllFunctionMerge(SqlFunctionCtx* pCtx) {
 
   SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
-  int32_t   start = pInput->startRowIndex;
-  char*     data = colDataGetData(pCol, start);
-  SHLLInfo* pInputInfo = (SHLLInfo*)varDataVal(data);
+  int32_t start = pInput->startRowIndex;
 
-  hllTransferInfo(pInputInfo, pInfo);
+  for(int32_t i = start; i < start + pInput->numOfRows; ++i) {
+    char* data = colDataGetData(pCol, i);
+    SHLLInfo* pInputInfo = (SHLLInfo*)varDataVal(data);
+    hllTransferInfo(pInputInfo, pInfo);
+  }
 
   SET_VAL(GET_RES_INFO(pCtx), 1, 1);
   return TSDB_CODE_SUCCESS;