Remove old schema snapshot upon adding new one

Author: CyberROFL

ydb/core/tx/datashard/datashard.cpp

@@ -1495,9 +1495,14 @@ void TDataShard::ScheduleRemoveAbandonedSchemaSnapshots() {
             Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
             break;
         }
-        if (snapshot.Schema->GetTableSchemaVersion() < it->second->GetTableSchemaVersion()) {
-            PendingSchemaSnapshotsToRemove.push_back(key);
+        if (SchemaSnapshotManager.HasReference(key)) {
+            continue;
+        }
+        if (snapshot.Schema->GetTableSchemaVersion() >= it->second->GetTableSchemaVersion()) {
+            continue;
         }
+
+        PendingSchemaSnapshotsToRemove.push_back(key);
     }
 
     if (wasEmpty && !PendingSchemaSnapshotsToRemove.empty()) {
@@ -1733,8 +1738,18 @@ void TDataShard::AddSchemaSnapshot(const TPathId& pathId, ui64 tableSchemaVersio
     Y_ABORT_UNLESS(TableInfos.contains(pathId.LocalPathId));
     auto tableInfo = TableInfos[pathId.LocalPathId];
 
-    const auto key = TSchemaSnapshotKey(pathId.OwnerId, pathId.LocalPathId, tableSchemaVersion);
+    const auto key = TSchemaSnapshotKey(pathId, tableSchemaVersion);
     SchemaSnapshotManager.AddSnapshot(txc.DB, key, TSchemaSnapshot(tableInfo, step, txId));
+
+    const auto& snapshots = SchemaSnapshotManager.GetSnapshots();
+    for (auto it = snapshots.lower_bound(TSchemaSnapshotKey(pathId, 1)); it != snapshots.end(); ++it) {
+        if (it->first == key) {
+            break;
+        }
+        if (!SchemaSnapshotManager.HasReference(it->first)) {
+            ScheduleRemoveSchemaSnapshot(it->first);
+        }
+    }
 }
 
 void TDataShard::PersistLastLoanTableTid(NIceDb::TNiceDb& db, ui32 localTid) {

ydb/core/tx/datashard/datashard_schema_snapshots.cpp

@@ -80,14 +80,16 @@ const TSchemaSnapshot* TSchemaSnapshotManager::FindSnapshot(const TSchemaSnapsho
     return Snapshots.FindPtr(key);
 }
 
-void TSchemaSnapshotManager::RemoveShapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key) {
+void TSchemaSnapshotManager::RemoveShapshot(NTable::TDatabase& db, const TSchemaSnapshotKey& key) {
     auto it = Snapshots.find(key);
     if (it == Snapshots.end()) {
         return;
     }
 
     Snapshots.erase(it);
-    PersistRemoveSnapshot(db, key);
+
+    NIceDb::TNiceDb nicedb(db);
+    PersistRemoveSnapshot(nicedb, key);
 }
 
 void TSchemaSnapshotManager::RenameSnapshots(NTable::TDatabase& db,

ydb/core/tx/datashard/datashard_schema_snapshots.h

@@ -33,7 +33,7 @@ class TSchemaSnapshotManager {
 
     bool AddSnapshot(NTable::TDatabase& db, const TSchemaSnapshotKey& key, const TSchemaSnapshot& snapshot);
     const TSchemaSnapshot* FindSnapshot(const TSchemaSnapshotKey& key) const;
-    void RemoveShapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key);
+    void RemoveShapshot(NTable::TDatabase& db, const TSchemaSnapshotKey& key);
     void RenameSnapshots(NTable::TDatabase& db, const TPathId& prevTableId, const TPathId& newTableId);
     const TSnapshots& GetSnapshots() const;
 

ydb/core/tx/datashard/remove_schema_snapshots.cpp

@@ -11,8 +11,6 @@ class TDataShard::TTxRemoveSchemaSnapshots: public NTabletFlatExecutor::TTransac
     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);
@@ -38,7 +36,7 @@ class TDataShard::TTxRemoveSchemaSnapshots: public NTabletFlatExecutor::TTransac
                 continue;
             }
 
-            Self->GetSchemaSnapshotManager().RemoveShapshot(db, key);
+            Self->GetSchemaSnapshotManager().RemoveShapshot(txc.DB, key);
             Self->PendingSchemaSnapshotsToRemove.pop_back();
         }