Remove the dependency on the S3 provider implementation

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -169,6 +169,8 @@
 #include <ydb/library/folder_service/folder_service.h>
 #include <ydb/library/folder_service/proto/config.pb.h>
 
+#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory_impl.h>
+
 #include <ydb/library/yql/minikql/comp_nodes/mkql_factories.h>
 #include <ydb/library/yql/parser/pg_wrapper/interface/comp_factory.h>
 #include <ydb/library/yql/utils/actor_log/log.h>
@@ -2146,16 +2148,19 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
 
         auto federatedQuerySetupFactory = NKqp::MakeKqpFederatedQuerySetupFactory(setup, appData, Config);
 
+        auto s3ActorsFactory = NYql::NDq::CreateS3ActorsFactory();
         auto proxy = NKqp::CreateKqpProxyService(Config.GetLogConfig(), Config.GetTableServiceConfig(),
             Config.GetQueryServiceConfig(),  Config.GetMetadataProviderConfig(), std::move(settings), Factories->QueryReplayBackendFactory, std::move(kqpProxySharedResources),
-            federatedQuerySetupFactory
+            federatedQuerySetupFactory, s3ActorsFactory
         );
         setup->LocalServices.push_back(std::make_pair(
             NKqp::MakeKqpProxyID(NodeId),
             TActorSetupCmd(proxy, TMailboxType::HTSwap, appData->UserPoolId)));
 
         // Create finalize script service
-        auto finalize = NKqp::CreateKqpFinalizeScriptService(Config.GetQueryServiceConfig(), Config.GetMetadataProviderConfig(), federatedQuerySetupFactory);
+        auto finalize = NKqp::CreateKqpFinalizeScriptService(
+            Config.GetQueryServiceConfig(), Config.GetMetadataProviderConfig(), federatedQuerySetupFactory, s3ActorsFactory
+        );
         setup->LocalServices.push_back(std::make_pair(
             NKqp::MakeKqpFinalizeScriptServiceId(NodeId),
             TActorSetupCmd(finalize, TMailboxType::HTSwap, appData->UserPoolId)));

ydb/core/driver_lib/run/ya.make

@@ -138,6 +138,7 @@ PEERDIR(
     ydb/library/yql/providers/yt/codec/codegen
     ydb/library/yql/providers/yt/comp_nodes/llvm14
     ydb/library/yql/providers/pq/cm_client
+    ydb/library/yql/providers/s3/actors
     ydb/library/yql/public/udf/service/exception_policy
     ydb/public/lib/base
     ydb/public/lib/deprecated/client

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

@@ -155,7 +155,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         ::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
         const ::NMonitoring::TDynamicCounterPtr& clientCounters,
         const TString& tenantName,
-        NActors::TMon* monitoring
+        NActors::TMon* monitoring,
+        std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
         )
         : YqSharedResources(yqSharedResources)
         , CredentialsProviderFactory(credentialsProviderFactory)
@@ -177,6 +178,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         , InternalServiceId(MakeInternalServiceActorId())
         , Monitoring(monitoring)
         , ComputeConfig(config.GetCompute())
+        , S3ActorsFactory(std::move(s3ActorsFactory))
     {
         Y_ENSURE(GetYqlDefaultModuleResolverWithContext(ModuleResolver));
     }
@@ -441,7 +443,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
             task.operation_id(),
             computeConnection,
             NProtoInterop::CastFromProto(task.result_ttl()),
-            std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end())
+            std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end()),
+            S3ActorsFactory
             );
 
         auto runActorId =
@@ -518,6 +521,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
     TActorId InternalServiceId;
     NActors::TMon* Monitoring;
     TComputeConfig ComputeConfig;
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
 };
 
 
@@ -536,7 +540,8 @@ NActors::IActor* CreatePendingFetcher(
     ::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
     const ::NMonitoring::TDynamicCounterPtr& clientCounters,
     const TString& tenantName,
-    NActors::TMon* monitoring)
+    NActors::TMon* monitoring,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory)
 {
     return new TPendingFetcher(
         yqSharedResources,
@@ -553,7 +558,8 @@ NActors::IActor* CreatePendingFetcher(
         std::move(pqCmConnections),
         clientCounters,
         tenantName,
-        monitoring);
+        monitoring,
+        std::move(s3ActorsFactory));
 }
 
 TActorId MakePendingFetcherId(ui32 nodeId) {

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

@@ -16,6 +16,7 @@
 #include <ydb/library/yql/providers/common/http_gateway/yql_http_gateway.h>
 #include <ydb/library/yql/providers/common/metrics/service_counters.h>
 #include <ydb/library/yql/providers/pq/cm_client/client.h>
+#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
 
 #include <ydb/public/lib/fq/scope.h>
 
@@ -51,7 +52,8 @@ NActors::IActor* CreatePendingFetcher(
     ::NPq::NConfigurationManager::IConnections::TPtr pqCmConnections,
     const ::NMonitoring::TDynamicCounterPtr& clientCounters,
     const TString& tenantName,
-    NActors::TMon* monitoring
+    NActors::TMon* monitoring,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
     );
 
 NActors::IActor* CreateRunActor(

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

@@ -32,7 +32,6 @@
 #include <ydb/library/yql/providers/s3/provider/yql_s3_provider.h>
 #include <ydb/library/yql/providers/solomon/gateway/yql_solomon_gateway.h>
 #include <ydb/library/yql/providers/solomon/provider/yql_solomon_provider.h>
-#include <ydb/library/yql/providers/s3/actors/yql_s3_applicator_actor.h>
 #include <ydb/library/yql/providers/s3/proto/sink.pb.h>
 #include <ydb/library/yql/sql/settings/translation_settings.h>
 #include <ydb/library/yql/minikql/mkql_alloc.h>
@@ -1782,7 +1781,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
                     }
                 }
 
-                Register(NYql::NDq::MakeS3ApplicatorActor(SelfId()
+                Register(Params.S3ActorsFactory->CreateS3ApplicatorActor(SelfId()
                                             , Params.S3Gateway
                                             , Params.QueryId
                                             , Params.JobId

ydb/core/fq/libs/actors/ya.make

@@ -79,6 +79,7 @@ PEERDIR(
     ydb/library/yql/providers/pq/cm_client
     ydb/library/yql/providers/pq/provider
     ydb/library/yql/providers/pq/task_meta
+    ydb/library/yql/providers/s3/actors_factory
     ydb/library/yql/providers/s3/provider
     ydb/library/yql/public/issue
     ydb/library/yql/public/issue/protos

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

@@ -57,7 +57,8 @@ TRunActorParams::TRunActorParams(
     const TString& operationId,
     const NFq::NConfig::TYdbStorageConfig& computeConnection,
     TDuration resultTtl,
-    std::map<TString, Ydb::TypedValue>&& queryParameters
+    std::map<TString, Ydb::TypedValue>&& queryParameters,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
     )
     : YqSharedResources(yqSharedResources)
     , CredentialsProviderFactory(credentialsProviderFactory)
@@ -111,6 +112,7 @@ TRunActorParams::TRunActorParams(
     , ComputeConnection(computeConnection)
     , ResultTtl(resultTtl)
     , QueryParameters(std::move(queryParameters))
+    , S3ActorsFactory(std::move(s3ActorsFactory))
     {
     }
 

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

@@ -14,6 +14,7 @@
 #include <ydb/library/yql/providers/dq/worker_manager/interface/counters.h>
 #include <ydb/library/yql/providers/pq/cm_client/client.h>
 #include <ydb/library/yql/providers/solomon/provider/yql_solomon_gateway.h>
+#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
 
 #include <ydb/public/lib/fq/scope.h>
 
@@ -75,7 +76,8 @@ struct TRunActorParams { // TODO2 : Change name
         const TString& operationId,
         const ::NFq::NConfig::TYdbStorageConfig& computeConnection,
         TDuration resultTtl,
-        std::map<TString, Ydb::TypedValue>&& queryParameters
+        std::map<TString, Ydb::TypedValue>&& queryParameters,
+        std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
     );
 
     TRunActorParams(const TRunActorParams& params) = default;
@@ -138,6 +140,7 @@ struct TRunActorParams { // TODO2 : Change name
     ::NFq::NConfig::TYdbStorageConfig ComputeConnection;
     TDuration ResultTtl;
     std::map<TString, Ydb::TypedValue> QueryParameters;
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
 };
 
 } /* NFq */

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

@@ -14,9 +14,11 @@ PEERDIR(
     ydb/core/fq/libs/grpc
     ydb/core/fq/libs/shared_resources
     ydb/library/yql/public/issue
+    ydb/library/yql/providers/common/http_gateway
     ydb/library/yql/providers/dq/provider
     ydb/library/yql/providers/generic/connector/api/service/protos
     ydb/library/yql/providers/generic/connector/libcpp
+    ydb/library/yql/providers/s3/actors_factory
 )
 
 YQL_LAST_ABI_VERSION()

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

@@ -39,8 +39,7 @@
 #include <ydb/library/yql/providers/dq/task_runner/tasks_runner_local.h>
 #include <ydb/library/yql/providers/dq/worker_manager/local_worker_manager.h>
 #include <ydb/library/yql/providers/generic/actors/yql_generic_provider_factories.h>
-#include <ydb/library/yql/providers/s3/actors/yql_s3_sink_factory.h>
-#include <ydb/library/yql/providers/s3/actors/yql_s3_source_factory.h>
+#include <ydb/library/yql/providers/s3/actors/yql_s3_actors_factory_impl.h>
 #include <ydb/library/yql/providers/s3/proto/retry_config.pb.h>
 #include <ydb/library/yql/providers/pq/async_io/dq_pq_read_actor.h>
 #include <ydb/library/yql/providers/pq/async_io/dq_pq_write_actor.h>
@@ -194,6 +193,8 @@ void Init(
         credentialsFactory = NYql::CreateSecuredServiceAccountCredentialsOverTokenAccessorFactory(tokenAccessorConfig.GetEndpoint(), tokenAccessorConfig.GetUseSsl(), caContent, tokenAccessorConfig.GetConnectionPoolSize());
     }
 
+    auto s3ActorsFactory = NYql::NDq::CreateS3ActorsFactory();
+
     if (protoConfig.GetPrivateApi().GetEnabled()) {
         const auto& s3readConfig = protoConfig.GetReadActorsFactoryConfig().GetS3ReadActorFactoryConfig();
         auto s3HttpRetryPolicy = NYql::GetHTTPDefaultRetryPolicy(NYql::THttpRetryPolicyOptions{.MaxTime = TDuration::Max(), .RetriedCurlCodes = NYql::FqRetriedCurlCodes()});
@@ -224,12 +225,13 @@ void Init(
         RegisterDqInputTransformLookupActorFactory(*asyncIoFactory);
         RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
         RegisterYdbReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory);
-        RegisterS3ReadActorFactory(*asyncIoFactory, credentialsFactory, httpGateway, s3HttpRetryPolicy, readActorFactoryCfg,
+
+        s3ActorsFactory->RegisterS3ReadActorFactory(*asyncIoFactory, credentialsFactory, httpGateway, s3HttpRetryPolicy, readActorFactoryCfg,
             yqCounters->GetSubgroup("subsystem", "S3ReadActor"));
-        RegisterS3WriteActorFactory(*asyncIoFactory, credentialsFactory,
+        s3ActorsFactory->RegisterS3WriteActorFactory(*asyncIoFactory, credentialsFactory,
             httpGateway, s3HttpRetryPolicy);
-        RegisterGenericProviderFactories(*asyncIoFactory, credentialsFactory, connectorClient);
 
+        RegisterGenericProviderFactories(*asyncIoFactory, credentialsFactory, connectorClient);
         RegisterDqPqWriteActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory, yqCounters->GetSubgroup("subsystem", "DqSinkTracker"));
         RegisterDQSolomonWriteActorFactory(*asyncIoFactory, credentialsFactory);
     }
@@ -339,7 +341,8 @@ void Init(
             std::move(pqCmConnections),
             clientCounters,
             tenant,
-            appData->Mon
+            appData->Mon,
+            s3ActorsFactory
             );
 
         actorRegistrator(MakePendingFetcherId(nodeId), fetcher);

ydb/core/kqp/compute_actor/kqp_compute_actor.cpp

@@ -10,8 +10,6 @@
 #include <ydb/core/kqp/runtime/kqp_sequencer_factory.h>
 #include <ydb/core/kqp/runtime/kqp_stream_lookup_factory.h>
 #include <ydb/library/yql/providers/generic/actors/yql_generic_provider_factories.h>
-#include <ydb/library/yql/providers/s3/actors/yql_s3_sink_factory.h>
-#include <ydb/library/yql/providers/s3/actors/yql_s3_source_factory.h>
 #include <ydb/core/formats/arrow/protos/ssa.pb.h>
 #include <ydb/library/yql/dq/proto/dq_tasks.pb.h>
 
@@ -69,7 +67,9 @@ namespace NKqp {
 
 NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
     TIntrusivePtr<TKqpCounters> counters,
-    std::optional<TKqpFederatedQuerySetup> federatedQuerySetup) {
+    std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
+    ) {
     auto factory = MakeIntrusive<NYql::NDq::TDqAsyncIoFactory>();
     RegisterStreamLookupActorFactory(*factory, counters);
     RegisterKqpReadActor(*factory, counters);
@@ -78,8 +78,8 @@ NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
 
     if (federatedQuerySetup) {
         auto s3HttpRetryPolicy = NYql::GetHTTPDefaultRetryPolicy(NYql::THttpRetryPolicyOptions{.RetriedCurlCodes = NYql::FqRetriedCurlCodes()});
-        RegisterS3ReadActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
-        RegisterS3WriteActorFactory(*factory,  federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
+        s3ActorsFactory->RegisterS3ReadActorFactory(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
+        s3ActorsFactory->RegisterS3WriteActorFactory(*factory,  federatedQuerySetup->CredentialsFactory, federatedQuerySetup->HttpGateway, s3HttpRetryPolicy);
 
         if (federatedQuerySetup->ConnectorClient) {
             RegisterGenericProviderFactories(*factory, federatedQuerySetup->CredentialsFactory, federatedQuerySetup->ConnectorClient);

ydb/core/kqp/compute_actor/kqp_compute_actor.h

@@ -6,6 +6,7 @@
 #include <ydb/core/scheme/scheme_tabledefs.h>
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h>
+#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
 
 namespace NKikimr {
 
@@ -59,7 +60,9 @@ IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vect
     const ui64 txId, const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId);
 
 NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
-    TIntrusivePtr<TKqpCounters> counters, std::optional<TKqpFederatedQuerySetup> federatedQuerySetup);
+    TIntrusivePtr<TKqpCounters> counters,
+    std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory);
 
 } // namespace NKqp
 } // namespace NKikimr

ydb/core/kqp/compute_actor/ya.make

@@ -25,7 +25,7 @@ PEERDIR(
     ydb/core/formats/arrow/protos
     ydb/library/yql/dq/actors/compute
     ydb/library/yql/providers/generic/actors
-    ydb/library/yql/providers/s3/actors
+    ydb/library/yql/providers/s3/actors_factory
     ydb/library/yql/public/issue
 )
 

ydb/core/kqp/finalize_script_service/kqp_finalize_script_actor.cpp

@@ -23,7 +23,8 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
     TScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
         const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
         const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
-        const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup)
+        const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
+        std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactor)
         : ReplyActor(request->Sender)
         , ExecutionId(request->Get()->Description.ExecutionId)
         , Database(request->Get()->Description.Database)
@@ -33,6 +34,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
         , MaximalSecretsSnapshotWaitTime(2 * TDuration::Seconds(metadataProviderConfig.GetRefreshPeriodSeconds()))
         , FederatedQuerySetup(federatedQuerySetup)
         , Compressor(queryServiceConfig.GetQueryArtifactsCompressionMethod(), queryServiceConfig.GetQueryArtifactsCompressionMinSize())
+        , S3ActorsFactor(std::move(s3ActorsFactor))
     {}
 
     void CompressScriptArtifacts() const {
@@ -166,7 +168,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
             return;
         }
 
-        Register(NYql::NDq::MakeS3ApplicatorActor(
+        Register(S3ActorsFactor->CreateS3ApplicatorActor(
             SelfId(),
             FederatedQuerySetup->HttpGateway,
             CreateGuidAsString(),
@@ -230,6 +232,7 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
     const TDuration MaximalSecretsSnapshotWaitTime;
     const std::optional<TKqpFederatedQuerySetup>& FederatedQuerySetup;
     const NFq::TCompressor Compressor;
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactor;
 
     TString CustomerSuppliedId;
     std::vector<NKqpProto::TKqpExternalSink> Sinks;
@@ -244,8 +247,10 @@ class TScriptFinalizerActor : public TActorBootstrapped<TScriptFinalizerActor> {
 IActor* CreateScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
     const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
     const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
-    const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup) {
-    return new TScriptFinalizerActor(std::move(request), queryServiceConfig, metadataProviderConfig, federatedQuerySetup);
+    const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
+    ) {
+    return new TScriptFinalizerActor(std::move(request), queryServiceConfig, metadataProviderConfig, federatedQuerySetup, std::move(s3ActorsFactory));
 }
 
 }  // namespace NKikimr::NKqp

ydb/core/kqp/finalize_script_service/kqp_finalize_script_actor.h

@@ -2,13 +2,15 @@
 
 #include <ydb/core/kqp/common/events/script_executions.h>
 #include <ydb/core/kqp/federated_query/kqp_federated_query_helpers.h>
+#include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
 
 
 namespace NKikimr::NKqp {
 
 IActor* CreateScriptFinalizerActor(TEvScriptFinalizeRequest::TPtr request,
     const NKikimrConfig::TQueryServiceConfig& queryServiceConfig,
     const NKikimrConfig::TMetadataProviderConfig& metadataProviderConfig,
-    const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup);
+    const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory);
 
 }  // namespace NKikimr::NKqp