Import changefeed's configuration from s3

ydb/core/protos/feature_flags.proto

@@ -193,4 +193,5 @@ message TFeatureFlags {
  // deny non-administrators the privilege of administering local users and groups
  optional bool EnableStrictUserManagement = 168 [default = false];
  optional bool EnableDatabaseAdmin = 169 [default = false];
+ optional bool EnableChangefeedsImport = 170 [default = false];
 }

ydb/core/protos/flat_scheme_op.proto

@@ -23,6 +23,7 @@ import "ydb/library/mkql_proto/protos/minikql.proto";
 import "ydb/public/api/protos/ydb_coordination.proto";
 import "ydb/public/api/protos/ydb_export.proto";
 import "ydb/public/api/protos/ydb_table.proto";
+import "ydb/public/api/protos/ydb_topic.proto";
 import "ydb/public/api/protos/ydb_value.proto";
 
 import "google/protobuf/empty.proto";
@@ -2124,3 +2125,11 @@ message TBackupBackupCollection {
  optional NKikimrProto.TPathID PathId = 2;
  optional string TargetDir = 3; // must be set on Rewrite
 }
+
+message TImportTableChangefeeds {
+ message TImportChangefeedTopic {
+ optional Ydb.Table.ChangefeedDescription Changefeed = 1;
+ optional Ydb.Topic.DescribeTopicResult Topic = 2;
+ }
+ repeated TImportChangefeedTopic Changefeeds = 1;
+}

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -4464,9 +4464,14 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  item.Metadata = NBackup::TMetadata::Deserialize(rowset.GetValue<Schema::ImportItems::Metadata>());
  }
 
+ if (rowset.HaveValue<Schema::ImportItems::Changefeeds>()) {
+ item.Changefeeds = rowset.GetValue<Schema::ImportItems::Changefeeds>();
+ }
+
  item.State = static_cast<TImportInfo::EState>(rowset.GetValue<Schema::ImportItems::State>());
  item.WaitTxId = rowset.GetValueOrDefault<Schema::ImportItems::WaitTxId>(InvalidTxId);
  item.NextIndexIdx = rowset.GetValueOrDefault<Schema::ImportItems::NextIndexIdx>(0);
+ item.NextChangefeedIdx = rowset.GetValueOrDefault<Schema::ImportItems::NextChangefeedIdx>(0);
  item.Issue = rowset.GetValueOrDefault<Schema::ImportItems::Issue>(TString());
 
  if (item.WaitTxId != InvalidTxId) {

ydb/core/tx/schemeshard/schemeshard_import.cpp

@@ -73,6 +73,9 @@ void TSchemeShard::FromXxportInfo(NKikimrImport::TImport& import, const TImportI
  case TImportInfo::EState::BuildIndexes:
  import.SetProgress(Ydb::Import::ImportProgress::PROGRESS_BUILD_INDEXES);
  break;
+ case TImportInfo::EState::CreateChangefeed:
+ import.SetProgress(Ydb::Import::ImportProgress::PROGRESS_CREATE_CHANGEFEEDS);
+ break;
  case TImportInfo::EState::Done:
  import.SetProgress(Ydb::Import::ImportProgress::PROGRESS_DONE);
  break;
@@ -163,6 +166,7 @@ void TSchemeShard::PersistImportItemState(NIceDb::TNiceDb& db, const TImportInfo
  NIceDb::TUpdate<Schema::ImportItems::State>(static_cast<ui8>(item.State)),
  NIceDb::TUpdate<Schema::ImportItems::WaitTxId>(item.WaitTxId),
  NIceDb::TUpdate<Schema::ImportItems::NextIndexIdx>(item.NextIndexIdx),
+ NIceDb::TUpdate<Schema::ImportItems::NextChangefeedIdx>(item.NextChangefeedIdx),
  NIceDb::TUpdate<Schema::ImportItems::Issue>(item.Issue)
  );
 }
@@ -189,6 +193,10 @@ void TSchemeShard::PersistImportItemScheme(NIceDb::TNiceDb& db, const TImportInf
  db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
  NIceDb::TUpdate<Schema::ImportItems::Metadata>(item.Metadata.Serialize())
  );
+
+ db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
+ NIceDb::TUpdate<Schema::ImportItems::Changefeeds>(item.Changefeeds)
+ );
 }
 
 void TSchemeShard::PersistImportItemPreparedCreationQuery(NIceDb::TNiceDb& db, const TImportInfo::TPtr importInfo, ui32 itemIdx) {

ydb/core/tx/schemeshard/schemeshard_import__create.cpp

@@ -526,6 +526,21 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  return true;
  }
 
+ void CreateChangefeed(TImportInfo::TPtr importInfo, ui32 itemIdx, TTxId txId) {
+ Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
+ auto& item = importInfo->Items.at(itemIdx);
+ item.SubState = ESubState::Proposed;
+
+ LOG_I("TImport::TTxProgress: CreateChangefeed propose"
+ << ": info# " << importInfo->ToString()
+ << ", item# " << item.ToString(itemIdx)
+ << ", txId# " << txId);
+
+ Y_ABORT_UNLESS(item.WaitTxId == InvalidTxId);
+
+ Send(Self->SelfId(), CreateChangefeedPropose(Self, txId, item));
+ }
+
  void AllocateTxId(TImportInfo::TPtr importInfo, ui32 itemIdx) {
  Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
  auto& item = importInfo->Items.at(itemIdx);
@@ -588,6 +603,25 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  return TTxId(ui64((*infoPtr)->Id));
  }
 
+ TTxId GetActiveCreateChangefeedTxId(TImportInfo::TPtr importInfo, ui32 itemIdx) {
+ Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
+ const auto& item = importInfo->Items.at(itemIdx);
+
+ Y_ABORT_UNLESS(item.State == EState::CreateChangefeed);
+ Y_ABORT_UNLESS(item.DstPathId);
+
+ if (!Self->PathsById.contains(item.DstPathId)) {
+ return InvalidTxId;
+ }
+
+ auto path = Self->PathsById.at(item.DstPathId);
+ if (path->PathState != NKikimrSchemeOp::EPathStateAlter) {
+ return InvalidTxId;
+ }
+
+ return path->LastTxId;
+ }
+
  static TString MakeIndexBuildUid(TImportInfo::TPtr importInfo, ui32 itemIdx) {
  Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
  const auto& item = importInfo->Items.at(itemIdx);
@@ -756,6 +790,7 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  case EState::CreateSchemeObject:
  case EState::Transferring:
  case EState::BuildIndexes:
+ case EState::CreateChangefeed:
  if (item.WaitTxId == InvalidTxId) {
  if (!IsCreatedByQuery(item) || item.PreparedCreationQuery) {
  AllocateTxId(importInfo, itemIdx);
@@ -781,6 +816,10 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  TTxId txId = InvalidTxId;
 
  switch (item.State) {
+ case EState::CreateChangefeed:
+ txId = GetActiveCreateChangefeedTxId(importInfo, itemIdx);
+ break;
+
  case EState::Transferring:
  if (!CancelTransferring(importInfo, itemIdx)) {
  txId = GetActiveRestoreTxId(importInfo, itemIdx);
@@ -1004,6 +1043,11 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  BuildIndex(importInfo, i, txId);
  itemIdx = i;
  break;
+
+ case EState::CreateChangefeed:
+ CreateChangefeed(importInfo, i, txId);
+ itemIdx = i;
+ break;
 
  default:
  break;
@@ -1064,6 +1108,8 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  txId = TTxId(record.GetPathCreateTxId());
  } else if (item.State == EState::Transferring) {
  txId = GetActiveRestoreTxId(importInfo, itemIdx);
+ } else if (item.State == EState::CreateChangefeed) {
+ txId = GetActiveCreateChangefeedTxId(importInfo, itemIdx);
  }
  }
 
@@ -1216,6 +1262,10 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  if (item.NextIndexIdx < item.Scheme.indexes_size()) {
  item.State = EState::BuildIndexes;
  AllocateTxId(importInfo, itemIdx);
+ } else if (item.NextChangefeedIdx < item.Changefeeds.changefeeds_size() &&
+ AppData()->FeatureFlags.GetEnableChangefeedsImport()) {
+ item.State = EState::CreateChangefeed;
+ AllocateTxId(importInfo, itemIdx);
  } else {
  item.State = EState::Done;
  }
@@ -1229,11 +1279,23 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
  } else {
  if (++item.NextIndexIdx < item.Scheme.indexes_size()) {
  AllocateTxId(importInfo, itemIdx);
+ } else if (item.NextChangefeedIdx < item.Changefeeds.changefeeds_size() &&
+ AppData()->FeatureFlags.GetEnableChangefeedsImport()) {
+ item.State = EState::CreateChangefeed;
+ AllocateTxId(importInfo, itemIdx);
  } else {
  item.State = EState::Done;
  }
  }
  break;
+
+ case EState::CreateChangefeed:
+ if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
+ AllocateTxId(importInfo, itemIdx);
+ } else {
+ item.State = EState::Done;
+ }
+ break;
 
  default:
  return SendNotificationsIfFinished(importInfo);

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.cpp

@@ -231,5 +231,65 @@ THolder<TEvIndexBuilder::TEvCancelRequest> CancelIndexBuildPropose(
  return MakeHolder<TEvIndexBuilder::TEvCancelRequest>(ui64(indexBuildId), domainPath.PathString(), ui64(indexBuildId));
 }
 
+THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
+ TSchemeShard* ss,
+ TTxId txId,
+ const TImportInfo::TItem& item
+) {
+ Y_ABORT_UNLESS(item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size());
+
+ const auto& importChangefeedTopic = item.Changefeeds.GetChangefeeds()[item.NextChangefeedIdx];
+ const auto& changefeed = importChangefeedTopic.GetChangefeed();
+ const auto& topic = importChangefeedTopic.GetTopic();
+
+ auto propose = MakeHolder<TEvSchemeShard::TEvModifySchemeTransaction>(ui64(txId), ss->TabletID());
+ auto& record = propose->Record;
+ auto& modifyScheme = *record.AddTransaction();
+ modifyScheme.SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpCreateCdcStream);
+ auto& cdcStream = *modifyScheme.MutableCreateCdcStream();
+
+ const TPath dstPath = TPath::Init(item.DstPathId, ss);
+ modifyScheme.SetWorkingDir(dstPath.Parent().PathString());
+ cdcStream.SetTableName(dstPath.LeafName());
+
+ TString error;
+ Ydb::StatusIds::StatusCode status;
+
+ auto& cdcStreamDescription = *cdcStream.MutableStreamDescription();
+ if (!FillChangefeedDescription(cdcStreamDescription, changefeed, status, error)) {
+ return nullptr;
+ }
+
+ if (topic.has_retention_period()) {
+ cdcStream.SetRetentionPeriodSeconds(topic.retention_period().seconds());
+ }
+
+ if (topic.has_partitioning_settings()) {
+ i64 minActivePartitions =
+ topic.partitioning_settings().min_active_partitions();
+ if (minActivePartitions < 0) {
+ return nullptr;
+ } else if (minActivePartitions == 0) {
+ minActivePartitions = 1;
+ }
+ cdcStream.SetTopicPartitions(minActivePartitions);
+
+ if (topic.partitioning_settings().has_auto_partitioning_settings()) {
+ auto& partitioningSettings = topic.partitioning_settings().auto_partitioning_settings();
+ cdcStream.SetTopicAutoPartitioning(partitioningSettings.strategy() != ::Ydb::Topic::AutoPartitioningStrategy::AUTO_PARTITIONING_STRATEGY_DISABLED);
+
+ i64 maxActivePartitions =
+ topic.partitioning_settings().max_active_partitions();
+ if (maxActivePartitions < 0) {
+ return nullptr;
+ } else if (maxActivePartitions == 0) {
+ maxActivePartitions = 50;
+ }
+ cdcStream.SetMaxPartitionCount(maxActivePartitions);
+ }
+ }
+ return propose;
+}
+
 } // NSchemeShard
 } // NKikimr

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.h

@@ -46,5 +46,11 @@ THolder<TEvIndexBuilder::TEvCancelRequest> CancelIndexBuildPropose(
  TTxId indexBuildId
 );
 
+THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
+ TSchemeShard* ss,
+ TTxId txId,
+ const TImportInfo::TItem& item
+);
+
 } // NSchemeShard
 } // NKikimr