YQ-3561 FQrun supported connections / bindings

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::IPqGateway::TPtr defaultPqGateway
         )
         : YqSharedResources(yqSharedResources)
         , CredentialsProviderFactory(credentialsProviderFactory)
@@ -180,6 +181,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
         , Monitoring(monitoring)
         , ComputeConfig(config.GetCompute())
         , S3ActorsFactory(std::move(s3ActorsFactory))
+        , DefaultPqGateway(std::move(defaultPqGateway))
     {
         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()),
+            DefaultPqGateway
             );
 
         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::IPqGateway::TPtr DefaultPqGateway;
 };
 
 
@@ -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::IPqGateway::TPtr defaultPqGateway)
 {
     return new TPendingFetcher(
         yqSharedResources,
@@ -585,7 +590,8 @@ NActors::IActor* CreatePendingFetcher(
         clientCounters,
         tenantName,
         monitoring,
-        std::move(s3ActorsFactory));
+        std::move(s3ActorsFactory),
+        defaultPqGateway);
 }
 
 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::IPqGateway::TPtr defaultPqGateway
     );
 
 NActors::IActor* CreateRunActor(

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

@@ -1977,7 +1977,7 @@ class TRunActor : public NActors::TActorBootstrapped<TRunActor> {
                 std::make_shared<NYql::TPqGatewayConfig>(gatewaysConfig.GetPq()),
                 Params.FunctionRegistry
             );
-            const auto pqGateway = NYql::CreatePqNativeGateway(pqServices);
+            const auto pqGateway = Params.DefaultPqGateway ? Params.DefaultPqGateway : NYql::CreatePqNativeGateway(pqServices);
             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::IPqGateway::TPtr defaultPqGateway
     )
     : YqSharedResources(yqSharedResources)
     , CredentialsProviderFactory(credentialsProviderFactory)
@@ -117,6 +118,7 @@ TRunActorParams::TRunActorParams(
     , QueryParameters(std::move(queryParameters))
     , S3ActorsFactory(std::move(s3ActorsFactory))
     , WorkloadManager(workloadManager)
+    , DefaultPqGateway(defaultPqGateway)
     {
     }
 

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::IPqGateway::TPtr defaultPqGateway
     );
 
     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::IPqGateway::TPtr DefaultPqGateway;
 };
 
 } /* NFq */

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

@@ -66,7 +66,8 @@ void Init(
     const IYqSharedResources::TPtr& iyqSharedResources,
     const std::function<IActor*(const NKikimrProto::NFolderService::TFolderServiceConfig& authConfig)>& folderServiceFactory,
     ui32 icPort,
-    const std::vector<NKikimr::NMiniKQL::TComputationNodeFactory>& additionalCompNodeFactories
+    const std::vector<NKikimr::NMiniKQL::TComputationNodeFactory>& additionalCompNodeFactories,
+    NYql::IPqGateway::TPtr defaultPqGateway
     )
 {
     Y_ABORT_UNLESS(iyqSharedResources, "No YQ shared resources created");
@@ -204,7 +205,7 @@ void Init(
             credentialsFactory,
             tenant,
             yqCounters->GetSubgroup("subsystem", "row_dispatcher"),
-            CreatePqNativeGateway(pqServices),
+            defaultPqGateway ? defaultPqGateway : CreatePqNativeGateway(pqServices),
             appData->Mon,
             appData->Counters);
         actorRegistrator(NFq::RowDispatcherServiceActorId(), rowDispatcher.release());
@@ -225,7 +226,7 @@ void Init(
             std::make_shared<NYql::TPqGatewayConfig>(protoConfig.GetGateways().GetPq()),
             appData->FunctionRegistry
         );
-        auto pqGateway = NYql::CreatePqNativeGateway(std::move(pqServices));
+        auto pqGateway = defaultPqGateway ? defaultPqGateway : NYql::CreatePqNativeGateway(std::move(pqServices));
         RegisterDqPqReadActorFactory(*asyncIoFactory, yqSharedResources->UserSpaceYdbDriver, credentialsFactory, pqGateway, 
             yqCounters->GetSubgroup("subsystem", "DqSourceTracker"), protoConfig.GetCommon().GetPqReconnectPeriod());
 
@@ -345,7 +346,8 @@ void Init(
             clientCounters,
             tenant,
             appData->Mon,
-            s3ActorsFactory
+            s3ActorsFactory,
+            defaultPqGateway
             );
 
         actorRegistrator(MakePendingFetcherId(nodeId), fetcher);

ydb/core/fq/libs/init/init.h

@@ -12,6 +12,7 @@
 #include <ydb/core/fq/libs/config/protos/audit.pb.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/actors/core/actor.h>
 
@@ -36,7 +37,8 @@ void Init(
     const IYqSharedResources::TPtr& yqSharedResources,
     const std::function<IActor*(const NKikimrProto::NFolderService::TFolderServiceConfig& authConfig)>& folderServiceFactory,
     ui32 icPort,
-    const std::vector<NKikimr::NMiniKQL::TComputationNodeFactory>& additionalCompNodeFactories
+    const std::vector<NKikimr::NMiniKQL::TComputationNodeFactory>& additionalCompNodeFactories,
+    NYql::IPqGateway::TPtr defaultPqGateway = nullptr
 );
 
 } // NFq

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

@@ -11,7 +11,7 @@ namespace NYql {
 NThreading::TFuture<void> TDummyPqGateway::OpenSession(const TString& sessionId, const TString& username) {
     with_lock (Mutex) {
         Y_ENSURE(sessionId);
-        Y_ENSURE(username);
+        Y_UNUSED(username);
 
         Y_ENSURE(!IsIn(OpenedSessions, sessionId), "Session " << sessionId << " is already opened in pq gateway");
         OpenedSessions.insert(sessionId);

ydb/tests/tools/fqrun/.gitignore

@@ -2,3 +2,6 @@ sync_dir
 
 *.log
 *.sql
+*.conf
+*.parquet
+*.json

ydb/tests/tools/fqrun/README.md

@@ -0,0 +1,35 @@
+# FQ run tool
+
+Tool can be used to execute streaming queries by using FQ proxy infrastructure.
+
+## Examples
+
+### Queries
+
+* Run select 42:
+    ```(bash)
+    ./fqrun -s "SELECT 42"
+    ```
+
+### Logs
+
+* Setup log settings:
+    ```(bash)
+    ./fqrun -s "SELECT 42" --log-default=warn --log FQ_RUN_ACTOR=trace --log-file query.log
+    ```
+
+### Cluster
+
+* Embedded UI:
+    ```(bash)
+    ./fqrun -M 32000
+    ```
+
+    Monitoring endpoint: https://localhost:32000
+
+* gRPC endpoint:
+    ```(bash)
+    ./fqrun -G 32000
+    ```
+
+    Connect with ydb CLI: `ydb -e grpc://localhost:32000 -d /Root`