YQ supported compute databases in fqrun and in memory cp storage (#16821)

Author: GrigoriyPA

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

@@ -159,7 +159,10 @@ class TComputeConfig {
             case NConfig::TYdbComputeControlPlane::TYPE_NOT_SET:
                 return {};
             case NConfig::TYdbComputeControlPlane::kSingle:
-                return controlPlane.GetSingle().GetWorkloadManagerConfig();
+                if (controlPlane.GetSingle().HasWorkloadManagerConfig()) {
+                    return controlPlane.GetSingle().GetWorkloadManagerConfig();
+                }
+                return controlPlane.GetDefaultWorkloadManagerConfig();
             case NConfig::TYdbComputeControlPlane::kCms:
                 return GetWorkloadManagerConfig(scope, controlPlane.GetCms().GetDatabaseMapping());
             case NConfig::TYdbComputeControlPlane::kYdbcp:
@@ -169,7 +172,10 @@ class TComputeConfig {
 
     NFq::NConfig::TWorkloadManagerConfig GetWorkloadManagerConfig(const TString& scope, const ::NFq::NConfig::TDatabaseMapping& databaseMapping) const {
         auto computeDatabaseConfig = GetComputeDatabaseConfig(scope, databaseMapping);
-        return computeDatabaseConfig.GetWorkloadManagerConfig();
+        if (computeDatabaseConfig.HasWorkloadManagerConfig()) {
+            return computeDatabaseConfig.GetWorkloadManagerConfig();
+        }
+        return ComputeConfig.GetYdb().GetControlPlane().GetDefaultWorkloadManagerConfig();
     }
 
     NFq::NConfig::TComputeDatabaseConfig GetComputeDatabaseConfig(const TString& scope, const ::NFq::NConfig::TDatabaseMapping& databaseMapping) const {

ydb/core/fq/libs/compute/ydb/control_plane/compute_database_control_plane_service.cpp

@@ -89,7 +89,7 @@ class TCreateDatabaseRequestActor : public NActors::TActorBootstrapped<TCreateDa
                 LOG_T("Scope: " << Scope << " Single control plane mode has been chosen");
                 const auto& singleConfig = Config.GetYdb().GetControlPlane().GetSingle();
                 *Result.mutable_connection() = singleConfig.GetConnection();
-                Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, singleConfig.GetConnection(), singleConfig.GetWorkloadManagerConfig()});
+                Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, singleConfig.GetConnection(), GetWorkloadManagerConfig(singleConfig)});
             }
             break;
             case NConfig::TYdbComputeControlPlane::kCms:
@@ -163,7 +163,7 @@ class TCreateDatabaseRequestActor : public NActors::TActorBootstrapped<TCreateDa
         }
 
         if (response.IsExists) {
-            Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), client->Config.GetWorkloadManagerConfig()});
+            Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), GetWorkloadManagerConfig(client->Config)});
         } else {
             auto invalidateSynchronizationEvent = std::make_unique<TEvControlPlaneStorage::TEvModifyDatabaseRequest>(Request->Get()->CloudId, Scope);
             invalidateSynchronizationEvent->Synchronized = false;
@@ -199,7 +199,7 @@ class TCreateDatabaseRequestActor : public NActors::TActorBootstrapped<TCreateDa
         }
 
         if (ev->Cookie == OnlyDatabaseCreateCookie) {
-            Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), client->Config.GetWorkloadManagerConfig()});
+            Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), GetWorkloadManagerConfig(client->Config)});
             return;
         }
         Send(ControlPlaneStorageServiceActorId(), new TEvControlPlaneStorage::TEvCreateDatabaseRequest{Request->Get()->CloudId, Scope, Result});
@@ -266,7 +266,7 @@ class TCreateDatabaseRequestActor : public NActors::TActorBootstrapped<TCreateDa
             return;
         }
 
-        Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), client->Config.GetWorkloadManagerConfig()});
+        Send(SynchronizationServiceActorId, new TEvYdbCompute::TEvSynchronizeRequest{Request.Get()->Get()->CloudId, Request.Get()->Get()->Scope, Result.connection(), GetWorkloadManagerConfig(client->Config)});
     }
 
     void Handle(TEvYdbCompute::TEvSynchronizeResponse::TPtr& ev) {
@@ -298,11 +298,20 @@ class TCreateDatabaseRequestActor : public NActors::TActorBootstrapped<TCreateDa
         }
     }
 
+private:
+    template <typename TComputeConfig>
+    NConfig::TWorkloadManagerConfig GetWorkloadManagerConfig(const TComputeConfig& config) const {
+        if (config.HasWorkloadManagerConfig()) {
+            return config.GetWorkloadManagerConfig();
+        }
+        return Config.GetYdb().GetControlPlane().GetDefaultWorkloadManagerConfig();
+    }
+
 private:
     TString Scope;
     std::shared_ptr<TDatabaseClients> Clients;
     TActorId SynchronizationServiceActorId;
-    NFq::NConfig::TComputeConfig Config;
+    NConfig::TComputeConfig Config;
     TEvYdbCompute::TEvCreateDatabaseRequest::TPtr Request;
     FederatedQuery::Internal::ComputeDatabaseInternal Result;
 

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

@@ -86,6 +86,7 @@ message TYdbComputeControlPlane {
     }
     string DatabasePrefix = 5;
     string DatabasesCacheReloadPeriod = 6;
+    TWorkloadManagerConfig DefaultWorkloadManagerConfig = 7;
 }
 
 message TYdbCompute {

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

@@ -680,31 +680,36 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvCreateDatabaseRequest : NActors::TEventLocal<TEvCreateDatabaseRequest, EvCreateDatabaseRequest> {
+        using TProto = FederatedQuery::Internal::ComputeDatabaseInternal;
+
         TEvCreateDatabaseRequest() = default;
 
         explicit TEvCreateDatabaseRequest(const TString& cloudId, const TString& scope, const FederatedQuery::Internal::ComputeDatabaseInternal& record)
             : CloudId(cloudId)
             , Scope(scope)
-            , Record(record)
+            , Request(record)
         {}
 
         size_t GetByteSize() const {
             return sizeof(*this)
                     + CloudId.size()
                     + Scope.size()
-                    + Record.ByteSizeLong();
+                    + Request.ByteSizeLong();
         }
 
         TString CloudId;
         TString Scope;
-        FederatedQuery::Internal::ComputeDatabaseInternal Record;
+        FederatedQuery::Internal::ComputeDatabaseInternal Request;
     };
 
     struct TEvCreateDatabaseResponse : NActors::TEventLocal<TEvCreateDatabaseResponse, EvCreateDatabaseResponse> {
+        using TProto = google::protobuf::Empty;
+
         static constexpr bool Auditable = false;
 
-        explicit TEvCreateDatabaseResponse()
-        {}
+        explicit TEvCreateDatabaseResponse(const google::protobuf::Empty& response = {}) {
+            Y_UNUSED(response);
+        }
 
         explicit TEvCreateDatabaseResponse(const NYql::TIssues& issues)
             : Issues(issues)
@@ -721,6 +726,7 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvDescribeDatabaseRequest : NActors::TEventLocal<TEvDescribeDatabaseRequest, EvDescribeDatabaseRequest> {
+        using TProto = google::protobuf::Empty;
 
         TEvDescribeDatabaseRequest() = default;
 
@@ -744,6 +750,8 @@ struct TEvControlPlaneStorage {
     struct TEvDescribeDatabaseResponse : NActors::TEventLocal<TEvDescribeDatabaseResponse, EvDescribeDatabaseResponse> {
         static constexpr bool Auditable = false;
 
+        using TProto = FederatedQuery::Internal::ComputeDatabaseInternal;
+
         explicit TEvDescribeDatabaseResponse(const FederatedQuery::Internal::ComputeDatabaseInternal& record)
             : Record(record)
         {}
@@ -767,6 +775,8 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvModifyDatabaseRequest : NActors::TEventLocal<TEvModifyDatabaseRequest, EvModifyDatabaseRequest> {
+        using TProto = google::protobuf::Empty;
+
         TEvModifyDatabaseRequest() = default;
 
         explicit TEvModifyDatabaseRequest(const TString& cloudId, const TString& scope)
@@ -776,10 +786,12 @@ struct TEvControlPlaneStorage {
 
         size_t GetByteSize() const {
             return sizeof(*this)
+                    + Request.ByteSizeLong()
                     + CloudId.size()
                     + Scope.size();
         }
 
+        google::protobuf::Empty Request;
         TString CloudId;
         TString Scope;
         TMaybe<bool> Synchronized;
@@ -788,10 +800,13 @@ struct TEvControlPlaneStorage {
     };
 
     struct TEvModifyDatabaseResponse : NActors::TEventLocal<TEvModifyDatabaseResponse, EvModifyDatabaseResponse> {
+        using TProto = google::protobuf::Empty;
+
         static constexpr bool Auditable = false;
 
-        explicit TEvModifyDatabaseResponse()
-        {}
+        explicit TEvModifyDatabaseResponse(const google::protobuf::Empty& response = {}) {
+            Y_UNUSED(response);
+        }
 
         explicit TEvModifyDatabaseResponse(const NYql::TIssues& issues)
             : Issues(issues)