feat(rust/ffi): support ADBC 1.1.0 rich errors from exported streams

rust/Cargo.toml

@@ -43,3 +43,8 @@ arrow-array = { version = ">=58, <60", default-features = false, features = [
 arrow-buffer = { version = ">=58, <60", default-features = false }
 arrow-schema = { version = ">=58, <60", default-features = false }
 arrow-select = { version = ">=58, <60", default-features = false }
+# The floor matters for the minimal-versions CI job: libc 0.2.0 (the "0.2"
+# minimum) predates the errno constants used by adbc_ffi and conflicts with
+# rustix's own floor. 0.2.182 matches the lowest version the rest of the
+# workspace already requires.
+libc = { version = "0.2.182", default-features = false }

rust/driver/dummy/src/lib.rs

@@ -67,6 +67,63 @@ impl RecordBatchReader for SingleBatchReader {
     }
 }
 
+/// SQL query that makes [`DummyStatement::execute`] return a stream which fails on
+/// the first call to `next`, carrying the rich [`Error`] from [`stream_error`].
+///
+/// Used to exercise the driver exporter's `ErrorFromArrayStream` implementation.
+pub const STREAM_ERROR_QUERY: &str = "TRIGGER_STREAM_ERROR";
+
+/// The [`Error`] emitted by the stream produced for [`STREAM_ERROR_QUERY`].
+pub fn stream_error() -> Error {
+    Error {
+        message: "the stream failed midway".into(),
+        status: Status::IO,
+        vendor_code: 42,
+        sqlstate: [
+            b'0' as std::os::raw::c_char,
+            b'8' as std::os::raw::c_char,
+            b'0' as std::os::raw::c_char,
+            b'0' as std::os::raw::c_char,
+            b'6' as std::os::raw::c_char,
+        ],
+        details: Some(vec![("detail-key".into(), b"detail-value".to_vec())]),
+    }
+}
+
+/// A [`RecordBatchReader`] whose first `next` yields the [`Error`] from
+/// [`stream_error`], wrapped as [`ArrowError::ExternalError`] exactly as a real
+/// driver would surface a mid-stream failure.
+#[derive(Debug)]
+pub struct FailingReader {
+    schema: SchemaRef,
+    error: Option<Error>,
+}
+
+impl FailingReader {
+    fn new() -> Self {
+        Self {
+            schema: Arc::new(get_table_schema()),
+            error: Some(stream_error()),
+        }
+    }
+}
+
+impl Iterator for FailingReader {
+    type Item = std::result::Result<RecordBatch, ArrowError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.error
+            .take()
+            .map(|error| Err(ArrowError::ExternalError(Box::new(error))))
+    }
+}
+
+impl RecordBatchReader for FailingReader {
+    fn schema(&self) -> SchemaRef {
+        self.schema.clone()
+    }
+}
+
 fn get_table_schema() -> Schema {
     Schema::new(vec![
         Field::new("a", DataType::UInt32, true),
@@ -817,6 +874,7 @@ impl Connection for DummyConnection {
 #[derive(Default)]
 pub struct DummyStatement {
     options: HashMap<OptionStatement, OptionValue>,
+    query: Option<String>,
 }
 
 impl Optionable for DummyStatement {
@@ -858,6 +916,9 @@ impl Statement for DummyStatement {
 
     fn execute(&mut self) -> Result<Box<dyn RecordBatchReader + Send + 'static>> {
         maybe_panic("StatementExecuteQuery");
+        if self.query.as_deref() == Some(STREAM_ERROR_QUERY) {
+            return Ok(Box::new(FailingReader::new()));
+        }
         let batch = get_table_data();
         let reader = SingleBatchReader::new(batch);
         Ok(Box::new(reader))
@@ -887,7 +948,8 @@ impl Statement for DummyStatement {
         Ok(())
     }
 
-    fn set_sql_query(&mut self, _query: impl AsRef<str>) -> Result<()> {
+    fn set_sql_query(&mut self, query: impl AsRef<str>) -> Result<()> {
+        self.query = Some(query.as_ref().to_string());
         Ok(())
     }
 

rust/driver/dummy/tests/driver_exporter_dummy.rs

@@ -650,3 +650,113 @@ fn test_statement_get_parameter_schema() {
     let native_schema = native_statement.get_parameter_schema().unwrap();
     assert_eq!(exported_schema, native_schema);
 }
+
+// ErrorFromArrayStream (ADBC 1.1.0): when a result stream fails mid-iteration, a
+// consumer must be able to recover the full `AdbcError` (status, vendor code and
+// structured details), not just an errno plus a message string.
+//
+// We drive the exported C driver directly here because the Rust driver manager
+// imports streams via arrow's `ArrowArrayStreamReader`, which has no notion of
+// this ADBC-specific entry point and so cannot exercise it.
+#[test]
+fn test_error_from_array_stream() {
+    use std::ffi::{CStr, CString};
+    use std::ptr::null_mut;
+
+    use adbc_core::constants;
+    use adbc_core::error::{AdbcStatusCode, Status};
+    use adbc_ffi::{
+        FFIDriver, FFI_AdbcConnection, FFI_AdbcDatabase, FFI_AdbcError, FFI_AdbcStatement,
+    };
+    use arrow_array::ffi::FFI_ArrowArray;
+    use arrow_array::ffi_stream::FFI_ArrowArrayStream;
+
+    use adbc_dummy::{stream_error, STREAM_ERROR_QUERY};
+
+    let driver = DummyDriver::ffi_driver();
+
+    unsafe {
+        let mut error = FFI_AdbcError::default();
+
+        let mut database = FFI_AdbcDatabase::default();
+        assert_eq!(
+            (driver.DatabaseNew.unwrap())(&mut database, &mut error),
+            constants::ADBC_STATUS_OK
+        );
+        assert_eq!(
+            (driver.DatabaseInit.unwrap())(&mut database, &mut error),
+            constants::ADBC_STATUS_OK
+        );
+
+        let mut connection = FFI_AdbcConnection::default();
+        assert_eq!(
+            (driver.ConnectionNew.unwrap())(&mut connection, &mut error),
+            constants::ADBC_STATUS_OK
+        );
+        assert_eq!(
+            (driver.ConnectionInit.unwrap())(&mut connection, &mut database, &mut error),
+            constants::ADBC_STATUS_OK
+        );
+
+        let mut statement = FFI_AdbcStatement::default();
+        assert_eq!(
+            (driver.StatementNew.unwrap())(&mut connection, &mut statement, &mut error),
+            constants::ADBC_STATUS_OK
+        );
+
+        let query = CString::new(STREAM_ERROR_QUERY).unwrap();
+        assert_eq!(
+            (driver.StatementSetSqlQuery.unwrap())(&mut statement, query.as_ptr(), &mut error),
+            constants::ADBC_STATUS_OK
+        );
+
+        // Execute to obtain the result stream.
+        let mut stream = FFI_ArrowArrayStream::empty();
+        assert_eq!(
+            (driver.StatementExecuteQuery.unwrap())(
+                &mut statement,
+                &mut stream,
+                null_mut(),
+                &mut error,
+            ),
+            constants::ADBC_STATUS_OK
+        );
+
+        // Iterating the stream fails on the first batch.
+        let mut array = FFI_ArrowArray::empty();
+        let rc = (stream.get_next.unwrap())(&mut stream, &mut array);
+        assert_ne!(rc, 0, "the stream was expected to fail");
+
+        // Recover the full error via ErrorFromArrayStream.
+        let mut status: AdbcStatusCode = constants::ADBC_STATUS_OK;
+        let error_ptr = (driver.ErrorFromArrayStream.unwrap())(&mut stream, &mut status);
+        assert!(!error_ptr.is_null(), "expected a recovered error");
+
+        let expected = stream_error();
+        assert_eq!(status, AdbcStatusCode::from(Status::IO));
+        assert_eq!(
+            CStr::from_ptr((*error_ptr).message).to_str().unwrap(),
+            expected.message
+        );
+        assert_eq!((*error_ptr).vendor_code, expected.vendor_code);
+        assert_eq!((*error_ptr).sqlstate, expected.sqlstate);
+
+        // Structured details survive too, retrievable via ErrorGetDetail*.
+        assert_eq!((driver.ErrorGetDetailCount.unwrap())(error_ptr), 1);
+        let detail = (driver.ErrorGetDetail.unwrap())(error_ptr, 0);
+        assert_eq!(CStr::from_ptr(detail.key).to_str().unwrap(), "detail-key");
+        assert_eq!(
+            std::slice::from_raw_parts(detail.value, detail.value_length),
+            b"detail-value"
+        );
+
+        // Cleanup. Releasing the stream frees the recovered error; the other
+        // handles are released via the driver.
+        if let Some(release) = stream.release {
+            release(&mut stream);
+        }
+        (driver.StatementRelease.unwrap())(&mut statement, &mut error);
+        (driver.ConnectionRelease.unwrap())(&mut connection, &mut error);
+        (driver.DatabaseRelease.unwrap())(&mut database, &mut error);
+    }
+}

rust/ffi/Cargo.toml

@@ -37,6 +37,7 @@ default = []
 adbc_core.workspace = true
 arrow-array.workspace = true
 arrow-schema.workspace = true
+libc.workspace = true
 
 [package.metadata.docs.rs]
 all-features = true

rust/ffi/src/driver_exporter.rs

@@ -132,7 +132,7 @@ impl<DriverType: Driver + Default + 'static> FFIDriver for DriverType {
             StatementSetSubstraitPlan: Some(statement_set_substrait_plan::<DriverType>),
             ErrorGetDetailCount: Some(error_get_detail_count),
             ErrorGetDetail: Some(error_get_detail),
-            ErrorFromArrayStream: None, // TODO(alexandreyc): what to do with this?
+            ErrorFromArrayStream: Some(crate::exported_stream::error_from_array_stream),
             DatabaseGetOption: Some(database_get_option::<DriverType>),
             DatabaseGetOptionBytes: Some(database_get_option_bytes::<DriverType>),
             DatabaseGetOptionDouble: Some(database_get_option_double::<DriverType>),
@@ -1117,7 +1117,7 @@ extern "C" fn connection_get_table_types<DriverType: Driver + 'static>(
 
         let reader = check_err!(connection.get_table_types(), error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1185,7 +1185,7 @@ extern "C" fn connection_get_info<DriverType: Driver + 'static>(
 
         let reader = check_err!(connection.get_info(info_codes), error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1263,7 +1263,7 @@ extern "C" fn connection_get_statistic_names<DriverType: Driver + 'static>(
 
         let reader = check_err!(connection.get_statistic_names(), error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1292,7 +1292,7 @@ extern "C" fn connection_read_partition<DriverType: Driver + 'static>(
             unsafe { std::slice::from_raw_parts(serialized_partition, serialized_length) };
         let reader = check_err!(connection.read_partition(partition), error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1326,7 +1326,7 @@ extern "C" fn connection_get_statistics<DriverType: Driver + 'static>(
         let reader = connection.get_statistics(catalog, db_schema, table_name, approximate);
         let reader = check_err!(reader, error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1383,7 +1383,7 @@ extern "C" fn connection_get_objects<DriverType: Driver + 'static>(
         );
         let reader = check_err!(reader, error);
         let reader = Box::new(reader);
-        let reader = FFI_ArrowArrayStream::new(reader);
+        let reader = crate::exported_stream::export_reader(reader);
         unsafe { std::ptr::write_unaligned(out, reader) };
 
         ADBC_STATUS_OK
@@ -1726,7 +1726,7 @@ extern "C" fn statement_execute_query<DriverType: Driver + 'static>(
         if !out.is_null() {
             let reader = check_err!(statement.execute(), error);
             let reader = Box::new(reader);
-            let reader = FFI_ArrowArrayStream::new(reader);
+            let reader = crate::exported_stream::export_reader(reader);
             unsafe { std::ptr::write_unaligned(out, reader) };
         } else {
             let rows_affected_value = check_err!(statement.execute_update(), error).unwrap_or(-1);