Import YDB C++ SDK 12

.github/import_generation.txt

@@ -1 +1 @@
-12
+13

.github/last_commit.txt

@@ -1 +1 @@
-ec36c2f7151e1e7718c2f0c0b2292383cab01e64
+6e380d54416c706390bbe2609c7a054de2bf3b89

.github/scripts/copy_sources.sh

@@ -31,7 +31,7 @@ cp -r $1/ydb/public/api/grpc $tmp_dir/src/api
 cp -r $1/ydb/public/api/protos $tmp_dir/src/api
 
 rm -r $tmp_dir/src/api/protos/out
-rm $tmp_dir/include/ydb-cpp-sdk/type_switcher.h $tmp_dir/src/version.h
+rm $tmp_dir/include/ydb-cpp-sdk/type_switcher.h $tmp_dir/include/ydb-cpp-sdk/client/proto/private.h $tmp_dir/src/version.h
 
 cp -r $2/util $tmp_dir
 cp -r $2/library $tmp_dir
@@ -53,6 +53,7 @@ cp $2/LICENSE $tmp_dir
 cp $2/README.md $tmp_dir
 
 cp $2/include/ydb-cpp-sdk/type_switcher.h $tmp_dir/include/ydb-cpp-sdk/type_switcher.h
+cp $2/include/ydb-cpp-sdk/client/proto/private.h $tmp_dir/include/ydb-cpp-sdk/client/proto/private.h
 cp $2/src/version.h $tmp_dir/src/version.h
 
 cd $2

include/ydb-cpp-sdk/client/discovery/discovery.h

@@ -48,6 +48,7 @@ struct TNodeRegistrationSettings : public TSimpleRequestSettings<TNodeRegistrati
     FLUENT_SETTING(std::string, DomainPath);
     FLUENT_SETTING_DEFAULT(bool, FixedNodeId, false);
     FLUENT_SETTING(std::string, Path);
+    FLUENT_SETTING(std::string, BridgePileName);
 };
 
 struct TEndpointInfo {
@@ -96,6 +97,7 @@ struct TNodeInfo {
     std::string Address;
     TNodeLocation Location;
     uint64_t Expire;
+    std::optional<uint32_t> BridgePileId;
 };
 
 class TNodeRegistrationResult : public TStatus {

include/ydb-cpp-sdk/client/proto/accessor.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "private.h"
+
 #include <src/api/protos/ydb_coordination.pb.h>
 #include <src/api/protos/ydb_export.pb.h>
 #include <src/api/protos/ydb_import.pb.h>
@@ -45,6 +47,10 @@ class TProtoAccessor {
     static const Ydb::Topic::DescribeConsumerResult& GetProto(const NYdb::NTopic::TConsumerDescription& consumerDescription);
     static const Ydb::Monitoring::SelfCheckResult& GetProto(const NYdb::NMonitoring::TSelfCheckResult& selfCheckResult);
     static const Ydb::Coordination::DescribeNodeResult& GetProto(const NYdb::NCoordination::TNodeDescription &describeNodeResult);
+#ifdef YDB_SDK_INTERNAL_CLIENTS
+    static const Ydb::Replication::DescribeReplicationResult& GetProto(const NYdb::NReplication::TDescribeReplicationResult& desc);
+    static const Ydb::View::DescribeViewResult& GetProto(const NYdb::NView::TDescribeViewResult& desc);
+#endif
 
     static NTable::TQueryStats FromProto(const Ydb::TableStats::QueryStats& queryStats);
     static NTable::TTableDescription FromProto(const Ydb::Table::CreateTableRequest& request);

include/ydb-cpp-sdk/client/query/query.h

@@ -124,15 +124,21 @@ class TResultSetMeta {
 public:
     TResultSetMeta() = default;
 
-    explicit TResultSetMeta(const std::vector<TColumn>& columns)
+    explicit TResultSetMeta(const std::vector<TColumn>& columns, uint64_t rowsCount = 0, bool finished = false)
         : Columns(columns)
+        , RowsCount(rowsCount)
+        , Finished(finished)
     {}
 
-    explicit TResultSetMeta(std::vector<TColumn>&& columns)
+    explicit TResultSetMeta(std::vector<TColumn>&& columns, uint64_t rowsCount = 0, bool finished = false)
         : Columns(std::move(columns))
+        , RowsCount(rowsCount)
+        , Finished(finished)
     {}
 
     std::vector<TColumn> Columns;
+    uint64_t RowsCount = 0;
+    bool Finished = false;
 };
 
 class TScriptExecutionOperation : public TOperation {

include/ydb-cpp-sdk/client/table/table.h

@@ -397,6 +397,8 @@ class TChangefeedDescription {
 
     // Enable virtual timestamps
     TChangefeedDescription& WithVirtualTimestamps();
+    // Enable schema changes
+    TChangefeedDescription& WithSchemaChanges();
     // Enable resolved timestamps
     TChangefeedDescription& WithResolvedTimestamps(const TDuration& interval);
     // Customise retention period of underlying topic (24h by default).
@@ -415,6 +417,7 @@ class TChangefeedDescription {
     EChangefeedFormat GetFormat() const;
     EChangefeedState GetState() const;
     bool GetVirtualTimestamps() const;
+    bool GetSchemaChanges() const;
     const std::optional<TDuration>& GetResolvedTimestamps() const;
     bool GetInitialScan() const;
     const std::unordered_map<std::string, std::string>& GetAttributes() const;
@@ -442,6 +445,7 @@ class TChangefeedDescription {
     EChangefeedFormat Format_;
     EChangefeedState State_ = EChangefeedState::Unknown;
     bool VirtualTimestamps_ = false;
+    bool SchemaChanges_ = false;
     std::optional<TDuration> ResolvedTimestamps_;
     std::optional<TDuration> RetentionPeriod_;
     bool InitialScan_ = false;

include/ydb-cpp-sdk/client/topic/control_plane.h

@@ -128,6 +128,7 @@ class TPartitionConsumerStats {
 class TPartitionLocation {
 public:
     TPartitionLocation(const Ydb::Topic::PartitionLocation& partitionLocation);
+    TPartitionLocation(std::int32_t nodeId, std::int64_t generation);
     int32_t GetNodeId() const;
     int64_t GetGeneration() const;
 

include/ydb-cpp-sdk/client/topic/read_events.h

@@ -1,10 +1,12 @@
 #pragma once
 
 #include "codecs.h"
+#include "control_plane.h"
 #include "events_common.h"
 
 #include <util/datetime/base.h>
 
+
 namespace NYdb::inline V3::NTopic {
 
 //! Partition session.
@@ -42,10 +44,14 @@ struct TPartitionSession: public TThrRefBase, public TPrintable<TPartitionSessio
     }
 
 protected:
+
     uint64_t PartitionSessionId;
     std::string TopicPath;
     std::string ReadSessionId;
     uint64_t PartitionId;
+    std::optional<TPartitionLocation> Location;
+    /*TDirectReadId*/ std::int64_t NextDirectReadId = 1;
+    std::optional</*TDirectReadId*/ std::int64_t> LastDirectReadId;
 };
 
 template<>

include/ydb-cpp-sdk/client/topic/read_session.h

@@ -193,6 +193,9 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
     //! AutoPartitioningSupport.
     FLUENT_SETTING_DEFAULT(bool, AutoPartitioningSupport, false);
 
+    // TODO(qyryq) Uncomment when direct read is ready.
+    // FLUENT_SETTING_DEFAULT(bool, DirectRead, false);
+
     //! Log.
     FLUENT_SETTING_OPTIONAL(TLog, Log);
 };

include/ydb-cpp-sdk/library/retry/retry_policy.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <util/datetime/base.h>
+#include <util/generic/maybe.h>
 #include <util/generic/typetraits.h>
 #include <util/random/random.h>
 
@@ -41,7 +42,7 @@ struct IRetryPolicy {
 
         //! Calculate delay before next retry if next retry is allowed.
         //! Returns empty maybe if retry is not allowed anymore.
-        [[nodiscard]] virtual std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) = 0;
+        [[nodiscard]] virtual TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) = 0;
     };
 
     virtual ~IRetryPolicy() = default;
@@ -81,8 +82,8 @@ struct TNoRetryPolicy : IRetryPolicy<TArgs...> {
     using IRetryState = typename IRetryPolicy<TArgs...>::IRetryState;
 
     struct TNoRetryState : IRetryState {
-        std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam...) override {
-            return std::nullopt;
+        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam...) override {
+            return {};
         }
     };
 
@@ -123,10 +124,10 @@ struct TExponentialBackoffPolicy : IRetryPolicy<TArgs...> {
         {
         }
 
-        std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
+        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
             const ERetryErrorClass errorClass = RetryClassFunction(args...);
             if (errorClass == ERetryErrorClass::NoRetry || AttemptsDone >= MaxRetries || StartTime && TInstant::Now() - StartTime >= MaxTime) {
-                return std::nullopt;
+                return {};
             }
 
             if (errorClass == ERetryErrorClass::LongRetry) {
@@ -212,10 +213,10 @@ struct TFixedIntervalPolicy : IRetryPolicy<TArgs...> {
         {
         }
 
-        std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
+        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
             const ERetryErrorClass errorClass = RetryClassFunction(args...);
             if (errorClass == ERetryErrorClass::NoRetry || AttemptsDone >= MaxRetries || StartTime && TInstant::Now() - StartTime >= MaxTime) {
-                return std::nullopt;
+                return {};
             }
 
             const TDuration delay = NRetryDetails::RandomizeDelay(errorClass == ERetryErrorClass::LongRetry ? LongRetryDelay : Delay);

src/api/grpc/draft/ydb_dynamic_config_v1.proto

@@ -49,4 +49,7 @@ service DynamicConfigService {
 
     // Generate dynamic config based on cluster's static config.
     rpc FetchStartupConfig(DynamicConfig.FetchStartupConfigRequest) returns (DynamicConfig.FetchStartupConfigResponse);
+
+    // Get configuration version of nodes.
+    rpc GetConfigurationVersion(DynamicConfig.GetConfigurationVersionRequest) returns (DynamicConfig.GetConfigurationVersionResponse);
 }

src/api/protos/draft/ydb_dynamic_config.proto

@@ -242,3 +242,42 @@ message FetchStartupConfigResult {
     // YAML document with generated dynamic config
     string config = 1;
 }
+
+message GetConfigurationVersionRequest {
+    Ydb.Operations.OperationParams operation_params = 1;
+    // List nodes with different configuration versions
+    bool list_nodes = 2;
+}
+
+message GetConfigurationVersionResponse {
+    // Result of request will be inside operation.
+    Ydb.Operations.Operation operation = 1;
+}
+
+message NodeInfo {
+    message Endpoint {
+        // Node's hostname
+        string hostname = 1;
+        // Node's port
+        uint32 port = 2;
+    }
+    // Node's id
+    uint32 node_id = 1;
+    // Node's endpoint
+    Endpoint endpoint = 2;
+}
+
+message GetConfigurationVersionResult {
+    // Count of nodes with V1 configuration version
+    uint32 v1_nodes = 1;
+    // List of nodes with V1 configuration version
+    repeated NodeInfo v1_nodes_list = 2;
+    // Count of nodes with V2 configuration version
+    uint32 v2_nodes = 3;
+    // List of nodes with V2 configuration version
+    repeated NodeInfo v2_nodes_list = 4;
+    // Count of nodes with unknown configuration version
+    uint32 unknown_nodes = 5;
+    // List of nodes with unknown configuration version
+    repeated NodeInfo unknown_nodes_list = 6;
+}

src/api/protos/ydb_discovery.proto

@@ -83,6 +83,7 @@ message NodeInfo {
     optional string address = 5;
     optional NodeLocation location = 6;
     optional uint64 expire = 7;
+    optional uint32 bridge_pile_id = 8;
 }
 
 message NodeRegistrationRequest {
@@ -94,6 +95,7 @@ message NodeRegistrationRequest {
     optional string domain_path = 6;
     optional bool fixed_node_id = 7;
     optional string path = 8;
+    optional string bridge_pile_name = 9;
 }
 
 message NodeRegistrationResult {

src/api/protos/ydb_query.proto

@@ -186,6 +186,8 @@ message ExecuteQueryRequest {
 
 message ResultSetMeta {
     repeated Ydb.Column columns = 1;
+    uint64 number_rows = 2;
+    bool finished = 3;
 }
 
 message ExecuteQueryResponsePart {

src/api/protos/ydb_scheme.proto

@@ -67,6 +67,7 @@ message Entry {
         VIEW = 20;
         RESOURCE_POOL = 21;
         TRANSFER = 23;
+        SYS_VIEW = 24;
     }
 
     // Name of scheme entry (dir2 of /dir1/dir2)

src/api/protos/ydb_table.proto

@@ -233,6 +233,8 @@ message Changefeed {
     google.protobuf.Duration resolved_timestamps_interval = 9;
     // Partitioning settings of underlying topic.
     Topic.PartitioningSettings topic_partitioning_settings = 10;
+    // Emit schema change events or not
+    bool schema_changes = 11;
 }
 
 message ChangefeedDescription {
@@ -274,6 +276,8 @@ message ChangefeedDescription {
     google.protobuf.Duration resolved_timestamps_interval = 8;
     // Progress of initial scan. If unspecified, initial scan was not launched.
     InitialScanProgress initial_scan_progress = 9;
+    // State of emitting of schema change events
+    bool schema_changes = 10;
 }
 
 message StoragePool {

src/client/discovery/discovery.cpp

@@ -77,7 +77,8 @@ TNodeInfo::TNodeInfo(const Ydb::Discovery::NodeInfo& info)
     , Address(info.address())
     , Location(info.location())
     , Expire(info.expire())
-    {}
+    , BridgePileId(info.has_bridge_pile_id() ? std::make_optional(info.bridge_pile_id()) : std::nullopt)
+{}
 
 TNodeRegistrationResult::TNodeRegistrationResult(TStatus&& status, const Ydb::Discovery::NodeRegistrationResult& proto)
     : TStatus(std::move(status))
@@ -208,6 +209,9 @@ class TDiscoveryClient::TImpl : public TClientImplCommon<TDiscoveryClient::TImpl
         if (!settings.Path_.empty()) {
             request.set_path(TStringType{settings.Path_});
         }
+        if (!settings.BridgePileName_.empty()) {
+            request.set_bridge_pile_name(TStringType{settings.BridgePileName_});
+        }
 
         auto requestLocation = request.mutable_location();
         const auto& location = settings.Location_;

src/client/persqueue_public/ut/read_session_ut.cpp

@@ -513,7 +513,6 @@ class TReadSessionImplTestSetup {
     std::shared_ptr<TCallbackContext<TSingleClusterReadSessionImpl>> CbContext;
     std::shared_ptr<TThreadPool> ThreadPool;
     ::IExecutor::TPtr DefaultExecutor;
-    std::shared_ptr<std::unordered_map<ECodec, THolder<NTopic::ICodec>>> ProvidedCodecs = std::make_shared<std::unordered_map<ECodec, THolder<NTopic::ICodec>>>();
 };
 
 class TReorderingExecutor : public ::IExecutor {
@@ -588,10 +587,6 @@ TReadSessionImplTestSetup::TReadSessionImplTestSetup() {
         .Counters(MakeIntrusive<NYdb::NPersQueue::TReaderCounters>(MakeIntrusive<::NMonitoring::TDynamicCounters>()));
 
     Log.SetFormatter(GetPrefixLogFormatter(""));
-
-    (*ProvidedCodecs)[ECodec::GZIP] = MakeHolder<NTopic::TGzipCodec>();
-    (*ProvidedCodecs)[ECodec::LZOP] = MakeHolder<NTopic::TUnsupportedCodec>();
-    (*ProvidedCodecs)[ECodec::ZSTD] = MakeHolder<NTopic::TZstdCodec>();
 }
 
 TReadSessionImplTestSetup::~TReadSessionImplTestSetup() noexcept(false) {

src/client/persqueue_public/ut/ut_utils/ut_utils.h

@@ -14,12 +14,22 @@ using NYdb::NTopic::IAsyncExecutor;
 
 namespace NYdb::NPersQueue::NTests {
 
+struct TPersQueueYdbSdkTestSetupSettings {
+    TString TestCaseName;
+    bool Start = true;
+    TVector<NKikimrServices::EServiceKikimr> LogServices = ::NPersQueue::TTestServer::LOGGED_SERVICES;
+    NActors::NLog::EPriority LogPriority = NActors::NLog::PRI_DEBUG;
+    ui32 NodeCount = NKikimr::NPersQueueTests::PQ_DEFAULT_NODE_COUNT;
+    size_t TopicPartitionsCount = 1;
+};
+
 class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
     THolder<NYdb::TDriver> Driver;
     THolder<NYdb::NPersQueue::TPersQueueClient> PersQueueClient;
 
     TAdaptiveLock Lock;
 public:
+    // TODO(qyryq) Delete this ctor in favor of TPersQueueYdbSdkTestSetupSettings.
     TPersQueueYdbSdkTestSetup(const TString& testCaseName, bool start = true,
                               const TVector<NKikimrServices::EServiceKikimr>& logServices = ::NPersQueue::TTestServer::LOGGED_SERVICES,
                               NActors::NLog::EPriority logPriority = NActors::NLog::PRI_DEBUG,
@@ -29,6 +39,17 @@ class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
     {
     }
 
+    TPersQueueYdbSdkTestSetup(TPersQueueYdbSdkTestSetupSettings settings)
+        : SDKTestSetup(
+            settings.TestCaseName,
+            settings.Start,
+            settings.LogServices,
+            settings.LogPriority,
+            settings.NodeCount,
+            settings.TopicPartitionsCount)
+    {
+    }
+
     ~TPersQueueYdbSdkTestSetup() {
         if (PersQueueClient) {
             PersQueueClient = nullptr;

src/client/query/query.cpp

@@ -53,7 +53,7 @@ TScriptExecutionOperation::TScriptExecutionOperation(TStatus&& status, Ydb::Oper
         for (const auto& column : resultSetMeta.columns()) {
             columns.emplace_back(column.name(), column.type());
         }
-        Metadata_.ResultSetsMeta.emplace_back(std::move(columns)); 
+        Metadata_.ResultSetsMeta.emplace_back(std::move(columns), resultSetMeta.number_rows(), resultSetMeta.finished()); 
     }
 
     if (metadata.has_script_content()) {