[C++ SDK] Supported topic-to-table transactions in Query Service (#15984)

Author: Gazizonoki

examples/CMakeLists.txt

@@ -4,5 +4,6 @@ add_subdirectory(pagination)
 add_subdirectory(secondary_index)
 add_subdirectory(secondary_index_builtin)
 add_subdirectory(topic_reader)
+add_subdirectory(topic_writer/transaction)
 add_subdirectory(ttl)
 add_subdirectory(vector_index)

examples/basic_example/basic_example.cpp

@@ -286,8 +286,8 @@ void MultiStep(TQueryClient client) {
         }
 
         // Get the active transaction id
-        auto txId = resultValue.GetTransaction()->GetId();
-        
+        auto tx = *resultValue.GetTransaction();
+
         // Processing the request result
         TResultSetParser parser(resultValue.GetResultSet(0));
         parser.TryNextRow();
@@ -328,7 +328,7 @@ void MultiStep(TQueryClient client) {
         // and commit it at the end of the second query execution.
         auto result2 = session.ExecuteQuery(
             query2,
-            TTxControl::Tx(txId).CommitTx(),
+            TTxControl::Tx(tx).CommitTx(),
             params2).GetValueSync();
 
         if (!result2.IsSuccess()) {
@@ -381,7 +381,7 @@ void ExplicitTcl(TQueryClient client) {
         // Execute query.
         // Transaction control settings continues active transaction (tx)
         auto updateResult = session.ExecuteQuery(query,
-            TTxControl::Tx(tx.GetId()),
+            TTxControl::Tx(tx),
             params).GetValueSync();
 
         if (!updateResult.IsSuccess()) {

examples/topic_reader/transaction/CMakeLists.txt

@@ -3,6 +3,7 @@ add_executable(read_from_topic_in_transaction)
 target_link_libraries(read_from_topic_in_transaction PUBLIC
   yutil
   YDB-CPP-SDK::Topic
+  YDB-CPP-SDK::Query
   getopt
 )
 

examples/topic_reader/transaction/application.cpp

@@ -20,10 +20,10 @@ TApplication::TApplication(const TOptions& options)
 
     Driver.emplace(config);
     TopicClient.emplace(*Driver);
-    TableClient.emplace(*Driver);
+    QueryClient.emplace(*Driver);
 
     CreateTopicReadSession(options);
-    CreateTableSession();
+    CreateQuerySession();
 
     TablePath = options.TablePath;
 }
@@ -40,15 +40,15 @@ void TApplication::CreateTopicReadSession(const TOptions& options)
     std::cout << "Topic session was created" << std::endl;
 }
 
-void TApplication::CreateTableSession()
+void TApplication::CreateQuerySession()
 {
-    NYdb::NTable::TCreateSessionSettings settings;
+    NYdb::NQuery::TCreateSessionSettings settings;
 
-    auto result = TableClient->GetSession(settings).GetValueSync();
+    auto result = QueryClient->GetSession(settings).GetValueSync();
 
-    TableSession = result.GetSession();
+    QuerySession = result.GetSession();
 
-    std::cout << "Table session was created" << std::endl;
+    std::cout << "Query session was created" << std::endl;
 }
 
 void TApplication::Run()
@@ -104,10 +104,10 @@ void TApplication::Finalize()
 void TApplication::BeginTransaction()
 {
     Y_ABORT_UNLESS(!Transaction);
-    Y_ABORT_UNLESS(TableSession);
+    Y_ABORT_UNLESS(QuerySession);
 
-    auto settings = NYdb::NTable::TTxSettings::SerializableRW();
-    auto result = TableSession->BeginTransaction(settings).GetValueSync();
+    auto settings = NYdb::NQuery::TTxSettings::SerializableRW();
+    auto result = QuerySession->BeginTransaction(settings).GetValueSync();
 
     Transaction = result.GetTransaction();
 }
@@ -116,7 +116,7 @@ void TApplication::CommitTransaction()
 {
     Y_ABORT_UNLESS(Transaction);
 
-    NYdb::NTable::TCommitTxSettings settings;
+    NYdb::NQuery::TCommitTxSettings settings;
 
     auto result = Transaction->Commit(settings).GetValueSync();
 
@@ -173,20 +173,16 @@ void TApplication::InsertRowsIntoTable()
 
     auto params = builder.Build();
 
-    NYdb::NTable::TExecDataQuerySettings settings;
-    settings.KeepInQueryCache(true);
-
-    auto runQuery = [this, &query, &params, &settings](NYdb::NTable::TSession) -> NYdb::TStatus {
+    auto runQuery = [this, &query, &params](NYdb::NQuery::TSession) -> NYdb::TStatus {
         auto result =
-            Transaction->GetSession().ExecuteDataQuery(query,
-                                                       NYdb::NTable::TTxControl::Tx(*Transaction),
-                                                       params,
-                                                       settings).GetValueSync();
+            Transaction->GetSession().ExecuteQuery(query,
+                                                   NYdb::NQuery::TTxControl::Tx(*Transaction),
+                                                   params).GetValueSync();
 
         return result;
     };
 
-    TableClient->RetryOperationSync(runQuery);
+    QueryClient->RetryQuerySync(runQuery);
 }
 
 void TApplication::AppendTableRow(const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message)

examples/topic_reader/transaction/application.h

@@ -4,7 +4,7 @@
 
 #include <ydb-cpp-sdk/client/driver/driver.h>
 #include <ydb-cpp-sdk/client/topic/client.h>
-#include <ydb-cpp-sdk/client/table/table.h>
+#include <ydb-cpp-sdk/client/query/client.h>
 
 #include <memory>
 #include <optional>
@@ -28,7 +28,7 @@ class TApplication {
     };
 
     void CreateTopicReadSession(const TOptions& options);
-    void CreateTableSession();
+    void CreateQuerySession();
 
     void BeginTransaction();
     void CommitTransaction();
@@ -40,10 +40,10 @@ class TApplication {
 
     std::optional<NYdb::TDriver> Driver;
     std::optional<NYdb::NTopic::TTopicClient> TopicClient;
-    std::optional<NYdb::NTable::TTableClient> TableClient;
+    std::optional<NYdb::NQuery::TQueryClient> QueryClient;
     std::shared_ptr<NYdb::NTopic::IReadSession> ReadSession;
-    std::optional<NYdb::NTable::TSession> TableSession;
-    std::optional<NYdb::NTable::TTransaction> Transaction;
+    std::optional<NYdb::NQuery::TSession> QuerySession;
+    std::optional<NYdb::NQuery::TTransaction> Transaction;
     std::vector<NYdb::NTopic::TReadSessionEvent::TStopPartitionSessionEvent> PendingStopEvents;
     std::vector<TRow> Rows;
     std::string TablePath;

examples/topic_writer/transaction/CMakeLists.txt

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

examples/topic_writer/transaction/main.cpp

@@ -1,8 +1,7 @@
 #include <ydb-cpp-sdk/client/topic/client.h>
-#include <ydb-cpp-sdk/client/table/table.h>
+#include <ydb-cpp-sdk/client/query/client.h>
 
-int main()
-{
+int main() {
     const std::string ENDPOINT = "HOST:PORT";
     const std::string DATABASE = "DATABASE";
     const std::string TOPIC = "PATH/TO/TOPIC";
@@ -12,24 +11,26 @@ int main()
     config.SetDatabase(DATABASE);
     NYdb::TDriver driver(config);
 
-    NYdb::NTable::TTableClient tableClient(driver);
-    auto getTableSessionResult = tableClient.GetSession().GetValueSync();
-    ThrowOnError(getTableSessionResult);
-    auto tableSession = getTableSessionResult.GetSession();
+    NYdb::NQuery::TQueryClient queryClient(driver);
+    auto getSessionResult = queryClient.GetSession().GetValueSync();
+    NYdb::NStatusHelpers::ThrowOnError(getSessionResult);
+    auto session = getSessionResult.GetSession();
 
     NYdb::NTopic::TTopicClient topicClient(driver);
-    auto topicSessionSettings = NYdb::NTopic::TWriteSessionSettings()
-        .Path(TOPIC)
-        .DeduplicationEnabled(true);
-    auto topicSession = topicClient.CreateSimpleBlockingWriteSession(topicSessionSettings);
 
-    auto beginTransactionResult = tableSession.BeginTransaction().GetValueSync();
-    ThrowOnError(beginTransactionResult);
-    auto transaction = beginTransactionResult.GetTransaction();
+    auto topicSession = topicClient.CreateSimpleBlockingWriteSession(
+        NYdb::NTopic::TWriteSessionSettings()
+            .Path(TOPIC)
+            .DeduplicationEnabled(true)
+    );
+
+    auto beginTxResult = session.BeginTransaction(NYdb::NQuery::TTxSettings()).GetValueSync();
+    NYdb::NStatusHelpers::ThrowOnError(beginTxResult);
+    auto tx = beginTxResult.GetTransaction();
 
     NYdb::NTopic::TWriteMessage writeMessage("message");
 
-    topicSession->Write(std::move(writeMessage), &transaction);
+    topicSession->Write(std::move(writeMessage), &tx);
 
-    transaction.Commit().GetValueSync();
+    tx.Commit().GetValueSync();
 }

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

@@ -8,6 +8,7 @@
 #include <ydb-cpp-sdk/client/driver/driver.h>
 #include <ydb-cpp-sdk/client/params/params.h>
 #include <ydb-cpp-sdk/client/retry/retry.h>
+#include <ydb-cpp-sdk/client/types/tx/tx.h>
 #include <ydb-cpp-sdk/client/types/request_settings.h>
 
 namespace NYdb::inline V3 {
@@ -168,30 +169,75 @@ class TCreateSessionResult: public TStatus {
     TSession Session_;
 };
 
-class TTransaction {
+class TTransaction : public TTransactionBase {
     friend class TQueryClient;
     friend class TExecuteQueryIterator::TReaderImpl;
+    friend class TExecQueryImpl;
+
 public:
-    const std::string& GetId() const {
-        return TxId_;
+    bool IsActive() const;
+
+    TAsyncCommitTransactionResult Commit(const TCommitTxSettings& settings = TCommitTxSettings());
+    TAsyncStatus Rollback(const TRollbackTxSettings& settings = TRollbackTxSettings());
+
+    TSession GetSession() const;
+
+    void AddPrecommitCallback(TPrecommitTransactionCallback cb) override;
+    void AddOnFailureCallback(TOnFailureTransactionCallback cb) override;
+
+private:
+    TTransaction(const TSession& session, const std::string& txId);
+
+    TAsyncStatus Precommit() const;
+    NThreading::TFuture<void> ProcessFailure() const;
+
+    class TImpl;
+
+    std::shared_ptr<TImpl> TransactionImpl_;
+};
+
+class TTxControl {
+    friend class TExecQueryImpl;
+    friend class TExecQueryInternal;
+
+public:
+    using TSelf = TTxControl;
+
+    static TTxControl Tx(const TTransaction& tx) {
+        return TTxControl(tx);
     }
 
-    bool IsActive() const {
-        return !TxId_.empty();
+    [[deprecated("This is bug-provoking API. Use TTxControl::Tx(TTransaction) instead. "
+                 "This constructor will be removed in upcomming release")]]
+    static TTxControl Tx(const std::string& txId) {
+        return TTxControl(txId);
     }
 
-    TAsyncCommitTransactionResult Commit(const TCommitTxSettings& settings = TCommitTxSettings());
-    TAsyncStatus Rollback(const TRollbackTxSettings& settings = TRollbackTxSettings());
+    static TTxControl BeginTx(const TTxSettings& settings = TTxSettings()) {
+        return TTxControl(settings);
+    }
 
-    TSession GetSession() const {
-        return Session_;
+    static TTxControl NoTx() {
+        return TTxControl();
     }
 
+    FLUENT_SETTING_FLAG(CommitTx);
+
+    bool HasTx() const { return !std::holds_alternative<std::monostate>(Tx_); }
+
 private:
-    TTransaction(const TSession& session, const std::string& txId);
+    TTxControl() {}
 
-    TSession Session_;
-    std::string TxId_;
+    TTxControl(const TTransaction& tx)
+        : Tx_(tx) {}
+
+    TTxControl(const TTxSettings& txSettings)
+        : Tx_(txSettings) {}
+    
+    TTxControl(const std::string& txId)
+        : Tx_(txId) {}
+
+    const std::variant<std::monostate, TTransaction, TTxSettings, std::string> Tx_;
 };
 
 class TBeginTransactionResult : public TStatus {

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

@@ -27,7 +27,7 @@ class TExecuteQueryPart;
 class TExecuteQueryIterator;
 
 class TTransaction;
-struct TTxControl;
+class TTxControl;
 
 class TQueryContent;
 class TResultSetMeta;

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

@@ -64,6 +64,12 @@ class TExecuteQueryIterator : public TStatus {
     : TStatus(std::move(status))
     , ReaderImpl_(impl) {}
 
+    TExecuteQueryIterator(
+        std::shared_ptr<TReaderImpl> impl,
+        TStatus&& status)
+    : TStatus(std::move(status))
+    , ReaderImpl_(impl) {}
+
     std::shared_ptr<TReaderImpl> ReaderImpl_;
 };