diff --git a/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp b/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp
index 8546a6c12431..427d958a1bc7 100644
--- a/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp
+++ b/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp
@@ -14,15 +14,13 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
         , std::shared_ptr<IKqpNodeState> state
         , TIntrusivePtr<NRm::TTxState> tx
         , TIntrusivePtr<NRm::TTaskState> task
-        , ui64 limit
-        , ui64 reasonableSpillingTreshold)
+        , ui64 limit)
     : NYql::NDq::TGuaranteeQuotaManager(limit, limit)
     , ResourceManager(std::move(resourceManager))
     , MemoryPool(memoryPool)
     , State(std::move(state))
     , Tx(std::move(tx))
     , Task(std::move(task))
-    , ReasonableSpillingTreshold(reasonableSpillingTreshold)
     {
     }
 
@@ -57,7 +55,7 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
     }
 
     bool IsReasonableToUseSpilling() const override {
-        return Tx->GetExtraMemoryAllocatedSize() >= ReasonableSpillingTreshold;
+        return Task->IsReasonableToStartSpilling();
     }
 
     TString MemoryConsumptionDetails() const override {
@@ -88,7 +86,6 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
     std::atomic<ui64> MkqlLightProgramMemoryLimit = 0;
     std::atomic<ui64> MkqlHeavyProgramMemoryLimit = 0;
     std::atomic<ui64> MinChannelBufferSize = 0;
-    std::atomic<ui64> ReasonableSpillingTreshold = 0;
 
 public:
     TKqpCaFactory(const NKikimrConfig::TTableServiceConfig::TResourceManager& config,
@@ -107,7 +104,6 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
         MkqlLightProgramMemoryLimit.store(config.GetMkqlLightProgramMemoryLimit());
         MkqlHeavyProgramMemoryLimit.store(config.GetMkqlHeavyProgramMemoryLimit());
         MinChannelBufferSize.store(config.GetMinChannelBufferSize());
-        ReasonableSpillingTreshold.store(config.GetReasonableSpillingTreshold());
     }
 
     TActorStartResult CreateKqpComputeActor(TCreateArgs&& args) override {
@@ -158,8 +154,7 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
             std::move(args.State),
             std::move(args.TxInfo),
             std::move(task),
-            limit,
-            ReasonableSpillingTreshold.load());
+            limit);
 
         auto runtimeSettings = args.RuntimeSettings;
         runtimeSettings.ExtraMemoryAllocationPool = args.MemoryPool;
diff --git a/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.h b/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.h
index 2157deffb0ab..034777781251 100644
--- a/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.h
+++ b/ydb/core/kqp/compute_actor/kqp_compute_actor_factory.h
@@ -122,6 +122,7 @@ struct IKqpNodeComputeActorFactory {
         const TInstant& Deadline;
         const bool ShareMailbox;
         const TMaybe<NYql::NDqProto::TRlPath>& RlPath;
+
         TComputeStagesWithScan* ComputesByStages = nullptr;
         std::shared_ptr<IKqpNodeState> State = nullptr;
         TComputeActorSchedulingOptions SchedulingOptions = {};
diff --git a/ydb/core/kqp/executer_actor/kqp_planner.cpp b/ydb/core/kqp/executer_actor/kqp_planner.cpp
index 0019ba5208d4..b3469b39c1b9 100644
--- a/ydb/core/kqp/executer_actor/kqp_planner.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_planner.cpp
@@ -223,6 +223,7 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
     }
 
     request.SetSchedulerGroup(UserRequestContext->PoolId);
+    request.SetMemoryPoolPercent(UserRequestContext->PoolConfig->QueryMemoryLimitPercentPerNode);
 
     return result;
 }
@@ -351,7 +352,8 @@ TString TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, ui32 computeTasksSize)
     NYql::NDq::TComputeRuntimeSettings settings;
     if (!TxInfo) {
         TxInfo = MakeIntrusive<NRm::TTxState>(
-            TxId, TInstant::Now(), ResourceManager_->GetCounters());
+            TxId, TInstant::Now(), ResourceManager_->GetCounters(),
+            UserRequestContext->PoolId, UserRequestContext->PoolConfig->QueryMemoryLimitPercentPerNode);
     }
 
     auto startResult = CaFactory_->CreateKqpComputeActor({
@@ -370,7 +372,7 @@ TString TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, ui32 computeTasksSize)
         .StatsMode = GetDqStatsMode(StatsMode),
         .Deadline = Deadline,
         .ShareMailbox = (computeTasksSize <= 1),
-        .RlPath = Nothing()
+        .RlPath = Nothing(),
     });
 
     if (const auto* rmResult = std::get_if<NRm::TKqpRMAllocateResult>(&startResult)) {
diff --git a/ydb/core/kqp/node_service/kqp_node_service.cpp b/ydb/core/kqp/node_service/kqp_node_service.cpp
index 30f1512893ec..37614b7ad080 100644
--- a/ydb/core/kqp/node_service/kqp_node_service.cpp
+++ b/ydb/core/kqp/node_service/kqp_node_service.cpp
@@ -205,7 +205,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
         }
 
         TIntrusivePtr<NRm::TTxState> txInfo = MakeIntrusive<NRm::TTxState>(
-            txId, TInstant::Now(), ResourceManager_->GetCounters());
+            txId, TInstant::Now(), ResourceManager_->GetCounters(),
+            msg.GetSchedulerGroup(), msg.GetMemoryPoolPercent());
 
         const ui32 tasksCount = msg.GetTasks().size();
         for (auto& dqTask: *msg.MutableTasks()) {
@@ -246,7 +247,7 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
                 .RlPath = rlPath,
                 .ComputesByStages = &computesByStage,
                 .State = State_,
-                .SchedulingOptions = std::move(schedulingOptions)
+                .SchedulingOptions = std::move(schedulingOptions),
             });
 
             if (const auto* rmResult = std::get_if<NRm::TKqpRMAllocateResult>(&result)) {
diff --git a/ydb/core/kqp/rm_service/kqp_rm_service.cpp b/ydb/core/kqp/rm_service/kqp_rm_service.cpp
index baf31d6bf8e7..c8049f19c939 100644
--- a/ydb/core/kqp/rm_service/kqp_rm_service.cpp
+++ b/ydb/core/kqp/rm_service/kqp_rm_service.cpp
@@ -20,6 +20,8 @@
 
 #include <ydb/library/yql/utils/yql_panic.h>
 
+#include <library/cpp/containers/absl_flat_hash/flat_hash_map.h>
+
 namespace NKikimr {
 namespace NKqp {
 namespace NRm {
@@ -43,42 +45,79 @@ using namespace NResourceBroker;
 
 namespace {
 
-template <typename T>
-class TLimitedResource {
+static constexpr double MYEPS = 1e-9;
+
+ui64 OverPercentage(ui64 limit, double percent) {
+    return static_cast<double>(limit) / 100 * (100 - percent) + MYEPS;
+}
+
+ui64 Percentage(ui64 limit, double percent) {
+    return static_cast<double>(limit) / 100 * percent + MYEPS;
+}
+
+class TMemoryResource : public TAtomicRefCount<TMemoryResource> {
 public:
-    explicit TLimitedResource(T limit)
-        : Limit(limit)
-        , Used(0) {}
+    explicit TMemoryResource(ui64 baseLimit, double memoryPoolPercent, double overPercent)
+        : BaseLimit(baseLimit)
+        , Used(0)
+        , MemoryPoolPercent(memoryPoolPercent)
+        , OverPercent(overPercent)
+        , SpillingCookie(MakeIntrusive<TMemoryResourceCookie>())
+    {
+        SetActualLimits();
+    }
 
-    T Available() const {
+    ui64 Available() const {
         return Limit > Used ? Limit - Used : 0;
     }
 
-    bool Has(T amount) const {
+    bool Has(ui64 amount) const {
         return Available() >= amount;
     }
 
-    bool Acquire(T value) {
+    bool AcquireIfAvailable(ui64 value) {
         if (Available() >= value) {
             Used += value;
+            UpdateCookie();
             return true;
         }
         return false;
     }
 
-    void Release(T value) {
+    TIntrusivePtr<TMemoryResourceCookie> GetSpillingCookie() const {
+        return SpillingCookie;
+    }
+
+    void UpdateCookie() {
+        SpillingCookie->SpillingPercentReached.store(Available() < OverLimit);
+    }
+
+    void Release(ui64 value) {
         if (Used > value) {
             Used -= value;
         } else {
             Used = 0;
         }
+
+        UpdateCookie();
+    }
+
+    void SetNewLimit(ui64 baseLimit, double memoryPoolPercent, double overPercent) {
+        if (abs(memoryPoolPercent - MemoryPoolPercent) < MYEPS && baseLimit != BaseLimit)
+            return;
+
+        BaseLimit = baseLimit;
+        MemoryPoolPercent = memoryPoolPercent;
+        OverPercent = overPercent;
+        SetActualLimits();
     }
 
-    void SetNewLimit(T limit) {
-        Limit = limit;
+    void SetActualLimits() {
+        Limit = Percentage(BaseLimit, MemoryPoolPercent);
+        OverLimit = OverPercentage(Limit, OverPercent);
     }
 
-    T GetLimit() const {
+    ui64 GetLimit() const {
         return Limit;
     }
 
@@ -87,8 +126,14 @@ class TLimitedResource {
     }
 
 private:
-    T Limit;
-    T Used;
+    ui64 BaseLimit;
+    ui64 OverLimit;
+    ui64 Limit;
+    ui64 Used;
+    double MemoryPoolPercent;
+    double OverPercent;
+
+    TIntrusivePtr<TMemoryResourceCookie> SpillingCookie;
 };
 
 struct TEvPrivate {
@@ -112,7 +157,8 @@ class TKqpResourceManager : public IKqpResourceManager {
         : Counters(counters)
         , ExecutionUnitsResource(config.GetComputeActorsCount())
         , ExecutionUnitsLimit(config.GetComputeActorsCount())
-        , ScanQueryMemoryResource(config.GetQueryMemoryLimit())
+        , SpillingPercent(config.GetSpillingPercent())
+        , TotalMemoryResource(MakeIntrusive<TMemoryResource>(config.GetQueryMemoryLimit(), (double)100, config.GetSpillingPercent()))
     {
         SetConfigValues(config);
     }
@@ -188,11 +234,7 @@ class TKqpResourceManager : public IKqpResourceManager {
         }
 
         bool hasScanQueryMemory = true;
-        ui64 queryMemoryLimit = 0;
 
-        // NOTE(gvit): the first memory request always satisfied.
-        // all other requests are not guaranteed to be satisfied.
-        // In the nearest future we need to implement several layers of memory requests.
         bool isFirstAllocationRequest = (resources.ExecutionUnits > 0 && resources.MemoryPool == EKqpMemoryPool::DataQuery);
         if (isFirstAllocationRequest) {
             TKqpResourcesRequest newRequest = resources;
@@ -210,17 +252,33 @@ class TKqpResourceManager : public IKqpResourceManager {
                 return result;
             }
 
-            hasScanQueryMemory = ScanQueryMemoryResource.Has(resources.Memory);
-            if (hasScanQueryMemory) {
-                ScanQueryMemoryResource.Acquire(resources.Memory);
-                queryMemoryLimit = QueryMemoryLimit.load();
+            hasScanQueryMemory = TotalMemoryResource->AcquireIfAvailable(resources.Memory);
+            task->TotalMemoryCookie = TotalMemoryResource->GetSpillingCookie();
+
+            if (hasScanQueryMemory && !tx->PoolId.empty() && tx->MemoryPoolPercent > 0) {
+                auto [it, success] = MemoryNamedPools.emplace(tx->PoolId, nullptr);
+
+                if (success) {
+                    it->second = MakeIntrusive<TMemoryResource>(TotalMemoryResource->GetLimit(), tx->MemoryPoolPercent, SpillingPercent.load());
+                } else {
+                    it->second->SetNewLimit(TotalMemoryResource->GetLimit(), tx->MemoryPoolPercent, SpillingPercent.load());
+                }
+
+                auto& poolMemory = it->second;
+                if (!poolMemory->AcquireIfAvailable(resources.Memory)) {
+                    hasScanQueryMemory = false;
+                    TotalMemoryResource->Release(resources.Memory);
+                }
+
+                task->PoolMemoryCookie = poolMemory->GetSpillingCookie();
             }
-        } // with_lock (Lock)
+        }
 
         if (!hasScanQueryMemory) {
             Counters->RmNotEnoughMemory->Inc();
             TStringBuilder reason;
-            reason << "TxId: " << txId << ", taskId: " << taskId << ". Not enough memory for query, requested: " << resources.Memory;
+            reason << "TxId: " << txId << ", taskId: " << taskId << ". Not enough memory for query, requested: " << resources.Memory
+                << ". " << tx->ToString();
             result.SetError(NKikimrKqp::TEvStartKqpTasksResponse::NOT_ENOUGH_MEMORY, reason);
             return result;
         }
@@ -232,28 +290,23 @@ class TKqpResourceManager : public IKqpResourceManager {
             if (!result) {
                 Counters->RmNotEnoughMemory->Inc();
                 with_lock (Lock) {
-                    ScanQueryMemoryResource.Release(resources.Memory);
-                } // with_lock (Lock)
+                    TotalMemoryResource->Release(resources.Memory);
+                    auto it = MemoryNamedPools.find(tx->PoolId);
+                    if (it != MemoryNamedPools.end()) {
+                        it->second->Release(resources.Memory);
+                    }
+                }
             }
         };
 
-        ui64 txTotalRequestedMemory = tx->GetExtraMemoryAllocatedSize() + resources.Memory;
-        if (txTotalRequestedMemory > queryMemoryLimit) {
-            TStringBuilder reason;
-            reason << "TxId: " << txId << ", taskId: " << taskId << ". Query memory limit exceeded: "
-                << "requested " << txTotalRequestedMemory;
-            result.SetError(NKikimrKqp::TEvStartKqpTasksResponse::QUERY_MEMORY_LIMIT_EXCEEDED, reason);
-            return result;
-        }
-
         bool allocated = ResourceBroker->SubmitTaskInstant(
             TEvResourceBroker::TEvSubmitTask(rbTaskId, rbTaskName, {0, resources.Memory}, "kqp_query", 0, {}),
             SelfId);
 
         if (!allocated) {
             TStringBuilder reason;
-            reason << "TxId: " << txId << ", taskId: " << taskId << ". Not enough ScanQueryMemory: "
-                << "requested " << resources.Memory;
+            reason << "TxId: " << txId << ", taskId: " << taskId << ". Not enough memory for query, requested: " << resources.Memory
+                << ". " << tx->ToString();
             LOG_AS_N(reason);
             result.SetError(NKikimrKqp::TEvStartKqpTasksResponse::NOT_ENOUGH_MEMORY, reason);
             return result;
@@ -302,14 +355,19 @@ class TKqpResourceManager : public IKqpResourceManager {
 
         if (resources.Memory > 0) {
             with_lock (Lock) {
-                ScanQueryMemoryResource.Release(resources.Memory);
-            } // with_lock (Lock)
+                TotalMemoryResource->Release(resources.Memory);
+                auto it = MemoryNamedPools.find(tx->PoolId);
+                if (it != MemoryNamedPools.end()) {
+                    it->second->Release(resources.Memory);
+                }
+            }
         }
 
-        LOG_AS_D("TxId: " << tx->TxId << ", taskId: " << task->TaskId << ". Released resources, "
-            << "ScanQueryMemory: " << resources.Memory << ", "
-            << "ExternalDataQueryMemory " << resources.ExternalMemory << ", "
-            << "ExecutionUnits " << resources.ExecutionUnits << ".");
+        LOG_AS_D("TxId: " << tx->TxId << ", taskId: " << task->TaskId
+            << ". Released resources, "
+            << "Memory: " << resources.Memory << ", "
+            << "Free Tier: " << resources.ExternalMemory << ", "
+            << "ExecutionUnits: " << resources.ExecutionUnits << ".");
 
         FireResourcesPublishing();
     }
@@ -346,7 +404,7 @@ class TKqpResourceManager : public IKqpResourceManager {
 
         with_lock (Lock) {
             result.ExecutionUnits = ExecutionUnitsResource.load();
-            result.Memory[EKqpMemoryPool::ScanQuery] = ScanQueryMemoryResource.Available();
+            result.Memory[EKqpMemoryPool::ScanQuery] = TotalMemoryResource->Available();
         }
 
         return result;
@@ -391,6 +449,7 @@ class TKqpResourceManager : public IKqpResourceManager {
         MinChannelBufferSize.store(config.GetMinChannelBufferSize());
         MaxTotalChannelBuffersSize.store(config.GetMaxTotalChannelBuffersSize());
         QueryMemoryLimit.store(config.GetQueryMemoryLimit());
+        SpillingPercent.store(config.GetSpillingPercent());
     }
 
     ui32 GetNodeId() override {
@@ -435,7 +494,8 @@ class TKqpResourceManager : public IKqpResourceManager {
     // limits (guarded by Lock)
     std::atomic<i32> ExecutionUnitsResource;
     std::atomic<i32> ExecutionUnitsLimit;
-    TLimitedResource<ui64> ScanQueryMemoryResource;
+    std::atomic<double> SpillingPercent;
+    TIntrusivePtr<TMemoryResource> TotalMemoryResource;
     std::atomic<i64> ExternalDataQueryMemory = 0;
 
     // current state
@@ -448,6 +508,8 @@ class TKqpResourceManager : public IKqpResourceManager {
     // state for resource info exchanger
     std::shared_ptr<TResourceSnapshotState> ResourceSnapshotState;
     TActorId ResourceInfoExchanger = TActorId();
+
+    absl::flat_hash_map<TString, TIntrusivePtr<TMemoryResource>> MemoryNamedPools;
 };
 
 struct TResourceManagers {
@@ -594,7 +656,7 @@ class TKqpResourceManagerActor : public TActorBootstrapped<TKqpResourceManagerAc
 
         if (queueConfig.GetLimit().GetMemory() > 0) {
             with_lock (ResourceManager->Lock) {
-                ResourceManager->ScanQueryMemoryResource.SetNewLimit(queueConfig.GetLimit().GetMemory());
+                ResourceManager->TotalMemoryResource->SetNewLimit(queueConfig.GetLimit().GetMemory(), (double)100, ResourceManager->SpillingPercent.load());
             }
             LOG_I("Total node memory for scan queries: " << queueConfig.GetLimit().GetMemory() << " bytes");
         }
@@ -711,7 +773,7 @@ class TKqpResourceManagerActor : public TActorBootstrapped<TKqpResourceManagerAc
             PRE() {
                 str << "State storage key: " << WbState.Tenant << Endl;
                 with_lock (ResourceManager->Lock) {
-                    str << "ScanQuery memory resource: " << ResourceManager->ScanQueryMemoryResource.ToString() << Endl;
+                    str << "ScanQuery memory resource: " << ResourceManager->TotalMemoryResource->ToString() << Endl;
                     str << "External DataQuery memory: " << ResourceManager->ExternalDataQueryMemory.load() << Endl;
                     str << "ExecutionUnits resource: " << ResourceManager->ExecutionUnitsResource.load() << Endl;
                 }
@@ -806,13 +868,13 @@ class TKqpResourceManagerActor : public TActorBootstrapped<TKqpResourceManagerAc
         ActorIdToProto(MakeKqpResourceManagerServiceID(SelfId().NodeId()), payload.MutableResourceManagerActorId()); // legacy
         with_lock (ResourceManager->Lock) {
             payload.SetAvailableComputeActors(ResourceManager->ExecutionUnitsResource.load()); // legacy
-            payload.SetTotalMemory(ResourceManager->ScanQueryMemoryResource.GetLimit()); // legacy
-            payload.SetUsedMemory(ResourceManager->ScanQueryMemoryResource.GetLimit() - ResourceManager->ScanQueryMemoryResource.Available()); // legacy
+            payload.SetTotalMemory(ResourceManager->TotalMemoryResource->GetLimit()); // legacy
+            payload.SetUsedMemory(ResourceManager->TotalMemoryResource->GetLimit() - ResourceManager->TotalMemoryResource->Available()); // legacy
 
             payload.SetExecutionUnits(ResourceManager->ExecutionUnitsResource.load());
             auto* pool = payload.MutableMemory()->Add();
             pool->SetPool(EKqpMemoryPool::ScanQuery);
-            pool->SetAvailable(ResourceManager->ScanQueryMemoryResource.Available());
+            pool->SetAvailable(ResourceManager->TotalMemoryResource->Available());
         }
 
         LOG_I("Send to publish resource usage for "
diff --git a/ydb/core/kqp/rm_service/kqp_rm_service.h b/ydb/core/kqp/rm_service/kqp_rm_service.h
index 78b519969f02..bdddf3078e33 100644
--- a/ydb/core/kqp/rm_service/kqp_rm_service.h
+++ b/ydb/core/kqp/rm_service/kqp_rm_service.h
@@ -55,6 +55,11 @@ struct TKqpResourcesRequest {
 
 class TTxState;
 
+class TMemoryResourceCookie : public TAtomicRefCount<TMemoryResourceCookie> {
+public:
+    std::atomic<bool> SpillingPercentReached{false};
+};
+
 class TTaskState : public TAtomicRefCount<TTaskState> {
     friend TTxState;
 
@@ -65,6 +70,8 @@ class TTaskState : public TAtomicRefCount<TTaskState> {
     ui64 ExternalDataQueryMemory = 0;
     ui64 ResourceBrokerTaskId = 0;
     ui32 ExecutionUnits = 0;
+    TIntrusivePtr<TMemoryResourceCookie> TotalMemoryCookie;
+    TIntrusivePtr<TMemoryResourceCookie> PoolMemoryCookie;
 
 public:
 
@@ -80,6 +87,11 @@ class TTaskState : public TAtomicRefCount<TTaskState> {
         return resources;
     }
 
+    bool IsReasonableToStartSpilling() {
+        return (PoolMemoryCookie && PoolMemoryCookie->SpillingPercentReached.load())
+            || (TotalMemoryCookie && TotalMemoryCookie->SpillingPercentReached.load());
+    }
+
     TKqpResourcesRequest FreeResourcesRequest() const {
         return TKqpResourcesRequest{
             .ExecutionUnits=ExecutionUnits,
@@ -101,21 +113,28 @@ class TTxState : public TAtomicRefCount<TTxState> {
     const ui64 TxId;
     const TInstant CreatedAt;
     TIntrusivePtr<TKqpCounters> Counters;
+    const TString PoolId;
+    const double MemoryPoolPercent;
+
 private:
     std::atomic<ui64> TxScanQueryMemory = 0;
     std::atomic<ui64> TxExternalDataQueryMemory = 0;
     std::atomic<ui32> TxExecutionUnits = 0;
 
 public:
-    explicit TTxState(ui64 txId, TInstant now, TIntrusivePtr<TKqpCounters> counters)
+    explicit TTxState(ui64 txId, TInstant now, TIntrusivePtr<TKqpCounters> counters, const TString& poolId, const double memoryPoolPercent)
         : TxId(txId)
         , CreatedAt(now)
         , Counters(std::move(counters))
+        , PoolId(poolId)
+        , MemoryPoolPercent(memoryPoolPercent)
     {}
 
     TString ToString() const {
         return TStringBuilder() << "TxResourcesInfo{ "
             << "TxId: " << TxId
+            << ", PoolId: " << PoolId
+            << ", MemoryPoolPercent: " << MemoryPoolPercent
             << ", memory initially granted resources: " << TxExternalDataQueryMemory.load()
             << ", extra allocations " << TxScanQueryMemory.load()
             << ", execution units: " << TxExecutionUnits.load()
diff --git a/ydb/core/kqp/rm_service/kqp_rm_ut.cpp b/ydb/core/kqp/rm_service/kqp_rm_ut.cpp
index 9c8dc4d318ef..c7053ab08731 100644
--- a/ydb/core/kqp/rm_service/kqp_rm_ut.cpp
+++ b/ydb/core/kqp/rm_service/kqp_rm_ut.cpp
@@ -184,7 +184,7 @@ class KqpRm : public TTestBase {
     }
 
     TIntrusivePtr<NRm::TTxState> MakeTx(ui64 txId, std::shared_ptr<NRm::IKqpResourceManager> rm) {
-        return MakeIntrusive<NRm::TTxState>(txId, TInstant::Now(), rm->GetCounters());
+        return MakeIntrusive<NRm::TTxState>(txId, TInstant::Now(), rm->GetCounters(), "", (double)100);
     }
 
     TIntrusivePtr<NRm::TTaskState> MakeTask(ui64 taskId, TIntrusivePtr<NRm::TTxState> tx) {
diff --git a/ydb/core/kqp/ut/query/kqp_limits_ut.cpp b/ydb/core/kqp/ut/query/kqp_limits_ut.cpp
index d6ae4c949b3a..3816e9562e6c 100644
--- a/ydb/core/kqp/ut/query/kqp_limits_ut.cpp
+++ b/ydb/core/kqp/ut/query/kqp_limits_ut.cpp
@@ -146,7 +146,7 @@ Y_UNIT_TEST_SUITE(KqpLimits) {
         )"), TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         result.GetIssues().PrintTo(Cerr);
 
-        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::OVERLOADED);
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::PRECONDITION_FAILED);
     }
 
     Y_UNIT_TEST(DatashardProgramSize) {
diff --git a/ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp b/ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp
index 686e3e08ca3d..691c3d308f19 100644
--- a/ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp
+++ b/ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp
@@ -32,13 +32,13 @@ NKikimrConfig::TAppConfig AppCfg() {
     return appCfg;
 }
 
-NKikimrConfig::TAppConfig AppCfgLowComputeLimits(ui64 reasonableTreshold) {
+NKikimrConfig::TAppConfig AppCfgLowComputeLimits(double reasonableTreshold) {
     NKikimrConfig::TAppConfig appCfg;
 
     auto* rm = appCfg.MutableTableServiceConfig()->MutableResourceManager();
     rm->SetMkqlLightProgramMemoryLimit(100);
     rm->SetMkqlHeavyProgramMemoryLimit(300);
-    rm->SetReasonableSpillingTreshold(reasonableTreshold);
+    rm->SetSpillingPercent(reasonableTreshold);
     appCfg.MutableTableServiceConfig()->SetEnableQueryServiceSpilling(true);
 
     auto* spilling = appCfg.MutableTableServiceConfig()->MutableSpillingServiceConfig()->MutableLocalFileConfig();
@@ -73,7 +73,7 @@ Y_UNIT_TEST(SpillingPragmaParseError) {
 }
 
 Y_UNIT_TEST_TWIN(SpillingInRuntimeNodes, EnabledSpilling) {
-    ui64 reasonableTreshold = EnabledSpilling ? 100 : 200_MB;
+    double reasonableTreshold = EnabledSpilling ? 0.01 : 100;
     Cerr << "cwd: " << NFs::CurrentWorkingDirectory() << Endl;
     TKikimrRunner kikimr(AppCfgLowComputeLimits(reasonableTreshold));
 
diff --git a/ydb/core/protos/kqp.proto b/ydb/core/protos/kqp.proto
index 9e89d926a065..d7893686f110 100644
--- a/ydb/core/protos/kqp.proto
+++ b/ydb/core/protos/kqp.proto
@@ -550,6 +550,7 @@ message TEvStartKqpTasksRequest {
     optional uint64 OutputChunkMaxSize = 7 [default = 0]; // 0 - use some default value
     optional string SerializedGUCSettings = 8;
     optional string SchedulerGroup = 9;
+    optional double MemoryPoolPercent = 10 [default = 100];
 }
 
 message TEvStartKqpTasksResponse {
diff --git a/ydb/core/protos/table_service_config.proto b/ydb/core/protos/table_service_config.proto
index dfeed653cda3..68ed40acd5e3 100644
--- a/ydb/core/protos/table_service_config.proto
+++ b/ydb/core/protos/table_service_config.proto
@@ -42,7 +42,7 @@ message TTableServiceConfig {
         optional TInfoExchangerSettings InfoExchangerSettings = 19;
         optional uint64 KqpPatternCachePatternAccessTimesBeforeTryToCompile = 20 [default = 5];
         optional uint64 KqpPatternCacheCompiledCapacityBytes = 21 [default = 104857600]; // 100 MiB
-        optional uint64 ReasonableSpillingTreshold = 22 [default = 104857600]; // 100 MiB
+        optional double SpillingPercent = 22 [default = 80]; // 100 MiB
     }
 
     message TSpillingServiceConfig {
@@ -315,6 +315,6 @@ message TTableServiceConfig {
         }
     };
     optional TComputePoolConfiguration PoolsConfiguration = 68;
-    
+
     optional bool EnableRowsDuplicationCheck = 69 [ default = false ];
 };