merge to yq stable YQ-3561 support fq run

ydb/core/driver_lib/run/config_helpers.cpp

@@ -1,5 +1,9 @@
 #include "config_helpers.h"
 
+#include <ydb/core/base/localdb.h>
+#include <ydb/core/protos/bootstrap.pb.h>
+#include <ydb/core/protos/resource_broker.pb.h>
+
 #include <ydb/library/actors/util/affinity.h>
 
 
@@ -114,4 +118,31 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
 
 }  // namespace NActorSystemConfigHelpers
 
+namespace NKikimrConfigHelpers {
+
+NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config) {
+    NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
+    auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
+        if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
+            resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
+        }
+        for (const auto& queue : resourceBrokerConfig.GetQueues()) {
+            if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
+                if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
+                    resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
+                }
+            }
+        }
+    };
+    if (config.HasBootstrapConfig() && config.GetBootstrapConfig().HasResourceBroker()) {
+        mergeResourceBrokerConfigs(config.GetBootstrapConfig().GetResourceBroker());
+    }
+    if (config.HasResourceBrokerConfig()) {
+        mergeResourceBrokerConfigs(config.GetResourceBrokerConfig());
+    }
+    return resourceBrokerSelfConfig;
+}
+
+}  // namespace NKikimrConfigHelpers
+
 }  // namespace NKikimr

ydb/core/driver_lib/run/config_helpers.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <ydb/core/memory_controller/memory_controller.h>
 #include <ydb/core/protos/config.pb.h>
 
 #include <ydb/library/actors/core/config.h>
@@ -15,4 +16,10 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
 
 }  // namespace NActorSystemConfigHelpers
 
+namespace NKikimrConfigHelpers {
+
+NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config);
+
+}  // namespace NKikimrConfigHelpers
+
 }  // namespace NKikimr

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -2049,28 +2049,8 @@ void TMemoryControllerInitializer::InitializeServices(
     NActors::TActorSystemSetup* setup,
     const NKikimr::TAppData* appData)
 {
-    NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
-    auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
-        if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
-            resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
-        }
-        for (const auto& queue : resourceBrokerConfig.GetQueues()) {
-            if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
-                if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
-                    resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
-                }
-            }
-        }
-    };
-    if (Config.HasBootstrapConfig() && Config.GetBootstrapConfig().HasResourceBroker()) {
-        mergeResourceBrokerConfigs(Config.GetBootstrapConfig().GetResourceBroker());
-    }
-    if (Config.HasResourceBrokerConfig()) {
-        mergeResourceBrokerConfigs(Config.GetResourceBrokerConfig());
-    }
-
     auto* actor = NMemory::CreateMemoryController(TDuration::Seconds(1), ProcessMemoryInfoProvider,
-        Config.GetMemoryControllerConfig(), resourceBrokerSelfConfig,
+        Config.GetMemoryControllerConfig(), NKikimrConfigHelpers::CreateMemoryControllerResourceBrokerConfig(Config),
         appData->Counters);
     setup->LocalServices.emplace_back(
         NMemory::MakeMemoryControllerId(0),

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

@@ -88,6 +88,19 @@ std::pair<TString, bool> ParseHttpEndpoint(const TString& endpoint) {
     return std::make_pair(ToString(host), scheme != "http");
 }
 
+std::pair<TString, TIpPort> ParseGrpcEndpoint(const TString& endpoint) {
+    TStringBuf scheme;
+    TStringBuf address;
+    TStringBuf uri;
+    NHttp::CrackURL(endpoint, scheme, address, uri);
+
+    TString hostname;
+    TIpPort port;
+    NHttp::CrackAddress(TString(address), hostname, port);
+
+    return {hostname, port};
+}
+
 void FillSolomonClusterConfig(NYql::TSolomonClusterConfig& clusterConfig,
     const TString& name,
     const TString& authToken,
@@ -230,8 +243,18 @@ void AddClustersFromConnections(
             clusterCfg->SetKind(NYql::EGenericDataSourceKind::YDB);
             clusterCfg->SetProtocol(NYql::EGenericProtocol::NATIVE);
             clusterCfg->SetName(connectionName);
-            clusterCfg->SetDatabaseId(db.database_id());
-            clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
+            if (const auto& databaseId = db.database_id()) {
+                clusterCfg->SetDatabaseId(databaseId);
+                clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
+            } else {
+                const auto& [host, port] = ParseGrpcEndpoint(db.endpoint());
+
+                auto& endpoint = *clusterCfg->MutableEndpoint();
+                endpoint.set_host(host);
+                endpoint.set_port(port);
+                clusterCfg->SetUseSsl(db.secure());
+                clusterCfg->SetDatabaseName(db.database());
+            }
             FillClusterAuth(*clusterCfg, db.auth(), authToken, accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
             break;

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

@@ -157,7 +157,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         const ::NMonitoring::TDynamicCounterPtr& clientCounters,
         const TString& tenantName,
         NActors::TMon* monitoring,
-        std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
+        std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
+        NYql::IPqGatewayFactory::TPtr pqGatewayFactory
         )
         : YqSharedResources(yqSharedResources)
         , CredentialsProviderFactory(credentialsProviderFactory)
@@ -180,6 +181,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         , Monitoring(monitoring)
         , ComputeConfig(config.GetCompute())
         , S3ActorsFactory(std::move(s3ActorsFactory))
+        , PqGatewayFactory(std::move(pqGatewayFactory))
     {
         Y_ENSURE(GetYqlDefaultModuleResolverWithContext(ModuleResolver));
     }
@@ -472,7 +474,8 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
             NProtoInterop::CastFromProto(task.result_ttl()),
             std::map<TString, Ydb::TypedValue>(task.parameters().begin(), task.parameters().end()),
             S3ActorsFactory,
-            ComputeConfig.GetWorkloadManagerConfig(task.scope())
+            ComputeConfig.GetWorkloadManagerConfig(task.scope()),
+            PqGatewayFactory
             );
 
         auto runActorId =
@@ -548,6 +551,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
     NActors::TMon* Monitoring;
     TComputeConfig ComputeConfig;
     std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
+    NYql::IPqGatewayFactory::TPtr PqGatewayFactory;
 };
 
 
@@ -567,7 +571,8 @@ NActors::IActor* CreatePendingFetcher(
     const ::NMonitoring::TDynamicCounterPtr& clientCounters,
     const TString& tenantName,
     NActors::TMon* monitoring,
-    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory)
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
+    NYql::IPqGatewayFactory::TPtr pqGatewayFactory)
 {
     return new TPendingFetcher(
         yqSharedResources,
@@ -585,7 +590,8 @@ NActors::IActor* CreatePendingFetcher(
         clientCounters,
         tenantName,
         monitoring,
-        std::move(s3ActorsFactory));
+        std::move(s3ActorsFactory),
+        std::move(pqGatewayFactory));
 }
 
 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 <yql/essentials/providers/common/metrics/service_counters.h>
 #include <ydb/library/yql/providers/pq/cm_client/client.h>
+#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
 #include <ydb/library/yql/providers/s3/actors_factory/yql_s3_actors_factory.h>
 
 #include <ydb/public/lib/fq/scope.h>
@@ -53,7 +54,8 @@ NActors::IActor* CreatePendingFetcher(
     const ::NMonitoring::TDynamicCounterPtr& clientCounters,
     const TString& tenantName,
     NActors::TMon* monitoring,
-    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory
+    std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
+    NYql::IPqGatewayFactory::TPtr pqGatewayFactory
     );
 
 NActors::IActor* CreateRunActor(

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

@@ -571,6 +571,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
                         SelfId(),
                         Params.QueryId,
                         Params.YqSharedResources->UserSpaceYdbDriver,
+                        Params.PqGatewayFactory->CreatePqGateway(),
                         Params.Resources.topic_consumers(),
                         PrepareReadRuleCredentials()
                     )
@@ -1435,6 +1436,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
                 SelfId(),
                 Params.QueryId,
                 Params.YqSharedResources->UserSpaceYdbDriver,
+                Params.PqGatewayFactory->CreatePqGateway(),
                 Params.Resources.topic_consumers(),
                 PrepareReadRuleCredentials()
             )
@@ -1970,14 +1972,8 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
         }
 
         {
-            NYql::TPqGatewayServices pqServices(
-                Params.YqSharedResources->UserSpaceYdbDriver,
-                Params.PqCmConnections,
-                Params.CredentialsFactory,
-                std::make_shared<NYql::TPqGatewayConfig>(gatewaysConfig.GetPq()),
-                Params.FunctionRegistry
-            );
-            const auto pqGateway = NYql::CreatePqNativeGateway(pqServices);
+            auto pqGateway = Params.PqGatewayFactory->CreatePqGateway();
+            pqGateway->UpdateClusterConfigs(std::make_shared<NYql::TPqGatewayConfig>(gatewaysConfig.GetPq()));
             dataProvidersInit.push_back(GetPqDataProviderInitializer(pqGateway, false, dbResolver));
         }
 

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

@@ -60,7 +60,8 @@ TRunActorParams::TRunActorParams(
     TDuration resultTtl,
     std::map<TString, Ydb::TypedValue>&& queryParameters,
     std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
-    const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager
+    const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager,
+    NYql::IPqGatewayFactory::TPtr pqGatewayFactory
     )
     : YqSharedResources(yqSharedResources)
     , CredentialsProviderFactory(credentialsProviderFactory)
@@ -117,6 +118,7 @@ TRunActorParams::TRunActorParams(
     , QueryParameters(std::move(queryParameters))
     , S3ActorsFactory(std::move(s3ActorsFactory))
     , WorkloadManager(workloadManager)
+    , PqGatewayFactory(std::move(pqGatewayFactory))
     {
     }
 

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

@@ -13,6 +13,7 @@
 #include <ydb/library/yql/providers/dq/provider/yql_dq_gateway.h>
 #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/pq/provider/yql_pq_gateway.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>
 
@@ -79,7 +80,8 @@ struct TRunActorParams { // TODO2 : Change name
         TDuration resultTtl,
         std::map<TString, Ydb::TypedValue>&& queryParameters,
         std::shared_ptr<NYql::NDq::IS3ActorsFactory> s3ActorsFactory,
-        const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager
+        const ::NFq::NConfig::TWorkloadManagerConfig& workloadManager,
+        NYql::IPqGatewayFactory::TPtr pqGatewayFactory
     );
 
     TRunActorParams(const TRunActorParams& params) = default;
@@ -145,6 +147,7 @@ struct TRunActorParams { // TODO2 : Change name
     std::map<TString, Ydb::TypedValue> QueryParameters;
     std::shared_ptr<NYql::NDq::IS3ActorsFactory> S3ActorsFactory;
     ::NFq::NConfig::TWorkloadManagerConfig WorkloadManager;
+    NYql::IPqGatewayFactory::TPtr PqGatewayFactory;
 };
 
 } /* NFq */

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

@@ -32,4 +32,6 @@ message TCommonConfig {
     bool ShowQueryTimeline = 16;
     uint64 MaxQueryTimelineSize = 17; // default: 200KB
     string PqReconnectPeriod = 18; // default: disabled
+    uint32 TopicClientHandlersExecutorThreadsNum = 19; // default: 0 that means use default from TopicClientSettings (1)
+    uint32 TopicClientCompressionExecutorThreadsNum = 20; // default: 0 that means use default from TopicClientSettings (2)
 }

ydb/core/fq/libs/control_plane_storage/control_plane_storage.h

@@ -40,7 +40,13 @@ namespace NFq {
 
 NActors::TActorId ControlPlaneStorageServiceActorId(ui32 nodeId = 0);
 
-NActors::IActor* CreateInMemoryControlPlaneStorageServiceActor(const NConfig::TControlPlaneStorageConfig& config);
+NActors::IActor* CreateInMemoryControlPlaneStorageServiceActor(
+    const NConfig::TControlPlaneStorageConfig& config,
+    const NYql::TS3GatewayConfig& s3Config,
+    const NConfig::TCommonConfig& common,
+    const NConfig::TComputeConfig& computeConfig,
+    const ::NMonitoring::TDynamicCounterPtr& counters,
+    const TString& tenantName);
 
 NActors::IActor* CreateYdbControlPlaneStorageServiceActor(
     const NConfig::TControlPlaneStorageConfig& config,

ydb/core/fq/libs/control_plane_storage/events/events.h

@@ -392,6 +392,7 @@ struct TEvControlPlaneStorage {
 
     // internal messages
     struct TEvWriteResultDataRequest : NActors::TEventLocal<TEvWriteResultDataRequest, EvWriteResultDataRequest> {
+        using TProto = Fq::Private::WriteTaskResultRequest;
 
         TEvWriteResultDataRequest() = default;
 
@@ -411,6 +412,8 @@ struct TEvControlPlaneStorage {
     struct TEvWriteResultDataResponse : NActors::TEventLocal<TEvWriteResultDataResponse, EvWriteResultDataResponse> {
         static constexpr bool Auditable = false;
 
+        using TProto = Fq::Private::WriteTaskResultResult;
+
         explicit TEvWriteResultDataResponse(
             const Fq::Private::WriteTaskResultResult& record)
             : Record(record)
@@ -434,6 +437,7 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvGetTaskRequest : NActors::TEventLocal<TEvGetTaskRequest, EvGetTaskRequest> {
+        using TProto = Fq::Private::GetTaskRequest;
 
         TEvGetTaskRequest() = default;
 
@@ -454,6 +458,8 @@ struct TEvControlPlaneStorage {
     struct TEvGetTaskResponse : NActors::TEventLocal<TEvGetTaskResponse, EvGetTaskResponse> {
         static constexpr bool Auditable = false;
 
+        using TProto = Fq::Private::GetTaskResult;
+
         explicit TEvGetTaskResponse(
             const Fq::Private::GetTaskResult& record)
             : Record(record)
@@ -499,6 +505,7 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvPingTaskRequest : NActors::TEventLocal<TEvPingTaskRequest, EvPingTaskRequest> {
+        using TProto = Fq::Private::PingTaskRequest;
 
         TEvPingTaskRequest() = default;
 
@@ -519,6 +526,8 @@ struct TEvControlPlaneStorage {
     struct TEvPingTaskResponse : NActors::TEventLocal<TEvPingTaskResponse, EvPingTaskResponse> {
         static constexpr bool Auditable = false;
 
+        using TProto = Fq::Private::PingTaskResult;
+
         explicit TEvPingTaskResponse(
             const Fq::Private::PingTaskResult& record)
             : Record(record)
@@ -542,6 +551,7 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvNodesHealthCheckRequest : NActors::TEventLocal<TEvNodesHealthCheckRequest, EvNodesHealthCheckRequest> {
+        using TProto = Fq::Private::NodesHealthCheckRequest;
 
         TEvNodesHealthCheckRequest() = default;
 
@@ -561,6 +571,8 @@ struct TEvControlPlaneStorage {
     struct TEvNodesHealthCheckResponse : NActors::TEventLocal<TEvNodesHealthCheckResponse, EvNodesHealthCheckResponse> {
         static constexpr bool Auditable = false;
 
+        using TProto = Fq::Private::NodesHealthCheckResult;
+
         explicit TEvNodesHealthCheckResponse(
             const Fq::Private::NodesHealthCheckResult& record)
             : Record(record)