Merge 6007823 into 1237379

Author: lex007in

ydb/core/tablet_flat/flat_executor.cpp

@@ -452,7 +452,7 @@ void TExecutor::Active(const TActorContext &ctx) {
 
     CompactionLogic = THolder<TCompactionLogic>(new TCompactionLogic(MemTableMemoryConsumersCollection.Get(), Logger.Get(), Broker.Get(), this, loadedState->Comp,
                                                                      Sprintf("tablet-%" PRIu64, Owner->TabletID())));
-    DataCleanupLogic = MakeHolder<TDataCleanupLogic>(static_cast<NActors::IActorOps*>(this), this, Owner, Logger.Get(), GcLogic.Get());
+    DataCleanupLogic = MakeHolder<TDataCleanupLogic>(static_cast<NActors::IActorOps*>(this), this, Owner, Logger.Get(), GcLogic.Get(), BorrowLogic.Get());
     LogicRedo->InstallCounters(Counters.Get(), nullptr);
 
     ResourceMetrics = MakeHolder<NMetrics::TResourceMetrics>(Owner->TabletID(), 0, Launcher);
@@ -3103,6 +3103,11 @@ void TExecutor::Handle(TEvTablet::TEvCommitResult::TPtr &ev, const TActorContext
                     CheckCollectionBarrier(x);
                 InFlySnapCollectionBarriers.erase(snapCollectionIt);
             }
+
+            DataCleanupLogic->OnLogCommited();
+            if (DataCleanupLogic->NeedLogSnaphot()) {
+                MakeLogSnapshot();
+            }
         }
         break;
     case ECommit::Snap:
@@ -3663,7 +3668,7 @@ void TExecutor::Handle(NOps::TEvResult *ops, TProdCompact *msg, bool cancelled)
 
     Y_ABORT_UNLESS(InFlyCompactionGcBarriers.emplace(commit->Step, ops->Barrier).second);
 
-    DataCleanupLogic->OnCompleteCompaction(tableId, CompactionLogic->GetFinishedCompactionInfo(tableId));
+    DataCleanupLogic->OnCompleteCompaction(tableId, Generation(), commit->Step, CompactionLogic->GetFinishedCompactionInfo(tableId));
 
     AttachLeaseCommit(commit.Get());
     CommitManager->Commit(commit);

ydb/core/tablet_flat/flat_executor_borrowlogic.cpp

@@ -602,4 +602,17 @@ bool TExecutorBorrowLogic::HasLoanedParts() const {
     return false;
 }
 
+bool TExecutorBorrowLogic::HasLoanedBlobsBefore(ui32 generation, ui32 step) const {
+    for (const auto &xpair : BorrowedInfo) {
+        if (xpair.second.BorrowInfo.FullBorrow) {
+            for (const auto& blobId : xpair.second.BorrowInfo.Keep) {
+                if (std::make_tuple(blobId.Generation(), blobId.Step()) < std::tie(generation, step)) {
+                    return true;
+                }
+            }
+        }
+    }
+    return false;
+}
+
 }}

ydb/core/tablet_flat/flat_executor_borrowlogic.h

@@ -171,6 +171,8 @@ class TExecutorBorrowLogic {
 
     // i.e. parts we own, but loaned to others
     bool HasLoanedParts() const;
+
+    bool HasLoanedBlobsBefore(ui32 generation, ui32 step) const;
 };
 
 }}

ydb/core/tablet_flat/flat_executor_data_cleanup_logic.cpp

@@ -2,12 +2,19 @@
 
 namespace NKikimr::NTabletFlatExecutor {
 
-TDataCleanupLogic::TDataCleanupLogic(IOps* ops, IExecutor* executor, ITablet* owner, NUtil::ILogger* logger, TExecutorGCLogic* gcLogic)
+TDataCleanupLogic::TDataCleanupLogic(
+    IOps* ops,
+    IExecutor* executor,
+    ITablet* owner,
+    NUtil::ILogger* logger,
+    TExecutorGCLogic* gcLogic,
+    TExecutorBorrowLogic* borrowLogic)
     : Ops(ops)
     , Executor(executor)
     , Owner(owner)
     , Logger(logger)
     , GcLogic(gcLogic)
+    , BorrowLogic(borrowLogic)
 {}
 
 bool TDataCleanupLogic::TryStartCleanup(ui64 dataCleanupGeneration, const TActorContext& ctx) {
@@ -66,6 +73,8 @@ void TDataCleanupLogic::WaitCompaction() {
 
 void TDataCleanupLogic::OnCompleteCompaction(
     ui32 tableId,
+    ui32 generation,
+    ui32 step,
     const TFinishedCompactionInfo& finishedCompactionInfo)
 {
     if (State != EDataCleanupState::WaitCompaction) {
@@ -75,10 +84,16 @@ void TDataCleanupLogic::OnCompleteCompaction(
     if (auto it = CompactingTables.find(tableId); it != CompactingTables.end()) {
         if (finishedCompactionInfo.Edge >= it->second.CompactionId) {
             CompactingTables.erase(it);
+            auto compactionTime = TGCTime(generation, step);
+            LastCompactionTime = Max(LastCompactionTime, compactionTime);
         }
     }
     if (CompactingTables.empty()) {
-        State = EDataCleanupState::PendingFirstSnapshot;
+        if (BorrowLogic->HasLoanedBlobsBefore(LastCompactionTime.Generation, LastCompactionTime.Step)) {
+            State = EDataCleanupState::WaitBorrowed;
+        } else {
+            State = EDataCleanupState::PendingFirstSnapshot;
+        }
     }
 }
 
@@ -110,6 +125,15 @@ void TDataCleanupLogic::OnMakeLogSnapshot(ui32 generation, ui32 step) {
     }
 }
 
+void TDataCleanupLogic::OnLogCommited() {
+    if (State != EDataCleanupState::WaitBorrowed) {
+        return;
+    }
+    if (!BorrowLogic->HasLoanedBlobsBefore(LastCompactionTime.Generation, LastCompactionTime.Step)) {
+        State = EDataCleanupState::PendingFirstSnapshot;
+    }
+}
+
 void TDataCleanupLogic::OnSnapshotCommited(ui32 generation, ui32 step) {
     switch (State) {
         case EDataCleanupState::WaitFirstSnapshot: {

ydb/core/tablet_flat/flat_executor_data_cleanup_logic.h

@@ -1,7 +1,8 @@
 #pragma once
 
-#include "tablet_flat_executor.h"
+#include "flat_executor_borrowlogic.h"
 #include "flat_executor_gclogic.h"
+#include "tablet_flat_executor.h"
 #include "util_fmt_logger.h"
 
 namespace NKikimr::NTabletFlatExecutor {
@@ -11,6 +12,7 @@ class TDataCleanupLogic {
         Idle,
         PendingCompaction,
         WaitCompaction,
+        WaitBorrowed,
         PendingFirstSnapshot,
         WaitFirstSnapshot,
         PendingSecondSnapshot,
@@ -31,14 +33,21 @@ class TDataCleanupLogic {
     using ITablet = NFlatExecutorSetup::ITablet;
     using ELnLev = NUtil::ELnLev;
 
-    TDataCleanupLogic(IOps* ops, IExecutor* executor, ITablet* owner, NUtil::ILogger* logger, TExecutorGCLogic* gcLogic);
+    TDataCleanupLogic(
+        IOps* ops,
+        IExecutor* executor,
+        ITablet* owner,
+        NUtil::ILogger* logger,
+        TExecutorGCLogic* gcLogic,
+        TExecutorBorrowLogic* borrowLogic);
 
     bool TryStartCleanup(ui64 dataCleanupGeneration, const TActorContext& ctx);
     void OnCompactionPrepared(ui32 tableId, ui64 compactionId);
     void WaitCompaction();
-    void OnCompleteCompaction(ui32 tableId, const TFinishedCompactionInfo& finishedCompactionInfo);
+    void OnCompleteCompaction(ui32 tableId, ui32 generation, ui32 step, const TFinishedCompactionInfo& finishedCompactionInfo);
     bool NeedLogSnaphot();
     void OnMakeLogSnapshot(ui32 generation, ui32 step);
+    void OnLogCommited();
     void OnSnapshotCommited(ui32 generation, ui32 step);
     void OnCollectedGarbage(const TActorContext& ctx);
     void OnGcForStepAckResponse(ui32 generation, ui32 step, const TActorContext& ctx);
@@ -53,12 +62,14 @@ class TDataCleanupLogic {
     ITablet* Owner;
     NUtil::ILogger* const Logger;
     TExecutorGCLogic* const GcLogic;
+    TExecutorBorrowLogic* const BorrowLogic;
 
     ui64 CurrentDataCleanupGeneration = 0;
     ui64 NextDataCleanupGeneration = 0;
     EDataCleanupState State = EDataCleanupState::Idle;
     THashMap<ui32, TCleanupTableInfo> CompactingTables; // tracks statuses of compaction
 
+    TGCTime LastCompactionTime;
     // two subsequent are snapshots required to force GC
     TGCTime FirstLogSnaphotStep;
     TGCTime SecondLogSnaphotStep;

ydb/core/tx/datashard/datashard_ut_data_cleanup.cpp

@@ -329,6 +329,46 @@ Y_UNIT_TEST_SUITE(DataCleanup) {
         CheckTableData(server, proxyDSs, "/Root/table-1");
         CheckTableData(server, proxyDSs, "/Root/table-2");
     }
+
+    Y_UNIT_TEST(BorrowerDataCleanedAfterCopyTable) {
+        auto [server, sender, tableShards, proxyDSs] = SetupWithTable(true);
+        auto& runtime = *server->GetRuntime();
+
+        auto txIdCopy = AsyncCreateCopyTable(server, sender, "/Root", "table-2", "/Root/table-1");
+        WaitTxNotification(server, sender, txIdCopy);
+        auto table2Shards = GetTableShards(server, sender, "/Root/table-2");
+        auto table2Id = ResolveTableId(server, sender, "/Root/table-2");
+
+        ExecSQL(server, sender, "DELETE FROM `/Root/table-1` WHERE key IN (1, 4);");
+        ExecSQL(server, sender, "DELETE FROM `/Root/table-2` WHERE key IN (1, 4);");
+        SimulateSleep(runtime, TDuration::Seconds(2));
+
+        ui64 dataCleanupGeneration = 24;
+        {
+            // cleanup for the first table should not be completed due to borrowed parts
+            auto request = MakeHolder<TEvDataShard::TEvForceDataCleanup>(dataCleanupGeneration);
+            runtime.SendToPipe(tableShards.at(0), sender, request.Release(), 0, GetPipeConfigWithRetries());
+
+            auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvForceDataCleanupResult>(sender, TDuration::Seconds(10));
+            UNIT_ASSERT(!ev);
+        }
+        {
+            // cleanup for the second table
+            auto request = MakeHolder<TEvDataShard::TEvForceDataCleanup>(dataCleanupGeneration);
+            runtime.SendToPipe(table2Shards.at(0), sender, request.Release(), 0, GetPipeConfigWithRetries());
+
+            auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvForceDataCleanupResult>(sender);
+            CheckResultEvent(*ev->Get(), table2Shards.at(0), dataCleanupGeneration);
+        }
+        {
+            // cleanup should be continued after compaction of the second table
+            auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvForceDataCleanupResult>(sender);
+            CheckResultEvent(*ev->Get(), tableShards.at(0), dataCleanupGeneration);
+        }
+
+        CheckTableData(server, proxyDSs, "/Root/table-1");
+        CheckTableData(server, proxyDSs, "/Root/table-2");
+    }
 }
 
 } // namespace NKikimr