userver: userver/storages/postgres/transaction.hpp Source File
Loading...
Searching...
No Matches
transaction.hpp
Go to the documentation of this file.
1#pragma once
2
3/// @file userver/storages/postgres/transaction.hpp
4/// @brief Transactions
5
6#include <memory>
7#include <string>
8
9#include <userver/storages/postgres/detail/connection_ptr.hpp>
10#include <userver/storages/postgres/detail/query_parameters.hpp>
11#include <userver/storages/postgres/detail/time_types.hpp>
12#include <userver/storages/postgres/options.hpp>
13#include <userver/storages/postgres/parameter_store.hpp>
14#include <userver/storages/postgres/portal.hpp>
15#include <userver/storages/postgres/postgres_fwd.hpp>
16#include <userver/storages/postgres/query.hpp>
17#include <userver/storages/postgres/result_set.hpp>
18
19USERVER_NAMESPACE_BEGIN
20
21namespace storages::postgres {
22
23/// @page pg_transactions uPg: Transactions
24///
25/// All queries that are run on a PostgreSQL cluster are executed inside
26/// a transaction, even if a single-query interface is used.
27///
28/// A uPg transaction can be started using all isolation levels and modes
29/// supported by PostgreSQL server as specified in documentation here
30/// https://www.postgresql.org/docs/current/static/sql-set-transaction.html.
31/// When starting a transaction, the options are specified using
32/// TransactionOptions structure.
33///
34/// For convenience and improvement of readability there are constants
35/// defined: Transaction::RW, Transaction::RO and Transaction::Deferrable.
36///
37/// @see TransactionOptions
38///
39/// Transaction object ensures that a transaction started in a PostgreSQL
40/// connection will be either committed or rolled back and the connection
41/// will returned back to a connection pool.
42///
43/// @todo Code snippet with transaction starting and committing
44///
45/// Next: @ref pg_run_queries
46///
47/// See also: @ref pg_process_results
48///
49/// ----------
50///
51/// @htmlonly <div class="bottom-nav"> @endhtmlonly
52/// ⇦ @ref pg_driver | @ref pg_run_queries ⇨
53/// @htmlonly </div> @endhtmlonly
54
55/// @page pg_run_queries uPg: Running queries
56///
57/// All queries are executed through a transaction object, event when being
58/// executed through singe-query interface, so here only executing queries
59/// with transaction will be covered. Single-query interface is basically
60/// the same except for additional options.
61///
62/// uPg provides means to execute text queries only. There is no query
63/// generation, but can be used by other tools to execute SQL queries.
64///
65/// @warning A query must contain a single query, multiple statements delimited
66/// by ';' are not supported.
67///
68/// All queries are parsed and prepared during the first invocation and are
69/// executed as prepared statements afterwards.
70///
71/// Any query execution can throw an exception. Please see @ref pg_errors for
72/// more information on possible errors.
73///
74/// @par Queries without parameters
75///
76/// Executing a query without any parameters is rather straightforward.
77/// @code
78/// auto trx = cluster->Begin(/* transaction options */);
79/// auto res = trx.Execute("select foo, bar from foobar");
80/// trx.Commit();
81/// @endcode
82///
83/// The cluster also provides interface for single queries
84/// @code
85/// auto res = cluster->Execute(/* transaction options */, /* the statement */);
86/// @endcode
87///
88/// @par Queries with parameters
89///
90/// uPg supports SQL dollar notation for parameter placeholders. The statement
91/// is prepared at first execution and then only arguments for a query is sent
92/// to the server.
93///
94/// A parameter can be of any type that is supported by the driver. See @ref
95/// pg_types for more information.
96///
97/// @code
98/// auto trx = cluster->Begin(/* transaction options */);
99/// auto res = trx.Execute(
100/// "select foo, bar from foobar where foo > $1 and bar = $2", 42, "baz");
101/// trx.Commit();
102/// @endcode
103///
104/// @see Transaction
105/// @see ResultSet
106///
107/// ----------
108///
109/// @htmlonly <div class="bottom-nav"> @endhtmlonly
110/// ⇦ @ref pg_transactions | @ref pg_process_results ⇨
111/// @htmlonly </div> @endhtmlonly
112
113// clang-format off
114/// @brief PostgreSQL transaction.
115///
116/// RAII wrapper for running transactions on PostgreSQL connections. Should be
117/// retrieved by calling storages::postgres::Cluster::Begin().
118///
119/// Non-copyable.
120///
121/// If the transaction is not explicitly finished (either committed or rolled back)
122/// it will roll itself back in the destructor.
123///
124/// @par Usage synopsis
125/// @code
126/// auto trx = someCluster.Begin(/* transaction options */);
127/// auto res = trx.Execute("select col1, col2 from schema.table");
128/// DoSomething(res);
129/// res = trx.Execute("update schema.table set col1 = $1 where col2 = $2", v1, v2);
130/// // If in the above lines an exception is thrown, then the transaction is
131/// // rolled back in the destructor of trx.
132/// trx.Commit();
133/// @endcode
134// clang-format on
135
137 public:
138 //@{
139 /** @name Shortcut transaction options constants */
140 /// Read-write read committed transaction
141 static constexpr TransactionOptions RW{};
142 /// Read-only read committed transaction
143 static constexpr TransactionOptions RO{TransactionOptions::kReadOnly};
144 /// Read-only serializable deferrable transaction
147 //@}
148
149 static constexpr std::size_t kDefaultRowsInChunk = 1024;
150
151 /// @cond
152 explicit Transaction(detail::ConnectionPtr&& conn,
153 const TransactionOptions& = RW,
154 OptionalCommandControl trx_cmd_ctl = {},
155 detail::SteadyClock::time_point trx_start_time =
156 detail::SteadyClock::now());
157
158 void SetName(std::string name);
159 /// @endcond
160
161 Transaction(Transaction&&) noexcept;
162 Transaction& operator=(Transaction&&) noexcept;
163
164 Transaction(const Transaction&) = delete;
165 Transaction& operator=(const Transaction&) = delete;
166
167 ~Transaction();
168 /// @name Query execution
169 /// @{
170 /// Execute statement with arbitrary parameters.
171 ///
172 /// Suspends coroutine for execution.
173 ///
174 /// @snippet storages/postgres/tests/landing_test.cpp TransacExec
175 template <typename... Args>
176 ResultSet Execute(const Query& query, const Args&... args) {
177 return Execute(OptionalCommandControl{}, query, args...);
178 }
179
180 /// Execute statement with arbitrary parameters and per-statement command
181 /// control.
182 ///
183 /// Suspends coroutine for execution.
184 ///
185 /// @warning Do NOT create a query string manually by embedding arguments!
186 /// It leads to vulnerabilities and bad performance. Either pass arguments
187 /// separately, or use storages::postgres::ParameterScope.
188 template <typename... Args>
189 ResultSet Execute(OptionalCommandControl statement_cmd_ctl,
190 const Query& query, const Args&... args) {
191 detail::StaticQueryParameters<sizeof...(args)> params;
192 params.Write(GetConnectionUserTypes(), args...);
193 return DoExecute(query, detail::QueryParameters{params}, statement_cmd_ctl);
194 }
195
196 /// Execute statement with stored parameters.
197 ///
198 /// Suspends coroutine for execution.
199 ///
200 /// @warning Do NOT create a query string manually by embedding arguments!
201 /// It leads to vulnerabilities and bad performance. Either pass arguments
202 /// separately, or use storages::postgres::ParameterScope.
203 ResultSet Execute(const Query& query, const ParameterStore& store) {
204 return Execute(OptionalCommandControl{}, query, store);
205 }
206
207 /// Execute statement with stored parameters and per-statement command
208 /// control.
209 ///
210 /// Suspends coroutine for execution.
211 ResultSet Execute(OptionalCommandControl statement_cmd_ctl,
212 const Query& query, const ParameterStore& store);
213
214 /// Execute statement that uses an array of arguments splitting that array in
215 /// chunks and executing the statement with a chunk of arguments.
216 ///
217 /// Useful for statements that unnest their arguments to avoid the need to
218 /// increase timeouts due to data amount growth.
219 template <typename Container>
220 void ExecuteBulk(const Query& query, const Container& args,
221 std::size_t chunk_rows = kDefaultRowsInChunk);
222
223 /// Execute statement that uses an array of arguments splitting that array in
224 /// chunks and executing the statement with a chunk of arguments.
225 ///
226 /// Useful for statements that unnest their arguments to avoid the need to
227 /// increase timeouts due to data amount growth.
228 template <typename Container>
229 void ExecuteBulk(OptionalCommandControl statement_cmd_ctl, const Query& query,
230 const Container& args,
231 std::size_t chunk_rows = kDefaultRowsInChunk);
232
233 /// Execute statement that uses an array of arguments transforming that array
234 /// into N arrays of corresponding fields and executing the statement
235 /// with a chunk of each of these arrays values.
236 /// Basically, a column-wise ExecuteBulk.
237 ///
238 /// Useful for statements that unnest their arguments to avoid the need to
239 /// increase timeouts due to data amount growth, but providing an explicit
240 /// mapping from `Container::value_type` to PG type is infeasible for some
241 /// reason (otherwise, use ExecuteBulk).
242 ///
243 /// @snippet storages/postgres/tests/arrays_pgtest.cpp ExecuteDecomposeBulk
244 template <typename Container>
245 void ExecuteDecomposeBulk(const Query& query, const Container& args,
246 std::size_t chunk_rows = kDefaultRowsInChunk);
247
248 /// Execute statement that uses an array of arguments transforming that array
249 /// into N arrays of corresponding fields and executing the statement
250 /// with a chunk of each of these arrays values.
251 /// Basically, a column-wise ExecuteBulk.
252 ///
253 /// Useful for statements that unnest their arguments to avoid the need to
254 /// increase timeouts due to data amount growth, but providing an explicit
255 /// mapping from `Container::value_type` to PG type is infeasible for some
256 /// reason (otherwise, use ExecuteBulk).
257 ///
258 /// @snippet storages/postgres/tests/arrays_pgtest.cpp ExecuteDecomposeBulk
259 template <typename Container>
260 void ExecuteDecomposeBulk(OptionalCommandControl statement_cmd_ctl,
261 const Query& query, const Container& args,
262 std::size_t chunk_rows = kDefaultRowsInChunk);
263
264 /// Create a portal for fetching results of a statement with arbitrary
265 /// parameters.
266 template <typename... Args>
267 Portal MakePortal(const Query& query, const Args&... args) {
268 return MakePortal(OptionalCommandControl{}, query, args...);
269 }
270
271 /// Create a portal for fetching results of a statement with arbitrary
272 /// parameters and per-statement command control.
273 template <typename... Args>
274 Portal MakePortal(OptionalCommandControl statement_cmd_ctl,
275 const Query& query, const Args&... args) {
276 detail::StaticQueryParameters<sizeof...(args)> params;
277 params.Write(GetConnectionUserTypes(), args...);
278 return MakePortal(PortalName{}, query, detail::QueryParameters{params},
279 statement_cmd_ctl);
280 }
281
282 /// Create a portal for fetching results of a statement with stored
283 /// parameters.
284 Portal MakePortal(const Query& query, const ParameterStore& store) {
285 return MakePortal(OptionalCommandControl{}, query, store);
286 }
287
288 /// Create a portal for fetching results of a statement with stored parameters
289 /// and per-statement command control.
290 Portal MakePortal(OptionalCommandControl statement_cmd_ctl,
291 const Query& query, const ParameterStore& store);
292
293 /// Set a connection parameter
294 /// https://www.postgresql.org/docs/current/sql-set.html
295 /// The parameter is set for this transaction only
296 void SetParameter(const std::string& param_name, const std::string& value);
297 //@}
298
299 /// Commit the transaction
300 /// Suspends coroutine until command complete.
301 /// After Commit or Rollback is called, the transaction is not usable any
302 /// more.
303 void Commit();
304 /// Rollback the transaction
305 /// Suspends coroutine until command complete.
306 /// After Commit or Rollback is called, the transaction is not usable any
307 /// more.
308 void Rollback();
309
310 /// Used in tests
311 OptionalCommandControl GetConnTransactionCommandControlDebug() const;
312 TimeoutDuration GetConnStatementTimeoutDebug() const;
313
314 private:
315 ResultSet DoExecute(const Query& query, const detail::QueryParameters& params,
316 OptionalCommandControl statement_cmd_ctl);
317 Portal MakePortal(const PortalName&, const Query& query,
318 const detail::QueryParameters& params,
319 OptionalCommandControl statement_cmd_ctl);
320
321 const UserTypes& GetConnectionUserTypes() const;
322
323 std::string name_;
324 detail::ConnectionPtr conn_;
325};
326
327template <typename Container>
328void Transaction::ExecuteBulk(const Query& query, const Container& args,
329 std::size_t chunk_rows) {
330 auto split = io::SplitContainer(args, chunk_rows);
331 for (auto&& chunk : split) {
332 Execute(query, chunk);
333 }
334}
335
336template <typename Container>
337void Transaction::ExecuteBulk(OptionalCommandControl statement_cmd_ctl,
338 const Query& query, const Container& args,
339 std::size_t chunk_rows) {
340 auto split = io::SplitContainer(args, chunk_rows);
341 for (auto&& chunk : split) {
342 Execute(statement_cmd_ctl, query, chunk);
343 }
344}
345
346template <typename Container>
347void Transaction::ExecuteDecomposeBulk(const Query& query,
348 const Container& args,
349 std::size_t chunk_rows) {
350 io::SplitContainerByColumns(args, chunk_rows)
351 .Perform([&query, this](const auto&... args) {
352 this->Execute(query, args...);
353 });
354}
355
356template <typename Container>
357void Transaction::ExecuteDecomposeBulk(OptionalCommandControl statement_cmd_ctl,
358 const Query& query,
359 const Container& args,
360 std::size_t chunk_rows) {
361 io::SplitContainerByColumns(args, chunk_rows)
362 .Perform([&query, &statement_cmd_ctl, this](const auto&... args) {
363 this->Execute(statement_cmd_ctl, query, args...);
364 });
365}
366
367} // namespace storages::postgres
368
369USERVER_NAMESPACE_END