Skip to content

Commit e84ca51

Browse files
andrewstalinandrewstalin
authored
Merge e235b77 into e00f237 
2 parents e00f237 + e235b77 commit e84ca51

Some content is hidden

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

48 files changed

+383
-220
lines changed

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2156,16 +2156,19 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
21562156

21572157
auto federatedQuerySetupFactory = NKqp::MakeKqpFederatedQuerySetupFactory(setup, appData, Config);
21582158

2159+
auto s3ActorsFactory = std::make_shared<NYql::NDq::TS3ActorsFactory>();
21592160
auto proxy = NKqp::CreateKqpProxyService(Config.GetLogConfig(), Config.GetTableServiceConfig(),
21602161
Config.GetQueryServiceConfig(), Config.GetMetadataProviderConfig(), std::move(settings), Factories->QueryReplayBackendFactory, std::move(kqpProxySharedResources),
2161-
federatedQuerySetupFactory
2162+
federatedQuerySetupFactory, s3ActorsFactory
21622163
);
21632164
setup->LocalServices.push_back(std::make_pair(
21642165
NKqp::MakeKqpProxyID(NodeId),
21652166
TActorSetupCmd(proxy, TMailboxType::HTSwap, appData->UserPoolId)));
21662167

21672168
// Create finalize script service
2168-
auto finalize = NKqp::CreateKqpFinalizeScriptService(Config.GetQueryServiceConfig(), Config.GetMetadataProviderConfig(), federatedQuerySetupFactory);
2169+
auto finalize = NKqp::CreateKqpFinalizeScriptService(
2170+
Config.GetQueryServiceConfig(), Config.GetMetadataProviderConfig(), federatedQuerySetupFactory, s3ActorsFactory
2171+
);
21692172
setup->LocalServices.push_back(std::make_pair(
21702173
NKqp::MakeKqpFinalizeScriptServiceId(NodeId),
21712174
TActorSetupCmd(finalize, TMailboxType::HTSwap, appData->UserPoolId)));

‎ydb/core/driver_lib/run/ya.make

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,7 @@ PEERDIR(
138138
ydb/library/yql/providers/yt/codec/codegen
139139
ydb/library/yql/providers/yt/comp_nodes/llvm14
140140
ydb/library/yql/providers/pq/cm_client
141+
ydb/library/yql/providers/s3/actors
141142
ydb/library/yql/public/udf/service/exception_policy
142143
ydb/public/lib/base
143144
ydb/public/lib/deprecated/client

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

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
155155
::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
156156
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
157157
const TString& tenantName,
158-
NActors::TMon* monitoring
158+
NActors::TMon* monitoring,
159+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
159160
)
160161
: YqSharedResources(yqSharedResources)
161162
, CredentialsProviderFactory(credentialsProviderFactory)
@@ -177,6 +178,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
177178
, InternalServiceId(MakeInternalServiceActorId())
178179
, Monitoring(monitoring)
179180
, ComputeConfig(config.GetCompute())
181+
, S3ActorsFactory(std::move(s3ActorsFactory))
180182
{
181183
Y_ENSURE(GetYqlDefaultModuleResolverWithContext(ModuleResolver));
182184
}
@@ -439,7 +441,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
439441
task.operation_id(),
440442
computeConnection,
441443
NProtoInterop::CastFromProto(task.result_ttl()),
442-
std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end())
444+
std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end()),
445+
S3ActorsFactory
443446
);
444447

445448
auto runActorId =
@@ -516,6 +519,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
516519
TActorId InternalServiceId;
517520
NActors::TMon* Monitoring;
518521
TComputeConfig ComputeConfig;
522+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
519523
};
520524

521525

@@ -534,7 +538,8 @@ NActors::IActor* CreatePendingFetcher(
534538
::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
535539
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
536540
const TString& tenantName,
537-
NActors::TMon* monitoring)
541+
NActors::TMon* monitoring,
542+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory)
538543
{
539544
return new TPendingFetcher(
540545
yqSharedResources,
@@ -551,7 +556,8 @@ NActors::IActor* CreatePendingFetcher(
551556
std::move(pqCmConnections),
552557
clientCounters,
553558
tenantName,
554-
monitoring);
559+
monitoring,
560+
std::move(s3ActorsFactory));
555561
}
556562

557563
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 <ydb/library/yql/providers/common/metrics/service_counters.h>
1818
#include <ydb/library/yql/providers/pq/cm_client/client.h>
19+
#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory.h>
1920

2021
#include <ydb/public/lib/fq/scope.h>
2122

@@ -51,7 +52,8 @@ NActors::IActor* CreatePendingFetcher(
5152
::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
5253
const ::NMonitoring::TDynamicCounterPtr& clientCounters,
5354
const TString& tenantName,
54-
NActors::TMon* monitoring
55+
NActors::TMon* monitoring,
56+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
5557
);
5658

5759
NActors::IActor* CreateRunActor(

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
#include <ydb/library/yql/providers/s3/provider/yql_s3_provider.h>
3333
#include <ydb/library/yql/providers/solomon/gateway/yql_solomon_gateway.h>
3434
#include <ydb/library/yql/providers/solomon/provider/yql_solomon_provider.h>
35-
#include <ydb/library/yql/providers/s3/actors/yql_s3_applicator_actor.h>
35+
#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory.h>
3636
#include <ydb/library/yql/providers/s3/proto/sink.pb.h>
3737
#include <ydb/library/yql/sql/settings/translation_settings.h>
3838
#include <ydb/library/yql/minikql/mkql_alloc.h>
@@ -1782,7 +1782,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
17821782
}
17831783
}
17841784

1785-
Register(NYql::NDq::MakeS3ApplicatorActor(SelfId()
1785+
Register(Params.S3ActorsFactory->CreateS3ApplicatorActor(SelfId()
17861786
, Params.S3Gateway
17871787
, Params.QueryId
17881788
, Params.JobId

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,8 @@ TRunActorParams::TRunActorParams(
5757
const TString& operationId,
5858
const NFq::NConfig::TYdbStorageConfig& computeConnection,
5959
TDuration resultTtl,
60-
std::map<TString, Ydb::TypedValue>&& queryParameters
60+
std::map<TString, Ydb::TypedValue>&& queryParameters,
61+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
6162
)
6263
: YqSharedResources(yqSharedResources)
6364
, CredentialsProviderFactory(credentialsProviderFactory)
@@ -111,6 +112,7 @@ TRunActorParams::TRunActorParams(
111112
, ComputeConnection(computeConnection)
112113
, ResultTtl(resultTtl)
113114
, QueryParameters(std::move(queryParameters))
115+
, S3ActorsFactory(std::move(s3ActorsFactory))
114116
{
115117
}
116118

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
#include <ydb/library/yql/providers/dq/worker_manager/interface/counters.h>
1515
#include <ydb/library/yql/providers/pq/cm_client/client.h>
1616
#include <ydb/library/yql/providers/solomon/provider/yql_solomon_gateway.h>
17+
#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory.h>
1718

1819
#include <ydb/public/lib/fq/scope.h>
1920

@@ -75,7 +76,8 @@ struct TRunActorParams { // TODO2 : Change name
7576
const TString& operationId,
7677
const ::NFq::NConfig::TYdbStorageConfig& computeConnection,
7778
TDuration resultTtl,
78-
std::map<TString, Ydb::TypedValue>&& queryParameters
79+
std::map<TString, Ydb::TypedValue>&& queryParameters,
80+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
7981
);
8082

8183
TRunActorParams(const TRunActorParams& params) = default;
@@ -138,6 +140,7 @@ struct TRunActorParams { // TODO2 : Change name
138140
::NFq::NConfig::TYdbStorageConfig ComputeConnection;
139141
TDuration ResultTtl;
140142
std::map<TString, Ydb::TypedValue> QueryParameters;
143+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
141144
};
142145

143146
} /* NFq */

‎ydb/core/fq/libs/compute/common/ya.make

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ PEERDIR(
1717
ydb/library/yql/providers/dq/provider
1818
ydb/library/yql/providers/generic/connector/api/service/protos
1919
ydb/library/yql/providers/generic/connector/libcpp
20+
ydb/library/yql/providers/common/http_gateway
2021
)
2122

2223
YQL_LAST_ABI_VERSION()

‎ydb/core/fq/libs/init/init.cpp

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,6 @@
3939
#include <ydb/library/yql/providers/dq/task_runner/tasks_runner_local.h>
4040
#include <ydb/library/yql/providers/dq/worker_manager/local_worker_manager.h>
4141
#include <ydb/library/yql/providers/generic/actors/yql_generic_provider_factories.h>
42-
#include <ydb/library/yql/providers/s3/actors/yql_s3_sink_factory.h>
43-
#include <ydb/library/yql/providers/s3/actors/yql_s3_source_factory.h>
4442
#include <ydb/library/yql/providers/s3/proto/retry_config.pb.h>
4543
#include <ydb/library/yql/providers/pq/async_io/dq_pq_read_actor.h>
4644
#include <ydb/library/yql/providers/pq/async_io/dq_pq_write_actor.h>
@@ -194,6 +192,8 @@ void Init(
194192
credentialsFactory = NYql::CreateSecuredServiceAccountCredentialsOverTokenAccessorFactory(tokenAccessorConfig.GetEndpoint(), tokenAccessorConfig.GetUseSsl(), caContent, tokenAccessorConfig.GetConnectionPoolSize());
195193
}
196194

195+
auto s3ActorsFactory = std::make_shared<NYql::NDq::TS3ActorsFactory>();
196+
197197
if (protoConfig.GetPrivateApi().GetEnabled()) {
198198
const auto& s3readConfig = protoConfig.GetReadActorsFactoryConfig().GetS3ReadActorFactoryConfig();
199199
auto s3HttpRetryPolicy = NYql::GetHTTPDefaultRetryPolicy(NYql::THttpRetryPolicyOptions{.MaxTime = TDuration::Max(), .RetriedCurlCodes = NYql::FqRetriedCurlCodes()});
@@ -224,12 +224,13 @@ void Init(
224224
RegisterDqInputTransformLookupActorFactory(*asyncIoFactory);
225225
RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
226226
RegisterYdbReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
227-
RegisterS3ReadActorFactory(*asyncIoFactory, credentialsFactory, httpGateway, s3HttpRetryPolicy, readActorFactoryCfg,
227+
228+
s3ActorsFactory->RegisterS3ReadActorFactory(*asyncIoFactory, credentialsFactory, httpGateway, s3HttpRetryPolicy, readActorFactoryCfg,
228229
yqCounters->GetSubgroup("subsystem", "S3ReadActor"));
229-
RegisterS3WriteActorFactory(*asyncIoFactory, credentialsFactory,
230+
s3ActorsFactory->RegisterS3WriteActorFactory(*asyncIoFactory, credentialsFactory,
230231
httpGateway, s3HttpRetryPolicy);
231-
RegisterGenericProviderFactories(*asyncIoFactory, credentialsFactory, connectorClient);
232232

233+
RegisterGenericProviderFactories(*asyncIoFactory, credentialsFactory, connectorClient);
233234
RegisterDqPqWriteActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory, yqCounters->GetSubgroup("subsystem", "DqSinkTracker"));
234235
RegisterDQSolomonWriteActorFactory(*asyncIoFactory, credentialsFactory);
235236
}
@@ -339,7 +340,8 @@ void Init(
339340
std::move(pqCmConnections),
340341
clientCounters,
341342
tenant,
342-
appData->Mon
343+
appData->Mon,
344+
s3ActorsFactory
343345
);
344346

345347
actorRegistrator(MakePendingFetcherId(nodeId), fetcher);

‎ydb/core/kqp/compute_actor/kqp_compute_actor.cpp

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,6 @@
1010
#include <ydb/core/kqp/runtime/kqp_sequencer_factory.h>
1111
#include <ydb/core/kqp/runtime/kqp_stream_lookup_factory.h>
1212
#include <ydb/library/yql/providers/generic/actors/yql_generic_provider_factories.h>
13-
#include <ydb/library/yql/providers/s3/actors/yql_s3_sink_factory.h>
14-
#include <ydb/library/yql/providers/s3/actors/yql_s3_source_factory.h>
1513
#include <ydb/core/formats/arrow/protos/ssa.pb.h>
1614
#include <ydb/library/yql/dq/proto/dq_tasks.pb.h>
1715

@@ -69,7 +67,9 @@ namespace NKqp {
6967

7068
NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
7169
TIntrusivePtr<TKqpCounters> counters,
72-
std::optional<TKqpFederatedQuerySetup> federatedQuerySetup) {
70+
std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
71+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
72+
) {
7373
auto factory = MakeIntrusive<NYql::NDq::TDqAsyncIoFactory>();
7474
RegisterStreamLookupActorFactory(*factory, counters);
7575
RegisterKqpReadActor(*factory, counters);
@@ -78,8 +78,8 @@ NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
7878

7979
if (federatedQuerySetup) {
8080
auto s3HttpRetryPolicy = NYql::GetHTTPDefaultRetryPolicy(NYql::THttpRetryPolicyOptions{.RetriedCurlCodes = NYql::FqRetriedCurlCodes()});
81-
RegisterS3ReadActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
82-
RegisterS3WriteActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
81+
s3ActorsFactory->RegisterS3ReadActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
82+
s3ActorsFactory->RegisterS3WriteActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
8383

8484
if (federatedQuerySetup->ConnectorClient) {
8585
RegisterGenericProviderFactories(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->ConnectorClient);

‎ydb/core/kqp/compute_actor/kqp_compute_actor.h

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
#include <ydb/core/scheme/scheme_tabledefs.h>
77
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
88
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h>
9+
#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory.h>
910

1011
namespace NKikimr {
1112

@@ -59,7 +60,9 @@ IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vect
5960
const ui64 txId, const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId);
6061

6162
NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
62-
TIntrusivePtr<TKqpCounters> counters, std::optional<TKqpFederatedQuerySetup> federatedQuerySetup);
63+
TIntrusivePtr<TKqpCounters> counters,
64+
std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
65+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory);
6366

6467
} // namespace NKqp
6568
} // namespace NKikimr

‎ydb/core/kqp/compute_actor/ya.make

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@ PEERDIR(
2424
ydb/core/formats/arrow/protos
2525
ydb/library/yql/dq/actors/compute
2626
ydb/library/yql/providers/generic/actors
27-
ydb/library/yql/providers/s3/actors
2827
ydb/library/yql/public/issue
2928
)
3029

‎ydb/core/kqp/finalize_script_service/kqp_finalize_script_actor.cpp

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,8 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
2323
TScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
2424
const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
2525
const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
26-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup)
26+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
27+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactor)
2728
: ReplyActor(request->Sender)
2829
, ExecutionId(request->Get()->Description.ExecutionId)
2930
, Database(request->Get()->Description.Database)
@@ -33,6 +34,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
3334
, MaximalSecretsSnapshotWaitTime(2 * TDuration::Seconds(metadataProviderConfig.GetRefreshPeriodSeconds()))
3435
, FederatedQuerySetup(federatedQuerySetup)
3536
, Compressor(queryServiceConfig.GetQueryArtifactsCompressionMethod(), queryServiceConfig.GetQueryArtifactsCompressionMinSize())
37+
, S3ActorsFactor(std::move(s3ActorsFactor))
3638
{}
3739

3840
void CompressScriptArtifacts() const {
@@ -166,7 +168,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
166168
return;
167169
}
168170

169-
Register(NYql::NDq::MakeS3ApplicatorActor(
171+
Register(S3ActorsFactor->CreateS3ApplicatorActor(
170172
SelfId(),
171173
FederatedQuerySetup->HttpGateway,
172174
CreateGuidAsString(),
@@ -230,6 +232,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
230232
const TDuration MaximalSecretsSnapshotWaitTime;
231233
const std::optional<TKqpFederatedQuerySetup>& FederatedQuerySetup;
232234
const NFq::TCompressor Compressor;
235+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactor;
233236

234237
TString CustomerSuppliedId;
235238
std::vector<NKqpProto::TKqpExternalSink> Sinks;
@@ -244,8 +247,10 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
244247
IActor* CreateScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
245248
const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
246249
const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
247-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup) {
248-
return new TScriptFinalizerActor(std::move(request), queryServiceConfig, metadataProviderConfig, federatedQuerySetup);
250+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
251+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
252+
) {
253+
return new TScriptFinalizerActor(std::move(request), queryServiceConfig, metadataProviderConfig, federatedQuerySetup, std::move(s3ActorsFactory));
249254
}
250255

251256
} // namespace NKikimr::NKqp

‎ydb/core/kqp/finalize_script_service/kqp_finalize_script_actor.h

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,15 @@
22

33
#include <ydb/core/kqp/common/events/script_executions.h>
44
#include <ydb/core/kqp/federated_query/kqp_federated_query_helpers.h>
5+
#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory.h>
56

67

78
namespace NKikimr::NKqp {
89

910
IActor* CreateScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
1011
const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
1112
const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
12-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup);
13+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
14+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory);
1315

1416
} // namespace NKikimr::NKqp

‎ydb/core/kqp/finalize_script_service/kqp_finalize_script_service.cpp

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,12 @@ class TKqpFinalizeScriptService : public TActorBootstrapped<TKqpFinalizeScriptSe
1414
public:
1515
TKqpFinalizeScriptService(const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
1616
const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
17-
IKqpFederatedQuerySetupFactory::TPtr federatedQuerySetupFactory)
17+
IKqpFederatedQuerySetupFactory::TPtr federatedQuerySetupFactory,
18+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory)
1819
: QueryServiceConfig(queryServiceConfig)
1920
, MetadataProviderConfig(metadataProviderConfig)
2021
, FederatedQuerySetupFactory(federatedQuerySetupFactory)
22+
, S3ActorsFactory(std::move(s3ActorsFactory))
2123
{}
2224

2325
void Bootstrap(const TActorContext &ctx) {
@@ -81,7 +83,8 @@ class TKqpFinalizeScriptService : public TActorBootstrapped<TKqpFinalizeScriptSe
8183
std::move(request),
8284
QueryServiceConfig,
8385
MetadataProviderConfig,
84-
FederatedQuerySetup
86+
FederatedQuerySetup,
87+
S3ActorsFactory
8588
));
8689
}
8790

@@ -131,14 +134,17 @@ class TKqpFinalizeScriptService : public TActorBootstrapped<TKqpFinalizeScriptSe
131134
ui32 FinalizationRequestsInFlight = 0;
132135
std::queue<TString> WaitingFinalizationExecutions;
133136
std::unordered_map<TString, std::vector<TEvScriptFinalizeRequest::TPtr>> FinalizationRequestsQueue;
137+
138+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
134139
};
135140

136141
} // anonymous namespace
137142

138143
IActor* CreateKqpFinalizeScriptService(const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
139144
const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
140-
IKqpFederatedQuerySetupFactory::TPtr federatedQuerySetupFactory) {
141-
return new TKqpFinalizeScriptService(queryServiceConfig, metadataProviderConfig, std::move(federatedQuerySetupFactory));
145+
IKqpFederatedQuerySetupFactory::TPtr federatedQuerySetupFactory,
146+
std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory) {
147+
return new TKqpFinalizeScriptService(queryServiceConfig, metadataProviderConfig, std::move(federatedQuerySetupFactory), std::move(s3ActorsFactory));
142148
}
143149

144150
} // namespace NKikimr::NKqp

0 commit comments

Comments
 (0)