Merge pull request #17734 from stanislav-shchetinin/auto-dropping-25-1

stanislav-shchetinin · web-flow · commit d8c1e82d11a8 · 2025-05-05T19:49:30.000+03:00
25-1: Auto-remove temporary export tables
diff --git a/ydb/core/protos/feature_flags.proto b/ydb/core/protos/feature_flags.proto
@@ -199,4 +199,5 @@ message TFeatureFlags {
     optional bool EnableChangefeedsExport = 174 [default = false];
     optional bool SwitchToConfigV2 = 179 [default = false];
     optional bool SwitchToConfigV1 = 180 [default = false];
+    optional bool EnableExportAutoDropping = 183 [default = false];
 }
diff --git a/ydb/core/tx/schemeshard/schemeshard_export.cpp b/ydb/core/tx/schemeshard/schemeshard_export.cpp
@@ -100,7 +100,8 @@ void TSchemeShard::FromXxportInfo(NKikimrExport::TExport& exprt, const TExportIn
     case TExportInfo::EState::CopyTables:
         exprt.SetProgress(Ydb::Export::ExportProgress::PROGRESS_PREPARING);
         break;
-
+    
+    case TExportInfo::EState::AutoDropping:
     case TExportInfo::EState::Transferring:
     case TExportInfo::EState::Done:
         for (ui32 itemIdx : xrange(exportInfo->Items.size())) {
diff --git a/ydb/core/tx/schemeshard/schemeshard_export__create.cpp b/ydb/core/tx/schemeshard/schemeshard_export__create.cpp
@@ -432,6 +432,18 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
         Send(Self->TxAllocatorClient, new TEvTxAllocatorClient::TEvAllocate(), 0, exportInfo->Id);
     }
 
+    void PrepareAutoDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db) {
+        bool isContinued = false;
+        PrepareDropping(ss, exportInfo, db, TExportInfo::EState::AutoDropping, [&](ui64 itemIdx) {
+            exportInfo->PendingDropItems.push_back(itemIdx);
+            isContinued = true;
+            AllocateTxId(exportInfo, itemIdx);
+        });
+        if (!isContinued) {
+            AllocateTxId(exportInfo);
+        }
+    }
+
     void SubscribeTx(TTxId txId) {
         Send(Self->SelfId(), new TEvSchemeShard::TEvNotifyTxCompletion(ui64(txId)));
     }
@@ -617,7 +629,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
                 const auto& item = exportInfo->Items.at(itemIdx);
 
                 if (item.WaitTxId == InvalidTxId) {
-                    if (item.SourcePathType == NKikimrSchemeOp::EPathTypeTable) {
+                    if (item.SourcePathType == NKikimrSchemeOp::EPathTypeTable && item.State <= EState::Transferring) {
                         pendingTables.emplace_back(itemIdx);
                     } else {
                         UploadScheme(exportInfo, itemIdx, ctx);
@@ -665,6 +677,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
             SendNotificationsIfFinished(exportInfo);
             break;
 
+        case EState::AutoDropping:
         case EState::Dropping:
             if (!exportInfo->AllItemsAreDropped()) {
                 for (ui32 itemIdx : xrange(exportInfo->Items.size())) {
@@ -749,6 +762,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
             }
             break;
 
+        case EState::AutoDropping:
         case EState::Dropping:
             if (exportInfo->PendingDropItems) {
                 itemIdx = PopFront(exportInfo->PendingDropItems);
@@ -821,6 +835,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
                 }
                 break;
 
+            case EState::AutoDropping:
             case EState::Dropping:
                 if (isMultipleMods || isNotExist) {
                     if (record.GetPathDropTxId()) {
@@ -925,6 +940,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
             Self->PersistExportItemState(db, exportInfo, itemIdx);
             break;
 
+        case EState::AutoDropping:
         case EState::Dropping:
             if (!exportInfo->AllItemsAreDropped()) {
                 Y_ABORT_UNLESS(itemIdx < exportInfo->Items.size());
@@ -1011,7 +1027,11 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
             Self->PersistExportItemState(db, exportInfo, itemIdx);
 
             if (AllOf(exportInfo->Items, &TExportInfo::TItem::IsDone)) {
-                EndExport(exportInfo, EState::Done, db);
+                if (!AppData()->FeatureFlags.GetEnableExportAutoDropping()) {
+                    EndExport(exportInfo, EState::Done, db);
+                } else {
+                    PrepareAutoDropping(Self, exportInfo, db);
+                }
             }
         } else if (exportInfo->State == EState::Cancellation) {
             item.State = EState::Cancelled;
@@ -1140,14 +1160,19 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
                 }
             }
             if (!itemHasIssues && AllOf(exportInfo->Items, &TExportInfo::TItem::IsDone)) {
-                exportInfo->State = EState::Done;
-                exportInfo->EndTime = TAppData::TimeProvider->Now();
+                if (!AppData()->FeatureFlags.GetEnableExportAutoDropping()) {
+                    exportInfo->State = EState::Done;
+                    exportInfo->EndTime = TAppData::TimeProvider->Now();
+                } else {
+                    PrepareAutoDropping(Self, exportInfo, db);
+                }
             }
 
             Self->PersistExportItemState(db, exportInfo, itemIdx);
             break;
         }
 
+        case EState::AutoDropping:
         case EState::Dropping:
             if (!exportInfo->AllItemsAreDropped()) {
                 Y_ABORT_UNLESS(itemIdx < exportInfo->Items.size());
@@ -1163,6 +1188,10 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
             } else {
                 SendNotificationsIfFinished(exportInfo, true); // for tests
 
+                if (exportInfo->State == EState::AutoDropping) {
+                    return EndExport(exportInfo, EState::Done, db);
+                }
+
                 if (exportInfo->Uid) {
                     Self->ExportsByUid.erase(exportInfo->Uid);
                 }
diff --git a/ydb/core/tx/schemeshard/schemeshard_export__forget.cpp b/ydb/core/tx/schemeshard/schemeshard_export__forget.cpp
@@ -77,23 +77,8 @@ struct TSchemeShard::TExport::TTxForget: public TSchemeShard::TXxport::TTxBase {
             LOG_D("TExport::TTxForget, dropping export tables"
                 << ", info: " << exportInfo->ToString()
             );
-            exportInfo->WaitTxId = InvalidTxId;
-            exportInfo->State = TExportInfo::EState::Dropping;
-            Self->PersistExportState(db, exportInfo);
-
-            for (ui32 itemIdx : xrange(exportInfo->Items.size())) {
-                auto& item = exportInfo->Items.at(itemIdx);
-
-                item.WaitTxId = InvalidTxId;
-                item.State = TExportInfo::EState::Dropped;
-
-                const TPath itemPath = TPath::Resolve(ExportItemPathName(Self, exportInfo, itemIdx), Self);
-                if (itemPath.IsResolved() && !itemPath.IsDeleted()) {
-                    item.State = TExportInfo::EState::Dropping;
-                }
-
-                Self->PersistExportItemState(db, exportInfo, itemIdx);
-            }
+            
+            PrepareDropping(Self, exportInfo, db);
 
             Progress = true;
         }
diff --git a/ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.cpp b/ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.cpp
@@ -319,5 +319,39 @@ TString ExportItemPathName(const TString& exportPathName, ui32 itemIdx) {
     return TStringBuilder() << exportPathName << "/" << itemIdx;
 }
 
+void PrepareDropping(
+        TSchemeShard* ss,
+        TExportInfo::TPtr exportInfo,
+        NIceDb::TNiceDb& db,
+        TExportInfo::EState droppingState,
+        std::function<void(ui64)> func)
+{
+    Y_ABORT_UNLESS(IsIn({TExportInfo::EState::AutoDropping, TExportInfo::EState::Dropping}, droppingState));
+
+    exportInfo->WaitTxId = InvalidTxId;
+    exportInfo->State = droppingState;
+    ss->PersistExportState(db, exportInfo);
+
+    for (ui32 itemIdx : xrange(exportInfo->Items.size())) {
+        auto& item = exportInfo->Items.at(itemIdx);
+
+        item.WaitTxId = InvalidTxId;
+        item.State = TExportInfo::EState::Dropped;
+        const TPath itemPath = TPath::Resolve(ExportItemPathName(ss, exportInfo, itemIdx), ss);
+        if (itemPath.IsResolved() && !itemPath.IsDeleted()) {
+            item.State = TExportInfo::EState::Dropping;
+            if (exportInfo->State == TExportInfo::EState::AutoDropping) {
+                func(itemIdx);
+            }
+        }
+
+        ss->PersistExportItemState(db, exportInfo, itemIdx);
+    }
+}
+
+void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db) {
+    PrepareDropping(ss, exportInfo, db, TExportInfo::EState::Dropping, [](ui64){});
+}
+
 } // NSchemeShard
 } // NKikimr
diff --git a/ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.h b/ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.h
@@ -47,5 +47,9 @@ THolder<TEvSchemeShard::TEvCancelTx> CancelPropose(
 TString ExportItemPathName(TSchemeShard* ss, const TExportInfo::TPtr exportInfo, ui32 itemIdx);
 TString ExportItemPathName(const TString& exportPathName, ui32 itemIdx);
 
+void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db,
+    TExportInfo::EState droppingState, std::function<void(ui64)> func);
+void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db); 
+
 } // NSchemeShard
 } // NKikimr
diff --git a/ydb/core/tx/schemeshard/schemeshard_info_types.h b/ydb/core/tx/schemeshard/schemeshard_info_types.h
@@ -2668,6 +2668,7 @@ struct TExportInfo: public TSimpleRefCount<TExportInfo> {
         Done = 240,
         Dropping = 241,
         Dropped = 242,
+        AutoDropping = 243,
         Cancellation = 250,
         Cancelled = 251,
     };
@@ -2782,12 +2783,16 @@ struct TExportInfo: public TSimpleRefCount<TExportInfo> {
         return State == EState::Dropping;
     }
 
+    bool IsAutoDropping() const {
+        return State == EState::AutoDropping;
+    }
+
     bool IsCancelling() const {
         return State == EState::Cancellation;
     }
 
     bool IsInProgress() const {
-        return IsPreparing() || IsWorking() || IsDropping() || IsCancelling();
+        return IsPreparing() || IsWorking() || IsDropping() || IsAutoDropping() || IsCancelling();
     }
 
     bool IsDone() const {
diff --git a/ydb/core/tx/schemeshard/ut_export/ut_export.cpp b/ydb/core/tx/schemeshard/ut_export/ut_export.cpp
diff --git a/ydb/core/tx/schemeshard/ut_export_reboots_s3/ut_export_reboots_s3.cpp b/ydb/core/tx/schemeshard/ut_export_reboots_s3/ut_export_reboots_s3.cpp
diff --git a/ydb/tests/functional/compatibility/test_compatibility.py b/ydb/tests/functional/compatibility/test_compatibility.py

Original file line number	Diff line number	Diff line change
`@@ -199,4 +199,5 @@ message TFeatureFlags {`
`199`	`199`	`optional bool EnableChangefeedsExport = 174 [default = false];`
`200`	`200`	`optional bool SwitchToConfigV2 = 179 [default = false];`
`201`	`201`	`optional bool SwitchToConfigV1 = 180 [default = false];`
	`202`	`+ optional bool EnableExportAutoDropping = 183 [default = false];`
`202`	`203`	`}`