Move ttl scans into a separate rb queue KIKIMR-14875

ref:561a800f3cf287375d6609a37dce1241eb386983

5 files changed, 61 insertions, 3 deletions

diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto
index ff170cd69a..e1b5a98d9e 100644
--- a/ydb/core/protos/config.proto
+++ b/ydb/core/protos/config.proto
@@ -1219,6 +1219,17 @@ message TImmediateControlsConfig {
             MinValue: 1000,
             MaxValue: 5000000,
             DefaultValue: 250000 }];
+
+        optional uint64 TtlReadAheadLo = 12 [(ControlOptions) = {
+            Description: "Override for ttl readahead (low watermark)",
+            MinValue: 0,
+            MaxValue: 67108864,
+            DefaultValue: 0 }];
+        optional uint64 TtlReadAheadHi = 13 [(ControlOptions) = {
+            Description: "Override for ttl readahead (high watermark)",
+            MinValue: 0,
+            MaxValue: 134217728,
+            DefaultValue: 0 }];
     }
 
     message TTxLimitControls {
@@ -1372,6 +1383,10 @@ message TDataShardConfig {
     optional string RestoreTaskName = 8 [default = "restore"];
     optional uint32 RestoreTaskPriority = 9 [default = 10];
     optional uint64 CleanupSnapshotPeriod = 10 [default = 15000]; // milliseconds
+    optional string TtlTaskName = 11 [default = "ttl"];
+    optional uint32 TtlTaskPriority = 12 [default = 10];
+    optional uint64 TtlReadAheadLo = 13 [default = 524288];
+    optional uint64 TtlReadAheadHi = 14 [default = 1048576];
 }
 
 message TCompactionConfig {
diff --git a/ydb/core/tablet/resource_broker.cpp b/ydb/core/tablet/resource_broker.cpp
index 3dd8ec02cc..7544bbe110 100644
--- a/ydb/core/tablet/resource_broker.cpp
+++ b/ydb/core/tablet/resource_broker.cpp
@@ -1340,6 +1340,11 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
     queue->SetWeight(100);
     queue->MutableLimit()->SetCpu(10);
 
+    queue = config.AddQueues();
+    queue->SetName("queue_ttl");
+    queue->SetWeight(100);
+    queue->MutableLimit()->SetCpu(2);
+
     auto task = config.AddTasks();
     task->SetName(NLocalDb::UnknownTaskName);
     task->SetQueueName(NLocalDb::DefaultQueueName);
@@ -1425,6 +1430,11 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
     task->SetQueueName("queue_build_index");
     task->SetDefaultDuration(TDuration::Minutes(10).GetValue());
 
+    task = config.AddTasks();
+    task->SetName("ttl");
+    task->SetQueueName("queue_ttl");
+    task->SetDefaultDuration(TDuration::Minutes(5).GetValue());
+
     config.MutableResourceLimit()->SetCpu(TotalCPU);
     config.MutableResourceLimit()->SetMemory(TotalMemory);
 
diff --git a/ydb/core/tx/datashard/datashard.cpp b/ydb/core/tx/datashard/datashard.cpp
index 0500e32d53..a2fe063ce6 100644
--- a/ydb/core/tx/datashard/datashard.cpp
+++ b/ydb/core/tx/datashard/datashard.cpp
@@ -140,6 +140,8 @@ TDataShard::TDataShard(const TActorId &tablet, TTabletStorageInfo *info)
     , ReadColumnsScanInUserPool(0, 0, 1)
     , BackupReadAheadLo(0, 0, 64*1024*1024)
     , BackupReadAheadHi(0, 0, 128*1024*1024)
+    , TtlReadAheadLo(0, 0, 64*1024*1024)
+    , TtlReadAheadHi(0, 0, 128*1024*1024)
     , EnablePrioritizedMvccSnapshotReads(1, 0, 1)
     , EnableUnprotectedMvccSnapshotReads(1, 0, 1)
     , EnableLeaderLeases(1, 0, 1)
@@ -307,6 +309,9 @@ void TDataShard::IcbRegister() {
         appData->Icb->RegisterSharedControl(BackupReadAheadLo, "DataShardControls.BackupReadAheadLo");
         appData->Icb->RegisterSharedControl(BackupReadAheadHi, "DataShardControls.BackupReadAheadHi");
 
+        appData->Icb->RegisterSharedControl(TtlReadAheadLo, "DataShardControls.TtlReadAheadLo");
+        appData->Icb->RegisterSharedControl(TtlReadAheadHi, "DataShardControls.TtlReadAheadHi");
+
         appData->Icb->RegisterSharedControl(EnablePrioritizedMvccSnapshotReads, "DataShardControls.PrioritizedMvccSnapshotReads");
         appData->Icb->RegisterSharedControl(EnableUnprotectedMvccSnapshotReads, "DataShardControls.UnprotectedMvccSnapshotReads");
 
diff --git a/ydb/core/tx/datashard/datashard__conditional_erase_rows.cpp b/ydb/core/tx/datashard/datashard__conditional_erase_rows.cpp
index 9f2ef822ee..d03c8ef2e6 100644
--- a/ydb/core/tx/datashard/datashard__conditional_erase_rows.cpp
+++ b/ydb/core/tx/datashard/datashard__conditional_erase_rows.cpp
@@ -578,9 +578,26 @@ void TDataShard::Handle(TEvDataShard::TEvConditionalEraseRowsRequest::TPtr& ev,
             const ui32 localTableId = userTable->LocalTid;
             Y_VERIFY(Executor()->Scheme().GetTableInfo(localTableId));
 
-            auto policy = Executor()->Scheme().GetTableInfo(localTableId)->CompactionPolicy;
-            auto opts = TScanOptions().SetResourceBroker(policy->BackupResourceBrokerTask, policy->DefaultTaskPriority + 2);
-            const ui64 scanId = QueueScan(localTableId, scan.Release(), localTxId, opts);
+            auto* appData = AppData(ctx);
+            const auto& taskName = appData->DataShardConfig.GetTtlTaskName();
+            const auto taskPrio = appData->DataShardConfig.GetTtlTaskPriority();
+
+            ui64 readAheadLo = appData->DataShardConfig.GetTtlReadAheadLo();
+            if (ui64 readAheadLoOverride = GetTtlReadAheadLoOverride(); readAheadLoOverride > 0) {
+                readAheadLo = readAheadLoOverride;
+            }
+
+            ui64 readAheadHi = appData->DataShardConfig.GetTtlReadAheadHi();
+            if (ui64 readAheadHiOverride = GetTtlReadAheadHiOverride(); readAheadHiOverride > 0) {
+                readAheadHi = readAheadHiOverride;
+            }
+
+            const ui64 scanId = QueueScan(localTableId, scan.Release(), localTxId,
+                TScanOptions()
+                    .SetResourceBroker(taskName, taskPrio)
+                    .SetReadAhead(readAheadLo, readAheadHi)
+                    .SetReadPrio(TScanOptions::EReadPrio::Low)
+            );
             InFlightCondErase.Enqueue(localTxId, scanId, condition);
         }
     } else {
diff --git a/ydb/core/tx/datashard/datashard_impl.h b/ydb/core/tx/datashard/datashard_impl.h
index fe83be478e..636ec29e77 100644
--- a/ydb/core/tx/datashard/datashard_impl.h
+++ b/ydb/core/tx/datashard/datashard_impl.h
@@ -1264,6 +1264,14 @@ public:
         return BackupReadAheadHi;
     }
 
+    ui64 GetTtlReadAheadLoOverride() const {
+        return TtlReadAheadLo;
+    }
+
+    ui64 GetTtlReadAheadHiOverride() const {
+        return TtlReadAheadHi;
+    }
+
     bool GetEnablePrioritizedMvccSnapshotReads() const {
         ui64 value = EnablePrioritizedMvccSnapshotReads;
         return value != 0;
@@ -2090,6 +2098,9 @@ private:
     TControlWrapper BackupReadAheadLo;
     TControlWrapper BackupReadAheadHi;
 
+    TControlWrapper TtlReadAheadLo;
+    TControlWrapper TtlReadAheadHi;
+
     TControlWrapper EnablePrioritizedMvccSnapshotReads;
     TControlWrapper EnableUnprotectedMvccSnapshotReads;