WIP

Author: Enjection

ydb/core/protos/flat_scheme_op.proto

@@ -346,6 +346,8 @@ message TTableDescription {
     optional bool AllowUnderSameOperation = 44 [default = false];
     // Create only as-well. Used for CopyTable to create table in desired state instead of default
     optional EPathState PathState = 46;
+    // Skip automatic index/impl table copying - indexes will be handled separately
+    optional bool OmitIndexes = 47 [default = false];
 }
 
 message TDictionaryEncodingSettings {
@@ -1282,6 +1284,11 @@ message TCopyTableConfig { //TTableDescription implemets copying a table in orig
     optional bool AllowUnderSameOperation = 7 [default = false];
 
     optional NKikimrSchemeOp.EPathState TargetPathTargetState = 8;
+    
+    // Map from index impl table name to CDC stream config for incremental backups
+    // Key: index impl table name (e.g., "indexImplTable")
+    // Value: CDC stream configuration to create on that index impl table
+    map<string, TCreateCdcStream> IndexImplTableCdcStreams = 9;
 }
 
 message TConsistentTableCopyingConfig {

ydb/core/tx/datashard/datashard_ut_incremental_backup.cpp

@@ -508,6 +508,127 @@ Y_UNIT_TEST_SUITE(IncrementalBackup) {
             "{ items { uint32_value: 3 } items { uint32_value: 30 } }");
     }
 
+    Y_UNIT_TEST(SimpleBackupRestoreWithIndex) {
+        TPortManager portManager;
+        TServer::TPtr server = new TServer(TServerSettings(portManager.GetPort(2134), {}, DefaultPQConfig())
+            .SetUseRealThreads(false)
+            .SetDomainName("Root")
+            .SetEnableBackupService(true)
+            .SetEnableChangefeedInitialScan(true)
+        );
+
+        auto& runtime = *server->GetRuntime();
+        const auto edgeActor = runtime.AllocateEdgeActor();
+
+        SetupLogging(runtime);
+        InitRoot(server, edgeActor);
+
+        // Create table with a global index
+        CreateShardedTable(server, edgeActor, "/Root", "TableWithIndex",
+            TShardedTableOptions()
+                .Columns({
+                    {"key", "Uint32", true, false},
+                    {"value", "Uint32", false, false},
+                    {"indexed", "Uint32", false, false}
+                })
+                .Indexes({
+                    {"idx", {"indexed"}, {}, NKikimrSchemeOp::EIndexTypeGlobal}
+                }));
+
+        // Insert test data
+        ExecSQL(server, edgeActor, R"(
+            UPSERT INTO `/Root/TableWithIndex` (key, value, indexed) VALUES
+            (1, 10, 100),
+            (2, 20, 200),
+            (3, 30, 300);
+        )");
+
+        // Verify index works before backup
+        auto beforeBackup = KqpSimpleExecSuccess(runtime, R"(
+            SELECT key FROM `/Root/TableWithIndex` VIEW idx WHERE indexed = 200
+        )");
+        UNIT_ASSERT_C(beforeBackup.find("uint32_value: 2") != TString::npos, 
+            "Index should work before backup: " << beforeBackup);
+
+        // Create backup collection
+        ExecSQL(server, edgeActor, R"(
+            CREATE BACKUP COLLECTION `TestCollection`
+              ( TABLE `/Root/TableWithIndex` )
+            WITH
+              ( STORAGE = 'cluster'
+              , INCREMENTAL_BACKUP_ENABLED = 'true'
+              );
+        )", false);
+
+        // Perform full backup
+        ExecSQL(server, edgeActor, R"(BACKUP `TestCollection`;)", false);
+        
+        // Wait longer for CDC streams to be fully created (AtTable phase completes async)
+        // The CDC AtTable phase increments schema versions and syncs indexes
+        SimulateSleep(server, TDuration::Seconds(5));
+
+        // Add version diagnostics after backup
+        {
+            Cerr << "========== VERSION DIAGNOSTICS AFTER BACKUP ==========" << Endl;
+            
+            auto describeTable = Ls(runtime, edgeActor, "/Root/TableWithIndex");
+            if (!describeTable->ResultSet.empty() && describeTable->ResultSet[0].TableId) {
+                Cerr << "Main table SchemaVersion: " << describeTable->ResultSet[0].TableId.SchemaVersion << Endl;
+                if (describeTable->ResultSet[0].Self) {
+                    Cerr << "Main table PathVersion: " << describeTable->ResultSet[0].Self->Info.GetPathVersion() << Endl;
+                }
+            }
+            
+            auto describeIndex = Ls(runtime, edgeActor, "/Root/TableWithIndex/idx");
+            if (!describeIndex->ResultSet.empty() && describeIndex->ResultSet[0].Self) {
+                Cerr << "Index PathVersion: " << describeIndex->ResultSet[0].Self->Info.GetPathVersion() << Endl;
+            }
+            
+            auto describeImplTable = Ls(runtime, edgeActor, "/Root/TableWithIndex/idx/indexImplTable");
+            if (!describeImplTable->ResultSet.empty() && describeImplTable->ResultSet[0].TableId) {
+                Cerr << "Index impl table SchemaVersion: " << describeImplTable->ResultSet[0].TableId.SchemaVersion << Endl;
+                if (describeImplTable->ResultSet[0].Self) {
+                    Cerr << "Index impl table PathVersion: " << describeImplTable->ResultSet[0].Self->Info.GetPathVersion() << Endl;
+                }
+            }
+            
+            Cerr << "======================================================" << Endl;
+        }
+
+        // Capture expected data
+        auto expectedData = KqpSimpleExecSuccess(runtime, R"(
+            SELECT key, value, indexed FROM `/Root/TableWithIndex` ORDER BY key
+        )");
+
+        // Drop table
+        ExecSQL(server, edgeActor, R"(DROP TABLE `/Root/TableWithIndex`;)", false);
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        // Restore from backup
+        ExecSQL(server, edgeActor, R"(RESTORE `TestCollection`;)", false);
+        runtime.SimulateSleep(TDuration::Seconds(5));
+
+        // Verify data is restored
+        auto actualData = KqpSimpleExecSuccess(runtime, R"(
+            SELECT key, value, indexed FROM `/Root/TableWithIndex` ORDER BY key
+        )");
+        UNIT_ASSERT_VALUES_EQUAL(expectedData, actualData);
+
+        // Verify index still exists and works after restore
+        auto afterRestore = KqpSimpleExecSuccess(runtime, R"(
+            SELECT key FROM `/Root/TableWithIndex` VIEW idx WHERE indexed = 200
+        )");
+        UNIT_ASSERT_C(afterRestore.find("uint32_value: 2") != TString::npos, 
+            "Index should work after restore: " << afterRestore);
+
+        // Verify index implementation table exists and has data
+        auto indexImplData = KqpSimpleExecSuccess(runtime, R"(
+            SELECT COUNT(*) FROM `/Root/TableWithIndex/idx/indexImplTable`
+        )");
+        UNIT_ASSERT_C(indexImplData.find("uint64_value: 3") != TString::npos, 
+            "Index impl table should have 3 rows: " << indexImplData);
+    }
+
     Y_UNIT_TEST(MultiBackup) {
         TPortManager portManager;
         TServer::TPtr server = new TServer(TServerSettings(portManager.GetPort(2134), {}, DefaultPQConfig())