WIP

.github/config/muted_ya.txt

@@ -49,6 +49,9 @@ ydb/core/transfer/ut/large TransferLarge.Transfer100KM_10P_RowTable_TopicAutoPar
 ydb/core/transfer/ut/large TransferLarge.Transfer1KM_1KP_RowTable_TopicAutoPartitioning
 ydb/core/tx/conveyor_composite/ut CompositeConveyorTests.TestUniformDistribution
 ydb/core/tx/datashard/ut_incremental_backup IncrementalBackup.ComplexRestoreBackupCollection+WithIncremental
+ydb/core/tx/datashard/ut_incremental_backup IncrementalBackup.E2EBackupCollection
+ydb/core/tx/datashard/ut_incremental_backup IncrementalBackup.SimpleBackupBackupCollection+WithIncremental
+ydb/core/tx/datashard/ut_incremental_backup IncrementalBackup.SimpleRestoreBackupCollection+WithIncremental
 ydb/core/tx/schemeshard/ut_background_cleaning TSchemeshardBackgroundCleaningTest.SchemeshardBackgroundCleaningTestCreateCleanManyTables
 ydb/core/tx/schemeshard/ut_index_build VectorIndexBuildTest.BaseCase
 ydb/core/tx/schemeshard/ut_login_large TSchemeShardLoginLargeTest.RemoveLogin_Many
@@ -315,4 +318,4 @@ ydb/tests/stress/log/tests test_workload.py.TestYdbLogWorkload.test[column]
 ydb/tests/stress/mixedpy test_mixed.py.TestYdbMixedWorkload.test[column]
 ydb/tests/stress/reconfig_state_storage_workload/tests sole chunk chunk
 ydb/tests/stress/reconfig_state_storage_workload/tests test_board_workload.py.TestReconfigStateStorageBoardWorkload.test_state_storage_board
-ydb/tools/stress_tool/ut TDeviceTestTool.PDiskTestLogWrite
+ydb/tools/stress_tool/ut TDeviceTestTool.PDiskTestLogWrite

ydb/core/protos/counters_schemeshard.proto

@@ -251,6 +251,9 @@ enum ESimpleCounters {
     COUNTER_SYS_VIEW_COUNT = 196                    [(CounterOpts) = {Name: "SysViewCount"}];
     COUNTER_IN_FLIGHT_OPS_TxCreateSysView = 197     [(CounterOpts) = {Name: "InFlightOps/CreateSysView"}];
     COUNTER_IN_FLIGHT_OPS_TxDropSysView = 198       [(CounterOpts) = {Name: "InFlightOps/DropSysView"}];
+
+    COUNTER_IN_FLIGHT_OPS_TxCreateLongIncrementalRestoreOp = 199       [(CounterOpts) = {Name: "InFlightOps/CreateLongIncrementalRestoreOp"}];
+    COUNTER_IN_FLIGHT_OPS_TxChangePathState = 200                     [(CounterOpts) = {Name: "InFlightOps/ChangePathState"}];
 }
 
 enum ECumulativeCounters {
@@ -404,6 +407,9 @@ enum ECumulativeCounters {
 
     COUNTER_FINISHED_OPS_TxCreateSysView = 119              [(CounterOpts) = {Name: "FinishedOps/CreateSysView"}];
     COUNTER_FINISHED_OPS_TxDropSysView = 120                [(CounterOpts) = {Name: "FinishedOps/DropSysView"}];
+
+    COUNTER_FINISHED_OPS_TxCreateLongIncrementalRestoreOp = 121       [(CounterOpts) = {Name: "FinishedOps/CreateLongIncrementalRestoreOp"}];
+    COUNTER_FINISHED_OPS_TxChangePathState = 122                     [(CounterOpts) = {Name: "FinishedOps/ChangePathState"}];
 }
 
 enum EPercentileCounters {

ydb/core/protos/flat_scheme_op.proto

@@ -1797,6 +1797,8 @@ message TModifyScheme {
 
     optional TSysViewDescription CreateSysView = 81;
 
+    optional TChangePathState ChangePathState = 82;
+
     // Some entries are grouped by semantics, so are out of order
 }
 
@@ -2235,3 +2237,17 @@ message TSysViewDescription {
     optional string Name = 1;
     optional NKikimrSysView.ESysViewType Type = 2;
 }
+
+message TLongIncrementalRestoreOp {
+    optional uint64 TxId = 1;
+    optional NKikimrProto.TPathID BackupCollectionPathId = 2;
+    optional string Id = 3;
+    repeated string TablePathList = 4;
+    optional string FullBackupTrimmedName = 5;
+    repeated string IncrementalBackupTrimmedNames = 6;
+}
+
+message TChangePathState {
+    optional string Path = 1;
+    optional EPathState TargetState = 2;
+}

ydb/core/protos/schemeshard/operations.proto

@@ -178,5 +178,11 @@ enum EOperationType {
     ESchemeOpCreateSysView = 116;
     ESchemeOpDropSysView = 117;
 
+    // Long Incremental Restore
+    ESchemeOpCreateLongIncrementalRestoreOp = 118;
+
+    // Change Path State
+    ESchemeOpChangePathState = 119;
+
     // Some entries are grouped by semantics, so are out of order
 }

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -3597,6 +3597,15 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                             txState.TargetPathTargetState = proto.GetTxCopyTableExtraData().GetTargetPathTargetState();
                         }
                     }
+                } else if (txState.TxType == TTxState::TxChangePathState) {
+                    if (!extraData.empty()) {
+                        NKikimrSchemeOp::TGenericTxInFlyExtraData proto;
+                        bool deserializeRes = ParseFromStringNoSizeLimit(proto, extraData);
+                        Y_ABORT_UNLESS(deserializeRes);
+                        if (proto.GetTxCopyTableExtraData().HasTargetPathTargetState()) {
+                            txState.TargetPathTargetState = proto.GetTxCopyTableExtraData().GetTargetPathTargetState();
+                        }
+                    }
                 }
 
                 Y_ABORT_UNLESS(txState.TxType != TTxState::TxInvalid);

ydb/core/tx/schemeshard/schemeshard__operation.cpp

@@ -1269,6 +1269,13 @@ ISubOperation::TPtr TOperation::RestorePart(TTxState::ETxType txType, TTxState::
     case TTxState::ETxType::TxDropSysView:
         return CreateDropSysView(NextPartId(), txState);
 
+    // ChangePathState
+    case TTxState::ETxType::TxChangePathState:
+        return CreateChangePathState(NextPartId(), txState);
+
+    case TTxState::ETxType::TxCreateLongIncrementalRestoreOp:
+        return CreateLongIncrementalRestoreOpControlPlane(NextPartId(), txState);
+
     case TTxState::ETxType::TxInvalid:
         Y_UNREACHABLE();
     }
@@ -1567,12 +1574,18 @@ TVector<ISubOperation::TPtr> TDefaultOperationFactory::MakeOperationParts(
         return CreateBackupIncrementalBackupCollection(op.NextPartId(), tx, context);
     case NKikimrSchemeOp::EOperationType::ESchemeOpRestoreBackupCollection:
         return CreateRestoreBackupCollection(op.NextPartId(), tx, context);
+    case NKikimrSchemeOp::EOperationType::ESchemeOpCreateLongIncrementalRestoreOp:
+        return {CreateLongIncrementalRestoreOpControlPlane(op.NextPartId(), tx)};
 
     // SysView
     case NKikimrSchemeOp::EOperationType::ESchemeOpCreateSysView:
         return {CreateNewSysView(op.NextPartId(), tx)};
     case NKikimrSchemeOp::EOperationType::ESchemeOpDropSysView:
         return {CreateDropSysView(op.NextPartId(), tx)};
+
+    // ChangePathState
+    case NKikimrSchemeOp::EOperationType::ESchemeOpChangePathState:
+        return CreateChangePathState(op.NextPartId(), tx, context);
     }
 
     Y_UNREACHABLE();

ydb/core/tx/schemeshard/schemeshard__operation_alter_cdc_stream.cpp

@@ -144,7 +144,13 @@ class TAlterCdcStream: public TSubOperation {
                 .NotDeleted()
                 .IsTable()
                 .NotAsyncReplicaTable()
-                .NotUnderOperation();
+                .NotUnderDeleting();
+
+            // Allow CDC operations on tables that are under incremental backup/restore
+            if (checks && tablePath.IsUnderOperation() &&
+                !tablePath.IsUnderOutgoingIncrementalRestore()) {
+                checks.NotUnderOperation();
+            }
 
             if (checks && !tablePath.IsInsideTableIndexPath()) {
                 checks.IsCommonSensePath();
@@ -374,8 +380,13 @@ class TAlterCdcStreamAtTable: public TSubOperation {
                 .NotDeleted()
                 .IsTable()
                 .NotAsyncReplicaTable()
-                .NotUnderDeleting()
-                .NotUnderOperation();
+                .NotUnderDeleting();
+
+            // Allow CDC operations on tables that are under incremental backup/restore
+            if (checks && tablePath.IsUnderOperation() &&
+                !tablePath.IsUnderOutgoingIncrementalRestore()) {
+                checks.NotUnderOperation();
+            }
 
             if (checks && !tablePath.IsInsideTableIndexPath()) {
                 checks.IsCommonSensePath();
@@ -498,8 +509,13 @@ std::variant<TStreamPaths, ISubOperation::TPtr> DoAlterStreamPathChecks(
             .IsResolved()
             .NotDeleted()
             .IsTable()
-            .NotAsyncReplicaTable()
-            .NotUnderOperation();
+            .NotAsyncReplicaTable();
+
+        // Allow CDC operations on tables that are under incremental backup/restore
+        if (checks && tablePath.IsUnderOperation() && 
+            !tablePath.IsUnderOutgoingIncrementalRestore()) {
+            checks.NotUnderOperation();
+        }
 
         if (checks && !tablePath.IsInsideTableIndexPath()) {
             checks.IsCommonSensePath();

ydb/core/tx/schemeshard/schemeshard__operation_alter_extsubdomain.cpp

@@ -2,6 +2,7 @@
 #include "schemeshard_impl.h"
 #include "schemeshard__operation_common.h"
 #include "schemeshard__operation_common_subdomain.h"
+#include "schemeshard__operation_states.h"
 #include "schemeshard_utils.h"  // for TransactionTemplate
 
 #include <ydb/core/base/subdomain.h>
@@ -561,33 +562,6 @@ class TCreateHive: public TSubOperationState {
     }
 };
 
-class TEmptyPropose: public TSubOperationState {
-private:
-    TOperationId OperationId;
-
-    TString DebugHint() const override {
-        return TStringBuilder() << "TEmptyPropose, operationId " << OperationId << ", ";
-    }
-
-public:
-    TEmptyPropose(TOperationId id)
-        : OperationId(id)
-    {
-        IgnoreMessages(DebugHint(), {});
-    }
-
-    bool ProgressState(TOperationContext& context) override {
-        TTxState* txState = context.SS->FindTx(OperationId);
-        Y_ABORT_UNLESS(txState);
-
-        LOG_I(DebugHint() << "ProgressState, operation type " << TTxState::TypeName(txState->TxType));
-
-        context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, TStepId(0));
-
-        return true;
-    }
-};
-
 class TAlterExtSubDomainCreateHive: public TSubOperation {
     static TTxState::ETxState NextState() {
         return TTxState::CreateParts;

ydb/core/tx/schemeshard/schemeshard__operation_base.h

@@ -0,0 +1,63 @@
+#pragma once
+
+#include "schemeshard__operation_part.h"
+#include "schemeshard_impl.h"
+
+namespace NKikimr::NSchemeShard {
+
+/**
+ * Base class for sub-operations that need proper context-aware state management.
+ * This class provides common implementations for operations that:
+ * 1. Always use SetState with context for proper state transitions
+ * 2. Only use the context-aware SelectStateFunc overload
+ * 
+ * Derived classes should override SelectStateFunc(TTxState::ETxState, TOperationContext&)
+ * instead of the non-context version.
+ * 
+ * For operations that need custom NextState logic with context, they should override
+ * StateDone to call their custom NextState method.
+ */
+class TSubOperationWithContext : public TSubOperation {
+protected:
+    using TSubOperation::SelectStateFunc;
+    using TSubOperation::NextState;
+    /**
+     * Properly handles state transitions by calling SetState with context
+     * and activating the transaction when moving to a valid next state.
+     * 
+     * Derived classes can override this if they need custom StateDone logic.
+     * 
+     * @param context The operation context containing database and completion handlers
+     */
+    void StateDone(TOperationContext& context) override {
+        if (GetState() == TTxState::Done) {
+            return;
+        }
+
+        TTxState::ETxState nextState;
+        nextState = NextState(GetState());
+
+        SetState(nextState, context);
+
+        if (nextState != TTxState::Invalid) {
+            context.OnComplete.ActivateTx(OperationId);
+        }
+    }
+
+    /**
+     * Default implementation that prevents use of the non-context SelectStateFunc.
+     * Derived classes should override SelectStateFunc(TTxState::ETxState, TOperationContext&)
+     * instead of this method.
+     * 
+     * @param state The transaction state (unused)
+     * @return Always aborts, forcing use of context-aware version
+     */
+    TSubOperationState::TPtr SelectStateFunc(TTxState::ETxState) override {
+        Y_ABORT("Unreachable code: TSubOperationWithContext should only use context-aware SelectStateFunc");
+    }
+
+public:
+    using TSubOperation::TSubOperation;
+};
+
+} // namespace NKikimr::NSchemeShard