Auto partitioning to PQ_V1 sdk

Author: niksaveliev

ydb/public/sdk/cpp/client/ydb_persqueue_public/impl/persqueue.cpp

@@ -86,11 +86,20 @@ TDescribeTopicResult::TDescribeTopicResult(TStatus status, const Ydb::PersQueue:
 }
 
 TDescribeTopicResult::TTopicSettings::TTopicSettings(const Ydb::PersQueue::V1::TopicSettings& settings) {
-
-    PartitionsCount_ = settings.partitions_count();
     RetentionPeriod_ = TDuration::MilliSeconds(settings.retention_period_ms());
     SupportedFormat_ = static_cast<EFormat>(settings.supported_format());
 
+    if (settings.has_auto_partitioning_settings()) {
+        PartitionsCount_ = settings.auto_partitioning_settings().min_active_partitions();
+        MaxPartitionsCount_ = settings.auto_partitioning_settings().max_active_partitions();
+        StabilizationWindow_ = TDuration::Seconds(settings.auto_partitioning_settings().partition_write_speed().stabilization_window().seconds());
+        UpUtilizationPercent_ = settings.auto_partitioning_settings().partition_write_speed().up_utilization_percent();
+        DownUtilizationPercent_ = settings.auto_partitioning_settings().partition_write_speed().down_utilization_percent();
+        AutoPartitioningStrategy_ = settings.auto_partitioning_settings().strategy();
+    } else {
+        PartitionsCount_ = settings.partitions_count();
+    }
+
     for (const auto& codec : settings.supported_codecs()) {
         SupportedCodecs_.push_back(static_cast<ECodec>(codec));
     }

ydb/public/sdk/cpp/client/ydb_persqueue_public/impl/persqueue_impl.h

@@ -54,6 +54,34 @@ class TPersQueueClient::TImpl : public TClientImplCommon<TPersQueueClient::TImpl
 
         props.set_partitions_count(settings.PartitionsCount_);
 
+        bool autoscalingSettingsDefined = false;
+        if (settings.MaxPartitionsCount_.Defined()) {
+            props.mutable_auto_partitioning_settings()->set_max_active_partitions(settings.PartitionsCount_);
+            autoscalingSettingsDefined = true;
+        }
+        if (settings.AutoPartitioningStrategy_.Defined()) {
+            props.mutable_auto_partitioning_settings()->set_strategy(*settings.AutoPartitioningStrategy_);
+            autoscalingSettingsDefined = true;
+        }
+        if (settings.DownUtilizationPercent_.Defined()) {
+            props.mutable_auto_partitioning_settings()->mutable_partition_write_speed()->set_down_utilization_percent(*settings.DownUtilizationPercent_);
+            autoscalingSettingsDefined = true;
+        }
+        if (settings.UpUtilizationPercent_.Defined()) {
+            props.mutable_auto_partitioning_settings()->mutable_partition_write_speed()->set_up_utilization_percent(*settings.UpUtilizationPercent_);
+            autoscalingSettingsDefined = true;
+        }
+        if (settings.StabilizationWindow_.Defined()) {
+            props.mutable_auto_partitioning_settings()->mutable_partition_write_speed()->mutable_stabilization_window()->set_seconds((*settings.StabilizationWindow_).Seconds());
+            autoscalingSettingsDefined = true;
+        }
+        if (!autoscalingSettingsDefined) {
+            props.set_partitions_count(settings.PartitionsCount_);
+        } else {
+            props.mutable_auto_partitioning_settings()->set_min_active_partitions(settings.PartitionsCount_);
+        }
+
+
         props.set_retention_period_ms(settings.RetentionPeriod_.MilliSeconds());
         props.set_supported_format(static_cast<Ydb::PersQueue::V1::TopicSettings::Format>(settings.SupportedFormat_));
         for (const auto& codec : settings.SupportedCodecs_) {

ydb/public/sdk/cpp/client/ydb_persqueue_public/include/control_plane.h

@@ -118,6 +118,12 @@ struct TDescribeTopicResult : public TStatus {
         }
         GETTER(TMaybe<TRemoteMirrorRule>, RemoteMirrorRule);
 
+        GETTER(TMaybe<ui64>, MaxPartitionsCount);
+        GETTER(TMaybe<TDuration>, StabilizationWindow);
+        GETTER(TMaybe<ui64>, UpUtilizationPercent);
+        GETTER(TMaybe<ui64>, DownUtilizationPercent);
+        GETTER(TMaybe<Ydb::PersQueue::V1::AutoPartitioningStrategy>, AutoPartitioningStrategy);
+
 
 #undef GETTER
 
@@ -139,6 +145,12 @@ struct TDescribeTopicResult : public TStatus {
         TMaybe<ui32> AbcId_;
         TMaybe<TString> AbcSlug_;
         TString FederationAccount_;
+
+        TMaybe<ui64> MaxPartitionsCount_;
+        TMaybe<TDuration> StabilizationWindow_;
+        TMaybe<ui64> UpUtilizationPercent_;
+        TMaybe<ui64> DownUtilizationPercent_;
+        TMaybe<Ydb::PersQueue::V1::AutoPartitioningStrategy> AutoPartitioningStrategy_;
     };
 
     TDescribeTopicResult(TStatus status, const Ydb::PersQueue::V1::DescribeTopicResult& result);
@@ -192,6 +204,7 @@ struct TReadRuleSettings {
 // Settings for topic.
 template <class TDerived>
 struct TTopicSettings : public TOperationRequestSettings<TDerived> {
+    friend class TPersQueueClient;
 
     struct TRemoteMirrorRuleSettings {
         TRemoteMirrorRuleSettings() {}
@@ -267,9 +280,23 @@ struct TTopicSettings : public TOperationRequestSettings<TDerived> {
         if (settings.RemoteMirrorRule()) {
             RemoteMirrorRule_ = TRemoteMirrorRuleSettings().SetSettings(settings.RemoteMirrorRule().GetRef());
         }
+
+        MaxPartitionsCount_ = settings.MaxPartitionsCount();
+        StabilizationWindow_ = settings.StabilizationWindow();
+        UpUtilizationPercent_ = settings.UpUtilizationPercent();
+        DownUtilizationPercent_ = settings.DownUtilizationPercent();
+        AutoPartitioningStrategy_ = settings.AutoPartitioningStrategy();
+
         return static_cast<TDerived&>(*this);
     }
 
+    private:
+        TMaybe<ui64> MaxPartitionsCount_;
+        TMaybe<TDuration> StabilizationWindow_;
+        TMaybe<ui64> UpUtilizationPercent_;
+        TMaybe<ui64> DownUtilizationPercent_;
+        TMaybe<Ydb::PersQueue::V1::AutoPartitioningStrategy> AutoPartitioningStrategy_;
+
 };