YQ-3644 added validations for resource pool parametres (#8831)

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -6264,6 +6264,20 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             );)").GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::GENERIC_ERROR);
         UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(), "Failed to parse property concurrent_query_limit:");
+
+        result = session.ExecuteSchemeQuery(TStringBuilder() << R"(
+            CREATE RESOURCE POOL MyResourcePool WITH (
+                CONCURRENT_QUERY_LIMIT=)" << NResourcePool::POOL_MAX_CONCURRENT_QUERY_LIMIT + 1 << R"(
+            );)").GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SCHEME_ERROR);
+        UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(), TStringBuilder() << "Invalid resource pool configuration, concurrent_query_limit is " << NResourcePool::POOL_MAX_CONCURRENT_QUERY_LIMIT + 1 << ", that exceeds limit in " << NResourcePool::POOL_MAX_CONCURRENT_QUERY_LIMIT);
+
+        result = session.ExecuteSchemeQuery(R"(
+            CREATE RESOURCE POOL MyResourcePool WITH (
+                QUEUE_SIZE=1
+            );)").GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SCHEME_ERROR);
+        UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(), "Invalid resource pool configuration, queue_size unsupported without concurrent_query_limit or database_load_cpu_threshold");
     }
 
     Y_UNIT_TEST(CreateResourcePool) {

ydb/core/kqp/workload_service/common/helpers.cpp

@@ -12,12 +12,7 @@ NYql::TIssues GroupIssues(const NYql::TIssues& issues, const TString& message) {
 }
 
 void ParsePoolSettings(const NKikimrSchemeOp::TResourcePoolDescription& description, NResourcePool::TPoolSettings& poolConfig) {
-    const auto& properties = description.GetProperties().GetProperties();
-    for (auto& [property, value] : poolConfig.GetPropertiesMap()) {
-        if (auto propertyIt = properties.find(property); propertyIt != properties.end()) {
-            std::visit(NResourcePool::TPoolSettings::TParser{propertyIt->second}, value);
-        }
-    }
+    poolConfig = NResourcePool::TPoolSettings(description.GetProperties().GetProperties());
 }
 
 ui64 SaturationSub(ui64 x, ui64 y) {

ydb/core/kqp/workload_service/ut/kqp_workload_service_actors_ut.cpp

@@ -132,7 +132,7 @@ Y_UNIT_TEST_SUITE(KqpWorkloadServiceActors) {
         // Check alter access
         TSampleQueries::CheckSuccess(ydb->ExecuteQuery(TStringBuilder() << R"(
             ALTER RESOURCE POOL )" << NResourcePool::DEFAULT_POOL_ID << R"( SET (
-                QUEUE_SIZE=1
+                QUERY_MEMORY_LIMIT_PERCENT_PER_NODE=1
             );
         )", settings));
 
@@ -205,7 +205,7 @@ Y_UNIT_TEST_SUITE(KqpWorkloadServiceSubscriptions) {
 
         ydb->ExecuteSchemeQuery(TStringBuilder() << R"(
             ALTER RESOURCE POOL )" << ydb->GetSettings().PoolId_ << R"( SET (
-                QUEUE_SIZE=42
+                CONCURRENT_QUERY_LIMIT=42
             );
         )");
 
@@ -214,7 +214,7 @@ Y_UNIT_TEST_SUITE(KqpWorkloadServiceSubscriptions) {
 
         const auto& config = response->Get()->Config;
         UNIT_ASSERT_C(config, "Pool config not found");
-        UNIT_ASSERT_VALUES_EQUAL(config->QueueSize, 42);
+        UNIT_ASSERT_VALUES_EQUAL(config->ConcurrentQueryLimit, 42);
     }
 
     Y_UNIT_TEST(TestResourcePoolSubscriptionAfterAclChange) {

ydb/core/kqp/workload_service/ut/kqp_workload_service_tables_ut.cpp

@@ -76,7 +76,7 @@ Y_UNIT_TEST_SUITE(KqpWorkloadServiceTables) {
     Y_UNIT_TEST(TestTablesIsNotCreatingForUnlimitedPool) {
         auto ydb = TYdbSetupSettings()
             .ConcurrentQueryLimit(-1)
-            .QueueSize(10)
+            .QueryMemoryLimitPercentPerNode(50)
             .Create();
 
         TSampleQueries::TSelect42::CheckResult(ydb->ExecuteQuery(TSampleQueries::TSelect42::Query));

ydb/core/resource_pools/resource_pool_settings.cpp

@@ -43,6 +43,14 @@ TString TPoolSettings::TExtractor::operator()(TDuration* setting) const {
 
 //// TPoolSettings
 
+TPoolSettings::TPoolSettings(const google::protobuf::Map<TString, TString>& properties) {
+    for (auto& [property, value] : GetPropertiesMap()) {
+        if (auto propertyIt = properties.find(property); propertyIt != properties.end()) {
+            std::visit(TPoolSettings::TParser{propertyIt->second}, value);
+        }
+    }
+}
+
 std::unordered_map<TString, TPoolSettings::TProperty> TPoolSettings::GetPropertiesMap(bool restricted) {
     std::unordered_map<TString, TProperty> properties = {
         {"concurrent_query_limit", &ConcurrentQueryLimit},
@@ -56,4 +64,13 @@ std::unordered_map<TString, TPoolSettings::TProperty> TPoolSettings::GetProperti
     return properties;
 }
 
+void TPoolSettings::Validate() const {
+    if (ConcurrentQueryLimit > POOL_MAX_CONCURRENT_QUERY_LIMIT) {
+        throw yexception() << "Invalid resource pool configuration, concurrent_query_limit is " << ConcurrentQueryLimit << ", that exceeds limit in " << POOL_MAX_CONCURRENT_QUERY_LIMIT;
+    }
+    if (QueueSize != -1 && ConcurrentQueryLimit == -1 && DatabaseLoadCpuThreshold < 0.0) {
+        throw yexception() << "Invalid resource pool configuration, queue_size unsupported without concurrent_query_limit or database_load_cpu_threshold";
+    }
+}
+
 }  // namespace NKikimr::NResourcePool

ydb/core/resource_pools/resource_pool_settings.h

@@ -2,13 +2,17 @@
 
 #include "settings_common.h"
 
+#include <contrib/libs/protobuf/src/google/protobuf/map.h>
+
 #include <util/datetime/base.h>
 
 
 namespace NKikimr::NResourcePool {
 
 inline constexpr char DEFAULT_POOL_ID[] = "default";
 
+inline constexpr i64 POOL_MAX_CONCURRENT_QUERY_LIMIT = 1000;
+
 struct TPoolSettings : public TSettingsBase {
     typedef double TPercent;
 
@@ -27,8 +31,13 @@ struct TPoolSettings : public TSettingsBase {
         TString operator()(TDuration* setting) const;
     };
 
+    TPoolSettings() = default;
+    TPoolSettings(const google::protobuf::Map<TString, TString>& properties);
+
     bool operator==(const TPoolSettings& other) const = default;
+
     std::unordered_map<TString, TProperty> GetPropertiesMap(bool restricted = false);
+    void Validate() const;
 
     i32 ConcurrentQueryLimit = -1;  // -1 = disabled
     i32 QueueSize = -1;  // -1 = disabled

ydb/core/resource_pools/resource_pool_settings_ut.cpp

@@ -73,6 +73,21 @@ Y_UNIT_TEST_SUITE(ResourcePoolTest) {
         UNIT_ASSERT_VALUES_EQUAL(std::visit(extractor, propertiesMap["query_cancel_after_seconds"]), "15");
         UNIT_ASSERT_VALUES_EQUAL(std::visit(extractor, propertiesMap["query_memory_limit_percent_per_node"]), "0.5");
     }
+
+    Y_UNIT_TEST(SettingsValidation) {
+        {  // Max concurrent query limit validation
+            TPoolSettings settings;
+            settings.ConcurrentQueryLimit = POOL_MAX_CONCURRENT_QUERY_LIMIT + 1;
+            UNIT_ASSERT_EXCEPTION_CONTAINS(settings.Validate(), yexception, TStringBuilder() << "Invalid resource pool configuration, concurrent_query_limit is " << settings.ConcurrentQueryLimit << ", that exceeds limit in " << POOL_MAX_CONCURRENT_QUERY_LIMIT);
+        }
+
+        {  // Unused queue size validation
+
+            TPoolSettings settings;
+            settings.QueueSize = 1;
+            UNIT_ASSERT_EXCEPTION_CONTAINS(settings.Validate(), yexception, TStringBuilder() << "Invalid resource pool configuration, queue_size unsupported without concurrent_query_limit or database_load_cpu_threshold");
+        }   
+    }
 }
 
 }  // namespace NKikimr

ydb/core/resource_pools/ya.make

@@ -6,6 +6,7 @@ SRCS(
 )
 
 PEERDIR(
+    contrib/libs/protobuf
     util
 )
 

ydb/core/tx/schemeshard/schemeshard__operation_alter_resource_pool.cpp

@@ -149,6 +149,7 @@ class TAlterResourcePool : public TSubOperation {
         Y_ABORT_UNLESS(oldResourcePoolInfo);
         const TResourcePoolInfo::TPtr resourcePoolInfo = NResourcePool::ModifyResourcePool(resourcePoolDescription, oldResourcePoolInfo);
         Y_ABORT_UNLESS(resourcePoolInfo);
+        RETURN_RESULT_UNLESS(NResourcePool::IsResourcePoolInfoValid(result, resourcePoolInfo));
 
         result->SetPathId(dstPath.Base()->PathId.LocalPathId);
         const TPathElement::TPtr resourcePool = ReplaceResourcePoolPathElement(dstPath);

ydb/core/tx/schemeshard/schemeshard__operation_common_resource_pool.cpp

@@ -1,6 +1,8 @@
 #include "schemeshard__operation_common_resource_pool.h"
 #include "schemeshard_impl.h"
 
+#include <ydb/core/resource_pools/resource_pool_settings.h>
+
 
 namespace NKikimr::NSchemeShard::NResourcePool {
 
@@ -90,6 +92,17 @@ bool IsDescriptionValid(const THolder<TProposeResponse>& result, const NKikimrSc
     return true;
 }
 
+bool IsResourcePoolInfoValid(const THolder<TProposeResponse>& result, const TResourcePoolInfo::TPtr& info) {
+    try {
+        NKikimr::NResourcePool::TPoolSettings settings(info->Properties.GetProperties());
+        settings.Validate();
+    } catch (...) {
+        result->SetError(NKikimrScheme::StatusSchemeError, CurrentExceptionMessage());
+        return false;
+    }
+    return true;
+}
+
 TTxState& CreateTransaction(const TOperationId& operationId, const TOperationContext& context, const TPathId& resourcePoolPathId, TTxState::ETxType txType) {
     Y_ABORT_UNLESS(!context.SS->FindTx(operationId));
     TTxState& txState = context.SS->CreateTx(operationId, txType, resourcePoolPathId);

ydb/core/tx/schemeshard/schemeshard__operation_common_resource_pool.h

@@ -24,6 +24,8 @@ bool IsApplyIfChecksPassed(const TTxTransaction& transaction, const THolder<TPro
 
 bool IsDescriptionValid(const THolder<TProposeResponse>& result, const NKikimrSchemeOp::TResourcePoolDescription& description);
 
+bool IsResourcePoolInfoValid(const THolder<TProposeResponse>& result, const TResourcePoolInfo::TPtr& info);
+
 TTxState& CreateTransaction(const TOperationId& operationId, const TOperationContext& context, const TPathId& resourcePoolPathId, TTxState::ETxType txType);
 
 void RegisterParentPathDependencies(const TOperationId& operationId, const TOperationContext& context, const TPath& parentPath);

ydb/core/tx/schemeshard/schemeshard__operation_create_resource_pool.cpp

@@ -173,6 +173,7 @@ class TCreateResourcePool : public TSubOperation {
 
         const TResourcePoolInfo::TPtr resourcePoolInfo = NResourcePool::CreateResourcePool(resourcePoolDescription, 1);
         Y_ABORT_UNLESS(resourcePoolInfo);
+        RETURN_RESULT_UNLESS(NResourcePool::IsResourcePoolInfoValid(result, resourcePoolInfo));
 
         AddPathInSchemeShard(result, dstPath, owner);
         const TPathElement::TPtr resourcePool = CreateResourcePoolPathElement(dstPath);

ydb/core/tx/schemeshard/ya.make

@@ -261,6 +261,7 @@ PEERDIR(
     ydb/core/persqueue/events
     ydb/core/persqueue/writer
     ydb/core/protos
+    ydb/core/resource_pools
     ydb/core/scheme
     ydb/core/statistics
     ydb/core/sys_view/partition_stats