From f49d91ac1a53c4c2f0582454db2d9e09f9dae892 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Fri, 21 Oct 2022 17:11:22 +0800
Subject: [PATCH] fix(stream): decode bloom filter

---
 source/dnode/vnode/src/tq/tqStreamStateSnap.c | 2 +-
 source/libs/stream/src/streamUpdate.c         | 8 ++++++--
 source/util/src/tbloomfilter.c                | 6 ++++--
 3 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/source/dnode/vnode/src/tq/tqStreamStateSnap.c b/source/dnode/vnode/src/tq/tqStreamStateSnap.c
index b4a7ce7737..92e5f8df7a 100644
--- a/source/dnode/vnode/src/tq/tqStreamStateSnap.c
+++ b/source/dnode/vnode/src/tq/tqStreamStateSnap.c
@@ -167,7 +167,7 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
   if (rollback) {
     ASSERT(0);
   } else {
-    code = tdbCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
+    code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
     if (code) goto _err;
   }
 
diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c
index 80410568e5..199892c241 100644
--- a/source/libs/stream/src/streamUpdate.c
+++ b/source/libs/stream/src/streamUpdate.c
@@ -175,11 +175,15 @@ void updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t pr
     maxTs = TMAX(maxTs, ts);
     SScalableBf *pSBf = getSBf(pInfo, ts);
     if (pSBf) {
-      tScalableBfPut(pSBf, &ts, sizeof(TSKEY));
+      SUpdateKey updateKey = {
+          .tbUid = tbUid,
+          .ts = ts,
+      };
+      tScalableBfPut(pSBf, &updateKey, sizeof(SUpdateKey));
     }
   }
   TSKEY *pMaxTs = taosHashGet(pInfo->pMap, &tbUid, sizeof(int64_t));
-  if (pMaxTs == NULL || *pMaxTs > tbUid) {
+  if (pMaxTs == NULL || *pMaxTs > maxTs) {
     taosHashPut(pInfo->pMap, &tbUid, sizeof(int64_t), &maxTs, sizeof(TSKEY));
   }
 }
diff --git a/source/util/src/tbloomfilter.c b/source/util/src/tbloomfilter.c
index b9c96dd606..84a78f3477 100644
--- a/source/util/src/tbloomfilter.c
+++ b/source/util/src/tbloomfilter.c
@@ -137,8 +137,10 @@ SBloomFilter *tBloomFilterDecode(SDecoder *pDecoder) {
     if (tDecodeU64(pDecoder, pUnits + i) < 0) goto _error;
   }
   if (tDecodeDouble(pDecoder, &pBF->errorRate) < 0) goto _error;
-  pBF->hashFn1 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_TIMESTAMP);
-  pBF->hashFn2 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_NCHAR);
+  /*pBF->hashFn1 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_TIMESTAMP);*/
+  /*pBF->hashFn2 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_NCHAR);*/
+  pBF->hashFn1 = taosFastHash;
+  pBF->hashFn2 = taosDJB2Hash;
   return pBF;
 
 _error:
-- 
GitLab