fix review

stanislav-shchetinin · Feb 21, 2025 · 5a159f3 · 5a159f3
1 parent ffd888a
commit 5a159f3
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 22 deletions.
diff --git a/ydb/core/tx/schemeshard/schemeshard_import__create.cpp b/ydb/core/tx/schemeshard/schemeshard_import__create.cpp
@@ -642,7 +642,7 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
         const auto& item = importInfo->Items.at(itemIdx);
 
         Y_ABORT_UNLESS(item.State == EState::CreateChangefeed);
-        Y_ABORT_UNLESS(item.DstPathId);
+        Y_ABORT_UNLESS(item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateConsumers);
         Y_ABORT_UNLESS(item.StreamImplPathId);
 
         if (!Self->PathsById.contains(item.StreamImplPathId)) {
@@ -851,14 +851,6 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
                 TTxId txId = InvalidTxId;
 
                 switch (item.State) {
-                case EState::CreateChangefeed:
-                    if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
-                        txId = GetActiveCreateChangefeedTxId(importInfo, itemIdx);
-                    } else {
-                        txId = GetActiveCreateConsumerTxId(importInfo, itemIdx);
-                    }
-                    break;
-
                 case EState::Transferring:
                     if (!CancelTransferring(importInfo, itemIdx)) {
                         txId = GetActiveRestoreTxId(importInfo, itemIdx);
@@ -1161,20 +1153,21 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
                 }
             }
 
-            if (record.GetStatus() == NKikimrScheme::StatusAlreadyExists && item.State == EState::CreateChangefeed) {
-                if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
-                    item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateConsumers;
-                    AllocateTxId(importInfo, itemIdx);
-                } else if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
-                    item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateChangefeed;
-                    AllocateTxId(importInfo, itemIdx);
-                } else {
-                    item.State = EState::Done;
+            if (txId == InvalidTxId) {
+
+                if (record.GetStatus() == NKikimrScheme::StatusAlreadyExists && item.State == EState::CreateChangefeed) {
+                    if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
+                        item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateConsumers;
+                        AllocateTxId(importInfo, itemIdx);
+                    } else if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
+                        item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateChangefeed;
+                        AllocateTxId(importInfo, itemIdx);
+                    } else {
+                        item.State = EState::Done;
+                    }
+                    return;
                 }
-                return;
-            }
 
-            if (txId == InvalidTxId) {
                 return CancelAndPersist(db, importInfo, itemIdx, record.GetReason(), "unhappy propose");
             }
 

diff --git a/ydb/core/tx/schemeshard/schemeshard_info_types.h b/ydb/core/tx/schemeshard/schemeshard_info_types.h
@@ -2869,7 +2869,7 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
         int NextChangefeedIdx = 0;
         TString Issue;
         int ViewCreationRetries = 0;
-        NKikimr::TPathId StreamImplPathId;
+        TPathId StreamImplPathId;
 
         TItem() = default;