fix: schema version handling

Enjection · Enjection · commit cab75265ae82 · 2025-11-08T02:02:53.000Z
diff --git a/ydb/core/tx/datashard/datashard_ut_incremental_backup.cpp b/ydb/core/tx/datashard/datashard_ut_incremental_backup.cpp
@@ -3012,6 +3012,89 @@ Y_UNIT_TEST_SUITE(IncrementalBackup) {
         UNIT_ASSERT_C(!indexBackupExists, "Index backup should NOT exist when OmitIndexes flag is set");
     }
 
+    Y_UNIT_TEST(Test) {
+        TPortManager portManager;
+        TServer::TPtr server = new TServer(TServerSettings(portManager.GetPort(2134), {}, DefaultPQConfig())
+            .SetUseRealThreads(false)
+            .SetDomainName("Root")
+            .SetEnableChangefeedInitialScan(true)
+            .SetEnableBackupService(true)
+            .SetEnableRealSystemViewPaths(false)
+        );
+
+        auto& runtime = *server->GetRuntime();
+        const auto edgeActor = runtime.AllocateEdgeActor();
+
+        SetupLogging(runtime);
+        InitRoot(server, edgeActor);
+
+        // Create first table with index
+        CreateShardedTable(server, edgeActor, "/Root", "Table1",
+            TShardedTableOptions()
+                .Columns({
+                    {"key", "Uint32", true, false},
+                    {"val1", "Uint32", false, false}
+                })
+                .Indexes({
+                    {"idx1", {"val1"}, {}, NKikimrSchemeOp::EIndexTypeGlobal}
+                }));
+
+        // Create second table with different index
+        CreateShardedTable(server, edgeActor, "/Root", "Table2",
+            TShardedTableOptions()
+                .Columns({
+                    {"key", "Uint32", true, false},
+                    {"val2", "Uint32", false, false}
+                })
+                .Indexes({
+                    {"idx2", {"val2"}, {}, NKikimrSchemeOp::EIndexTypeGlobal}
+                }));
+
+        // Insert data into both tables
+        ExecSQL(server, edgeActor, R"(
+            UPSERT INTO `/Root/Table1` (key, val1) VALUES (1, 100), (2, 200);
+            UPSERT INTO `/Root/Table2` (key, val2) VALUES (1, 1000), (2, 2000);
+        )");
+
+        // Create backup collection with both tables
+        ExecSQL(server, edgeActor, R"(
+            CREATE BACKUP COLLECTION `MultiTableCollection`
+              ( TABLE `/Root/Table1`
+              , TABLE `/Root/Table2`
+              )
+            WITH
+              ( STORAGE = 'cluster'
+              , INCREMENTAL_BACKUP_ENABLED = 'true'
+              );
+        )", false);
+
+        // Full backup
+        ExecSQL(server, edgeActor, R"(BACKUP `MultiTableCollection`;)", false);
+        SimulateSleep(server, TDuration::Seconds(1));
+
+        // Modify both tables
+        ExecSQL(server, edgeActor, R"(
+            UPSERT INTO `/Root/Table1` (key, val1) VALUES (3, 300);
+            UPSERT INTO `/Root/Table2` (key, val2) VALUES (3, 3000);
+        )");
+
+        // Incremental backup
+        ExecSQL(server, edgeActor, R"(BACKUP `MultiTableCollection` INCREMENTAL;)", false);
+        SimulateSleep(server, TDuration::Seconds(5));
+
+        // Capture expected states
+        ExecSQL(server, edgeActor, R"(
+            SELECT key, val1 FROM `/Root/Table1` ORDER BY key
+        )");
+        
+        ExecSQL(server, edgeActor, R"(
+            SELECT key, val2 FROM `/Root/Table2` ORDER BY key
+        )");
+
+        // Drop both tables
+        ExecSQL(server, edgeActor, R"(DROP TABLE `/Root/Table1`;)", false);
+    }
+
 } // Y_UNIT_TEST_SUITE(IncrementalBackup)
 
 } // NKikimr
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp
@@ -124,8 +124,6 @@ bool TProposeAtTable::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOpera
     Y_ABORT_UNLESS(context.SS->Tables.contains(pathId));
     auto table = context.SS->Tables.at(pathId);
 
-    table->AlterVersion += 1;
-
     NIceDb::TNiceDb db(context.GetDB());
 
     bool isContinuousBackupStream = false;
@@ -147,32 +145,119 @@ bool TProposeAtTable::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOpera
     }
 
     // Check if this is an index implementation table
-    // If so, we need to sync the parent index version to match the impl table version
-    // Do this ONLY for continuous backup operations
+    bool isIndexImplTable = false;
     TPathId parentPathId = path->ParentPathId;
-    if (parentPathId && context.SS->PathsById.contains(parentPathId) && isContinuousBackupStream) {
+    TPathId grandParentPathId;
+    if (parentPathId && context.SS->PathsById.contains(parentPathId)) {
         auto parentPath = context.SS->PathsById.at(parentPathId);
         if (parentPath->IsTableIndex()) {
-            Y_ABORT_UNLESS(context.SS->Indexes.contains(parentPathId));
+            isIndexImplTable = true;
+            grandParentPathId = parentPath->ParentPathId;
+        }
+    }
+
+    // For index impl tables: synchronize with parent table version
+    // This applies to ALL operations on impl tables, not just continuous backup
+    if (isIndexImplTable && grandParentPathId && context.SS->Tables.contains(grandParentPathId)) {
+        auto parentTable = context.SS->Tables.at(grandParentPathId);
+        ui64 currentImplVersion = table->AlterVersion;
+        ui64 currentParentVersion = parentTable->AlterVersion;
+
+        // Impl table should stay synchronized with parent table
+        // If behind or equal, sync to parent version
+        if (currentImplVersion <= currentParentVersion) {
+            table->AlterVersion = currentParentVersion;
+            LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
+                        DebugHint() << " Synchronized index impl table version to parent table"
+                                   << ", implTablePathId: " << pathId
+                                   << ", parentTablePathId: " << grandParentPathId
+                                   << ", oldImplVersion: " << currentImplVersion
+                                   << ", parentVersion: " << currentParentVersion
+                                   << ", newImplVersion: " << table->AlterVersion
+                                   << ", continuous backup: " << (isContinuousBackupStream ? "yes" : "no")
+                                   << ", at schemeshard: " << context.SS->SelfTabletId());
+        } else {
+            // Impl version is ahead (shouldn't happen, but handle gracefully)
+            table->AlterVersion += 1;
+            LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
+                        DebugHint() << " WARNING: Impl table version ahead of parent, incrementing"
+                                   << ", implTablePathId: " << pathId
+                                   << ", implVersion: " << currentImplVersion
+                                   << ", parentVersion: " << currentParentVersion
+                                   << ", newImplVersion: " << table->AlterVersion
+                                   << ", at schemeshard: " << context.SS->SelfTabletId());
+        }
+
+        // ALWAYS sync parent index entity version to match impl table
+        // This is critical to avoid version mismatches
+        if (context.SS->Indexes.contains(parentPathId)) {
             auto index = context.SS->Indexes.at(parentPathId);
-            
             index->AlterVersion = table->AlterVersion;
-            
-            // Persist the index version update directly to database
+
+            // Persist the index version update
             db.Table<Schema::TableIndex>().Key(parentPathId.LocalPathId).Update(
                 NIceDb::TUpdate<Schema::TableIndex::AlterVersion>(index->AlterVersion)
             );
-            
+
+            auto parentPath = context.SS->PathsById.at(parentPathId);
             context.SS->ClearDescribePathCaches(parentPath);
             context.OnComplete.PublishToSchemeBoard(OperationId, parentPathId);
 
             LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
-                        DebugHint() << " Synced parent index version with impl table"
+                        DebugHint() << " Synced parent index entity version to match impl table"
                                    << ", indexPathId: " << parentPathId
-                                   << ", indexName: " << parentPath->Name
                                    << ", newVersion: " << index->AlterVersion
                                    << ", at schemeshard: " << context.SS->SelfTabletId());
         }
+    } else {
+        table->AlterVersion += 1;
+        LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
+                    DebugHint() << " Incremented table version"
+                               << ", pathId: " << pathId
+                               << ", newVersion: " << table->AlterVersion
+                               << ", isIndexImpl: " << (isIndexImplTable ? "yes" : "no")
+                               << ", at schemeshard: " << context.SS->SelfTabletId());
+    }
+
+    // For parent tables with indexes in continuous backup:
+    // Sync the index entity version to match the parent table
+    // Note: impl table version is synced during impl table's own CDC operation
+    if (isContinuousBackupStream && !isIndexImplTable) {
+        for (const auto& [childName, childPathId] : path->GetChildren()) {
+            auto childPath = context.SS->PathsById.at(childPathId);
+
+            // Skip non-index children (CDC streams, etc.)
+            if (!childPath->IsTableIndex()) {
+                continue;
+            }
+
+            // Skip deleted indexes
+            if (childPath->Dropped()) {
+                continue;
+            }
+
+            // Sync parent index version with parent table version
+            if (context.SS->Indexes.contains(childPathId)) {
+                auto index = context.SS->Indexes.at(childPathId);
+                index->AlterVersion = table->AlterVersion;
+
+                // Persist the index version update
+                db.Table<Schema::TableIndex>().Key(childPathId.LocalPathId).Update(
+                    NIceDb::TUpdate<Schema::TableIndex::AlterVersion>(index->AlterVersion)
+                );
+
+                context.SS->ClearDescribePathCaches(childPath);
+                context.OnComplete.PublishToSchemeBoard(OperationId, childPathId);
+
+                LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
+                            DebugHint() << " Synced parent index version with parent table"
+                                       << ", parentTable: " << path->Name
+                                       << ", indexName: " << childName
+                                       << ", indexPathId: " << childPathId
+                                       << ", newVersion: " << index->AlterVersion
+                                       << ", at schemeshard: " << context.SS->SelfTabletId());
+            }
+        }
     }
 
     context.SS->PersistTableAlterVersion(db, pathId, table);
