[stable-25-3] backport incr backup restore for indexes

.github/config/muted_ya.txt

@@ -20,6 +20,26 @@ ydb/core/kqp/ut/query KqpAnalyze.AnalyzeTable+ColumnStore
 ydb/core/kqp/ut/scheme KqpOlapScheme.AddPgColumnWithStore
 ydb/core/persqueue/ut/ut_with_sdk TopicTimestamp.TimestampRead_40MB_Topic_offset+middle_WithRestart
 ydb/core/statistics/aggregator/ut AnalyzeColumnshard.AnalyzeRebootColumnShard
+ydb/core/sys_view/ut AuthSystemView.AuthGroupMembers_Access
+ydb/core/sys_view/ut AuthSystemView.AuthGroups_Access
+ydb/core/sys_view/ut AuthSystemView.AuthOwners_Access
+ydb/core/sys_view/ut AuthSystemView.AuthPermissions_Access
+ydb/core/sys_view/ut SystemView.ShowCreateTableChangefeeds
+ydb/core/sys_view/ut SystemView.SystemViewFailOps+EnableRealSystemViewPaths
+ydb/core/tablet_flat/ut_large TFlatTableLongTxLarge.LargeDeltaChain
+ydb/core/tablet_flat/ut_large unittest.sole chunk
+ydb/core/transfer/ut/large TransferLarge.Transfer100KM_10P_LocalRead_TopicAutoPartitioning
+ydb/core/transfer/ut/large TransferLarge.Transfer100KM_10P_RowTable_TopicAutoPartitioning
+ydb/core/tx/datashard/ut_order DataShardTxOrder.RandomPointsAndRanges
+ydb/core/tx/datashard/ut_order unittest.[*/*] chunk
+ydb/core/tx/schemeshard/ut_background_cleaning TSchemeshardBackgroundCleaningTest.SchemeshardBackgroundCleaningTestCreateCleanManyTables
+ydb/core/tx/schemeshard/ut_base_reboots TTablesWithReboots.CopyIndexedTableWithReboots
+ydb/core/tx/schemeshard/ut_base_reboots unittest.[*/*] chunk
+ydb/core/tx/schemeshard/ut_cdc_stream_reboots TCdcStreamWithRebootsTests.CreateDropRecreate[TabletReboots]
+ydb/core/tx/schemeshard/ut_cdc_stream_reboots unittest.[*/*] chunk
+ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.CancelOnOnSingleTopic
+ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.CancelOnSingleShardTableWithChangefeed
+ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.CancelShouldSucceedOnMultiShardTable
 ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.CancelShouldSucceedOnSingleShardTableWithUniqueIndex
 ydb/core/tx/schemeshard/ut_vector_index_build_reboots VectorIndexBuildTestReboots.BaseCase+Prefixed[TabletReboots]
 ydb/core/tx/schemeshard/ut_vector_index_build_reboots unittest.[*/*] chunk

ydb/core/kqp/host/kqp_gateway_proxy.cpp

@@ -1339,7 +1339,8 @@ class TKqpGatewayProxy : public IKikimrGateway {
             op.SetPrefix(settings.Prefix);
 
             if (settings.Settings.IncrementalBackupEnabled) {
-                op.MutableIncrementalBackupConfig();
+                auto* config = op.MutableIncrementalBackupConfig();
+                config->SetOmitIndexes(settings.Settings.OmitIndexes);
             }
 
             auto errOpt = std::visit(

ydb/core/kqp/provider/yql_kikimr_exec.cpp

@@ -1028,6 +1028,13 @@ namespace {
                         "INCREMENTAL_BACKUP_ENABLED must be true or false"));
                     return false;
                 }
+            } else if (name == "omit_indexes") {
+                auto value = ToString(setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value());
+                if (!TryFromString(value, dstSettings.OmitIndexes)) {
+                    ctx.AddError(TIssue(ctx.GetPosition(pos),
+                        "OMIT_INDEXES must be true or false"));
+                    return false;
+                }
             } else if (name == "storage") {
                 auto value = ToString(setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value());
                 if (to_lower(value) != "cluster") {

ydb/core/kqp/provider/yql_kikimr_gateway.h

@@ -1003,7 +1003,8 @@ struct TAnalyzeSettings {
 };
 
 struct TBackupCollectionSettings {
-    bool IncrementalBackupEnabled;
+    bool IncrementalBackupEnabled = false;
+    bool OmitIndexes = false;
 };
 
 struct TCreateBackupCollectionSettings {

ydb/core/kqp/provider/yql_kikimr_type_ann.cpp

@@ -2396,6 +2396,7 @@ virtual TStatus HandleCreateTable(TKiCreateTable create, TExprContext& ctx) over
         const THashSet<TString> supportedSettings = {
             "incremental_backup_enabled",
             "storage",
+            "omit_indexes",
         };
 
         if (!CheckBackupCollectionSettings(node.BackupCollectionSettings(), supportedSettings, ctx)) {

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 {
@@ -1273,6 +1275,11 @@ message TCopyTableConfig { //TTableDescription implemets copying a table in orig
     optional bool AllowUnderSameOperation = 7 [default = false];
 
     optional NKikimrSchemeOp.EPathState TargetPathTargetState = 8;
+
+    // Map from index name to CDC stream config for incremental backups
+    // Key: index name (e.g., "age_index", "name_index")
+    // Value: CDC stream configuration to create on that index's impl table
+    map<string, TCreateCdcStream> IndexImplTableCdcStreams = 9;
 }
 
 message TConsistentTableCopyingConfig {
@@ -2302,7 +2309,7 @@ message TBackupCollectionDescription {
     }
 
     message TIncrementalBackupConfig {
-
+        optional bool OmitIndexes = 1 [default = false];
     }
 
     oneof Entries {
@@ -2316,6 +2323,8 @@ message TBackupCollectionDescription {
     oneof Storage {
         google.protobuf.Empty Cluster = 7;
     }
+
+    optional bool OmitIndexes = 9 [default = false];
 }
 
 message TBackupBackupCollection {

ydb/core/tx/datashard/datashard_ut_common_kqp.h

@@ -3,6 +3,7 @@
 #include <ydb/core/tx/datashard/ut_common/datashard_ut_common.h>
 #include <ydb/core/grpc_services/local_rpc/local_rpc.h>
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/result/result.h>
+#include <ydb/library/ut/ut.h>
 
 namespace NKikimr {
 namespace NDataShard {
@@ -182,6 +183,14 @@ namespace NKqpHelpers {
         return FormatResult(response);
     }
 
+    inline TString KqpSimpleExecSuccess(TTestActorRuntime& runtime, const TString& query, bool staleRo = false, const TString& database = {}, NYdb::NUt::TTestContext testCtx = NYdb::NUt::TTestContext()) {
+        auto response = AwaitResponse(runtime, KqpSimpleSend(runtime, query, staleRo, database));
+        CTX_UNIT_ASSERT_VALUES_EQUAL_C(response.operation().status(), Ydb::StatusIds::SUCCESS,
+            "Query failed: " << query << ", status: " << response.operation().status()
+            << ", issues: " << response.operation().issues());
+        return FormatResult(response);
+    }
+
     inline auto KqpSimpleStaleRoSend(TTestActorRuntime& runtime, const TString& query, const TString& database = {}) {
         return KqpSimpleSend(runtime, query, true, database);
     }