refactor(odbc): enhance column binding and buffer settings handling

This commit refines the column binding logic in the OdbcBridge by removing unnecessary parameters and improving the handling of maximum column sizes. It also updates the OdbcConnectOptions to assert that the max_column_size is greater than zero, ensuring valid configurations. Additionally, tests are modified to accommodate the new function signatures, enhancing the robustness of data handling in ODBC connections.

Author: lovasoa

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

@@ -16,19 +16,16 @@ use std::sync::Arc;
 // Bulk fetch implementation using columnar buffers instead of row-by-row fetching
 // This provides significant performance improvements by fetching rows in batches
 // and avoiding the slow `next_row()` method from odbc-api
-
+#[derive(Debug)]
 struct ColumnBinding {
     column: OdbcColumn,
     buffer_desc: BufferDesc,
 }
 
-fn build_bindings<C>(
+fn build_bindings<C: Cursor>(
     cursor: &mut C,
-    buffer_settings: &OdbcBufferSettings,
-) -> Result<Vec<ColumnBinding>, Error>
-where
-    C: ResultSetMetadata,
-{
+    max_column_size: usize,
+) -> Result<Vec<ColumnBinding>, Error> {
     let column_count = cursor.num_result_cols().unwrap_or(0);
     let mut bindings = Vec::with_capacity(column_count as usize);
     for index in 1..=column_count {
@@ -37,18 +34,18 @@ where
             .col_nullability(index as u16)
             .unwrap_or(Nullability::Unknown)
             .could_be_nullable();
-        let buffer_desc = map_buffer_desc(
-            cursor,
-            index as u16,
-            &column.type_info,
-            nullable,
-            buffer_settings,
-        )?;
+        let buffer_desc = map_buffer_desc(&column.type_info, nullable, max_column_size)?;
         bindings.push(ColumnBinding {
             column,
             buffer_desc,
         });
     }
+    dbg!(&bindings);
+    log::trace!(
+        "built {} ODBC batch column bindings: {:?}",
+        bindings.len(),
+        bindings
+    );
     Ok(bindings)
 }
 
@@ -131,14 +128,15 @@ fn to_param(arg: OdbcArgumentValue) -> Box<dyn odbc_api::parameter::InputParamet
     }
 }
 
-fn handle_cursor<C>(mut cursor: C, tx: &ExecuteSender, buffer_settings: OdbcBufferSettings)
-where
-    C: Cursor + ResultSetMetadata,
-{
+fn handle_cursor<C: Cursor + ResultSetMetadata>(
+    mut cursor: C,
+    tx: &ExecuteSender,
+    buffer_settings: OdbcBufferSettings,
+) {
     match buffer_settings.max_column_size {
-        Some(_) => {
+        Some(max_column_size) => {
             // Buffered mode - use batch fetching with columnar buffers
-            let bindings = match build_bindings(&mut cursor, &buffer_settings) {
+            let bindings = match build_bindings(&mut cursor, max_column_size) {
                 Ok(b) => b,
                 Err(e) => {
                     send_error(tx, e);
@@ -197,34 +195,17 @@ where
     }
 }
 
-fn map_buffer_desc<C>(
-    _cursor: &mut C,
-    _column_index: u16,
+fn map_buffer_desc(
     type_info: &OdbcTypeInfo,
     nullable: bool,
-    buffer_settings: &OdbcBufferSettings,
-) -> Result<BufferDesc, Error>
-where
-    C: ResultSetMetadata,
-{
+    max_column_size: usize,
+) -> Result<BufferDesc, Error> {
     use odbc_api::DataType;
 
+    // Some drivers report datatype lengths that are smaller than the actual data,
+    // so we cannot use it to build the BufferDesc.
     let data_type = type_info.data_type();
-
-    // Helper function to determine buffer length with fallback
-    let max_column_size = buffer_settings.max_column_size.unwrap_or(4096);
-
-    let buffer_length = |length: Option<std::num::NonZeroUsize>| {
-        if let Some(length) = length {
-            if length.get() < 255 {
-                length.get()
-            } else {
-                max_column_size
-            }
-        } else {
-            max_column_size
-        }
-    };
+    let max_str_len = max_column_size;
 
     let buffer_desc = match data_type {
         // Integer types - all map to I64
@@ -241,31 +222,22 @@ where
         DataType::Time { .. } => BufferDesc::Time { nullable },
         DataType::Timestamp { .. } => BufferDesc::Timestamp { nullable },
         // Binary types
-        DataType::Binary { length }
-        | DataType::Varbinary { length }
-        | DataType::LongVarbinary { length } => BufferDesc::Binary {
-            length: buffer_length(length),
-        },
+        DataType::Binary { .. } | DataType::Varbinary { .. } | DataType::LongVarbinary { .. } => {
+            BufferDesc::Binary {
+                length: max_column_size,
+            }
+        }
         // Text types
-        DataType::Char { length }
-        | DataType::WChar { length }
-        | DataType::Varchar { length }
-        | DataType::WVarchar { length }
-        | DataType::LongVarchar { length }
-        | DataType::WLongVarchar { length }
-        | DataType::Other {
-            column_size: length,
-            ..
-        } => BufferDesc::Text {
-            max_str_len: buffer_length(length),
-        },
+        DataType::Char { .. }
+        | DataType::WChar { .. }
+        | DataType::Varchar { .. }
+        | DataType::WVarchar { .. }
+        | DataType::LongVarchar { .. }
+        | DataType::WLongVarchar { .. }
+        | DataType::Other { .. } => BufferDesc::Text { max_str_len },
         // Fallback cases
-        DataType::Unknown => BufferDesc::Text {
-            max_str_len: max_column_size,
-        },
-        DataType::Decimal { .. } | DataType::Numeric { .. } => BufferDesc::Text {
-            max_str_len: max_column_size,
-        },
+        DataType::Unknown => BufferDesc::Text { max_str_len },
+        DataType::Decimal { .. } | DataType::Numeric { .. } => BufferDesc::Text { max_str_len },
     };
 
     Ok(buffer_desc)

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

@@ -140,7 +140,13 @@ impl OdbcConnectOptions {
     ///
     /// - When set to `Some(value)`: Enables buffered mode with batch fetching
     /// - When set to `None`: Enables unbuffered mode with row-by-row processing
+    ///
+    /// # Panics
+    /// Panics if `max_column_size` is less than 0.
     pub fn max_column_size(&mut self, max_column_size: Option<usize>) -> &mut Self {
+        if let Some(size) = max_column_size {
+            assert!(size > 0, "max_column_size must be greater than 0");
+        }
         self.buffer_settings.max_column_size = max_column_size;
         self
     }

tests/odbc/odbc.rs

@@ -440,7 +440,7 @@ async fn test_with_buffer_settings<F, T>(
     test_fn: F,
 ) -> anyhow::Result<()>
 where
-    F: Fn(OdbcConnection) -> T,
+    F: Fn(OdbcConnection, OdbcBufferSettings) -> T,
     T: std::future::Future<Output = anyhow::Result<()>>,
 {
     use sqlx_oldapi::odbc::OdbcConnectOptions;
@@ -451,7 +451,7 @@ where
         opts.buffer_settings(buf_settings);
 
         let conn = OdbcConnection::connect_with(&opts).await?;
-        test_fn(conn).await?;
+        test_fn(conn, buf_settings).await?;
     }
     Ok(())
 }
@@ -472,7 +472,7 @@ async fn it_handles_binary_data() -> anyhow::Result<()> {
         },
     ];
 
-    test_with_buffer_settings(&buffer_settings, |mut conn| async move {
+    test_with_buffer_settings(&buffer_settings, |mut conn, buf_settings| async move {
         let stmt = conn.prepare("SELECT ? AS binary_data").await?;
         let row = stmt
             .query_as::<(Vec<u8>,)>()
@@ -482,7 +482,10 @@ async fn it_handles_binary_data() -> anyhow::Result<()> {
             .expect("query failed")
             .expect("row expected");
 
-        assert_eq!(row.0, binary_data);
+        assert_eq!(
+            row.0, binary_data,
+            "failed to query {binary_data:?} with buffer settings {buf_settings:?}"
+        );
         Ok(())
     })
     .await
@@ -504,7 +507,7 @@ async fn it_handles_text_as_utf8_binary() -> anyhow::Result<()> {
         },
     ];
 
-    test_with_buffer_settings(&buffer_settings, |mut conn| async move {
+    test_with_buffer_settings(&buffer_settings, |mut conn, buf_settings| async move {
         let stmt = conn.prepare("SELECT ? AS text_data").await?;
         let row = stmt
             .query_as::<(Vec<u8>,)>()
@@ -514,7 +517,11 @@ async fn it_handles_text_as_utf8_binary() -> anyhow::Result<()> {
             .expect("query failed")
             .expect("row expected");
 
-        assert_eq!(row.0, text.as_bytes());
+        assert_eq!(
+            row.0,
+            text.as_bytes(),
+            "failed to query {text} with buffer settings {buf_settings:?}"
+        );
         Ok(())
     })
     .await
@@ -1178,7 +1185,7 @@ async fn it_works_with_buffered_and_unbuffered_mode() -> anyhow::Result<()> {
         },
     ];
 
-    test_with_buffer_settings(&buffer_settings, |mut conn| async move {
+    test_with_buffer_settings(&buffer_settings, |mut conn, buf_settings| async move {
         let select = (0..count)
             .map(|i| format!("SELECT {i} AS n, '{}' as aas", "a".repeat(i)))
             .collect::<Vec<_>>()
@@ -1207,7 +1214,11 @@ async fn it_works_with_buffered_and_unbuffered_mode() -> anyhow::Result<()> {
                 .to_owned()
                 .try_decode::<String>()
                 .expect("SELECT aas should be a string");
-            assert_eq!(aas, "a".repeat(i));
+            assert_eq!(
+                aas,
+                "a".repeat(i),
+                "failed to query {i} with buffer settings {buf_settings:?}"
+            );
         }
         Ok(())
     })