Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_text/tests/annealer_corpus/diesel_pg.rs

57.7 KiB, 1 run

created by r1870400018:11726, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1pub(super) mod copy;
2pub(crate) mod cursor;
3mod raw;
4mod result;
5mod row;
6mod stmt;
7
8use self::copy::{CopyFromSink, CopyToBuffer};
9use self::cursor::*;
10use self::private::{ConnectionAndTransactionManager, CopyFromWrapper, QueryFragmentHelper};
11use self::raw::{PgTransactionStatus, RawConnection};
12use self::stmt::Statement;
13use crate::RunQueryDsl;
14use crate::connection::instrumentation::{DynInstrumentation, Instrumentation, StrQueryHelper};
15use crate::connection::statement_cache::{MaybeCached, StatementCache};
16use crate::connection::*;
17use crate::expression::QueryMetadata;
18use crate::pg::backend::PgNotification;
19use crate::pg::metadata_lookup::{GetPgMetadataCache, PgMetadataCache};
20use crate::pg::query_builder::copy::InternalCopyFromQuery;
21use crate::pg::{Pg, TransactionBuilder};
22use crate::query_builder::bind_collector::RawBytesBindCollector;
23use crate::query_builder::*;
24use crate::result::ConnectionError::CouldntSetupConfiguration;
25use crate::result::*;
26use alloc::ffi::CString;
27use core::ffi as libc;
28use core::fmt::Debug;
29
30use super::query_builder::copy::{CopyFromExpression, CopyTarget, CopyToCommand};
31
32pub(super) use self::result::PgResult;
33
34/// The connection string expected by `PgConnection::establish`
35/// should be a PostgreSQL connection string, as documented at
36/// <https://www.postgresql.org/docs/9.4/static/libpq-connect.html#LIBPQ-CONNSTRING>
37///
38/// # Supported loading model implementations
39///
40/// * [`DefaultLoadingMode`]
41/// * [`PgRowByRowLoadingMode`]
42///
43/// If you are unsure which loading mode is the correct one for your application,
44/// you likely want to use the `DefaultLoadingMode` as it's simpler and more intuitive.
45/// However, if you plan to process each row on its own, you should use the `PgRowByRowLoadingMode` and
46/// you can also expect a performance boost.
47///
48/// Due to the fact that `PgConnection` supports multiple loading modes
49/// it is **required** to always specify the used loading mode
50/// when calling [`RunQueryDsl::load_iter`]
51///
52/// ## `DefaultLoadingMode`
53///
54/// By using this mode `PgConnection` defaults to loading all response values at **once**
55/// and only performs deserialization afterward for the `DefaultLoadingMode`.
56///
57/// This loading mode allows users to perform hold more than one iterator at once using
58/// the same connection:
59/// ```rust
60/// # include!("../../doctest_setup.rs");
61/// #
62/// # fn main() {
63/// # run_test().unwrap();
64/// # }
65/// #
66/// # fn run_test() -> QueryResult<()> {
67/// # use schema::users;
68/// # let connection = &mut establish_connection();
69/// use diesel::connection::DefaultLoadingMode;
70///
71/// let iter1 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
72/// let iter2 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
73///
74/// for r in iter1 {
75/// let (id, name) = r?;
76/// println!("Id: {} Name: {}", id, name);
77/// }
78///
79/// for r in iter2 {
80/// let (id, name) = r?;
81/// println!("Id: {} Name: {}", id, name);
82/// }
83/// # Ok(())
84/// # }
85/// ```
86///
87/// ## `PgRowByRowLoadingMode`
88///
89/// By using this mode `PgConnection` defaults to loading each row of the result set
90/// separately. This might be desired for huge result sets.
91///
92/// This loading mode **prevents** creating more than one iterator at once using
93/// the same connection. The following code is **not** allowed:
94///
95/// ```compile_fail
96/// # include!("../../doctest_setup.rs");
97/// #
98/// # fn main() {
99/// # run_test().unwrap();
100/// # }
101/// #
102/// # fn run_test() -> QueryResult<()> {
103/// # use schema::users;
104/// # let connection = &mut establish_connection();
105/// use diesel::pg::PgRowByRowLoadingMode;
106///
107/// let iter1 = users::table.load_iter::<(i32, String), PgRowByRowLoadingMode>(connection)?;
108/// // creating a second iterator generates an compiler error
109/// let iter2 = users::table.load_iter::<(i32, String), PgRowByRowLoadingMode>(connection)?;
110///
111/// for r in iter1 {
112/// let (id, name) = r?;
113/// println!("Id: {} Name: {}", id, name);
114/// }
115///
116/// for r in iter2 {
117/// let (id, name) = r?;
118/// println!("Id: {} Name: {}", id, name);
119/// }
120/// # Ok(())
121/// # }
122/// ```
123#[allow(missing_debug_implementations)]
124#[cfg(feature = "postgres")]
125pub struct PgConnection {
126 statement_cache: StatementCache<Pg, Statement>,
127 metadata_cache: PgMetadataCache,
128 connection_and_transaction_manager: ConnectionAndTransactionManager,
129}
130
131// according to libpq documentation a connection can be transferred to other threads
132#[allow(unsafe_code)]
133unsafe impl Send for PgConnection {}
134
135impl SimpleConnection for PgConnection {
136 #[allow(unsafe_code)] // use of unsafe function
137 fn batch_execute(&mut self, query: &str) -> QueryResult<()> {
138 self.connection_and_transaction_manager
139 .instrumentation
140 .on_connection_event(InstrumentationEvent::StartQuery {
141 query: &StrQueryHelper::new(query),
142 });
143 let c_query = CString::new(query)?;
144 let inner_result = unsafe {
145 self.connection_and_transaction_manager
146 .raw_connection
147 .exec(c_query.as_ptr())
148 };
149 update_transaction_manager_status(
150 inner_result.and_then(|raw_result| {
151 PgResult::new(
152 raw_result,
153 &self.connection_and_transaction_manager.raw_connection,
154 )
155 }),
156 &mut self.connection_and_transaction_manager,
157 &|callback| callback(&StrQueryHelper::new(query)),
158 true,
159 )?;
160 Ok(())
161 }
162}
163
164/// A [`PgConnection`] specific loading mode to load rows one by one
165///
166/// See the documentation of [`PgConnection`] for details
167#[derive(Debug, Copy, Clone)]
168pub struct PgRowByRowLoadingMode;
169
170impl ConnectionSealed for PgConnection {}
171
172impl Connection for PgConnection {
173 type Backend = Pg;
174 type TransactionManager = AnsiTransactionManager;
175
176 fn establish(database_url: &str) -> ConnectionResult<PgConnection> {
177 let mut instrumentation = DynInstrumentation::default_instrumentation();
178 instrumentation.on_connection_event(InstrumentationEvent::StartEstablishConnection {
179 url: database_url,
180 });
181 let r = RawConnection::establish(database_url).and_then(|raw_conn| {
182 let mut conn = PgConnection {
183 connection_and_transaction_manager: ConnectionAndTransactionManager {
184 raw_connection: raw_conn,
185 transaction_state: AnsiTransactionManager::default(),
186 instrumentation: DynInstrumentation::none(),
187 },
188 statement_cache: StatementCache::new(),
189 metadata_cache: PgMetadataCache::new(),
190 };
191 conn.set_config_options()
192 .map_err(CouldntSetupConfiguration)?;
193 Ok(conn)
194 });
195 instrumentation.on_connection_event(InstrumentationEvent::FinishEstablishConnection {
196 url: database_url,
197 error: r.as_ref().err(),
198 });
199 let mut conn = r?;
200 conn.connection_and_transaction_manager.instrumentation = instrumentation;
201 Ok(conn)
202 }
203
204 fn execute_returning_count<T>(&mut self, source: &T) -> QueryResult<usize>
205 where
206 T: QueryFragment<Pg> + QueryId,
207 {
208 update_transaction_manager_status(
209 self.with_prepared_query(
210 Box::new(source),
211 true,
212 &mut |query, params, conn, _source| {
213 let res = query
214 .execute(&mut conn.raw_connection, &params, false)
215 .and_then(|r| r.rows_affected());
216 // according to https://www.postgresql.org/docs/current/libpq-async.html
217 // `PQgetResult` needs to be called till a null pointer is returned
218 while conn.raw_connection.get_next_result()?.is_some() {}
219 res
220 },
221 ),
222 &mut self.connection_and_transaction_manager,
223 &|callback| source.instrumentation(callback),
224 true,
225 )
226 }
227
228 fn transaction_state(&mut self) -> &mut AnsiTransactionManager
229 where
230 Self: Sized,
231 {
232 &mut self.connection_and_transaction_manager.transaction_state
233 }
234
235 fn instrumentation(&mut self) -> &mut dyn Instrumentation {
236 &mut *self.connection_and_transaction_manager.instrumentation
237 }
238
239 fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
240 self.connection_and_transaction_manager.instrumentation = instrumentation.into();
241 }
242
243 fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
244 self.statement_cache.set_cache_size(size);
245 }
246}
247
248impl<B> LoadConnection<B> for PgConnection
249where
250 Self: self::private::PgLoadingMode<B>,
251{
252 type Cursor<'conn, 'query> = <Self as self::private::PgLoadingMode<B>>::Cursor<'conn, 'query>;
253 type Row<'conn, 'query> = <Self as self::private::PgLoadingMode<B>>::Row<'conn, 'query>;
254
255 fn load<'conn, 'query, T>(
256 &'conn mut self,
257 source: T,
258 ) -> QueryResult<Self::Cursor<'conn, 'query>>
259 where
260 T: Query + QueryFragment<Self::Backend> + QueryId + 'query,
261 Self::Backend: QueryMetadata<T::SqlType>,
262 {
263 self.with_prepared_query(
264 Box::new(source),
265 false,
266 &mut |stmt, params, conn, source| {
267 use self::private::PgLoadingMode;
268 let result = inner_load(stmt, params, conn, &*source, Self::USE_ROW_BY_ROW_MODE)?;
269 Self::get_cursor(conn, result, source)
270 },
271 )
272 }
273}
274
275fn inner_load(
276 stmt: MaybeCached<'_, Statement>,
277 params: Vec<Option<Vec<u8>>>,
278 conn: &mut ConnectionAndTransactionManager,
279 source: &dyn QueryFragmentHelper<crate::result::Error>,
280 row_by_row: bool,
281) -> Result<PgResult, Error> {
282 let result = stmt.execute(&mut conn.raw_connection, &params, row_by_row);
283 update_transaction_manager_status(
284 result,
285 conn,
286 &|callback| source.instrumentation(callback),
287 false,
288 )
289}
290
291impl GetPgMetadataCache for PgConnection {
292 fn get_metadata_cache(&mut self) -> &mut PgMetadataCache {
293 &mut self.metadata_cache
294 }
295}
296
297#[inline(always)]
298fn update_transaction_manager_status<T>(
299 query_result: QueryResult<T>,
300 conn: &mut ConnectionAndTransactionManager,
301 instrumentation_callback: &dyn Fn(
302 &mut dyn FnMut(&dyn crate::connection::instrumentation::DebugQuery),
303 ),
304 final_call: bool,
305) -> QueryResult<T> {
306 /// avoid monomorphizing for every result type - this part will not be inlined
307 fn non_generic_inner(conn: &mut ConnectionAndTransactionManager, is_err: bool) {
308 let raw_conn: &mut RawConnection = &mut conn.raw_connection;
309 let tm: &mut AnsiTransactionManager = &mut conn.transaction_state;
310 // libpq keeps track of the transaction status internally, and that is accessible
311 // via `transaction_status`. We can use that to update the AnsiTransactionManager
312 // status
313 match raw_conn.transaction_status() {
314 PgTransactionStatus::InError => {
315 tm.status.set_requires_rollback_maybe_up_to_top_level(true)
316 }
317 PgTransactionStatus::Unknown => tm.status.set_in_error(),
318 PgTransactionStatus::Idle => {
319 // This is useful in particular for commit attempts (even
320 // if `COMMIT` errors it still exits transaction)
321
322 // This may repair the transaction manager
323 tm.status = TransactionManagerStatus::Valid(Default::default())
324 }
325 PgTransactionStatus::InTransaction => {
326 let transaction_status = &mut tm.status;
327 // If we weren't an error, it is possible that we were a transaction start
328 // -> we should tolerate any state
329 if is_err {
330 // An error may not have us enter a transaction, so if we weren't in one
331 // we may not be in one now
332 if !matches!(transaction_status, TransactionManagerStatus::Valid(valid_tm) if valid_tm.transaction_depth().is_some())
333 {
334 // -> transaction manager is broken
335 transaction_status.set_in_error()
336 }
337 } else {
338 // If transaction was InError before, but now it's not (because we attempted
339 // a rollback), we may pretend it's fixed because
340 // if it isn't Postgres *will* tell us again.
341
342 // Fun fact: if we have not received an `InTransaction` status however,
343 // postgres will *not* warn us that transaction is broken when attempting to
344 // commit, so we may think that commit has succeeded but in fact it hasn't.
345 tm.status.set_requires_rollback_maybe_up_to_top_level(false)
346 }
347 }
348 PgTransactionStatus::Active => {
349 // This is a transient state for libpq - nothing we can deduce here.
350 }
351 }
352 }
353
354 fn non_generic_instrumentation(
355 query_result: Result<(), &Error>,
356 conn: &mut ConnectionAndTransactionManager,
357 instrumentation_callback: &dyn Fn(
358 &mut dyn FnMut(&dyn crate::connection::instrumentation::DebugQuery),
359 ),
360 final_call: bool,
361 ) {
362 if let Err(e) = query_result {
363 instrumentation_callback(&mut |query| {
364 conn.instrumentation
365 .on_connection_event(InstrumentationEvent::FinishQuery {
366 query,
367 error: Some(e),
368 })
369 });
370 } else if final_call {
371 instrumentation_callback(&mut |query| {
372 conn.instrumentation
373 .on_connection_event(InstrumentationEvent::FinishQuery { query, error: None });
374 });
375 }
376 }
377
378 non_generic_inner(conn, query_result.is_err());
379 non_generic_instrumentation(
380 query_result.as_ref().map(|_| ()),
381 conn,
382 instrumentation_callback,
383 final_call,
384 );
385 query_result
386}
387
388#[cfg(feature = "r2d2")]
389impl crate::r2d2::R2D2Connection for PgConnection {
390 fn ping(&mut self) -> QueryResult<()> {
391 crate::r2d2::CheckConnectionQuery.execute(self).map(|_| ())
392 }
393
394 fn is_broken(&mut self) -> bool {
395 AnsiTransactionManager::is_broken_transaction_manager(self)
396 }
397}
398
399impl MultiConnectionHelper for PgConnection {
400 fn to_any<'a>(
401 lookup: &mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup,
402 ) -> &mut (dyn core::any::Any + 'a) {
403 lookup.as_any()
404 }
405
406 fn from_any(
407 lookup: &mut dyn core::any::Any,
408 ) -> Option<&mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup> {
409 lookup
410 .downcast_mut::<Self>()
411 .map(|conn| conn as &mut dyn super::PgMetadataLookup)
412 }
413}
414
415impl PgConnection {
416 /// Build a transaction, specifying additional details such as isolation level
417 ///
418 /// See [`TransactionBuilder`] for more examples.
419 ///
420 /// [`TransactionBuilder`]: crate::pg::TransactionBuilder
421 ///
422 /// ```rust
423 /// # include!("../../doctest_setup.rs");
424 /// #
425 /// # fn main() {
426 /// # run_test().unwrap();
427 /// # }
428 /// #
429 /// # fn run_test() -> QueryResult<()> {
430 /// # use schema::users::dsl::*;
431 /// # let conn = &mut connection_no_transaction();
432 /// conn.build_transaction()
433 /// .read_only()
434 /// .serializable()
435 /// .deferrable()
436 /// .run(|conn| Ok(()))
437 /// # }
438 /// ```
439 pub fn build_transaction(&mut self) -> TransactionBuilder<'_, Self> {
440 TransactionBuilder::new(self)
441 }
442
443 pub(crate) fn copy_from<S, T>(&mut self, target: S) -> Result<usize, S::Error>
444 where
445 S: CopyFromExpression<T>,
446 {
447 let query = CopyFromWrapper(core::cell::RefCell::new(InternalCopyFromQuery::new(target)));
448 let res =
449 self.with_prepared_query(Box::new(query), false, &mut |stmt, binds, conn, source| {
450 fn inner_copy_in<S, T>(
451 stmt: MaybeCached<'_, Statement>,
452 conn: &mut ConnectionAndTransactionManager,
453 binds: Vec<Option<Vec<u8>>>,
454 source: &dyn QueryFragmentHelper<S::Error>,
455 ) -> Result<usize, S::Error>
456 where
457 S: CopyFromExpression<T>,
458 {
459 let _res = stmt.execute(&mut conn.raw_connection, &binds, false)?;
460 let mut copy_in = CopyFromSink::new(&mut conn.raw_connection);
461 let r = source.write_copy_from(&mut copy_in);
462 copy_in.finish(r.as_ref().err().map(|e| e.to_string()))?;
463 let next_res = conn.raw_connection.get_next_result()?.ok_or_else(|| {
464 crate::result::Error::DeserializationError(
465 "Failed to receive result from the database".into(),
466 )
467 })?;
468 let rows = next_res.rows_affected();
469 while let Some(_r) = conn.raw_connection.get_next_result()? {}
470 r?;
471 rows.map_err(Into::into)
472 }
473
474 let rows = inner_copy_in::<S, T>(stmt, conn, binds, &*source);
475 if let Err(ref e) = rows {
476 let database_error = crate::result::Error::DatabaseError(
477 crate::result::DatabaseErrorKind::Unknown,
478 Box::new(e.to_string()),
479 );
480 source.instrumentation(&mut |query| {
481 conn.instrumentation.on_connection_event(
482 InstrumentationEvent::FinishQuery {
483 query,
484 error: Some(&database_error),
485 },
486 );
487 });
488 } else {
489 source.instrumentation(&mut |query| {
490 conn.instrumentation.on_connection_event(
491 InstrumentationEvent::FinishQuery { query, error: None },
492 );
493 });
494 }
495
496 rows
497 })?;
498
499 Ok(res)
500 }
501
502 pub(crate) fn copy_to<T>(&mut self, command: CopyToCommand<T>) -> QueryResult<CopyToBuffer<'_>>
503 where
504 T: CopyTarget,
505 {
506 let res = self.with_prepared_query::<_, Error>(
507 Box::new(command),
508 false,
509 &mut |stmt, binds, conn, source| {
510 let res = stmt.execute(&mut conn.raw_connection, &binds, false);
511 source.instrumentation(&mut |query| {
512 conn.instrumentation
513 .on_connection_event(InstrumentationEvent::FinishQuery {
514 query,
515 error: res.as_ref().err(),
516 });
517 });
518 Ok(CopyToBuffer::new(&mut conn.raw_connection, res?))
519 },
520 )?;
521 Ok(res)
522 }
523
524 fn with_prepared_query<'conn, 'query, R, E>(
525 &'conn mut self,
526 source: Box<dyn QueryFragmentHelper<E> + 'query>,
527 execute_returning_count: bool,
528 f: &mut dyn FnMut(
529 MaybeCached<'_, Statement>,
530 Vec<Option<Vec<u8>>>,
531 &'conn mut ConnectionAndTransactionManager,
532 Box<dyn QueryFragmentHelper<E> + 'query>,
533 ) -> Result<R, E>,
534 ) -> Result<R, E>
535 where
536 E: From<crate::result::Error>,
537 {
538 fn prepare_query_non_generic_inner<'a, E>(
539 connection_and_transaction_manager: &mut ConnectionAndTransactionManager,
540 cache: &'a mut StatementCache<Pg, Statement>,
541 source: &dyn QueryFragmentHelper<E>,
542 execute_returning_count: bool,
543 bind_collector: RawBytesBindCollector<Pg>,
544 ) -> QueryResult<(
545 Vec<Option<Vec<u8>>>,
546 Result<MaybeCached<'a, Statement>, Error>,
547 )> {
548 let binds = bind_collector.binds;
549 let metadata = bind_collector.metadata;
550 let query = cache.cached_statement_non_generic(
551 source.query_id(),
552 source,
553 &Pg,
554 &metadata,
555 &mut connection_and_transaction_manager.raw_connection,
556 Statement::prepare,
557 &mut *connection_and_transaction_manager.instrumentation,
558 );
559 if !execute_returning_count && let Err(ref e) = query {
560 source.instrumentation(&mut |query| {
561 connection_and_transaction_manager
562 .instrumentation
563 .on_connection_event(InstrumentationEvent::FinishQuery {
564 query,
565 error: Some(e),
566 });
567 });
568 }
569 Ok((binds, query))
570 }
571
572 let bind_collector = self.collect_binds(&*source)?;
573 let (binds, query) = prepare_query_non_generic_inner(
574 &mut self.connection_and_transaction_manager,
575 &mut self.statement_cache,
576 &*source,
577 execute_returning_count,
578 bind_collector,
579 )?;
580
581 f(
582 query?,
583 binds,
584 &mut self.connection_and_transaction_manager,
585 source,
586 )
587 }
588
589 fn collect_binds<E>(
590 &mut self,
591 source: &dyn QueryFragmentHelper<E>,
592 ) -> Result<RawBytesBindCollector<Pg>, crate::result::Error> {
593 source.instrumentation(&mut |query| {
594 self.connection_and_transaction_manager
595 .instrumentation
596 .on_connection_event(InstrumentationEvent::StartQuery { query });
597 });
598 let mut bind_collector = RawBytesBindCollector::<Pg>::new();
599 source.collect_binds(&mut bind_collector, self)?;
600 Ok(bind_collector)
601 }
602
603 fn set_config_options(&mut self) -> QueryResult<()> {
604 crate::sql_query("SET TIME ZONE 'UTC'").execute(self)?;
605 crate::sql_query("SET CLIENT_ENCODING TO 'UTF8'").execute(self)?;
606 self.connection_and_transaction_manager
607 .raw_connection
608 .set_notice_processor(noop_notice_processor);
609 Ok(())
610 }
611
612 /// See Postgres documentation for SQL commands [NOTIFY][] and [LISTEN][]
613 ///
614 /// The returned iterator can yield items even after a None value when new notifications have been received.
615 /// The iterator can be polled again after a `None` value was received as new notifications might have
616 /// been send in the mean time.
617 ///
618 /// [NOTIFY]: https://www.postgresql.org/docs/current/sql-notify.html
619 /// [LISTEN]: https://www.postgresql.org/docs/current/sql-listen.html
620 ///
621 /// ## Example
622 ///
623 /// ```
624 /// # include!("../../doctest_setup.rs");
625 /// #
626 /// # fn main() {
627 /// # run_test().unwrap();
628 /// # }
629 /// #
630 /// # fn run_test() -> QueryResult<()> {
631 /// # let connection = &mut establish_connection();
632 ///
633 /// // register the notifications channel we want to receive notifications for
634 /// diesel::sql_query("LISTEN example_channel").execute(connection)?;
635 /// // send some notification
636 /// // this is usually done from a different connection/thread/application
637 /// diesel::sql_query("NOTIFY example_channel, 'additional data'").execute(connection)?;
638 ///
639 /// for result in connection.notifications_iter() {
640 /// let notification = result.unwrap();
641 /// assert_eq!(notification.channel, "example_channel");
642 /// assert_eq!(notification.payload, "additional data");
643 ///
644 /// println!(
645 /// "Notification received from server process with id {}.",
646 /// notification.process_id
647 /// );
648 /// }
649 /// # Ok(())
650 /// # }
651 /// ```
652 pub fn notifications_iter(&mut self) -> impl Iterator<Item = QueryResult<PgNotification>> + '_ {
653 let conn = &self.connection_and_transaction_manager.raw_connection;
654 core::iter::from_fn(move || conn.pq_notifies().transpose())
655 }
656}
657
658extern "C" fn noop_notice_processor(_: *mut libc::c_void, _message: *const libc::c_char) {}
659
660mod private {
661 use super::*;
662
663 #[allow(missing_debug_implementations)]
664 pub struct ConnectionAndTransactionManager {
665 pub(super) raw_connection: RawConnection,
666 pub(super) transaction_state: AnsiTransactionManager,
667 pub(super) instrumentation: DynInstrumentation,
668 }
669
670 pub trait PgLoadingMode<B> {
671 const USE_ROW_BY_ROW_MODE: bool;
672 type Cursor<'conn, 'query>: Iterator<Item = QueryResult<Self::Row<'conn, 'query>>>;
673 type Row<'conn, 'query>: crate::row::Row<'conn, Pg>;
674
675 fn get_cursor<'conn, 'query>(
676 raw_connection: &'conn mut ConnectionAndTransactionManager,
677 result: PgResult,
678 source: Box<dyn QueryFragmentHelper<crate::result::Error> + 'query>,
679 ) -> QueryResult<Self::Cursor<'conn, 'query>>;
680 }
681
682 impl PgLoadingMode<DefaultLoadingMode> for PgConnection {
683 const USE_ROW_BY_ROW_MODE: bool = false;
684 type Cursor<'conn, 'query> = Cursor;
685 type Row<'conn, 'query> = self::row::PgRow;
686
687 fn get_cursor<'conn, 'query>(
688 conn: &'conn mut ConnectionAndTransactionManager,
689 result: PgResult,
690 source: Box<dyn QueryFragmentHelper<crate::result::Error> + 'query>,
691 ) -> QueryResult<Self::Cursor<'conn, 'query>> {
692 update_transaction_manager_status(
693 Cursor::new(result, &mut conn.raw_connection),
694 conn,
695 &|callback| source.instrumentation(callback),
696 true,
697 )
698 }
699 }
700
701 impl PgLoadingMode<PgRowByRowLoadingMode> for PgConnection {
702 const USE_ROW_BY_ROW_MODE: bool = true;
703 type Cursor<'conn, 'query> = RowByRowCursor<'conn, 'query>;
704 type Row<'conn, 'query> = self::row::PgRow;
705
706 fn get_cursor<'conn, 'query>(
707 raw_connection: &'conn mut ConnectionAndTransactionManager,
708 result: PgResult,
709 source: Box<dyn QueryFragmentHelper<crate::result::Error> + 'query>,
710 ) -> QueryResult<Self::Cursor<'conn, 'query>> {
711 Ok(RowByRowCursor::new(result, raw_connection, source))
712 }
713 }
714
715 // this trait exists to turn generic query types
716 // into trait objects to deduplicate the connection code
717 pub trait QueryFragmentHelper<E>:
718 crate::connection::statement_cache::QueryFragmentForCachedStatement<crate::pg::Pg>
719 {
720 fn query_id(&self) -> Option<core::any::TypeId>;
721
722 fn instrumentation(
723 &self,
724 callback: &mut dyn FnMut(&dyn crate::connection::instrumentation::DebugQuery),
725 );
726
727 fn collect_binds(
728 &self,
729 bind_collector: &mut RawBytesBindCollector<crate::pg::Pg>,
730 conn: &mut PgConnection,
731 ) -> QueryResult<()>;
732
733 fn write_copy_from(&self, _sink: &mut CopyFromSink<'_>) -> Result<(), E> {
734 Ok(())
735 }
736 }
737
738 // any type that implement `QueryFragment` and `QueryId` works in this location
739 impl<T> QueryFragmentHelper<diesel::result::Error> for T
740 where
741 T: QueryFragment<crate::pg::Pg> + QueryId,
742 {
743 fn query_id(&self) -> Option<core::any::TypeId> {
744 <T as QueryId>::query_id()
745 }
746
747 fn instrumentation(
748 &self,
749 callback: &mut dyn FnMut(&dyn crate::connection::instrumentation::DebugQuery),
750 ) {
751 callback(&crate::debug_query(self))
752 }
753
754 fn collect_binds(
755 &self,
756 bind_collector: &mut RawBytesBindCollector<crate::pg::Pg>,
757 conn: &mut PgConnection,
758 ) -> QueryResult<()> {
759 <Self as QueryFragment<diesel::pg::Pg>>::collect_binds(
760 self,
761 bind_collector,
762 conn,
763 &crate::pg::Pg,
764 )
765 }
766 }
767
768 // This wrapper exists as we need to have custom behavior for copy from
769 // statements (fn write_copy_from is relevant)
770 pub(super) struct CopyFromWrapper<S, T>(
771 pub(super) core::cell::RefCell<InternalCopyFromQuery<S, T>>,
772 );
773
774 impl<S, T> crate::connection::statement_cache::QueryFragmentForCachedStatement<Pg>
775 for CopyFromWrapper<S, T>
776 where
777 InternalCopyFromQuery<S, T>:
778 crate::connection::statement_cache::QueryFragmentForCachedStatement<Pg>,
779 {
780 fn construct_sql(&self, backend: &Pg) -> QueryResult<String> {
781 self.0.borrow().construct_sql(backend)
782 }
783
784 fn is_safe_to_cache_prepared(&self, backend: &Pg) -> QueryResult<bool> {
785 self.0.borrow().is_safe_to_cache_prepared(backend)
786 }
787 }
788
789 impl<S, T> QueryFragmentHelper<S::Error> for CopyFromWrapper<S, T>
790 where
791 S: CopyFromExpression<T>,
792 InternalCopyFromQuery<S, T>: QueryFragmentHelper<crate::result::Error>,
793 Self: crate::connection::statement_cache::QueryFragmentForCachedStatement<Pg>,
794 {
795 fn query_id(&self) -> Option<core::any::TypeId> {
796 self.0.borrow().query_id()
797 }
798
799 fn instrumentation(
800 &self,
801 callback: &mut dyn FnMut(&dyn crate::connection::instrumentation::DebugQuery),
802 ) {
803 callback(&crate::debug_query(&*self.0.borrow()))
804 }
805
806 fn collect_binds(
807 &self,
808 bind_collector: &mut RawBytesBindCollector<crate::pg::Pg>,
809 conn: &mut PgConnection,
810 ) -> QueryResult<()> {
811 <InternalCopyFromQuery<S, T> as QueryFragmentHelper<crate::result::Error>>::collect_binds(
812 &*self.0.borrow(),
813 bind_collector,
814 conn,
815 )
816 }
817
818 fn write_copy_from(&self, sink: &mut CopyFromSink<'_>) -> Result<(), S::Error> {
819 self.0.borrow_mut().target.callback(sink)
820 }
821 }
822}
823
824#[cfg(test)]
825// that's a false positive for `panic!`/`assert!` on rust 2018
826#[allow(clippy::uninlined_format_args)]
827mod tests {
828 extern crate dotenvy;
829
830 use super::*;
831 use crate::prelude::*;
832 use crate::result::Error::DatabaseError;
833 use std::num::NonZeroU32;
834
835 fn connection() -> PgConnection {
836 crate::test_helpers::pg_connection_no_transaction()
837 }
838
839 #[diesel_test_helper::test]
840 fn notifications_arrive() {
841 use crate::sql_query;
842
843 let conn = &mut connection();
844 sql_query("LISTEN test_notifications")
845 .execute(conn)
846 .unwrap();
847 sql_query("NOTIFY test_notifications, 'first'")
848 .execute(conn)
849 .unwrap();
850 sql_query("NOTIFY test_notifications, 'second'")
851 .execute(conn)
852 .unwrap();
853
854 let notifications = conn
855 .notifications_iter()
856 .map(Result::unwrap)
857 .collect::<Vec<_>>();
858
859 assert_eq!(2, notifications.len());
860 assert_eq!(notifications[0].channel, "test_notifications");
861 assert_eq!(notifications[1].channel, "test_notifications");
862 assert_eq!(notifications[0].payload, "first");
863 assert_eq!(notifications[1].payload, "second");
864
865 let next_notification = conn.notifications_iter().next();
866 assert!(
867 next_notification.is_none(),
868 "Got a next notification, while not expecting one: {next_notification:?}"
869 );
870
871 sql_query("NOTIFY test_notifications")
872 .execute(conn)
873 .unwrap();
874 assert_eq!(
875 conn.notifications_iter().next().unwrap().unwrap().payload,
876 ""
877 );
878 }
879
880 #[diesel_test_helper::test]
881 fn malformed_sql_query() {
882 let connection = &mut connection();
883 let query =
884 crate::sql_query("SELECT not_existent FROM also_not_there;").execute(connection);
885
886 if let Err(DatabaseError(_, string)) = query {
887 assert_eq!(Some(26), string.statement_position());
888 } else {
889 unreachable!();
890 }
891 }
892
893 table! {
894 users {
895 id -> Integer,
896 name -> Text,
897 }
898 }
899
900 #[diesel_test_helper::test]
901 fn transaction_manager_returns_an_error_when_attempting_to_commit_outside_of_a_transaction() {
902 use crate::PgConnection;
903 use crate::connection::{AnsiTransactionManager, TransactionManager};
904 use crate::result::Error;
905
906 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
907 assert_eq!(
908 None,
909 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
910 conn
911 ).transaction_depth().expect("Transaction depth")
912 );
913 let result = AnsiTransactionManager::commit_transaction(conn);
914 assert!(matches!(result, Err(Error::NotInTransaction)))
915 }
916
917 #[diesel_test_helper::test]
918 fn transaction_manager_returns_an_error_when_attempting_to_rollback_outside_of_a_transaction() {
919 use crate::PgConnection;
920 use crate::connection::{AnsiTransactionManager, TransactionManager};
921 use crate::result::Error;
922
923 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
924 assert_eq!(
925 None,
926 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
927 conn
928 ).transaction_depth().expect("Transaction depth")
929 );
930 let result = AnsiTransactionManager::rollback_transaction(conn);
931 assert!(matches!(result, Err(Error::NotInTransaction)))
932 }
933
934 #[diesel_test_helper::test]
935 fn postgres_transaction_is_rolled_back_upon_syntax_error() {
936 use std::num::NonZeroU32;
937
938 use crate::connection::{AnsiTransactionManager, TransactionManager};
939 use crate::pg::connection::raw::PgTransactionStatus;
940 use crate::*;
941 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
942 assert_eq!(
943 None,
944 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
945 conn
946 ).transaction_depth().expect("Transaction depth")
947 );
948 let _result = conn.build_transaction().run(|conn| {
949 assert_eq!(
950 NonZeroU32::new(1),
951 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
952 conn
953 ).transaction_depth().expect("Transaction depth")
954 );
955 // In Postgres, a syntax error breaks the transaction block
956 let query_result = sql_query("SELECT_SYNTAX_ERROR 1").execute(conn);
957 assert!(query_result.is_err());
958 assert_eq!(
959 PgTransactionStatus::InError,
960 conn.connection_and_transaction_manager.raw_connection.transaction_status()
961 );
962 query_result
963 });
964 assert_eq!(
965 None,
966 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
967 conn
968 ).transaction_depth().expect("Transaction depth")
969 );
970 assert_eq!(
971 PgTransactionStatus::Idle,
972 conn.connection_and_transaction_manager
973 .raw_connection
974 .transaction_status()
975 );
976 }
977
978 #[diesel_test_helper::test]
979 fn nested_postgres_transaction_is_rolled_back_upon_syntax_error() {
980 use std::num::NonZeroU32;
981
982 use crate::connection::{AnsiTransactionManager, TransactionManager};
983 use crate::pg::connection::raw::PgTransactionStatus;
984 use crate::*;
985 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
986 assert_eq!(
987 None,
988 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
989 conn
990 ).transaction_depth().expect("Transaction depth")
991 );
992 let result = conn.build_transaction().run(|conn| {
993 assert_eq!(
994 NonZeroU32::new(1),
995 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
996 conn
997 ).transaction_depth().expect("Transaction depth")
998 );
999 let result = conn.build_transaction().run(|conn| {
1000 assert_eq!(
1001 NonZeroU32::new(2),
1002 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1003 conn
1004 ).transaction_depth().expect("Transaction depth")
1005 );
1006 sql_query("SELECT_SYNTAX_ERROR 1").execute(conn)
1007 });
1008 assert!(result.is_err());
1009 assert_eq!(
1010 NonZeroU32::new(1),
1011 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1012 conn
1013 ).transaction_depth().expect("Transaction depth")
1014 );
1015 let query_result = sql_query("SELECT 1").execute(conn);
1016 assert!(query_result.is_ok());
1017 assert_eq!(
1018 PgTransactionStatus::InTransaction,
1019 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1020 );
1021 query_result
1022 });
1023 assert!(result.is_ok());
1024 assert_eq!(
1025 PgTransactionStatus::Idle,
1026 conn.connection_and_transaction_manager
1027 .raw_connection
1028 .transaction_status()
1029 );
1030 assert_eq!(
1031 None,
1032 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1033 conn
1034 ).transaction_depth().expect("Transaction depth")
1035 );
1036 }
1037
1038 #[diesel_test_helper::test]
1039 // This function uses collect with an side effect (spawning threads)
1040 // so this is a false positive from clippy
1041 #[allow(clippy::needless_collect)]
1042 fn postgres_transaction_depth_is_tracked_properly_on_serialization_failure() {
1043 use crate::pg::connection::raw::PgTransactionStatus;
1044 use crate::result::DatabaseErrorKind::SerializationFailure;
1045 use crate::result::Error::DatabaseError;
1046 use crate::*;
1047 use std::sync::{Arc, Barrier};
1048 use std::thread;
1049
1050 table! {
1051 #[sql_name = "pg_transaction_depth_is_tracked_properly_on_commit_failure"]
1052 serialization_example {
1053 id -> Serial,
1054 class -> Integer,
1055 }
1056 }
1057
1058 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1059
1060 sql_query(
1061 "DROP TABLE IF EXISTS pg_transaction_depth_is_tracked_properly_on_commit_failure;",
1062 )
1063 .execute(conn)
1064 .unwrap();
1065 sql_query(
1066 r#"
1067 CREATE TABLE pg_transaction_depth_is_tracked_properly_on_commit_failure (
1068 id SERIAL PRIMARY KEY,
1069 class INTEGER NOT NULL
1070 )
1071 "#,
1072 )
1073 .execute(conn)
1074 .unwrap();
1075
1076 insert_into(serialization_example::table)
1077 .values(&vec![
1078 serialization_example::class.eq(1),
1079 serialization_example::class.eq(2),
1080 ])
1081 .execute(conn)
1082 .unwrap();
1083
1084 let before_barrier = Arc::new(Barrier::new(2));
1085 let after_barrier = Arc::new(Barrier::new(2));
1086 let threads = (1..3)
1087 .map(|i| {
1088 let before_barrier = before_barrier.clone();
1089 let after_barrier = after_barrier.clone();
1090 thread::spawn(move || {
1091 use crate::connection::AnsiTransactionManager;
1092 use crate::connection::TransactionManager;
1093 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1094 assert_eq!(None, <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1095
1096 let result = conn.build_transaction().serializable().run(|conn| {
1097 assert_eq!(NonZeroU32::new(1), <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1098
1099 let _ = serialization_example::table
1100 .filter(serialization_example::class.eq(i))
1101 .count()
1102 .execute(conn)?;
1103
1104 let other_i = if i == 1 { 2 } else { 1 };
1105 let q = insert_into(serialization_example::table)
1106 .values(serialization_example::class.eq(other_i));
1107 before_barrier.wait();
1108
1109 let r = q.execute(conn);
1110 after_barrier.wait();
1111 r
1112 });
1113 assert_eq!(
1114 PgTransactionStatus::Idle,
1115 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1116 );
1117
1118 assert_eq!(None, <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1119 result
1120 })
1121 })
1122 .collect::<Vec<_>>();
1123
1124 let mut results = threads
1125 .into_iter()
1126 .map(|t| t.join().unwrap())
1127 .collect::<Vec<_>>();
1128
1129 results.sort_by_key(|r| r.is_err());
1130
1131 assert!(results[0].is_ok(), "Got {:?} instead", results);
1132 assert!(
1133 matches!(&results[1], Err(DatabaseError(SerializationFailure, _))),
1134 "Got {:?} instead",
1135 results
1136 );
1137 assert_eq!(
1138 PgTransactionStatus::Idle,
1139 conn.connection_and_transaction_manager
1140 .raw_connection
1141 .transaction_status()
1142 );
1143 }
1144
1145 #[diesel_test_helper::test]
1146 // This function uses collect with an side effect (spawning threads)
1147 // so this is a false positive from clippy
1148 #[allow(clippy::needless_collect)]
1149 fn postgres_transaction_depth_is_tracked_properly_on_nested_serialization_failure() {
1150 use crate::pg::connection::raw::PgTransactionStatus;
1151 use crate::result::DatabaseErrorKind::SerializationFailure;
1152 use crate::result::Error::DatabaseError;
1153 use crate::*;
1154 use std::sync::{Arc, Barrier};
1155 use std::thread;
1156
1157 table! {
1158 #[sql_name = "pg_nested_transaction_depth_is_tracked_properly_on_commit_failure"]
1159 serialization_example {
1160 id -> Serial,
1161 class -> Integer,
1162 }
1163 }
1164
1165 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1166
1167 sql_query(
1168 "DROP TABLE IF EXISTS pg_nested_transaction_depth_is_tracked_properly_on_commit_failure;",
1169 )
1170 .execute(conn)
1171 .unwrap();
1172 sql_query(
1173 r#"
1174 CREATE TABLE pg_nested_transaction_depth_is_tracked_properly_on_commit_failure (
1175 id SERIAL PRIMARY KEY,
1176 class INTEGER NOT NULL
1177 )
1178 "#,
1179 )
1180 .execute(conn)
1181 .unwrap();
1182
1183 insert_into(serialization_example::table)
1184 .values(&vec![
1185 serialization_example::class.eq(1),
1186 serialization_example::class.eq(2),
1187 ])
1188 .execute(conn)
1189 .unwrap();
1190
1191 let before_barrier = Arc::new(Barrier::new(2));
1192 let after_barrier = Arc::new(Barrier::new(2));
1193 let threads = (1..3)
1194 .map(|i| {
1195 let before_barrier = before_barrier.clone();
1196 let after_barrier = after_barrier.clone();
1197 thread::spawn(move || {
1198 use crate::connection::AnsiTransactionManager;
1199 use crate::connection::TransactionManager;
1200 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1201 assert_eq!(None, <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1202
1203 let result = conn.build_transaction().serializable().run(|conn| {
1204 assert_eq!(NonZeroU32::new(1), <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1205 let r = conn.transaction(|conn| {
1206 assert_eq!(NonZeroU32::new(2), <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1207
1208 let _ = serialization_example::table
1209 .filter(serialization_example::class.eq(i))
1210 .count()
1211 .execute(conn)?;
1212
1213 let other_i = if i == 1 { 2 } else { 1 };
1214 let q = insert_into(serialization_example::table)
1215 .values(serialization_example::class.eq(other_i));
1216 before_barrier.wait();
1217
1218 let r = q.execute(conn);
1219 after_barrier.wait();
1220 r
1221 });
1222 assert_eq!(NonZeroU32::new(1), <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1223 assert_eq!(
1224 PgTransactionStatus::InTransaction,
1225 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1226 );
1227 r
1228 });
1229 assert_eq!(
1230 PgTransactionStatus::Idle,
1231 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1232 );
1233
1234 assert_eq!(None, <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(conn).transaction_depth().expect("Transaction depth"));
1235 result
1236 })
1237 })
1238 .collect::<Vec<_>>();
1239
1240 let mut results = threads
1241 .into_iter()
1242 .map(|t| t.join().unwrap())
1243 .collect::<Vec<_>>();
1244
1245 results.sort_by_key(|r| r.is_err());
1246
1247 assert!(results[0].is_ok(), "Got {:?} instead", results);
1248 assert!(
1249 matches!(&results[1], Err(DatabaseError(SerializationFailure, _))),
1250 "Got {:?} instead",
1251 results
1252 );
1253 assert_eq!(
1254 PgTransactionStatus::Idle,
1255 conn.connection_and_transaction_manager
1256 .raw_connection
1257 .transaction_status()
1258 );
1259 }
1260
1261 #[diesel_test_helper::test]
1262 fn postgres_transaction_is_rolled_back_upon_deferred_constraint_failure() {
1263 use crate::connection::{AnsiTransactionManager, TransactionManager};
1264 use crate::pg::connection::raw::PgTransactionStatus;
1265 use crate::result::Error;
1266 use crate::*;
1267
1268 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1269 assert_eq!(
1270 None,
1271 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1272 conn
1273 ).transaction_depth().expect("Transaction depth")
1274 );
1275 let result: Result<_, Error> = conn.build_transaction().run(|conn| {
1276 assert_eq!(
1277 NonZeroU32::new(1),
1278 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1279 conn
1280 ).transaction_depth().expect("Transaction depth")
1281 );
1282 sql_query("DROP TABLE IF EXISTS deferred_constraint_commit").execute(conn)?;
1283 sql_query("CREATE TABLE deferred_constraint_commit(id INT UNIQUE INITIALLY DEFERRED)")
1284 .execute(conn)?;
1285 sql_query("INSERT INTO deferred_constraint_commit VALUES(1)").execute(conn)?;
1286 let result =
1287 sql_query("INSERT INTO deferred_constraint_commit VALUES(1)").execute(conn);
1288 assert!(result.is_ok());
1289 assert_eq!(
1290 PgTransactionStatus::InTransaction,
1291 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1292 );
1293 Ok(())
1294 });
1295 assert_eq!(
1296 None,
1297 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1298 conn
1299 ).transaction_depth().expect("Transaction depth")
1300 );
1301 assert_eq!(
1302 PgTransactionStatus::Idle,
1303 conn.connection_and_transaction_manager
1304 .raw_connection
1305 .transaction_status()
1306 );
1307 assert!(result.is_err());
1308 }
1309
1310 #[diesel_test_helper::test]
1311 fn postgres_transaction_is_rolled_back_upon_deferred_trigger_failure() {
1312 use crate::connection::{AnsiTransactionManager, TransactionManager};
1313 use crate::pg::connection::raw::PgTransactionStatus;
1314 use crate::result::Error;
1315 use crate::*;
1316
1317 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1318 assert_eq!(
1319 None,
1320 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1321 conn
1322 ).transaction_depth().expect("Transaction depth")
1323 );
1324 let result: Result<_, Error> = conn.build_transaction().run(|conn| {
1325 assert_eq!(
1326 NonZeroU32::new(1),
1327 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1328 conn
1329 ).transaction_depth().expect("Transaction depth")
1330 );
1331 sql_query("DROP TABLE IF EXISTS deferred_trigger_commit").execute(conn)?;
1332 sql_query("CREATE TABLE deferred_trigger_commit(id INT UNIQUE INITIALLY DEFERRED)")
1333 .execute(conn)?;
1334 sql_query(
1335 r#"
1336 CREATE OR REPLACE FUNCTION transaction_depth_blow_up()
1337 RETURNS trigger
1338 LANGUAGE plpgsql
1339 AS $$
1340 DECLARE
1341 BEGIN
1342 IF NEW.value = 42 THEN
1343 RAISE EXCEPTION 'Transaction kaboom';
1344 END IF;
1345 RETURN NEW;
1346
1347 END;$$;
1348 "#,
1349 )
1350 .execute(conn)?;
1351
1352 sql_query(
1353 r#"
1354 CREATE CONSTRAINT TRIGGER transaction_depth_trigger
1355 AFTER INSERT ON "deferred_trigger_commit"
1356 DEFERRABLE INITIALLY DEFERRED
1357 FOR EACH ROW
1358 EXECUTE PROCEDURE transaction_depth_blow_up()
1359 "#,
1360 )
1361 .execute(conn)?;
1362 let result = sql_query("INSERT INTO deferred_trigger_commit VALUES(42)").execute(conn);
1363 assert!(result.is_ok());
1364 assert_eq!(
1365 PgTransactionStatus::InTransaction,
1366 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1367 );
1368 Ok(())
1369 });
1370 assert_eq!(
1371 None,
1372 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1373 conn
1374 ).transaction_depth().expect("Transaction depth")
1375 );
1376 assert_eq!(
1377 PgTransactionStatus::Idle,
1378 conn.connection_and_transaction_manager
1379 .raw_connection
1380 .transaction_status()
1381 );
1382 assert!(result.is_err());
1383 }
1384
1385 #[diesel_test_helper::test]
1386 fn nested_postgres_transaction_is_rolled_back_upon_deferred_trigger_failure() {
1387 use crate::connection::{AnsiTransactionManager, TransactionManager};
1388 use crate::pg::connection::raw::PgTransactionStatus;
1389 use crate::result::Error;
1390 use crate::*;
1391
1392 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1393 assert_eq!(
1394 None,
1395 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1396 conn
1397 ).transaction_depth().expect("Transaction depth")
1398 );
1399 let result: Result<_, Error> = conn.build_transaction().run(|conn| {
1400 assert_eq!(
1401 NonZeroU32::new(1),
1402 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1403 conn
1404 ).transaction_depth().expect("Transaction depth")
1405 );
1406 sql_query("DROP TABLE IF EXISTS deferred_trigger_nested_commit").execute(conn)?;
1407 sql_query(
1408 "CREATE TABLE deferred_trigger_nested_commit(id INT UNIQUE INITIALLY DEFERRED)",
1409 )
1410 .execute(conn)?;
1411 sql_query(
1412 r#"
1413 CREATE OR REPLACE FUNCTION transaction_depth_blow_up()
1414 RETURNS trigger
1415 LANGUAGE plpgsql
1416 AS $$
1417 DECLARE
1418 BEGIN
1419 IF NEW.value = 42 THEN
1420 RAISE EXCEPTION 'Transaction kaboom';
1421 END IF;
1422 RETURN NEW;
1423
1424 END;$$;
1425 "#,
1426 )
1427 .execute(conn)?;
1428
1429 sql_query(
1430 r#"
1431 CREATE CONSTRAINT TRIGGER transaction_depth_trigger
1432 AFTER INSERT ON "deferred_trigger_nested_commit"
1433 DEFERRABLE INITIALLY DEFERRED
1434 FOR EACH ROW
1435 EXECUTE PROCEDURE transaction_depth_blow_up()
1436 "#,
1437 )
1438 .execute(conn)?;
1439 let inner_result: Result<_, Error> = conn.build_transaction().run(|conn| {
1440 let result = sql_query("INSERT INTO deferred_trigger_nested_commit VALUES(42)")
1441 .execute(conn);
1442 assert!(result.is_ok());
1443 Ok(())
1444 });
1445 assert!(inner_result.is_err());
1446 assert_eq!(
1447 PgTransactionStatus::InTransaction,
1448 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1449 );
1450 Ok(())
1451 });
1452 assert_eq!(
1453 None,
1454 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1455 conn
1456 ).transaction_depth().expect("Transaction depth")
1457 );
1458 assert_eq!(
1459 PgTransactionStatus::Idle,
1460 conn.connection_and_transaction_manager
1461 .raw_connection
1462 .transaction_status()
1463 );
1464 assert!(result.is_ok(), "Expected success, got {:?}", result);
1465 }
1466
1467 #[diesel_test_helper::test]
1468 fn nested_postgres_transaction_is_rolled_back_upon_deferred_constraint_failure() {
1469 use crate::connection::{AnsiTransactionManager, TransactionManager};
1470 use crate::pg::connection::raw::PgTransactionStatus;
1471 use crate::result::Error;
1472 use crate::*;
1473
1474 let conn = &mut crate::test_helpers::pg_connection_no_transaction();
1475 assert_eq!(
1476 None,
1477 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1478 conn
1479 ).transaction_depth().expect("Transaction depth")
1480 );
1481 let result: Result<_, Error> = conn.build_transaction().run(|conn| {
1482 assert_eq!(
1483 NonZeroU32::new(1),
1484 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1485 conn
1486 ).transaction_depth().expect("Transaction depth")
1487 );
1488 sql_query("DROP TABLE IF EXISTS deferred_constraint_nested_commit").execute(conn)?;
1489 sql_query("CREATE TABLE deferred_constraint_nested_commit(id INT UNIQUE INITIALLY DEFERRED)").execute(conn)?;
1490 let inner_result: Result<_, Error> = conn.build_transaction().run(|conn| {
1491 assert_eq!(
1492 NonZeroU32::new(2),
1493 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1494 conn
1495 ).transaction_depth().expect("Transaction depth")
1496 );
1497 sql_query("INSERT INTO deferred_constraint_nested_commit VALUES(1)").execute(conn)?;
1498 let result = sql_query("INSERT INTO deferred_constraint_nested_commit VALUES(1)").execute(conn);
1499 assert!(result.is_ok());
1500 Ok(())
1501 });
1502 assert!(inner_result.is_err());
1503 assert_eq!(
1504 PgTransactionStatus::InTransaction,
1505 conn.connection_and_transaction_manager.raw_connection.transaction_status()
1506 );
1507 assert_eq!(
1508 NonZeroU32::new(1),
1509 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1510 conn
1511 ).transaction_depth().expect("Transaction depth")
1512 );
1513 sql_query("INSERT INTO deferred_constraint_nested_commit VALUES(1)").execute(conn)
1514 });
1515 assert_eq!(
1516 None,
1517 <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
1518 conn
1519 ).transaction_depth().expect("Transaction depth")
1520 );
1521 assert_eq!(
1522 PgTransactionStatus::Idle,
1523 conn.connection_and_transaction_manager
1524 .raw_connection
1525 .transaction_status()
1526 );
1527 assert!(result.is_ok());
1528 }
1529}