Import YDB C++ SDK 36

.github/import_generation.txt

@@ -1 +1 @@
-36
+37

.github/last_commit.txt

@@ -1 +1 @@
-3b235ed1f2fc3977cfc6f99a74123c0097ef9795
+abde8e11540c2ccb2c8b2d9d15b2bed3d3fbaf7e

CHANGELOG.md

@@ -0,0 +1,8 @@
+* EXPERIMENTAL! Added `IProducer` interface to the SDK. This interface is used to write messages to a topic.
+Each message can be associated with a partitioning key, which is used to determine the partition to which the message will be written.
+
+* Added gRPC load balancing policy option for `TDriver`. Default policy: `round_robin`.
+
+* Removed the `layout` field from `FulltextIndexSettings` and replaced it with separate index types in `TableIndexDescription`.
+  `layout` was a preliminary version of API, actual YDB release 26-1 uses separate index types, so please note that creating
+  full text indexes via gRPC won't work with previous versions of SDK.

examples/CMakeLists.txt

@@ -6,6 +6,7 @@ add_subdirectory(secondary_index_builtin)
 add_subdirectory(time)
 add_subdirectory(topic_reader)
 add_subdirectory(topic_writer/transaction)
+add_subdirectory(topic_writer/producer/basic_write)
 add_subdirectory(ttl)
 add_subdirectory(vector_index)
 add_subdirectory(vector_index_builtin)

examples/executor/main.cpp

@@ -4,20 +4,13 @@
 
 #include <library/cpp/getopt/last_getopt.h>
 
-#include <util/thread/pool.h>
-
 #include <thread>
 
 
 void ExecutorExample(const std::string& endpoint, const std::string& database) {
     auto driverConfig = NYdb::CreateFromEnvironment(endpoint + "/?database=" + database)
-        .SetExecutor(NYdb::CreateThreadPoolExecutorAdapter(
-            std::make_shared<TThreadPool>(TThreadPool::TParams()
-                .SetBlocking(true)
-                .SetCatching(false)
-                .SetForkAware(false)),
-            std::thread::hardware_concurrency())
-        );
+        .SetExecutor(NYdb::CreateThreadPoolExecutor(std::thread::hardware_concurrency(), 0)
+    );
 
     NYdb::TDriver driver(driverConfig);
     NYdb::NQuery::TQueryClient client(driver);

examples/topic_writer/producer/basic_write/CMakeLists.txt

@@ -0,0 +1,36 @@
+add_executable(topic_writer_producer_buffer_overloaded)
+
+target_link_libraries(topic_writer_producer_buffer_overloaded
+  PUBLIC
+    yutil
+    YDB-CPP-SDK::Topic
+    YDB-CPP-SDK::Query
+)
+
+target_sources(topic_writer_producer_buffer_overloaded
+  PRIVATE
+    main.cpp
+)
+
+vcs_info(topic_writer_producer_buffer_overloaded)
+
+if (CMAKE_SYSTEM_PROCESSOR STREQUAL "x86_64" OR CMAKE_SYSTEM_PROCESSOR STREQUAL "AMD64")
+  target_link_libraries(topic_writer_producer_buffer_overloaded PUBLIC
+    cpuid_check
+  )
+endif()
+
+if (CMAKE_SYSTEM_NAME STREQUAL "Linux")
+  target_link_options(topic_writer_producer_buffer_overloaded PRIVATE
+    -ldl
+    -lrt
+    -Wl,--no-as-needed
+    -lpthread
+  )
+elseif (CMAKE_SYSTEM_NAME STREQUAL "Darwin")
+  target_link_options(topic_writer_producer_buffer_overloaded PRIVATE
+    -Wl,-platform_version,macos,11.0,11.0
+    -framework
+    CoreFoundation
+  )
+endif()

examples/topic_writer/producer/basic_write/main.cpp

@@ -0,0 +1,123 @@
+#include <thread>
+#include <ydb-cpp-sdk/client/topic/client.h>
+#include <ydb-cpp-sdk/client/query/client.h>
+
+#include <util/generic/serialized_enum.h>
+
+std::shared_ptr<NYdb::NTopic::IProducer> CreateProducer(const std::string& topic, NYdb::NTopic::TTopicClient& topicClient) {
+    NYdb::NTopic::TProducerSettings producerSettings;
+    producerSettings.Path(topic);
+    producerSettings.Codec(NYdb::NTopic::ECodec::ZSTD);
+    producerSettings.ProducerIdPrefix("producer_basic");
+    producerSettings.PartitionChooserStrategy(NYdb::NTopic::TProducerSettings::EPartitionChooserStrategy::Bound);
+    producerSettings.SubSessionIdleTimeout(TDuration::Seconds(30));
+    producerSettings.MaxBlock(TDuration::Seconds(30));
+    producerSettings.MaxMemoryUsage(100_MB);
+    return topicClient.CreateProducer(producerSettings);
+}
+
+std::string GetResultStatus(const NYdb::NTopic::TWriteResult& result) {
+    switch (result.Status) {
+        case NYdb::NTopic::EWriteStatus::Queued:
+            return "Queued";
+        case NYdb::NTopic::EWriteStatus::Error:
+            return "Error";
+        case NYdb::NTopic::EWriteStatus::Timeout:
+            return "Timeout";
+    }
+}
+
+std::string GetResultStatus(const NYdb::NTopic::TFlushResult& result) {
+    switch (result.Status) {
+        case NYdb::NTopic::EFlushStatus::Success:
+            return "Success";
+        case NYdb::NTopic::EFlushStatus::ProducerClosed:
+            return "ProducerClosed";
+    }
+}
+
+std::string GetErrorMessage(const NYdb::NTopic::TFlushResult& result) {
+    std::string errorMessage = "error occurred while flushing messages";
+    errorMessage += ", flush status: " + GetResultStatus(result);
+    errorMessage += ", last written sequence number: " + ToString(result.LastWrittenSeqNo);
+    if (result.ClosedDescription) {
+        errorMessage += ", producer is closed: ";
+        errorMessage += result.ClosedDescription.value().DebugString();
+    }
+    return errorMessage;
+}
+
+std::string GetErrorMessage(const NYdb::NTopic::TWriteResult& result) {
+    std::string errorMessage = "error occurred while writing message";
+    errorMessage += ", write status: " + GetResultStatus(result);
+    if (result.ErrorMessage) {
+        errorMessage += ", reason: ";
+        errorMessage += result.ErrorMessage.value();
+    }
+    if (result.ClosedDescription) {
+        errorMessage += ", producer is closed: ";
+        errorMessage += result.ClosedDescription.value().DebugString();
+    }
+    return errorMessage;
+}
+
+void WriteWithHandlingResult(std::shared_ptr<NYdb::NTopic::IProducer> producer, NYdb::NTopic::TWriteMessage&& writeMessage) {
+    static constexpr size_t MAX_RETRIES = 10;
+
+    for (size_t retries = 0; retries < MAX_RETRIES; retries++) {
+        auto writeResult = producer->Write(std::move(writeMessage));
+        if (writeResult.IsSuccess()) {
+            // if write was successful, we can continue writing messages
+            continue;
+        }
+
+        if (writeResult.IsError()) {
+            // this means that some non retryable error occurred, for example, producer was closed due to user error
+            // in this case we need to stop retrying and see the close description (to simplify the example, we just print it to standard error)
+            std::cerr << GetErrorMessage(writeResult) << std::endl;
+            return;
+        }
+
+        if (writeResult.IsTimeout()) {
+            // when timeout occurs this means that producer's buffer is overloaded by memory (see MaxMemoryUsage setting)
+            // so we need to wait for some time and try to write again later
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            continue;
+        }
+    }
+
+    auto flushResult = producer->Flush().GetValueSync();
+    if (flushResult.IsSuccess()) {
+        // if flush was successful, we can return, because all messages were written to the server
+        return;
+    }
+
+    if (flushResult.IsClosed()) {
+        // if flush was not successful, this means that producer was closed due to non retryable error
+        // in this case we should see the close description (to simplify the example, we just print it to standard error)
+        std::cerr << GetErrorMessage(flushResult) << std::endl;
+    }
+}
+
+int main() {
+    const std::string ENDPOINT = "HOST:PORT";
+    const std::string DATABASE = "DATABASE";
+    const std::string TOPIC = "PATH/TO/TOPIC";
+
+    NYdb::TDriverConfig config;
+    config.SetEndpoint(ENDPOINT);
+    config.SetDatabase(DATABASE);
+    NYdb::TDriver driver(config);
+
+    NYdb::NTopic::TTopicClient topicClient(driver);
+
+    auto producer = CreateProducer(TOPIC, topicClient);
+    auto messageData = std::string(1_KB, 'a');
+
+    for (int i = 0; i < 10; i++) {
+        NYdb::NTopic::TWriteMessage writeMessage(messageData);
+        writeMessage.Key("key" + ToString(i));
+        WriteWithHandlingResult(producer, std::move(writeMessage));
+    }
+    return 0;
+}

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

@@ -122,6 +122,7 @@ class TDriverConfig {
     //! Params is a optionally field to set policy settings
     //! default: EBalancingPolicy::UsePreferableLocation
     TDriverConfig& SetBalancingPolicy(EBalancingPolicy policy, const std::string& params = "");
+
     //! Set grpc level keep alive. If keepalive ping was delayed more than given timeout
     //! internal grpc routine fails request with TRANSIENT_FAILURE or TRANSPORT_UNAVAILABLE error
     //! Note: this timeout should not be too small to prevent fail due to
@@ -131,6 +132,11 @@ class TDriverConfig {
     TDriverConfig& SetGRpcKeepAliveTimeout(TDuration timeout);
     TDriverConfig& SetGRpcKeepAlivePermitWithoutCalls(bool permitWithoutCalls);
 
+    //! Set grpc load balancing policy
+    //! policy - name of the load balancing policy, see grpc documentation for available policies
+    //! default: "round_robin"
+    TDriverConfig& SetGRpcLoadBalancingPolicy(const std::string& policy);
+
     //! Set inactive socket timeout.
     //! Used to close connections, that were inactive for given time.
     //! Closes unused connections every 1/10 of timeout, so deletion time is approximate.

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

@@ -169,6 +169,7 @@ struct TImportFromFsSettings : public TOperationRequestSettings<TImportFromFsSet
     FLUENT_SETTING_OPTIONAL(uint32_t, NumberOfRetries);
     FLUENT_SETTING_OPTIONAL(bool, NoACL);
     FLUENT_SETTING_OPTIONAL(bool, SkipChecksumValidation);
+    FLUENT_SETTING_DEFAULT(EIndexPopulationMode, IndexPopulationMode, EIndexPopulationMode::Build);
     FLUENT_SETTING_VECTOR(std::string, ExcludeRegexp);
 };
 

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

@@ -230,6 +230,14 @@ class TReadReplicasSettings {
 };
 
 struct TGlobalIndexSettings {
+    static constexpr const int VectorKMeansTreeLevelTablePosition = 0;
+    static constexpr const int VectorKMeansTreePostingTablePosition = 1;
+    static constexpr const int VectorKMeansTreePrefixTablePosition = 2;
+    static constexpr const int FulltextRelevanceDictTablePosition = 0;
+    static constexpr const int FulltextRelevanceDocsTablePosition = 1;
+    static constexpr const int FulltextRelevanceStatsTablePosition = 2;
+    static constexpr const int FulltextRelevancePostingTablePosition = 3;
+
     using TUniformOrExplicitPartitions = std::variant<std::monostate, uint64_t, TExplicitPartitions>;
 
     TPartitioningSettings PartitioningSettings;
@@ -305,12 +313,6 @@ struct TKMeansTreeSettings {
 
 struct TFulltextIndexSettings {
 public:
-    enum class ELayout {
-        Unspecified = 0,
-        Flat,
-        FlatRelevance,
-    };
-
     enum class ETokenizer {
         Unspecified = 0,
         Whitespace,
@@ -337,7 +339,6 @@ struct TFulltextIndexSettings {
         std::optional<TAnalyzers> Analyzers;
     };
 
-    std::optional<ELayout> Layout;
     std::vector<TColumnAnalyzers> Columns;
 
     static TFulltextIndexSettings FromProto(const Ydb::Table::FulltextIndexSettings& proto);
@@ -458,6 +459,8 @@ class TChangefeedDescription {
     TChangefeedDescription& WithRetentionPeriod(const TDuration& value);
     // Initial scan will output the current state of the table first
     TChangefeedDescription& WithInitialScan();
+    // Enable UserSIDs
+    TChangefeedDescription& WithUserSIDs();
     // Attributes
     TChangefeedDescription& AddAttribute(const std::string& key, const std::string& value);
     TChangefeedDescription& SetAttributes(const std::unordered_map<std::string, std::string>& attrs);
@@ -473,6 +476,7 @@ class TChangefeedDescription {
     bool GetSchemaChanges() const;
     const std::optional<TDuration>& GetResolvedTimestamps() const;
     bool GetInitialScan() const;
+    bool GetUserSIDs() const;
     const std::unordered_map<std::string, std::string>& GetAttributes() const;
     const std::string& GetAwsRegion() const;
     const std::optional<TInitialScanProgress>& GetInitialScanProgress() const;
@@ -502,6 +506,7 @@ class TChangefeedDescription {
     std::optional<TDuration> ResolvedTimestamps_;
     std::optional<TDuration> RetentionPeriod_;
     bool InitialScan_ = false;
+    bool UserSIDs_ = false;
     std::unordered_map<std::string, std::string> Attributes_;
     std::string AwsRegion_;
     std::optional<TInitialScanProgress> InitialScanProgress_;
@@ -808,8 +813,8 @@ class TTableDescription {
     void AddVectorKMeansTreeIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const TKMeansTreeSettings& indexSettings);
     void AddVectorKMeansTreeIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns, const TKMeansTreeSettings& indexSettings);
     // fulltext
-    void AddFulltextIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const TFulltextIndexSettings& indexSettings);
-    void AddFulltextIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns, const TFulltextIndexSettings& indexSettings);
+    void AddFulltextIndex(const std::string& indexName, EIndexType type, const std::vector<std::string>& indexColumns, const TFulltextIndexSettings& indexSettings);
+    void AddFulltextIndex(const std::string& indexName, EIndexType type, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns, const TFulltextIndexSettings& indexSettings);
 
     // default
     void AddSecondaryIndex(const std::string& indexName, const std::vector<std::string>& indexColumns);
@@ -1054,6 +1059,8 @@ class TTableBuilder {
     // fulltext
     TTableBuilder& AddFulltextIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const TFulltextIndexSettings& indexSettings);
     TTableBuilder& AddFulltextIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns, const TFulltextIndexSettings& indexSettings);
+    TTableBuilder& AddFulltextRelevanceIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const TFulltextIndexSettings& indexSettings);
+    TTableBuilder& AddFulltextRelevanceIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns, const TFulltextIndexSettings& indexSettings);
 
     // default
     TTableBuilder& AddSecondaryIndex(const std::string& indexName, const std::vector<std::string>& indexColumns, const std::vector<std::string>& dataColumns);

include/ydb-cpp-sdk/client/topic/CMakeLists.txt

@@ -26,4 +26,10 @@ generate_enum_serilization(client-ydb_topic-include
   include/ydb-cpp-sdk/client/topic/write_events.h
 )
 
+generate_enum_serilization(client-ydb_topic-include
+  ${YDB_SDK_SOURCE_DIR}/include/ydb-cpp-sdk/client/topic/producer.h
+  INCLUDE_HEADERS
+  include/ydb-cpp-sdk/client/topic/producer.h
+)
+
 _ydb_sdk_install_targets(TARGETS client-ydb_topic-include)

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

@@ -3,6 +3,7 @@
 #include "control_plane.h"
 #include "read_session.h"
 #include "write_session.h"
+#include "producer.h"
 
 namespace NYdb::inline V3::NTopic {
 
@@ -49,11 +50,14 @@ class TTopicClient {
     //! Create write session.
     std::shared_ptr<ISimpleBlockingWriteSession> CreateSimpleBlockingWriteSession(const TWriteSessionSettings& settings);
 
-    //! Create simple blocking keyed write session. Experimental feature. DO NOT USE IN PRODUCTION.
-    std::shared_ptr<ISimpleBlockingKeyedWriteSession> CreateSimpleBlockingKeyedWriteSession(const TKeyedWriteSessionSettings& settings);
+    //! Create producer. Experimental feature. DO NOT USE IN PRODUCTION.
+    std::shared_ptr<IProducer> CreateProducer(const TProducerSettings& settings);
 
-    //! Create keyed write session. Experimental feature. DO NOT USE IN PRODUCTION.
-    std::shared_ptr<IKeyedWriteSession> CreateKeyedWriteSession(const TKeyedWriteSessionSettings& settings);
+    //! Create typed producer.
+    template<typename T>
+    std::shared_ptr<TTypedProducer<T>> CreateTypedProducer(const TProducerSettings& settings) {
+        return std::make_shared<TTypedProducer<T>>(CreateProducer(settings));
+    }
 
     //! Create write session.
     std::shared_ptr<IWriteSession> CreateWriteSession(const TWriteSessionSettings& settings);