Refactor getting SchedulableReadFactory by datashard (#41123)

24 files changed, 763 insertions, 285 deletions

diff --git a/ydb/core/base/appdata_fwd.h b/ydb/core/base/appdata_fwd.h
index 48314307f9d..790cde296dc 100644
--- a/ydb/core/base/appdata_fwd.h
+++ b/ydb/core/base/appdata_fwd.h
@@ -182,6 +182,11 @@ namespace NAudit {
     class TAuditConfig;
 }
 
+namespace NKqp::NScheduler {
+    class TComputeScheduler;
+    using TComputeSchedulerPtr = std::shared_ptr<TComputeScheduler>; // TODO: duplicates forward declaration
+}
+
 struct TAppData {
     static const ui32 MagicTag = 0x2991AAF8;
     const ui32 Magic;
@@ -336,6 +341,8 @@ struct TAppData {
     // Immutable snapshot registry for fast snapshot queries
     TIntrusivePtr<IImmutableSnapshotRegistryHolder> SnapshotRegistryHolder;
 
+    NKqp::NScheduler::TComputeSchedulerPtr KqpComputeScheduler;
+
     TAppData(
             ui32 sysPoolId, ui32 userPoolId, ui32 ioPoolId, ui32 batchPoolId,
             TMap<TString, ui32> servicePools,
diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
index 89d59f60919..09df8c88eb4 100644
--- a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
+++ b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -2411,24 +2411,6 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
                 NKqp::MakeKqpWarmupActorId(NodeId),
                 TActorSetupCmd(warmupActor, TMailboxType::HTSwap, appData->UserPoolId)));
         }
-
-        // Create Compute Scheduler service
-        {
-            NKqp::NScheduler::TOptions schedulerOptions {
-                .DelayParams = {
-                    .MaxDelay = TDuration::MicroSeconds(Config.GetTableServiceConfig().GetComputeSchedulerSettings().GetMaxTaskDelayUs()),
-                    .MinDelay = TDuration::MicroSeconds(Config.GetTableServiceConfig().GetComputeSchedulerSettings().GetMinTaskDelayUs()),
-                    .AttemptBonus = TDuration::MicroSeconds(Config.GetTableServiceConfig().GetComputeSchedulerSettings().GetAttemptTaskBonusUs()),
-                    .MaxRandomDelay = TDuration::MicroSeconds(Config.GetTableServiceConfig().GetComputeSchedulerSettings().GetMaxTaskRandomDelayUs()),
-                },
-                .UpdateFairSharePeriod = TDuration::MilliSeconds(Config.GetTableServiceConfig().GetComputeSchedulerSettings().GetUpdateFairShareMs()),
-            };
-            auto* computeSchedulerActor = NKqp::CreateKqpComputeSchedulerService(schedulerOptions);
-            setup->LocalServices.push_back(std::make_pair(
-                NKqp::MakeKqpSchedulerServiceId(NodeId),
-                TActorSetupCmd(computeSchedulerActor, TMailboxType::HTSwap, appData->UserPoolId)
-            ));
-        }
     }
 }
 
diff --git a/ydb/core/driver_lib/run/run.cpp b/ydb/core/driver_lib/run/run.cpp
index 6b6b6e1e367..803f564a9f6 100644
--- a/ydb/core/driver_lib/run/run.cpp
+++ b/ydb/core/driver_lib/run/run.cpp
@@ -6,6 +6,7 @@
 
 #include <ydb/core/kqp/compile_service/kqp_warmup_compile_actor.h>
 #include <ydb/core/kqp/common/simple/services.h>
+#include <ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h>
 #include <ydb/core/memory_controller/memory_controller.h>
 #include <ydb/library/actors/core/callstack.h>
 #include <ydb/library/actors/core/events.h>
@@ -1644,11 +1645,13 @@ void TKikimrRunner::InitializeAppData(const TKikimrRunConfig& runConfig)
 
     // setup resource profiles
     AppData->ResourceProfiles = new TResourceProfiles;
-    if (runConfig.AppConfig.GetBootstrapConfig().ResourceProfilesSize())
+    if (runConfig.AppConfig.GetBootstrapConfig().ResourceProfilesSize()) {
         AppData->ResourceProfiles->LoadProfiles(runConfig.AppConfig.GetBootstrapConfig().GetResourceProfiles());
+    }
 
-    if (runConfig.AppConfig.GetBootstrapConfig().HasEnableIntrospection())
+    if (runConfig.AppConfig.GetBootstrapConfig().HasEnableIntrospection()) {
         AppData->EnableIntrospection = runConfig.AppConfig.GetBootstrapConfig().GetEnableIntrospection();
+    }
 
     if (runConfig.AppConfig.HasClusterDiagnosticsConfig()) {
         AppData->ClusterDiagnosticsConfig.CopyFrom(runConfig.AppConfig.GetClusterDiagnosticsConfig());
@@ -1658,6 +1661,8 @@ void TKikimrRunner::InitializeAppData(const TKikimrRunConfig& runConfig)
         AppData->LongTxServiceConfig.CopyFrom(runConfig.AppConfig.GetLongTxServiceConfig());
     }
 
+    AppData->KqpComputeScheduler = NKqp::CreateKqpComputeScheduler(Counters, runConfig.AppConfig);
+
     TAppDataInitializersList appDataInitializers;
     // setup domain info
     appDataInitializers.AddAppDataInitializer(new TDomainsInitializer(runConfig));
diff --git a/ydb/core/kqp/node_service/kqp_node_service.cpp b/ydb/core/kqp/node_service/kqp_node_service.cpp
index c69cb11c2e6..3f9cc099c4b 100644
--- a/ydb/core/kqp/node_service/kqp_node_service.cpp
+++ b/ydb/core/kqp/node_service/kqp_node_service.cpp
@@ -315,7 +315,7 @@ private:
         ptr->StartRetryDelay = TDuration::MilliSeconds(settings.GetStartDelayMs());
         ptr->MaxShardAttempts = settings.GetMaxShardRetries();
         ptr->MaxShardResolves = settings.GetMaxShardResolves();
-        ptr->UnsertaintyRatio = settings.GetUnsertaintyRatio();
+        ptr->UncertaintyRatio = settings.GetUnsertaintyRatio();
         ptr->Multiplier = settings.GetMultiplier();
         if (settings.GetMaxTotalRetries()) {
             ptr->MaxTotalRetries = settings.GetMaxTotalRetries();
diff --git a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
index 8d369f117a6..b6ad409e14e 100644
--- a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
+++ b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
@@ -382,22 +382,10 @@ public:
         TActivationContext::ActorSystem()->RegisterLocalService(
             MakeKqpWorkloadServiceId(SelfId().NodeId()), KqpWorkloadService);
 
-        // A lot of tests create ProxyService but don't create KqpServiceInitializer
-        if (!TActivationContext::ActorSystem()->LookupLocalService(MakeKqpSchedulerServiceId(SelfId().NodeId()))) {
-            NScheduler::TOptions schedulerOptions {
-                .DelayParams = {
-                    .MaxDelay = TDuration::MicroSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetMaxTaskDelayUs()),
-                    .MinDelay = TDuration::MicroSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetMinTaskDelayUs()),
-                    .AttemptBonus = TDuration::MicroSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetAttemptTaskBonusUs()),
-                    .MaxRandomDelay = TDuration::MicroSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetMaxTaskRandomDelayUs()),
-                },
-                .UpdateFairSharePeriod = TDuration::MilliSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetUpdateFairShareMs()),
-            };
-
-            KqpComputeSchedulerService = TActivationContext::Register(CreateKqpComputeSchedulerService(schedulerOptions));
-            TActivationContext::ActorSystem()->RegisterLocalService(
-                MakeKqpSchedulerServiceId(SelfId().NodeId()), KqpComputeSchedulerService);
-        }
+        auto updateFairSharePeriod = TDuration::MilliSeconds(TableServiceConfig.GetComputeSchedulerSettings().GetUpdateFairShareMs());
+        KqpComputeSchedulerService = TActivationContext::Register(CreateKqpComputeSchedulerService(updateFairSharePeriod));
+        TActivationContext::ActorSystem()->RegisterLocalService(
+            NKqp::MakeKqpSchedulerServiceId(SelfId().NodeId()), KqpComputeSchedulerService);
 
         NActors::TMon* mon = AppData()->Mon;
         if (mon) {
@@ -516,9 +504,7 @@ public:
         Send(KqpNodeService, new TEvents::TEvPoison);
 
         Send(KqpWorkloadService, new TEvents::TEvPoison());
-        if (KqpComputeSchedulerService) {
-            Send(KqpComputeSchedulerService, new TEvents::TEvPoison());
-        }
+        Send(KqpComputeSchedulerService, new TEvents::TEvPoison());
         Send(KqpQueryTextCacheService, new TEvents::TEvPoison());
         if (RowDispatcherService) {
             Send(RowDispatcherService, new TEvents::TEvPoison());
diff --git a/ydb/core/kqp/runtime/kqp_buffer_lookup_actor.cpp b/ydb/core/kqp/runtime/kqp_buffer_lookup_actor.cpp
index a0833984b1e..69ae765f9e4 100644
--- a/ydb/core/kqp/runtime/kqp_buffer_lookup_actor.cpp
+++ b/ydb/core/kqp/runtime/kqp_buffer_lookup_actor.cpp
@@ -484,8 +484,10 @@ public:
             case Ydb::StatusIds::OVERLOADED: {
                 CA_LOG_D("OVERLOADED was received from tablet: " << shardId << "."
                     << getIssues().ToOneLineString());
-                const bool isThrottled = record.HasThrottled() && record.GetThrottled();
-                if (!RetryTableRead(record.GetReadId(), false, isThrottled)) {
+                const std::optional<TDuration> throttleDelay = record.HasThrottleDelayMs()
+                    ? std::make_optional(TDuration::MilliSeconds(record.GetThrottleDelayMs()))
+                    : std::nullopt;
+                if (!RetryTableRead(record.GetReadId(), false, throttleDelay)) {
                     return RuntimeError(
                         NYql::NDqProto::StatusIds::OVERLOADED,
                         NYql::TIssuesIds::KIKIMR_OVERLOADED,
@@ -622,19 +624,24 @@ public:
         }
     }
 
-    bool RetryTableRead(const ui64 failedReadId, bool allowInstantRetry, bool isThrottled = false) {
+    bool RetryTableRead(const ui64 failedReadId, bool allowInstantRetry, std::optional<TDuration> throttleDelay = std::nullopt) {
         auto& failedRead = ReadIdToState.at(failedReadId);
         auto& lookupState = CookieToLookupState.at(failedRead.LookupCookie);
         CA_LOG_D("Retry reading of table: " << lookupState.Worker->GetTablePath() << ", failedReadId: " << failedReadId
             << ", shardId: " << failedRead.ShardId);
         failedRead.Blocked = true;
 
-        if (!isThrottled && failedRead.RetryAttempts >= MaxShardRetries()) {
-            return false;
+        TDuration delay;
+        if (!throttleDelay) {
+            if (failedRead.RetryAttempts >= MaxShardRetries()) {
+                return false;
+            }
+            ++failedRead.RetryAttempts;
+            delay = CalcDelay(failedRead.RetryAttempts, allowInstantRetry);
+        } else {
+            delay = *throttleDelay;
         }
 
-        ++failedRead.RetryAttempts;
-        auto delay = CalcDelay(failedRead.RetryAttempts, allowInstantRetry);
         if (delay == TDuration::Zero()) {
             DoRetryTableRead(failedReadId, lookupState, failedRead);
         } else {
diff --git a/ydb/core/kqp/runtime/kqp_full_text_source.cpp b/ydb/core/kqp/runtime/kqp_full_text_source.cpp
index 9aaa0ce65e1..c0d05784bf5 100644
--- a/ydb/core/kqp/runtime/kqp_full_text_source.cpp
+++ b/ydb/core/kqp/runtime/kqp_full_text_source.cpp
@@ -3200,16 +3200,21 @@ public:
     }
 
     void ScheduleReadRetry(ui64 readId, ui64 shardId, bool allowInstantRetry, NYql::NDqProto::StatusIds::StatusCode errorStatus,
-        bool isThrottled = false)
+        std::optional<TDuration> throttleDelay = std::nullopt)
     {
         auto maxRetries = MaxShardRetries();
-        if (!isThrottled && ReadsState.CheckShardRetriesExeeded(shardId, maxRetries)) {
-            RuntimeError(TStringBuilder() << "Max retries (" << maxRetries << ") exceeded for read " << readId
-                << " on shard " << shardId, errorStatus);
-            return;
+        TDuration delay;
+        if (!throttleDelay) {
+            if (ReadsState.CheckShardRetriesExeeded(shardId, maxRetries)) {
+                RuntimeError(TStringBuilder() << "Max retries (" << maxRetries << ") exceeded for read " << readId
+                    << " on shard " << shardId, errorStatus);
+                return;
+            }
+            delay = ReadsState.GetDelay(shardId, allowInstantRetry);
+        } else {
+            delay = *throttleDelay;
         }
 
-        auto delay = ReadsState.GetDelay(shardId, allowInstantRetry);
         if (delay > TDuration::Zero()) {
             TlsActivationContext->Schedule(delay, new IEventHandle(this->SelfId(), this->SelfId(), new TEvPrivate::TEvRetrySingleRead(readId)));
         } else {
@@ -3231,8 +3236,10 @@ public:
 
         switch (statusCode) {
             case Ydb::StatusIds::OVERLOADED: {
-                const bool isThrottled = record.HasThrottled() && record.GetThrottled();
-                ScheduleReadRetry(readId, shardId, false, NYql::NDqProto::StatusIds::OVERLOADED, isThrottled);
+                const std::optional<TDuration> throttleDelay = record.HasThrottleDelayMs()
+                    ? std::make_optional(TDuration::MilliSeconds(record.GetThrottleDelayMs()))
+                    : std::nullopt;
+                ScheduleReadRetry(readId, shardId, false, NYql::NDqProto::StatusIds::OVERLOADED, throttleDelay);
                 return;
             }
             case Ydb::StatusIds::INTERNAL_ERROR: {
diff --git a/ydb/core/kqp/runtime/kqp_read_actor.cpp b/ydb/core/kqp/runtime/kqp_read_actor.cpp
index 67269a97daa..14319006f54 100644
--- a/ydb/core/kqp/runtime/kqp_read_actor.cpp
+++ b/ydb/core/kqp/runtime/kqp_read_actor.cpp
@@ -79,7 +79,7 @@ public:
         size_t RetryAttempt = 0;
         size_t SuccessBatches = 0;
 
-        TMaybe<ui32> NodeId = {};
+        TMaybe<ui32> NodeId;
         bool IsFirst = false;
         bool IsFake = false;
 
@@ -752,14 +752,15 @@ public:
         return state->RetryAttempt + 1 > MaxShardRetries();
     }
 
-    void RetryRead(ui64 id, bool allowInstantRetry = true, bool throttled = false) {
+    void RetryRead(ui64 id, bool allowInstantRetry = true, std::optional<TDuration> throttleDelay = std::nullopt) {
         if (!Reads[id] || Reads[id].Finished) {
             return;
         }
 
         auto* state = Reads[id].Shard;
 
-        if (!throttled) {
+        TDuration delay;
+        if (!throttleDelay) {
             if (CheckTotalRetriesExeeded()) {
                 return RuntimeError(TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' retry limit exceeded",
                     NDqProto::StatusIds::UNAVAILABLE);
@@ -770,10 +771,12 @@ public:
                 ResetRead(id);
                 return ResolveShard(state);
             }
+            ++state->RetryAttempt;
+            delay = CalcDelay(state->RetryAttempt, allowInstantRetry);
+        } else {
+            delay = *throttleDelay;
         }
-        ++state->RetryAttempt;
 
-        auto delay = CalcDelay(state->RetryAttempt, allowInstantRetry && !throttled); // TODO: account potential quota shortage
         if (delay == TDuration::Zero()) {
             return DoRetryRead(id);
         }
@@ -1031,13 +1034,15 @@ public:
                 break;
             }
             case Ydb::StatusIds::OVERLOADED: {
-                const bool isThrottled = record.HasThrottled() && record.GetThrottled();
-                if (!isThrottled && (CheckTotalRetriesExeeded() || CheckShardRetriesExeeded(id))) {
+                const std::optional<TDuration> throttleDelay = record.HasThrottleDelayMs()
+                    ? std::make_optional(TDuration::MilliSeconds(record.GetThrottleDelayMs()))
+                    : std::nullopt;
+                if (!throttleDelay && (CheckTotalRetriesExeeded() || CheckShardRetriesExeeded(id))) {
                     return replyError(
                         TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' retry limit exceeded.",
                         NYql::NDqProto::StatusIds::OVERLOADED);
                 }
-                return RetryRead(id, false, isThrottled);
+                return RetryRead(id, false, throttleDelay);
             }
             case Ydb::StatusIds::INTERNAL_ERROR: {
                 if (CheckTotalRetriesExeeded() || CheckShardRetriesExeeded(id)) {
diff --git a/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp b/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp
index 1d0c0179000..658c1b38d61 100644
--- a/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp
+++ b/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp
@@ -74,7 +74,7 @@ TDuration TIteratorReadBackoffSettings::CalcShardDelay(size_t attempt, bool allo
         delay = Min(delay, MaxRetryDelay);
     }
 
-    delay *= (1 - UnsertaintyRatio * RandomNumber<double>());
+    delay *= (1 - UncertaintyRatio * RandomNumber<double>());
 
     return delay;
 }
diff --git a/ydb/core/kqp/runtime/kqp_read_iterator_common.h b/ydb/core/kqp/runtime/kqp_read_iterator_common.h
index a2968935a2b..70bffcac372 100644
--- a/ydb/core/kqp/runtime/kqp_read_iterator_common.h
+++ b/ydb/core/kqp/runtime/kqp_read_iterator_common.h
@@ -11,7 +11,7 @@ struct TIteratorReadBackoffSettings : TAtomicRefCount<TIteratorReadBackoffSettin
     TDuration StartRetryDelay = TDuration::MilliSeconds(5);
     size_t MaxShardAttempts = 10;
     size_t MaxShardResolves = 3;
-    double UnsertaintyRatio = 0.5;
+    double UncertaintyRatio = 0.5;
     double Multiplier = 2.0;
     TDuration MaxRetryDelay = TDuration::Seconds(1);
 
@@ -54,4 +54,4 @@ THolder<NKikimr::TEvDataShard::TEvReadAck> GetDefaultReadAckSettings();
 void SetDefaultReadAckSettings(const NKikimrTxDataShard::TEvReadAck&);
 
 } // namespace NKqp
-} // namespace NKikimr
\ No newline at end of file
+} // namespace NKikimr
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp b/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp
index 356d8117a73..71ce74b6d75 100644
--- a/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp
+++ b/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp
@@ -693,8 +693,10 @@ private:
                 return ResolveTableShards();
             }
             case Ydb::StatusIds::OVERLOADED: {
-                const bool isThrottled = record.HasThrottled() && record.GetThrottled();
-                if (!isThrottled && (CheckTotalRetriesExeeded() || Reads.CheckShardRetriesExeeded(read))) {
+                const std::optional<TDuration> throttleDelay = record.HasThrottleDelayMs()
+                    ? std::make_optional(TDuration::MilliSeconds(record.GetThrottleDelayMs()))
+                    : std::nullopt;
+                if (!throttleDelay && (CheckTotalRetriesExeeded() || Reads.CheckShardRetriesExeeded(read))) {
                     return replyError(
                         TStringBuilder() << "Table '" << StreamLookupWorker->GetTablePath() << "' retry limit exceeded.",
                         NYql::NDqProto::StatusIds::OVERLOADED);
@@ -702,7 +704,7 @@ private:
                 CA_LOG_D("OVERLOADED was received from tablet: " << read.ShardId << "."
                     << getIssues().ToOneLineString());
                 read.SetBlocked();
-                return RetryTableRead(read, /*allowInstantRetry = */false, isThrottled);
+                return RetryTableRead(read, /*allowInstantRetry = */false, throttleDelay);
             }
             case Ydb::StatusIds::INTERNAL_ERROR: {
                 if (CheckTotalRetriesExeeded() || Reads.CheckShardRetriesExeeded(read)) {
@@ -1160,11 +1162,12 @@ private:
         return limit && TotalRetryAttempts + 1 > *limit;
     }
 
-    void RetryTableRead(TReadState& failedRead, bool allowInstantRetry = true, bool isThrottled = false) {
+    void RetryTableRead(TReadState& failedRead, bool allowInstantRetry = true, std::optional<TDuration> throttleDelay = std::nullopt) {
         CA_LOG_D("Retry reading of table: " << StreamLookupWorker->GetTablePath() << ", readId: " << failedRead.Id
             << ", shardId: " << failedRead.ShardId);
 
-        if (!isThrottled) {
+        TDuration delay;
+        if (!throttleDelay) {
             if (CheckTotalRetriesExeeded()) {
                 return RuntimeError(TStringBuilder() << "Table '" << StreamLookupWorker->GetTablePath() << "' retry limit exceeded",
                     NYql::NDqProto::StatusIds::UNAVAILABLE);
@@ -1176,16 +1179,20 @@ private:
                 Reads.eraseRead(failedRead);
                 return ResolveTableShards();
             }
+
+            delay = Reads.CalcDelayForShard(failedRead, allowInstantRetry);
+        } else {
+            delay = *throttleDelay;
         }
 
-        auto delay = Reads.CalcDelayForShard(failedRead, allowInstantRetry);
+
         if (delay == TDuration::Zero()) {
             auto guard = BindAllocator();
             StreamLookupWorker->RebuildRequest(failedRead.ShardId, failedRead.Id, OperationId);
             Reads.eraseRead(failedRead);
             ScheduleNextReads();
         } else {
-            CA_LOG_D("Schedule retry atempt for readId: " << failedRead.Id << " after " << delay);
+            CA_LOG_D("Schedule retry attempt for readId: " << failedRead.Id << " after " << delay);
             TlsActivationContext->Schedule(
                 delay, new IEventHandle(SelfId(), SelfId(), new TEvPrivate::TEvRetryRead(failedRead.Id, failedRead.LastSeqNo, /*instantStart = */ true))
             );
diff --git a/ydb/core/kqp/runtime/scheduler/fwd.h b/ydb/core/kqp/runtime/scheduler/fwd.h
index 9acd6b52b9a..1326fed6caf 100644
--- a/ydb/core/kqp/runtime/scheduler/fwd.h
+++ b/ydb/core/kqp/runtime/scheduler/fwd.h
@@ -68,6 +68,12 @@ namespace NKikimr::NKqp::NScheduler {
         const TDuration MaxRandomDelay;
     };
 
+    struct TOptions {
+        bool Enabled = true;
+        TDelayParams DelayParams;
+        NHdrf::NSnapshot::ELeafFairShare FairShareMode = NHdrf::NSnapshot::ELeafFairShare::EQUAL_TO_PARENT;
+    };
+
 } // namespace NKikimr::NKqp::NScheduler
 
 Y_DECLARE_OUT_SPEC(inline, NKikimr::NKqp::NScheduler::NHdrf::TId, out, id) {
diff --git a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.cpp b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.cpp
index 3c7f124a03e..ddd3f482dda 100644
--- a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.cpp
+++ b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.cpp
@@ -1,6 +1,5 @@
 #include "kqp_compute_scheduler_service.h"
 
-#include "kqp_schedulable_read.h"
 #include "log.h"
 #include "tree/dynamic.h"
 
@@ -11,6 +10,7 @@
 #include <ydb/core/kqp/common/events/workload_service.h>
 #include <ydb/core/kqp/common/simple/services.h>
 #include <ydb/core/protos/feature_flags.pb.h>
+#include <ydb/core/protos/table_service_config.pb.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/actors/core/events.h>
 #include <ydb/library/actors/core/subsystems/stats.h>
@@ -26,11 +26,11 @@ constexpr double Epsilon = 1e-8;
 
 class TComputeSchedulerService : public NActors::TActorBootstrapped<TComputeSchedulerService> {
 public:
-    explicit TComputeSchedulerService(const NScheduler::TOptions& options) : Options(options) {}
+    explicit TComputeSchedulerService(const TDuration& updateFairSharePeriod) : UpdateFairSharePeriod(updateFairSharePeriod) {}
 
     void Bootstrap() {
-        auto counters = MakeIntrusive<TKqpCounters>(AppData()->Counters, &NActors::TActivationContext::AsActorContext());
-        Scheduler = std::make_shared<NScheduler::TComputeScheduler>(counters, Options.DelayParams);
+        Scheduler = AppData()->KqpComputeScheduler;
+        Y_ENSURE(Scheduler);
 
         Send(
             NConsole::MakeConfigsDispatcherID(SelfId().NodeId()),
@@ -38,8 +38,7 @@ public:
             NActors::IEventHandle::FlagTrackDelivery
         );
 
-        Enabled = AppData()->FeatureFlags.GetEnableResourcePoolsScheduler();
-        if (Enabled) {
+        if (Scheduler->IsEnabled()) {
             LOG_I("Enabled on start");
         } else {
             LOG_I("Disabled on start");
@@ -48,7 +47,7 @@ public:
         Scheduler->SetTotalCpuLimit(CalculateTotalCpuLimit()); // TODO: take total cpu limit from outside
 
         Become(&TComputeSchedulerService::State);
-        Schedule(Options.UpdateFairSharePeriod, new NActors::TEvents::TEvWakeup());
+        Schedule(UpdateFairSharePeriod, new NActors::TEvents::TEvWakeup());
     }
 
     STATEFN(State) {
@@ -66,8 +65,6 @@ public:
 
             hFunc(NActors::TEvents::TEvWakeup, Handle);
 
-            hFunc(TEvGetReadFactory, Handle);
-
             default:
                 LOG_E("Unexpected event: " << ev->GetTypeRewrite());
         }
@@ -80,8 +77,8 @@ public:
     void Handle(NConsole::TEvConsole::TEvConfigNotificationRequest::TPtr& ev) {
         const auto& event = ev->Get()->Record;
 
-        Enabled = event.GetConfig().GetFeatureFlags().GetEnableResourcePoolsScheduler();
-        if (Enabled) {
+        Scheduler->ToggleEnabled(event.GetConfig().GetFeatureFlags().GetEnableResourcePoolsScheduler());
+        if (Scheduler->IsEnabled()) {
             LOG_I("Become enabled");
         } else {
             LOG_I("Become disabled");
@@ -197,7 +194,7 @@ public:
         };
 
         auto response = MakeHolder<TEvQueryResponse>();
-        if (Enabled) {
+        if (Scheduler->IsEnabled()) {
             auto query = Scheduler->AddOrUpdateQuery(databaseId, poolId.empty() ? NKikimr::NResourcePool::DEFAULT_POOL_ID : poolId, queryId, attrs);
             response->Query = query;
             LOG_D("Add query: " << databaseId << "/" << poolId << ", TxId: " << queryId);
@@ -216,13 +213,7 @@ public:
 
     void Handle(NActors::TEvents::TEvWakeup::TPtr&) {
         Scheduler->UpdateFairShare();
-        Schedule(Options.UpdateFairSharePeriod, new NActors::TEvents::TEvWakeup());
-    }
-
-    void Handle(TEvGetReadFactory::TPtr& ev) {
-        auto response = MakeHolder<TEvReadFactoryResponse>();
-        response->Factory = std::make_unique<TSchedulableReadFactory>(Scheduler);
-        Send(ev->Sender, response.Release(), 0, 0);
+        Schedule(UpdateFairSharePeriod, new NActors::TEvents::TEvWakeup());
     }
 
 private:
@@ -248,9 +239,8 @@ private:
     }
 
 private:
-    bool Enabled = true;
     TComputeSchedulerPtr Scheduler;
-    const NScheduler::TOptions Options;
+    const TDuration UpdateFairSharePeriod;
 
     struct TPoolParams {
         bool IsFirstRemoval = false;
@@ -266,10 +256,11 @@ namespace NKikimr::NKqp {
 
 namespace NScheduler {
 
-TComputeScheduler::TComputeScheduler(const TIntrusivePtr<TKqpCounters>& counters, const TDelayParams& delayParams, NHdrf::NSnapshot::ELeafFairShare fairShareMode)
-    : Root(std::make_shared<TRoot>(counters))
-    , DelayParams(delayParams)
-    , FairShareMode(fairShareMode)
+TComputeScheduler::TComputeScheduler(const TIntrusivePtr<TKqpCounters>& counters, const TOptions& options)
+    : Enabled(options.Enabled)
+    , Root(std::make_shared<TRoot>(counters))
+    , DelayParams(options.DelayParams)
+    , FairShareMode(options.FairShareMode)
     , KqpCounters(counters)
 {
     auto group = counters->GetKqpCounters();
@@ -350,6 +341,10 @@ TQueryPtr TComputeScheduler::AddOrUpdateQuery(const NHdrf::TDatabaseId& database
 }
 
 NHdrf::NDynamic::TQueryPtr TComputeScheduler::GetReadQuery(const NHdrf::TDatabaseId& databaseId, const NHdrf::TPoolId& poolId) const {
+    if (!IsEnabled()) {
+        return {};
+    }
+
     TReadGuard lock(Mutex);
 
     auto databaseAndPoolId = std::make_pair(databaseId, poolId);
@@ -396,13 +391,29 @@ void TComputeScheduler::UpdateFairShare() {
 
 } // namespace NScheduler
 
-IActor* CreateKqpComputeSchedulerService(const NScheduler::TOptions& options) {
-    Y_ENSURE(options.UpdateFairSharePeriod > TDuration::Zero());
+NScheduler::TComputeSchedulerPtr CreateKqpComputeScheduler(const NMonitoring::TDynamicCounterPtr& counters, const NKikimrConfig::TAppConfig& appConfig) {
+    const auto& schedulerSettings = appConfig.GetTableServiceConfig().GetComputeSchedulerSettings();
+
+    auto options = TOptions{
+        .Enabled = appConfig.GetFeatureFlags().GetEnableResourcePoolsScheduler(),
+        .DelayParams = TDelayParams{
+            .MaxDelay = TDuration::MicroSeconds(schedulerSettings.GetMaxTaskDelayUs()),
+            .MinDelay = TDuration::MicroSeconds(schedulerSettings.GetMinTaskDelayUs()),
+            .AttemptBonus = TDuration::MicroSeconds(schedulerSettings.GetAttemptTaskBonusUs()),
+            .MaxRandomDelay = TDuration::MicroSeconds(schedulerSettings.GetMaxTaskRandomDelayUs()),
+        }
+    };
+
     Y_ENSURE(options.DelayParams.MaxDelay > TDuration::Zero());
     Y_ENSURE(options.DelayParams.MinDelay > TDuration::Zero());
     Y_ENSURE(options.DelayParams.AttemptBonus > TDuration::Zero());
     Y_ENSURE(options.DelayParams.MaxRandomDelay > TDuration::Zero());
-    return new TComputeSchedulerService(options);
+    return std::make_shared<NScheduler::TComputeScheduler>(MakeIntrusive<NKqp::TKqpCounters>(counters), options);
+}
+
+IActor* CreateKqpComputeSchedulerService(const TDuration& updateFairSharePeriod) {
+    Y_ENSURE(updateFairSharePeriod > TDuration::Zero());
+    return new TComputeSchedulerService(updateFairSharePeriod);
 }
 
 } // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h
index 6483aeec4d9..cb09bced996 100644
--- a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h
+++ b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h
@@ -10,8 +10,14 @@ namespace NKikimr::NKqp::NScheduler {
 
 class TComputeScheduler : public std::enable_shared_from_this<TComputeScheduler> {
 public:
-    TComputeScheduler(const TIntrusivePtr<TKqpCounters>& counters, const TDelayParams& delayParams,
-        NHdrf::NSnapshot::ELeafFairShare fairShareMode = NHdrf::NSnapshot::ELeafFairShare::EQUAL_TO_PARENT);
+    TComputeScheduler(const TIntrusivePtr<TKqpCounters>& counters, const TOptions& options);
+
+    void ToggleEnabled(bool enable) {
+        Enabled = enable;
+    }
+    bool IsEnabled() const {
+        return Enabled;
+    }
 
     void SetTotalCpuLimit(ui64 cpu);
     ui64 GetTotalCpuLimit() const;
@@ -29,6 +35,8 @@ public:
 private:
     static constexpr NHdrf::TQueryId READ_QUERY_ID = -1;
 
+    std::atomic<bool> Enabled;
+
     TRWMutex Mutex;
     NHdrf::NDynamic::TRootPtr Root;                                // protected by Mutex
     THashMap<NHdrf::TQueryId, NHdrf::NDynamic::TQueryPtr> Queries; // protected by Mutex
@@ -48,11 +56,6 @@ private:
 
 using TComputeSchedulerPtr = std::shared_ptr<TComputeScheduler>;
 
-struct TOptions {
-    TDelayParams DelayParams;
-    TDuration UpdateFairSharePeriod;
-};
-
 struct TEvents {
     enum : ui32 {
         EvAddDatabase = EventSpaceBegin(TKikimrEvents::ES_KQP) + 400,
@@ -62,11 +65,6 @@ struct TEvents {
         EvAddQuery,
         EvRemoveQuery,
         EvQueryResponse,
-
-        // Because datashard may get EvRead from another node, it's hard to use EvAddQuery+EvQueryResponse.
-        // These messages return factory that can create schedulable objects on-the-fly.
-        EvGetReadFactory,
-        EvReadFactoryResponse,
     };
 };
 
@@ -111,16 +109,15 @@ struct TEvQueryResponse : public TEventLocal<TEvQueryResponse, TEvents::EvQueryR
     NHdrf::NDynamic::TQueryPtr Query;
 };
 
-struct TEvGetReadFactory : public TEventLocal<TEvGetReadFactory, TEvents::EvGetReadFactory> {
-    // TODO: datashard id?
-};
-
-struct TEvReadFactoryResponse : public TEventLocal<TEvReadFactoryResponse, TEvents::EvReadFactoryResponse> {
-    TSchedulableReadFactoryPtr Factory;
-};
-
 } // namespace NKikimr::NKqp::NScheduler
 
+namespace NKikimrConfig {
+    class TAppConfig;
+}
+
 namespace NKikimr::NKqp {
-    IActor* CreateKqpComputeSchedulerService(const NScheduler::TOptions& options);
+    NScheduler::TComputeSchedulerPtr CreateKqpComputeScheduler(
+        const NMonitoring::TDynamicCounterPtr& counters,
+        const NKikimrConfig::TAppConfig& appConfig);
+    IActor* CreateKqpComputeSchedulerService(const TDuration& updateFairSharePeriod);
 }
diff --git a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_ut.cpp b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_ut.cpp
index 2f88104ea45..380782502ec 100644
--- a/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_ut.cpp
+++ b/ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_ut.cpp
@@ -21,8 +21,6 @@ namespace {
         .MaxRandomDelay = TDuration::MicroSeconds(100),
     };
 
-    constexpr TDuration kDefaultUpdateFairSharePeriod = TDuration::MilliSeconds(500);
-
     std::vector<TSchedulableTaskPtr> CreateDemandTasks(const NHdrf::NDynamic::TQueryPtr& query, ui64 demand) {
         std::vector<TSchedulableTaskPtr> tasks;
 
@@ -57,15 +55,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kQueryDemand = 2;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -112,9 +110,9 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
+            .FairShareMode = NHdrf::NSnapshot::ELeafFairShare::DEFAULT_FIFO,
         };
-        TComputeScheduler scheduler(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::DEFAULT_FIFO);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -164,9 +162,9 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
+            .FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT,
         };
-        TComputeScheduler scheduler(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -212,9 +210,9 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
+            .FairShareMode = NHdrf::NSnapshot::ELeafFairShare::EQUAL_TO_PARENT,
         };
-        TComputeScheduler scheduler(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::EQUAL_TO_PARENT);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -261,9 +259,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -312,15 +309,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kQueryDemand = 4;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -366,15 +363,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kWeightedFairShare = 4;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -432,15 +429,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kWeightedFairShare = 4;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -550,15 +547,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kCpuLimit = 20;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -653,9 +650,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -700,9 +696,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -750,9 +745,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -795,9 +789,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -825,16 +818,16 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kQueryDemand = 5;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
 
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -908,15 +901,15 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kQueryDemand = 5;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -990,16 +983,16 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         constexpr ui64 kRoundedDemand = 2;
 
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
-        const TOptions options{
+        TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
 
         std::unique_ptr<TComputeScheduler> scheduler;
         if (DefaultFairShareMode) {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams);
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         } else {
-            scheduler = std::make_unique<TComputeScheduler>(counters, options.DelayParams, NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT);
+            options.FairShareMode = NHdrf::NSnapshot::ELeafFairShare::ALLOW_OVERLIMIT;
+            scheduler = std::make_unique<TComputeScheduler>(counters, options);
         }
         scheduler->SetTotalCpuLimit(kCpuLimit);
 
@@ -1093,9 +1086,8 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
         auto counters = MakeIntrusive<TKqpCounters>(MakeIntrusive<NMonitoring::TDynamicCounters>());
         const TOptions options{
             .DelayParams = kDefaultDelayParams,
-            .UpdateFairSharePeriod = kDefaultUpdateFairSharePeriod
         };
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
@@ -1124,7 +1116,7 @@ Y_UNIT_TEST_SUITE(KqpComputeScheduler) {
             .DelayParams = kDefaultDelayParams,
         };
 
-        TComputeScheduler scheduler(counters, options.DelayParams);
+        TComputeScheduler scheduler(counters, options);
         scheduler.SetTotalCpuLimit(kCpuLimit);
 
         const TString databaseId = "db1";
diff --git a/ydb/core/kqp/runtime/scheduler/kqp_schedulable_read.cpp b/ydb/core/kqp/runtime/scheduler/kqp_schedulable_read.cpp
index df526624930..d49169e683c 100644
--- a/ydb/core/kqp/runtime/scheduler/kqp_schedulable_read.cpp
+++ b/ydb/core/kqp/runtime/scheduler/kqp_schedulable_read.cpp
@@ -73,12 +73,11 @@ void TSchedulableRead::ReturnQuota(NHPTimer::STime elapsedCycles) {
 }
 
 TDuration TSchedulableRead::EstimateQuotaDelay(TDuration expectedQuota) const {
-    Y_ASSERT(QuotaPerSecond != 0);
+    const auto expectedQuotaMs = std::min(expectedQuota.MilliSeconds(), MaxQuotaMs);
 
-    auto expectedQuotaMs = std::min(expectedQuota.MilliSeconds(), MaxQuotaMs);
-
-    if (AvailableQuotaMs >= static_cast<i64>(expectedQuotaMs)) {
+    if (QuotaPerSecond == 0 || AvailableQuotaMs >= static_cast<i64>(expectedQuotaMs)) {
         // Quota available, but TryIncreaseUsage() failed (fair-share exhausted)
+        // TODO: handle queries to the pool with 0% limit somewhere else, and don't allow them to execute.
         return TDuration::MilliSeconds(10) + RandomDuration(TDuration::MilliSeconds(1));
     }
 
@@ -92,7 +91,9 @@ TDuration TSchedulableRead::EstimateQuotaDelay(TDuration expectedQuota) const {
 
 TSchedulableReadFactory::TSchedulableReadFactory(TComputeSchedulerPtr scheduler)
     : Scheduler(std::move(scheduler))
-{}
+{
+    Y_ENSURE(Scheduler);
+}
 
 TSchedulableReadPtr TSchedulableReadFactory::Get(const NHdrf::TDatabaseId& databaseId, const NHdrf::TPoolId& poolId) const {
     const auto databaseAndPoolId = std::make_pair(databaseId, poolId);
diff --git a/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp b/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp
index 53ac20d2d03..9a10ae3ad34 100644
--- a/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp
+++ b/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp
@@ -2615,8 +2615,9 @@ Y_UNIT_TEST_SUITE(KqpScan) {
                 );
             )").GetValueSync();
             UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
-
-            result = session.ExecuteDataQuery(R"(
+        }
+        {
+            auto result = session.ExecuteDataQuery(R"(
                 REPLACE INTO `/Root/TestTable` (Key, Value) VALUES
                     ('SomeString1', '100'),
                     ('SomeString2', '200'),
@@ -2675,7 +2676,7 @@ Y_UNIT_TEST_SUITE(KqpScan) {
 
         server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPUTE, NActors::NLog::EPriority::PRI_DEBUG);
 
-        auto runtime = server->GetRuntime();
+        auto* runtime = server->GetRuntime();
         auto sender = runtime->AllocateEdgeActor();
         auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
 
@@ -2787,6 +2788,539 @@ Y_UNIT_TEST_SUITE(KqpScan) {
         )");
     }
 
+    // Throttled OVERLOADED responses (from KQP Compute Scheduler quota) must not be counted
+    // against MaxTotalRetries. The test sets a low MaxTotalRetries and feeds many throttled
+    // responses; query must complete successfully.
+    Y_UNIT_TEST(StreamLookupThrottleDoesNotExhaustTotalRetries) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamIdxLookupJoin(true);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetStartDelayMs(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxDelayMs(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(1.0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardRetries(100);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardResolves(100);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(2);
+
+        TPortManager tp;
+        ui16 mbusport = tp.GetPort(2134);
+        auto settings = Tests::TServerSettings(mbusport)
+            .SetDomainName("Root")
+            .SetUseRealThreads(false)
+            .SetAppConfig(appConfig);
+
+        Tests::TServer::TPtr server = new Tests::TServer(settings);
+        auto* runtime = server->GetRuntime();
+        auto sender = runtime->AllocateEdgeActor();
+        auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
+        InitRoot(server, sender);
+
+        constexpr int kThrottleCount = 5;  // > MaxTotalRetries
+        int throttleResponded = 0;
+
+        auto captureEvents = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+            if (ev->GetTypeRewrite() == NKikimr::TEvDataShard::TEvRead::EventType) {
+                if (throttleResponded < kThrottleCount) {
+                    auto& record = ev->Get<NKikimr::TEvDataShard::TEvRead>()->Record;
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    resp->Record.SetThrottleDelayMs(1);
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++throttleResponded;
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        auto createSession = [&]() {
+            runtime->Send(new IEventHandle(kqpProxy, sender, new TEvKqp::TEvCreateSessionRequest()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvCreateSessionResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+            return reply->Get()->Record.GetResponse().GetSessionId();
+        };
+
+        auto createTable = [&](const TString& sessionId, const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->SetSessionId(sessionId);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DDL);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+        };
+
+        auto sendQuery = [&](const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
+            ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            ev->Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
+            ActorIdToProto(sender, ev->Record.MutableRequestActorId());
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL_C(
+                reply->Get()->Record.GetYdbStatus(),
+                Ydb::StatusIds::SUCCESS,
+                reply->Get()->Record.GetResponse().DebugString());
+        };
+
+        createTable(createSession(), R"(
+            --!syntax_v1
+            CREATE TABLE `/Root/Table1` (Key uint32, Value uint32, PRIMARY KEY(Key));
+        )");
+
+        runtime->SetEventFilter(captureEvents);
+
+        sendQuery(R"(
+            $data = AsList(AsStruct(1u AS Key, 1u AS Value));
+            SELECT a.Value, b.Value
+            FROM AS_TABLE($data) a
+            JOIN `/Root/Table1` b
+            ON a.Key = b.Key;
+        )");
+
+        UNIT_ASSERT_VALUES_EQUAL(throttleResponded, kThrottleCount);
+    }
+
+    // Throttled OVERLOADED responses must not consume the per-shard retry budget.
+    // The test feeds N throttled responses (N > MaxShardRetries), then a single non-throttled
+    // OVERLOADED, then lets the real datashard answer.
+    // Bug: throttle bumps shardState.RetryAttempts -> CheckShardRetriesExeeded triggers ResolveShard
+    // on the first non-throttle error -> with MaxShardResolves=1 it exceeds and query fails.
+    // Fix: throttle does not bump the counter, the non-throttle retry succeeds.
+    Y_UNIT_TEST(StreamLookupThrottleDoesNotExhaustShardRetries) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamIdxLookupJoin(true);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetStartDelayMs(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxDelayMs(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(1.0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardRetries(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardResolves(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(100);
+
+        TPortManager tp;
+        ui16 mbusport = tp.GetPort(2134);
+        auto settings = Tests::TServerSettings(mbusport)
+            .SetDomainName("Root")
+            .SetUseRealThreads(false)
+            .SetAppConfig(appConfig);
+
+        Tests::TServer::TPtr server = new Tests::TServer(settings);
+        auto* runtime = server->GetRuntime();
+        auto sender = runtime->AllocateEdgeActor();
+        auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
+        InitRoot(server, sender);
+
+        constexpr int kThrottleCount = 5;  // > MaxShardRetries
+        int throttleResponded = 0;
+        int nonThrottleResponded = 0;
+
+        auto captureEvents = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+            if (ev->GetTypeRewrite() == NKikimr::TEvDataShard::TEvRead::EventType) {
+                auto& record = ev->Get<NKikimr::TEvDataShard::TEvRead>()->Record;
+                if (throttleResponded < kThrottleCount) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    resp->Record.SetThrottleDelayMs(1);
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++throttleResponded;
+                    return true;
+                }
+                if (nonThrottleResponded == 0) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    // No Throttled flag.
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++nonThrottleResponded;
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        auto createSession = [&]() {
+            runtime->Send(new IEventHandle(kqpProxy, sender, new TEvKqp::TEvCreateSessionRequest()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvCreateSessionResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+            return reply->Get()->Record.GetResponse().GetSessionId();
+        };
+
+        auto createTable = [&](const TString& sessionId, const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->SetSessionId(sessionId);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DDL);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+        };
+
+        auto sendQuery = [&](const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
+            ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            ev->Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
+            ActorIdToProto(sender, ev->Record.MutableRequestActorId());
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL_C(
+                reply->Get()->Record.GetYdbStatus(),
+                Ydb::StatusIds::SUCCESS,
+                reply->Get()->Record.GetResponse().DebugString());
+        };
+
+        createTable(createSession(), R"(
+            --!syntax_v1
+            CREATE TABLE `/Root/Table1` (Key uint32, Value uint32, PRIMARY KEY(Key));
+        )");
+
+        runtime->SetEventFilter(captureEvents);
+
+        sendQuery(R"(
+            $data = AsList(AsStruct(1u AS Key, 1u AS Value));
+            SELECT a.Value, b.Value
+            FROM AS_TABLE($data) a
+            JOIN `/Root/Table1` b
+            ON a.Key = b.Key;
+        )");
+
+        UNIT_ASSERT_VALUES_EQUAL(throttleResponded, kThrottleCount);
+        UNIT_ASSERT_VALUES_EQUAL(nonThrottleResponded, 1);
+    }
+
+    // Throttled OVERLOADED responses (from KQP Compute Scheduler quota) must not be counted
+    // against MaxTotalRetries in kqp_read_actor. The test sets a low MaxTotalRetries and feeds
+    // many throttled responses; query must complete successfully.
+    Y_UNIT_TEST(ReadActorThrottleDoesNotExhaustTotalRetries) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetStartDelayMs(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxDelayMs(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(1.0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardRetries(100);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardResolves(100);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(2);
+
+        TPortManager tp;
+        ui16 mbusport = tp.GetPort(2134);
+        auto settings = Tests::TServerSettings(mbusport)
+            .SetDomainName("Root")
+            .SetUseRealThreads(false)
+            .SetAppConfig(appConfig);
+
+        Tests::TServer::TPtr server = new Tests::TServer(settings);
+        auto* runtime = server->GetRuntime();
+        auto sender = runtime->AllocateEdgeActor();
+        auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
+        InitRoot(server, sender);
+
+        constexpr int kThrottleCount = 5;  // > MaxTotalRetries
+        int throttleResponded = 0;
+
+        auto captureEvents = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+            if (ev->GetTypeRewrite() == NKikimr::TEvDataShard::TEvRead::EventType) {
+                if (throttleResponded < kThrottleCount) {
+                    auto& record = ev->Get<NKikimr::TEvDataShard::TEvRead>()->Record;
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    resp->Record.SetThrottleDelayMs(1);
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++throttleResponded;
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        auto createSession = [&]() {
+            runtime->Send(new IEventHandle(kqpProxy, sender, new TEvKqp::TEvCreateSessionRequest()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvCreateSessionResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+            return reply->Get()->Record.GetResponse().GetSessionId();
+        };
+
+        auto createTable = [&](const TString& sessionId, const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->SetSessionId(sessionId);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DDL);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+        };
+
+        auto sendQuery = [&](const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
+            ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            ev->Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
+            ActorIdToProto(sender, ev->Record.MutableRequestActorId());
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL_C(
+                reply->Get()->Record.GetYdbStatus(),
+                Ydb::StatusIds::SUCCESS,
+                reply->Get()->Record.GetResponse().DebugString());
+        };
+
+        createTable(createSession(), R"(
+            --!syntax_v1
+            CREATE TABLE `/Root/Table1` (Key uint32, Value uint32, PRIMARY KEY(Key));
+        )");
+
+        runtime->SetEventFilter(captureEvents);
+
+        sendQuery(R"(
+            SELECT * FROM `/Root/Table1`;
+        )");
+
+        UNIT_ASSERT_VALUES_EQUAL(throttleResponded, kThrottleCount);
+    }
+
+    // Throttled OVERLOADED responses must not consume the per-shard retry budget in kqp_read_actor.
+    // The test feeds N throttled responses (N > MaxShardRetries), then a single non-throttled
+    // OVERLOADED, then lets the real datashard answer.
+    // Bug: throttle bumps state->RetryAttempt -> CheckShardRetriesExeeded triggers on the first
+    // non-throttle error -> with MaxShardResolves=1 it exceeds and query fails.
+    // Fix: throttle does not bump the counter, the non-throttle retry succeeds.
+    Y_UNIT_TEST(ReadActorThrottleDoesNotExhaustShardRetries) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetStartDelayMs(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxDelayMs(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(1.0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardRetries(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardResolves(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(100);
+
+        TPortManager tp;
+        ui16 mbusport = tp.GetPort(2134);
+        auto settings = Tests::TServerSettings(mbusport)
+            .SetDomainName("Root")
+            .SetUseRealThreads(false)
+            .SetAppConfig(appConfig);
+
+        Tests::TServer::TPtr server = new Tests::TServer(settings);
+        auto* runtime = server->GetRuntime();
+        auto sender = runtime->AllocateEdgeActor();
+        auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
+        InitRoot(server, sender);
+
+        constexpr int kThrottleCount = 5;  // > MaxShardRetries
+        int throttleResponded = 0;
+        int nonThrottleResponded = 0;
+
+        auto captureEvents = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+            if (ev->GetTypeRewrite() == NKikimr::TEvDataShard::TEvRead::EventType) {
+                auto& record = ev->Get<NKikimr::TEvDataShard::TEvRead>()->Record;
+                if (throttleResponded < kThrottleCount) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    resp->Record.SetThrottleDelayMs(1);
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++throttleResponded;
+                    return true;
+                }
+                if (nonThrottleResponded == 0) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    // No Throttled flag.
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++nonThrottleResponded;
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        auto createSession = [&]() {
+            runtime->Send(new IEventHandle(kqpProxy, sender, new TEvKqp::TEvCreateSessionRequest()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvCreateSessionResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+            return reply->Get()->Record.GetResponse().GetSessionId();
+        };
+
+        auto createTable = [&](const TString& sessionId, const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->SetSessionId(sessionId);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DDL);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+        };
+
+        auto sendQuery = [&](const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
+            ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            ev->Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
+            ActorIdToProto(sender, ev->Record.MutableRequestActorId());
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL_C(
+                reply->Get()->Record.GetYdbStatus(),
+                Ydb::StatusIds::SUCCESS,
+                reply->Get()->Record.GetResponse().DebugString());
+        };
+
+        createTable(createSession(), R"(
+            --!syntax_v1
+            CREATE TABLE `/Root/Table1` (Key uint32, Value uint32, PRIMARY KEY(Key));
+        )");
+
+        runtime->SetEventFilter(captureEvents);
+
+        sendQuery(R"(
+            SELECT * FROM `/Root/Table1`;
+        )");
+
+        UNIT_ASSERT_VALUES_EQUAL(throttleResponded, kThrottleCount);
+        UNIT_ASSERT_VALUES_EQUAL(nonThrottleResponded, 1);
+    }
+
+    // Throttled OVERLOADED responses must not consume the per-shard retry budget in
+    // kqp_buffer_lookup_actor (triggered by INSERT/UPSERT into a table with a UNIQUE secondary
+    // index). The test feeds N throttled responses (N > MaxShardRetries), then a single
+    // non-throttled OVERLOADED.
+    // Bug: throttle bumps failedRead.RetryAttempts -> on the next non-throttle response
+    // the check `!isThrottled && RetryAttempts >= MaxShardRetries` fails the retry,
+    // RetryTableRead returns false, and the actor replies with OVERLOADED.
+    // Fix: throttle does not bump the counter, the non-throttle retry succeeds.
+    // Note: TKqpBufferLookupActor has no MaxTotalRetries / MaxShardResolves limits — only
+    // MaxShardRetries — so only the shard-retry variant is meaningful here.
+    Y_UNIT_TEST(BufferLookupThrottleDoesNotExhaustShardRetries) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetStartDelayMs(1);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxDelayMs(2);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(1.0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxShardRetries(2);
+        appConfig.MutableTableServiceConfig()->SetEnableIndexStreamWrite(true);
+
+        TPortManager tp;
+        ui16 mbusport = tp.GetPort(2134);
+        auto settings = Tests::TServerSettings(mbusport)
+            .SetDomainName("Root")
+            .SetUseRealThreads(false)
+            .SetAppConfig(appConfig);
+
+        Tests::TServer::TPtr server = new Tests::TServer(settings);
+        auto* runtime = server->GetRuntime();
+        auto sender = runtime->AllocateEdgeActor();
+        auto kqpProxy = MakeKqpProxyID(runtime->GetNodeId(0));
+        InitRoot(server, sender);
+
+        constexpr int kThrottleCount = 5;  // > MaxShardRetries
+        int throttleResponded = 0;
+        int nonThrottleResponded = 0;
+
+        auto captureEvents = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+            if (ev->GetTypeRewrite() == NKikimr::TEvDataShard::TEvRead::EventType) {
+                if (runtime->FindActorName(ev->Sender) != "KQP_BUFFER_LOOKUP_ACTOR") {
+                    return false;
+                }
+                auto& record = ev->Get<NKikimr::TEvDataShard::TEvRead>()->Record;
+                if (throttleResponded < kThrottleCount) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    resp->Record.SetThrottleDelayMs(1);
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++throttleResponded;
+                    return true;
+                }
+                if (nonThrottleResponded == 0) {
+                    auto resp = MakeHolder<NKikimr::TEvDataShard::TEvReadResult>();
+                    resp->Record.SetReadId(record.GetReadId());
+                    resp->Record.MutableStatus()->SetCode(Ydb::StatusIds::OVERLOADED);
+                    // No Throttled flag.
+                    runtime->Send(new IEventHandle(ev->Sender, ev->GetRecipientRewrite(), resp.Release()));
+                    ++nonThrottleResponded;
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        auto createSession = [&]() {
+            runtime->Send(new IEventHandle(kqpProxy, sender, new TEvKqp::TEvCreateSessionRequest()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvCreateSessionResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+            return reply->Get()->Record.GetResponse().GetSessionId();
+        };
+
+        auto createTable = [&](const TString& sessionId, const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->SetSessionId(sessionId);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DDL);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL(reply->Get()->Record.GetYdbStatus(), Ydb::StatusIds::SUCCESS);
+        };
+
+        auto sendQuery = [&](const TString& queryText) {
+            auto ev = std::make_unique<NKqp::TEvKqp::TEvQueryRequest>();
+            ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
+            ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
+            ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
+            ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
+            ev->Record.MutableRequest()->SetQuery(queryText);
+            ev->Record.MutableRequest()->SetUsePublicResponseDataFormat(true);
+            ActorIdToProto(sender, ev->Record.MutableRequestActorId());
+            runtime->Send(new IEventHandle(kqpProxy, sender, ev.release()));
+            auto reply = runtime->GrabEdgeEventRethrow<TEvKqp::TEvQueryResponse>(sender);
+            UNIT_ASSERT_VALUES_EQUAL_C(
+                reply->Get()->Record.GetYdbStatus(),
+                Ydb::StatusIds::SUCCESS,
+                reply->Get()->Record.GetResponse().DebugString());
+        };
+
+        createTable(createSession(), R"(
+            --!syntax_v1
+            CREATE TABLE `/Root/Table1` (
+                Key uint32,
+                Value uint32 NOT NULL,
+                PRIMARY KEY(Key),
+                INDEX UniqIdx GLOBAL UNIQUE SYNC ON (Value)
+            );
+        )");
+
+        runtime->SetEventFilter(captureEvents);
+
+        sendQuery(R"(
+            UPSERT INTO `/Root/Table1` (Key, Value) VALUES (1u, 1u);
+        )");
+
+        UNIT_ASSERT_VALUES_EQUAL(throttleResponded, kThrottleCount);
+        UNIT_ASSERT_VALUES_EQUAL(nonThrottleResponded, 1);
+    }
+
     Y_UNIT_TEST(DecimalColumnCsvBulkUpsertScan) {
         TKikimrSettings settings;
         settings.AppConfig.MutableTableServiceConfig()->SetEnableKqpScanQuerySourceRead(true);
diff --git a/ydb/core/protos/tx_datashard.proto b/ydb/core/protos/tx_datashard.proto
index eff25dec09d..48ccf934d95 100644
--- a/ydb/core/protos/tx_datashard.proto
+++ b/ydb/core/protos/tx_datashard.proto
@@ -2424,8 +2424,8 @@ message TEvReadResult {
     optional uint64 DeferredVictimQuerySpanId = 16;
 
     // Indicates that initial read was throttled by KQP Compute Scheduler due to lack of read quota.
-    // The reader should retry after some reasonable delay.
-    optional bool Throttled = 17;
+    // The reader should retry only after this delay.
+    optional uint64 ThrottleDelayMs = 17;
 
     // Data for the possibly partial result
     oneof ReadResult {
diff --git a/ydb/core/testlib/test_client.cpp b/ydb/core/testlib/test_client.cpp
index 247b8ca0c2d..3b2c1f0523f 100644
--- a/ydb/core/testlib/test_client.cpp
+++ b/ydb/core/testlib/test_client.cpp
@@ -1,5 +1,6 @@
 #include "test_client.h"
 
+#include <ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h>
 #include <ydb/core/kqp/federated_query/actors/kqp_federated_query_actors.h>
 #include <ydb/core/testlib/basics/runtime.h>
 #include <ydb/core/base/path.h>
@@ -1446,6 +1447,9 @@ namespace Tests {
                 });
             }
 
+            auto counters = MakeIntrusive<::NMonitoring::TDynamicCounters>();
+            Runtime->GetAppData(nodeIdx).KqpComputeScheduler = NKqp::CreateKqpComputeScheduler(counters, *Settings->AppConfig);
+
             IActor* kqpProxyService = NKqp::CreateKqpProxyService(Settings->AppConfig->GetLogConfig(),
                                                                   Settings->AppConfig->GetTableServiceConfig(),
                                                                   Settings->AppConfig->GetQueryServiceConfig(),
diff --git a/ydb/core/tx/datashard/datashard.cpp b/ydb/core/tx/datashard/datashard.cpp
index 427bf566db3..3782103646f 100644
--- a/ydb/core/tx/datashard/datashard.cpp
+++ b/ydb/core/tx/datashard/datashard.cpp
@@ -402,9 +402,10 @@ void TDataShard::OnActivateExecutor(const TActorContext& ctx) {
         Execute(CreateTxInitSchema(), ctx);
         Become(&TThis::StateInactive);
 
-        // Get factory from KQP Scheduler and schedule delayed empty response as fail-safe measure
-        ctx.Send(NKqp::MakeKqpSchedulerServiceId(ctx.SelfID.NodeId()), new NKqp::NScheduler::TEvGetReadFactory, IEventHandle::FlagTrackDelivery);
-        ctx.Schedule(TDuration::Seconds(1), new NKqp::NScheduler::TEvReadFactoryResponse);
+        // In tests the scheduler may be uninitialized
+        if (auto scheduler = AppData()->KqpComputeScheduler) {
+            SchedulableReadFactory = std::make_unique<NKqp::NScheduler::TSchedulableReadFactory>(scheduler);
+        }
     } else {
         SyncConfig();
         State = TShardState::Readonly;
@@ -2858,10 +2859,6 @@ bool TDataShard::NeedMediatorStateRestored() const {
 }
 
 void TDataShard::CheckMediatorStateRestored() {
-    if (!SchedulableReadFactory) {
-        return;
-    }
-
     if (!MediatorStateWaiting ||
         !RegistrationSended ||
         !MediatorTimeCastEntry ||
@@ -4069,8 +4066,8 @@ void TDataShard::DoPeriodicTasks(const TActorContext &ctx) {
         LOG_NOTICE_S(ctx, NKikimrServices::TX_DATASHARD, "Stoped key access sampling at datashard: " << TabletID());
     }
 
-    if (SchedulableReadFactory && *SchedulableReadFactory) {
-        (*SchedulableReadFactory)->CleanupReadsCache();
+    if (SchedulableReadFactory) {
+        SchedulableReadFactory->CleanupReadsCache();
     }
 
     if (!PeriodicWakeupPending) {
@@ -4965,20 +4962,6 @@ void TDataShard::OnTableCreated(TTransactionContext &txc, const TActorContext &c
     }
 }
 
-void TDataShard::Handle(NKqp::NScheduler::TEvReadFactoryResponse::TPtr& ev) {
-    if (!SchedulableReadFactory) {
-        SchedulableReadFactory = std::move(ev->Get()->Factory);
-    }
-    CheckMediatorStateRestored();
-}
-
-void TDataShard::HandleInactive(TEvents::TEvUndelivered::TPtr& ev) {
-    if (ev->Get()->SourceType == NKqp::NScheduler::TEvGetReadFactory::EventType) {
-        SchedulableReadFactory = nullptr;
-        CheckMediatorStateRestored();
-    }
-}
-
 } // NDataShard
 
 TString TEvDataShard::TEvRead::ToString() const {
diff --git a/ydb/core/tx/datashard/datashard__read_iterator.cpp b/ydb/core/tx/datashard/datashard__read_iterator.cpp
index 903053aa0c0..bd8bd1eddf6 100644
--- a/ydb/core/tx/datashard/datashard__read_iterator.cpp
+++ b/ydb/core/tx/datashard/datashard__read_iterator.cpp
@@ -1385,8 +1385,11 @@ std::unique_ptr<TEvDataShard::TEvReadResult> MakeEvReadResult(ui32 nodeId) {
 }
 
 
+const TDuration MaxTimePerIteration = TDuration::MilliSeconds(10);
+
 const NHPTimer::STime TReader::MaxCyclesPerIteration =
-    /* 10ms */ (NHPTimer::GetCyclesPerSecond() + 99) / 100;
+    ((NHPTimer::GetCyclesPerSecond() * MaxTimePerIteration.MicroSeconds()) + TDuration::Seconds(1).MicroSeconds() - 1)
+    / TDuration::Seconds(1).MicroSeconds();
 
 } // namespace
 
@@ -2079,14 +2082,14 @@ public:
         }
 
         const auto& schedulableRead = state.SchedulableRead;
-        if (schedulableRead && !schedulableRead->TryConsumeQuota(TDuration::MilliSeconds(10))) {
+        if (schedulableRead && !schedulableRead->TryConsumeQuota(MaxTimePerIteration)) {
             SetStatusError(
                 Result->Record,
                 Ydb::StatusIds::OVERLOADED,
                 TStringBuilder() << "Read quota for resource pool exceeded" // TODO: add pool id
                     << " (shard# " << Self->TabletID()
                     << " node# " << ctx.SelfID.NodeId() << ")");
-            Result->Record.SetThrottled(true);
+            Result->Record.SetThrottleDelayMs(schedulableRead->EstimateQuotaDelay(MaxTimePerIteration).MilliSeconds());
             return EExecutionStatus::DelayComplete;
         }
 
@@ -3402,14 +3405,14 @@ public:
         // Try to consume schedulable read quota before reading
         const auto& schedulableRead = state.SchedulableRead;
         if (schedulableRead) {
-            if (!schedulableRead->TryConsumeQuota(TDuration::MilliSeconds(10))) {
+            if (!schedulableRead->TryConsumeQuota(MaxTimePerIteration)) {
                 // KQP read quota exhausted, reschedule with delay.
                 // Keep ReadContinuePending=true so that ReadAck doesn't schedule a duplicate TEvReadContinue while we wait.
                 state.ReadContinuePending = true;
                 Reader.reset();
                 Result.reset();
                 ctx.Schedule(
-                    schedulableRead->EstimateQuotaDelay(TDuration::MilliSeconds(10)),
+                    schedulableRead->EstimateQuotaDelay(MaxTimePerIteration),
                     new TEvDataShard::TEvReadContinue(LocalReadId));
                 return true;
             }
@@ -3792,8 +3795,8 @@ void TDataShard::Handle(TEvDataShard::TEvRead::TPtr& ev, const TActorContext& ct
     }
 
     NKqp::NScheduler::TSchedulableReadPtr schedulableRead;
-    if (record.HasPoolId() && !record.GetPoolId().empty() && SchedulableReadFactory && *SchedulableReadFactory) {
-        schedulableRead = (*SchedulableReadFactory)->Get(record.GetDatabaseId(), record.GetPoolId());
+    if (record.HasPoolId() && !record.GetPoolId().empty() && SchedulableReadFactory) {
+        schedulableRead = SchedulableReadFactory->Get(record.GetDatabaseId(), record.GetPoolId());
     }
 
     ui64 localReadId = NextTieBreakerIndex++;
diff --git a/ydb/core/tx/datashard/datashard_impl.h b/ydb/core/tx/datashard/datashard_impl.h
index e7b1b085743..61657d99731 100644
--- a/ydb/core/tx/datashard/datashard_impl.h
+++ b/ydb/core/tx/datashard/datashard_impl.h
@@ -42,7 +42,7 @@
 #include <ydb/library/ydb_issue/issue_helpers.h>
 #include <ydb/core/change_exchange/change_exchange.h>
 #include <ydb/core/engine/mkql_engine_flat_host.h>
-#include <ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h>
+#include <ydb/core/kqp/runtime/scheduler/fwd.h>
 #include <ydb/core/statistics/events.h>
 #include <ydb/core/tablet/pipe_tracker.h>
 #include <ydb/core/tablet/tablet_exception.h>
@@ -1481,10 +1481,6 @@ class TDataShard
 
     void Handle(TEvDataShard::TEvVacuum::TPtr& ev, const TActorContext& ctx);
 
-    void Handle(NKqp::NScheduler::TEvReadFactoryResponse::TPtr& ev);
-
-    void HandleInactive(TEvents::TEvUndelivered::TPtr& ev);
-
     void HandleByReplicationSourceOffsetsServer(STATEFN_SIG);
 
     void DoPeriodicTasks(const TActorContext &ctx);
@@ -3211,9 +3207,7 @@ private:
     TIntrusiveList<TGlobalTxIdAwaiter> GlobalTxIdAwaiters;
     TVector<ui64> GlobalTxIdCache;
 
-    // If value is un-set then wait in StateInactive.
-    // Also it's fine if value is set and nullptr - it means: don't use the read factory.
-    std::optional<NKqp::NScheduler::TSchedulableReadFactoryPtr> SchedulableReadFactory;
+    NKqp::NScheduler::TSchedulableReadFactoryPtr SchedulableReadFactory;
 
 public:
     struct TBreakerInfo {
@@ -3312,8 +3306,6 @@ protected:
             HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
             HFunc(TEvPrivate::TEvBuildTableStatsResult, Handle);
             HFunc(TEvPrivate::TEvBuildTableStatsError, Handle);
-            hFunc(NKqp::NScheduler::TEvReadFactoryResponse, Handle);
-            hFunc(TEvents::TEvUndelivered, HandleInactive);
         default:
             if (!HandleDefaultEvents(ev, SelfId())) {
                 ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateInactive unhandled event type: " << ev->GetTypeRewrite()
@@ -3462,7 +3454,6 @@ protected:
             HFunc(TEvPrivate::TEvStatisticsScanFinished, Handle);
             HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
             HFunc(TEvDataShard::TEvVacuum, Handle);
-            IgnoreFunc(NKqp::NScheduler::TEvReadFactoryResponse); // ignore self-scheduled fail-safe response from previous stage
             default:
                 if (!HandleDefaultEvents(ev, SelfId())) {
                     ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateWork unhandled event type: " << ev->GetTypeRewrite() << " event: " << ev->ToString());
diff --git a/ydb/core/tx/datashard/ut_read_iterator/datashard_ut_read_iterator_scheduler.cpp b/ydb/core/tx/datashard/ut_read_iterator/datashard_ut_read_iterator_scheduler.cpp
index 0b24c04716e..b7b4092212d 100644
--- a/ydb/core/tx/datashard/ut_read_iterator/datashard_ut_read_iterator_scheduler.cpp
+++ b/ydb/core/tx/datashard/ut_read_iterator/datashard_ut_read_iterator_scheduler.cpp
@@ -16,11 +16,6 @@ const TString TEST_POOL_ID = "test_pool";
 //
 // Parameters:
 //   readLimitMs           — ReadLimit for the test pool (nullopt = unlimited).
-//   blockSchedulerFactory — Drop every TEvGetReadFactory event so the leader
-//                           shard never receives a real factory from the scheduler.
-//                           The constructor then advances simulated time by 2 s to
-//                           let the shard's built-in 1-second fail-safe timer fire,
-//                           completing mediator-state init with a null factory.
 //   withFollower          — Create the table with one follower and route all
 //                           SendRead() calls through a ForceFollower pipe.
 //                           Followers never request a scheduler factory themselves,
@@ -41,7 +36,6 @@ struct TSchedulerTestHelper {
     bool WithFollower = false;
 
     explicit TSchedulerTestHelper(std::optional<ui64> readLimitMs = std::nullopt,
-                                  bool blockSchedulerFactory = false,
                                   bool withFollower = false,
                                   ui32 numShards = 1)
         : WithFollower(withFollower)
@@ -67,19 +61,6 @@ struct TSchedulerTestHelper {
         runtime.SetLogPriority(NKikimrServices::TX_PROXY, NLog::PRI_INFO);
         runtime.SetLogPriority(NKikimrServices::KQP_COMPUTE_SCHEDULER, NLog::PRI_TRACE);
 
-        if (blockSchedulerFactory) {
-            // Drop TEvGetReadFactory so the scheduler never sends a real factory
-            // to the leader shard.  The leader will rely on its 1-second fail-safe
-            // timer instead.  Follower shards never send this event at all, so
-            // dropping it has no effect on their behaviour.
-            runtime.SetObserverFunc([](TAutoPtr<IEventHandle>& ev) {
-                if (ev->GetTypeRewrite() == NKqp::NScheduler::TEvGetReadFactory::EventType) {
-                    return TTestActorRuntimeBase::EEventAction::DROP;
-                }
-                return TTestActorRuntimeBase::EEventAction::PROCESS;
-            });
-        }
-
         {
             auto ev = std::make_unique<TEvAddDatabase>(TEST_DATABASE_ID);
             runtime.Send(MakeKqpSchedulerServiceId(runtime.GetFirstNodeId()), Sender, ev.release());
@@ -122,12 +103,6 @@ struct TSchedulerTestHelper {
                 ClientIds.push_back(LeaderClientIds.back());
             }
         }
-
-        if (blockSchedulerFactory) {
-            // Advance simulated time past the 1-second fail-safe timer so the
-            // leader shard completes mediator-state restoration with a null factory.
-            runtime.SimulateSleep(TDuration::Seconds(2));
-        }
     }
 
     ui64 TabletId() const { return TabletIds[0]; }
@@ -222,7 +197,6 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
     //   quota machinery entirely (SchedulableReadFactory is never set on followers).
     Y_UNIT_TEST_TWIN(ShouldReadWithSchedulerPoolId, WithFollower) {
         TSchedulerTestHelper helper(/*readLimitMs=*/std::nullopt,
-                                    /*blockSchedulerFactory=*/false,
                                     /*withFollower=*/WithFollower);
         helper.Upsert(1, 100);
 
@@ -264,7 +238,6 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
     Y_UNIT_TEST_TWIN(ShouldContinuationReadWithSchedulerPoolId, WithFollower) {
         constexpr ui32 kRows = 5;
         TSchedulerTestHelper helper(/*readLimitMs=*/std::nullopt,
-                                    /*blockSchedulerFactory=*/false,
                                     /*withFollower=*/WithFollower);
         helper.UpsertMany(1, kRows);
 
@@ -298,7 +271,6 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
     // serves the same data successfully, demonstrating the bypass.
     Y_UNIT_TEST_TWIN(ShouldReadWhenQuotaExhausted, WithFollower) {
         TSchedulerTestHelper helper(/*readLimitMs=*/0u,
-                                    /*blockSchedulerFactory=*/false,
                                     /*withFollower=*/WithFollower);
         helper.Upsert(1, 100);
 
@@ -356,7 +328,6 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
     // True for both leader and follower.
     Y_UNIT_TEST_TWIN(ShouldReadWithoutPoolIdUnaffectedByScheduler, WithFollower) {
         TSchedulerTestHelper helper(/*readLimitMs=*/std::nullopt,
-                                    /*blockSchedulerFactory=*/false,
                                     /*withFollower=*/WithFollower);
         helper.Upsert(3, 300);
 
@@ -375,11 +346,10 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
     // When the datashard receives a read request with a pool ID that is not registered
     // in ComputeScheduler, the read must succeed without quota enforcement.
     //
-    // For the leader: SchedulableRead is nullptr, quota check is bypassed.
+    // For the leader: factory returns nullptr for the unknown pool (no read query is registered), so quota check is bypassed.
     // For the follower: quota is never applied regardless of pool registration.
     Y_UNIT_TEST_TWIN(ShouldReadWithUnknownPoolId, WithFollower) {
         TSchedulerTestHelper helper(/*readLimitMs=*/std::nullopt,
-                                    /*blockSchedulerFactory=*/false,
                                     /*withFollower=*/WithFollower);
         helper.Upsert(1, 100);
 
@@ -396,34 +366,6 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler) {
         UNIT_ASSERT_VALUES_EQUAL(cells[3].AsValue<ui32>(), 100u);
     }
 
-    // When the datashard never receives a factory response from the scheduler
-    // (e.g. the scheduler is temporarily unavailable), it falls back to reading
-    // without quota after the built-in 1-second fail-safe timer fires.
-    // Reads with a PoolId must still succeed on both leader and follower.
-    //
-    // Note: followers never request a factory in the first place, so they are
-    // always in this "no-factory" state.  The blockSchedulerFactory flag here
-    // affects only the leader; it verifies that the leader's fallback path also
-    // works correctly.
-    Y_UNIT_TEST_TWIN(ShouldReadSuccessfullyWhenSchedulerDoesNotRespond, WithFollower) {
-        TSchedulerTestHelper helper(/*readLimitMs=*/std::nullopt,
-                                    /*blockSchedulerFactory=*/true,
-                                    /*withFollower=*/WithFollower);
-        helper.Upsert(1, 100);
-
-        auto request = helper.MakeReadRequest(1);
-        AddKeyQuery(*request, {1, 1, 1});
-
-        auto result = helper.SendRead(request.release());
-        UNIT_ASSERT_C(result, "Expected read to succeed when scheduler did not respond");
-        UNIT_ASSERT_VALUES_EQUAL(result->Record.GetStatus().GetCode(), Ydb::StatusIds::SUCCESS);
-        UNIT_ASSERT_VALUES_EQUAL(result->GetRowsCount(), 1);
-
-        auto cells = result->GetCells(0);
-        UNIT_ASSERT_VALUES_EQUAL(cells.size(), 4u);
-        UNIT_ASSERT_VALUES_EQUAL(cells[3].AsValue<ui32>(), 100u);
-    }
-
 } // Y_UNIT_TEST_SUITE(DataShardReadIteratorScheduler)
 
 } // namespace NKikimr
diff --git a/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp b/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp
index dc86ee89dbe..c6bdaf84ccd 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp
+++ b/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp
@@ -9,6 +9,7 @@
 #include <ydb/core/kqp/common/simple/services.h>
 #include <ydb/core/kqp/proxy_service/kqp_proxy_service.h>
 #include <ydb/core/kqp/rm_service/kqp_rm_service.h>
+#include <ydb/core/kqp/runtime/scheduler/kqp_compute_scheduler_service.h>
 #include <ydb/core/metering/metering.h>
 #include <ydb/core/protos/schemeshard/operations.pb.h>
 #include <ydb/core/tablet_flat/tablet_flat_executed.h>
@@ -583,6 +584,13 @@ void SetupKqpProxy(TTestActorRuntime& runtime, ui32 nodeIdx) {
     NKikimrConfig::TTableServiceConfig tableServiceConfig;
     SetupKqpResourceManager(runtime, tableServiceConfig, nodeIdx);
 
+    // Used by KqpComputeSchedulerService
+    {
+        NKikimrConfig::TAppConfig appConfig;
+        auto counters = MakeIntrusive<::NMonitoring::TDynamicCounters>();
+        runtime.GetAppData(nodeIdx).KqpComputeScheduler = NKqp::CreateKqpComputeScheduler(counters, appConfig);
+    }
+
     NKikimrConfig::TLogConfig logConfig;
     NKikimrConfig::TQueryServiceConfig queryServiceConfig;
     auto federatedQuerySetupFactory = std::make_shared<NKqp::TKqpFederatedQuerySetupFactoryNoop>();