diff options
| -rw-r--r-- | ydb/library/yql/providers/dq/global_worker_manager/global_worker_manager_ut.cpp | 29 | ||||
| -rw-r--r-- | ydb/library/yql/providers/dq/global_worker_manager/ut/ya.make | 7 |
2 files changed, 36 insertions, 0 deletions
diff --git a/ydb/library/yql/providers/dq/global_worker_manager/global_worker_manager_ut.cpp b/ydb/library/yql/providers/dq/global_worker_manager/global_worker_manager_ut.cpp index 07438e01c69..6a1913f578e 100644 --- a/ydb/library/yql/providers/dq/global_worker_manager/global_worker_manager_ut.cpp +++ b/ydb/library/yql/providers/dq/global_worker_manager/global_worker_manager_ut.cpp @@ -10,6 +10,13 @@ #include <ydb/library/yql/providers/dq/common/attrs.h> #include <ydb/library/yql/providers/dq/actors/dynamic_nameserver.h> #include <ydb/library/yql/providers/dq/actors/resource_allocator.h> +#include <ydb/library/yql/minikql/invoke_builtins/mkql_builtins.h> +#include <ydb/library/yql/dq/integration/transform/yql_dq_task_transform.h> +#include <ydb/library/yql/dq/comp_nodes/yql_common_dq_factory.h> +#include <ydb/library/yql/providers/common/comp_nodes/yql_factory.h> +#include <ydb/library/yql/minikql/comp_nodes/mkql_factories.h> +#include <ydb/library/yql/dq/transform/yql_common_dq_transform.h> +#include <ydb/library/yql/providers/dq/task_runner/tasks_runner_local.h> using namespace NYql; using namespace NActors; @@ -178,11 +185,26 @@ public: TActorSetupCmd{gwmActor, TMailboxType::Simple, 0}); // Local WM. + FunctionRegistry_ = CreateFunctionRegistry(NKikimr::NMiniKQL::CreateBuiltinRegistry()); + auto dqCompFactory = NKikimr::NMiniKQL::GetCompositeWithBuiltinFactory({ + NYql::GetCommonDqFactory(), + NKikimr::NMiniKQL::GetYqlFactory() + }); + + auto dqTaskTransformFactory = NYql::CreateCompositeTaskTransformFactory({ + NYql::CreateCommonDqTaskTransformFactory() + }); + + auto patternCache = std::make_shared<NKikimr::NMiniKQL::TComputationPatternLRUCache>(NKikimr::NMiniKQL::TComputationPatternLRUCache::Config(200_MB, 200_MB)); + + auto factory = NTaskRunnerProxy::CreateFactory(FunctionRegistry_.Get(), dqCompFactory, dqTaskTransformFactory, patternCache, true); for (ui32 i = 1; i < nodesNumber; i++) { NYql::NDqs::TLocalWorkerManagerOptions lwmOptions; lwmOptions.TaskRunnerInvokerFactory = new NDqs::TTaskRunnerInvokerFactory(); lwmOptions.TaskRunnerActorFactory = NYql::NDq::NTaskRunnerActor::CreateTaskRunnerActorFactory( lwmOptions.Factory, lwmOptions.TaskRunnerInvokerFactory); + lwmOptions.FunctionRegistry = FunctionRegistry_.Get(); + lwmOptions.Factory = factory; auto localWM = CreateLocalWorkerManager(lwmOptions); ActorRuntime_->AddLocalService(MakeWorkerManagerActorID(NodeId(i)), TActorSetupCmd{localWM, TMailboxType::Simple, 0}, i); @@ -192,6 +214,10 @@ public: ActorRuntime_->Initialize(); + for (ui32 i = 1; i < nodesNumber; i++) { + ActorRuntime_->GetLogSettings(i)->Mask = 0xffffffff; + } + NActors::TDispatchOptions options; options.FinalEvents.emplace_back(NActors::TEvents::TSystem::Bootstrap, nodesNumber); ActorRuntime_->DispatchEvents(options); @@ -360,6 +386,8 @@ public: TActorId RegisterResourceAllocator(const ui32 workersCount, const TActorId& execActor) const { TIntrusivePtr<NMonitoring::TDynamicCounters> counters = MakeIntrusive<NMonitoring::TDynamicCounters>(); auto gwmActor = MakeWorkerManagerActorID(NodeId()); + TVector<NYql::NDqProto::TDqTask> tasks(workersCount); + //auto allocator = CreateResourceAllocator(gwmActor, execActor, execActor, workersCount, "TraceId", new TDqConfiguration(), counters, tasks, "sync"); auto allocator = CreateResourceAllocator(gwmActor, execActor, execActor, workersCount, "TraceId", new TDqConfiguration(), counters); const auto allocatorId = ActorRuntime_->Register(allocator); return allocatorId; @@ -434,6 +462,7 @@ public: } THolder<NActors::TTestActorRuntimeBase> ActorRuntime_; + TIntrusivePtr<NKikimr::NMiniKQL::IFunctionRegistry> FunctionRegistry_; }; UNIT_TEST_SUITE_REGISTRATION(TGlobalWorkerManagerTest) diff --git a/ydb/library/yql/providers/dq/global_worker_manager/ut/ya.make b/ydb/library/yql/providers/dq/global_worker_manager/ut/ya.make index b1ae373296d..1a24af97ff2 100644 --- a/ydb/library/yql/providers/dq/global_worker_manager/ut/ya.make +++ b/ydb/library/yql/providers/dq/global_worker_manager/ut/ya.make @@ -13,6 +13,13 @@ PEERDIR( ydb/library/yql/dq/actors/compute ydb/library/yql/minikql/computation/llvm14 ydb/library/yql/minikql/comp_nodes/llvm14 + + ydb/library/yql/dq/integration/transform + ydb/library/yql/dq/comp_nodes + ydb/library/yql/providers/common/comp_nodes + ydb/library/yql/minikql/comp_nodes + ydb/library/yql/dq/transform + ydb/library/yql/providers/dq/task_runner ) SRCS( |