table.hpp

#pragma once
 
#include <ydb-cpp-sdk/client/table/table.h>
 
#include <userver/dynamic_config/source.hpp>
#include <userver/utils/statistics/fwd.hpp>
 
#include <userver/ydb/builder.hpp>
#include <userver/ydb/query.hpp>
#include <userver/ydb/response.hpp>
#include <userver/ydb/settings.hpp>
#include <userver/ydb/transaction.hpp>
 
namespace NMonitoring {
class TMetricRegistry;
}  // namespace NMonitoring
 
USERVER_NAMESPACE_BEGIN
 
namespace tracing {
class Span;
}  // namespace tracing
 
namespace utils {
class RetryBudget;
}  // namespace utils
 
namespace ydb {
 
namespace impl {
struct Stats;
struct TableSettings;
class Driver;
struct RequestContext;
enum class IsStreaming : bool {};
}  // namespace impl
 
using ScanQuerySettings = NYdb::NTable::TStreamExecScanQuerySettings;
 
class TableClient final {
 public:
  /// @cond
  // For internal use only.
  TableClient(impl::TableSettings settings,
              OperationSettings operation_settings,
              dynamic_config::Source config_source,
              std::shared_ptr<impl::Driver> driver);
 
  ~TableClient();
  /// @endcond
 
  /// Query for creating/deleting tables
  void ExecuteSchemeQuery(const std::string& query);
 
  void MakeDirectory(const std::string& path);
  void RemoveDirectory(const std::string& path);
 
  NYdb::NScheme::TDescribePathResult DescribePath(std::string_view path);
  NYdb::NScheme::TListDirectoryResult ListDirectory(std::string_view path);
 
  NYdb::NTable::TDescribeTableResult DescribeTable(std::string_view path);
  void CreateTable(std::string_view path,
                   NYdb::NTable::TTableDescription&& table_desc);
  void DropTable(std::string_view path);
 
  /// @name Data queries execution
  /// Execute a single data query outside of transactions. Query parameters are
  /// passed in `Args` as "string key - value" pairs:
  ///
  /// @code
  /// client.ExecuteDataQuery(query, "name1", value1, "name2", value2, ...);
  /// @endcode
  ///
  /// Use ydb::PreparedArgsBuilder for storing a generic buffer of query params
  /// if needed.
  ///
  /// @{
  template <typename... Args>
  ExecuteResponse ExecuteDataQuery(const Query& query, Args&&... args);
 
  template <typename... Args>
  ExecuteResponse ExecuteDataQuery(OperationSettings settings,
                                   const Query& query, Args&&... args);
 
  ExecuteResponse ExecuteDataQuery(OperationSettings settings,
                                   const Query& query,
                                   PreparedArgsBuilder&& builder);
 
  ExecuteResponse ExecuteDataQuery(QuerySettings query_settings,
                                   OperationSettings settings,
                                   const Query& query,
                                   PreparedArgsBuilder&& builder);
  /// @}
 
  /// @name Transactions
  /// @brief Begin a transaction with the specified name. The settings are used
  /// for the `BEGIN` statement.
  /// @see ydb::Transaction
  ///
  /// @{
  Transaction Begin(std::string transaction_name,
                    OperationSettings settings = {});
 
  Transaction Begin(std::string transaction_name, TransactionMode tx_mode);
  /// @}
 
  /// Builder for storing dynamic query params.
  PreparedArgsBuilder GetBuilder() const;
 
  /// Efficiently write large ranges of table data.
  void BulkUpsert(std::string_view table, NYdb::TValue&& rows,
                  OperationSettings settings = {});
 
  /// Efficiently write large ranges of table data.
  /// The passed range of structs is serialized to TValue.
  template <typename RangeOfStructs>
  void BulkUpsert(std::string_view table, const RangeOfStructs& rows,
                  OperationSettings settings = {});
 
  /// Efficiently read large ranges of table data.
  ReadTableResults ReadTable(
      std::string_view table,
      NYdb::NTable::TReadTableSettings&& read_settings = {},
      OperationSettings settings = {});
 
  /// @name Scan queries execution
  /// A separate data access interface designed primarily for performing
  /// analytical ad-hoc queries.
  /// @{
  template <typename... Args>
  ScanQueryResults ExecuteScanQuery(const Query& query, Args&&... args);
 
  template <typename... Args>
  ScanQueryResults ExecuteScanQuery(ScanQuerySettings&& scan_settings,
                                    OperationSettings settings,
                                    const Query& query, Args&&... args);
 
  ScanQueryResults ExecuteScanQuery(ScanQuerySettings&& scan_settings,
                                    OperationSettings settings,
                                    const Query& query,
                                    PreparedArgsBuilder&& builder);
  /// @}
 
  /// @cond
  // For internal use only.
  friend void DumpMetric(utils::statistics::Writer& writer,
                         const TableClient& table_client);
  /// @endcond
 
  /// Get native table client
  /// @warning Use with care! Facilities from
  /// `<core/include/userver/drivers/subscribable_futures.hpp>` can help with
  /// non-blocking wait operations.
  NYdb::NTable::TTableClient& GetNativeTableClient();
 
  utils::RetryBudget& GetRetryBudget();
 
 private:
  friend class Transaction;
  friend struct impl::RequestContext;
 
  std::string JoinDbPath(std::string_view path) const;
 
  void Select1();
 
  NYdb::NTable::TExecDataQuerySettings ToExecQuerySettings(
      QuerySettings query_settings) const;
 
  template <typename... Args>
  PreparedArgsBuilder MakeBuilder(Args&&... args);
 
  // Func: (TSession, const std::string& full_path, const Settings&)
  //       -> NThreading::TFuture<T>
  //       OR
  //       (TTableClient&, const std::string& full_path, const Settings&)
  //       -> NThreading::TFuture<T>
  // ExecuteSchemeQueryImpl -> T
  template <typename Settings, typename Func>
  auto ExecuteWithPathImpl(std::string_view path,
                           std::string_view operation_name,
                           OperationSettings&& settings, Func&& func);
 
  dynamic_config::Source config_source_;
  const OperationSettings default_settings_;
  const bool keep_in_query_cache_;
  std::unique_ptr<impl::Stats> stats_;
  std::shared_ptr<impl::Driver> driver_;
  std::unique_ptr<NYdb::NScheme::TSchemeClient> scheme_client_;
  std::unique_ptr<NYdb::NTable::TTableClient> table_client_;
};
 
template <typename... Args>
PreparedArgsBuilder TableClient::MakeBuilder(Args&&... args) {
  auto builder = GetBuilder();
  builder.AddParams(std::forward<Args>(args)...);
  return builder;
}
 
template <typename... Args>
ExecuteResponse TableClient::ExecuteDataQuery(const Query& query,
                                              Args&&... args) {
  return ExecuteDataQuery(OperationSettings{}, query,
                          MakeBuilder(std::forward<Args>(args)...));
}
 
template <typename... Args>
ExecuteResponse TableClient::ExecuteDataQuery(OperationSettings settings,
                                              const Query& query,
                                              Args&&... args) {
  return ExecuteDataQuery(settings, query,
                          MakeBuilder(std::forward<Args>(args)...));
}
 
template <typename RangeOfStructs>
void TableClient::BulkUpsert(std::string_view table, const RangeOfStructs& rows,
                             OperationSettings settings) {
  NYdb::TValueBuilder builder;
  ydb::Write(builder, rows);
  BulkUpsert(table, builder.Build(), std::move(settings));
}
 
template <typename... Args>
ScanQueryResults TableClient::ExecuteScanQuery(const Query& query,
                                               Args&&... args) {
  return ExecuteScanQuery(ScanQuerySettings{}, OperationSettings{}, query,
                          MakeBuilder(std::forward<Args>(args)...));
}
 
template <typename... Args>
ScanQueryResults TableClient::ExecuteScanQuery(
    ScanQuerySettings&& scan_settings, OperationSettings settings,
    const Query& query, Args&&... args) {
  return ExecuteScanQuery(std::move(scan_settings), std::move(settings), query,
                          MakeBuilder(std::forward<Args>(args)...));
}
 
}  // namespace ydb
 
USERVER_NAMESPACE_END