From be34546edd8935917f05e663dede14a0b919e5af Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 5 Jun 2023 15:19:06 +0800
Subject: [PATCH 1/8] enh: enable interp fill value support scarlar expression

---
 source/libs/parser/src/parTranslater.c | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index c10ee5d988..57f708c5cd 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -707,6 +707,10 @@ static bool isWindowPseudoColumnFunc(const SNode* pNode) {
   return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsWindowPseudoColumnFunc(((SFunctionNode*)pNode)->funcId));
 }
 
+static bool isInterpFunc(const SNode* pNode) {
+  return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsInterpFunc(((SFunctionNode*)pNode)->funcId));
+}
+
 static bool isInterpPseudoColumnFunc(const SNode* pNode) {
   return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsInterpPseudoColumnFunc(((SFunctionNode*)pNode)->funcId));
 }
@@ -3006,7 +3010,7 @@ static int32_t translateOrderBy(STranslateContext* pCxt, SSelectStmt* pSelect) {
 }
 
 static EDealRes needFillImpl(SNode* pNode, void* pContext) {
-  if (isAggFunc(pNode) && FUNCTION_TYPE_GROUP_KEY != ((SFunctionNode*)pNode)->funcType) {
+  if ((isAggFunc(pNode) || isInterpFunc(pNode)) && FUNCTION_TYPE_GROUP_KEY != ((SFunctionNode*)pNode)->funcType) {
     *(bool*)pContext = true;
     return DEAL_RES_END;
   }
@@ -3517,6 +3521,9 @@ static int32_t translateInterpFill(STranslateContext* pCxt, SSelectStmt* pSelect
   if (TSDB_CODE_SUCCESS == code) {
     code = checkFill(pCxt, (SFillNode*)pSelect->pFill, (SValueNode*)pSelect->pEvery, true);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = checkFillValues(pCxt, (SFillNode*)pSelect->pFill, pSelect->pProjectionList);
+  }
 
   return code;
 }

From 6b670b7f0801b490e7e5e66e25428be233529eb1 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 5 Jun 2023 15:48:15 +0800
Subject: [PATCH 2/8] fix desc

---
 source/libs/parser/src/parTranslater.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index 57f708c5cd..be3c41e2ee 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -3034,7 +3034,7 @@ static int32_t convertFillValue(STranslateContext* pCxt, SDataType dt, SNodeList
     code = scalarCalculateConstants(pCaseFunc, &pCell->pNode);
   }
   if (TSDB_CODE_SUCCESS == code && QUERY_NODE_VALUE != nodeType(pCell->pNode)) {
-    code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "Fill value is just a constant");
+    code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "Fill value can only accept constant");
   } else if (TSDB_CODE_SUCCESS != code) {
     code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "Filled data type mismatch");
   }

From 005182c3c925169d45392feb0755beaf50efca2a Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 5 Jun 2023 16:40:48 +0800
Subject: [PATCH 3/8] fix multiple interp issue

---
 source/libs/parser/src/parTranslater.c | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index be3c41e2ee..f049af6747 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -3041,7 +3041,7 @@ static int32_t convertFillValue(STranslateContext* pCxt, SDataType dt, SNodeList
   return code;
 }
 
-static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeList* pProjectionList) {
+static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeList* pProjectionList, bool isInterpFill) {
   if (FILL_MODE_VALUE != pFill->mode && FILL_MODE_VALUE_F != pFill->mode) {
     return TSDB_CODE_SUCCESS;
   }
@@ -3058,10 +3058,13 @@ static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeL
       if (TSDB_CODE_SUCCESS != code) {
         return code;
       }
-      ++fillNo;
+
+      if (!isInterpFill) {
+        ++fillNo;
+      }
     }
   }
-  if (fillNo != LIST_LENGTH(pFillValues->pNodeList)) {
+  if (!isInterpFill && fillNo != LIST_LENGTH(pFillValues->pNodeList)) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "Filled values number mismatch");
   }
   return TSDB_CODE_SUCCESS;
@@ -3072,7 +3075,7 @@ static int32_t translateFillValues(STranslateContext* pCxt, SSelectStmt* pSelect
       NULL == ((SIntervalWindowNode*)pSelect->pWindow)->pFill) {
     return TSDB_CODE_SUCCESS;
   }
-  return checkFillValues(pCxt, (SFillNode*)((SIntervalWindowNode*)pSelect->pWindow)->pFill, pSelect->pProjectionList);
+  return checkFillValues(pCxt, (SFillNode*)((SIntervalWindowNode*)pSelect->pWindow)->pFill, pSelect->pProjectionList, false);
 }
 
 static int32_t rewriteProjectAlias(SNodeList* pProjectionList) {
@@ -3522,7 +3525,7 @@ static int32_t translateInterpFill(STranslateContext* pCxt, SSelectStmt* pSelect
     code = checkFill(pCxt, (SFillNode*)pSelect->pFill, (SValueNode*)pSelect->pEvery, true);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = checkFillValues(pCxt, (SFillNode*)pSelect->pFill, pSelect->pProjectionList);
+    code = checkFillValues(pCxt, (SFillNode*)pSelect->pFill, pSelect->pProjectionList, true);
   }
 
   return code;

From 384aa4d70bc1751037ec9af1be813eba490af5ad Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 5 Jun 2023 16:41:09 +0800
Subject: [PATCH 4/8] add test cases

---
 tests/system-test/2-query/interp.py | 128 ++++++++++++++++++++++++++++
 1 file changed, 128 insertions(+)

diff --git a/tests/system-test/2-query/interp.py b/tests/system-test/2-query/interp.py
index 121d4dcff6..019b449dfd 100644
--- a/tests/system-test/2-query/interp.py
+++ b/tests/system-test/2-query/interp.py
@@ -219,6 +219,56 @@ class TDTestCase:
         tdSql.checkData(2, 0, 12)
         tdSql.checkData(3, 0, 12)
 
+        ## test fill value with scalar expression
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, 1 + 2)")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, 1.0 + 2.0)")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, 1 + 2.5)")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, 1 + '2')")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, 1 + '2.0')")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, '3' + 'abc')")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
+        tdSql.query(f"select interp(c0) from {dbname}.{tbname} range('2020-02-01 00:00:16', '2020-02-01 00:00:19') every(1s) fill(value, '2' + '1abc')")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 0, 3)
+        tdSql.checkData(1, 0, 3)
+        tdSql.checkData(2, 0, 3)
+        tdSql.checkData(3, 0, 3)
+
         tdLog.printNoPrefix("==========step5:fill prev")
 
         ## {. . .}
@@ -1837,6 +1887,84 @@ class TDTestCase:
         tdSql.checkData(59, 1, 123)
         tdSql.checkData(60, 1, 60)   #
 
+        tdSql.query(f"select interp(c0),interp(c1) from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(value, 123 + 123)")
+        tdSql.checkRows(61)
+        tdSql.checkCols(2)
+        tdSql.checkData(0,  0, 0)    #
+        tdSql.checkData(1,  0, 246)
+        tdSql.checkData(4,  0, 246)
+        tdSql.checkData(5,  0, None) #
+        tdSql.checkData(6,  0, 246)
+        tdSql.checkData(9,  0, 246)
+        tdSql.checkData(10, 0, 10)   #
+        tdSql.checkData(11, 0, 246)
+        tdSql.checkData(14, 0, 246)
+        tdSql.checkData(15, 0, None) #
+        tdSql.checkData(16, 0, 246)
+        tdSql.checkData(19, 0, 246)
+        tdSql.checkData(20, 0, 20)   #
+        tdSql.checkData(21, 0, 246)
+        tdSql.checkData(24, 0, 246)
+        tdSql.checkData(25, 0, None) #
+        tdSql.checkData(26, 0, 246)
+        tdSql.checkData(29, 0, 246)
+        tdSql.checkData(30, 0, 30)   #
+        tdSql.checkData(31, 0, 246)
+        tdSql.checkData(34, 0, 246)
+        tdSql.checkData(35, 0, 35)   #
+        tdSql.checkData(36, 0, 246)
+        tdSql.checkData(39, 0, 246)
+        tdSql.checkData(40, 0, 40)   #
+        tdSql.checkData(41, 0, 246)
+        tdSql.checkData(44, 0, 246)
+        tdSql.checkData(45, 0, None) #
+        tdSql.checkData(46, 0, 246)
+        tdSql.checkData(49, 0, 246)
+        tdSql.checkData(50, 0, 50)   #
+        tdSql.checkData(51, 0, 246)
+        tdSql.checkData(54, 0, 246)
+        tdSql.checkData(55, 0, None) #
+        tdSql.checkData(59, 0, 246)
+        tdSql.checkData(60, 0, 55)   #
+
+        tdSql.checkData(0,  1, None) #
+        tdSql.checkData(1,  1, 246)
+        tdSql.checkData(4,  1, 246)
+        tdSql.checkData(5,  1, None) #
+        tdSql.checkData(6,  1, 246)
+        tdSql.checkData(9,  1, 246)
+        tdSql.checkData(10, 1, 10)   #
+        tdSql.checkData(11, 1, 246)
+        tdSql.checkData(14, 1, 246)
+        tdSql.checkData(15, 1, None) #
+        tdSql.checkData(16, 1, 246)
+        tdSql.checkData(19, 1, 246)
+        tdSql.checkData(20, 1, None) #
+        tdSql.checkData(21, 1, 246)
+        tdSql.checkData(24, 1, 246)
+        tdSql.checkData(25, 1, None) #
+        tdSql.checkData(26, 1, 246)
+        tdSql.checkData(29, 1, 246)
+        tdSql.checkData(30, 1, 30)   #
+        tdSql.checkData(31, 1, 246)
+        tdSql.checkData(34, 1, 246)
+        tdSql.checkData(35, 1, None) #
+        tdSql.checkData(36, 1, 246)
+        tdSql.checkData(39, 1, 246)
+        tdSql.checkData(40, 1, 40)   #
+        tdSql.checkData(41, 1, 246)
+        tdSql.checkData(44, 1, 246)
+        tdSql.checkData(45, 1, 45)   #
+        tdSql.checkData(46, 1, 246)
+        tdSql.checkData(49, 1, 246)
+        tdSql.checkData(50, 1, None) #
+        tdSql.checkData(51, 1, 246)
+        tdSql.checkData(54, 1, 246)
+        tdSql.checkData(55, 1, None) #
+        tdSql.checkData(56, 1, 246)
+        tdSql.checkData(59, 1, 246)
+        tdSql.checkData(60, 1, 60)   #
+
         # test fill prev
         tdSql.query(f"select interp(c0),interp(c1) from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(prev)")
         tdSql.checkRows(61)

From 1e9c4d5facfaf7dbae9c2d6c30636af6193e563c Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Wed, 7 Jun 2023 10:45:27 +0800
Subject: [PATCH 5/8] opt stream block dispatch

---
 include/common/tdatablock.h             |  1 +
 include/libs/stream/tstream.h           |  1 +
 source/common/src/tdatablock.c          | 24 +++++++---
 source/dnode/vnode/src/tq/tqSink.c      |  6 +--
 source/libs/stream/src/streamDispatch.c | 63 ++++++++++++++++++-------
 source/libs/stream/src/streamTask.c     |  4 ++
 6 files changed, 71 insertions(+), 28 deletions(-)

diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h
index 53fc07c3f3..6cb7d88523 100644
--- a/include/common/tdatablock.h
+++ b/include/common/tdatablock.h
@@ -248,6 +248,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pData
                                     tb_uid_t suid);
 
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
+int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);
 
 static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) {
   return blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock)) + blockDataGetSize(pBlock);
diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 8316e6ef50..51f2de481d 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -327,6 +327,7 @@ struct SStreamTask {
   int64_t             checkpointingId;
   int32_t             checkpointAlignCnt;
   struct SStreamMeta* pMeta;
+  SSHashObj*          pNameMap;
 };
 
 // meta
diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c
index 24e978b0ea..033fbb0ef1 100644
--- a/source/common/src/tdatablock.c
+++ b/source/common/src/tdatablock.c
@@ -2465,19 +2465,31 @@ _end:
 }
 
 char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
-  if (stbFullName[0] == 0) {
+  char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
+  if (!pBuf) {
     return NULL;
   }
+  int32_t code = buildCtbNameByGroupIdImpl(stbFullName, groupId, pBuf);
+  if (code != TSDB_CODE_SUCCESS) {
+    taosMemoryFree(pBuf);
+    return NULL;
+  }
+  return pBuf;
+}
+
+int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, char* cname) {
+  if (stbFullName[0] == 0) {
+    return TSDB_CODE_FAILED;
+  }
 
   SArray* tags = taosArrayInit(0, sizeof(SSmlKv));
   if (tags == NULL) {
-    return NULL;
+    return TSDB_CODE_FAILED;
   }
 
-  void* cname = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
   if (cname == NULL) {
     taosArrayDestroy(tags);
-    return NULL;
+    return TSDB_CODE_FAILED;
   }
 
   SSmlKv pTag = {.key = "group_id",
@@ -2499,9 +2511,9 @@ char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
   taosArrayDestroy(tags);
 
   if ((rname.ctbShortName && rname.ctbShortName[0]) == 0) {
-    return NULL;
+    return TSDB_CODE_FAILED;
   }
-  return rname.ctbShortName;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c
index db1b5ed902..9349c6eb0d 100644
--- a/source/dnode/vnode/src/tq/tqSink.c
+++ b/source/dnode/vnode/src/tq/tqSink.c
@@ -298,10 +298,8 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
         if (res == TSDB_CODE_SUCCESS) {
           memcpy(ctbName, pTableSinkInfo->tbName, strlen(pTableSinkInfo->tbName));
         } else {
-          char* tmp = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
-          memcpy(ctbName, tmp, strlen(tmp));
-          memcpy(pTableSinkInfo->tbName, tmp, strlen(tmp));
-          taosMemoryFree(tmp);
+          buildCtbNameByGroupIdImpl(stbFullName, pDataBlock->info.id.groupId, ctbName);
+          memcpy(pTableSinkInfo->tbName, ctbName, strlen(ctbName));
           tqDebug("vgId:%d, gropuId:%" PRIu64 " datablock table name is null", TD_VID(pVnode),
                   pDataBlock->info.id.groupId);
         }
diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index 9cb0a56644..922a1f5345 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -15,6 +15,13 @@
 
 #include "streamInc.h"
 
+#define MAX_BLOCK_NAME_NUM 1024
+
+typedef struct SBlockName {
+  uint32_t hashValue;
+  char     parTbName[TSDB_TABLE_NAME_LEN];
+} SBlockName;
+
 int32_t tEncodeStreamDispatchReq(SEncoder* pEncoder, const SStreamDispatchReq* pReq) {
   if (tStartEncode(pEncoder) < 0) return -1;
   if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1;
@@ -331,26 +338,46 @@ FAIL:
 
 int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, SSDataBlock* pDataBlock, int32_t vgSz,
                                 int64_t groupId) {
-  char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
-  if (ctbName == NULL) {
-    return -1;
-  }
-
-  if (pDataBlock->info.parTbName[0]) {
-    snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName);
-  } else {
-    char* ctbShortName = buildCtbNameByGroupId(pTask->shuffleDispatcher.stbFullName, groupId);
-    snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, ctbShortName);
-    taosMemoryFree(ctbShortName);
-  }
-
+  uint32_t   hashValue = 0;
   SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
+  if (pTask->pNameMap == NULL) {
+    pTask->pNameMap = tSimpleHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
+  }
 
-  /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
-  SUseDbRsp* pDbInfo = &pTask->shuffleDispatcher.dbInfo;
-  uint32_t   hashValue =
-      taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
-  taosMemoryFree(ctbName);
+  void* pVal = tSimpleHashGet(pTask->pNameMap, &groupId, sizeof(int64_t));
+  if (pVal) {
+    SBlockName* pBln = (SBlockName*)pVal;
+    hashValue = pBln->hashValue;
+    if (!pDataBlock->info.parTbName[0]) {
+      memcpy(pDataBlock->info.parTbName, pBln->parTbName, strlen(pBln->parTbName));
+    }
+  } else {
+    char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
+    if (ctbName == NULL) {
+      return -1;
+    }
+
+    if (pDataBlock->info.parTbName[0]) {
+      snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName);
+    } else {
+      buildCtbNameByGroupIdImpl(pTask->shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName);
+      snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName);
+    }
+
+    SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
+
+    /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
+    SUseDbRsp* pDbInfo = &pTask->shuffleDispatcher.dbInfo;
+    hashValue =
+        taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
+    taosMemoryFree(ctbName);
+    SBlockName bln = {0};
+    bln.hashValue = hashValue;
+    memcpy(bln.parTbName, pDataBlock->info.parTbName, strlen(pDataBlock->info.parTbName));
+    if (tSimpleHashGetSize(pTask->pNameMap) < MAX_BLOCK_NAME_NUM) {
+      tSimpleHashPut(pTask->pNameMap, &groupId, sizeof(int64_t), &bln, sizeof(SBlockName));
+    }
+  }
 
   bool found = false;
   // TODO: optimize search
diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c
index a0caffd41f..284d1ecab6 100644
--- a/source/libs/stream/src/streamTask.c
+++ b/source/libs/stream/src/streamTask.c
@@ -224,5 +224,9 @@ void tFreeStreamTask(SStreamTask* pTask) {
     taosMemoryFree((void*)pTask->id.idStr);
   }
 
+  if (pTask->pNameMap) {
+    tSimpleHashCleanup(pTask->pNameMap);
+  }
+
   taosMemoryFree(pTask);
 }

From 150e9ae2960da9c8cc146034c0688f3217a9966a Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 7 Jun 2023 16:55:03 +0800
Subject: [PATCH 6/8] make interp fill multiple col logic same as window fill

---
 source/libs/parser/src/parTranslater.c | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index f049af6747..5c841deda0 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -3041,7 +3041,7 @@ static int32_t convertFillValue(STranslateContext* pCxt, SDataType dt, SNodeList
   return code;
 }
 
-static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeList* pProjectionList, bool isInterpFill) {
+static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeList* pProjectionList) {
   if (FILL_MODE_VALUE != pFill->mode && FILL_MODE_VALUE_F != pFill->mode) {
     return TSDB_CODE_SUCCESS;
   }
@@ -3059,12 +3059,10 @@ static int32_t checkFillValues(STranslateContext* pCxt, SFillNode* pFill, SNodeL
         return code;
       }
 
-      if (!isInterpFill) {
-        ++fillNo;
-      }
+      ++fillNo;
     }
   }
-  if (!isInterpFill && fillNo != LIST_LENGTH(pFillValues->pNodeList)) {
+  if (fillNo != LIST_LENGTH(pFillValues->pNodeList)) {
     return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "Filled values number mismatch");
   }
   return TSDB_CODE_SUCCESS;
@@ -3075,7 +3073,7 @@ static int32_t translateFillValues(STranslateContext* pCxt, SSelectStmt* pSelect
       NULL == ((SIntervalWindowNode*)pSelect->pWindow)->pFill) {
     return TSDB_CODE_SUCCESS;
   }
-  return checkFillValues(pCxt, (SFillNode*)((SIntervalWindowNode*)pSelect->pWindow)->pFill, pSelect->pProjectionList, false);
+  return checkFillValues(pCxt, (SFillNode*)((SIntervalWindowNode*)pSelect->pWindow)->pFill, pSelect->pProjectionList);
 }
 
 static int32_t rewriteProjectAlias(SNodeList* pProjectionList) {
@@ -3525,7 +3523,7 @@ static int32_t translateInterpFill(STranslateContext* pCxt, SSelectStmt* pSelect
     code = checkFill(pCxt, (SFillNode*)pSelect->pFill, (SValueNode*)pSelect->pEvery, true);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = checkFillValues(pCxt, (SFillNode*)pSelect->pFill, pSelect->pProjectionList, true);
+    code = checkFillValues(pCxt, (SFillNode*)pSelect->pFill, pSelect->pProjectionList);
   }
 
   return code;

From c612d945ec73114f93623f0116c190608605fc45 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 7 Jun 2023 18:27:29 +0800
Subject: [PATCH 7/8] fix fill value bug

---
 source/libs/executor/src/timesliceoperator.c | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/source/libs/executor/src/timesliceoperator.c b/source/libs/executor/src/timesliceoperator.c
index 3e4055876d..2421343bd7 100644
--- a/source/libs/executor/src/timesliceoperator.c
+++ b/source/libs/executor/src/timesliceoperator.c
@@ -257,7 +257,8 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
 
 
   // output the result
-  bool hasInterp = true;
+  int32_t fillColIndex = 0;
+  bool       hasInterp = true;
   for (int32_t j = 0; j < pExprSup->numOfExprs; ++j) {
     SExprInfo* pExprInfo = &pExprSup->pExprInfo[j];
 
@@ -307,7 +308,7 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
 
       case TSDB_FILL_SET_VALUE:
       case TSDB_FILL_SET_VALUE_F: {
-        SVariant* pVar = &pSliceInfo->pFillColInfo[j].fillVal;
+        SVariant* pVar = &pSliceInfo->pFillColInfo[fillColIndex].fillVal;
 
         if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
           float v = 0;
@@ -342,6 +343,8 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
           }
           colDataSetVal(pDst, rows, (char*)&v, false);
         }
+
+        ++fillColIndex;
         break;
       }
 

From 46baeefea73e2c6edf038482a96e373335d7edef Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 7 Jun 2023 18:39:33 +0800
Subject: [PATCH 8/8] fix test cases

---
 tests/system-test/2-query/interp.py | 190 ++++++++++++++--------------
 1 file changed, 97 insertions(+), 93 deletions(-)

diff --git a/tests/system-test/2-query/interp.py b/tests/system-test/2-query/interp.py
index 019b449dfd..eb6aeec95a 100644
--- a/tests/system-test/2-query/interp.py
+++ b/tests/system-test/2-query/interp.py
@@ -1809,47 +1809,10 @@ class TDTestCase:
         tdSql.checkData(60, 1, 60)   #
 
         # test fill value
-        tdSql.query(f"select interp(c0),interp(c1) from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(value, 123)")
+        tdSql.query(f"select _irowts, interp(c0), _irowts, interp(c1), _irowts from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(value, 123, 456)")
         tdSql.checkRows(61)
-        tdSql.checkCols(2)
-        tdSql.checkData(0,  0, 0)    #
-        tdSql.checkData(1,  0, 123)
-        tdSql.checkData(4,  0, 123)
-        tdSql.checkData(5,  0, None) #
-        tdSql.checkData(6,  0, 123)
-        tdSql.checkData(9,  0, 123)
-        tdSql.checkData(10, 0, 10)   #
-        tdSql.checkData(11, 0, 123)
-        tdSql.checkData(14, 0, 123)
-        tdSql.checkData(15, 0, None) #
-        tdSql.checkData(16, 0, 123)
-        tdSql.checkData(19, 0, 123)
-        tdSql.checkData(20, 0, 20)   #
-        tdSql.checkData(21, 0, 123)
-        tdSql.checkData(24, 0, 123)
-        tdSql.checkData(25, 0, None) #
-        tdSql.checkData(26, 0, 123)
-        tdSql.checkData(29, 0, 123)
-        tdSql.checkData(30, 0, 30)   #
-        tdSql.checkData(31, 0, 123)
-        tdSql.checkData(34, 0, 123)
-        tdSql.checkData(35, 0, 35)   #
-        tdSql.checkData(36, 0, 123)
-        tdSql.checkData(39, 0, 123)
-        tdSql.checkData(40, 0, 40)   #
-        tdSql.checkData(41, 0, 123)
-        tdSql.checkData(44, 0, 123)
-        tdSql.checkData(45, 0, None) #
-        tdSql.checkData(46, 0, 123)
-        tdSql.checkData(49, 0, 123)
-        tdSql.checkData(50, 0, 50)   #
-        tdSql.checkData(51, 0, 123)
-        tdSql.checkData(54, 0, 123)
-        tdSql.checkData(55, 0, None) #
-        tdSql.checkData(59, 0, 123)
-        tdSql.checkData(60, 0, 55)   #
-
-        tdSql.checkData(0,  1, None) #
+        tdSql.checkCols(5)
+        tdSql.checkData(0,  1, 0)    #
         tdSql.checkData(1,  1, 123)
         tdSql.checkData(4,  1, 123)
         tdSql.checkData(5,  1, None) #
@@ -1861,7 +1824,7 @@ class TDTestCase:
         tdSql.checkData(15, 1, None) #
         tdSql.checkData(16, 1, 123)
         tdSql.checkData(19, 1, 123)
-        tdSql.checkData(20, 1, None) #
+        tdSql.checkData(20, 1, 20)   #
         tdSql.checkData(21, 1, 123)
         tdSql.checkData(24, 1, 123)
         tdSql.checkData(25, 1, None) #
@@ -1870,64 +1833,64 @@ class TDTestCase:
         tdSql.checkData(30, 1, 30)   #
         tdSql.checkData(31, 1, 123)
         tdSql.checkData(34, 1, 123)
-        tdSql.checkData(35, 1, None) #
+        tdSql.checkData(35, 1, 35)   #
         tdSql.checkData(36, 1, 123)
         tdSql.checkData(39, 1, 123)
         tdSql.checkData(40, 1, 40)   #
         tdSql.checkData(41, 1, 123)
         tdSql.checkData(44, 1, 123)
-        tdSql.checkData(45, 1, 45)   #
+        tdSql.checkData(45, 1, None) #
         tdSql.checkData(46, 1, 123)
         tdSql.checkData(49, 1, 123)
-        tdSql.checkData(50, 1, None) #
+        tdSql.checkData(50, 1, 50)   #
         tdSql.checkData(51, 1, 123)
         tdSql.checkData(54, 1, 123)
         tdSql.checkData(55, 1, None) #
-        tdSql.checkData(56, 1, 123)
         tdSql.checkData(59, 1, 123)
-        tdSql.checkData(60, 1, 60)   #
+        tdSql.checkData(60, 1, 55)   #
 
-        tdSql.query(f"select interp(c0),interp(c1) from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(value, 123 + 123)")
+        tdSql.checkData(0,  3, None) #
+        tdSql.checkData(1,  3, 456)
+        tdSql.checkData(4,  3, 456)
+        tdSql.checkData(5,  3, None) #
+        tdSql.checkData(6,  3, 456)
+        tdSql.checkData(9,  3, 456)
+        tdSql.checkData(10, 3, 10)   #
+        tdSql.checkData(11, 3, 456)
+        tdSql.checkData(14, 3, 456)
+        tdSql.checkData(15, 3, None) #
+        tdSql.checkData(16, 3, 456)
+        tdSql.checkData(19, 3, 456)
+        tdSql.checkData(20, 3, None) #
+        tdSql.checkData(21, 3, 456)
+        tdSql.checkData(24, 3, 456)
+        tdSql.checkData(25, 3, None) #
+        tdSql.checkData(26, 3, 456)
+        tdSql.checkData(29, 3, 456)
+        tdSql.checkData(30, 3, 30)   #
+        tdSql.checkData(31, 3, 456)
+        tdSql.checkData(34, 3, 456)
+        tdSql.checkData(35, 3, None) #
+        tdSql.checkData(36, 3, 456)
+        tdSql.checkData(39, 3, 456)
+        tdSql.checkData(40, 3, 40)   #
+        tdSql.checkData(41, 3, 456)
+        tdSql.checkData(44, 3, 456)
+        tdSql.checkData(45, 3, 45)   #
+        tdSql.checkData(46, 3, 456)
+        tdSql.checkData(49, 3, 456)
+        tdSql.checkData(50, 3, None) #
+        tdSql.checkData(51, 3, 456)
+        tdSql.checkData(54, 3, 456)
+        tdSql.checkData(55, 3, None) #
+        tdSql.checkData(56, 3, 456)
+        tdSql.checkData(59, 3, 456)
+        tdSql.checkData(60, 3, 60)   #
+
+        tdSql.query(f"select _isfilled, interp(c0), _isfilled, interp(c1), _isfilled from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(value, 123 + 123, 234 + 234)")
         tdSql.checkRows(61)
-        tdSql.checkCols(2)
-        tdSql.checkData(0,  0, 0)    #
-        tdSql.checkData(1,  0, 246)
-        tdSql.checkData(4,  0, 246)
-        tdSql.checkData(5,  0, None) #
-        tdSql.checkData(6,  0, 246)
-        tdSql.checkData(9,  0, 246)
-        tdSql.checkData(10, 0, 10)   #
-        tdSql.checkData(11, 0, 246)
-        tdSql.checkData(14, 0, 246)
-        tdSql.checkData(15, 0, None) #
-        tdSql.checkData(16, 0, 246)
-        tdSql.checkData(19, 0, 246)
-        tdSql.checkData(20, 0, 20)   #
-        tdSql.checkData(21, 0, 246)
-        tdSql.checkData(24, 0, 246)
-        tdSql.checkData(25, 0, None) #
-        tdSql.checkData(26, 0, 246)
-        tdSql.checkData(29, 0, 246)
-        tdSql.checkData(30, 0, 30)   #
-        tdSql.checkData(31, 0, 246)
-        tdSql.checkData(34, 0, 246)
-        tdSql.checkData(35, 0, 35)   #
-        tdSql.checkData(36, 0, 246)
-        tdSql.checkData(39, 0, 246)
-        tdSql.checkData(40, 0, 40)   #
-        tdSql.checkData(41, 0, 246)
-        tdSql.checkData(44, 0, 246)
-        tdSql.checkData(45, 0, None) #
-        tdSql.checkData(46, 0, 246)
-        tdSql.checkData(49, 0, 246)
-        tdSql.checkData(50, 0, 50)   #
-        tdSql.checkData(51, 0, 246)
-        tdSql.checkData(54, 0, 246)
-        tdSql.checkData(55, 0, None) #
-        tdSql.checkData(59, 0, 246)
-        tdSql.checkData(60, 0, 55)   #
-
-        tdSql.checkData(0,  1, None) #
+        tdSql.checkCols(5)
+        tdSql.checkData(0,  1, 0)    #
         tdSql.checkData(1,  1, 246)
         tdSql.checkData(4,  1, 246)
         tdSql.checkData(5,  1, None) #
@@ -1939,7 +1902,7 @@ class TDTestCase:
         tdSql.checkData(15, 1, None) #
         tdSql.checkData(16, 1, 246)
         tdSql.checkData(19, 1, 246)
-        tdSql.checkData(20, 1, None) #
+        tdSql.checkData(20, 1, 20)   #
         tdSql.checkData(21, 1, 246)
         tdSql.checkData(24, 1, 246)
         tdSql.checkData(25, 1, None) #
@@ -1948,22 +1911,59 @@ class TDTestCase:
         tdSql.checkData(30, 1, 30)   #
         tdSql.checkData(31, 1, 246)
         tdSql.checkData(34, 1, 246)
-        tdSql.checkData(35, 1, None) #
+        tdSql.checkData(35, 1, 35)   #
         tdSql.checkData(36, 1, 246)
         tdSql.checkData(39, 1, 246)
         tdSql.checkData(40, 1, 40)   #
         tdSql.checkData(41, 1, 246)
         tdSql.checkData(44, 1, 246)
-        tdSql.checkData(45, 1, 45)   #
+        tdSql.checkData(45, 1, None) #
         tdSql.checkData(46, 1, 246)
         tdSql.checkData(49, 1, 246)
-        tdSql.checkData(50, 1, None) #
+        tdSql.checkData(50, 1, 50)   #
         tdSql.checkData(51, 1, 246)
         tdSql.checkData(54, 1, 246)
         tdSql.checkData(55, 1, None) #
-        tdSql.checkData(56, 1, 246)
         tdSql.checkData(59, 1, 246)
-        tdSql.checkData(60, 1, 60)   #
+        tdSql.checkData(60, 1, 55)   #
+
+        tdSql.checkData(0,  3, None) #
+        tdSql.checkData(1,  3, 468)
+        tdSql.checkData(4,  3, 468)
+        tdSql.checkData(5,  3, None) #
+        tdSql.checkData(6,  3, 468)
+        tdSql.checkData(9,  3, 468)
+        tdSql.checkData(10, 3, 10)   #
+        tdSql.checkData(11, 3, 468)
+        tdSql.checkData(14, 3, 468)
+        tdSql.checkData(15, 3, None) #
+        tdSql.checkData(16, 3, 468)
+        tdSql.checkData(19, 3, 468)
+        tdSql.checkData(20, 3, None) #
+        tdSql.checkData(21, 3, 468)
+        tdSql.checkData(24, 3, 468)
+        tdSql.checkData(25, 3, None) #
+        tdSql.checkData(26, 3, 468)
+        tdSql.checkData(29, 3, 468)
+        tdSql.checkData(30, 3, 30)   #
+        tdSql.checkData(31, 3, 468)
+        tdSql.checkData(34, 3, 468)
+        tdSql.checkData(35, 3, None) #
+        tdSql.checkData(36, 3, 468)
+        tdSql.checkData(39, 3, 468)
+        tdSql.checkData(40, 3, 40)   #
+        tdSql.checkData(41, 3, 468)
+        tdSql.checkData(44, 3, 468)
+        tdSql.checkData(45, 3, 45)   #
+        tdSql.checkData(46, 3, 468)
+        tdSql.checkData(49, 3, 468)
+        tdSql.checkData(50, 3, None) #
+        tdSql.checkData(51, 3, 468)
+        tdSql.checkData(54, 3, 468)
+        tdSql.checkData(55, 3, None) #
+        tdSql.checkData(56, 3, 468)
+        tdSql.checkData(59, 3, 468)
+        tdSql.checkData(60, 3, 60)   #
 
         # test fill prev
         tdSql.query(f"select interp(c0),interp(c1) from {dbname}.{tbname1} range('2020-02-02 00:00:00', '2020-02-02 00:01:00') every(1s) fill(prev)")
@@ -2138,7 +2138,7 @@ class TDTestCase:
           tdSql.checkData(3, i, None)
           tdSql.checkData(4, i, None)
 
-        tdSql.query(f"select interp(c0),interp(c1),interp(c2),interp(c3) from {dbname}.{tbname} range('2020-02-09 00:00:05', '2020-02-13 00:00:05') every(1d) fill(value, 1)")
+        tdSql.query(f"select interp(c0),interp(c1),interp(c2),interp(c3) from {dbname}.{tbname} range('2020-02-09 00:00:05', '2020-02-13 00:00:05') every(1d) fill(value, 1, 1, 1, 1)")
         tdSql.checkRows(5)
         tdSql.checkCols(4)
 
@@ -2564,6 +2564,10 @@ class TDTestCase:
         tdSql.error(f"select interp(c0) from {dbname}.{tbname} where _isfilled = true range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
         tdSql.error(f"select interp(c0) from {dbname}.{tbname} where _irowts > 0 range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
 
+        # fill value number mismatch
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(value, 1, 2)")
+        tdSql.error(f"select interp(c0), interp(c1) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(value, 1)")
+