Speedup bulkupsert (#17333) (#19793)

eivanov89 · github-actions[bot] · commit 76f4ca693d69 · 2025-06-27T00:15:52.000Z
diff --git a/.github/last_commit.txt b/.github/last_commit.txt
@@ -1 +1 @@
-71f3186a075dd5aa34e2840ef29fc3958a89d2fa
+81ac0de27c1d3727dcc4ef3a2f2581f441e57e41
diff --git a/include/ydb-cpp-sdk/client/value/value.h b/include/ydb-cpp-sdk/client/value/value.h
@@ -262,10 +262,17 @@ struct TUuidValue {
     } Buf_;
 };
 
+namespace NTable {
+
+class TTableClient;
+
+} // namespace NTable
+
 //! Representation of YDB value.
 class TValue {
     friend class TValueParser;
     friend class TProtoAccessor;
+    friend class ::NYdb::Dev::NTable::TTableClient;
 public:
     TValue(const TType& type, const Ydb::Value& valueProto);
     TValue(const TType& type, Ydb::Value&& valueProto);
diff --git a/include/ydb-cpp-sdk/library/retry/retry_policy.h b/include/ydb-cpp-sdk/library/retry/retry_policy.h
@@ -1,7 +1,6 @@
 #pragma once
 
 #include <util/datetime/base.h>
-#include <util/generic/maybe.h>
 #include <util/generic/typetraits.h>
 #include <util/random/random.h>
 
@@ -42,7 +41,7 @@ struct IRetryPolicy {
 
         //! Calculate delay before next retry if next retry is allowed.
         //! Returns empty maybe if retry is not allowed anymore.
-        [[nodiscard]] virtual TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) = 0;
+        [[nodiscard]] virtual std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) = 0;
     };
 
     virtual ~IRetryPolicy() = default;
@@ -82,8 +81,8 @@ struct TNoRetryPolicy : IRetryPolicy<TArgs...> {
     using IRetryState = typename IRetryPolicy<TArgs...>::IRetryState;
 
     struct TNoRetryState : IRetryState {
-        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam...) override {
-            return {};
+        std::optional<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam...) override {
+            return std::nullopt;
         }
     };
 
@@ -124,10 +123,10 @@ struct TExponentialBackoffPolicy : IRetryPolicy<TArgs...> {
         {
         }
 
-        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
+        std::optional<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 {};
+                return std::nullopt;
             }
 
             if (errorClass == ERetryErrorClass::LongRetry) {
@@ -213,10 +212,10 @@ struct TFixedIntervalPolicy : IRetryPolicy<TArgs...> {
         {
         }
 
-        TMaybe<TDuration> GetNextRetryDelay(typename TTypeTraits<TArgs>::TFuncParam... args) override {
+        std::optional<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 {};
+                return std::nullopt;
             }
 
             const TDuration delay = NRetryDetails::RandomizeDelay(errorClass == ERetryErrorClass::LongRetry ? LongRetryDelay : Delay);
diff --git a/src/client/table/impl/table_client.cpp b/src/client/table/impl/table_client.cpp
@@ -989,11 +989,16 @@ void TTableClient::TImpl::SetStatCollector(const NSdkStats::TStatCollector::TCli
     SessionRemovedDueBalancing.Set(collector.SessionRemovedDueBalancing);
 }
 
-TAsyncBulkUpsertResult TTableClient::TImpl::BulkUpsert(const std::string& table, TValue&& rows, const TBulkUpsertSettings& settings) {
+TAsyncBulkUpsertResult TTableClient::TImpl::BulkUpsert(const std::string& table, TValue&& rows, const TBulkUpsertSettings& settings, bool canMove) {
     auto request = MakeOperationRequest<Ydb::Table::BulkUpsertRequest>(settings);
     request.set_table(TStringType{table});
-    *request.mutable_rows()->mutable_type() = TProtoAccessor::GetProto(rows.GetType());
-    *request.mutable_rows()->mutable_value() = rows.GetProto();
+    if (canMove) {
+        request.mutable_rows()->mutable_type()->Swap(&rows.GetType().GetProto());
+        request.mutable_rows()->mutable_value()->Swap(&rows.GetProto());
+    } else {
+        *request.mutable_rows()->mutable_type() = TProtoAccessor::GetProto(rows.GetType());
+        *request.mutable_rows()->mutable_value() = rows.GetProto();
+    }
 
     auto promise = NewPromise<TBulkUpsertResult>();
 
diff --git a/src/client/table/impl/table_client.h b/src/client/table/impl/table_client.h
@@ -138,7 +138,7 @@ class TTableClient::TImpl: public TClientImplCommon<TTableClient::TImpl>, public
 
     void SetStatCollector(const NSdkStats::TStatCollector::TClientStatCollector& collector);
 
-    TAsyncBulkUpsertResult BulkUpsert(const std::string& table, TValue&& rows, const TBulkUpsertSettings& settings);
+    TAsyncBulkUpsertResult BulkUpsert(const std::string& table, TValue&& rows, const TBulkUpsertSettings& settings, bool canMove);
     TAsyncBulkUpsertResult BulkUpsert(const std::string& table, EDataFormat format,
         const std::string& data, const std::string& schema, const TBulkUpsertSettings& settings);
 
diff --git a/src/client/table/table.cpp b/src/client/table/table.cpp
@@ -1460,7 +1460,7 @@ NThreading::TFuture<void> TTableClient::Stop() {
 TAsyncBulkUpsertResult TTableClient::BulkUpsert(const std::string& table, TValue&& rows,
     const TBulkUpsertSettings& settings)
 {
-    return Impl_->BulkUpsert(table, std::move(rows), settings);
+    return Impl_->BulkUpsert(table, std::move(rows), settings, rows.Impl_.use_count() == 1);
 }
 
 TAsyncBulkUpsertResult TTableClient::BulkUpsert(const std::string& table, EDataFormat format,
diff --git a/src/library/retry/retry.cpp b/src/library/retry/retry.cpp
@@ -17,9 +17,9 @@ class TRetryOptionsWithRetCodePolicy : public IRetryPolicy<bool> {
         {
         }
 
-        TMaybe<TDuration> GetNextRetryDelay(bool ret) override {
+        std::optional<TDuration> GetNextRetryDelay(bool ret) override {
             if (ret || Attempt == Opts.RetryCount) {
-                return {};
+                return std::nullopt;
             }
             return Opts.GetTimeToSleep(Attempt++);
         }
diff --git a/src/library/retry/retry.h b/src/library/retry/retry.h
@@ -104,9 +104,9 @@ class TRetryOptionsPolicy : public IRetryPolicy<const TException&> {
         {
         }
 
-        TMaybe<TDuration> GetNextRetryDelay(const TException&) override {
+        std::optional<TDuration> GetNextRetryDelay(const TException&) override {
             if (Attempt == Opts.RetryCount) {
-                return {};
+                return std::nullopt;
             }
             return Opts.GetTimeToSleep(Attempt++);
         }
@@ -151,7 +151,7 @@ std::optional<TResult> DoWithRetry(std::function<TResult()> func, const typename
                 retryState = retryPolicy->CreateRetryState();
             }
 
-            if (const TMaybe<TDuration> delay = retryState->GetNextRetryDelay(ex)) {
+            if (const std::optional<TDuration> delay = retryState->GetNextRetryDelay(ex)) {
                 if (*delay) {
                     if (sleepFunction) {
                         sleepFunction(*delay);
@@ -167,7 +167,7 @@ std::optional<TResult> DoWithRetry(std::function<TResult()> func, const typename
             }
         }
     }
-    return {};
+    return std::nullopt;
 }
 
 template <typename TResult, typename TException = yexception>
@@ -204,7 +204,7 @@ TRetCode DoWithRetryOnRetCode(std::function<TRetCode()> func, const typename IRe
     auto retryState = retryPolicy->CreateRetryState();
     while (true) {
         TRetCode code = func();
-        if (const TMaybe<TDuration> delay = retryState->GetNextRetryDelay(code)) {
+        if (const std::optional<TDuration> delay = retryState->GetNextRetryDelay(code)) {
             if (*delay) {
                 if (sleepFunction) {
                     sleepFunction(*delay);

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-71f3186a075dd5aa34e2840ef29fc3958a89d2fa`
	`1`	`+81ac0de27c1d3727dcc4ef3a2f2581f441e57e41`
Original file line number	Diff line number	Diff line change
`@@ -1460,7 +1460,7 @@ NThreading::TFuture<void> TTableClient::Stop() {`
`1460`	`1460`	`TAsyncBulkUpsertResult TTableClient::BulkUpsert(const std::string& table, TValue&& rows,`
`1461`	`1461`	`const TBulkUpsertSettings& settings)`
`1462`	`1462`	`{`
`1463`		`- return Impl_->BulkUpsert(table, std::move(rows), settings);`
	`1463`	`+ return Impl_->BulkUpsert(table, std::move(rows), settings, rows.Impl_.use_count() == 1);`
`1464`	`1464`	`}`
`1465`	`1465`
`1466`	`1466`	`TAsyncBulkUpsertResult TTableClient::BulkUpsert(const std::string& table, EDataFormat format,`
Original file line number	Diff line number	Diff line change
`@@ -17,9 +17,9 @@ class TRetryOptionsWithRetCodePolicy : public IRetryPolicy<bool> {`
`17`	`17`	`{`
`18`	`18`	`}`
`19`	`19`
`20`		`- TMaybe<TDuration> GetNextRetryDelay(bool ret) override {`
	`20`	`+ std::optional<TDuration> GetNextRetryDelay(bool ret) override {`
`21`	`21`	`if (ret \|\| Attempt == Opts.RetryCount) {`
`22`		`- return {};`
	`22`	`+ return std::nullopt;`
`23`	`23`	`}`
`24`	`24`	`return Opts.GetTimeToSleep(Attempt++);`
`25`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -104,9 +104,9 @@ class TRetryOptionsPolicy : public IRetryPolicy<const TException&> {`
`104`	`104`	`{`
`105`	`105`	`}`
`106`	`106`
`107`		`- TMaybe<TDuration> GetNextRetryDelay(const TException&) override {`
	`107`	`+ std::optional<TDuration> GetNextRetryDelay(const TException&) override {`
`108`	`108`	`if (Attempt == Opts.RetryCount) {`
`109`		`- return {};`
	`109`	`+ return std::nullopt;`
`110`	`110`	`}`
`111`	`111`	`return Opts.GetTimeToSleep(Attempt++);`
`112`	`112`	`}`
`@@ -151,7 +151,7 @@ std::optional<TResult> DoWithRetry(std::function<TResult()> func, const typename`
`151`	`151`	`retryState = retryPolicy->CreateRetryState();`
`152`	`152`	`}`
`153`	`153`
`154`		`- if (const TMaybe<TDuration> delay = retryState->GetNextRetryDelay(ex)) {`
	`154`	`+ if (const std::optional<TDuration> delay = retryState->GetNextRetryDelay(ex)) {`
`155`	`155`	`if (*delay) {`
`156`	`156`	`if (sleepFunction) {`
`157`	`157`	`sleepFunction(*delay);`
`@@ -167,7 +167,7 @@ std::optional<TResult> DoWithRetry(std::function<TResult()> func, const typename`
`167`	`167`	`}`
`168`	`168`	`}`
`169`	`169`	`}`
`170`		`- return {};`
	`170`	`+ return std::nullopt;`
`171`	`171`	`}`
`172`	`172`
`173`	`173`	`template <typename TResult, typename TException = yexception>`
`@@ -204,7 +204,7 @@ TRetCode DoWithRetryOnRetCode(std::function<TRetCode()> func, const typename IRe`
`204`	`204`	`auto retryState = retryPolicy->CreateRetryState();`
`205`	`205`	`while (true) {`
`206`	`206`	`TRetCode code = func();`
`207`		`- if (const TMaybe<TDuration> delay = retryState->GetNextRetryDelay(code)) {`
	`207`	`+ if (const std::optional<TDuration> delay = retryState->GetNextRetryDelay(code)) {`
`208`	`208`	`if (*delay) {`
`209`	`209`	`if (sleepFunction) {`
`210`	`210`	`sleepFunction(*delay);`