transaction.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/storages/postgres/transaction.hpp
/// @brief Transactions
 
#include <memory>
#include <string>
 
#include <userver/storages/postgres/detail/connection_ptr.hpp>
#include <userver/storages/postgres/detail/query_parameters.hpp>
#include <userver/storages/postgres/detail/time_types.hpp>
#include <userver/storages/postgres/options.hpp>
#include <userver/storages/postgres/parameter_store.hpp>
#include <userver/storages/postgres/portal.hpp>
#include <userver/storages/postgres/postgres_fwd.hpp>
#include <userver/storages/postgres/query.hpp>
#include <userver/storages/postgres/result_set.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace storages::postgres {
 
/// @page pg_transactions uPg: Transactions
///
/// All queries that are run on a PostgreSQL cluster are executed inside
/// a transaction, even if a single-query interface is used.
///
/// A uPg transaction can be started using all isolation levels and modes
/// supported by PostgreSQL server as specified in documentation here
/// https://www.postgresql.org/docs/current/static/sql-set-transaction.html.
/// When starting a transaction, the options are specified using
/// TransactionOptions structure.
///
/// For convenience and improvement of readability there are constants
/// defined: Transaction::RW, Transaction::RO and Transaction::Deferrable.
///
/// @see TransactionOptions
///
/// Transaction object ensures that a transaction started in a PostgreSQL
/// connection will be either committed or rolled back and the connection
/// will returned back to a connection pool.
///
/// @todo Code snippet with transaction starting and committing
///
/// Next: @ref pg_run_queries
///
/// See also: @ref pg_process_results
///
/// ----------
///
/// @htmlonly <div class="bottom-nav"> @endhtmlonly
/// ⇦ @ref pg_driver | @ref pg_run_queries ⇨
/// @htmlonly </div> @endhtmlonly
 
/// @page pg_run_queries uPg: Running queries
///
/// All queries are executed through a transaction object, event when being
/// executed through singe-query interface, so here only executing queries
/// with transaction will be covered. Single-query interface is basically
/// the same except for additional options.
///
/// uPg provides means to execute text queries only. There is no query
/// generation, but can be used by other tools to execute SQL queries.
///
/// @warning A query must contain a single query, multiple statements delimited
/// by ';' are not supported.
///
/// All queries are parsed and prepared during the first invocation and are
/// executed as prepared statements afterwards.
///
/// Any query execution can throw an exception. Please see @ref pg_errors for
/// more information on possible errors.
///
/// @par Queries without parameters
///
/// Executing a query without any parameters is rather straightforward.
/// @code
/// auto trx = cluster->Begin(/* transaction options */);
/// auto res = trx.Execute("select foo, bar from foobar");
/// trx.Commit();
/// @endcode
///
/// The cluster also provides interface for single queries
/// @code
/// auto res = cluster->Execute(/* transaction options */, /* the statement */);
/// @endcode
///
/// @par Queries with parameters
///
/// uPg supports SQL dollar notation for parameter placeholders. The statement
/// is prepared at first execution and then only arguments for a query is sent
/// to the server.
///
/// A parameter can be of any type that is supported by the driver. See @ref
/// pg_types for more information.
///
/// @code
/// auto trx = cluster->Begin(/* transaction options */);
/// auto res = trx.Execute(
///     "select foo, bar from foobar where foo > $1 and bar = $2", 42, "baz");
/// trx.Commit();
/// @endcode
///
/// @see Transaction
/// @see ResultSet
///
/// ----------
///
/// @htmlonly <div class="bottom-nav"> @endhtmlonly
/// ⇦ @ref pg_transactions | @ref pg_process_results ⇨
/// @htmlonly </div> @endhtmlonly
 
// clang-format off
/// @brief PostgreSQL transaction.
///
/// RAII wrapper for running transactions on PostgreSQL connections. Should be
/// retrieved by calling storages::postgres::Cluster::Begin().
///
/// Non-copyable.
///
/// If the transaction is not explicitly finished (either committed or rolled back)
/// it will roll itself back in the destructor.
///
/// @par Usage synopsis
/// @code
/// auto trx = someCluster.Begin(/* transaction options */);
/// auto res = trx.Execute("select col1, col2 from schema.table");
/// DoSomething(res);
/// res = trx.Execute("update schema.table set col1 = $1 where col2 = $2", v1, v2);
/// // If in the above lines an exception is thrown, then the transaction is
/// // rolled back in the destructor of trx.
/// trx.Commit();
/// @endcode
// clang-format on
 
class Transaction {
public:
    //@{
    /** @name Shortcut transaction options constants */
    /// Read-write read committed transaction
    static constexpr TransactionOptions RW{};
    /// Read-only read committed transaction
    static constexpr TransactionOptions RO{TransactionOptions::kReadOnly};
    /// Read-only serializable deferrable transaction
    static constexpr TransactionOptions Deferrable{TransactionOptions::Deferrable()};
    //@}
 
    static constexpr std::size_t kDefaultRowsInChunk = 1024;
 
    /// @cond
    explicit Transaction(
        detail::ConnectionPtr&& conn,
        const TransactionOptions& = RW,
        OptionalCommandControl trx_cmd_ctl = {},
        detail::SteadyClock::time_point trx_start_time = detail::SteadyClock::now()
    );
 
    void SetName(std::string name);
    /// @endcond
 
    Transaction(Transaction&&) noexcept;
    Transaction& operator=(Transaction&&) noexcept;
 
    Transaction(const Transaction&) = delete;
    Transaction& operator=(const Transaction&) = delete;
 
    ~Transaction();
    /// @name Query execution
    /// @{
    /// Execute statement with arbitrary parameters.
    ///
    /// Suspends coroutine for execution.
    ///
    /// @snippet storages/postgres/tests/landing_test.cpp TransacExec
    template <typename... Args>
    ResultSet Execute(const Query& query, const Args&... args) {
        return Execute(OptionalCommandControl{}, query, args...);
    }
 
    /// Execute statement with arbitrary parameters and per-statement command
    /// control.
    ///
    /// Suspends coroutine for execution.
    ///
    /// @warning Do NOT create a query string manually by embedding arguments!
    /// It leads to vulnerabilities and bad performance. Either pass arguments
    /// separately, or use storages::postgres::ParameterScope.
    template <typename... Args>
    ResultSet Execute(OptionalCommandControl statement_cmd_ctl, const Query& query, const Args&... args) {
        detail::StaticQueryParameters<sizeof...(args)> params;
        params.Write(GetConnectionUserTypes(), args...);
        return DoExecute(query, detail::QueryParameters{params}, statement_cmd_ctl);
    }
 
    /// Execute statement with stored parameters.
    ///
    /// Suspends coroutine for execution.
    ///
    /// @warning Do NOT create a query string manually by embedding arguments!
    /// It leads to vulnerabilities and bad performance. Either pass arguments
    /// separately, or use storages::postgres::ParameterScope.
    ResultSet Execute(const Query& query, const ParameterStore& store) {
        return Execute(OptionalCommandControl{}, query, store);
    }
 
    /// Execute statement with stored parameters and per-statement command
    /// control.
    ///
    /// Suspends coroutine for execution.
    ///
    /// @warning Do NOT create a query string manually by embedding arguments!
    /// It leads to vulnerabilities and bad performance. Either pass arguments
    /// separately, or use storages::postgres::ParameterScope.
    ResultSet Execute(OptionalCommandControl statement_cmd_ctl, const Query& query, const ParameterStore& store);
 
    /// Execute statement that uses an array of arguments splitting that array in
    /// chunks and executing the statement with a chunk of arguments.
    ///
    /// Useful for statements that unnest their arguments to avoid the need to
    /// increase timeouts due to data amount growth.
    template <typename Container>
    void ExecuteBulk(const Query& query, const Container& args, std::size_t chunk_rows = kDefaultRowsInChunk);
 
    /// Execute statement that uses an array of arguments splitting that array in
    /// chunks and executing the statement with a chunk of arguments.
    ///
    /// Useful for statements that unnest their arguments to avoid the need to
    /// increase timeouts due to data amount growth.
    template <typename Container>
    void ExecuteBulk(
        OptionalCommandControl statement_cmd_ctl,
        const Query& query,
        const Container& args,
        std::size_t chunk_rows = kDefaultRowsInChunk
    );
 
    /// Execute statement that uses an array of arguments transforming that array
    /// into N arrays of corresponding fields and executing the statement
    /// with a chunk of each of these arrays values.
    /// Basically, a column-wise ExecuteBulk.
    ///
    /// Useful for statements that unnest their arguments to avoid the need to
    /// increase timeouts due to data amount growth, but providing an explicit
    /// mapping from `Container::value_type` to PG type is infeasible for some
    /// reason (otherwise, use ExecuteBulk).
    ///
    /// @snippet storages/postgres/tests/arrays_pgtest.cpp ExecuteDecomposeBulk
    template <typename Container>
    void ExecuteDecomposeBulk(const Query& query, const Container& args, std::size_t chunk_rows = kDefaultRowsInChunk);
 
    /// Execute statement that uses an array of arguments transforming that array
    /// into N arrays of corresponding fields and executing the statement
    /// with a chunk of each of these arrays values.
    /// Basically, a column-wise ExecuteBulk.
    ///
    /// Useful for statements that unnest their arguments to avoid the need to
    /// increase timeouts due to data amount growth, but providing an explicit
    /// mapping from `Container::value_type` to PG type is infeasible for some
    /// reason (otherwise, use ExecuteBulk).
    ///
    /// @snippet storages/postgres/tests/arrays_pgtest.cpp ExecuteDecomposeBulk
    template <typename Container>
    void ExecuteDecomposeBulk(
        OptionalCommandControl statement_cmd_ctl,
        const Query& query,
        const Container& args,
        std::size_t chunk_rows = kDefaultRowsInChunk
    );
 
    /// Create a portal for fetching results of a statement with arbitrary
    /// parameters.
    template <typename... Args>
    Portal MakePortal(const Query& query, const Args&... args) {
        return MakePortal(OptionalCommandControl{}, query, args...);
    }
 
    /// Create a portal for fetching results of a statement with arbitrary
    /// parameters and per-statement command control.
    template <typename... Args>
    Portal MakePortal(OptionalCommandControl statement_cmd_ctl, const Query& query, const Args&... args) {
        detail::StaticQueryParameters<sizeof...(args)> params;
        params.Write(GetConnectionUserTypes(), args...);
        return MakePortal(PortalName{}, query, detail::QueryParameters{params}, statement_cmd_ctl);
    }
 
    /// Create a portal for fetching results of a statement with stored
    /// parameters.
    Portal MakePortal(const Query& query, const ParameterStore& store) {
        return MakePortal(OptionalCommandControl{}, query, store);
    }
 
    /// Create a portal for fetching results of a statement with stored parameters
    /// and per-statement command control.
    Portal MakePortal(OptionalCommandControl statement_cmd_ctl, const Query& query, const ParameterStore& store);
 
    /// Set a connection parameter
    /// https://www.postgresql.org/docs/current/sql-set.html
    /// The parameter is set for this transaction only
    void SetParameter(const std::string& param_name, const std::string& value);
    //@}
 
    /// Commit the transaction
    /// Suspends coroutine until command complete.
    /// After Commit or Rollback is called, the transaction is not usable any
    /// more.
    void Commit();
    /// Rollback the transaction
    /// Suspends coroutine until command complete.
    /// After Commit or Rollback is called, the transaction is not usable any
    /// more.
    void Rollback();
 
    /// Used in tests
    OptionalCommandControl GetConnTransactionCommandControlDebug() const;
    TimeoutDuration GetConnStatementTimeoutDebug() const;
 
private:
    ResultSet
    DoExecute(const Query& query, const detail::QueryParameters& params, OptionalCommandControl statement_cmd_ctl);
    Portal MakePortal(
        const PortalName&,
        const Query& query,
        const detail::QueryParameters& params,
        OptionalCommandControl statement_cmd_ctl
    );
 
    const UserTypes& GetConnectionUserTypes() const;
 
    std::string name_;
    detail::ConnectionPtr conn_;
};
 
template <typename Container>
void Transaction::ExecuteBulk(const Query& query, const Container& args, std::size_t chunk_rows) {
    auto split = io::SplitContainer(args, chunk_rows);
    for (auto&& chunk : split) {
        Execute(query, chunk);
    }
}
 
template <typename Container>
void Transaction::ExecuteBulk(
    OptionalCommandControl statement_cmd_ctl,
    const Query& query,
    const Container& args,
    std::size_t chunk_rows
) {
    auto split = io::SplitContainer(args, chunk_rows);
    for (auto&& chunk : split) {
        Execute(statement_cmd_ctl, query, chunk);
    }
}
 
template <typename Container>
void Transaction::ExecuteDecomposeBulk(const Query& query, const Container& args, std::size_t chunk_rows) {
    io::SplitContainerByColumns(args, chunk_rows).Perform([&query, this](const auto&... args) {
        this->Execute(query, args...);
    });
}
 
template <typename Container>
void Transaction::ExecuteDecomposeBulk(
    OptionalCommandControl statement_cmd_ctl,
    const Query& query,
    const Container& args,
    std::size_t chunk_rows
) {
    io::SplitContainerByColumns(args, chunk_rows).Perform([&query, &statement_cmd_ctl, this](const auto&... args) {
        this->Execute(statement_cmd_ctl, query, args...);
    });
}
 
}  // namespace storages::postgres
 
USERVER_NAMESPACE_END