Reset RowsCache

6 files changed, 20 insertions, 0 deletions

diff --git a/ydb/core/tx/datashard/change_collector_async_index.cpp b/ydb/core/tx/datashard/change_collector_async_index.cpp
index cc00c433e83..d012f961899 100644
--- a/ydb/core/tx/datashard/change_collector_async_index.cpp
+++ b/ydb/core/tx/datashard/change_collector_async_index.cpp
@@ -187,6 +187,11 @@ bool TAsyncIndexChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
     return true;
 }
 
+void TAsyncIndexChangeCollector::Reset() {
+    TBaseChangeCollector::Reset();
+    RowsCache.Reset();
+}
+
 auto TAsyncIndexChangeCollector::CacheTags(const TTableId& tableId) const {
     Y_VERIFY(Self->GetUserTables().contains(tableId.PathId.LocalPathId));
     auto userTable = Self->GetUserTables().at(tableId.PathId.LocalPathId);
diff --git a/ydb/core/tx/datashard/change_collector_async_index.h b/ydb/core/tx/datashard/change_collector_async_index.h
index 9160f6d8992..c6404f03f96 100644
--- a/ydb/core/tx/datashard/change_collector_async_index.h
+++ b/ydb/core/tx/datashard/change_collector_async_index.h
@@ -57,6 +57,8 @@ public:
     bool Collect(const TTableId& tableId, NTable::ERowOp rop,
         TArrayRef<const TRawTypeValue> key, TArrayRef<const NTable::TUpdateOp> updates) override;
 
+    void Reset() override;
+
 private:
     TRowVersion ReadVersion;
 
diff --git a/ydb/core/tx/datashard/change_collector_cdc_stream.cpp b/ydb/core/tx/datashard/change_collector_cdc_stream.cpp
index a64d9977752..0f4c6de4008 100644
--- a/ydb/core/tx/datashard/change_collector_cdc_stream.cpp
+++ b/ydb/core/tx/datashard/change_collector_cdc_stream.cpp
@@ -165,6 +165,11 @@ bool TCdcStreamChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
     return true;
 }
 
+void TCdcStreamChangeCollector::Reset() {
+    TBaseChangeCollector::Reset();
+    RowsCache.Reset();
+}
+
 TMaybe<TRowState> TCdcStreamChangeCollector::GetCurrentState(ui32 tid, TArrayRef<const TRawTypeValue> key,
         TArrayRef<const TTag> keyTags, TArrayRef<const TTag> valueTags)
 {
diff --git a/ydb/core/tx/datashard/change_collector_cdc_stream.h b/ydb/core/tx/datashard/change_collector_cdc_stream.h
index 631a91c24e7..83dd37376c1 100644
--- a/ydb/core/tx/datashard/change_collector_cdc_stream.h
+++ b/ydb/core/tx/datashard/change_collector_cdc_stream.h
@@ -28,6 +28,8 @@ public:
     bool Collect(const TTableId& tableId, NTable::ERowOp rop,
         TArrayRef<const TRawTypeValue> key, TArrayRef<const NTable::TUpdateOp> updates) override;
 
+    void Reset() override;
+
 private:
     TRowVersion ReadVersion;
 
diff --git a/ydb/core/tx/datashard/change_collector_helpers.cpp b/ydb/core/tx/datashard/change_collector_helpers.cpp
index fce849a0764..2a13eb57ff3 100644
--- a/ydb/core/tx/datashard/change_collector_helpers.cpp
+++ b/ydb/core/tx/datashard/change_collector_helpers.cpp
@@ -106,6 +106,10 @@ void TRowsCache::UpdateCachedRow(ui32 tid, ERowOp rop,
     }
 }
 
+void TRowsCache::Reset() {
+    Rows.clear();
+}
+
 const TRowsCache::TRow* TRowsCache::CacheRow(ui32 tid, EReady ready,
         TArrayRef<const TRawTypeValue> key, TArrayRef<const TTag> tags, const TRowState& rowState)
 {
diff --git a/ydb/core/tx/datashard/change_collector_helpers.h b/ydb/core/tx/datashard/change_collector_helpers.h
index b41c5986f27..54bba063cdf 100644
--- a/ydb/core/tx/datashard/change_collector_helpers.h
+++ b/ydb/core/tx/datashard/change_collector_helpers.h
@@ -27,6 +27,8 @@ public:
     void UpdateCachedRow(ui32 tid, NTable::ERowOp rop,
         TArrayRef<const TRawTypeValue> key, TArrayRef<const NTable::TUpdateOp> updates);
 
+    void Reset();
+
 private:
     THashMap<ui32, THashMap<TString, TRow>> Rows;