Skip to content

Commit f5580fd

Browse files
GrigoriyPAHor911kardymondsfedor-mironyumkam
authored
merge to yq stable YQ-3561 support fq run (#15132)
Co-authored-by: Hor911 <[email protected]> Co-authored-by: Dmitry Kardymon <[email protected]> Co-authored-by: Fiodar Miron <[email protected]> Co-authored-by: yumkam <[email protected]>
1 parent ac1ae9a commit f5580fd

File tree

91 files changed

+5530
-1746
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

91 files changed

+5530
-1746
lines changed

‎ydb/core/driver_lib/run/config_helpers.cpp‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
#include "config_helpers.h"
22

3+
#include <ydb/core/base/localdb.h>
4+
#include <ydb/core/protos/bootstrap.pb.h>
5+
#include <ydb/core/protos/resource_broker.pb.h>
6+
37
#include <ydb/library/actors/util/affinity.h>
48

59

@@ -114,4 +118,31 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
114118

115119
} // namespace NActorSystemConfigHelpers
116120

121+
namespace NKikimrConfigHelpers {
122+
123+
NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config) {
124+
NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
125+
auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
126+
if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
127+
resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
128+
}
129+
for (const auto& queue : resourceBrokerConfig.GetQueues()) {
130+
if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
131+
if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
132+
resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
133+
}
134+
}
135+
}
136+
};
137+
if (config.HasBootstrapConfig() && config.GetBootstrapConfig().HasResourceBroker()) {
138+
mergeResourceBrokerConfigs(config.GetBootstrapConfig().GetResourceBroker());
139+
}
140+
if (config.HasResourceBrokerConfig()) {
141+
mergeResourceBrokerConfigs(config.GetResourceBrokerConfig());
142+
}
143+
return resourceBrokerSelfConfig;
144+
}
145+
146+
} // namespace NKikimrConfigHelpers
147+
117148
} // namespace NKikimr

‎ydb/core/driver_lib/run/config_helpers.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#pragma once
22

3+
#include <ydb/core/memory_controller/memory_controller.h>
34
#include <ydb/core/protos/config.pb.h>
45

56
#include <ydb/library/actors/core/config.h>
@@ -15,4 +16,10 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
1516

1617
} // namespace NActorSystemConfigHelpers
1718

19+
namespace NKikimrConfigHelpers {
20+
21+
NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config);
22+
23+
} // namespace NKikimrConfigHelpers
24+
1825
} // namespace NKikimr

‎ydb/core/driver_lib/run/kikimr_services_initializers.cpp‎

Lines changed: 1 addition & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -2049,28 +2049,8 @@ void TMemoryControllerInitializer::InitializeServices(
20492049
NActors::TActorSystemSetup* setup,
20502050
const NKikimr::TAppData* appData)
20512051
{
2052-
NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
2053-
auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
2054-
if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
2055-
resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
2056-
}
2057-
for (const auto& queue : resourceBrokerConfig.GetQueues()) {
2058-
if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
2059-
if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
2060-
resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
2061-
}
2062-
}
2063-
}
2064-
};
2065-
if (Config.HasBootstrapConfig() && Config.GetBootstrapConfig().HasResourceBroker()) {
2066-
mergeResourceBrokerConfigs(Config.GetBootstrapConfig().GetResourceBroker());
2067-
}
2068-
if (Config.HasResourceBrokerConfig()) {
2069-
mergeResourceBrokerConfigs(Config.GetResourceBrokerConfig());
2070-
}
2071-
20722052
auto* actor = NMemory::CreateMemoryController(TDuration::Seconds(1), ProcessMemoryInfoProvider,
2073-
Config.GetMemoryControllerConfig(), resourceBrokerSelfConfig,
2053+
Config.GetMemoryControllerConfig(), NKikimrConfigHelpers::CreateMemoryControllerResourceBrokerConfig(Config),
20742054
appData->Counters);
20752055
setup->LocalServices.emplace_back(
20762056
NMemory::MakeMemoryControllerId(0),

‎ydb/core/fq/libs/actors/clusters_from_connections.cpp‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,19 @@ std::pair<TString, bool> ParseHttpEndpoint(const TString& endpoint) {
8888
return std::make_pair(ToString(host), scheme != "http");
8989
}
9090

91+
std::pair<TString, TIpPort> ParseGrpcEndpoint(const TString& endpoint) {
92+
TStringBuf scheme;
93+
TStringBuf address;
94+
TStringBuf uri;
95+
NHttp::CrackURL(endpoint, scheme, address, uri);
96+
97+
TString hostname;
98+
TIpPort port;
99+
NHttp::CrackAddress(TString(address), hostname, port);
100+
101+
return {hostname, port};
102+
}
103+
91104
void FillSolomonClusterConfig(NYql::TSolomonClusterConfig& clusterConfig,
92105
const TString& name,
93106
const TString& authToken,
@@ -230,8 +243,18 @@ void AddClustersFromConnections(
230243
clusterCfg->SetKind(NYql::EGenericDataSourceKind::YDB);
231244
clusterCfg->SetProtocol(NYql::EGenericProtocol::NATIVE);
232245
clusterCfg->SetName(connectionName);
233-
clusterCfg->SetDatabaseId(db.database_id());
234-
clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
246+
if (const auto& databaseId = db.database_id()) {
247+
clusterCfg->SetDatabaseId(databaseId);
248+
clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
249+
} else {
250+
const auto& [host, port] = ParseGrpcEndpoint(db.endpoint());
251+
252+
auto& endpoint = *clusterCfg->MutableEndpoint();
253+
endpoint.set_host(host);
254+
endpoint.set_port(port);
255+
clusterCfg->SetUseSsl(db.secure());
256+
clusterCfg->SetDatabaseName(db.database());
257+
}
235258
FillClusterAuth(*clusterCfg, db.auth(), authToken, accountIdSignatures);
236259
clusters.emplace(connectionName, GenericProviderName);
237260
break;

‎ydb/core/fq/libs/actors/pending_fetcher.cpp‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
157157
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
158158
const TString& tenantName,
159159
NActors::TMon* monitoring,
160-
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
160+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
161+
NYql::IPqGatewayFactory::TPtr pqGatewayFactory
161162
)
162163
: YqSharedResources(yqSharedResources)
163164
, CredentialsProviderFactory(credentialsProviderFactory)
@@ -180,6 +181,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
180181
, Monitoring(monitoring)
181182
, ComputeConfig(config.GetCompute())
182183
, S3ActorsFactory(std::move(s3ActorsFactory))
184+
, PqGatewayFactory(std::move(pqGatewayFactory))
183185
{
184186
Y_ENSURE(GetYqlDefaultModuleResolverWithContext(ModuleResolver));
185187
}
@@ -472,7 +474,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
472474
NProtoInterop::CastFromProto(task.result_ttl()),
473475
std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end()),
474476
S3ActorsFactory,
475-
ComputeConfig.GetWorkloadManagerConfig(task.scope())
477+
ComputeConfig.GetWorkloadManagerConfig(task.scope()),
478+
PqGatewayFactory
476479
);
477480

478481
auto runActorId =
@@ -548,6 +551,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
548551
NActors::TMon* Monitoring;
549552
TComputeConfig ComputeConfig;
550553
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
554+
NYql::IPqGatewayFactory::TPtr PqGatewayFactory;
551555
};
552556

553557

@@ -567,7 +571,8 @@ NActors::IActor* CreatePendingFetcher(
567571
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
568572
const TString& tenantName,
569573
NActors::TMon* monitoring,
570-
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory)
574+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
575+
NYql::IPqGatewayFactory::TPtr pqGatewayFactory)
571576
{
572577
return new TPendingFetcher(
573578
yqSharedResources,
@@ -585,7 +590,8 @@ NActors::IActor* CreatePendingFetcher(
585590
clientCounters,
586591
tenantName,
587592
monitoring,
588-
std::move(s3ActorsFactory));
593+
std::move(s3ActorsFactory),
594+
std::move(pqGatewayFactory));
589595
}
590596

591597
TActorId MakePendingFetcherId(ui32 nodeId) {

‎ydb/core/fq/libs/actors/proxy.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#include <ydb/library/yql/providers/common/http_gateway/yql_http_gateway.h>
1717
#include <yql/essentials/providers/common/metrics/service_counters.h>
1818
#include <ydb/library/yql/providers/pq/cm_client/client.h>
19+
#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
1920
#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
2021

2122
#include <ydb/public/lib/fq/scope.h>
@@ -53,7 +54,8 @@ NActors::IActor* CreatePendingFetcher(
5354
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
5455
const TString& tenantName,
5556
NActors::TMon* monitoring,
56-
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
57+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
58+
NYql::IPqGatewayFactory::TPtr pqGatewayFactory
5759
);
5860

5961
NActors::IActor* CreateRunActor(

‎ydb/core/fq/libs/actors/run_actor.cpp‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -571,6 +571,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
571571
SelfId(),
572572
Params.QueryId,
573573
Params.YqSharedResources->UserSpaceYdbDriver,
574+
Params.PqGatewayFactory->CreatePqGateway(),
574575
Params.Resources.topic_consumers(),
575576
PrepareReadRuleCredentials()
576577
)
@@ -1435,6 +1436,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
14351436
SelfId(),
14361437
Params.QueryId,
14371438
Params.YqSharedResources->UserSpaceYdbDriver,
1439+
Params.PqGatewayFactory->CreatePqGateway(),
14381440
Params.Resources.topic_consumers(),
14391441
PrepareReadRuleCredentials()
14401442
)
@@ -1970,14 +1972,8 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
19701972
}
19711973

19721974
{
1973-
NYql::TPqGatewayServices pqServices(
1974-
Params.YqSharedResources->UserSpaceYdbDriver,
1975-
Params.PqCmConnections,
1976-
Params.CredentialsFactory,
1977-
std::make_shared<NYql::TPqGatewayConfig>(gatewaysConfig.GetPq()),
1978-
Params.FunctionRegistry
1979-
);
1980-
const auto pqGateway = NYql::CreatePqNativeGateway(pqServices);
1975+
auto pqGateway = Params.PqGatewayFactory->CreatePqGateway();
1976+
pqGateway->UpdateClusterConfigs(std::make_shared<NYql::TPqGatewayConfig>(gatewaysConfig.GetPq()));
19811977
dataProvidersInit.push_back(GetPqDataProviderInitializer(pqGateway, false, dbResolver));
19821978
}
19831979

‎ydb/core/fq/libs/compute/common/run_actor_params.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,8 @@ TRunActorParams::TRunActorParams(
6060
TDuration resultTtl,
6161
std::map<TString, Ydb::TypedValue>&& queryParameters,
6262
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
63-
const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager
63+
const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager,
64+
NYql::IPqGatewayFactory::TPtr pqGatewayFactory
6465
)
6566
: YqSharedResources(yqSharedResources)
6667
, CredentialsProviderFactory(credentialsProviderFactory)
@@ -117,6 +118,7 @@ TRunActorParams::TRunActorParams(
117118
, QueryParameters(std::move(queryParameters))
118119
, S3ActorsFactory(std::move(s3ActorsFactory))
119120
, WorkloadManager(workloadManager)
121+
, PqGatewayFactory(std::move(pqGatewayFactory))
120122
{
121123
}
122124

‎ydb/core/fq/libs/compute/common/run_actor_params.h‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
#include <ydb/library/yql/providers/dq/provider/yql_dq_gateway.h>
1414
#include <ydb/library/yql/providers/dq/worker_manager/interface/counters.h>
1515
#include <ydb/library/yql/providers/pq/cm_client/client.h>
16+
#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
1617
#include <ydb/library/yql/providers/solomon/provider/yql_solomon_gateway.h>
1718
#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
1819

@@ -79,7 +80,8 @@ struct TRunActorParams { // TODO2 : Change name
7980
TDuration resultTtl,
8081
std::map<TString, Ydb::TypedValue>&& queryParameters,
8182
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
82-
const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager
83+
const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager,
84+
NYql::IPqGatewayFactory::TPtr pqGatewayFactory
8385
);
8486

8587
TRunActorParams(const TRunActorParams& params) = default;
@@ -145,6 +147,7 @@ struct TRunActorParams { // TODO2 : Change name
145147
std::map<TString, Ydb::TypedValue> QueryParameters;
146148
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
147149
::NFq::NConfig::TWorkloadManagerConfig WorkloadManager;
150+
NYql::IPqGatewayFactory::TPtr PqGatewayFactory;
148151
};
149152

150153
} /* NFq */

‎ydb/core/fq/libs/config/protos/common.proto‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,4 +32,6 @@ message TCommonConfig {
3232
bool ShowQueryTimeline = 16;
3333
uint64 MaxQueryTimelineSize = 17; // default: 200KB
3434
string PqReconnectPeriod = 18; // default: disabled
35+
uint32 TopicClientHandlersExecutorThreadsNum = 19; // default: 0 that means use default from TopicClientSettings (1)
36+
uint32 TopicClientCompressionExecutorThreadsNum = 20; // default: 0 that means use default from TopicClientSettings (2)
3537
}

0 commit comments

Comments
 (0)