feat: Add ODBC driver support

This commit introduces the ODBC driver for SQLx, enabling database connectivity via ODBC.

Co-authored-by: contact <[email protected]>

Author: cursoragent

.github/workflows/sqlx.yml

@@ -33,20 +33,20 @@ jobs:
         run: |
           cargo clippy --manifest-path sqlx-core/Cargo.toml \
             --no-default-features \
-            --features offline,all-databases,all-types,migrate,runtime-${{ matrix.runtime }}-${{ matrix.tls }} \
+            --features offline,all-databases,all-types,migrate,odbc,runtime-${{ matrix.runtime }}-${{ matrix.tls }} \
             -- -D warnings
       - name: Run clippy for root with all features
         run: |
           cargo clippy \
             --no-default-features \
-            --features offline,all-databases,all-types,migrate,runtime-${{ matrix.runtime }}-${{ matrix.tls }},macros \
+            --features offline,all-databases,all-types,migrate,odbc,runtime-${{ matrix.runtime }}-${{ matrix.tls }},macros \
             -- -D warnings
       - name: Run clippy for all targets
         run: |
           cargo clippy \
             --no-default-features \
             --all-targets \
-            --features offline,all-databases,migrate,runtime-${{ matrix.runtime }}-${{ matrix.tls }} \
+            --features offline,all-databases,migrate,odbc,runtime-${{ matrix.runtime }}-${{ matrix.tls }} \
             -- -D warnings
 
   test:
@@ -74,7 +74,7 @@ jobs:
       - run:
           cargo test
             --manifest-path sqlx-core/Cargo.toml
-            --features offline,all-databases,all-types,runtime-${{ matrix.runtime }}-${{ matrix.tls }}
+            --features offline,all-databases,all-types,odbc,runtime-${{ matrix.runtime }}-${{ matrix.tls }}
 
   cli:
     name: CLI Binaries

sqlx-core/src/odbc/column.rs

@@ -11,13 +11,19 @@ pub struct OdbcColumn {
 impl Column for OdbcColumn {
     type Database = Odbc;
 
-    fn ordinal(&self) -> usize { self.ordinal }
-    fn name(&self) -> &str { &self.name }
-    fn type_info(&self) -> &OdbcTypeInfo { &self.type_info }
+    fn ordinal(&self) -> usize {
+        self.ordinal
+    }
+    fn name(&self) -> &str {
+        &self.name
+    }
+    fn type_info(&self) -> &OdbcTypeInfo {
+        &self.type_info
+    }
 }
 
 mod private {
-    use crate::column::private_column::Sealed;
     use super::OdbcColumn;
+    use crate::column::private_column::Sealed;
     impl Sealed for OdbcColumn {}
 }

sqlx-core/src/odbc/connection/executor.rs

@@ -2,63 +2,18 @@ use crate::describe::Describe;
 use crate::error::Error;
 use crate::executor::{Execute, Executor};
 use crate::logger::QueryLogger;
-use crate::odbc::{Odbc, OdbcColumn, OdbcConnection, OdbcQueryResult, OdbcRow, OdbcStatement, OdbcTypeInfo};
+use crate::odbc::{
+    Odbc, OdbcColumn, OdbcConnection, OdbcQueryResult, OdbcRow, OdbcStatement, OdbcTypeInfo,
+};
 use either::Either;
 use futures_core::future::BoxFuture;
 use futures_core::stream::BoxStream;
 use futures_util::TryStreamExt;
-use std::pin::Pin;
 use odbc_api::Cursor;
 use std::borrow::Cow;
+use std::pin::Pin;
 
-impl OdbcConnection {
-    async fn run<'e, 'c: 'e>(
-        &'c mut self,
-        sql: &'e str,
-    ) -> Result<impl futures_core::Stream<Item = Result<Either<OdbcQueryResult, OdbcRow>, Error>> + 'e, Error> {
-        let mut logger = QueryLogger::new(sql, self.log_settings.clone());
-
-        Ok(Box::pin(try_stream! {
-            let guard = self.worker.shared.conn.lock().await;
-            match guard.execute(sql, (), None) {
-                Ok(Some(mut cursor)) => {
-                    use odbc_api::ResultSetMetadata;
-                    let mut columns = Vec::new();
-                    if let Ok(count) = cursor.num_result_cols() {
-                        for i in 1..=count { // ODBC columns are 1-based
-                            let mut cd = odbc_api::ColumnDescription::default();
-                            let _ = cursor.describe_col(i as u16, &mut cd);
-                            let name = String::from_utf8(cd.name).unwrap_or_else(|_| format!("col{}", i-1));
-                            columns.push(OdbcColumn { name, type_info: OdbcTypeInfo { name: format!("{:?}", cd.data_type), is_null: false }, ordinal: (i-1) as usize });
-                        }
-                    }
-                    while let Some(mut row) = cursor.next_row().map_err(|e| Error::from(e))? {
-                        let mut values = Vec::with_capacity(columns.len());
-                        for i in 1..=columns.len() {
-                            let mut buf = Vec::new();
-                            let not_null = row.get_text(i as u16, &mut buf).map_err(|e| Error::from(e))?;
-                            if not_null {
-                                let ti = OdbcTypeInfo { name: "TEXT".into(), is_null: false };
-                                values.push((ti, Some(buf)));
-                            } else {
-                                let ti = OdbcTypeInfo { name: "TEXT".into(), is_null: true };
-                                values.push((ti, None));
-                            }
-                        }
-                        logger.increment_rows_returned();
-                        r#yield!(Either::Right(OdbcRow { columns: columns.clone(), values }));
-                    }
-                    r#yield!(Either::Left(OdbcQueryResult { rows_affected: 0 }));
-                }
-                Ok(None) => {
-                    r#yield!(Either::Left(OdbcQueryResult { rows_affected: 0 }));
-                }
-                Err(e) => return Err(Error::from(e)),
-            }
-            Ok(())
-        }))
-    }
-}
+// run method removed; fetch_many implements streaming directly
 
 impl<'c> Executor<'c> for &'c mut OdbcConnection {
     type Database = Odbc;
@@ -127,7 +82,9 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
         let mut s = self.fetch_many(query);
         Box::pin(async move {
             while let Some(v) = s.try_next().await? {
-                if let Either::Right(r) = v { return Ok(Some(r)); }
+                if let Either::Right(r) = v {
+                    return Ok(Some(r));
+                }
             }
             Ok(None)
         })
@@ -143,7 +100,11 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
     {
         Box::pin(async move {
             // Basic statement metadata: no parameter/column info without executing
-            Ok(OdbcStatement { sql: Cow::Borrowed(sql), columns: Vec::new(), parameters: 0 })
+            Ok(OdbcStatement {
+                sql: Cow::Borrowed(sql),
+                columns: Vec::new(),
+                parameters: 0,
+            })
         })
     }
 

sqlx-core/src/odbc/connection/mod.rs

@@ -1,12 +1,12 @@
 use crate::connection::{Connection, LogSettings};
 use crate::error::Error;
-use crate::transaction::Transaction;
 use crate::odbc::{Odbc, OdbcConnectOptions};
+use crate::transaction::Transaction;
 use futures_core::future::BoxFuture;
 use futures_util::future;
 
-mod worker;
 mod executor;
+mod worker;
 
 pub(crate) use worker::ConnectionWorker;
 
@@ -23,7 +23,10 @@ pub struct OdbcConnection {
 impl OdbcConnection {
     pub(crate) async fn establish(options: &OdbcConnectOptions) -> Result<Self, Error> {
         let worker = ConnectionWorker::establish(options.clone()).await?;
-        Ok(Self { worker, log_settings: LogSettings::default() })
+        Ok(Self {
+            worker,
+            log_settings: LogSettings::default(),
+        })
     }
 }
 

sqlx-core/src/odbc/connection/worker.rs

@@ -19,11 +19,21 @@ pub(crate) struct Shared {
 }
 
 enum Command {
-    Ping { tx: oneshot::Sender<()> },
-    Shutdown { tx: oneshot::Sender<()> },
-    Begin { tx: oneshot::Sender<Result<(), Error>> },
-    Commit { tx: oneshot::Sender<Result<(), Error>> },
-    Rollback { tx: oneshot::Sender<Result<(), Error>> },
+    Ping {
+        tx: oneshot::Sender<()>,
+    },
+    Shutdown {
+        tx: oneshot::Sender<()>,
+    },
+    Begin {
+        tx: oneshot::Sender<Result<(), Error>>,
+    },
+    Commit {
+        tx: oneshot::Sender<Result<(), Error>>,
+    },
+    Rollback {
+        tx: oneshot::Sender<Result<(), Error>>,
+    },
 }
 
 impl ConnectionWorker {
@@ -39,18 +49,25 @@ impl ConnectionWorker {
                 // to 'static, as ODBC connection borrows it. This is acceptable for long-lived
                 // process and mirrors SQLite approach to background workers.
                 let env = Box::leak(Box::new(odbc_api::Environment::new().unwrap()));
-                let conn = match env.connect_with_connection_string(options.connection_string(), Default::default()) {
+                let conn = match env
+                    .connect_with_connection_string(options.connection_string(), Default::default())
+                {
                     Ok(c) => c,
                     Err(e) => {
                         let _ = establish_tx.send(Err(Error::Configuration(e.to_string().into())));
                         return;
                     }
                 };
 
-                let shared = Arc::new(Shared { conn: Mutex::new(conn, true) });
+                let shared = Arc::new(Shared {
+                    conn: Mutex::new(conn, true),
+                });
 
                 if establish_tx
-                    .send(Ok(Self { command_tx: tx.clone(), shared: Arc::clone(&shared) }))
+                    .send(Ok(Self {
+                        command_tx: tx.clone(),
+                        shared: Arc::clone(&shared),
+                    }))
                     .is_err()
                 {
                     return;
@@ -67,20 +84,35 @@ impl ConnectionWorker {
                         }
                         Command::Begin { tx } => {
                             let res = if let Some(mut guard) = shared.conn.try_lock() {
-                                match guard.execute("BEGIN", (), None) { Ok(_) => Ok(()), Err(e) => Err(Error::Configuration(e.to_string().into())) }
-                            } else { Ok(()) };
+                                match guard.execute("BEGIN", (), None) {
+                                    Ok(_) => Ok(()),
+                                    Err(e) => Err(Error::Configuration(e.to_string().into())),
+                                }
+                            } else {
+                                Ok(())
+                            };
                             let _ = tx.send(res);
                         }
                         Command::Commit { tx } => {
                             let res = if let Some(mut guard) = shared.conn.try_lock() {
-                                match guard.execute("COMMIT", (), None) { Ok(_) => Ok(()), Err(e) => Err(Error::Configuration(e.to_string().into())) }
-                            } else { Ok(()) };
+                                match guard.execute("COMMIT", (), None) {
+                                    Ok(_) => Ok(()),
+                                    Err(e) => Err(Error::Configuration(e.to_string().into())),
+                                }
+                            } else {
+                                Ok(())
+                            };
                             let _ = tx.send(res);
                         }
                         Command::Rollback { tx } => {
                             let res = if let Some(mut guard) = shared.conn.try_lock() {
-                                match guard.execute("ROLLBACK", (), None) { Ok(_) => Ok(()), Err(e) => Err(Error::Configuration(e.to_string().into())) }
-                            } else { Ok(()) };
+                                match guard.execute("ROLLBACK", (), None) {
+                                    Ok(_) => Ok(()),
+                                    Err(e) => Err(Error::Configuration(e.to_string().into())),
+                                }
+                            } else {
+                                Ok(())
+                            };
                             let _ = tx.send(res);
                         }
                         Command::Shutdown { tx } => {

sqlx-core/src/odbc/error.rs

@@ -7,19 +7,33 @@ use std::fmt::{Display, Formatter, Result as FmtResult};
 pub struct OdbcDatabaseError(pub OdbcApiError);
 
 impl Display for OdbcDatabaseError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult { Display::fmt(&self.0, f) }
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        Display::fmt(&self.0, f)
+    }
 }
 
 impl std::error::Error for OdbcDatabaseError {}
 
 impl DatabaseError for OdbcDatabaseError {
-    fn message(&self) -> &str { "ODBC error" }
-    fn code(&self) -> Option<Cow<'_, str>> { None }
-    fn as_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) { self }
-    fn as_error_mut(&mut self) -> &mut (dyn std::error::Error + Send + Sync + 'static) { self }
-    fn into_error(self: Box<Self>) -> Box<dyn std::error::Error + Send + Sync + 'static> { self }
+    fn message(&self) -> &str {
+        "ODBC error"
+    }
+    fn code(&self) -> Option<Cow<'_, str>> {
+        None
+    }
+    fn as_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) {
+        self
+    }
+    fn as_error_mut(&mut self) -> &mut (dyn std::error::Error + Send + Sync + 'static) {
+        self
+    }
+    fn into_error(self: Box<Self>) -> Box<dyn std::error::Error + Send + Sync + 'static> {
+        self
+    }
 }
 
 impl From<OdbcApiError> for crate::error::Error {
-    fn from(value: OdbcApiError) -> Self { crate::error::Error::Database(Box::new(OdbcDatabaseError(value))) }
+    fn from(value: OdbcApiError) -> Self {
+        crate::error::Error::Database(Box::new(OdbcDatabaseError(value)))
+    }
 }

sqlx-core/src/odbc/mod.rs

@@ -2,30 +2,30 @@
 
 use crate::executor::Executor;
 
+mod arguments;
+mod column;
 mod connection;
 mod database;
+mod error;
+mod options;
+mod query_result;
 mod row;
-mod column;
-mod value;
-mod type_info;
 mod statement;
-mod query_result;
 mod transaction;
-mod options;
-mod error;
-mod arguments;
+mod type_info;
+mod value;
 
+pub use arguments::{OdbcArgumentValue, OdbcArguments};
+pub use column::OdbcColumn;
 pub use connection::OdbcConnection;
 pub use database::Odbc;
 pub use options::OdbcConnectOptions;
 pub use query_result::OdbcQueryResult;
 pub use row::OdbcRow;
-pub use column::OdbcColumn;
 pub use statement::OdbcStatement;
 pub use transaction::OdbcTransactionManager;
 pub use type_info::OdbcTypeInfo;
 pub use value::{OdbcValue, OdbcValueRef};
-pub use arguments::{OdbcArguments, OdbcArgumentValue};
 
 /// An alias for [`Pool`][crate::pool::Pool], specialized for ODBC.
 pub type OdbcPool = crate::pool::Pool<Odbc>;

sqlx-core/src/odbc/options/mod.rs

@@ -15,12 +15,16 @@ pub struct OdbcConnectOptions {
 }
 
 impl OdbcConnectOptions {
-    pub fn connection_string(&self) -> &str { &self.conn_str }
+    pub fn connection_string(&self) -> &str {
+        &self.conn_str
+    }
 }
 
 impl Debug for OdbcConnectOptions {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        f.debug_struct("OdbcConnectOptions").field("conn_str", &"<redacted>").finish()
+        f.debug_struct("OdbcConnectOptions")
+            .field("conn_str", &"<redacted>")
+            .finish()
     }
 }
 
@@ -29,7 +33,10 @@ impl FromStr for OdbcConnectOptions {
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         // Use full string as ODBC connection string or DSN
-        Ok(Self { conn_str: s.to_owned(), log_settings: LogSettings::default() })
+        Ok(Self {
+            conn_str: s.to_owned(),
+            log_settings: LogSettings::default(),
+        })
     }
 }