Merge 2083d4d into c692b6f

Author: kardymonds

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/init/init.cpp

@@ -56,6 +56,17 @@ namespace NFq {
 
 using namespace NKikimr;
 
+NYdb::NTopic::TTopicClientSettings GetCommonTopicClientSettings(const NFq::NConfig::TCommonConfig& config) {
+    NYdb::NTopic::TTopicClientSettings settings;
+    if (config.GetTopicClientHandlersExecutorThreadsNum()) {
+        settings.DefaultHandlersExecutor(NYdb::NTopic::CreateThreadPoolExecutor(config.GetTopicClientHandlersExecutorThreadsNum()));
+    }
+    if (config.GetTopicClientCompressionExecutorThreadsNum()) {
+        settings.DefaultCompressionExecutor(NYdb::NTopic::CreateThreadPoolExecutor(config.GetTopicClientCompressionExecutorThreadsNum()));
+    }
+    return settings;
+}
+
 void Init(
     const NFq::NConfig::TConfig& protoConfig,
     ui32 nodeId,
@@ -189,13 +200,18 @@ void Init(
         credentialsFactory = NYql::CreateSecuredServiceAccountCredentialsOverTokenAccessorFactory(tokenAccessorConfig.GetEndpoint(), tokenAccessorConfig.GetUseSsl(), caContent, tokenAccessorConfig.GetConnectionPoolSize());
     }
 
+    TMaybe<NYdb::NTopic::TTopicClientSettings> commonTopicClientSettings;
+
     if (protoConfig.GetRowDispatcher().GetEnabled()) {
+        commonTopicClientSettings = GetCommonTopicClientSettings(protoConfig.GetCommon());
         NYql::TPqGatewayServices pqServices(
             yqSharedResources->UserSpaceYdbDriver,
             nullptr,
             nullptr,
             std::make_shared<NYql::TPqGatewayConfig>(),
-            nullptr);
+            nullptr,
+            nullptr,
+            *commonTopicClientSettings);
 
         auto rowDispatcher = NFq::NewRowDispatcherService(
             protoConfig.GetRowDispatcher(),
@@ -223,7 +239,9 @@ void Init(
             pqCmConnections,
             credentialsFactory,
             std::make_shared<NYql::TPqGatewayConfig>(protoConfig.GetGateways().GetPq()),
-            appData->FunctionRegistry
+            appData->FunctionRegistry,
+            nullptr,
+            commonTopicClientSettings.GetOrElse(GetCommonTopicClientSettings(protoConfig.GetCommon()))
         );
         auto pqGateway = NYql::CreatePqNativeGateway(std::move(pqServices));
         RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory, pqGateway, 

ydb/core/fq/libs/row_dispatcher/topic_session.cpp

@@ -411,12 +411,11 @@ void TTopicSession::SubscribeOnNextEvent() {
 }
 
 NYdb::NTopic::TTopicClientSettings TTopicSession::GetTopicClientSettings(const NYql::NPq::NProto::TDqPqTopicSource& sourceParams) const {
-    NYdb::NTopic::TTopicClientSettings opts;
-    opts.Database(Database)
+    return PqGateway->GetTopicClientSettings()
+        .Database(Database)
         .DiscoveryEndpoint(Endpoint)
         .SslCredentials(NYdb::TSslCredentials(sourceParams.GetUseSsl()))
         .CredentialsProviderFactory(CredentialsProviderFactory);
-    return opts;
 }
 
 NYql::ITopicClient& TTopicSession::GetTopicClient(const NYql::NPq::NProto::TDqPqTopicSource& sourceParams) {

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor.cpp

@@ -161,7 +161,7 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public NYql::NDq:
     }
 
     NYdb::NTopic::TTopicClientSettings GetTopicClientSettings() const {
-        NYdb::NTopic::TTopicClientSettings opts;
+        NYdb::NTopic::TTopicClientSettings opts = PqGateway->GetTopicClientSettings();
         opts.Database(SourceParams.GetDatabase())
             .DiscoveryEndpoint(SourceParams.GetEndpoint())
             .SslCredentials(NYdb::TSslCredentials(SourceParams.GetUseSsl()))

ydb/library/yql/providers/pq/async_io/dq_pq_write_actor.cpp

@@ -312,7 +312,7 @@ class TDqPqWriteActor : public NActors::TActor<TDqPqWriteActor>, public IDqCompu
     }
 
     NYdb::NTopic::TTopicClientSettings GetTopicClientSettings() {
-        return NYdb::NTopic::TTopicClientSettings()
+        return PqGateway->GetTopicClientSettings()
             .Database(SinkParams.GetDatabase())
             .DiscoveryEndpoint(SinkParams.GetEndpoint())
             .SslCredentials(NYdb::TSslCredentials(SinkParams.GetUseSsl()))

ydb/library/yql/providers/pq/gateway/dummy/yql_pq_dummy_gateway.cpp

@@ -80,4 +80,8 @@ void TDummyPqGateway::UpdateClusterConfigs(
     Y_UNUSED(secure);
 }
 
+NYdb::NTopic::TTopicClientSettings TDummyPqGateway::GetTopicClientSettings() {
+    return NYdb::NTopic::TTopicClientSettings();
+}
+
 } // namespace NYql

ydb/library/yql/providers/pq/gateway/dummy/yql_pq_dummy_gateway.h

@@ -58,6 +58,7 @@ class TDummyPqGateway : public IPqGateway {
         bool secure) override;
 
     ITopicClient::TPtr GetTopicClient(const NYdb::TDriver& driver, const NYdb::NTopic::TTopicClientSettings& settings) override;
+    NYdb::NTopic::TTopicClientSettings GetTopicClientSettings() const override;
 
     using TClusterNPath = std::pair<TString, TString>;
 private:

ydb/library/yql/providers/pq/gateway/native/yql_pq_gateway.cpp

@@ -41,6 +41,7 @@ class TPqNativeGateway : public IPqGateway {
         bool secure) override;
 
     ITopicClient::TPtr GetTopicClient(const NYdb::TDriver& driver, const NYdb::NTopic::TTopicClientSettings& settings) override;
+    NYdb::NTopic::TTopicClientSettings GetTopicClientSettings() const override;
 
 private:
     void InitClusterConfigs();
@@ -56,6 +57,7 @@ class TPqNativeGateway : public IPqGateway {
     NYdb::TDriver YdbDriver;
     TPqClusterConfigsMapPtr ClusterConfigs;
     THashMap<TString, TPqSession::TPtr> Sessions;
+    TMaybe<NYdb::NTopic::TTopicClientSettings> CommonTopicClientSettings;
 };
 
 TPqNativeGateway::TPqNativeGateway(const TPqGatewayServices& services)
@@ -65,6 +67,7 @@ TPqNativeGateway::TPqNativeGateway(const TPqGatewayServices& services)
     , CredentialsFactory(services.CredentialsFactory)
     , CmConnections(services.CmConnections)
     , YdbDriver(services.YdbDriver)
+    , CommonTopicClientSettings(services.CommonTopicClientSettings)
 {
     Y_UNUSED(FunctionRegistry);
     InitClusterConfigs();
@@ -144,6 +147,10 @@ ITopicClient::TPtr TPqNativeGateway::GetTopicClient(const NYdb::TDriver& driver,
     return MakeIntrusive<TNativeTopicClient>(driver, settings);
 }
 
+NYdb::NTopic::TTopicClientSettings TPqNativeGateway::GetTopicClientSettings() {
+    return CommonTopicClientSettings.GetOrElse(NYdb::NTopic::TTopicClientSettings());
+}
+
 TPqNativeGateway::~TPqNativeGateway() {
     Sessions.clear();
 }

ydb/library/yql/providers/pq/gateway/native/yql_pq_gateway.h

@@ -26,20 +26,23 @@ struct TPqGatewayServices {
     ISecuredServiceAccountCredentialsFactory::TPtr CredentialsFactory;
     ::NPq::NConfigurationManager::IConnections::TPtr CmConnections;
     NYdb::TDriver YdbDriver;
+    TMaybe<NYdb::NTopic::TTopicClientSettings> CommonTopicClientSettings;
 
     TPqGatewayServices(
         NYdb::TDriver driver,
         ::NPq::NConfigurationManager::IConnections::TPtr cmConnections,
         ISecuredServiceAccountCredentialsFactory::TPtr credentialsFactory,
         TPqGatewayConfigPtr config,
         const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry,
-        IMetricsRegistryPtr metrics = nullptr)
+        IMetricsRegistryPtr metrics = nullptr,
+        TMaybe<NYdb::NTopic::TTopicClientSettings> commonTopicClientSettings = Nothing())
         : FunctionRegistry(functionRegistry)
         , Config(std::move(config))
         , Metrics(std::move(metrics))
         , CredentialsFactory(std::move(credentialsFactory))
         , CmConnections(std::move(cmConnections))
         , YdbDriver(std::move(driver))
+        , CommonTopicClientSettings(commonTopicClientSettings)
     {
     }
 };

ydb/library/yql/providers/pq/provider/yql_pq_gateway.h

@@ -36,6 +36,8 @@ struct IPqGateway : public TThrRefBase {
         const TString& endpoint,
         const TString& database,
         bool secure) = 0;
+
+    virtual NYdb::NTopic::TTopicClientSettings GetTopicClientSettings() const = 0;
 };
 
 } // namespace NYql