From dac948692cbbd01fe980826d12fcf540ccd61ba8 Mon Sep 17 00:00:00 2001 From: makostrov Date: Thu, 28 Sep 2023 01:39:52 +0300 Subject: [kqp] убрать PackFrom/Unpack для настроек чтения из source KIKIMR-19032 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit KIKIMR-19032 --- .../kqp/compute_actor/kqp_pure_compute_actor.cpp | 443 ++++++++++----------- .../kqp/compute_actor/kqp_pure_compute_actor.h | 71 ++++ ydb/core/kqp/executer_actor/kqp_data_executer.cpp | 2 +- ydb/core/kqp/executer_actor/kqp_planner.cpp | 29 +- ydb/core/kqp/executer_actor/kqp_planner.h | 2 +- ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp | 14 +- ydb/core/kqp/executer_actor/kqp_tasks_graph.h | 4 +- ydb/core/kqp/runtime/kqp_read_actor.cpp | 157 ++++---- .../dq/actors/compute/dq_compute_actor_async_io.h | 2 + .../compute/dq_compute_actor_async_io_factory.h | 29 ++ .../yql/dq/actors/compute/dq_compute_actor_impl.h | 12 +- ydb/library/yql/dq/runtime/dq_tasks_runner.h | 13 + 12 files changed, 443 insertions(+), 335 deletions(-) create mode 100644 ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h 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 -#include -#include -#include -#include -#include -#include - -#include - +#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 { - using TBase = TDqComputeActorBase; +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 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()) { + 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 TKqpComputeActor::DoBootstrap() { + const TActorSystem* actorSystem = TlsActivationContext->ActorSystem(); -public: - static constexpr NKikimrServices::TActivity::EType ActorActivityType() { - return NKikimrServices::TActivity::KQP_COMPUTE_ACTOR; + 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); + }; } - 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 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()) { - 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()); - } + TDqTaskRunnerContext execCtx; + + 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(); + + TDqTaskRunnerSettings settings; + settings.CollectBasicStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_BASIC; + settings.CollectProfileStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_PROFILE; + + settings.OptLLVM = (GetTask().HasUseLlvm() && GetTask().GetUseLlvm()) ? "--compile-options=disable-opt" : "OFF"; + settings.UseCacheForLLVM = AppData()->FeatureFlags.GetEnableLLVMCache(); + + for (const auto& [paramsName, paramsValue] : GetTask().GetTaskParams()) { + settings.TaskParams[paramsName] = paramsValue; } - void DoBootstrap() { - const TActorSystem* actorSystem = TlsActivationContext->ActorSystem(); + for (const auto& [paramsName, paramsValue] : GetTask().GetSecureParams()) { + settings.SecureParams[paramsName] = paramsValue; + } - 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); - }; - } + for (const auto& readRange : GetTask().GetReadRanges()) { + settings.ReadRanges.push_back(readRange); + } - TDqTaskRunnerContext execCtx; + auto taskRunner = MakeDqTaskRunner(execCtx, settings, logger); + SetTaskRunner(taskRunner); - 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(); + auto wakeup = [this]{ ContinueExecute(); }; + try { + PrepareTaskRunner(TKqpTaskRunnerExecutionContext(std::get(TxId), RuntimeSettings.UseSpilling, + std::move(wakeup), TlsActivationContext->AsActorContext())); + } catch (const NMiniKQL::TKqpEnsureFail& e) { + InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage()); + return; + } - TDqTaskRunnerSettings settings; - settings.CollectBasicStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_BASIC; - settings.CollectProfileStats = RuntimeSettings.StatsMode >= NYql::NDqProto::DQ_STATS_MODE_PROFILE; + TSmallVec columns; - settings.OptLLVM = (GetTask().HasUseLlvm() && GetTask().GetUseLlvm()) ? "--compile-options=disable-opt" : "OFF"; - settings.UseCacheForLLVM = AppData()->FeatureFlags.GetEnableLLVMCache(); + TVector ranges; + if (Meta) { + YQL_ENSURE(ComputeCtx.GetTableScans().empty()); - for (const auto& [paramsName, paramsValue] : GetTask().GetTaskParams()) { - settings.TaskParams[paramsName] = paramsValue; - } + ComputeCtx.AddTableScan(0, *Meta, GetStatsMode()); + ScanData = &ComputeCtx.GetTableScan(0); - for (const auto& [paramsName, paramsValue] : GetTask().GetSecureParams()) { - settings.SecureParams[paramsName] = paramsValue; + 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)); } - for (const auto& readRange : GetTask().GetReadRanges()) { - settings.ReadRanges.push_back(readRange); + const auto& protoRanges = Meta->GetReads()[0].GetKeyRanges(); + for (auto& range : protoRanges) { + ranges.emplace_back(range); } + } + + if (ScanData) { + ScanData->TaskId = GetTask().GetId(); + ScanData->TableReader = CreateKqpTableReader(*ScanData); - auto taskRunner = MakeDqTaskRunner(execCtx, settings, logger); - SetTaskRunner(taskRunner); + auto scanActor = NSysView::CreateSystemViewScan(SelfId(), 0, ScanData->TableId, ranges, columns); - auto wakeup = [this]{ ContinueExecute(); }; - try { - PrepareTaskRunner(TKqpTaskRunnerExecutionContext(std::get(TxId), RuntimeSettings.UseSpilling, - std::move(wakeup), TlsActivationContext->AsActorContext())); - } catch (const NMiniKQL::TKqpEnsureFail& e) { - InternalError((TIssuesIds::EIssueCode) e.GetCode(), e.GetMessage()); + if (!scanActor) { + InternalError(TIssuesIds::DEFAULT_ERROR, TStringBuilder() + << "Failed to create system view scan, table id: " << ScanData->TableId); return; } - TSmallVec columns; - - TVector ranges; - if (Meta) { - YQL_ENSURE(ComputeCtx.GetTableScans().empty()); - - ComputeCtx.AddTableScan(0, *Meta, GetStatsMode()); - ScanData = &ComputeCtx.GetTableScan(0); + SysViewActorId = Register(scanActor.Release()); + Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(MemoryLimits.ChannelBufferSize)); + } - 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)); - } + ContinueExecute(); + Become(&TKqpComputeActor::StateFunc); +} - const auto& protoRanges = Meta->GetReads()[0].GetKeyRanges(); - for (auto& range : protoRanges) { - ranges.emplace_back(range); - } +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()); + } - if (ScanData) { - ScanData->TaskId = GetTask().GetId(); - ScanData->TableReader = CreateKqpTableReader(*ScanData); + ReportEventElapsedTime(); +} - auto scanActor = NSysView::CreateSystemViewScan(SelfId(), 0, ScanData->TableId, ranges, columns); +ui64 TKqpComputeActor::CalcMkqlMemoryLimit() { + return TBase::CalcMkqlMemoryLimit() + ComputeCtx.GetTableScans().size() * MemoryLimits.ChannelBufferSize; +} - if (!scanActor) { - InternalError(TIssuesIds::DEFAULT_ERROR, TStringBuilder() - << "Failed to create system view scan, table id: " << ScanData->TableId); - return; - } +void TKqpComputeActor::FillExtraStats(NDqProto::TDqComputeActorStats* dst, bool last) { + if (last && SysViewActorId && ScanData && dst->TasksSize() > 0) { + YQL_ENSURE(dst->TasksSize() == 1); - SysViewActorId = Register(scanActor.Release()); - Send(SysViewActorId, new TEvKqpCompute::TEvScanDataAck(MemoryLimits.ChannelBufferSize)); - } + auto* taskStats = dst->MutableTasks(0); + auto* tableStats = taskStats->AddTables(); - ContinueExecute(); - Become(&TKqpComputeActor::StateFunc); - } + tableStats->SetTablePath(ScanData->TablePath); - 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()); + if (auto* x = ScanData->BasicStats.get()) { + tableStats->SetReadRows(x->Rows); + tableStats->SetReadBytes(x->Bytes); + // TODO: CpuTime } - ReportEventElapsedTime(); + if (auto* x = ScanData->ProfileStats.get()) { + // save your profile stats here + } } +} -protected: - ui64 CalcMkqlMemoryLimit() override { - return TBase::CalcMkqlMemoryLimit() + ComputeCtx.GetTableScans().size() * MemoryLimits.ChannelBufferSize; +void TKqpComputeActor::PassAway() { + if (SysViewActorId) { + Send(SysViewActorId, new TEvents::TEvPoison); } -public: - void FillExtraStats(NDqProto::TDqComputeActorStats* dst, bool last) { - if (last && SysViewActorId && ScanData && dst->TasksSize() > 0) { - YQL_ENSURE(dst->TasksSize() == 1); - - auto* taskStats = dst->MutableTasks(0); - auto* tableStats = taskStats->AddTables(); - - tableStats->SetTablePath(ScanData->TablePath); - - if (auto* x = ScanData->BasicStats.get()) { - tableStats->SetReadRows(x->Rows); - tableStats->SetReadBytes(x->Bytes); - // TODO: CpuTime - } - - 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(); } } -private: - void PassAway() override { - if (SysViewActorId) { - Send(SysViewActorId, new TEvents::TEvPoison); - } + TBase::PassAway(); +} - if (TaskRunner) { - if (TaskRunner->IsAllocatorAttached()) { - ComputeCtx.Clear(); - } else { - auto guard = TaskRunner->BindAllocator(GetMkqlMemoryLimit()); - ComputeCtx.Clear(); - } - } +void TKqpComputeActor::HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr& ev) { + Y_VERIFY_DEBUG(ScanData); - TBase::PassAway(); - } + auto& msg = ev->Get()->Record; -private: - void HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr& ev) { - Y_VERIFY_DEBUG(ScanData); + Y_VERIFY_DEBUG(SysViewActorId == ActorIdFromProto(msg.GetScanActorId())); - auto& msg = ev->Get()->Record; + CA_LOG_D("Got sysview scan initial event, scan actor: " << SysViewActorId << ", scanId: 0"); + Send(ev->Sender, new TEvKqpCompute::TEvScanDataAck(GetMemoryLimits().ChannelBufferSize)); + return; +} - Y_VERIFY_DEBUG(SysViewActorId == ActorIdFromProto(msg.GetScanActorId())); +void TKqpComputeActor::HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev) { + Y_VERIFY_DEBUG(ScanData); + Y_VERIFY_DEBUG(SysViewActorId == ev->Sender); - CA_LOG_D("Got sysview scan initial event, scan actor: " << SysViewActorId << ", scanId: 0"); - Send(ev->Sender, new TEvKqpCompute::TEvScanDataAck(GetMemoryLimits().ChannelBufferSize)); - return; - } + auto& msg = *ev->Get(); - void HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev) { - Y_VERIFY_DEBUG(ScanData); - Y_VERIFY_DEBUG(SysViewActorId == ev->Sender); - - 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 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 +#include +#include +#include +#include +#include +#include + +#include + + +namespace NKikimr { +namespace NKqp { + +class TKqpComputeActor : public TDqComputeActorBase { + using TBase = TDqComputeActorBase; + +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 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 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 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 +#include + using namespace NActors; namespace NKikimr::NKqp { @@ -151,7 +153,7 @@ std::unique_ptr 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& taskSourceSettings = static_cast(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 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 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(); - 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 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 stateHolder = MakeHolder(Settings.GetShardIdHint()); + THolder stateHolder = MakeHolder(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(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 columns; columns.reserve(ResultColumns.size()); for (const auto& column : ResultColumns) { @@ -547,7 +551,7 @@ public: auto keyDesc = MakeHolder(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 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> 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 ranges; if (state->HasRanges()) { - state->FillUnprocessedRanges(ranges, KeyColumnTypes, Settings.GetReverse()); + state->FillUnprocessedRanges(ranges, KeyColumnTypes, Settings->GetReverse()); } TVector 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 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(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 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 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 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 Arena; TVector ResultColumns; TVector KeyColumnTypes; @@ -1436,10 +1441,10 @@ private: void RegisterKqpReadActor(NYql::NDq::TDqAsyncIoFactory& factory, TIntrusivePtr counters) { - factory.RegisterSource( + factory.RegisterSourceWithPtr( 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(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 Alloc; IMemoryQuotaManager::TPtr MemoryQuotaManager; + const google::protobuf::Message* SourceSettings = nullptr; // used only in case if we execute compute actor locally + TIntrusivePtr 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 @@ -20,6 +20,11 @@ concept TSourceCreatorFunc = requires(T f, TProto&& settings, IDqAsyncIoFactory: { f(std::move(settings), std::move(args)) } -> TCastsToAsyncInputPair; }; +template +concept TSourceCreatorFuncMove = requires(T f, TProto* settings, IDqAsyncIoFactory::TSourceArguments&& args) { + { f(settings, std::move(args)) } -> TCastsToAsyncInputPair; +}; + template 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 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(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(), + "Source \"" << type << "\" settings are expected to have protobuf type " << TProtoMsg::descriptor()->full_name() + << ", but got " << settingsAny.type_url()); + TProtoMsg* settingsPtr = args.Arena->Allocate(); + YQL_ENSURE(settingsAny.UnpackTo(settingsPtr), "Failed to unpack settings of type \"" << type << "\""); + return creator(std::move(settingsPtr), std::move(args)); + } + }); + } + template 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& MutableTaskSourceSettings() { + return Task.MutableSourceSettings(); + } + protected: void SetTaskRunner(const TIntrusivePtr& 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& GetSourceSettings() const { + return SourceSettings; + } + + TVector& MutableSourceSettings() { + return SourceSettings; + } + + const TIntrusivePtr& GetArena() const { + return Arena; + } + private: // external callback to retrieve parameter value. @@ -399,6 +411,7 @@ private: NDqProto::TDqTask* Task_ = nullptr; std::unique_ptr HeapTask_; TIntrusivePtr Arena; + TVector SourceSettings; // used only in case if we execute compute actor locally }; class IDqTaskRunner : public TSimpleRefCount, private TNonCopyable { -- cgit v1.3