Add datashard interface for data cleanup operation (#12896)

Author: lex007in

ydb/core/protos/tx_datashard.proto

@@ -2265,3 +2265,18 @@ message TEvInMemoryStateResponse {
     // Will be set for incomplete replies
     optional bytes ContinuationToken = 3;
 }
+
+// Forces cleanup-related operations in datashard tablet:
+// compaction, GC, GC log and other temporary data erasure.
+// Once the operation is done, any deleted data not used initially
+// will be included in garbage collection requests for the blobstorage.
+// Note: deleted data may still be used for a while, e.g. for history for MVCC.
+message TEvForceDataCleanup {
+    optional uint64 DataCleanupGeneration = 1; // increasing sequence
+}
+
+// Returns DataCleanupGeneration of the last successfully completed TEvForceDataCleanup request.
+// Intermediate requests and corresponding TEvForceDataCleanupResult's may be skipped.
+message TEvForceDataCleanupResult {
+    optional uint64 DataCleanupGeneration = 1; // from corresponding request
+}

ydb/core/tx/datashard/datashard.h

@@ -346,6 +346,9 @@ namespace TEvDataShard {
         EvInMemoryStateRequest,
         EvInMemoryStateResponse,
 
+        EvForceDataCleanup,
+        EvForceDataCleanupResult,
+
         EvEnd
     };
 
@@ -1562,6 +1565,24 @@ namespace TEvDataShard {
         }
     };
 
+    struct TEvForceDataCleanup : TEventPB<TEvForceDataCleanup, NKikimrTxDataShard::TEvForceDataCleanup,
+                                          TEvDataShard::EvForceDataCleanup> {
+        TEvForceDataCleanup() = default;
+
+        TEvForceDataCleanup(ui64 dataCleanupGeneration) {
+            Record.SetDataCleanupGeneration(dataCleanupGeneration);
+        }
+    };
+
+    struct TEvForceDataCleanupResult : TEventPB<TEvForceDataCleanupResult, NKikimrTxDataShard::TEvForceDataCleanupResult,
+                                                TEvDataShard::EvForceDataCleanupResult> {
+        TEvForceDataCleanupResult() = default;
+
+        TEvForceDataCleanupResult(ui64 dataCleanupGeneration) {
+            Record.SetDataCleanupGeneration(dataCleanupGeneration);
+        }
+    };
+
     /**
      * This message is used to ask datashard to compact any borrowed parts it has
      * for the specified user table.

ydb/core/tx/datashard/datashard__data_cleanup.cpp

@@ -0,0 +1,12 @@
+#include "datashard_impl.h"
+
+namespace NKikimr::NDataShard {
+
+void TDataShard::Handle(TEvDataShard::TEvForceDataCleanup::TPtr& ev, const TActorContext& ctx) {
+    // TODO: implement
+    const auto& record = ev->Get()->Record;
+    auto result = MakeHolder<TEvDataShard::TEvForceDataCleanupResult>(record.GetDataCleanupGeneration());
+    ctx.Send(ev->Sender, std::move(result));
+}
+
+} // namespace NKikimr::NDataShard

ydb/core/tx/datashard/datashard_impl.h

@@ -1389,6 +1389,8 @@ class TDataShard
 
     void Handle(TEvIncrementalRestoreScan::TEvFinished::TPtr& ev, const TActorContext& ctx);
 
+    void Handle(TEvDataShard::TEvForceDataCleanup::TPtr& ev, const TActorContext& ctx);
+
     void HandleByReplicationSourceOffsetsServer(STATEFN_SIG);
 
     void DoPeriodicTasks(const TActorContext &ctx);
@@ -3211,6 +3213,7 @@ class TDataShard
             HFunc(TEvPrivate::TEvStatisticsScanFinished, Handle);
             HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
             HFunc(TEvIncrementalRestoreScan::TEvFinished, Handle);
+            HFunc(TEvDataShard::TEvForceDataCleanup, Handle);
             default:
                 if (!HandleDefaultEvents(ev, SelfId())) {
                     ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateWork unhandled event type: " << ev->GetTypeRewrite() << " event: " << ev->ToString());

ydb/core/tx/datashard/datashard_ut_data_cleanup.cpp

@@ -0,0 +1,32 @@
+#include <ydb/core/tx/datashard/ut_common/datashard_ut_common.h>
+
+namespace NKikimr {
+
+using namespace Tests;
+
+Y_UNIT_TEST_SUITE(DataCleanup) {
+    Y_UNIT_TEST(ForceDataCleanup) {
+        TPortManager pm;
+        TServerSettings serverSettings(pm.GetPort(2134));
+        serverSettings.SetDomainName("Root")
+            .SetUseRealThreads(false);
+
+        Tests::TServer::TPtr server = new TServer(serverSettings);
+        auto &runtime = *server->GetRuntime();
+        auto sender = runtime.AllocateEdgeActor();
+
+        InitRoot(server, sender);
+
+        auto [shards, tableId] = CreateShardedTable(server, sender, "/Root", "table-1", 1);
+
+        ui64 expectedDataCleanupGeneration = 42;
+        auto request = MakeHolder<TEvDataShard::TEvForceDataCleanup>(expectedDataCleanupGeneration);
+
+        runtime.SendToPipe(shards.at(0), sender, request.Release(), 0, GetPipeConfigWithRetries());
+
+        auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvForceDataCleanupResult>(sender);
+        UNIT_ASSERT_VALUES_EQUAL(ev->Get()->Record.GetDataCleanupGeneration(), expectedDataCleanupGeneration);
+    }
+}
+
+} // namespace NKikimr

ydb/core/tx/datashard/ut_data_cleanup/ya.make

@@ -0,0 +1,25 @@
+UNITTEST_FOR(ydb/core/tx/datashard)
+
+FORK_SUBTESTS()
+
+SPLIT_FACTOR(1)
+
+IF (SANITIZER_TYPE == "thread" OR WITH_VALGRIND)
+    SIZE(LARGE)
+    TAG(ya:fat)
+ELSE()
+    SIZE(MEDIUM)
+ENDIF()
+
+PEERDIR(
+    ydb/core/testlib/default
+    ydb/core/tx/datashard/ut_common
+)
+
+YQL_LAST_ABI_VERSION()
+
+SRCS(
+    datashard_ut_data_cleanup.cpp
+)
+
+END()

ydb/core/tx/datashard/ya.make

@@ -55,6 +55,7 @@ SRCS(
     datashard__compact_borrowed.cpp
     datashard__compaction.cpp
     datashard__conditional_erase_rows.cpp
+    datashard__data_cleanup.cpp
     datashard__engine_host.cpp
     datashard__engine_host.h
     datashard__get_state_tx.cpp
@@ -309,6 +310,7 @@ RECURSE_FOR_TESTS(
     ut_change_exchange
     ut_column_stats
     ut_compaction
+    ut_data_cleanup
     ut_erase_rows
     ut_external_blobs
     ut_followers