Merge c3d5f34 into b3c26e4

Author: Enjection

ydb/core/protos/flat_scheme_op.proto

@@ -926,9 +926,12 @@ message TAlterContinuousBackup {
     message TStop {
     }
 
+    message TTakeIncrementalBackup {
+    }
+
     oneof Action {
         TStop Stop = 2;
-        // TODO(innokentii): something like TakeIncremental
+        TTakeIncrementalBackup TakeIncrementalBackup = 3;
     }
 }
 

ydb/core/protos/pqconfig.proto

@@ -305,6 +305,10 @@ enum EConsumerScalingSupport {
     FULL_SUPPORT = 3;
 }
 
+message TOffloadConfig {
+
+}
+
 message TPQTabletConfig {
     optional uint64 CacheSize = 1 [default = 104857600]; //100Mb, per tablet
     optional TPartitionConfig PartitionConfig = 2; //mandatory
@@ -410,6 +414,8 @@ message TPQTabletConfig {
     optional TPartitionStrategy PartitionStrategy = 35;
 
     repeated TPartition AllPartitions = 36; // filled by schemeshard
+
+    optional TOffloadConfig OffloadConfig = 38;
 }
 
 message THeartbeat {

ydb/core/tx/schemeshard/schemeshard__operation_alter_continuous_backup.cpp

@@ -11,6 +11,30 @@
 
 namespace NKikimr::NSchemeShard {
 
+void DoAlterPqPart(const TOperationId& opId, const TPath& topicPath, TTopicInfo::TPtr topic, TVector<ISubOperation::TPtr>& result)
+{
+    auto outTx = TransactionTemplate(topicPath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpAlterPersQueueGroup);
+    // outTx.SetFailOnExist(!acceptExisted);
+
+    outTx.SetAllowAccessToPrivatePaths(true);
+
+    auto& desc = *outTx.MutableAlterPersQueueGroup();
+    desc.SetPathId(topicPath.Base()->PathId.LocalPathId);
+
+    NKikimrPQ::TPQTabletConfig tabletConfig;
+    if (!topic->TabletConfig.empty()) {
+        bool parseOk = ParseFromStringNoSizeLimit(tabletConfig, topic->TabletConfig);
+        Y_ABORT_UNLESS(parseOk, "Previously serialized pq tablet config cannot be parsed");
+    }
+
+    auto& pqConfig = *desc.MutablePQTabletConfig();
+    pqConfig.CopyFrom(tabletConfig);
+    pqConfig.ClearPartitionKeySchema();
+    pqConfig.MutableOffloadConfig();
+
+    result.push_back(CreateAlterPQ(NextPartId(opId, result), outTx));
+}
+
 TVector<ISubOperation::TPtr> CreateAlterContinuousBackup(TOperationId opId, const TTxTransaction& tx, TOperationContext& context) {
     Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::EOperationType::ESchemeOpAlterContinuousBackup);
 
@@ -24,6 +48,8 @@ TVector<ISubOperation::TPtr> CreateAlterContinuousBackup(TOperationId opId, cons
     }
 
     const auto [tablePath, streamPath] = std::get<NCdc::TStreamPaths>(checksResult);
+    const auto topicPath = streamPath.Child("streamImpl");
+    TTopicInfo::TPtr topic = context.SS->Topics.at(topicPath.Base()->PathId);
 
     TString errStr;
     if (!context.SS->CheckApplyIf(tx, errStr)) {
@@ -40,6 +66,7 @@ TVector<ISubOperation::TPtr> CreateAlterContinuousBackup(TOperationId opId, cons
 
     switch (cbOp.GetActionCase()) {
     case NKikimrSchemeOp::TAlterContinuousBackup::kStop:
+    case NKikimrSchemeOp::TAlterContinuousBackup::kTakeIncrementalBackup:
         alterCdcStreamOp.MutableDisable();
         break;
     default:
@@ -51,6 +78,10 @@ TVector<ISubOperation::TPtr> CreateAlterContinuousBackup(TOperationId opId, cons
 
     NCdc::DoAlterStream(alterCdcStreamOp, opId, workingDirPath, tablePath, result);
 
+    if (cbOp.GetActionCase() == NKikimrSchemeOp::TAlterContinuousBackup::kTakeIncrementalBackup) {
+        DoAlterPqPart(opId, topicPath, topic, result);
+    }
+
     return result;
 }
 

ydb/core/tx/schemeshard/ut_continuous_backup/ut_continuous_backup.cpp

@@ -64,4 +64,48 @@ Y_UNIT_TEST_SUITE(TContinuousBackupTests) {
         TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl"), {NLs::PathNotExist});
         TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl/streamImpl"), {NLs::PathNotExist});
     }
+
+    Y_UNIT_TEST(TakeIncrementalBackup) {
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        ui64 txId = 100;
+
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "Table"
+            Columns { Name: "key" Type: "Uint64" }
+            Columns { Name: "value" Type: "Uint64" }
+            KeyColumnNames: ["key"]
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        TestCreateContinuousBackup(runtime, ++txId, "/MyRoot", R"(
+            TableName: "Table"
+            ContinuousBackupDescription {
+            }
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl"), {
+            NLs::PathExist,
+            NLs::StreamMode(NKikimrSchemeOp::ECdcStreamModeUpdate),
+            NLs::StreamFormat(NKikimrSchemeOp::ECdcStreamFormatProto),
+            NLs::StreamState(NKikimrSchemeOp::ECdcStreamStateReady),
+            NLs::StreamVirtualTimestamps(false),
+        });
+        TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl/streamImpl"), {
+            NLs::PathExist,
+            NLs::HasNotOffloadConfig,
+        });
+
+        TestAlterContinuousBackup(runtime, ++txId, "/MyRoot", R"(
+            TableName: "Table"
+            TakeIncrementalBackup {}
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl/streamImpl"), {
+            NLs::PathExist,
+            NLs::HasOffloadConfig,
+        });
+    }
 } // TCdcStreamWithInitialScanTests

ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp

@@ -1251,6 +1251,11 @@ TCheckFunc ServerlessComputeResourcesMode(NKikimrSubDomains::EServerlessComputeR
     };
 }
 
+void HasOffloadConfigBase(const NKikimrScheme::TEvDescribeSchemeResult& record, TInverseTag inverse) {
+    UNIT_ASSERT(inverse.Value xor record.GetPathDescription().GetPersQueueGroup()
+        .GetPQTabletConfig().HasOffloadConfig());
+}
+
 #undef DESCRIBE_ASSERT_EQUAL
 #undef DESCRIBE_ASSERT_GE
 #undef DESCRIBE_ASSERT

ydb/core/tx/schemeshard/ut_helpers/ls_checks.h

@@ -167,6 +167,14 @@ namespace NLs {
     TCheckFunc SharedHive(ui64 sharedHiveId);
     TCheckFunc ServerlessComputeResourcesMode(NKikimrSubDomains::EServerlessComputeResourcesMode serverlessComputeResourcesMode);
 
+    struct TInverseTag {
+        bool Value = false;
+    };
+
+    void HasOffloadConfigBase(const NKikimrScheme::TEvDescribeSchemeResult& record, TInverseTag inverse);
+    inline void HasOffloadConfig(const NKikimrScheme::TEvDescribeSchemeResult& record) { return HasOffloadConfigBase(record, {}); };
+    inline void HasNotOffloadConfig(const NKikimrScheme::TEvDescribeSchemeResult& record) { return HasOffloadConfigBase(record, {.Value = true}); };
+
     template<class TCheck>
     void PerformAllChecks(const NKikimrScheme::TEvDescribeSchemeResult& result, TCheck&& check) {
         check(result);