[kqp] убрать PackFrom/Unpack для настроек чтения из source KIKIMR-19032

KIKIMR-19032

12 files changed, 442 insertions, 334 deletions

diff --git a/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.cpp b/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.cpp
index b3c639e6696..2e907093de3 100644
--- a/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.cpp
+++ b/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.cpp
@@ -1,313 +1,274 @@
-#include "kqp_compute_actor.h"
-#include "kqp_compute_actor_impl.h"
-
-#include <ydb/core/base/appdata.h>
-#include <ydb/core/protos/tx_datashard.pb.h>
-#include <ydb/core/kqp/rm_service/kqp_rm_service.h>
-#include <ydb/core/kqp/runtime/kqp_compute.h>
-#include <ydb/core/kqp/runtime/kqp_scan_data.h>
-#include <ydb/core/sys_view/scan.h>
-#include <ydb/library/yverify_stream/yverify_stream.h>
-
-#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h>
-
+#include "kqp_pure_compute_actor.h"
 
 namespace NKikimr {
 namespace NKqp {
 
-namespace {
-
-using namespace NYql;
-using namespace NYql::NDq;
-
-inline bool IsDebugLogEnabled(const TActorSystem* actorSystem) {
+bool TKqpComputeActor::IsDebugLogEnabled(const TActorSystem* actorSystem) {
     auto* settings = actorSystem->LoggerSettings();
     return settings && settings->Satisfies(NActors::NLog::EPriority::PRI_DEBUG, NKikimrServices::KQP_TASKS_RUNNER);
 }
 
-class TKqpComputeActor : public TDqComputeActorBase<TKqpComputeActor> {
-    using TBase = TDqComputeActorBase<TKqpComputeActor>;
-
-public:
-    static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
-        return NKikimrServices::TActivity::KQP_COMPUTE_ACTOR;
-    }
-
-    TKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
-        IDqAsyncIoFactory::TPtr asyncIoFactory,
-        const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry,
-        const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
-        NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena)
-        : TBase(executerId, txId, task, std::move(asyncIoFactory), functionRegistry, settings, memoryLimits, /* ownMemoryQuota = */ true, /* passExceptions = */ true, /*taskCounters = */ nullptr, std::move(traceId), std::move(arena))
-        , ComputeCtx(settings.StatsMode)
-    {
-        if (GetTask().GetMeta().Is<NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta>()) {
-            Meta.ConstructInPlace();
-            YQL_ENSURE(GetTask().GetMeta().UnpackTo(Meta.Get()), "Invalid task meta: " << GetTask().GetMeta().DebugString());
-            YQL_ENSURE(Meta->GetReads().size() == 1);
-            YQL_ENSURE(!Meta->GetReads()[0].GetKeyRanges().empty());
-            YQL_ENSURE(!Meta->GetTable().GetSysViewInfo().empty());
-        }
+TKqpComputeActor::TKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
+    IDqAsyncIoFactory::TPtr asyncIoFactory,
+    const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry,
+    const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
+    NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena)
+    : TBase(executerId, txId, task, std::move(asyncIoFactory), functionRegistry, settings, memoryLimits, /* ownMemoryQuota = */ true, /* passExceptions = */ true, /*taskCounters = */ nullptr, std::move(traceId), std::move(arena))
+    , ComputeCtx(settings.StatsMode)
+{
+    if (GetTask().GetMeta().Is<NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta>()) {
+        Meta.ConstructInPlace();
+        YQL_ENSURE(GetTask().GetMeta().UnpackTo(Meta.Get()), "Invalid task meta: " << GetTask().GetMeta().DebugString());
+        YQL_ENSURE(Meta->GetReads().size() == 1);
+        YQL_ENSURE(!Meta->GetReads()[0].GetKeyRanges().empty());
+        YQL_ENSURE(!Meta->GetTable().GetSysViewInfo().empty());
     }
+}
 
-    void DoBootstrap() {
-        const TActorSystem* actorSystem = TlsActivationContext->ActorSystem();
-
-        TLogFunc logger;
-        if (IsDebugLogEnabled(actorSystem)) {
-            logger = [actorSystem, txId = this->GetTxId(), taskId = GetTask().GetId()] (const TString& message) {
-                LOG_DEBUG_S(*actorSystem, NKikimrServices::KQP_TASKS_RUNNER, "TxId: " << txId
-                    << ", task: " << taskId << ": " << message);
-            };
-        }
+void TKqpComputeActor::DoBootstrap() {
+    const TActorSystem* actorSystem = TlsActivationContext->ActorSystem();
 
-        TDqTaskRunnerContext execCtx;
+    TLogFunc logger;
+    if (IsDebugLogEnabled(actorSystem)) {
+        logger = [actorSystem, txId = this->GetTxId(), taskId = GetTask().GetId()] (const TString& message) {
+            LOG_DEBUG_S(*actorSystem, NKikimrServices::KQP_TASKS_RUNNER, "TxId: " << txId
+                << ", task: " << taskId << ": " << message);
+        };
+    }
 
-        execCtx.FuncRegistry = AppData()->FunctionRegistry;
-        execCtx.RandomProvider = TAppData::RandomProvider.Get();
-        execCtx.TimeProvider = TAppData::TimeProvider.Get();
-        execCtx.ComputeCtx = &ComputeCtx;
-        execCtx.ComputationFactory = NMiniKQL::GetKqpActorComputeFactory(&ComputeCtx);
-        execCtx.ApplyCtx = nullptr;
-        execCtx.Alloc = nullptr;
-        execCtx.TypeEnv = nullptr;
-        execCtx.PatternCache = GetKqpResourceManager()->GetPatternCache();
+    TDqTaskRunnerContext execCtx;
 
-        TDqTaskRunnerSettings settings;
-        settings.CollectBasicStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_BASIC;
-        settings.CollectProfileStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_PROFILE;
+    execCtx.FuncRegistry = AppData()->FunctionRegistry;
+    execCtx.RandomProvider = TAppData::RandomProvider.Get();
+    execCtx.TimeProvider = TAppData::TimeProvider.Get();
+    execCtx.ComputeCtx = &ComputeCtx;
+    execCtx.ComputationFactory = NMiniKQL::GetKqpActorComputeFactory(&ComputeCtx);
+    execCtx.ApplyCtx = nullptr;
+    execCtx.Alloc = nullptr;
+    execCtx.TypeEnv = nullptr;
+    execCtx.PatternCache = GetKqpResourceManager()->GetPatternCache();
 
-        settings.OptLLVM = (GetTask().HasUseLlvm() && GetTask().GetUseLlvm()) ? "--compile-options=disable-opt" : "OFF";
-        settings.UseCacheForLLVM = AppData()->FeatureFlags.GetEnableLLVMCache();
+    TDqTaskRunnerSettings settings;
+    settings.CollectBasicStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_BASIC;
+    settings.CollectProfileStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_PROFILE;
 
-        for (const auto& [paramsName, paramsValue] : GetTask().GetTaskParams()) {
-            settings.TaskParams[paramsName] = paramsValue;
-        }
+    settings.OptLLVM = (GetTask().HasUseLlvm() && GetTask().GetUseLlvm()) ? "--compile-options=disable-opt" : "OFF";
+    settings.UseCacheForLLVM = AppData()->FeatureFlags.GetEnableLLVMCache();
 
-        for (const auto& [paramsName, paramsValue] : GetTask().GetSecureParams()) {
-            settings.SecureParams[paramsName] = paramsValue;
-        }
+    for (const auto& [paramsName, paramsValue] : GetTask().GetTaskParams()) {
+        settings.TaskParams[paramsName] = paramsValue;
+    }
 
-        for (const auto& readRange : GetTask().GetReadRanges()) {
-            settings.ReadRanges.push_back(readRange);
-        }
+    for (const auto& [paramsName, paramsValue] : GetTask().GetSecureParams()) {
+        settings.SecureParams[paramsName] = paramsValue;
+    }
 
-        auto taskRunner = MakeDqTaskRunner(execCtx, settings, logger);
-        SetTaskRunner(taskRunner);
+    for (const auto& readRange : GetTask().GetReadRanges()) {
+        settings.ReadRanges.push_back(readRange);
+    }
 
-        auto wakeup = [this]{ ContinueExecute(); };
-        try {
-            PrepareTaskRunner(TKqpTaskRunnerExecutionContext(std::get<ui64>(TxId), RuntimeSettings.UseSpilling,
-                std::move(wakeup), TlsActivationContext->AsActorContext()));
-        } catch (const NMiniKQL::TKqpEnsureFail& e) {
-            InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage());
-            return;
-        }
+    auto taskRunner = MakeDqTaskRunner(execCtx, settings, logger);
+    SetTaskRunner(taskRunner);
 
-        TSmallVec<NMiniKQL::TKqpScanComputeContext::TColumn> columns;
+    auto wakeup = [this]{ ContinueExecute(); };
+    try {
+        PrepareTaskRunner(TKqpTaskRunnerExecutionContext(std::get<ui64>(TxId), RuntimeSettings.UseSpilling,
+            std::move(wakeup), TlsActivationContext->AsActorContext()));
+    } catch (const NMiniKQL::TKqpEnsureFail& e) {
+        InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage());
+        return;
+    }
 
-        TVector<TSerializedTableRange> ranges;
-        if (Meta) {
-            YQL_ENSURE(ComputeCtx.GetTableScans().empty());
+    TSmallVec<NMiniKQL::TKqpScanComputeContext::TColumn> columns;
 
-            ComputeCtx.AddTableScan(0, *Meta, GetStatsMode());
-            ScanData = &ComputeCtx.GetTableScan(0);
+    TVector<TSerializedTableRange> ranges;
+    if (Meta) {
+        YQL_ENSURE(ComputeCtx.GetTableScans().empty());
 
-            columns.reserve(Meta->ColumnsSize());
-            for (const auto& column : ScanData->GetColumns()) {
-                NMiniKQL::TKqpScanComputeContext::TColumn c;
-                c.Tag = column.Tag;
-                c.Type = column.Type;
-                columns.emplace_back(std::move(c));
-            }
+        ComputeCtx.AddTableScan(0, *Meta, GetStatsMode());
+        ScanData = &ComputeCtx.GetTableScan(0);
 
-            const auto& protoRanges = Meta->GetReads()[0].GetKeyRanges();
-            for (auto& range : protoRanges) {
-                ranges.emplace_back(range);
-            }
+        columns.reserve(Meta->ColumnsSize());
+        for (const auto& column : ScanData->GetColumns()) {
+            NMiniKQL::TKqpScanComputeContext::TColumn c;
+            c.Tag = column.Tag;
+            c.Type = column.Type;
+            columns.emplace_back(std::move(c));
         }
 
-        if (ScanData) {
-            ScanData->TaskId = GetTask().GetId();
-            ScanData->TableReader = CreateKqpTableReader(*ScanData);
+        const auto& protoRanges = Meta->GetReads()[0].GetKeyRanges();
+        for (auto& range : protoRanges) {
+            ranges.emplace_back(range);
+        }
+    }
 
-            auto scanActor = NSysView::CreateSystemViewScan(SelfId(), 0, ScanData->TableId, ranges, columns);
+    if (ScanData) {
+        ScanData->TaskId = GetTask().GetId();
+        ScanData->TableReader = CreateKqpTableReader(*ScanData);
 
-            if (!scanActor) {
-                InternalError(TIssuesIds::DEFAULT_ERROR, TStringBuilder()
-                    << "Failed to create system view scan, table id: " << ScanData->TableId);
-                return;
-            }
+        auto scanActor = NSysView::CreateSystemViewScan(SelfId(), 0, ScanData->TableId, ranges, columns);
 
-            SysViewActorId = Register(scanActor.Release());
-            Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(MemoryLimits.ChannelBufferSize));
+        if (!scanActor) {
+            InternalError(TIssuesIds::DEFAULT_ERROR, TStringBuilder()
+                << "Failed to create system view scan, table id: " << ScanData->TableId);
+            return;
         }
 
-        ContinueExecute();
-        Become(&TKqpComputeActor::StateFunc);
+        SysViewActorId = Register(scanActor.Release());
+        Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(MemoryLimits.ChannelBufferSize));
     }
 
-    STFUNC(StateFunc) {
-        try {
-            switch (ev->GetTypeRewrite()) {
-                hFunc(TEvKqpCompute::TEvScanInitActor, HandleExecute);
-                hFunc(TEvKqpCompute::TEvScanData, HandleExecute);
-                hFunc(TEvKqpCompute::TEvScanError, HandleExecute);
-                default:
-                    BaseStateFuncBody(ev);
-            }
-        } catch (const TMemoryLimitExceededException& e) {
-            InternalError(TIssuesIds::KIKIMR_PRECONDITION_FAILED, TStringBuilder()
-                << "Mkql memory limit exceeded, limit: " << GetMkqlMemoryLimit()
-                << ", host: " << HostName()
-                << ", canAllocateExtraMemory: " << CanAllocateExtraMemory);
-        } catch (const NMiniKQL::TKqpEnsureFail& e) {
-            InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage());
-        } catch (const yexception& e) {
-            InternalError(TIssuesIds::DEFAULT_ERROR, e.what());
-        }
+    ContinueExecute();
+    Become(&TKqpComputeActor::StateFunc);
+}
 
-        ReportEventElapsedTime();
+STFUNC(TKqpComputeActor::StateFunc) {
+    try {
+        switch (ev->GetTypeRewrite()) {
+            hFunc(TEvKqpCompute::TEvScanInitActor, HandleExecute);
+            hFunc(TEvKqpCompute::TEvScanData, HandleExecute);
+            hFunc(TEvKqpCompute::TEvScanError, HandleExecute);
+            default:
+                BaseStateFuncBody(ev);
+        }
+    } catch (const TMemoryLimitExceededException& e) {
+        InternalError(TIssuesIds::KIKIMR_PRECONDITION_FAILED, TStringBuilder()
+            << "Mkql memory limit exceeded, limit: " << GetMkqlMemoryLimit()
+            << ", host: " << HostName()
+            << ", canAllocateExtraMemory: " << CanAllocateExtraMemory);
+    } catch (const NMiniKQL::TKqpEnsureFail& e) {
+        InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage());
+    } catch (const yexception& e) {
+        InternalError(TIssuesIds::DEFAULT_ERROR, e.what());
     }
 
-protected:
-    ui64 CalcMkqlMemoryLimit() override {
-        return TBase::CalcMkqlMemoryLimit() + ComputeCtx.GetTableScans().size() * MemoryLimits.ChannelBufferSize;
-    }
+    ReportEventElapsedTime();
+}
 
-public:
-    void FillExtraStats(NDqProto::TDqComputeActorStats* dst, bool last) {
-        if (last && SysViewActorId && ScanData && dst->TasksSize() > 0) {
-            YQL_ENSURE(dst->TasksSize() == 1);
+ui64 TKqpComputeActor::CalcMkqlMemoryLimit() {
+    return TBase::CalcMkqlMemoryLimit() + ComputeCtx.GetTableScans().size() * MemoryLimits.ChannelBufferSize;
+}
 
-            auto* taskStats = dst->MutableTasks(0);
-            auto* tableStats = taskStats->AddTables();
+void TKqpComputeActor::FillExtraStats(NDqProto::TDqComputeActorStats* dst, bool last) {
+    if (last && SysViewActorId && ScanData && dst->TasksSize() > 0) {
+        YQL_ENSURE(dst->TasksSize() == 1);
 
-            tableStats->SetTablePath(ScanData->TablePath);
+        auto* taskStats = dst->MutableTasks(0);
+        auto* tableStats = taskStats->AddTables();
 
-            if (auto* x = ScanData->BasicStats.get()) {
-                tableStats->SetReadRows(x->Rows);
-                tableStats->SetReadBytes(x->Bytes);
-                // TODO: CpuTime
-            }
+        tableStats->SetTablePath(ScanData->TablePath);
 
-            if (auto* x = ScanData->ProfileStats.get()) {
-                // save your profile stats here
-            }
+        if (auto* x = ScanData->BasicStats.get()) {
+            tableStats->SetReadRows(x->Rows);
+            tableStats->SetReadBytes(x->Bytes);
+            // TODO: CpuTime
         }
-    }
 
-private:
-    void PassAway() override {
-        if (SysViewActorId) {
-            Send(SysViewActorId, new TEvents::TEvPoison);
+        if (auto* x = ScanData->ProfileStats.get()) {
+            // save your profile stats here
         }
+    }
+}
 
-        if (TaskRunner) {
-            if (TaskRunner->IsAllocatorAttached()) {
-                ComputeCtx.Clear();
-            } else {
-                auto guard = TaskRunner->BindAllocator(GetMkqlMemoryLimit());
-                ComputeCtx.Clear();
-            }
-        }
+void TKqpComputeActor::PassAway() {
+    if (SysViewActorId) {
+        Send(SysViewActorId, new TEvents::TEvPoison);
+    }
 
-        TBase::PassAway();
+    if (TaskRunner) {
+        if (TaskRunner->IsAllocatorAttached()) {
+            ComputeCtx.Clear();
+        } else {
+            auto guard = TaskRunner->BindAllocator(GetMkqlMemoryLimit());
+            ComputeCtx.Clear();
+        }
     }
 
-private:
-    void HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr& ev) {
-        Y_VERIFY_DEBUG(ScanData);
+    TBase::PassAway();
+}
+
+void TKqpComputeActor::HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr& ev) {
+    Y_VERIFY_DEBUG(ScanData);
 
-        auto& msg = ev->Get()->Record;
+    auto& msg = ev->Get()->Record;
 
-        Y_VERIFY_DEBUG(SysViewActorId == ActorIdFromProto(msg.GetScanActorId()));
+    Y_VERIFY_DEBUG(SysViewActorId == ActorIdFromProto(msg.GetScanActorId()));
 
-        CA_LOG_D("Got sysview scan initial event, scan actor: " << SysViewActorId << ", scanId: 0");
-        Send(ev->Sender, new TEvKqpCompute::TEvScanDataAck(GetMemoryLimits().ChannelBufferSize));
-        return;
-    }
+    CA_LOG_D("Got sysview scan initial event, scan actor: " << SysViewActorId << ", scanId: 0");
+    Send(ev->Sender, new TEvKqpCompute::TEvScanDataAck(GetMemoryLimits().ChannelBufferSize));
+    return;
+}
 
-    void HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev) {
-        Y_VERIFY_DEBUG(ScanData);
-        Y_VERIFY_DEBUG(SysViewActorId == ev->Sender);
+void TKqpComputeActor::HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev) {
+    Y_VERIFY_DEBUG(ScanData);
+    Y_VERIFY_DEBUG(SysViewActorId == ev->Sender);
 
-        auto& msg = *ev->Get();
+    auto& msg = *ev->Get();
 
-        ui64 bytes = 0;
-        ui64 rowsCount = 0;
-        {
-            auto guard = TaskRunner->BindAllocator();
-            switch (msg.GetDataFormat()) {
-                case NKikimrTxDataShard::EScanDataFormat::UNSPECIFIED:
-                case NKikimrTxDataShard::EScanDataFormat::CELLVEC: {
-                    if (!msg.Rows.empty()) {
-                        bytes = ScanData->AddData(msg.Rows, {}, TaskRunner->GetHolderFactory());
-                        rowsCount = msg.Rows.size();
-                    }
-                    break;
+    ui64 bytes = 0;
+    ui64 rowsCount = 0;
+    {
+        auto guard = TaskRunner->BindAllocator();
+        switch (msg.GetDataFormat()) {
+            case NKikimrTxDataShard::EScanDataFormat::UNSPECIFIED:
+            case NKikimrTxDataShard::EScanDataFormat::CELLVEC: {
+                if (!msg.Rows.empty()) {
+                    bytes = ScanData->AddData(msg.Rows, {}, TaskRunner->GetHolderFactory());
+                    rowsCount = msg.Rows.size();
                 }
-                case NKikimrTxDataShard::EScanDataFormat::ARROW: {
-                    if(msg.ArrowBatch != nullptr) {
-                        bytes = ScanData->AddData(*msg.ArrowBatch, {}, TaskRunner->GetHolderFactory());
-                        rowsCount = msg.ArrowBatch->num_rows();
-                    }
-                    break;
+                break;
+            }
+            case NKikimrTxDataShard::EScanDataFormat::ARROW: {
+                if(msg.ArrowBatch != nullptr) {
+                    bytes = ScanData->AddData(*msg.ArrowBatch, {}, TaskRunner->GetHolderFactory());
+                    rowsCount = msg.ArrowBatch->num_rows();
                 }
+                break;
             }
         }
+    }
 
-        CA_LOG_D("Got sysview scandata, rows: " << rowsCount << ", bytes: " << bytes
-            << ", finished: " << msg.Finished << ", from: " << SysViewActorId);
+    CA_LOG_D("Got sysview scandata, rows: " << rowsCount << ", bytes: " << bytes
+        << ", finished: " << msg.Finished << ", from: " << SysViewActorId);
 
-        if (msg.Finished) {
-            CA_LOG_D("Finishing rows buffer");
-            ScanData->Finish();
-        }
+    if (msg.Finished) {
+        CA_LOG_D("Finishing rows buffer");
+        ScanData->Finish();
+    }
 
-        if (Y_UNLIKELY(ScanData->ProfileStats)) {
-            ScanData->ProfileStats->Messages++;
-            ScanData->ProfileStats->ScanCpuTime += msg.CpuTime; // TODO: not implemented yet
-            ScanData->ProfileStats->ScanWaitTime += msg.WaitTime;
-            if (msg.PageFault) {
-                ScanData->ProfileStats->PageFaults += msg.PageFaults;
-                ScanData->ProfileStats->MessagesByPageFault++;
-            }
+    if (Y_UNLIKELY(ScanData->ProfileStats)) {
+        ScanData->ProfileStats->Messages++;
+        ScanData->ProfileStats->ScanCpuTime += msg.CpuTime; // TODO: not implemented yet
+        ScanData->ProfileStats->ScanWaitTime += msg.WaitTime;
+        if (msg.PageFault) {
+            ScanData->ProfileStats->PageFaults += msg.PageFaults;
+            ScanData->ProfileStats->MessagesByPageFault++;
         }
+    }
 
-        ui64 freeSpace = GetMemoryLimits().ChannelBufferSize > ScanData->GetStoredBytes()
-            ? GetMemoryLimits().ChannelBufferSize - ScanData->GetStoredBytes()
-            : 0;
-
-        if (freeSpace > 0) {
-            CA_LOG_D("Send scan data ack, freeSpace: " << freeSpace);
+    ui64 freeSpace = GetMemoryLimits().ChannelBufferSize > ScanData->GetStoredBytes()
+        ? GetMemoryLimits().ChannelBufferSize - ScanData->GetStoredBytes()
+        : 0;
 
-            Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(freeSpace));
-        }
+    if (freeSpace > 0) {
+        CA_LOG_D("Send scan data ack, freeSpace: " << freeSpace);
 
-        DoExecute();
+        Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(freeSpace));
     }
 
-    void HandleExecute(TEvKqpCompute::TEvScanError::TPtr& ev) {
-        Y_VERIFY_DEBUG(ScanData);
-
-        TIssues issues;
-        Ydb::StatusIds::StatusCode status = ev->Get()->Record.GetStatus();
-        IssuesFromMessage(ev->Get()->Record.GetIssues(), issues);
+    DoExecute();
+}
 
-        State = NDqProto::COMPUTE_STATE_FAILURE;
-        ReportStateAndMaybeDie(YdbStatusToDqStatus(status), issues);
-    }
+void TKqpComputeActor::HandleExecute(TEvKqpCompute::TEvScanError::TPtr& ev) {
+    Y_VERIFY_DEBUG(ScanData);
 
-private:
-    NMiniKQL::TKqpScanComputeContext ComputeCtx;
-    TMaybe<NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta> Meta;
-    NMiniKQL::TKqpScanComputeContext::TScanData* ScanData = nullptr;
-    TActorId SysViewActorId;
-    const TDqTaskRunnerParameterProvider ParameterProvider;
-};
+    TIssues issues;
+    Ydb::StatusIds::StatusCode status = ev->Get()->Record.GetStatus();
+    IssuesFromMessage(ev->Get()->Record.GetIssues(), issues);
 
-} // anonymous namespace
+    State = NDqProto::COMPUTE_STATE_FAILURE;
+    ReportStateAndMaybeDie(YdbStatusToDqStatus(status), issues);
+}
 
 IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
     IDqAsyncIoFactory::TPtr asyncIoFactory,
diff --git a/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h b/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h
new file mode 100644
index 00000000000..3ecae9079a9
--- /dev/null
+++ b/ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h
@@ -0,0 +1,71 @@
+#pragma once
+
+#include "kqp_compute_actor.h"
+#include "kqp_compute_actor_impl.h"
+
+#include <ydb/core/base/appdata.h>
+#include <ydb/core/protos/tx_datashard.pb.h>
+#include <ydb/core/kqp/rm_service/kqp_rm_service.h>
+#include <ydb/core/kqp/runtime/kqp_compute.h>
+#include <ydb/core/kqp/runtime/kqp_scan_data.h>
+#include <ydb/core/sys_view/scan.h>
+#include <ydb/library/yverify_stream/yverify_stream.h>
+
+#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h>
+
+
+namespace NKikimr {
+namespace NKqp {
+
+class TKqpComputeActor : public TDqComputeActorBase<TKqpComputeActor> {
+    using TBase = TDqComputeActorBase<TKqpComputeActor>;
+
+public:
+    static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
+        return NKikimrServices::TActivity::KQP_COMPUTE_ACTOR;
+    }
+
+    TKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
+        IDqAsyncIoFactory::TPtr asyncIoFactory,
+        const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry,
+        const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
+        NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena);
+
+    void DoBootstrap();
+
+    STFUNC(StateFunc);
+
+protected:
+    ui64 CalcMkqlMemoryLimit() override;
+
+public:
+    void FillExtraStats(NDqProto::TDqComputeActorStats* dst, bool last);
+
+private:
+    void PassAway() override;
+
+private:
+    void HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr& ev);
+    
+    void HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev);
+
+    void HandleExecute(TEvKqpCompute::TEvScanError::TPtr& ev);
+
+    bool IsDebugLogEnabled(const TActorSystem* actorSystem);
+
+private:
+    NMiniKQL::TKqpScanComputeContext ComputeCtx;
+    TMaybe<NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta> Meta;
+    NMiniKQL::TKqpScanComputeContext::TScanData* ScanData = nullptr;
+    TActorId SysViewActorId;
+    const TDqTaskRunnerParameterProvider ParameterProvider;
+};
+
+IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
+    IDqAsyncIoFactory::TPtr asyncIoFactory,
+    const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry,
+    const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
+    NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena);
+
+} // namespace NKqp
+} // namespace NKikimr
diff --git a/ydb/core/kqp/executer_actor/kqp_data_executer.cpp b/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
index d5dd1fb68ec..916984da92d 100644
--- a/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
@@ -1783,7 +1783,7 @@ private:
         for (auto& task : TasksGraph.GetTasks()) {
             auto& stageInfo = TasksGraph.GetStageInfo(task.StageId);
             if (task.Meta.ShardId && (task.Meta.Reads || task.Meta.Writes)) {
-                NYql::NDqProto::TDqTask* protoTask = ArenaSerializeTaskToProto(TasksGraph, task);
+                NYql::NDqProto::TDqTask* protoTask = ArenaSerializeTaskToProto(TasksGraph, task, true);
                 datashardTasks[task.Meta.ShardId].emplace_back(protoTask);
             } else if (stageInfo.Meta.IsSysView()) {
                 computeTasks.emplace_back(task.Id);
diff --git a/ydb/core/kqp/executer_actor/kqp_planner.cpp b/ydb/core/kqp/executer_actor/kqp_planner.cpp
index 2bed312e676..f4516628803 100644
--- a/ydb/core/kqp/executer_actor/kqp_planner.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_planner.cpp
@@ -9,6 +9,8 @@
 
 #include <util/generic/set.h>
 
+#include <ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h>
+
 using namespace NActors;
 
 namespace NKikimr::NKqp {
@@ -151,7 +153,7 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
 
     for (ui64 taskId : requestData.TaskIds) {
         const auto& task = TasksGraph.GetTask(taskId);
-        NYql::NDqProto::TDqTask* serializedTask = ArenaSerializeTaskToProto(TasksGraph, task);
+        NYql::NDqProto::TDqTask* serializedTask = ArenaSerializeTaskToProto(TasksGraph, task, /* serializeAsyncIoSettings = */ true);
         request.AddTasks()->Swap(serializedTask);
     }
 
@@ -293,10 +295,12 @@ const IKqpGateway::TKqpSnapshot& TKqpPlanner::GetSnapshot() const {
     return TasksGraph.GetMeta().Snapshot;
 }
 
-void TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, bool shareMailbox) {
+// optimizeProtoForLocalExecution - if we want to execute compute actor locally and don't want to serialize & then deserialize proto message
+// instead we just give ptr to proto message and after that we swap/copy it
+void TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, bool shareMailbox, bool optimizeProtoForLocalExecution) {
 
     auto& task = TasksGraph.GetTask(taskId);
-    NYql::NDqProto::TDqTask* taskDesc = ArenaSerializeTaskToProto(TasksGraph, task);
+    NYql::NDqProto::TDqTask* taskDesc = ArenaSerializeTaskToProto(TasksGraph, task, /* serializeAsyncIoSettings = */ !optimizeProtoForLocalExecution);
 
     NYql::NDq::TComputeRuntimeSettings settings;
     if (Deadline) {
@@ -323,6 +327,17 @@ void TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, bool shareMailbox) {
     auto computeActor = NKikimr::NKqp::CreateKqpComputeActor(ExecuterId, TxId, taskDesc, AsyncIoFactory,
         AppData()->FunctionRegistry, settings, limits, NWilson::TTraceId(), TasksGraph.GetMeta().GetArenaIntrusivePtr());
 
+    if (optimizeProtoForLocalExecution) {
+        TVector<google::protobuf::Message*>& taskSourceSettings = static_cast<TKqpComputeActor*>(computeActor)->MutableTaskSourceSettings();
+        taskSourceSettings.assign(task.Inputs.size(), nullptr);
+        for (size_t i = 0; i < task.Inputs.size(); ++i) {
+            const auto input = task.Inputs[i];
+            if (input.Type() == NYql::NDq::TTaskInputType::Source && Y_LIKELY(input.Meta.SourceSettings)) {
+                taskSourceSettings[i] = (&(*input.Meta.SourceSettings));
+            }
+        }
+    }
+
     auto computeActorId = shareMailbox ? TlsActivationContext->AsActorContext().RegisterWithSameMailbox(computeActor) : TlsActivationContext->AsActorContext().Register(computeActor);
     task.ComputeActorId = computeActorId;
 
@@ -365,7 +380,7 @@ std::unique_ptr<IEventHandle> TKqpPlanner::PlanExecution() {
     if (IsDataQuery) {
         bool shareMailbox = (ComputeTasks.size() <= 1);
         for (ui64 taskId : ComputeTasks) {
-            ExecuteDataComputeTask(taskId, shareMailbox);
+            ExecuteDataComputeTask(taskId, shareMailbox, /* optimizeProtoForLocalExecution = */ true);
         }
         ComputeTasks.clear();
     }
@@ -373,9 +388,9 @@ std::unique_ptr<IEventHandle> TKqpPlanner::PlanExecution() {
     if (nComputeTasks == 0 && TasksPerNode.size() == 1 && (AsyncIoFactory != nullptr) && DoOptimization && IsDataQuery) {
         // query affects a single key or shard, so it might be more effective
         // to execute this task locally so we can avoid useless overhead for remote task launching.
-        for(auto& [shardId, tasks]: TasksPerNode) {
-            for(ui64 taskId: tasks) {
-                ExecuteDataComputeTask(taskId, true);
+        for (auto& [shardId, tasks]: TasksPerNode) {
+            for (ui64 taskId: tasks) {
+                ExecuteDataComputeTask(taskId, true, /* optimizeProtoForLocalExecution = */ true);
             }
         }
 
diff --git a/ydb/core/kqp/executer_actor/kqp_planner.h b/ydb/core/kqp/executer_actor/kqp_planner.h
index 5adde16c328..d8f5da48ad7 100644
--- a/ydb/core/kqp/executer_actor/kqp_planner.h
+++ b/ydb/core/kqp/executer_actor/kqp_planner.h
@@ -63,7 +63,7 @@ public:
 private:
     
     const IKqpGateway::TKqpSnapshot& GetSnapshot() const;
-    void ExecuteDataComputeTask(ui64 taskId, bool shareMailbox);
+    void ExecuteDataComputeTask(ui64 taskId, bool shareMailbox, bool optimizeProtoForLocalExecution);
     void PrepareToProcess();
     TString GetEstimationsInfo() const;
 
diff --git a/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp b/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
index 219cfaeca6e..9d76aaf9ef4 100644
--- a/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
@@ -997,7 +997,7 @@ void FillOutputDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TTaskOutpu
     }
 }
 
-void FillInputDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TTaskInput& inputDesc, const TTaskInput& input) {
+void FillInputDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TTaskInput& inputDesc, const TTaskInput& input, bool serializeAsyncIoSettings) {
     const auto& snapshot = tasksGraph.GetMeta().Snapshot;
     const auto& lockTxId = tasksGraph.GetMeta().LockTxId;
 
@@ -1015,7 +1015,9 @@ void FillInputDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TTaskInput&
                     input.Meta.SourceSettings->SetUseFollowers(tasksGraph.GetMeta().UseFollowers);
                 }
 
-                inputDesc.MutableSource()->MutableSettings()->PackFrom(*input.Meta.SourceSettings);
+                if (serializeAsyncIoSettings) {
+                    inputDesc.MutableSource()->MutableSettings()->PackFrom(*input.Meta.SourceSettings);
+                }
             } else {
                 YQL_ENSURE(input.SourceSettings);
                 inputDesc.MutableSource()->MutableSettings()->CopyFrom(*input.SourceSettings);
@@ -1066,7 +1068,7 @@ void FillInputDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TTaskInput&
     }
 }
 
-void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, NYql::NDqProto::TDqTask* result) {
+void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, NYql::NDqProto::TDqTask* result, bool serializeAsyncIoSettings) {
     auto& stageInfo = tasksGraph.GetStageInfo(task.StageId);
     ActorIdToProto(task.Meta.ExecuterId, result->MutableExecuter()->MutableActorId());
     result->SetId(task.Id);
@@ -1089,7 +1091,7 @@ void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, N
     }
 
     for (const auto& input : task.Inputs) {
-        FillInputDesc(tasksGraph, *result->AddInputs(), input);
+        FillInputDesc(tasksGraph, *result->AddInputs(), input, serializeAsyncIoSettings);
     }
 
     for (const auto& output : task.Outputs) {
@@ -1110,9 +1112,9 @@ void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, N
     FillTaskMeta(stageInfo, task, *result);
 }
 
-NYql::NDqProto::TDqTask* ArenaSerializeTaskToProto(TKqpTasksGraph& tasksGraph, const TTask& task) {
+NYql::NDqProto::TDqTask* ArenaSerializeTaskToProto(TKqpTasksGraph& tasksGraph, const TTask& task, bool serializeAsyncIoSettings) {
     NYql::NDqProto::TDqTask* result = tasksGraph.GetMeta().Allocate<NYql::NDqProto::TDqTask>();
-    SerializeTaskToProto(tasksGraph, task, result);
+    SerializeTaskToProto(tasksGraph, task, result, serializeAsyncIoSettings);
     return result;
 }
 
diff --git a/ydb/core/kqp/executer_actor/kqp_tasks_graph.h b/ydb/core/kqp/executer_actor/kqp_tasks_graph.h
index 5b519187592..b0398100b49 100644
--- a/ydb/core/kqp/executer_actor/kqp_tasks_graph.h
+++ b/ydb/core/kqp/executer_actor/kqp_tasks_graph.h
@@ -249,8 +249,8 @@ void BuildKqpTaskGraphResultChannels(TKqpTasksGraph& tasksGraph, const TKqpPhyTx
 void BuildKqpStageChannels(TKqpTasksGraph& tasksGraph, const TStageInfo& stageInfo,
     ui64 txId, bool enableSpilling);
 
-NYql::NDqProto::TDqTask* ArenaSerializeTaskToProto(TKqpTasksGraph& tasksGraph, const TTask& task);
-void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, NYql::NDqProto::TDqTask* message);
+NYql::NDqProto::TDqTask* ArenaSerializeTaskToProto(TKqpTasksGraph& tasksGraph, const TTask& task, bool serializeAsyncIoSettings);
+void SerializeTaskToProto(const TKqpTasksGraph& tasksGraph, const TTask& task, NYql::NDqProto::TDqTask* message, bool serializeAsyncIoSettings);
 void FillTableMeta(const TStageInfo& stageInfo, NKikimrTxDataShard::TKqpTransaction_TTableMeta* meta);
 void FillChannelDesc(const TKqpTasksGraph& tasksGraph, NYql::NDqProto::TChannel& channelDesc,
     const NYql::NDq::TChannel& channel, const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion);
diff --git a/ydb/core/kqp/runtime/kqp_read_actor.cpp b/ydb/core/kqp/runtime/kqp_read_actor.cpp
index 00d76293578..0f330211aee 100644
--- a/ydb/core/kqp/runtime/kqp_read_actor.cpp
+++ b/ydb/core/kqp/runtime/kqp_read_actor.cpp
@@ -383,10 +383,11 @@ public:
 
 public:
     TKqpReadActor(
-        NKikimrTxDataShard::TKqpReadRangesSourceSettings&& settings,
+        const NKikimrTxDataShard::TKqpReadRangesSourceSettings* settings,
         const NYql::NDq::TDqAsyncIoFactory::TSourceArguments& args,
         TIntrusivePtr<TKqpCounters> counters)
-        : Settings(std::move(settings))
+        : Settings(settings)
+        , Arena(args.Arena)
         , LogPrefix(TStringBuilder() << "TxId: " << args.TxId << ", task: " << args.TaskId << ", CA Id " << args.ComputeActorId << ". ")
         , ComputeActorId(args.ComputeActorId)
         , InputIndex(args.InputIndex)
@@ -397,14 +398,17 @@ public:
         , UseFollowers(false)
         , PipeCacheId(MainPipeCacheId)
     {
+        Y_VERIFY(Arena);
+        Y_VERIFY(settings->GetArena() == Arena->Get());
+
         TableId = TTableId(
-            Settings.GetTable().GetTableId().GetOwnerId(),
-            Settings.GetTable().GetTableId().GetTableId(),
-            Settings.GetTable().GetSysViewInfo(),
-            Settings.GetTable().GetTableId().GetSchemaVersion()
+            Settings->GetTable().GetTableId().GetOwnerId(),
+            Settings->GetTable().GetTableId().GetTableId(),
+            Settings->GetTable().GetSysViewInfo(),
+            Settings->GetTable().GetTableId().GetSchemaVersion()
         );
 
-        if (Settings.GetUseFollowers() && !Snapshot.IsValid()) {
+        if (Settings->GetUseFollowers() && !Snapshot.IsValid()) {
             // reading from followers is allowed only of snapshot is not specified and
             // specific flag is set. otherwise we always read from main replicas.
             PipeCacheId = FollowersPipeCacheId;
@@ -413,22 +417,22 @@ public:
 
         InitResultColumns();
 
-        KeyColumnTypes.reserve(Settings.GetKeyColumnTypes().size());
-        for (size_t i = 0; i < Settings.KeyColumnTypesSize(); ++i) {
-            auto typeId = Settings.GetKeyColumnTypes(i);
+        KeyColumnTypes.reserve(Settings->GetKeyColumnTypes().size());
+        for (size_t i = 0; i < Settings->KeyColumnTypesSize(); ++i) {
+            auto typeId = Settings->GetKeyColumnTypes(i);
             KeyColumnTypes.push_back(
                 NScheme::TTypeInfo(
                     (NScheme::TTypeId)typeId,
                     (typeId == NScheme::NTypeIds::Pg) ?
                         NPg::TypeDescFromPgTypeId(
-                            Settings.GetKeyColumnTypeInfos(i).GetPgTypeId()
+                            Settings->GetKeyColumnTypeInfos(i).GetPgTypeId()
                         ) : nullptr));
         }
         Counters->ReadActorsCount->Inc();
-        Snapshot = IKqpGateway::TKqpSnapshot(Settings.GetSnapshot().GetStep(), Settings.GetSnapshot().GetTxId());
+        Snapshot = IKqpGateway::TKqpSnapshot(Settings->GetSnapshot().GetStep(), Settings->GetSnapshot().GetTxId());
 
-        if (Settings.HasMaxInFlightShards()) {
-            MaxInFlight = Settings.GetMaxInFlightShards();
+        if (Settings->HasMaxInFlightShards()) {
+            MaxInFlight = Settings->GetMaxInFlightShards();
         }
     }
 
@@ -464,28 +468,28 @@ public:
 
     void StartTableScan() {
         ScanStarted = true;
-        THolder<TShardState> stateHolder = MakeHolder<TShardState>(Settings.GetShardIdHint());
+        THolder<TShardState> stateHolder = MakeHolder<TShardState>(Settings->GetShardIdHint());
         PendingShards.PushBack(stateHolder.Get());
         auto& state = *stateHolder.Release();
 
-        if (Settings.HasFullRange()) {
-            state.AddRange(TSerializedTableRange(Settings.GetFullRange()));
+        if (Settings->HasFullRange()) {
+            state.AddRange(TSerializedTableRange(Settings->GetFullRange()));
         } else {
-            YQL_ENSURE(Settings.HasRanges());
-            if (Settings.GetRanges().KeyRangesSize() > 0) {
-                YQL_ENSURE(Settings.GetRanges().KeyPointsSize() == 0);
-                for (const auto& range : Settings.GetRanges().GetKeyRanges()) {
+            YQL_ENSURE(Settings->HasRanges());
+            if (Settings->GetRanges().KeyRangesSize() > 0) {
+                YQL_ENSURE(Settings->GetRanges().KeyPointsSize() == 0);
+                for (const auto& range : Settings->GetRanges().GetKeyRanges()) {
                     state.AddRange(TSerializedTableRange(range));
                 }
             } else {
-                YQL_ENSURE(Settings.GetRanges().KeyPointsSize() > 0);
-                for (const auto& point : Settings.GetRanges().GetKeyPoints()) {
+                YQL_ENSURE(Settings->GetRanges().KeyPointsSize() > 0);
+                for (const auto& point : Settings->GetRanges().GetKeyPoints()) {
                     state.AddPoint(TSerializedCellVec(point));
                 }
             }
         }
 
-        if (!Settings.HasShardIdHint()) {
+        if (!Settings->HasShardIdHint()) {
             InFlightShards.PushBack(&state);
             ResolveShard(&state);
         } else {
@@ -495,15 +499,15 @@ public:
     }
 
     bool StartShards() {
-        const ui32 maxAllowedInFlight = Settings.GetSorted() ? 1 : MaxInFlight;
-        CA_LOG_D("effective maxinflight " << maxAllowedInFlight << " sorted " << Settings.GetSorted());
+        const ui32 maxAllowedInFlight = Settings->GetSorted() ? 1 : MaxInFlight;
+        CA_LOG_D("effective maxinflight " << maxAllowedInFlight << " sorted " << Settings->GetSorted());
         bool isFirst = true;
         while (!PendingShards.Empty() && RunningReads() + 1 <= maxAllowedInFlight) {
             if (isFirst) {
                 CA_LOG_D("BEFORE: " << PendingShards.Size() << "." << RunningReads());
                 isFirst = false;
             }
-            if (Settings.GetReverse()) {
+            if (Settings->GetReverse()) {
                 auto state = THolder<TShardState>(PendingShards.PopBack());
                 InFlightShards.PushBack(state.Get());
                 StartRead(state.Release());
@@ -525,7 +529,7 @@ public:
 
     void ResolveShard(TShardState* state) {
         if (state->ResolveAttempt >= ::MaxShardResolves()) {
-            RuntimeError(TStringBuilder() << "Table '" << Settings.GetTable().GetTablePath() << "' resolve limit exceeded",
+            RuntimeError(TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' resolve limit exceeded",
                 NDqProto::StatusIds::UNAVAILABLE);
             return;
         }
@@ -533,7 +537,7 @@ public:
         Counters->IteratorsShardResolve->Inc();
         state->ResolveAttempt++;
 
-        auto range = state->GetBounds(Settings.GetReverse());
+        auto range = state->GetBounds(Settings->GetReverse());
         TVector<TKeyDesc::TColumnOp> columns;
         columns.reserve(ResultColumns.size());
         for (const auto& column : ResultColumns) {
@@ -547,7 +551,7 @@ public:
         auto keyDesc = MakeHolder<TKeyDesc>(TableId, range, TKeyDesc::ERowOperation::Read,
             KeyColumnTypes, columns);
 
-        CA_LOG_D("Sending TEvResolveKeySet update for table '" << Settings.GetTable().GetTablePath() << "'"
+        CA_LOG_D("Sending TEvResolveKeySet update for table '" << Settings->GetTable().GetTablePath() << "'"
             << ", range: " << DebugPrintRange(KeyColumnTypes, range, *AppData()->TypeRegistry)
             << ", attempt #" << state->ResolveAttempt);
 
@@ -563,7 +567,7 @@ public:
     }
 
     void HandleResolve(TEvTxProxySchemeCache::TEvResolveKeySetResult::TPtr& ev) {
-        CA_LOG_D("Received TEvResolveKeySetResult update for table '" << Settings.GetTable().GetTablePath() << "'");
+        CA_LOG_D("Received TEvResolveKeySetResult update for table '" << Settings->GetTable().GetTablePath() << "'");
 
         auto* request = ev->Get()->Request.Get();
         THolder<TShardState> state;
@@ -575,7 +579,7 @@ public:
         }
 
         if (request->ErrorCount > 0 || !state) {
-            CA_LOG_E("Resolve request failed for table '" << Settings.GetTable().GetTablePath() << "', ErrorCount# " << request->ErrorCount);
+            CA_LOG_E("Resolve request failed for table '" << Settings->GetTable().GetTablePath() << "', ErrorCount# " << request->ErrorCount);
 
             auto statusCode = NDqProto::StatusIds::UNAVAILABLE;
             TString error;
@@ -588,19 +592,19 @@ public:
                     switch (x.Status) {
                         case NSchemeCache::TSchemeCacheRequest::EStatus::PathErrorNotExist:
                             statusCode = NDqProto::StatusIds::SCHEME_ERROR;
-                            error = TStringBuilder() << "Table '" << Settings.GetTable().GetTablePath() << "' not exists.";
+                            error = TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' not exists.";
                             break;
                         case NSchemeCache::TSchemeCacheRequest::EStatus::TypeCheckError:
                             statusCode = NDqProto::StatusIds::SCHEME_ERROR;
-                            error = TStringBuilder() << "Table '" << Settings.GetTable().GetTablePath() << "' scheme changed.";
+                            error = TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' scheme changed.";
                             break;
                         case NSchemeCache::TSchemeCacheRequest::EStatus::LookupError:
                             statusCode = NDqProto::StatusIds::UNAVAILABLE;
-                            error = TStringBuilder() << "Failed to resolve table '" << Settings.GetTable().GetTablePath() << "'.";
+                            error = TStringBuilder() << "Failed to resolve table '" << Settings->GetTable().GetTablePath() << "'.";
                             break;
                         default:
                             statusCode = NDqProto::StatusIds::SCHEME_ERROR;
-                            error = TStringBuilder() << "Unresolved table '" << Settings.GetTable().GetTablePath() << "'. Status: " << x.Status;
+                            error = TStringBuilder() << "Unresolved table '" << Settings->GetTable().GetTablePath() << "'. Status: " << x.Status;
                             break;
                     }
                 }
@@ -628,7 +632,7 @@ public:
         }
 
         if (keyDesc->GetPartitions().empty()) {
-            TString error = TStringBuilder() << "No partitions to read from '" << Settings.GetTable().GetTablePath() << "'";
+            TString error = TStringBuilder() << "No partitions to read from '" << Settings->GetTable().GetTablePath() << "'";
             CA_LOG_E(error);
             return RuntimeError(error, NDqProto::StatusIds::SCHEME_ERROR);
         }
@@ -638,17 +642,17 @@ public:
         TVector<THolder<TShardState>> newShards;
         newShards.reserve(keyDesc->GetPartitions().size());
 
-        auto bounds = state->GetBounds(Settings.GetReverse());
+        auto bounds = state->GetBounds(Settings->GetReverse());
         size_t pointIndex = 0;
         size_t rangeIndex = 0;
         TVector<TSerializedTableRange> ranges;
         if (state->HasRanges()) {
-            state->FillUnprocessedRanges(ranges, KeyColumnTypes, Settings.GetReverse());
+            state->FillUnprocessedRanges(ranges, KeyColumnTypes, Settings->GetReverse());
         }
 
         TVector<TSerializedCellVec> points;
         if (state->HasPoints()) {
-            state->FillUnprocessedPoints(points, Settings.GetReverse());
+            state->FillUnprocessedPoints(points, Settings->GetReverse());
         }
 
         for (ui64 idx = 0; idx < keyDesc->GetPartitions().size(); ++idx) {
@@ -717,7 +721,7 @@ public:
 
         YQL_ENSURE(!newShards.empty());
         Counters->IteratorsReadSplits->Add(newShards.size() - 1);
-        if (Settings.GetReverse()) {
+        if (Settings->GetReverse()) {
             for (size_t i = 0; i < newShards.size(); ++i) {
                 PendingShards.PushBack(newShards[i].Release());
             }
@@ -766,7 +770,7 @@ public:
         TotalRetries += 1;
         auto limit = ::MaxTotalRetries();
         if (limit && TotalRetries > *limit) {
-            return RuntimeError(TStringBuilder() << "Table '" << Settings.GetTable().GetTablePath() << "' retry limit exceeded",
+            return RuntimeError(TStringBuilder() << "Table '" << Settings->GetTable().GetTablePath() << "' retry limit exceeded",
                 NDqProto::StatusIds::UNAVAILABLE);
         }
 
@@ -812,8 +816,8 @@ public:
 
     void StartRead(TShardState* state) {
         TMaybe<ui64> limit;
-        if (Settings.GetItemsLimit()) {
-            limit = Settings.GetItemsLimit() - Min(Settings.GetItemsLimit(), ReceivedRowCount);
+        if (Settings->GetItemsLimit()) {
+            limit = Settings->GetItemsLimit() - Min(Settings->GetItemsLimit(), ReceivedRowCount);
 
             if (*limit == 0) {
                 return;
@@ -823,8 +827,8 @@ public:
         auto ev = ::DefaultReadSettings();
         auto& record = ev->Record;
 
-        state->FillEvRead(*ev, KeyColumnTypes, Settings.GetReverse());
-        for (const auto& column : Settings.GetColumns()) {
+        state->FillEvRead(*ev, KeyColumnTypes, Settings->GetReverse());
+        for (const auto& column : Settings->GetColumns()) {
             if (!IsSystemColumn(column.GetId())) {
                 record.AddColumns(column.GetId());
             }
@@ -848,34 +852,34 @@ public:
 
         record.SetReadId(id);
 
-        record.MutableTableId()->SetOwnerId(Settings.GetTable().GetTableId().GetOwnerId());
-        record.MutableTableId()->SetTableId(Settings.GetTable().GetTableId().GetTableId());
-        record.MutableTableId()->SetSchemaVersion(Settings.GetTable().GetSchemaVersion());
+        record.MutableTableId()->SetOwnerId(Settings->GetTable().GetTableId().GetOwnerId());
+        record.MutableTableId()->SetTableId(Settings->GetTable().GetTableId().GetTableId());
+        record.MutableTableId()->SetSchemaVersion(Settings->GetTable().GetSchemaVersion());
 
-        record.SetReverse(Settings.GetReverse());
+        record.SetReverse(Settings->GetReverse());
         if (limit) {
             record.SetMaxRows(*limit);
             record.SetTotalRowsLimit(*limit);
         }
         record.SetMaxBytes(Min<ui64>(record.GetMaxBytes(), BufSize));
 
-        record.SetResultFormat(Settings.GetDataFormat());
+        record.SetResultFormat(Settings->GetDataFormat());
 
-        if (Settings.HasLockTxId() && BrokenLocks.empty()) {
-            record.SetLockTxId(Settings.GetLockTxId());
+        if (Settings->HasLockTxId() && BrokenLocks.empty()) {
+            record.SetLockTxId(Settings->GetLockTxId());
         }
 
-        if (Settings.HasLockNodeId()) {
-            record.SetLockNodeId(Settings.GetLockNodeId());
+        if (Settings->HasLockNodeId()) {
+            record.SetLockNodeId(Settings->GetLockNodeId());
         }
 
-        CA_LOG_D(TStringBuilder() << "Send EvRead to shardId: " << state->TabletId << ", tablePath: " << Settings.GetTable().GetTablePath()
+        CA_LOG_D(TStringBuilder() << "Send EvRead to shardId: " << state->TabletId << ", tablePath: " << Settings->GetTable().GetTablePath()
             << ", ranges: " << DebugPrintRanges(KeyColumnTypes, ev->Ranges, *AppData()->TypeRegistry)
             << ", limit: " << limit
             << ", readId = " << id
             << ", reverse = " << record.GetReverse()
-            << " snapshot = (txid=" << Settings.GetSnapshot().GetTxId() << ",step=" << Settings.GetSnapshot().GetStep() << ")"
-            << " lockTxId = " << Settings.GetLockTxId());
+            << " snapshot = (txid=" << Settings->GetSnapshot().GetTxId() << ",step=" << Settings->GetSnapshot().GetStep() << ")"
+            << " lockTxId = " << Settings->GetLockTxId());
 
         Counters->CreatedIterators->Inc();
         ReadIdByTabletId[state->TabletId].push_back(id);
@@ -1065,7 +1069,7 @@ public:
             return {};
         }
         YQL_ENSURE(packed == 0);
-        if (Settings.ColumnsSize() == 0) {
+        if (Settings->ColumnsSize() == 0) {
             batch->resize(result->Get()->GetRowsCount(), HolderFactory.GetEmptyContainer());
         } else {
             TVector<NUdf::TUnboxedValue*> editAccessors(result->Get()->GetRowsCount());
@@ -1073,7 +1077,7 @@ public:
 
             for (ui64 rowIndex = 0; rowIndex < result->Get()->GetRowsCount(); ++rowIndex) {
                 batch->emplace_back(HolderFactory.CreateDirectArrayHolder(
-                    Settings.columns_size(),
+                    Settings->columns_size(),
                     editAccessors[rowIndex])
                 );
             }
@@ -1111,7 +1115,7 @@ public:
         }
         TStringBuilder builder;
         TVector<NScheme::TTypeInfo> types;
-        for (auto& column : Settings.GetColumns()) {
+        for (auto& column : Settings->GetColumns()) {
             types.push_back(NScheme::TTypeInfo((NScheme::TTypeId)column.GetType()));
         }
 
@@ -1130,7 +1134,7 @@ public:
         for (size_t rowIndex = packed; rowIndex < result->Get()->GetRowsCount(); ++rowIndex) {
             const auto& row = result->Get()->GetCells(rowIndex);
             NUdf::TUnboxedValue* rowItems = nullptr;
-            batch->emplace_back(HolderFactory.CreateDirectArrayHolder(Settings.ColumnsSize(), rowItems));
+            batch->emplace_back(HolderFactory.CreateDirectArrayHolder(Settings->ColumnsSize(), rowItems));
 
             i64 rowSize = 0;
             size_t columnIndex = 0;
@@ -1167,7 +1171,7 @@ public:
     }
 
     bool LimitReached() const {
-        return Settings.GetItemsLimit() && ProcessedRowCount >= Settings.GetItemsLimit();
+        return Settings->GetItemsLimit() && ProcessedRowCount >= Settings->GetItemsLimit();
     }
 
     i64 GetAsyncInputData(
@@ -1210,7 +1214,7 @@ public:
                 resultBatch.push_back(std::move((*batch)[result.ProcessedRows]));
                 ProcessedRowCount += 1;
                 bytes += rowSize.AllocatedBytes;
-                if (ProcessedRowCount == Settings.GetItemsLimit()) {
+                if (ProcessedRowCount == Settings->GetItemsLimit()) {
                     finished = true;
                     CA_LOG_D(TStringBuilder() << " returned async data because limit reached");
                     return bytes;
@@ -1223,8 +1227,8 @@ public:
                 auto& record = msg.Record;
                 if (!Reads[id].Finished) {
                     TMaybe<ui64> limit;
-                    if (Settings.GetItemsLimit()) {
-                        limit = Settings.GetItemsLimit() - Min(Settings.GetItemsLimit(), ReceivedRowCount);
+                    if (Settings->GetItemsLimit()) {
+                        limit = Settings->GetItemsLimit() - Min(Settings->GetItemsLimit(), ReceivedRowCount);
                     }
 
                     if (!limit || *limit > 0) {
@@ -1276,7 +1280,7 @@ public:
             << " running reads " << RunningReads()
             << " pending shards " << PendingShards.Size()
             << " finished = " << finished
-            << " has limit " << (Settings.GetItemsLimit() != 0)
+            << " has limit " << (Settings->GetItemsLimit() != 0)
             << " limit reached " << LimitReached());
 
         return bytes;
@@ -1287,13 +1291,13 @@ public:
             NDqProto::TDqTableStats* tableStats = nullptr;
             for (size_t i = 0; i < stats->TablesSize(); ++i) {
                 auto* table = stats->MutableTables(i);
-                if (table->GetTablePath() == Settings.GetTable().GetTablePath()) {
+                if (table->GetTablePath() == Settings->GetTable().GetTablePath()) {
                     tableStats = table;
                 }
             }
             if (!tableStats) {
                 tableStats = stats->AddTables();
-                tableStats->SetTablePath(Settings.GetTable().GetTablePath());
+                tableStats->SetTablePath(Settings->GetTable().GetTablePath());
 
             }
 
@@ -1368,9 +1372,9 @@ private:
     }
 
     void InitResultColumns() {
-        ResultColumns.reserve(Settings.ColumnsSize());
-        for (size_t resultColumnIndex = 0; resultColumnIndex < Settings.ColumnsSize(); ++resultColumnIndex) {
-            const auto& srcColumn = Settings.GetColumns(resultColumnIndex);
+        ResultColumns.reserve(Settings->ColumnsSize());
+        for (size_t resultColumnIndex = 0; resultColumnIndex < Settings->ColumnsSize(); ++resultColumnIndex) {
+            const auto& srcColumn = Settings->GetColumns(resultColumnIndex);
             TResultColumn column;
             column.Tag = srcColumn.GetId();
             column.TypeInfo = MakeTypeInfo(srcColumn);
@@ -1386,7 +1390,8 @@ private:
         NScheme::TTypeInfo TypeInfo;
     };
 
-    NKikimrTxDataShard::TKqpReadRangesSourceSettings Settings;
+    const NKikimrTxDataShard::TKqpReadRangesSourceSettings* Settings;
+    TIntrusivePtr<NActors::TProtoArenaHolder> Arena;
 
     TVector<TResultColumn> ResultColumns;
     TVector<NScheme::TTypeInfo> KeyColumnTypes;
@@ -1436,10 +1441,10 @@ private:
 
 
 void RegisterKqpReadActor(NYql::NDq::TDqAsyncIoFactory& factory, TIntrusivePtr<TKqpCounters> counters) {
-    factory.RegisterSource<NKikimrTxDataShard::TKqpReadRangesSourceSettings>(
+    factory.RegisterSourceWithPtr<NKikimrTxDataShard::TKqpReadRangesSourceSettings>(
         TString(NYql::KqpReadRangesSourceName),
-        [counters] (NKikimrTxDataShard::TKqpReadRangesSourceSettings&& settings, NYql::NDq::TDqAsyncIoFactory::TSourceArguments&& args) {
-            auto* actor = new TKqpReadActor(std::move(settings), args, counters);
+        [counters] (const NKikimrTxDataShard::TKqpReadRangesSourceSettings* settings, NYql::NDq::TDqAsyncIoFactory::TSourceArguments&& args) {
+            auto* actor = new TKqpReadActor(settings, args, counters);
             return std::make_pair<NYql::NDq::IDqComputeActorAsyncInput*, IActor*>(actor, actor);
         });
 }
diff --git a/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h b/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h
index 5f1dade0e84..39b20c49d89 100644
--- a/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h
+++ b/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h
@@ -192,6 +192,8 @@ public:
         ::NMonitoring::TDynamicCounterPtr TaskCounters;
         std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
         IMemoryQuotaManager::TPtr MemoryQuotaManager;
+        const google::protobuf::Message* SourceSettings = nullptr;  // used only in case if we execute compute actor locally
+        TIntrusivePtr<NActors::TProtoArenaHolder> Arena;  // Arena for SourceSettings
     };
 
     struct TSinkArguments {
diff --git a/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h b/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h
index e5be6d2fcd1..53b3e2762e3 100644
--- a/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h
+++ b/ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h
@@ -21,6 +21,11 @@ concept TSourceCreatorFunc = requires(T f, TProto&& settings, IDqAsyncIoFactory:
 };
 
 template <class T, class TProto>
+concept TSourceCreatorFuncMove = requires(T f, TProto* settings, IDqAsyncIoFactory::TSourceArguments&& args) {
+    { f(settings, std::move(args)) } -> TCastsToAsyncInputPair;
+};
+
+template <class T, class TProto>
 concept TInputTransformCreatorFunc = requires(T f, TProto&& settings, IDqAsyncIoFactory::TInputTransformArguments&& args) {
     { f(std::move(settings), std::move(args)) } -> TCastsToAsyncInputPair;
 };
@@ -49,6 +54,30 @@ public:
     // Registration
     void RegisterSource(const TString& type, TSourceCreatorFunction creator);
 
+
+    // This function expects to have Arena where SourceSettings are allocated or will be allocated
+    template <class TProtoMsg, TSourceCreatorFuncMove<TProtoMsg> TCreatorFunc>
+    void RegisterSourceWithPtr(const TString& type, TCreatorFunc creator) {  // WithArena
+        RegisterSource(type,
+            [creator = std::move(creator), type](TSourceArguments&& args)
+            {
+                YQL_ENSURE(args.Arena, "args are expected to have Arena for SourceSettings");
+                if (args.SourceSettings != nullptr) {
+                    const TProtoMsg* settingsPtr = dynamic_cast<const TProtoMsg*>(args.SourceSettings);
+                    YQL_ENSURE(settingsPtr);
+                    return creator(settingsPtr, std::move(args));
+                } else {
+                    const google::protobuf::Any& settingsAny = args.InputDesc.GetSource().GetSettings();
+                    YQL_ENSURE(settingsAny.Is<TProtoMsg>(),
+                        "Source \"" << type << "\" settings are expected to have protobuf type " << TProtoMsg::descriptor()->full_name()
+                        << ", but got " << settingsAny.type_url());
+                    TProtoMsg* settingsPtr = args.Arena->Allocate<TProtoMsg>();
+                    YQL_ENSURE(settingsAny.UnpackTo(settingsPtr), "Failed to unpack settings of type \"" << type << "\"");
+                    return creator(std::move(settingsPtr), std::move(args));
+                }
+        });
+    }
+
     template <class TProtoMsg, TSourceCreatorFunc<TProtoMsg> TCreatorFunc>
     void RegisterSource(const TString& type, TCreatorFunc creator) {
         RegisterSource(type,
diff --git a/ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h b/ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h
index e980fa9880b..612dc5066c1 100644
--- a/ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h
+++ b/ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h
@@ -1513,6 +1513,12 @@ protected:
         return MemoryLimits;
     }
 
+public:
+
+    TVector<google::protobuf::Message*>& MutableTaskSourceSettings() {
+        return Task.MutableSourceSettings();
+    }
+
 protected:
     void SetTaskRunner(const TIntrusivePtr<IDqTaskRunner>& taskRunner) {
         TaskRunner = taskRunner;
@@ -1559,6 +1565,8 @@ protected:
             Y_VERIFY(inputDesc.HasSource());
             source.Type = inputDesc.GetSource().GetType();
             const ui64 i = inputIndex; // Crutch for clang
+            const auto& settings = Task.GetSourceSettings();
+            Y_VERIFY(settings.empty() || inputIndex < settings.size());
             CA_LOG_D("Create source for input " << i << " " << inputDesc);
             try {
                 std::tie(source.AsyncInput, source.Actor) = AsyncIoFactory->CreateDqSource(
@@ -1574,7 +1582,9 @@ protected:
                         .HolderFactory = holderFactory,
                         .TaskCounters = TaskCounters,
                         .Alloc = TaskRunner ? TaskRunner->GetAllocatorPtr() : nullptr,
-                        .MemoryQuotaManager = MemoryLimits.MemoryQuotaManager
+                        .MemoryQuotaManager = MemoryLimits.MemoryQuotaManager,
+                        .SourceSettings = (!settings.empty() ? settings.at(inputIndex) : nullptr),
+                        .Arena = Task.GetArena()
                     });
             } catch (const std::exception& ex) {
                 throw yexception() << "Failed to create source " << inputDesc.GetSource().GetType() << ": " << ex.what();
diff --git a/ydb/library/yql/dq/runtime/dq_tasks_runner.h b/ydb/library/yql/dq/runtime/dq_tasks_runner.h
index b50f2b6f511..83bbca81b2a 100644
--- a/ydb/library/yql/dq/runtime/dq_tasks_runner.h
+++ b/ydb/library/yql/dq/runtime/dq_tasks_runner.h
@@ -392,6 +392,18 @@ public:
         return Task_->HasUseLlvm();
     }
 
+    const TVector<google::protobuf::Message*>& GetSourceSettings() const {
+        return SourceSettings;
+    }
+
+    TVector<google::protobuf::Message*>& MutableSourceSettings() {
+        return SourceSettings;
+    }
+
+    const TIntrusivePtr<NActors::TProtoArenaHolder>& GetArena() const {
+        return Arena;
+    }
+
 private:
 
     // external callback to retrieve parameter value.
@@ -399,6 +411,7 @@ private:
     NDqProto::TDqTask* Task_ = nullptr;
     std::unique_ptr<NDqProto::TDqTask> HeapTask_;
     TIntrusivePtr<NActors::TProtoArenaHolder> Arena;
+    TVector<google::protobuf::Message*> SourceSettings;  // used only in case if we execute compute actor locally
 };
 
 class IDqTaskRunner : public TSimpleRefCount<IDqTaskRunner>, private TNonCopyable {