sqlx_core_oldapi/odbc/connection/
mod.rs

1use crate::common::StatementCache;
2use crate::connection::{Connection, LogSettings};
3use crate::error::Error;
4use crate::odbc::{
5    Odbc, OdbcArguments, OdbcBufferSettings, OdbcColumn, OdbcConnectOptions, OdbcQueryResult,
6    OdbcRow, OdbcTypeInfo,
7};
8use crate::transaction::Transaction;
9use either::Either;
10use sqlx_rt::spawn_blocking;
11mod odbc_bridge;
12use crate::odbc::{OdbcStatement, OdbcStatementMetadata};
13use futures_core::future::BoxFuture;
14use futures_util::future;
15use odbc_api::{
16    handles::StatementConnection, ConnectionTransitions, Prepared, ResultSetMetadata,
17    SharedConnection,
18};
19use odbc_bridge::{establish_connection, execute_sql};
20use std::borrow::Cow;
21use std::sync::{Arc, Mutex};
22
23mod executor;
24
25type PreparedStatement = Prepared<StatementConnection<SharedConnection<'static>>>;
26type SharedPreparedStatement = Arc<Mutex<PreparedStatement>>;
27
28struct CollectedColumns {
29    columns: Vec<OdbcColumn>,
30    deferred: bool,
31}
32
33fn collect_columns(
34    prepared: &mut PreparedStatement,
35    parameter_count: usize,
36    allow_deferred_result_columns: bool,
37) -> Result<CollectedColumns, Error> {
38    let count = match prepared.num_result_cols() {
39        Ok(count) => count,
40        Err(error) if allow_deferred_result_columns && parameter_count > 0 => {
41            log::debug!("ODBC prepare deferred result columns until execution: {error}");
42            validate_parameter_metadata(prepared, parameter_count)?;
43            return Ok(CollectedColumns {
44                columns: Vec::new(),
45                deferred: true,
46            });
47        }
48        Err(error) => return Err(error.into()),
49    };
50
51    let mut columns = Vec::with_capacity(count as usize);
52    for i in 1..=count {
53        columns.push(describe_column(prepared, i as u16)?);
54    }
55    Ok(CollectedColumns {
56        columns,
57        deferred: false,
58    })
59}
60
61fn validate_parameter_metadata(
62    prepared: &mut PreparedStatement,
63    parameter_count: usize,
64) -> Result<(), Error> {
65    for index in 1..=parameter_count {
66        let parameter_number = u16::try_from(index)
67            .map_err(|_| Error::Protocol(format!("ODBC parameter index {index} exceeds u16")))?;
68        prepared.describe_param(parameter_number)?;
69    }
70    Ok(())
71}
72
73fn collect_statement_metadata(
74    prepared: &mut PreparedStatement,
75    allow_deferred_result_columns: bool,
76) -> Result<(OdbcStatementMetadata, bool), Error> {
77    let parameters = usize::from(prepared.num_params()?);
78    let collected = collect_columns(prepared, parameters, allow_deferred_result_columns)?;
79    let metadata_complete = !(collected.deferred || parameters > 0 && collected.columns.is_empty());
80
81    Ok((
82        OdbcStatementMetadata {
83            columns: collected.columns,
84            parameters,
85        },
86        metadata_complete,
87    ))
88}
89
90pub(super) fn describe_column<S>(stmt: &mut S, index: u16) -> Result<OdbcColumn, Error>
91where
92    S: ResultSetMetadata,
93{
94    let mut cd = odbc_api::ColumnDescription::default();
95    stmt.describe_col(index, &mut cd)?;
96
97    Ok(OdbcColumn {
98        name: decode_column_name(cd.name, index),
99        type_info: OdbcTypeInfo::new(cd.data_type),
100        ordinal: usize::from(
101            index
102                .checked_sub(1)
103                .ok_or_else(|| Error::Protocol("ODBC column indices are 1-based".into()))?,
104        ),
105    })
106}
107
108pub(super) trait ColumnNameDecode {
109    fn decode_or_default(self, index: u16) -> String;
110}
111
112impl ColumnNameDecode for Vec<u8> {
113    fn decode_or_default(self, index: u16) -> String {
114        String::from_utf8(self).unwrap_or_else(|_| format!("col{}", index - 1))
115    }
116}
117
118impl ColumnNameDecode for Vec<u16> {
119    fn decode_or_default(self, index: u16) -> String {
120        String::from_utf16(&self).unwrap_or_else(|_| format!("col{}", index - 1))
121    }
122}
123
124pub(super) fn decode_column_name<T: ColumnNameDecode>(name: T, index: u16) -> String {
125    name.decode_or_default(index)
126}
127
128/// A connection to an ODBC-accessible database.
129///
130/// ODBC uses a blocking C API, so we offload blocking calls to the runtime's blocking
131/// thread-pool via `spawn_blocking` and synchronize access with a mutex.
132pub struct OdbcConnection {
133    pub(crate) conn: SharedConnection<'static>,
134    pub(crate) stmt_cache: StatementCache<SharedPreparedStatement>,
135    pub(crate) buffer_settings: OdbcBufferSettings,
136    pub(crate) log_settings: LogSettings,
137}
138
139impl std::fmt::Debug for OdbcConnection {
140    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
141        f.debug_struct("OdbcConnection")
142            .field("conn", &self.conn)
143            .field("buffer_settings", &self.buffer_settings)
144            .finish()
145    }
146}
147
148impl OdbcConnection {
149    pub(crate) async fn with_conn<R, F, S>(&mut self, operation: S, f: F) -> Result<R, Error>
150    where
151        R: Send + 'static,
152        F: FnOnce(&mut odbc_api::Connection<'static>) -> Result<R, Error> + Send + 'static,
153        S: std::fmt::Display + Send + 'static,
154    {
155        let conn = Arc::clone(&self.conn);
156        spawn_blocking(move || {
157            let mut conn_guard = conn.lock().map_err(|_| {
158                Error::Protocol(format!("ODBC {}: failed to lock connection", operation))
159            })?;
160            f(&mut conn_guard)
161        })
162        .await
163    }
164
165    pub(crate) async fn establish(options: &OdbcConnectOptions) -> Result<Self, Error> {
166        let shared_conn = spawn_blocking({
167            let options = options.clone();
168            move || {
169                let conn = establish_connection(&options)?;
170                let shared_conn = odbc_api::SharedConnection::new(std::sync::Mutex::new(conn));
171                Ok::<_, Error>(shared_conn)
172            }
173        })
174        .await?;
175
176        Ok(Self {
177            conn: shared_conn,
178            stmt_cache: StatementCache::new(options.statement_cache_capacity),
179            buffer_settings: options.buffer_settings,
180            log_settings: options.log_settings.clone(),
181        })
182    }
183
184    pub(crate) async fn ping_blocking(&mut self) -> Result<(), Error> {
185        self.with_conn("ping", move |conn| {
186            conn.execute("SELECT 1", (), None)?;
187            Ok(())
188        })
189        .await
190    }
191
192    pub(crate) async fn begin_blocking(&mut self) -> Result<(), Error> {
193        self.with_conn("begin", move |conn| {
194            conn.set_autocommit(false)?;
195            Ok(())
196        })
197        .await
198    }
199
200    pub(crate) async fn commit_blocking(&mut self) -> Result<(), Error> {
201        self.with_conn("commit", move |conn| {
202            conn.commit()?;
203            conn.set_autocommit(true)?;
204            Ok(())
205        })
206        .await
207    }
208
209    pub(crate) async fn rollback_blocking(&mut self) -> Result<(), Error> {
210        self.with_conn("rollback", move |conn| {
211            conn.rollback()?;
212            conn.set_autocommit(true)?;
213            Ok(())
214        })
215        .await
216    }
217
218    /// Launches a background task to execute the SQL statement and send the results to the returned channel.
219    pub(crate) async fn execute_stream(
220        &mut self,
221        sql: &str,
222        args: Option<OdbcArguments>,
223        persistent: bool,
224    ) -> Result<flume::Receiver<Result<Either<OdbcQueryResult, OdbcRow>, Error>>, Error> {
225        let statement = if args.is_some() {
226            MaybePrepared::Prepared(self.prepared_statement(sql, persistent).await?)
227        } else {
228            MaybePrepared::NotPrepared(sql.to_string())
229        };
230
231        let (tx, rx) = flume::bounded(64);
232
233        let sql_owned = sql.to_string();
234        let conn = Arc::clone(&self.conn);
235        let buffer_settings = self.buffer_settings;
236        let log_settings = self.log_settings.clone();
237        sqlx_rt::spawn(sqlx_rt::spawn_blocking(move || {
238            let mut logger = crate::logger::QueryLogger::new(&sql_owned, log_settings);
239            let result = conn
240                .lock()
241                .map_err(|_| Error::Protocol("ODBC execute: failed to lock connection".into()))
242                .and_then(|mut conn| {
243                    execute_sql(
244                        &mut conn,
245                        statement,
246                        args,
247                        &tx,
248                        buffer_settings,
249                        &mut logger,
250                    )
251                });
252
253            if let Err(e) = result {
254                let _ = tx.send(Err(e));
255            }
256        }));
257
258        Ok(rx)
259    }
260
261    async fn prepared_statement(
262        &mut self,
263        sql: &str,
264        store_to_cache: bool,
265    ) -> Result<SharedPreparedStatement, Error> {
266        if let Some(cached) = self.stmt_cache.get_mut(sql) {
267            return Ok(Arc::clone(cached));
268        }
269
270        let conn = Arc::clone(&self.conn);
271        let sql_owned = sql.to_string();
272        let prepared =
273            spawn_blocking(move || conn.into_prepared(&sql_owned).map_err(Error::from)).await?;
274        let prepared = Arc::new(Mutex::new(prepared));
275
276        if store_to_cache && self.stmt_cache.is_enabled() {
277            self.stmt_cache.insert(sql, Arc::clone(&prepared));
278        }
279
280        Ok(prepared)
281    }
282
283    pub(crate) async fn clear_cached_statements(&mut self) -> Result<(), Error> {
284        while self.stmt_cache.remove_lru().is_some() {}
285        Ok(())
286    }
287
288    async fn prepare_with_metadata_policy<'a>(
289        &mut self,
290        sql: &'a str,
291        store_to_cache: bool,
292        allow_deferred_result_columns: bool,
293    ) -> Result<OdbcStatement<'a>, Error> {
294        let prepared = self.prepared_statement(sql, false).await?;
295
296        let (metadata, metadata_complete) = spawn_blocking({
297            let prepared = Arc::clone(&prepared);
298            move || {
299                let mut prepared = prepared.lock().map_err(|_| {
300                    Error::Protocol("ODBC prepare: failed to lock prepared statement".into())
301                })?;
302                collect_statement_metadata(&mut prepared, allow_deferred_result_columns)
303            }
304        })
305        .await?;
306
307        if !allow_deferred_result_columns && !metadata_complete {
308            return Err(Error::Protocol(
309                "ODBC driver did not provide result-column metadata before execution".into(),
310            ));
311        }
312
313        if store_to_cache && metadata_complete && self.stmt_cache.is_enabled() {
314            self.stmt_cache.insert(sql, prepared);
315        }
316
317        Ok(OdbcStatement {
318            sql: Cow::Borrowed(sql),
319            metadata,
320        })
321    }
322
323    pub async fn prepare<'a>(&mut self, sql: &'a str) -> Result<OdbcStatement<'a>, Error> {
324        self.prepare_with_metadata_policy(sql, true, true).await
325    }
326
327    pub(crate) async fn describe_statement<'a>(
328        &mut self,
329        sql: &'a str,
330    ) -> Result<OdbcStatement<'a>, Error> {
331        self.prepare_with_metadata_policy(sql, false, false).await
332    }
333}
334
335pub(crate) enum MaybePrepared {
336    Prepared(SharedPreparedStatement),
337    NotPrepared(String),
338}
339
340impl std::fmt::Debug for MaybePrepared {
341    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
342        match self {
343            MaybePrepared::Prepared(_) => f.debug_tuple("Prepared").finish(),
344            MaybePrepared::NotPrepared(sql) => f.debug_tuple("NotPrepared").field(sql).finish(),
345        }
346    }
347}
348
349impl Connection for OdbcConnection {
350    type Database = Odbc;
351
352    type Options = OdbcConnectOptions;
353
354    fn close(self) -> BoxFuture<'static, Result<(), Error>> {
355        Box::pin(async move {
356            // Drop connection by moving Arc and letting it fall out of scope.
357            drop(self);
358            Ok(())
359        })
360    }
361
362    fn close_hard(self) -> BoxFuture<'static, Result<(), Error>> {
363        Box::pin(async move { Ok(()) })
364    }
365
366    fn ping(&mut self) -> BoxFuture<'_, Result<(), Error>> {
367        Box::pin(self.ping_blocking())
368    }
369
370    fn begin(&mut self) -> BoxFuture<'_, Result<Transaction<'_, Self::Database>, Error>>
371    where
372        Self: Sized,
373    {
374        Transaction::begin(self)
375    }
376
377    fn cached_statements_size(&self) -> usize {
378        self.stmt_cache.len()
379    }
380
381    #[doc(hidden)]
382    fn flush(&mut self) -> BoxFuture<'_, Result<(), Error>> {
383        Box::pin(future::ok(()))
384    }
385
386    #[doc(hidden)]
387    fn should_flush(&self) -> bool {
388        false
389    }
390
391    fn clear_cached_statements(&mut self) -> BoxFuture<'_, Result<(), Error>> {
392        Box::pin(self.clear_cached_statements())
393    }
394
395    fn dbms_name(&mut self) -> BoxFuture<'_, Result<String, Error>> {
396        Box::pin(async move {
397            self.with_conn("dbms_name", move |conn| {
398                Ok(conn.database_management_system_name()?)
399            })
400            .await
401        })
402    }
403}
sqlx_core_oldapi/odbc/connection/mod.rs

sqlx_core_oldapi/odbc/connection/
mod.rs
