[backups] add incremetnal backup to offload config (#5236)

Author: Enjection

ydb/core/protos/pqconfig.proto

@@ -5,13 +5,13 @@ import "ydb/public/api/protos/annotations/sensitive.proto";
 
 import "ydb/core/protos/base.proto";
 import "ydb/core/protos/msgbus_kv.proto";
-import "ydb/core/protos/node_limits.proto";
 import "ydb/core/protos/netclassifier.proto";
-import "ydb/library/services/services.proto";
+import "ydb/core/protos/node_limits.proto";
+import "ydb/core/scheme/protos/pathid.proto";
 import "ydb/core/scheme/protos/type_info.proto";
 
 import "ydb/library/actors/protos/actors.proto";
-
+import "ydb/library/services/services.proto";
 
 package NKikimrPQ;
 option java_package = "ru.yandex.kikimr.proto";
@@ -306,7 +306,14 @@ enum EConsumerScalingSupport {
 }
 
 message TOffloadConfig {
+    message TIncrementalBackup {
+        optional string DstPath = 1;
+        optional NKikimrProto.TPathID DstPathId = 2;
+    }
 
+    oneof Strategy {
+        TIncrementalBackup IncrementalBackup = 1;
+    }
 }
 
 message TPQTabletConfig {

ydb/core/tx/schemeshard/schemeshard__operation_alter_continuous_backup.cpp

@@ -11,7 +11,7 @@
 
 namespace NKikimr::NSchemeShard {
 
-void DoAlterPqPart(const TOperationId& opId, const TPath& topicPath, TTopicInfo::TPtr topic, TVector<ISubOperation::TPtr>& result)
+void DoAlterPqPart(const TOperationId& opId, const TPath& tablePath, const TPath& topicPath, TTopicInfo::TPtr topic, TVector<ISubOperation::TPtr>& result)
 {
     auto outTx = TransactionTemplate(topicPath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpAlterPersQueueGroup);
     // outTx.SetFailOnExist(!acceptExisted);
@@ -30,7 +30,8 @@ void DoAlterPqPart(const TOperationId& opId, const TPath& topicPath, TTopicInfo:
     auto& pqConfig = *desc.MutablePQTabletConfig();
     pqConfig.CopyFrom(tabletConfig);
     pqConfig.ClearPartitionKeySchema();
-    pqConfig.MutableOffloadConfig();
+    auto& ib = *pqConfig.MutableOffloadConfig()->MutableIncrementalBackup();
+    ib.SetDstPath(tablePath.PathString());
 
     result.push_back(CreateAlterPQ(NextPartId(opId, result), outTx));
 }
@@ -45,7 +46,14 @@ void DoCreateIncBackupTable(const TOperationId& opId, const TPath& dst, NKikimrS
     desc.CopyFrom(tableDesc);
     desc.SetName(dst.LeafName());
 
-    auto col = desc.AddColumns();
+    auto& replicationConfig = *desc.MutableReplicationConfig();
+    replicationConfig.SetMode(NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY);
+    replicationConfig.SetConsistency(NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK);
+
+    // TODO: remove NotNull from all columns for correct deletion writing
+    // TODO: cleanup all sequences
+
+    auto* col = desc.AddColumns();
     col->SetName("__incrBackupImpl_deleted");
     col->SetType("Bool");
 
@@ -107,7 +115,7 @@ TVector<ISubOperation::TPtr> CreateAlterContinuousBackup(TOperationId opId, cons
 
     if (cbOp.GetActionCase() == NKikimrSchemeOp::TAlterContinuousBackup::kTakeIncrementalBackup) {
         DoCreateIncBackupTable(opId, backupTablePath, schema, result);
-        DoAlterPqPart(opId, topicPath, topic, result);
+        DoAlterPqPart(opId, backupTablePath, topicPath, topic, result);
     }
 
     return result;

ydb/core/tx/schemeshard/schemeshard__operation_alter_pq.cpp

@@ -184,6 +184,13 @@ class TAlterPQ: public TSubOperation {
             const TString databasePath = TPath::Init(context.SS->RootPathId(), context.SS).PathString();
             alterConfig.SetYdbDatabasePath(databasePath);
 
+
+            if (alterConfig.HasOffloadConfig()) {
+                // TODO: check validity
+                auto* pathId = alterConfig.MutableOffloadConfig()->MutableIncrementalBackup()->MutableDstPathId();
+                PathIdFromPathId(TPath::Resolve(alterConfig.GetOffloadConfig().GetIncrementalBackup().GetDstPath(), context.SS).Base()->PathId, pathId);
+            }
+
             alterConfig.MutablePartitionKeySchema()->Swap(tabletConfig->MutablePartitionKeySchema());
             Y_PROTOBUF_SUPPRESS_NODISCARD alterConfig.SerializeToString(&params->TabletConfig);
             alterConfig.Swap(tabletConfig);

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

@@ -103,9 +103,21 @@ Y_UNIT_TEST_SUITE(TContinuousBackupTests) {
         )");
         env.TestWaitNotification(runtime, txId);
 
+        auto pathInfo = DescribePrivatePath(runtime, "/MyRoot/Table/incBackupImpl");
+        auto ownerId = pathInfo.GetPathOwnerId();
+        auto localId = pathInfo.GetPathId();
+
         TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/continuousBackupImpl/streamImpl"), {
             NLs::PathExist,
-            NLs::HasOffloadConfig,
+            NLs::HasOffloadConfig(Sprintf(R"(
+                    IncrementalBackup: {
+                        DstPath: "/MyRoot/Table/incBackupImpl"
+                        DstPathId: {
+                            OwnerId: %)" PRIu64 R"(
+                            LocalId: %)" PRIu64 R"(
+                        }
+                    }
+                )", ownerId, localId)),
         });
 
         TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table/incBackupImpl"), {

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

@@ -13,6 +13,10 @@
 
 #include <library/cpp/testing/unittest/registar.h>
 
+#include <contrib/libs/protobuf/src/google/protobuf/text_format.h>
+#include <contrib/libs/protobuf/src/google/protobuf/util/field_comparator.h>
+#include <contrib/libs/protobuf/src/google/protobuf/util/message_differencer.h>
+
 namespace NSchemeShardUT_Private {
 namespace NLs {
 
@@ -1251,9 +1255,22 @@ TCheckFunc ServerlessComputeResourcesMode(NKikimrSubDomains::EServerlessComputeR
     };
 }
 
-void HasOffloadConfigBase(const NKikimrScheme::TEvDescribeSchemeResult& record, TInverseTag inverse) {
+void HasOffloadConfigBase(const NKikimrScheme::TEvDescribeSchemeResult& record, const TString* const config, TInverseTag inverse) {
     UNIT_ASSERT(inverse.Value xor record.GetPathDescription().GetPersQueueGroup()
         .GetPQTabletConfig().HasOffloadConfig());
+
+    if (config) {
+        NKikimrPQ::TOffloadConfig expectedConfig;
+
+        bool parseResult = google::protobuf::TextFormat::ParseFromString(*config, &expectedConfig);
+        UNIT_ASSERT(parseResult);
+
+        google::protobuf::util::MessageDifferencer md;
+        auto fieldComparator = google::protobuf::util::DefaultFieldComparator();
+        md.set_field_comparator(&fieldComparator);
+        auto& givenConfig = record.GetPathDescription().GetPersQueueGroup().GetPQTabletConfig().GetOffloadConfig();
+        UNIT_ASSERT(md.Compare(expectedConfig, givenConfig));
+    }
 }
 
 #undef DESCRIBE_ASSERT_EQUAL

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

@@ -171,9 +171,13 @@ namespace NLs {
         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}); };
+    void HasOffloadConfigBase(const NKikimrScheme::TEvDescribeSchemeResult& record, const TString* const config, TInverseTag inverse);
+    inline TCheckFunc HasOffloadConfig(TString config) {
+        return [config] (const NKikimrScheme::TEvDescribeSchemeResult& record) {
+            return HasOffloadConfigBase(record, config ? &config : nullptr, {});
+        };
+    }
+    inline void HasNotOffloadConfig(const NKikimrScheme::TEvDescribeSchemeResult& record) { return HasOffloadConfigBase(record, nullptr, {.Value = true}); }
 
     template<class TCheck>
     void PerformAllChecks(const NKikimrScheme::TEvDescribeSchemeResult& result, TCheck&& check) {