diesel/mysql_like/connection/
mod.rs

1mod bind;
2mod error_information;
3mod raw;
4mod stmt;
5mod url;
6
7use core::num::NonZeroU64;
8
9use self::raw::RawConnection;
10use self::stmt::Statement;
11use self::stmt::iterator::StatementIterator;
12use self::url::ConnectionOptions;
13use crate::RunQueryDsl;
14use crate::connection::instrumentation::{DebugQuery, DynInstrumentation, StrQueryHelper};
15use crate::connection::statement_cache::{MaybeCached, StatementCache};
16use crate::connection::*;
17use crate::expression::QueryMetadata;
18use crate::mysql_like::MysqlLikeBackend;
19use crate::query_builder::bind_collector::RawBytesBindCollector;
20use crate::query_builder::*;
21use crate::result::*;
22
23#[allow(missing_debug_implementations, missing_copy_implementations)]
24/// A connection to a MySQL database. Connection URLs should be in the form
25/// `mysql://[user[:password]@]host/database_name[?unix_socket=socket-path&ssl_mode=SSL_MODE*&ssl_ca=/etc/ssl/certs/ca-certificates.crt&ssl_cert=/etc/ssl/certs/client-cert.crt&ssl_key=/etc/ssl/certs/client-key.crt]`
26///
27///* `host` can be an IP address or a hostname. If it is set to `localhost`, a connection
28///  will be attempted through the socket at `/tmp/mysql.sock`. If you want to connect to
29///  a local server via TCP (e.g. docker containers), use `0.0.0.0` or `127.0.0.1` instead.
30/// * `unix_socket` expects the path to the unix socket
31/// * `ssl_ca` accepts a path to the system's certificate roots
32/// * `ssl_cert` accepts a path to the client's certificate file
33/// * `ssl_key` accepts a path to the client's private key file
34/// * `ssl_mode` expects a value defined for MySQL client command option `--ssl-mode`
35///   See <https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#option_general_ssl-mode>
36///
37/// # Supported loading model implementations
38///
39/// * [`DefaultLoadingMode`](crate::connection::DefaultLoadingMode)
40///
41/// As `MysqlConnection` only supports a single loading mode implementation
42/// it is **not required** to explicitly specify a loading mode
43/// when calling [`RunQueryDsl::load_iter()`] or [`LoadConnection::load`]
44///
45/// ## DefaultLoadingMode
46///
47/// `MysqlConnection` only supports a single loading mode, which loads
48/// values row by row from the result set.
49///
50/// ```rust,dejadoc
51/// # include!("../../doctest_setup.rs");
52/// #
53/// # fn main() {
54/// #     run_test().unwrap();
55/// # }
56/// #
57/// # fn run_test() -> QueryResult<()> {
58/// #     use schema::users;
59/// #     let connection = &mut establish_connection();
60/// use diesel::connection::DefaultLoadingMode;
61/// {
62///     // scope to restrict the lifetime of the iterator
63///     let iter1 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
64///
65///     for r in iter1 {
66///         let (id, name) = r?;
67///         println!("Id: {} Name: {}", id, name);
68///     }
69/// }
70///
71/// // works without specifying the loading mode
72/// let iter2 = users::table.load_iter::<(i32, String), _>(connection)?;
73///
74/// for r in iter2 {
75///     let (id, name) = r?;
76///     println!("Id: {} Name: {}", id, name);
77/// }
78/// #   Ok(())
79/// # }
80/// ```
81///
82/// This mode does **not support** creating
83/// multiple iterators using the same connection.
84///
85/// ```compile_fail,dejadoc
86/// # include!("../../doctest_setup.rs");
87/// #
88/// # fn main() {
89/// #     run_test().unwrap();
90/// # }
91/// #
92/// # fn run_test() -> QueryResult<()> {
93/// #     use schema::users;
94/// #     let connection = &mut establish_connection();
95/// use diesel::connection::DefaultLoadingMode;
96///
97/// let iter1 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
98/// let iter2 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
99///
100/// for r in iter1 {
101///     let (id, name) = r?;
102///     println!("Id: {} Name: {}", id, name);
103/// }
104///
105/// for r in iter2 {
106///     let (id, name) = r?;
107///     println!("Id: {} Name: {}", id, name);
108/// }
109/// #   Ok(())
110/// # }
111/// ```
112pub struct MysqlLikeConnection<DB: MysqlLikeBackend> {
113    raw_connection: RawConnection,
114    transaction_state: AnsiTransactionManager,
115    statement_cache: StatementCache<DB, Statement<DB>>,
116    instrumentation: DynInstrumentation,
117}
118
119// mysql connection can be shared between threads according to libmysqlclients documentation
120#[allow(unsafe_code)]
121unsafe impl<DB: MysqlLikeBackend> Send for MysqlLikeConnection<DB> {}
122
123impl<DB: MysqlLikeBackend> SimpleConnection for MysqlLikeConnection<DB> {
124    fn batch_execute(&mut self, query: &str) -> QueryResult<()> {
125        self.instrumentation
126            .on_connection_event(InstrumentationEvent::StartQuery {
127                query: &StrQueryHelper::new(query),
128            });
129        let r = self
130            .raw_connection
131            .enable_multi_statements(|| self.raw_connection.execute(query));
132        self.instrumentation
133            .on_connection_event(InstrumentationEvent::FinishQuery {
134                query: &StrQueryHelper::new(query),
135                error: r.as_ref().err(),
136            });
137        r
138    }
139}
140
141impl<DB: MysqlLikeBackend> ConnectionSealed for MysqlLikeConnection<DB> {}
142
143impl<DB: MysqlLikeBackend> Connection for MysqlLikeConnection<DB> {
144    type Backend = DB;
145    type TransactionManager = AnsiTransactionManager;
146
147    /// Establishes a new connection to the MySQL database
148    /// `database_url` may be enhanced by GET parameters
149    /// `mysql://[user[:password]@]host[:port]/database_name[?unix_socket=socket-path&ssl_mode=SSL_MODE*&ssl_ca=/etc/ssl/certs/ca-certificates.crt&ssl_cert=/etc/ssl/certs/client-cert.crt&ssl_key=/etc/ssl/certs/client-key.crt&local_infile=true]`
150    ///
151    /// * `host` can be an IP address or a hostname. If it is set to `localhost`, a connection
152    ///   will be attempted through the socket at `/tmp/mysql.sock`. If you want to connect to
153    ///   a local server via TCP (e.g. docker containers), use `0.0.0.0` or `127.0.0.1` instead.
154    /// * `unix_socket` expects the path to the unix socket
155    /// * `ssl_ca` accepts a path to the system's certificate roots
156    /// * `ssl_cert` accepts a path to the client's certificate file
157    /// * `ssl_key` accepts a path to the client's private key file
158    /// * `ssl_mode` expects a value defined for MySQL client command option `--ssl-mode`
159    ///   See <https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#option_general_ssl-mode>
160    /// * `local_infile` expects a boolean to enable or disable LOAD DATA LOCAL
161    fn establish(database_url: &str) -> ConnectionResult<Self> {
162        let mut instrumentation = DynInstrumentation::default_instrumentation();
163        instrumentation.on_connection_event(InstrumentationEvent::StartEstablishConnection {
164            url: database_url,
165        });
166
167        let establish_result = Self::establish_inner(database_url);
168        instrumentation.on_connection_event(InstrumentationEvent::FinishEstablishConnection {
169            url: database_url,
170            error: establish_result.as_ref().err(),
171        });
172        let mut conn = establish_result?;
173        conn.instrumentation = instrumentation;
174        Ok(conn)
175    }
176
177    fn execute_returning_count<T>(&mut self, source: &T) -> QueryResult<usize>
178    where
179        T: QueryFragment<Self::Backend> + QueryId,
180    {
181        #[allow(unsafe_code)] // call to unsafe function
182        update_transaction_manager_status(
183            prepared_query(
184                &source,
185                &mut self.statement_cache,
186                &mut self.raw_connection,
187                &mut *self.instrumentation,
188            )
189            .and_then(|stmt| {
190                // we have not called result yet, so calling `execute` is
191                // fine
192                let stmt_use = unsafe { stmt.execute() }?;
193                stmt_use.affected_rows()
194            }),
195            &mut self.transaction_state,
196            &mut self.instrumentation,
197            &crate::debug_query(source),
198        )
199    }
200
201    fn transaction_state(&mut self) -> &mut AnsiTransactionManager {
202        &mut self.transaction_state
203    }
204
205    fn instrumentation(&mut self) -> &mut dyn Instrumentation {
206        &mut *self.instrumentation
207    }
208
209    fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
210        self.instrumentation = instrumentation.into();
211    }
212
213    fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
214        self.statement_cache.set_cache_size(size);
215    }
216}
217
218#[inline(always)]
219fn update_transaction_manager_status<T>(
220    query_result: QueryResult<T>,
221    transaction_manager: &mut AnsiTransactionManager,
222    instrumentation: &mut DynInstrumentation,
223    query: &dyn DebugQuery,
224) -> QueryResult<T> {
225    fn non_generic_inner(
226        query_result: Result<(), &Error>,
227        transaction_manager: &mut AnsiTransactionManager,
228        instrumentation: &mut DynInstrumentation,
229        query: &dyn DebugQuery,
230    ) {
231        if let Err(Error::DatabaseError(DatabaseErrorKind::SerializationFailure, _)) = query_result
232        {
233            transaction_manager
234                .status
235                .set_requires_rollback_maybe_up_to_top_level(true)
236        }
237        instrumentation.on_connection_event(InstrumentationEvent::FinishQuery {
238            query,
239            error: query_result.err(),
240        });
241    }
242
243    non_generic_inner(
244        query_result.as_ref().map(|_| ()),
245        transaction_manager,
246        instrumentation,
247        query,
248    );
249    query_result
250}
251
252impl<DB: MysqlLikeBackend> LoadConnection<DefaultLoadingMode> for MysqlLikeConnection<DB> {
253    type Cursor<'conn, 'query> = self::stmt::iterator::StatementIterator<'conn, DB>;
254    type Row<'conn, 'query> = self::stmt::iterator::MysqlRow<DB>;
255
256    fn load<'conn, 'query, T>(
257        &'conn mut self,
258        source: T,
259    ) -> QueryResult<Self::Cursor<'conn, 'query>>
260    where
261        T: Query + QueryFragment<Self::Backend> + QueryId + 'query,
262        Self::Backend: QueryMetadata<T::SqlType>,
263    {
264        update_transaction_manager_status(
265            prepared_query(
266                &source,
267                &mut self.statement_cache,
268                &mut self.raw_connection,
269                &mut *self.instrumentation,
270            )
271            .and_then(|stmt| {
272                let mut metadata = Vec::new();
273                DB::row_metadata(&mut (), &mut metadata);
274                StatementIterator::from_stmt(stmt, &metadata)
275            }),
276            &mut self.transaction_state,
277            &mut self.instrumentation,
278            &crate::debug_query(&source),
279        )
280    }
281}
282
283#[cfg(feature = "r2d2")]
284impl<DB: MysqlLikeBackend> crate::r2d2::R2D2Connection for MysqlLikeConnection<DB> {
285    fn ping(&mut self) -> QueryResult<()> {
286        crate::r2d2::CheckConnectionQuery.execute(self).map(|_| ())
287    }
288
289    fn is_broken(&mut self) -> bool {
290        AnsiTransactionManager::is_broken_transaction_manager(self)
291    }
292}
293
294impl<DB: MysqlLikeBackend> MultiConnectionHelper for MysqlLikeConnection<DB> {
295    fn to_any<'a>(
296        lookup: &mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup,
297    ) -> &mut (dyn core::any::Any + 'a) {
298        lookup
299    }
300
301    fn from_any(
302        lookup: &mut dyn core::any::Any,
303    ) -> Option<&mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup> {
304        lookup.downcast_mut()
305    }
306}
307
308fn prepared_query<'a, DB: MysqlLikeBackend + Default, T: QueryFragment<DB> + QueryId>(
309    source: &'_ T,
310    statement_cache: &'a mut StatementCache<DB, Statement<DB>>,
311    raw_connection: &'a mut RawConnection,
312    instrumentation: &mut dyn Instrumentation,
313) -> QueryResult<MaybeCached<'a, Statement<DB>>> {
314    instrumentation.on_connection_event(InstrumentationEvent::StartQuery {
315        query: &crate::debug_query(source),
316    });
317    let mut stmt = statement_cache.cached_statement(
318        source,
319        &DB::default(),
320        &[],
321        &*raw_connection,
322        RawConnection::prepare,
323        instrumentation,
324    )?;
325
326    let mut bind_collector = RawBytesBindCollector::new();
327    source.collect_binds(&mut bind_collector, &mut (), &DB::default())?;
328    let binds = bind_collector
329        .metadata
330        .into_iter()
331        .zip(bind_collector.binds);
332    stmt.bind(binds)?;
333    Ok(stmt)
334}
335
336impl<DB: MysqlLikeBackend> MysqlLikeConnection<DB> {
337    /// Returns the version of the server this connection is connected to.
338    ///
339    /// # Errors
340    ///
341    /// Fails with [`Error::DeserializationError`] if the server reports a version it cannot parse.
342    ///
343    /// # Example
344    ///
345    /// ```rust
346    /// # include!("../../doctest_setup.rs");
347    /// use diesel::connection::DatabaseVersion;
348    /// #
349    /// # fn main() {
350    /// #     run_test().unwrap();
351    /// # }
352    /// #
353    /// # fn run_test() -> QueryResult<()> {
354    /// #     let conn = &mut establish_connection();
355    /// let supports_window_functions = conn.version()? >= DatabaseVersion::new(8, 0, 0);
356    /// #     Ok(())
357    /// # }
358    /// ```
359    pub fn version(&self) -> QueryResult<DatabaseVersion> {
360        let info = self.raw_connection.server_info().ok_or_else(|| {
361            Error::DeserializationError("The server did not report its version".into())
362        })?;
363        parse_server_version(info.to_bytes()).ok_or_else(|| {
364            Error::DeserializationError(
365                ::alloc::__export::must_use({
        ::alloc::fmt::format(format_args!("Unrecognized server version `{0}`",
                info.to_string_lossy()))
    })format!("Unrecognized server version `{}`", info.to_string_lossy()).into(),
366            )
367        })
368    }
369
370    /// Executes `source` and returns its `mysql_stmt_insert_id`, zero mapped
371    /// to `None`. Public entry point: [`InsertStatement::execute_returning_id`].
372    pub(crate) fn execute_returning_id<T>(&mut self, source: &T) -> QueryResult<Option<NonZeroU64>>
373    where
374        T: QueryFragment<DB> + QueryId,
375    {
376        #[allow(unsafe_code)] // call to unsafe function
377        update_transaction_manager_status(
378            prepared_query(
379                &source,
380                &mut self.statement_cache,
381                &mut self.raw_connection,
382                &mut *self.instrumentation,
383            )
384            .and_then(|stmt| {
385                // SAFETY: `prepared_query` returned this statement freshly
386                // bound, so no result set is pending, which is `execute`'s
387                // requirement.
388                let stmt_use = unsafe { stmt.execute() }?;
389                Ok(NonZeroU64::new(stmt_use.insert_id()))
390            }),
391            &mut self.transaction_state,
392            &mut self.instrumentation,
393            &crate::debug_query(source),
394        )
395    }
396
397    fn set_config_options(&mut self) -> QueryResult<()> {
398        crate::sql_query("SET time_zone = '+00:00';").execute(self)?;
399        crate::sql_query("SET character_set_client = 'utf8mb4'").execute(self)?;
400        crate::sql_query("SET character_set_connection = 'utf8mb4'").execute(self)?;
401        crate::sql_query("SET character_set_results = 'utf8mb4'").execute(self)?;
402        Ok(())
403    }
404
405    fn establish_inner(database_url: &str) -> Result<MysqlLikeConnection<DB>, ConnectionError> {
406        use crate::ConnectionError::CouldntSetupConfiguration;
407
408        let raw_connection = RawConnection::new();
409        let connection_options = ConnectionOptions::parse::<DB>(database_url)?;
410        raw_connection.connect(&connection_options)?;
411        let mut conn = MysqlLikeConnection {
412            raw_connection,
413            transaction_state: AnsiTransactionManager::default(),
414            statement_cache: StatementCache::new(),
415            instrumentation: DynInstrumentation::none(),
416        };
417        conn.set_config_options()
418            .map_err(CouldntSetupConfiguration)?;
419        Ok(conn)
420    }
421}
422
423/// MariaDB before 11.0 prefixes its version with this, which only MariaDB Connector/C strips.
424const MARIADB_REPLICATION_VERSION_PREFIX: &[u8] = b"5.5.5-";
425
426fn parse_server_version(info: &[u8]) -> Option<DatabaseVersion> {
427    let info = match info.strip_prefix(MARIADB_REPLICATION_VERSION_PREFIX) {
428        Some(version) if version.first().is_some_and(u8::is_ascii_digit) => version,
429        _ => info,
430    };
431    let end = info
432        .iter()
433        .position(|&byte| !byte.is_ascii_digit() && byte != b'.')
434        .unwrap_or(info.len());
435    core::str::from_utf8(&info[..end]).ok()?.parse().ok()
436}
437
438#[cfg(test)]
439mod tests {
440    use super::*;
441
442    #[diesel_test_helper::test]
443    fn parse_server_version_reads_mysql_and_mariadb_formats() {
444        let v = DatabaseVersion::new;
445        assert_eq!(parse_server_version(b"8.4.11"), Some(v(8, 4, 11)));
446        assert_eq!(
447            parse_server_version(b"8.0.36-0ubuntu0.22.04.1"),
448            Some(v(8, 0, 36))
449        );
450        assert_eq!(
451            parse_server_version(b"11.4.2-MariaDB-ubu2404"),
452            Some(v(11, 4, 2))
453        );
454        assert_eq!(
455            parse_server_version(b"5.5.5-10.11.6-MariaDB-log"),
456            Some(v(10, 11, 6))
457        );
458        assert_eq!(parse_server_version(b"5.5.5-log"), Some(v(5, 5, 5)));
459        assert_eq!(parse_server_version(b"8.4"), None);
460        assert_eq!(parse_server_version(b"8.4.11.2"), None);
461        assert_eq!(parse_server_version(b"8..4"), None);
462        assert_eq!(parse_server_version(b"8..4.11"), None);
463        assert_eq!(parse_server_version(b".8.4.11"), None);
464        assert_eq!(parse_server_version(b"8.4.11."), None);
465        assert_eq!(parse_server_version(b"8.4.4294967296"), None);
466        assert_eq!(parse_server_version(b"MariaDB"), None);
467    }
468}
diesel/mysql_like/connection/mod.rs

diesel/mysql_like/connection/
mod.rs
