feat: Enhance ODBC execution with parameterized SQL handling

This commit introduces a new function for executing SQL commands with parameters in the ODBC connection worker. It improves the handling of SQL execution by allowing for parameterized queries, enhancing flexibility and security. Additionally, it refines the argument conversion process for better integration with the ODBC API.

Author: lovasoa

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

@@ -26,9 +26,19 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
         let sql = _query.sql().to_string();
         let mut args = _query.take_arguments();
         Box::pin(try_stream! {
-            let rx = if let Some(a) = args.take() {
-                let new_sql = interpolate_sql_with_odbc_args(&sql, &a.values);
-                self.worker.execute_stream(&new_sql).await?
+            let rx = if let Some(mut a) = args.take() {
+                let vals: Vec<OdbcArgumentValue<'static>> = std::mem::take(&mut a.values)
+                    .into_iter()
+                    .map(|v| match v {
+                        OdbcArgumentValue::Text(s) => OdbcArgumentValue::Text(s),
+                        OdbcArgumentValue::Bytes(b) => OdbcArgumentValue::Bytes(b),
+                        OdbcArgumentValue::Int(i) => OdbcArgumentValue::Int(i),
+                        OdbcArgumentValue::Float(f) => OdbcArgumentValue::Float(f),
+                        OdbcArgumentValue::Null => OdbcArgumentValue::Null,
+                        OdbcArgumentValue::Phantom(_) => OdbcArgumentValue::Null,
+                    })
+                    .collect();
+                self.worker.execute_stream_with_args(&sql, vals).await?
             } else {
                 self.worker.execute_stream(&sql).await?
             };
@@ -84,43 +94,3 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
         Box::pin(async move { Err(Error::Protocol("ODBC describe not implemented".into())) })
     }
 }
-
-fn interpolate_sql_with_odbc_args(sql: &str, args: &[OdbcArgumentValue<'_>]) -> String {
-    let mut result = String::with_capacity(sql.len() + args.len() * 8);
-    let mut arg_iter = args.iter();
-    for ch in sql.chars() {
-        if ch == '?' {
-            if let Some(arg) = arg_iter.next() {
-                match arg {
-                    OdbcArgumentValue::Int(i) => result.push_str(&i.to_string()),
-                    OdbcArgumentValue::Float(f) => result.push_str(&format!("{}", f)),
-                    OdbcArgumentValue::Text(s) => {
-                        result.push('\'');
-                        for c in s.chars() {
-                            if c == '\'' {
-                                result.push('\'');
-                            }
-                            result.push(c);
-                        }
-                        result.push('\'');
-                    }
-                    OdbcArgumentValue::Bytes(b) => {
-                        result.push_str("X'");
-                        for byte in b {
-                            result.push_str(&format!("{:02X}", byte));
-                        }
-                        result.push('\'');
-                    }
-                    OdbcArgumentValue::Null | OdbcArgumentValue::Phantom(_) => {
-                        result.push_str("NULL")
-                    }
-                }
-            } else {
-                result.push('?');
-            }
-        } else {
-            result.push(ch);
-        }
-    }
-    result
-}

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

@@ -11,7 +11,7 @@ use crate::odbc::{
 #[allow(unused_imports)]
 use crate::row::Row as SqlxRow;
 use either::Either;
-use odbc_api::{Cursor, CursorRow, ResultSetMetadata};
+use odbc_api::{Cursor, CursorRow, IntoParameter, ResultSetMetadata};
 
 #[derive(Debug)]
 pub(crate) struct ConnectionWorker {
@@ -115,12 +115,10 @@ impl ConnectionWorker {
                         Command::Execute { sql, tx } => {
                             with_conn(&shared, |conn| execute_sql(conn, &sql, &tx));
                         }
-                        Command::ExecuteWithArgs {
-                            sql,
-                            args: _args,
-                            tx,
-                        } => {
-                            with_conn(&shared, |conn| execute_sql(conn, &sql, &tx));
+                        Command::ExecuteWithArgs { sql, args, tx } => {
+                            with_conn(&shared, |conn| {
+                                execute_sql_with_params(conn, &sql, args, &tx)
+                            });
                         }
                     }
                 }
@@ -254,6 +252,65 @@ fn execute_sql(
     }
 }
 
+fn execute_sql_with_params(
+    conn: &odbc_api::Connection<'static>,
+    sql: &str,
+    args: Vec<OdbcArgumentValue<'static>>,
+    tx: &flume::Sender<Result<Either<OdbcQueryResult, OdbcRow>, Error>>,
+) {
+    if args.is_empty() {
+        dispatch_execute(conn, sql, (), tx);
+        return;
+    }
+
+    let mut params: Vec<Box<dyn odbc_api::parameter::InputParameter>> =
+        Vec::with_capacity(args.len());
+    for a in args {
+        params.push(to_param(a));
+    }
+    dispatch_execute(conn, sql, &params[..], tx);
+}
+
+fn to_param(
+    arg: OdbcArgumentValue<'static>,
+) -> Box<dyn odbc_api::parameter::InputParameter + 'static> {
+    match arg {
+        OdbcArgumentValue::Int(i) => Box::new(i.into_parameter()),
+        OdbcArgumentValue::Float(f) => Box::new(f.into_parameter()),
+        OdbcArgumentValue::Text(s) => Box::new(s.into_parameter()),
+        OdbcArgumentValue::Bytes(b) => Box::new(b.into_parameter()),
+        OdbcArgumentValue::Null | OdbcArgumentValue::Phantom(_) => {
+            Box::new(Option::<i32>::None.into_parameter())
+        }
+    }
+}
+
+fn dispatch_execute<P>(
+    conn: &odbc_api::Connection<'static>,
+    sql: &str,
+    params: P,
+    tx: &flume::Sender<Result<Either<OdbcQueryResult, OdbcRow>, Error>>,
+) where
+    P: odbc_api::ParameterCollectionRef,
+{
+    match conn.execute(sql, params, None) {
+        Ok(Some(mut cursor)) => {
+            let columns = collect_columns(&mut cursor);
+            if let Err(e) = stream_rows(&mut cursor, &columns, tx) {
+                let _ = tx.send(Err(e));
+                return;
+            }
+            let _ = tx.send(Ok(Either::Left(OdbcQueryResult { rows_affected: 0 })));
+        }
+        Ok(None) => {
+            let _ = tx.send(Ok(Either::Left(OdbcQueryResult { rows_affected: 0 })));
+        }
+        Err(e) => {
+            let _ = tx.send(Err(Error::from(e)));
+        }
+    }
+}
+
 fn collect_columns<C>(cursor: &mut C) -> Vec<OdbcColumn>
 where
     C: ResultSetMetadata,

test.sh

@@ -9,3 +9,5 @@ docker compose -f tests/docker-compose.yml run -it -p 3306:3306 --name mysql_8 m
 DATABASE_URL='mysql://root:password@localhost/sqlx' cargo test --features any,mysql,macros,all-types,runtime-actix-rustls --
 
 DATABASE_URL='sqlite://./tests/sqlite/sqlite.db' cargo test --features any,sqlite,macros,all-types,runtime-actix-rustls --
+
+ATABASE_URL='DSN=SQLX_PG_55432;UID=postgres;PWD=password' cargo test --no-default-features --features odbc,macros,runtime-tokio-rustls --test odbc