query_queue.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/storages/postgres/query_queue.hpp
/// @brief An utility to execute multiple queries in a single network
/// round-trip.
 
#include <userver/storages/postgres/options.hpp>
#include <userver/storages/postgres/query.hpp>
#include <userver/storages/postgres/result_set.hpp>
 
#include <userver/storages/postgres/detail/connection_ptr.hpp>
#include <userver/storages/postgres/detail/query_parameters.hpp>
 
#include <userver/utils/any_movable.hpp>
#include <userver/utils/fast_pimpl.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace storages::postgres {
 
/// @brief A container to enqueue queries in FIFO order and execute them all
/// within a single network round-trip.
///
/// Acquired from storages::postgres::Cluster, one is expected to `Push`
/// some queries into the queue and then `Collect` them into vector of results.
///
/// From a client point of view `Collect` is transactional: either all the
/// queries succeed or `Collect` rethrows the first error encountered. However,
/// this is *NOT* the case for the server: server treats all the provided
/// queries independently and is likely to  execute subsequent queries even
/// after prior failures.
///
/// @warning If an explicit transaction ("BEGIN") or a modifying query is added
/// into the queue the behavior is unspecified. *Don't do that*.
///
/// @note Queries may or may not be sent to the server prior to `Collect` call.
///
/// @note Requires both pipelining and prepared statements to be enabled in the
/// driver, UINVARIANTS (that is, throws in release builds and aborts in debug
/// ones) that on construction.
class QueryQueue final {
public:
    QueryQueue(CommandControl default_cc, detail::ConnectionPtr&& conn);
 
    QueryQueue(QueryQueue&&) noexcept;
    QueryQueue& operator=(QueryQueue&&) noexcept;
 
    QueryQueue(const QueryQueue&) = delete;
    QueryQueue& operator=(const QueryQueue&) = delete;
 
    ~QueryQueue();
 
    /// Reserve internal storage to hold this amount of queries.
    void Reserve(std::size_t size);
 
    /// Add a query into the queue with specified command-control.
    /// CommandControl is used as following: 'execute' is used as a timeout for
    /// preparing the statement (if needed), 'statement' is used as a statement
    /// timeout for later execution.
    template <typename... Args>
    void Push(CommandControl cc, const Query& query, const Args&... args);
 
    /// Add a query into the queue with default command-control.
    template <typename... Args>
    void Push(const Query& query, const Args&... args);
 
    /// Collect results of all the queued queries, with specified timeout.
    /// Either returns a vector of N `ResultSet`s, where N is the number of
    /// queries enqueued, or rethrow the first error encountered, be that a query
    /// execution error or a timeout.
    [[nodiscard]] std::vector<ResultSet> Collect(TimeoutDuration timeout);
 
    /// Collect results of all the queued queries, with default timeout.
    /// Either returns a vector of N `ResultSet`s, where N is the number of
    /// queries enqueued, or rethrow the first error encountered, be that a query
    /// execution error or a timeout.
    [[nodiscard]] std::vector<ResultSet> Collect();
 
private:
    struct ParamsHolder final {
        // We only need to know what's here at construction (and at construction we
        // do know), after that we just need to preserve lifetime.
        USERVER_NAMESPACE::utils::AnyMovable any_params{};
        // This here knows what's there actually is in params.
        detail::QueryParameters params_proxy;
    };
 
    const UserTypes& GetConnectionUserTypes() const;
 
    void DoPush(CommandControl cc, const Query& query, ParamsHolder&& params);
 
    void ValidateUsage() const;
 
    CommandControl default_cc_;
    detail::ConnectionPtr conn_;
 
    struct QueriesStorage;
    USERVER_NAMESPACE::utils::FastPimpl<QueriesStorage, 48, 8> queries_storage_;
};
 
template <typename... Args>
void QueryQueue::Push(CommandControl cc, const Query& query, const Args&... args) {
    ParamsHolder holder{};
    auto& params = holder.any_params.Emplace<detail::StaticQueryParameters<sizeof...(args)>>();
    params.Write(GetConnectionUserTypes(), args...);
    holder.params_proxy = detail::QueryParameters{params};
    DoPush(cc, query, std::move(holder));
}
 
template <typename... Args>
void QueryQueue::Push(const Query& query, const Args&... args) {
    Push(default_cc_, query, args...);
}
 
}  // namespace storages::postgres
 
USERVER_NAMESPACE_END