Remove schema snapshots, add change records to queue at init

CyberROFL · CyberROFL · commit 2e9f97227426 · 2024-07-10T13:51:15.000+03:00
diff --git a/ydb/core/protos/counters_datashard.proto b/ydb/core/protos/counters_datashard.proto
@@ -490,4 +490,5 @@ enum ETxTypes {
     TXTYPE_CLEANUP_VOLATILE = 80                          [(TxTypeOpts) = {Name: "TxCleanupVolatile"}];
     TXTYPE_PLAN_PREDICTED_TXS = 81                        [(TxTypeOpts) = {Name: "TxPlanPredictedTxs"}];
     TXTYPE_WRITE = 82                                     [(TxTypeOpts) = {Name: "TxWrite"}];
+    TXTYPE_REMOVE_SCHEMA_SNAPSHOTS = 83                   [(TxTypeOpts) = {Name: "TxRemoveSchemaSnapshots"}];
 }
diff --git a/ydb/core/tx/datashard/datashard.cpp b/ydb/core/tx/datashard/datashard.cpp
@@ -869,8 +869,10 @@ void TDataShard::PersistChangeRecord(NIceDb::TNiceDb& db, const TChangeRecord& r
             Y_VERIFY_S(it != ChangesQueue.end(), "Cannot find change record: " << order);
 
             if (it->second.SchemaSnapshotAcquired) {
-                SchemaSnapshotManager.ReleaseReference(
-                    TSchemaSnapshotKey(it->second.TableId, it->second.SchemaVersion));
+                const auto snapshotKey = TSchemaSnapshotKey(it->second.TableId, it->second.SchemaVersion);
+                if (const auto last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
+                    ScheduleRemoveSchemaSnapshot(snapshotKey);
+                }
             }
 
             ChangesQueue.erase(it);
@@ -998,8 +1000,10 @@ void TDataShard::CommitLockChangeRecords(NIceDb::TNiceDb& db, ui64 lockId, ui64
             Y_VERIFY_S(cIt != ChangesQueue.end(), "Cannot find change record: " << order);
 
             if (cIt->second.SchemaSnapshotAcquired) {
-                SchemaSnapshotManager.ReleaseReference(
-                    TSchemaSnapshotKey(cIt->second.TableId, cIt->second.SchemaVersion));
+                const auto snapshotKey = TSchemaSnapshotKey(cIt->second.TableId, cIt->second.SchemaVersion);
+                if (const auto last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
+                    ScheduleRemoveSchemaSnapshot(snapshotKey);
+                }
             }
 
             ChangesQueue.erase(cIt);
@@ -1067,23 +1071,9 @@ void TDataShard::RemoveChangeRecord(NIceDb::TNiceDb& db, ui64 order) {
     ChangesQueueBytes -= record.BodySize;
 
     if (record.SchemaSnapshotAcquired) {
-        Y_ABORT_UNLESS(record.TableId);
-        auto tableIt = TableInfos.find(record.TableId.LocalPathId);
-
-        if (tableIt != TableInfos.end()) {
-            const auto snapshotKey = TSchemaSnapshotKey(record.TableId, record.SchemaVersion);
-            const bool last = SchemaSnapshotManager.ReleaseReference(snapshotKey);
-
-            if (last) {
-                const auto* snapshot = SchemaSnapshotManager.FindSnapshot(snapshotKey);
-                Y_ABORT_UNLESS(snapshot);
-
-                if (snapshot->Schema->GetTableSchemaVersion() < tableIt->second->GetTableSchemaVersion()) {
-                    SchemaSnapshotManager.RemoveShapshot(db, snapshotKey);
-                }
-            }
-        } else {
-            Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
+        const auto snapshotKey = TSchemaSnapshotKey(record.TableId, record.SchemaVersion);
+        if (const bool last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
+            ScheduleRemoveSchemaSnapshot(snapshotKey);
         }
     }
 
@@ -1303,6 +1293,14 @@ bool TDataShard::LoadChangeRecords(NIceDb::TNiceDb& db, TVector<IDataShardChange
             .SchemaVersion = schemaVersion,
         });
 
+        auto res = ChangesQueue.emplace(records.back().Order, records.back());
+        Y_VERIFY_S(res.second, "Duplicate change record: " << records.back().Order);
+
+        if (res.first->second.SchemaVersion) {
+            res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
+                TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
+        }
+
         if (!rowset.Next()) {
             return false;
         }
@@ -1401,6 +1399,14 @@ bool TDataShard::LoadChangeRecordCommits(NIceDb::TNiceDb& db, TVector<IDataShard
             });
             entry.Count++;
             needSort = true;
+
+            auto res = ChangesQueue.emplace(records.back().Order, records.back());
+            Y_VERIFY_S(res.second, "Duplicate change record: " << records.back().Order);
+
+            if (res.first->second.SchemaVersion) {
+                res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
+                    TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
+            }
         }
 
         LockChangeRecords.erase(lockId);
@@ -1459,6 +1465,46 @@ void TDataShard::ScheduleRemoveAbandonedLockChanges() {
     }
 }
 
+void TDataShard::ScheduleRemoveSchemaSnapshot(const TSchemaSnapshotKey& key) {
+    Y_ABORT_UNLESS(!SchemaSnapshotManager.HasReference(key));
+
+    const auto* snapshot = SchemaSnapshotManager.FindSnapshot(key);
+    Y_ABORT_UNLESS(snapshot);
+
+    auto it = TableInfos.find(key.PathId);
+    if (it == TableInfos.end()) {
+        Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
+        return;
+    }
+
+    if (snapshot->Schema->GetTableSchemaVersion() < it->second->GetTableSchemaVersion()) {
+        bool wasEmpty = PendingSchemaSnapshotsToRemove.empty();
+        PendingSchemaSnapshotsToRemove.push_back(key);
+        if (wasEmpty) {
+            Send(SelfId(), new TEvPrivate::TEvRemoveSchemaSnapshots);
+        }
+    }
+}
+
+void TDataShard::ScheduleRemoveAbandonedSchemaSnapshots() {
+    bool wasEmpty = PendingSchemaSnapshotsToRemove.empty();
+
+    for (const auto& [key, snapshot] : SchemaSnapshotManager.GetSnapshots()) {
+        auto it = TableInfos.find(key.PathId);
+        if (it == TableInfos.end()) {
+            Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
+            break;
+        }
+        if (snapshot.Schema->GetTableSchemaVersion() < it->second->GetTableSchemaVersion()) {
+            PendingSchemaSnapshotsToRemove.push_back(key);
+        }
+    }
+
+    if (wasEmpty && !PendingSchemaSnapshotsToRemove.empty()) {
+        Send(SelfId(), new TEvPrivate::TEvRemoveSchemaSnapshots);
+    }
+}
+
 void TDataShard::PersistSchemeTxResult(NIceDb::TNiceDb &db, const TSchemaOperation &op) {
     db.Table<Schema::SchemaOperations>().Key(op.TxId).Update(
         NIceDb::TUpdate<Schema::SchemaOperations::Success>(op.Success),
diff --git a/ydb/core/tx/datashard/datashard__init.cpp b/ydb/core/tx/datashard/datashard__init.cpp
@@ -425,6 +425,12 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
             return false;
     }
 
+    if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::SchemaSnapshots::TableId)) {
+        if (!Self->SchemaSnapshotManager.Load(db)) {
+            return false;
+        }
+    }
+
     if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::ChangeRecords::TableId)) {
         if (!Self->LoadChangeRecords(db, ChangeRecords)) {
             return false;
@@ -512,12 +518,6 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
         }
     }
 
-    if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::SchemaSnapshots::TableId)) {
-        if (!Self->SchemaSnapshotManager.Load(db)) {
-            return false;
-        }
-    }
-
     if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::Locks::TableId)) {
         TDataShardLocksDb locksDb(*Self, txc);
         if (!Self->SysLocks.Load(locksDb)) {
@@ -547,6 +547,7 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
     Self->SubscribeNewLocks();
 
     Self->ScheduleRemoveAbandonedLockChanges();
+    Self->ScheduleRemoveAbandonedSchemaSnapshots();
 
     return true;
 }
diff --git a/ydb/core/tx/datashard/datashard_impl.h b/ydb/core/tx/datashard/datashard_impl.h
@@ -241,6 +241,7 @@ class TDataShard
     class TTxCdcStreamScanProgress;
     class TTxCdcStreamEmitHeartbeats;
     class TTxUpdateFollowerReadEdge;
+    class TTxRemoveSchemaSnapshots;
 
     template <typename T> friend class TTxDirectBase;
     class TTxUploadRows;
@@ -374,6 +375,7 @@ class TDataShard
             EvPlanPredictedTxs,
             EvStatisticsScanFinished,
             EvTableStatsError,
+            EvRemoveSchemaSnapshots,
             EvEnd
         };
 
@@ -595,6 +597,8 @@ class TDataShard
         struct TEvPlanPredictedTxs : public TEventLocal<TEvPlanPredictedTxs, EvPlanPredictedTxs> {};
 
         struct TEvStatisticsScanFinished : public TEventLocal<TEvStatisticsScanFinished, EvStatisticsScanFinished> {};
+
+        struct TEvRemoveSchemaSnapshots : public TEventLocal<TEvRemoveSchemaSnapshots, EvRemoveSchemaSnapshots> {};
     };
 
     struct Schema : NIceDb::Schema {
@@ -1383,6 +1387,8 @@ class TDataShard
 
     void Handle(TEvPrivate::TEvPlanPredictedTxs::TPtr& ev, const TActorContext& ctx);
 
+    void Handle(TEvPrivate::TEvRemoveSchemaSnapshots::TPtr& ev, const TActorContext& ctx);
+
     void HandleByReplicationSourceOffsetsServer(STATEFN_SIG);
 
     void DoPeriodicTasks(const TActorContext &ctx);
@@ -1920,6 +1926,8 @@ class TDataShard
     bool LoadChangeRecordCommits(NIceDb::TNiceDb& db, TVector<IDataShardChangeCollector::TChange>& records);
     void ScheduleRemoveLockChanges(ui64 lockId);
     void ScheduleRemoveAbandonedLockChanges();
+    void ScheduleRemoveSchemaSnapshot(const TSchemaSnapshotKey& key);
+    void ScheduleRemoveAbandonedSchemaSnapshots();
 
     static void PersistCdcStreamScanLastKey(NIceDb::TNiceDb& db, const TSerializedCellVec& value,
         const TPathId& tablePathId, const TPathId& streamPathId);
@@ -2871,6 +2879,7 @@ class TDataShard
     THashMap<ui64, TUncommittedLockChangeRecords> LockChangeRecords; // ui64 is lock id
     THashMap<ui64, TCommittedLockChangeRecords> CommittedLockChangeRecords; // ui64 is lock id
     TVector<ui64> PendingLockChangeRecordsToRemove;
+    TVector<TSchemaSnapshotKey> PendingSchemaSnapshotsToRemove;
 
     // in
     THashMap<ui64, TInChangeSender> InChangeSenders; // ui64 is shard id
@@ -2991,6 +3000,7 @@ class TDataShard
             HFuncTraced(TEvMediatorTimecast::TEvNotifyPlanStep, Handle);
             HFuncTraced(TEvPrivate::TEvMediatorRestoreBackup, Handle);
             HFuncTraced(TEvPrivate::TEvRemoveLockChangeRecords, Handle);
+            HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
         default:
             if (!HandleDefaultEvents(ev, SelfId())) {
                 ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateInactive unhandled event type: " << ev->GetTypeRewrite()
@@ -3119,6 +3129,7 @@ class TDataShard
             HFunc(TEvPrivate::TEvPlanPredictedTxs, Handle);
             HFunc(NStat::TEvStatistics::TEvStatisticsRequest, Handle);
             HFunc(TEvPrivate::TEvStatisticsScanFinished, Handle);
+            HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
             default:
                 if (!HandleDefaultEvents(ev, SelfId())) {
                     ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateWork unhandled event type: " << ev->GetTypeRewrite() << " event: " << ev->ToString());
diff --git a/ydb/core/tx/datashard/datashard_schema_snapshots.cpp b/ydb/core/tx/datashard/datashard_schema_snapshots.cpp
@@ -119,6 +119,10 @@ void TSchemaSnapshotManager::RenameSnapshots(NTable::TDatabase& db,
     }
 }
 
+const TSchemaSnapshotManager::TSnapshots& TSchemaSnapshotManager::GetSnapshots() const {
+    return Snapshots;
+}
+
 bool TSchemaSnapshotManager::AcquireReference(const TSchemaSnapshotKey& key) {
     auto it = Snapshots.find(key);
     if (it == Snapshots.end()) {
@@ -152,6 +156,15 @@ bool TSchemaSnapshotManager::ReleaseReference(const TSchemaSnapshotKey& key) {
     return true;
 }
 
+bool TSchemaSnapshotManager::HasReference(const TSchemaSnapshotKey& key) const {
+    auto refIt = References.find(key);
+    if (refIt != References.end()) {
+        return refIt->second;
+    } else {
+        return false;
+    }
+}
+
 void TSchemaSnapshotManager::PersistAddSnapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key, const TSchemaSnapshot& snapshot) {
     using Schema = TDataShard::Schema;
     db.Table<Schema::SchemaSnapshots>()
diff --git a/ydb/core/tx/datashard/datashard_schema_snapshots.h b/ydb/core/tx/datashard/datashard_schema_snapshots.h
@@ -23,6 +23,8 @@ struct TSchemaSnapshot {
 };
 
 class TSchemaSnapshotManager {
+    using TSnapshots = TMap<TSchemaSnapshotKey, TSchemaSnapshot, TLess<void>>;
+
 public:
     explicit TSchemaSnapshotManager(const TDataShard* self);
 
@@ -33,17 +35,19 @@ class TSchemaSnapshotManager {
     const TSchemaSnapshot* FindSnapshot(const TSchemaSnapshotKey& key) const;
     void RemoveShapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key);
     void RenameSnapshots(NTable::TDatabase& db, const TPathId& prevTableId, const TPathId& newTableId);
+    const TSnapshots& GetSnapshots() const;
 
     bool AcquireReference(const TSchemaSnapshotKey& key);
     bool ReleaseReference(const TSchemaSnapshotKey& key);
+    bool HasReference(const TSchemaSnapshotKey& key) const;
 
 private:
     void PersistAddSnapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key, const TSchemaSnapshot& snapshot);
     void PersistRemoveSnapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key);
 
 private:
     const TDataShard* Self;
-    TMap<TSchemaSnapshotKey, TSchemaSnapshot, TLess<void>> Snapshots;
+    TSnapshots Snapshots;
     THashMap<TSchemaSnapshotKey, size_t> References;
 
 }; // TSchemaSnapshotManager
diff --git a/ydb/core/tx/datashard/remove_schema_snapshots.cpp b/ydb/core/tx/datashard/remove_schema_snapshots.cpp
@@ -0,0 +1,56 @@
+#include "datashard_impl.h"
+
+namespace NKikimr::NDataShard {
+
+class TDataShard::TTxRemoveSchemaSnapshots: public NTabletFlatExecutor::TTransactionBase<TDataShard> {
+public:
+    TTxRemoveSchemaSnapshots(TDataShard* self)
+        : TBase(self)
+    { }
+
+    TTxType GetTxType() const override { return TXTYPE_REMOVE_SCHEMA_SNAPSHOTS; }
+
+    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+        NIceDb::TNiceDb db(txc.DB);
+
+        while (!Self->PendingSchemaSnapshotsToRemove.empty()) {
+            const auto key = Self->PendingSchemaSnapshotsToRemove.back();
+            const auto* snapshot = Self->GetSchemaSnapshotManager().FindSnapshot(key);
+
+            if (!snapshot) {
+                Self->PendingSchemaSnapshotsToRemove.pop_back();
+                continue;
+            }
+
+            if (Self->GetSchemaSnapshotManager().HasReference(key)) {
+                Self->PendingSchemaSnapshotsToRemove.pop_back();
+                continue;
+            }
+
+            auto table = Self->FindUserTable(TPathId(key.OwnerId, key.PathId));
+            if (!table) {
+                Self->PendingSchemaSnapshotsToRemove.pop_back();
+                continue;
+            }
+
+            if (snapshot->Schema->GetTableSchemaVersion() >= table->GetTableSchemaVersion()) {
+                Self->PendingSchemaSnapshotsToRemove.pop_back();
+                continue;
+            }
+
+            Self->GetSchemaSnapshotManager().RemoveShapshot(db, key);
+            Self->PendingSchemaSnapshotsToRemove.pop_back();
+        }
+
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {
+    }
+};
+
+void TDataShard::Handle(TEvPrivate::TEvRemoveSchemaSnapshots::TPtr&, const TActorContext& ctx) {
+    Execute(new TTxRemoveSchemaSnapshots(this), ctx);
+}
+
+} // namespace NKikimr::NDataShard
diff --git a/ydb/core/tx/datashard/ya.make b/ydb/core/tx/datashard/ya.make
@@ -189,6 +189,7 @@ SRCS(
     receive_snapshot_unit.cpp
     remove_lock_change_records.cpp
     remove_locks.cpp
+    remove_schema_snapshots.cpp
     range_ops.cpp
     read_iterator.h
     restore_unit.cpp

Original file line number	Diff line number	Diff line change
`@@ -490,4 +490,5 @@ enum ETxTypes {`
`490`	`490`	`TXTYPE_CLEANUP_VOLATILE = 80 [(TxTypeOpts) = {Name: "TxCleanupVolatile"}];`
`491`	`491`	`TXTYPE_PLAN_PREDICTED_TXS = 81 [(TxTypeOpts) = {Name: "TxPlanPredictedTxs"}];`
`492`	`492`	`TXTYPE_WRITE = 82 [(TxTypeOpts) = {Name: "TxWrite"}];`
	`493`	`+ TXTYPE_REMOVE_SCHEMA_SNAPSHOTS = 83 [(TxTypeOpts) = {Name: "TxRemoveSchemaSnapshots"}];`
`493`	`494`	`}`