wait tables clearing before new initialization in tablet (#5419)

ivanmorozov333 · web-flow · commit c72cf7b6b4ec · 2024-06-11T10:19:38.000+03:00
diff --git a/ydb/core/tx/columnshard/columnshard__init.cpp b/ydb/core/tx/columnshard/columnshard__init.cpp
@@ -223,6 +223,8 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
         Self->SharingSessionsManager = local;
     }
 
+    Self->ProgressTxController->StartOperators();
+
     Self->UpdateInsertTableCounters();
     Self->UpdateIndexCounters();
     Self->UpdateResourceMetrics(ctx, {});
diff --git a/ydb/core/tx/columnshard/columnshard_impl.cpp b/ydb/core/tx/columnshard/columnshard_impl.cpp
@@ -72,6 +72,7 @@ TColumnShard::TColumnShard(TTabletStorageInfo* info, const TActorId& tablet)
     , StatsReportInterval(NYDBTest::TControllers::GetColumnShardController()->GetStatsReportInterval(TSettings::DefaultStatsReportInterval))
     , InFlightReadsTracker(StoragesManager)
     , TablesManager(StoragesManager, info->TabletID)
+    , Subscribers(std::make_shared<NSubscriber::TManager>(*this))
     , PipeClientCache(NTabletPipe::CreateBoundedClientCache(new NTabletPipe::TBoundedClientCacheConfig(), GetPipeClientConfig()))
     , InsertTable(std::make_unique<NOlap::TInsertTable>())
     , SubscribeCounters(std::make_shared<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters>())
diff --git a/ydb/core/tx/columnshard/columnshard_impl.h b/ydb/core/tx/columnshard/columnshard_impl.h
@@ -28,6 +28,8 @@
 #include "data_sharing/common/transactions/tx_extension.h"
 #include "data_sharing/modification/events/change_owning.h"
 
+#include "subscriber/abstract/manager/manager.h"
+
 #include <ydb/core/base/tablet_pipecache.h>
 #include <ydb/core/tablet/tablet_counters.h>
 #include <ydb/core/tablet/tablet_pipe_client_cache.h>
@@ -84,10 +86,11 @@ class TGeneralCompactColumnEngineChanges;
 
 namespace NKikimr::NColumnShard {
 
-
+class TTxFinishAsyncTransaction;
 class TTxInsertTableCleanup;
 class TTxRemoveSharedBlobs;
 class TOperationsManager;
+class TWaitEraseTablesTxSubscriber;
 
 extern bool gAllowLogBatchingDefaultValue;
 
@@ -150,6 +153,8 @@ class TColumnShard
     friend class TTxApplyNormalizer;
     friend class TTxMonitoring;
     friend class TTxRemoveSharedBlobs;
+    friend class TTxFinishAsyncTransaction;
+    friend class TWaitEraseTablesTxSubscriber;
 
     friend class NOlap::TCleanupPortionsColumnEngineChanges;
     friend class NOlap::TCleanupTablesColumnEngineChanges;
@@ -490,6 +495,7 @@ class TColumnShard
 
     TInFlightReadsTracker InFlightReadsTracker;
     TTablesManager TablesManager;
+    std::shared_ptr<NSubscriber::TManager> Subscribers;
     std::shared_ptr<TTiersManager> Tiers;
     std::unique_ptr<TTabletCountersBase> TabletCountersPtr;
     TTabletCountersBase* TabletCounters;
diff --git a/ydb/core/tx/columnshard/data_sharing/common/context/context.cpp b/ydb/core/tx/columnshard/data_sharing/common/context/context.cpp
@@ -12,9 +12,7 @@ NKikimrColumnShardDataSharingProto::TTransferContext TTransferContext::Serialize
     for (auto&& i : SourceTabletIds) {
         result.AddSourceTabletIds((ui64)i);
     }
-    if (SnapshotBarrier) {
-        SnapshotBarrier->SerializeToProto(*result.MutableSnapshotBarrier());
-    }
+    GetSnapshotBarrierVerified().SerializeToProto(*result.MutableSnapshotBarrier());
     result.SetMoving(Moving);
     return result;
 }
@@ -59,7 +57,7 @@ bool TTransferContext::IsEqualTo(const TTransferContext& context) const {
 }
 
 TString TTransferContext::DebugString() const {
-    return TStringBuilder() << "{from=" << (ui64)DestinationTabletId << ";moving=" << Moving << ";snapshot=" << (SnapshotBarrier ? SnapshotBarrier->DebugString() : "NO") << "}";
+    return TStringBuilder() << "{from=" << (ui64)DestinationTabletId << ";moving=" << Moving << ";snapshot=" << SnapshotBarrier.DebugString() << "}";
 }
 
 TTransferContext::TTransferContext(const TTabletId destination, const THashSet<TTabletId>& sources, const TSnapshot& snapshotBarrier, const bool moving, const std::optional<ui64> txId)
@@ -74,13 +72,8 @@ TTransferContext::TTransferContext(const TTabletId destination, const THashSet<T
 }
 
 const NKikimr::NOlap::TSnapshot& TTransferContext::GetSnapshotBarrierVerified() const {
-    AFL_VERIFY(!!SnapshotBarrier);
-    return *SnapshotBarrier;
-}
-
-void TTransferContext::SetSnapshotBarrier(const TSnapshot& snapshot) {
-    AFL_VERIFY(!SnapshotBarrier || *SnapshotBarrier == snapshot);
-    SnapshotBarrier = snapshot;
+    AFL_VERIFY(SnapshotBarrier.Valid());
+    return SnapshotBarrier;
 }
 
 }
diff --git a/ydb/core/tx/columnshard/data_sharing/common/context/context.h b/ydb/core/tx/columnshard/data_sharing/common/context/context.h
@@ -15,7 +15,7 @@ class TTransferContext {
     YDB_READONLY(TTabletId, DestinationTabletId, (TTabletId)0);
     YDB_READONLY_DEF(THashSet<TTabletId>, SourceTabletIds);
     YDB_READONLY(bool, Moving, false);
-    std::optional<TSnapshot> SnapshotBarrier;
+    TSnapshot SnapshotBarrier = TSnapshot::Zero();
     YDB_READONLY_DEF(std::optional<ui64>, TxId);
 public:
     TTransferContext() = default;
@@ -24,8 +24,6 @@ class TTransferContext {
 
     const TSnapshot& GetSnapshotBarrierVerified() const;
 
-    void SetSnapshotBarrier(const TSnapshot& snapshot);
-
     TTransferContext(const TTabletId destination, const THashSet<TTabletId>& sources, const TSnapshot& snapshotBarrier, const bool moving, const std::optional<ui64> txId = {});
     NKikimrColumnShardDataSharingProto::TTransferContext SerializeToProto() const;
     TConclusionStatus DeserializeFromProto(const NKikimrColumnShardDataSharingProto::TTransferContext& proto);
diff --git a/ydb/core/tx/columnshard/data_sharing/destination/session/destination.h b/ydb/core/tx/columnshard/data_sharing/destination/session/destination.h
@@ -90,10 +90,6 @@ class TDestinationSession: public TCommonSession {
 public:
     bool TryTakePortionBlobs(const TVersionedIndex& vIndex, const TPortionInfo& portion);
 
-    void SetBarrierSnapshot(const TSnapshot& value) {
-        TransferContext.SetSnapshotBarrier(value);
-    }
-
     TSourceCursorForDestination& GetCursorVerified(const TTabletId& tabletId) {
         auto it = Cursors.find(tabletId);
         AFL_VERIFY(it != Cursors.end());
diff --git a/ydb/core/tx/columnshard/engines/changes/cleanup_tables.cpp b/ydb/core/tx/columnshard/engines/changes/cleanup_tables.cpp
@@ -3,6 +3,7 @@
 #include <ydb/core/tx/columnshard/engines/column_engine_logs.h>
 #include <ydb/core/tx/columnshard/blobs_action/blob_manager_db.h>
 #include <ydb/core/tx/columnshard/columnshard_schema.h>
+#include <ydb/core/tx/columnshard/subscriber/events/tables_erased/event.h>
 #include <util/string/join.h>
 
 namespace NKikimr::NOlap {
@@ -25,6 +26,7 @@ void TCleanupTablesColumnEngineChanges::DoWriteIndexOnComplete(NColumnShard::TCo
     for (auto&& t : TablesToDrop) {
         self->TablesManager.TryFinalizeDropPathOnComplete(t);
     }
+    self->Subscribers->OnEvent(std::make_shared<NColumnShard::NSubscriber::TEventTablesErased>(TablesToDrop));
 }
 
 void TCleanupTablesColumnEngineChanges::DoStart(NColumnShard::TColumnShard& self) {
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/events/event.cpp b/ydb/core/tx/columnshard/subscriber/abstract/events/event.cpp
@@ -0,0 +1,4 @@
+#include "event.h"
+
+namespace NKikimr::NColumnShard::NSubscriber {
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/events/event.h b/ydb/core/tx/columnshard/subscriber/abstract/events/event.h
@@ -0,0 +1,32 @@
+#pragma once
+#include <ydb/library/accessor/accessor.h>
+#include <util/generic/string.h>
+
+namespace NKikimr::NColumnShard::NSubscriber {
+
+enum class EEventType {
+    Undefined,
+    TablesErased
+};
+
+class ISubscriptionEvent {
+private:
+    YDB_READONLY(EEventType, Type, EEventType::Undefined);
+    virtual TString DoDebugString() const {
+        return "";
+    }
+public:
+    virtual ~ISubscriptionEvent() = default;
+
+    ISubscriptionEvent(const EEventType type)
+        : Type(type)
+    {
+
+    }
+
+    TString DebugString() const {
+        return DoDebugString();
+    }
+};
+
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/events/ya.make b/ydb/core/tx/columnshard/subscriber/abstract/events/ya.make
@@ -0,0 +1,12 @@
+LIBRARY()
+
+SRCS(
+    event.cpp
+)
+
+PEERDIR(
+)
+
+GENERATE_ENUM_SERIALIZATION(event.h)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/manager/manager.cpp b/ydb/core/tx/columnshard/subscriber/abstract/manager/manager.cpp
@@ -0,0 +1,4 @@
+#include "manager.h"
+
+namespace NKikimr::NColumnShard::NSubscriber {
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/manager/manager.h b/ydb/core/tx/columnshard/subscriber/abstract/manager/manager.h
@@ -0,0 +1,81 @@
+#pragma once
+#include <ydb/core/tx/columnshard/subscriber/abstract/subscriber/subscriber.h>
+#include <ydb/core/tx/columnshard/subscriber/abstract/events/event.h>
+
+#include <ydb/library/actors/core/log.h>
+#include <util/generic/hash.h>
+#include <vector>
+
+namespace NKikimr::NColumnShard::NSubscriber {
+
+class TManager {
+private:
+    class TSharedPtrHashContainer {
+    private:
+        std::shared_ptr<ISubscriber> Object;
+    public:
+        TSharedPtrHashContainer(const std::shared_ptr<ISubscriber>& obj)
+            : Object(obj)
+        {
+
+        }
+
+        TSharedPtrHashContainer() {
+            AFL_VERIFY(!!Object);
+        }
+
+        ISubscriber* operator->() const {
+            return Object.get();
+        }
+
+        explicit operator size_t() const {
+            return (size_t)Object.get();
+        }
+
+        bool operator==(const TSharedPtrHashContainer& object) const {
+            return Object == object.Object;
+        }
+    };
+    TColumnShard& Owner;
+    THashMap<EEventType, THashSet<TSharedPtrHashContainer>> Subscribers;
+public:
+    TManager(TColumnShard& owner)
+        : Owner(owner)
+    {
+
+    }
+
+    void RegisterSubscriber(const std::shared_ptr<ISubscriber>& s) {
+        for (auto&& et : s->GetEventTypes()) {
+            AFL_VERIFY(Subscribers[et].emplace(s).second);
+        }
+    }
+
+    void OnEvent(const std::shared_ptr<ISubscriptionEvent>& ev) {
+        auto it = Subscribers.find(ev->GetType());
+        if (it == Subscribers.end()) {
+            AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "on_event_subscriber_skipped")("event", ev->GetType())("details", ev->DebugString());
+            return;
+        } else {
+            AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "on_event_subscriber")("event", ev->GetType())("details", ev->DebugString());
+        }
+        std::vector<TSharedPtrHashContainer> toRemove;
+        for (auto&& i : it->second) {
+            i->OnEvent(ev, Owner);
+            if (i->IsFinished()) {
+                toRemove.emplace_back(i);
+            }
+        }
+        for (auto&& i : toRemove) {
+            for (auto&& evType : i->GetEventTypes()) {
+                auto it = Subscribers.find(evType);
+                AFL_VERIFY(it != Subscribers.end());
+                it->second.erase(i);
+                if (it->second.empty()) {
+                    Subscribers.erase(it);
+                }
+            }
+        }
+    }
+};
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/manager/ya.make b/ydb/core/tx/columnshard/subscriber/abstract/manager/ya.make
@@ -0,0 +1,11 @@
+LIBRARY()
+
+SRCS(
+    manager.cpp
+)
+
+PEERDIR(
+    ydb/library/actors/core
+)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/subscriber/subscriber.cpp b/ydb/core/tx/columnshard/subscriber/abstract/subscriber/subscriber.cpp
@@ -0,0 +1,4 @@
+#include "subscriber.h"
+
+namespace NKikimr::NColumnShard::NSubscriber {
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/subscriber/subscriber.h b/ydb/core/tx/columnshard/subscriber/abstract/subscriber/subscriber.h
@@ -0,0 +1,26 @@
+#pragma once
+#include <ydb/core/tx/columnshard/subscriber/abstract/events/event.h>
+#include <memory>
+#include <set>
+
+namespace NKikimr::NColumnShard {
+class TColumnShard;
+}
+
+namespace NKikimr::NColumnShard::NSubscriber {
+
+class ISubscriber {
+private:
+    virtual bool DoOnEvent(const std::shared_ptr<ISubscriptionEvent>& ev, TColumnShard& shard) = 0;
+public:
+    bool OnEvent(const std::shared_ptr<ISubscriptionEvent>& ev, TColumnShard& shard) {
+        return DoOnEvent(ev, shard);
+    }
+
+    virtual std::set<EEventType> GetEventTypes() const = 0;
+    virtual bool IsFinished() const = 0;
+
+    virtual ~ISubscriber() = default;
+};
+
+}
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/subscriber/ya.make b/ydb/core/tx/columnshard/subscriber/abstract/subscriber/ya.make
@@ -0,0 +1,10 @@
+LIBRARY()
+
+SRCS(
+    subscriber.cpp
+)
+
+PEERDIR(
+)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/abstract/ya.make b/ydb/core/tx/columnshard/subscriber/abstract/ya.make
@@ -0,0 +1,9 @@
+LIBRARY()
+
+PEERDIR(
+    ydb/core/tx/columnshard/subscriber/abstract/events
+    ydb/core/tx/columnshard/subscriber/abstract/events
+    ydb/core/tx/columnshard/subscriber/abstract/subscriber
+)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/events/tables_erased/event.cpp b/ydb/core/tx/columnshard/subscriber/events/tables_erased/event.cpp
@@ -0,0 +1,10 @@
+#include "event.h"
+#include <util/string/join.h>
+
+namespace NKikimr::NColumnShard::NSubscriber {
+
+TString TEventTablesErased::DoDebugString() const {
+    return "paths=" + JoinSeq(",", PathIds);
+}
+
+}
diff --git a/ydb/core/tx/columnshard/subscriber/events/tables_erased/event.h b/ydb/core/tx/columnshard/subscriber/events/tables_erased/event.h
@@ -0,0 +1,19 @@
+#pragma once
+#include <ydb/core/tx/columnshard/subscriber/abstract/events/event.h>
+#include <util/generic/hash_set.h>
+
+namespace NKikimr::NColumnShard::NSubscriber {
+class TEventTablesErased: public ISubscriptionEvent {
+private:
+    using TBase = ISubscriptionEvent;
+    YDB_READONLY_DEF(THashSet<ui64>, PathIds);
+    virtual TString DoDebugString() const override;
+public:
+    TEventTablesErased(const THashSet<ui64>& pathIds)
+        : TBase(EEventType::TablesErased)
+        , PathIds(pathIds)
+    {
+
+    }
+};
+}
diff --git a/ydb/core/tx/columnshard/subscriber/events/tables_erased/ya.make b/ydb/core/tx/columnshard/subscriber/events/tables_erased/ya.make
@@ -0,0 +1,10 @@
+LIBRARY()
+
+SRCS(
+    event.cpp
+)
+
+PEERDIR(
+)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/events/ya.make b/ydb/core/tx/columnshard/subscriber/events/ya.make
@@ -0,0 +1,7 @@
+LIBRARY()
+
+PEERDIR(
+    ydb/core/tx/columnshard/subscriber/events/tables_erased
+)
+
+END()
diff --git a/ydb/core/tx/columnshard/subscriber/ya.make b/ydb/core/tx/columnshard/subscriber/ya.make
diff --git a/ydb/core/tx/columnshard/tables_manager.cpp b/ydb/core/tx/columnshard/tables_manager.cpp
diff --git a/ydb/core/tx/columnshard/transactions/operators/schema.cpp b/ydb/core/tx/columnshard/transactions/operators/schema.cpp
diff --git a/ydb/core/tx/columnshard/transactions/operators/schema.h b/ydb/core/tx/columnshard/transactions/operators/schema.h
diff --git a/ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.cpp b/ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.cpp
diff --git a/ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.h b/ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.h
diff --git a/ydb/core/tx/columnshard/transactions/transactions/ya.make b/ydb/core/tx/columnshard/transactions/transactions/ya.make
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.h b/ydb/core/tx/columnshard/transactions/tx_controller.h
diff --git a/ydb/core/tx/columnshard/ya.make b/ydb/core/tx/columnshard/ya.make

Original file line number	Diff line number	Diff line change
`@@ -223,6 +223,8 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)`
`223`	`223`	`Self->SharingSessionsManager = local;`
`224`	`224`	`}`
`225`	`225`
	`226`	`+ Self->ProgressTxController->StartOperators();`
	`227`	`+`
`226`	`228`	`Self->UpdateInsertTableCounters();`
`227`	`229`	`Self->UpdateIndexCounters();`
`228`	`230`	`Self->UpdateResourceMetrics(ctx, {});`
Original file line number	Diff line number	Diff line change
`@@ -12,9 +12,7 @@ NKikimrColumnShardDataSharingProto::TTransferContext TTransferContext::Serialize`
`12`	`12`	`for (auto&& i : SourceTabletIds) {`
`13`	`13`	`result.AddSourceTabletIds((ui64)i);`
`14`	`14`	`}`
`15`		`- if (SnapshotBarrier) {`
`16`		`- SnapshotBarrier->SerializeToProto(*result.MutableSnapshotBarrier());`
`17`		`- }`
	`15`	`+ GetSnapshotBarrierVerified().SerializeToProto(*result.MutableSnapshotBarrier());`
`18`	`16`	`result.SetMoving(Moving);`
`19`	`17`	`return result;`
`20`	`18`	`}`
`@@ -59,7 +57,7 @@ bool TTransferContext::IsEqualTo(const TTransferContext& context) const {`
`59`	`57`	`}`
`60`	`58`
`61`	`59`	`TString TTransferContext::DebugString() const {`
`62`		`- return TStringBuilder() << "{from=" << (ui64)DestinationTabletId << ";moving=" << Moving << ";snapshot=" << (SnapshotBarrier ? SnapshotBarrier->DebugString() : "NO") << "}";`
	`60`	`+ return TStringBuilder() << "{from=" << (ui64)DestinationTabletId << ";moving=" << Moving << ";snapshot=" << SnapshotBarrier.DebugString() << "}";`
`63`	`61`	`}`
`64`	`62`
`65`	`63`	`TTransferContext::TTransferContext(const TTabletId destination, const THashSet<TTabletId>& sources, const TSnapshot& snapshotBarrier, const bool moving, const std::optional<ui64> txId)`
`@@ -74,13 +72,8 @@ TTransferContext::TTransferContext(const TTabletId destination, const THashSet<T`
`74`	`72`	`}`
`75`	`73`
`76`	`74`	`const NKikimr::NOlap::TSnapshot& TTransferContext::GetSnapshotBarrierVerified() const {`
`77`		`- AFL_VERIFY(!!SnapshotBarrier);`
`78`		`- return *SnapshotBarrier;`
`79`		`-}`
`80`		`-`
`81`		`-void TTransferContext::SetSnapshotBarrier(const TSnapshot& snapshot) {`
`82`		`- AFL_VERIFY(!SnapshotBarrier \|\| *SnapshotBarrier == snapshot);`
`83`		`- SnapshotBarrier = snapshot;`
	`75`	`+ AFL_VERIFY(SnapshotBarrier.Valid());`
	`76`	`+ return SnapshotBarrier;`
`84`	`77`	`}`
`85`	`78`
`86`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@`
`3`	`3`	`#include <ydb/core/tx/columnshard/engines/column_engine_logs.h>`
`4`	`4`	`#include <ydb/core/tx/columnshard/blobs_action/blob_manager_db.h>`
`5`	`5`	`#include <ydb/core/tx/columnshard/columnshard_schema.h>`
	`6`	`+#include <ydb/core/tx/columnshard/subscriber/events/tables_erased/event.h>`
`6`	`7`	`#include <util/string/join.h>`
`7`	`8`
`8`	`9`	`namespace NKikimr::NOlap {`
`@@ -25,6 +26,7 @@ void TCleanupTablesColumnEngineChanges::DoWriteIndexOnComplete(NColumnShard::TCo`
`25`	`26`	`for (auto&& t : TablesToDrop) {`
`26`	`27`	`self->TablesManager.TryFinalizeDropPathOnComplete(t);`
`27`	`28`	`}`
	`29`	`+ self->Subscribers->OnEvent(std::make_shared<NColumnShard::NSubscriber::TEventTablesErased>(TablesToDrop));`
`28`	`30`	`}`
`29`	`31`
`30`	`32`	`void TCleanupTablesColumnEngineChanges::DoStart(NColumnShard::TColumnShard& self) {`
-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
 +#include "event.h"
++
 +namespace NKikimr::NColumnShard::NSubscriber {
 +}