This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 8cc3971a6 fix(rust/ffi): avoid undefined behavior in exported drivers
(#4871)
8cc3971a6 is described below
commit 8cc3971a682a31ae63a406b092e94aef2f805ccc
Author: David Li <[email protected]>
AuthorDate: Sun Oct 11 16:36:16 2026 -0700
fix(rust/ffi): avoid undefined behavior in exported drivers (#4871)
- Use & instead of &mut to avoid violating borrowing rules
- Use Mutex to ensure even misbehaved clients don't violate safety
- Use Mutex to also provide interior mutability
- Some minor refactoring
Depends on #4861.
Assisted-by: GPT-6 Astra
---
.github/workflows/rust.yml | 2 +-
rust/driver_manager/tests/cancellation.rs | 38 +--
rust/driver_manager/tests/common/dummy.rs | 307 +++++++++++++++++++++
rust/driver_manager/tests/common/mod.rs | 36 +++
rust/driver_manager/tests/exporter_cancellation.rs | 107 +++++++
rust/ffi/src/driver_exporter.rs | 272 +++++++++++-------
6 files changed, 620 insertions(+), 142 deletions(-)
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 66fd8c4f4..5e17b06ff 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -297,4 +297,4 @@ jobs:
- name: adbc_driver_manager (miri)
working-directory: rust
run: |
- cargo +nightly miri test -p adbc_driver_manager --test cancellation
+ cargo +nightly miri test -p adbc_driver_manager --test cancellation
--test exporter_cancellation
diff --git a/rust/driver_manager/tests/cancellation.rs
b/rust/driver_manager/tests/cancellation.rs
index f9d260bf4..959bc38a3 100644
--- a/rust/driver_manager/tests/cancellation.rs
+++ b/rust/driver_manager/tests/cancellation.rs
@@ -35,6 +35,10 @@ use adbc_ffi::{
};
use arrow_array::ffi_stream::FFI_ArrowArrayStream;
+mod common;
+
+use common::AtomicFlag;
+
const TIMEOUT: Duration = Duration::from_secs(2);
const CALLBACK_TIMEOUT: Duration = Duration::from_secs(10);
@@ -42,40 +46,6 @@ const CALLBACK_TIMEOUT: Duration = Duration::from_secs(10);
// Simulated driver implementation
// ------------------------------------------------------------
-struct AtomicFlag {
- mu: std::sync::Mutex<bool>,
- cv: std::sync::Condvar,
-}
-
-impl AtomicFlag {
- fn new(value: bool) -> Self {
- Self {
- mu: std::sync::Mutex::new(value),
- cv: std::sync::Condvar::new(),
- }
- }
-
- fn clear(&self) {
- let mut guard = self.mu.lock().unwrap();
- *guard = false;
- self.cv.notify_all();
- }
-
- fn set(&self) {
- let mut guard = self.mu.lock().unwrap();
- *guard = true;
- self.cv.notify_all();
- }
-
- fn wait(&self, timeout: Duration) -> bool {
- let (guard, _) = self
- .cv
- .wait_timeout_while(self.mu.lock().unwrap(), timeout, |flag|
!*flag)
- .unwrap();
- *guard
- }
-}
-
// Track what's been done with the driver so we can verify invariants later.
struct DriverObjectState {
// count of non-cancel operations
diff --git a/rust/driver_manager/tests/common/dummy.rs
b/rust/driver_manager/tests/common/dummy.rs
new file mode 100644
index 000000000..acac42995
--- /dev/null
+++ b/rust/driver_manager/tests/common/dummy.rs
@@ -0,0 +1,307 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use adbc_core::error::{Error, Result, Status};
+use adbc_core::options::{
+ InfoCode, ObjectDepth, OptionConnection, OptionDatabase, OptionStatement,
OptionValue,
+};
+use adbc_core::{Connection, Database, Driver, Optionable, Statement};
+use arrow_array::{RecordBatch, RecordBatchReader};
+
+use super::AtomicFlag;
+
+// hackery for test
+pub static IN_COMMIT: std::sync::LazyLock<AtomicFlag> =
+ std::sync::LazyLock::new(|| AtomicFlag::new(false));
+
+#[derive(Default)]
+pub struct DummyDriver {}
+
+impl Driver for DummyDriver {
+ type DatabaseType = DummyDatabase;
+
+ fn new_database(&mut self) -> Result<Self::DatabaseType> {
+ Ok(Self::DatabaseType::default())
+ }
+
+ fn new_database_with_opts(
+ &mut self,
+ opts: impl IntoIterator<Item = (<Self::DatabaseType as
Optionable>::Option, OptionValue)>,
+ ) -> Result<Self::DatabaseType> {
+ let mut database = Self::DatabaseType::default();
+ for (key, value) in opts {
+ database.set_option(key, value)?;
+ }
+ Ok(database)
+ }
+}
+
+#[derive(Default)]
+pub struct DummyDatabase {}
+
+impl Optionable for DummyDatabase {
+ type Option = OptionDatabase;
+
+ fn set_option(&mut self, _key: Self::Option, _value: OptionValue) ->
Result<()> {
+ Ok(())
+ }
+
+ fn get_option_bytes(&self, _key: Self::Option) -> Result<Vec<u8>> {
+ todo!()
+ }
+
+ fn get_option_double(&self, _key: Self::Option) -> Result<f64> {
+ todo!()
+ }
+
+ fn get_option_int(&self, _key: Self::Option) -> Result<i64> {
+ todo!()
+ }
+
+ fn get_option_string(&self, _key: Self::Option) -> Result<String> {
+ todo!()
+ }
+}
+
+impl Database for DummyDatabase {
+ type ConnectionType = DummyConnection;
+
+ fn new_connection(&self) -> Result<Self::ConnectionType> {
+ Ok(Self::ConnectionType {
+ flag: std::sync::Arc::new(AtomicFlag::new(false)),
+ })
+ }
+
+ fn new_connection_with_opts(
+ &self,
+ opts: impl IntoIterator<Item = (<Self::ConnectionType as
Optionable>::Option, OptionValue)>,
+ ) -> Result<Self::ConnectionType> {
+ let mut connection = Self::ConnectionType {
+ flag: std::sync::Arc::new(AtomicFlag::new(false)),
+ };
+ for (key, value) in opts {
+ connection.set_option(key, value)?;
+ }
+ Ok(connection)
+ }
+}
+
+pub struct DummyConnection {
+ flag: std::sync::Arc<AtomicFlag>,
+}
+
+impl Optionable for DummyConnection {
+ type Option = OptionConnection;
+
+ fn set_option(&mut self, _key: Self::Option, _value: OptionValue) ->
Result<()> {
+ Ok(())
+ }
+
+ fn get_option_bytes(&self, _key: Self::Option) -> Result<Vec<u8>> {
+ todo!()
+ }
+
+ fn get_option_double(&self, _key: Self::Option) -> Result<f64> {
+ todo!()
+ }
+
+ fn get_option_int(&self, _key: Self::Option) -> Result<i64> {
+ todo!()
+ }
+
+ fn get_option_string(&self, _key: Self::Option) -> Result<String> {
+ todo!()
+ }
+}
+
+impl Connection for DummyConnection {
+ type StatementType = DummyStatement;
+
+ fn new_statement(&mut self) -> Result<Self::StatementType> {
+ Ok(Self::StatementType::default())
+ }
+
+ fn cancel(&mut self) -> Result<()> {
+ todo!()
+ }
+
+ fn get_cancel_handle(&self) -> Box<dyn adbc_core::CancelHandle> {
+ struct CancelHandle {
+ flag: std::sync::Arc<AtomicFlag>,
+ }
+
+ impl adbc_core::CancelHandle for CancelHandle {
+ fn try_cancel(&self) -> Result<()> {
+ self.flag.set();
+ Ok(())
+ }
+ }
+
+ Box::new(CancelHandle {
+ flag: self.flag.clone(),
+ })
+ }
+
+ fn commit(&mut self) -> Result<()> {
+ IN_COMMIT.set();
+
+ if self.flag.wait(std::time::Duration::from_secs(5)) {
+ // properly cancelled
+ Ok(())
+ } else {
+ Err(Error::with_message_and_status(
+ "Commit was not cancelled",
+ Status::Internal,
+ ))
+ }
+ }
+
+ fn get_info(
+ &self,
+ _codes: Option<std::collections::HashSet<InfoCode>>,
+ ) -> Result<Box<dyn RecordBatchReader + Send + 'static>> {
+ todo!()
+ }
+
+ fn get_objects(
+ &self,
+ _depth: ObjectDepth,
+ _catalog: Option<&str>,
+ _db_schema: Option<&str>,
+ _table_name: Option<&str>,
+ _table_type: Option<Vec<&str>>,
+ _column_name: Option<&str>,
+ ) -> Result<Box<dyn RecordBatchReader + Send + 'static>> {
+ todo!()
+ }
+
+ fn get_statistics(
+ &self,
+ _catalog: Option<&str>,
+ _db_schema: Option<&str>,
+ _table_name: Option<&str>,
+ _approximate: bool,
+ ) -> Result<Box<dyn RecordBatchReader + Send + 'static>> {
+ todo!()
+ }
+
+ fn get_statistic_names(&self) -> Result<Box<dyn RecordBatchReader + Send +
'static>> {
+ todo!()
+ }
+
+ fn get_table_schema(
+ &self,
+ _catalog: Option<&str>,
+ _db_schema: Option<&str>,
+ _table_name: &str,
+ ) -> Result<arrow_schema::Schema> {
+ todo!()
+ }
+
+ fn get_table_types(&self) -> Result<Box<dyn RecordBatchReader + Send +
'static>> {
+ todo!()
+ }
+
+ fn read_partition(
+ &self,
+ _partition: impl AsRef<[u8]>,
+ ) -> Result<Box<dyn RecordBatchReader + Send + 'static>> {
+ todo!()
+ }
+
+ fn rollback(&mut self) -> Result<()> {
+ Ok(())
+ }
+}
+
+#[derive(Default)]
+pub struct DummyStatement {
+ // make sure this struct isn't zero-sized
+ _state: i32,
+}
+
+impl Optionable for DummyStatement {
+ type Option = OptionStatement;
+
+ fn set_option(&mut self, _key: Self::Option, _value: OptionValue) ->
Result<()> {
+ Ok(())
+ }
+
+ fn get_option_bytes(&self, _key: Self::Option) -> Result<Vec<u8>> {
+ todo!()
+ }
+
+ fn get_option_double(&self, _key: Self::Option) -> Result<f64> {
+ todo!()
+ }
+
+ fn get_option_int(&self, _key: Self::Option) -> Result<i64> {
+ todo!()
+ }
+
+ fn get_option_string(&self, _key: Self::Option) -> Result<String> {
+ todo!()
+ }
+}
+
+impl Statement for DummyStatement {
+ fn bind(&mut self, _batch: RecordBatch) -> Result<()> {
+ Ok(())
+ }
+
+ fn bind_stream(&mut self, _reader: Box<dyn RecordBatchReader + Send>) ->
Result<()> {
+ Ok(())
+ }
+
+ fn cancel(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn execute(&mut self) -> Result<Box<dyn RecordBatchReader + Send +
'static>> {
+ todo!()
+ }
+
+ fn execute_partitions(&mut self) -> Result<adbc_core::PartitionedResult> {
+ todo!()
+ }
+
+ fn execute_schema(&mut self) -> Result<arrow_schema::Schema> {
+ todo!()
+ }
+
+ fn execute_update(&mut self) -> Result<Option<i64>> {
+ Ok(Some(0))
+ }
+
+ fn get_parameter_schema(&self) -> Result<arrow_schema::Schema> {
+ todo!()
+ }
+
+ fn prepare(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn set_sql_query(&mut self, _query: impl AsRef<str>) -> Result<()> {
+ Ok(())
+ }
+
+ fn set_substrait_plan(&mut self, _plan: impl AsRef<[u8]>) -> Result<()> {
+ Ok(())
+ }
+}
+
+adbc_ffi::export_driver!(AdbcDummyInit, DummyDriver);
diff --git a/rust/driver_manager/tests/common/mod.rs
b/rust/driver_manager/tests/common/mod.rs
index f0b60d208..e546f6798 100644
--- a/rust/driver_manager/tests/common/mod.rs
+++ b/rust/driver_manager/tests/common/mod.rs
@@ -35,6 +35,42 @@ use arrow_array::{
use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaRef};
use arrow_select::concat::concat_batches;
+pub mod dummy;
+
+pub struct AtomicFlag {
+ mu: std::sync::Mutex<bool>,
+ cv: std::sync::Condvar,
+}
+
+impl AtomicFlag {
+ pub fn new(value: bool) -> Self {
+ Self {
+ mu: std::sync::Mutex::new(value),
+ cv: std::sync::Condvar::new(),
+ }
+ }
+
+ pub fn clear(&self) {
+ let mut guard = self.mu.lock().unwrap();
+ *guard = false;
+ self.cv.notify_all();
+ }
+
+ pub fn set(&self) {
+ let mut guard = self.mu.lock().unwrap();
+ *guard = true;
+ self.cv.notify_all();
+ }
+
+ pub fn wait(&self, timeout: std::time::Duration) -> bool {
+ let (guard, _) = self
+ .cv
+ .wait_timeout_while(self.mu.lock().unwrap(), timeout, |flag|
!*flag)
+ .unwrap();
+ *guard
+ }
+}
+
pub struct SingleBatchReader {
batch: Option<RecordBatch>,
schema: SchemaRef,
diff --git a/rust/driver_manager/tests/exporter_cancellation.rs
b/rust/driver_manager/tests/exporter_cancellation.rs
new file mode 100644
index 000000000..186dd47e3
--- /dev/null
+++ b/rust/driver_manager/tests/exporter_cancellation.rs
@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Test that the driver exporter and cancellation do not have undefined
+// behavior by introducing overlapping mutable borrows. Run under Miri.
+
+use std::sync::{Arc, Mutex};
+
+use adbc_core::error::{Result, Status};
+use adbc_core::options::AdbcVersion;
+use adbc_core::{CancelHandle, Connection, Database, Driver, Statement};
+use adbc_driver_manager::ManagedDriver;
+use adbc_ffi::FFI_AdbcDriverInitFunc;
+use arrow_array::{RecordBatch, RecordBatchReader};
+use arrow_schema::{ArrowError, Schema, SchemaRef};
+
+mod common;
+
+struct CancelOnDropReader {
+ cancel: Box<dyn CancelHandle>,
+ status: Arc<Mutex<Option<Status>>>,
+}
+
+impl Iterator for CancelOnDropReader {
+ type Item = std::result::Result<RecordBatch, ArrowError>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ None
+ }
+}
+
+impl RecordBatchReader for CancelOnDropReader {
+ fn schema(&self) -> SchemaRef {
+ Arc::new(Schema::empty())
+ }
+}
+
+impl Drop for CancelOnDropReader {
+ fn drop(&mut self) {
+ let status = match self.cancel.try_cancel() {
+ Ok(()) => Status::Ok,
+ Err(error) => error.status,
+ };
+ if let Ok(mut recorded) = self.status.lock() {
+ *recorded = Some(status);
+ }
+ }
+}
+
+#[test]
+fn statement_cancellation() -> Result<()> {
+ let mut driver = ManagedDriver::load_static(
+ &(common::dummy::AdbcDummyInit as FFI_AdbcDriverInitFunc),
+ AdbcVersion::V110,
+ )?;
+ let database = driver.new_database()?;
+ let mut connection = database.new_connection()?;
+ let mut statement = connection.new_statement()?;
+ let status = Arc::new(Mutex::new(None));
+ let reader = CancelOnDropReader {
+ cancel: statement.get_cancel_handle(),
+ status: status.clone(),
+ };
+
+ // The dummy driver drops the reader while bind_stream's mutable statement
+ // borrow is live. Its destructor re-enters statement_private_data through
+ // cancellation, creating a potential overlapping mutable borrow of the
+ // exporter state. This was fixed, but this ensures this case continues to
+ // work and should be run via miri.
+ statement.bind_stream(Box::new(reader))?;
+ assert_eq!(*status.lock().unwrap(), Some(Status::Unknown));
+ Ok(())
+}
+
+#[test]
+fn connection_cancellation() -> Result<()> {
+ let mut driver = ManagedDriver::load_static(
+ &(common::dummy::AdbcDummyInit as FFI_AdbcDriverInitFunc),
+ AdbcVersion::V110,
+ )?;
+ let database = driver.new_database()?;
+ let mut connection = database.new_connection()?;
+
+ let handle = connection.get_cancel_handle();
+ let guard = std::thread::spawn(move || {
+
assert!(common::dummy::IN_COMMIT.wait(std::time::Duration::from_secs(5)));
+ handle.try_cancel().unwrap();
+ });
+
+ connection.commit()?;
+ guard.join().unwrap();
+ Ok(())
+}
diff --git a/rust/ffi/src/driver_exporter.rs b/rust/ffi/src/driver_exporter.rs
index 2c775f718..cd356768f 100644
--- a/rust/ffi/src/driver_exporter.rs
+++ b/rust/ffi/src/driver_exporter.rs
@@ -19,6 +19,7 @@ use std::collections::{HashMap, HashSet};
use std::ffi::{CStr, CString};
use std::hash::Hash;
use std::os::raw::{c_char, c_int, c_void};
+use std::sync::Mutex;
use arrow_array::StructArray;
use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema, from_ffi};
@@ -62,33 +63,46 @@ impl<DriverType: Driver> ExportedDatabase<DriverType> {
}
struct InitializedConnection<DriverType: Driver> {
- connection: ConnectionType<DriverType>,
+ connection: Mutex<ConnectionType<DriverType>>,
cancel_handle: Box<dyn adbc_core::CancelHandle>,
}
enum ExportedConnection<DriverType: Driver> {
/// Pre-init options
- Options(HashMap<OptionConnection, OptionValue>),
+ Options(Mutex<HashMap<OptionConnection, OptionValue>>),
/// Initialized connection
Connection(InitializedConnection<DriverType>),
}
+type ConnectionTuple<'a, DriverType> = (
+ Option<std::sync::MutexGuard<'a, HashMap<OptionConnection, OptionValue>>>,
+ Option<std::sync::MutexGuard<'a, ConnectionType<DriverType>>>,
+);
+
impl<DriverType: Driver> ExportedConnection<DriverType> {
- fn tuple(
- &mut self,
- ) -> (
- Option<&mut HashMap<OptionConnection, OptionValue>>,
- Option<&mut ConnectionType<DriverType>>,
- ) {
+ fn tuple(&self) -> Result<ConnectionTuple<'_, DriverType>> {
match self {
- Self::Options(options) => (Some(options), None),
- Self::Connection(connection) => (None, Some(&mut
connection.connection)),
+ Self::Options(options) => options
+ .lock()
+ .map_err(|_| {
+ Error::with_message_and_status("Connection mutex
poisoned", Status::Internal)
+ })
+ .map(|guard| (Some(guard), None)),
+ Self::Connection(connection) => connection
+ .connection
+ .lock()
+ .map_err(|_| {
+ Error::with_message_and_status("Connection mutex
poisoned", Status::Internal)
+ })
+ .map(|guard| (None, Some(guard))),
}
}
- fn try_connection(&mut self) -> Result<&mut ConnectionType<DriverType>> {
+ fn try_connection(&self) -> Result<std::sync::MutexGuard<'_,
ConnectionType<DriverType>>> {
match self {
- Self::Connection(connection) => Ok(&mut connection.connection),
+ Self::Connection(connection) =>
connection.connection.lock().map_err(|_| {
+ Error::with_message_and_status("Connection mutex poisoned",
Status::Internal)
+ }),
_ => Err(Error::with_message_and_status(
"Connection not initialized",
Status::InvalidState,
@@ -96,9 +110,9 @@ impl<DriverType: Driver> ExportedConnection<DriverType> {
}
}
- fn try_cancel(&mut self) -> Result<&mut dyn adbc_core::CancelHandle> {
+ fn try_cancel(&self) -> Result<&dyn adbc_core::CancelHandle> {
match self {
- Self::Connection(connection) =>
Ok(connection.cancel_handle.as_mut()),
+ Self::Connection(connection) =>
Ok(connection.cancel_handle.as_ref()),
_ => Err(Error::with_message_and_status(
"Connection not initialized",
Status::InvalidState,
@@ -107,10 +121,18 @@ impl<DriverType: Driver> ExportedConnection<DriverType> {
}
}
-struct ExportedStatement<DriverType: Driver>(
- StatementType<DriverType>,
- Box<dyn adbc_core::CancelHandle>,
-);
+struct ExportedStatement<DriverType: Driver> {
+ stmt: Mutex<StatementType<DriverType>>,
+ cancel: Box<dyn adbc_core::CancelHandle>,
+}
+
+impl<DriverType: Driver> ExportedStatement<DriverType> {
+ fn try_statement(&self) -> Result<std::sync::MutexGuard<'_,
StatementType<DriverType>>> {
+ self.stmt.lock().map_err(|_| {
+ Error::with_message_and_status("Statement mutex poisoned",
Status::Internal)
+ })
+ }
+}
pub trait FFIDriver {
fn ffi_driver() -> FFI_AdbcDriver;
@@ -298,16 +320,15 @@ macro_rules! check_not_null {
#[macro_export]
macro_rules! pointer_as_mut {
($ptr:ident, $err_out:expr) => {
- match unsafe { $ptr.as_mut() } {
- Some(p) => p,
- None => {
- let error = adbc_core::error::Error::with_message_and_status(
- format!("Passed null pointer for argument {:?}",
stringify!($ptr)),
- adbc_core::error::Status::InvalidArguments,
- );
- unsafe { $crate::export_error($err_out, error) };
- return adbc_core::error::Status::InvalidArguments.into();
- }
+ if let Some(p) = unsafe { $ptr.as_mut() } {
+ p
+ } else {
+ let error = adbc_core::error::Error::with_message_and_status(
+ format!("Passed null pointer for argument {:?}",
stringify!($ptr)),
+ adbc_core::error::Status::InvalidArguments,
+ );
+ unsafe { $crate::export_error($err_out, error) };
+ return adbc_core::error::Status::InvalidArguments.into();
}
};
}
@@ -827,10 +848,10 @@ unsafe fn maybe_str<'a>(str: *const c_char) ->
Result<Option<&'a str>> {
// Connection
unsafe fn connection_private_data<'a, DriverType: Driver>(
- connection: &mut FFI_AdbcConnection,
-) -> Result<&'a mut ExportedConnection<DriverType>> {
- let exported = connection.private_data as *mut
ExportedConnection<DriverType>;
- exported.as_mut().ok_or(Error::with_message_and_status(
+ connection: *mut FFI_AdbcConnection,
+) -> Result<&'a ExportedConnection<DriverType>> {
+ let exported = (*connection).private_data as *mut
ExportedConnection<DriverType>;
+ exported.as_ref().ok_or(Error::with_message_and_status(
"Uninitialized connection",
Status::InvalidState,
))
@@ -843,7 +864,7 @@ unsafe fn connection_set_option_impl<DriverType: Driver,
Value: Into<OptionValue
value: Value,
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
assert!(!key.is_null());
let exported =
check_err!(connection_private_data::<DriverType>(connection), error);
@@ -851,13 +872,22 @@ unsafe fn connection_set_option_impl<DriverType: Driver,
Value: Into<OptionValue
match exported {
ExportedConnection::Options(options) => {
+ let mut options = check_err!(
+ options.lock().map_err(|_| {
+ Error::with_message_and_status("Connection mutex
poisoned", Status::Internal)
+ }),
+ error
+ );
options.insert(key.into(), value.into());
}
ExportedConnection::Connection(connection) => {
- check_err!(
- connection.connection.set_option(key.into(), value.into()),
+ let mut connection = check_err!(
+ connection.connection.lock().map_err(|_| {
+ Error::with_message_and_status("Connection mutex
poisoned", Status::Internal)
+ }),
error
);
+ check_err!(connection.set_option(key.into(), value.into()), error);
}
}
@@ -871,7 +901,9 @@ extern "C" fn connection_new<DriverType: Driver>(
catch_panic(error, || {
let connection = pointer_as_mut!(connection, error);
- let exported =
Box::new(ExportedConnection::<DriverType>::Options(HashMap::new()));
+ let exported =
Box::new(ExportedConnection::<DriverType>::Options(Mutex::new(
+ HashMap::new(),
+ )));
connection.private_data = Box::into_raw(exported) as *mut c_void;
ADBC_STATUS_OK
@@ -887,8 +919,15 @@ extern "C" fn connection_init<DriverType: Driver>(
let connection = pointer_as_mut!(connection, error);
let database = pointer_as_mut!(database, error);
+ // It's OK to have a &mut here since the caller can't yet get a cancel
handle
let exported_connection = check_err!(
- unsafe { connection_private_data::<DriverType>(connection) },
+ unsafe {
+ let exported = connection.private_data as *mut
ExportedConnection<DriverType>;
+ exported.as_mut().ok_or(Error::with_message_and_status(
+ "Uninitialized connection",
+ Status::InvalidState,
+ ))
+ },
error
);
let exported_database = check_err!(
@@ -899,6 +938,16 @@ extern "C" fn connection_init<DriverType: Driver>(
if let ExportedConnection::Options(options) = exported_connection {
let connection = match exported_database {
ExportedDatabase::Database(database) => {
+ let options = check_err!(
+ options.get_mut().map_err(|_| {
+ Error::with_message_and_status(
+ "Connection mutex poisoned",
+ Status::Internal,
+ )
+ }),
+ error
+ );
+
database.new_connection_with_opts(options.clone())
}
_ => Err(Error::with_message_and_status(
@@ -908,8 +957,9 @@ extern "C" fn connection_init<DriverType: Driver>(
};
let connection = check_err!(connection, error);
let cancel_handle = connection.get_cancel_handle();
+
*exported_connection =
ExportedConnection::Connection(InitializedConnection {
- connection,
+ connection: Mutex::new(connection),
cancel_handle,
});
} else {
@@ -1022,7 +1072,7 @@ extern "C" fn connection_get_option<DriverType: Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(key, error);
check_not_null!(value, error);
check_not_null!(length, error);
@@ -1031,9 +1081,10 @@ extern "C" fn connection_get_option<DriverType: Driver>(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let (options, connection) = exported.tuple();
+ let (mut options, mut connection) = check_err!(exported.tuple(),
error);
- let optvalue = unsafe { get_option(connection, options, key) };
+ let optvalue =
+ unsafe { get_option(connection.as_deref_mut(),
options.as_deref_mut(), key) };
let optvalue = check_err!(optvalue, error);
check_err!(unsafe { copy_string(&optvalue, value, length) }, error);
@@ -1048,7 +1099,7 @@ extern "C" fn connection_get_option_int<DriverType:
Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(key, error);
check_not_null!(value, error);
@@ -1056,9 +1107,12 @@ extern "C" fn connection_get_option_int<DriverType:
Driver>(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let (options, connection) = exported.tuple();
+ let (mut options, mut connection) = check_err!(exported.tuple(),
error);
- let optvalue = check_err!(unsafe { get_option_int(connection, options,
key) }, error);
+ let optvalue = check_err!(
+ unsafe { get_option_int(connection.as_deref_mut(),
options.as_deref_mut(), key) },
+ error
+ );
unsafe { std::ptr::write_unaligned(value, optvalue) };
ADBC_STATUS_OK
@@ -1072,7 +1126,7 @@ extern "C" fn connection_get_option_double<DriverType:
Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(key, error);
check_not_null!(value, error);
@@ -1080,10 +1134,10 @@ extern "C" fn connection_get_option_double<DriverType:
Driver>(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let (options, connection) = exported.tuple();
+ let (mut options, mut connection) = check_err!(exported.tuple(),
error);
let optvalue = check_err!(
- unsafe { get_option_double(connection, options, key) },
+ unsafe { get_option_double(connection.as_deref_mut(),
options.as_deref_mut(), key) },
error
);
unsafe { std::ptr::write_unaligned(value, optvalue) };
@@ -1100,7 +1154,7 @@ extern "C" fn connection_get_option_bytes<DriverType:
Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(key, error);
check_not_null!(value, error);
check_not_null!(length, error);
@@ -1109,9 +1163,10 @@ extern "C" fn connection_get_option_bytes<DriverType:
Driver>(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let (options, connection) = exported.tuple();
+ let (mut options, mut connection) = check_err!(exported.tuple(),
error);
- let optvalue = unsafe { get_option_bytes(connection, options, key) };
+ let optvalue =
+ unsafe { get_option_bytes(connection.as_deref_mut(),
options.as_deref_mut(), key) };
let optvalue = check_err!(optvalue, error);
unsafe { copy_bytes(&optvalue, value, length) };
@@ -1125,7 +1180,7 @@ extern "C" fn connection_get_table_types<DriverType:
Driver + 'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(out, error);
let exported = check_err!(
@@ -1152,7 +1207,7 @@ extern "C" fn connection_get_table_schema<DriverType:
Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(table_name, error);
check_not_null!(schema, error);
@@ -1183,7 +1238,7 @@ extern "C" fn connection_get_info<DriverType: Driver +
'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(out, error);
let exported = check_err!(
@@ -1215,13 +1270,13 @@ extern "C" fn connection_commit<DriverType: Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
let exported = check_err!(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let connection = check_err!(exported.try_connection(), error);
+ let mut connection = check_err!(exported.try_connection(), error);
check_err!(connection.commit(), error);
ADBC_STATUS_OK
@@ -1233,13 +1288,13 @@ extern "C" fn connection_rollback<DriverType: Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
let exported = check_err!(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let connection = check_err!(exported.try_connection(), error);
+ let mut connection = check_err!(exported.try_connection(), error);
check_err!(connection.rollback(), error);
ADBC_STATUS_OK
@@ -1251,7 +1306,7 @@ extern "C" fn connection_cancel<DriverType: Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
let exported = check_err!(
unsafe { connection_private_data::<DriverType>(connection) },
@@ -1270,7 +1325,7 @@ extern "C" fn connection_get_statistic_names<DriverType:
Driver + 'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(out, error);
let exported = check_err!(
@@ -1296,7 +1351,7 @@ extern "C" fn connection_read_partition<DriverType:
Driver + 'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(serialized_partition, error);
check_not_null!(out, error);
@@ -1327,7 +1382,7 @@ extern "C" fn connection_get_statistics<DriverType:
Driver + 'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(out, error);
let catalog = check_err!(unsafe { maybe_str(catalog) }, error);
@@ -1363,7 +1418,7 @@ extern "C" fn connection_get_objects<DriverType: Driver +
'static>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
check_not_null!(out, error);
let depth = check_err!(ObjectDepth::try_from(depth), error);
@@ -1413,10 +1468,10 @@ extern "C" fn connection_get_objects<DriverType: Driver
+ 'static>(
// SAFETY: Will panic if `statement` is null.
unsafe fn statement_private_data<'a, DriverType: Driver>(
statement: *mut FFI_AdbcStatement,
-) -> Result<&'a mut ExportedStatement<DriverType>> {
+) -> Result<&'a ExportedStatement<DriverType>> {
assert!(!statement.is_null());
let exported = (*statement).private_data as *mut
ExportedStatement<DriverType>;
- exported.as_mut().ok_or(Error::with_message_and_status(
+ exported.as_ref().ok_or(Error::with_message_and_status(
"Uninitialized statement",
Status::InvalidState,
))
@@ -1433,8 +1488,9 @@ unsafe fn statement_set_option_impl<DriverType: Driver,
Value: Into<OptionValue>
assert!(!key.is_null());
let exported = check_err!(statement_private_data::<DriverType>(statement),
error);
+ let mut stmt = check_err!(exported.try_statement(), error);
let key = check_err!(CStr::from_ptr(key).to_str(), error);
- check_err!(exported.0.set_option(key.into(), value.into()), error);
+ check_err!(stmt.set_option(key.into(), value.into()), error);
ADBC_STATUS_OK
}
@@ -1444,22 +1500,21 @@ extern "C" fn statement_new<DriverType: Driver>(
error: *mut FFI_AdbcError,
) -> AdbcStatusCode {
catch_panic(error, || {
- let connection = pointer_as_mut!(connection, error);
+ check_not_null!(connection, error);
let statement = pointer_as_mut!(statement, error);
let exported_connection = check_err!(
unsafe { connection_private_data::<DriverType>(connection) },
error
);
- let inner_connection =
check_err!(exported_connection.try_connection(), error);
-
+ let mut inner_connection =
check_err!(exported_connection.try_connection(), error);
let inner_statement = check_err!(inner_connection.new_statement(),
error);
let cancel_handle = inner_statement.get_cancel_handle();
- let exported = Box::new(ExportedStatement::<DriverType>(
- inner_statement,
- cancel_handle,
- ));
+ let exported = Box::new(ExportedStatement::<DriverType> {
+ stmt: Mutex::new(inner_statement),
+ cancel: cancel_handle,
+ });
statement.private_data = Box::into_raw(exported) as *mut c_void;
ADBC_STATUS_OK
@@ -1568,7 +1623,8 @@ extern "C" fn statement_get_option<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let optvalue = unsafe { get_option(Some(&mut exported.0), None, key) };
+ let mut stmt = check_err!(exported.try_statement(), error);
+ let optvalue = unsafe { get_option(Some(&mut *stmt), None, key) };
let optvalue = check_err!(optvalue, error);
check_err!(unsafe { copy_string(&optvalue, value, length) }, error);
@@ -1591,8 +1647,9 @@ extern "C" fn statement_get_option_int<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
+ let mut stmt = check_err!(exported.try_statement(), error);
let optvalue = check_err!(
- unsafe { get_option_int(Some(&mut exported.0), None, key) },
+ unsafe { get_option_int(Some(&mut *stmt), None, key) },
error
);
unsafe { std::ptr::write_unaligned(value, optvalue) };
@@ -1616,8 +1673,9 @@ extern "C" fn statement_get_option_double<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
+ let mut stmt = check_err!(exported.try_statement(), error);
let optvalue = check_err!(
- unsafe { get_option_double(Some(&mut exported.0), None, key) },
+ unsafe { get_option_double(Some(&mut *stmt), None, key) },
error
);
unsafe { std::ptr::write_unaligned(value, optvalue) };
@@ -1643,7 +1701,8 @@ extern "C" fn statement_get_option_bytes<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let optvalue = unsafe { get_option_bytes(Some(&mut exported.0), None,
key) };
+ let mut stmt = check_err!(exported.try_statement(), error);
+ let optvalue = unsafe { get_option_bytes(Some(&mut *stmt), None, key)
};
let optvalue = check_err!(optvalue, error);
unsafe { copy_bytes(&optvalue, value, length) };
ADBC_STATUS_OK
@@ -1665,7 +1724,7 @@ extern "C" fn statement_bind<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
let schema = unsafe { schema.as_ref().unwrap() };
let data = unsafe { FFI_ArrowArray::from_raw(values) };
@@ -1701,7 +1760,8 @@ extern "C" fn statement_bind_stream<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut stmt = check_err!(exported.try_statement(), error);
+ let statement = &mut *stmt;
let reader = check_err!(unsafe {
ArrowArrayStreamReader::from_raw(stream) }, error);
let reader = Box::new(reader);
@@ -1722,8 +1782,7 @@ extern "C" fn statement_cancel<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let cancel_handle = &mut exported.1;
- check_err!(cancel_handle.try_cancel(), error);
+ check_err!(exported.cancel.try_cancel(), error);
ADBC_STATUS_OK
})
@@ -1742,7 +1801,7 @@ extern "C" fn statement_execute_query<DriverType: Driver
+ 'static>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
if !out.is_null() {
let reader = check_err!(statement.execute(), error);
@@ -1776,7 +1835,7 @@ extern "C" fn statement_execute_schema<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
let schema_value = check_err!(statement.execute_schema(), error);
let schema_value: FFI_ArrowSchema =
check_err!(schema_value.try_into(), error);
@@ -1802,7 +1861,7 @@ extern "C" fn statement_execute_partitions<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
let result = check_err!(statement.execute_partitions(), error);
@@ -1831,7 +1890,7 @@ extern "C" fn statement_prepare<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
check_err!(statement.prepare(), error);
ADBC_STATUS_OK
})
@@ -1850,7 +1909,7 @@ extern "C" fn statement_set_sql_query<DriverType: Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
let query = check_err!(unsafe { CStr::from_ptr(query).to_str() },
error);
check_err!(statement.set_sql_query(query), error);
@@ -1873,7 +1932,7 @@ extern "C" fn statement_set_substrait_plan<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &mut exported.0;
+ let mut statement = check_err!(exported.try_statement(), error);
let plan = unsafe { std::slice::from_raw_parts(plan, length) };
check_err!(statement.set_substrait_plan(plan), error);
@@ -1895,7 +1954,7 @@ extern "C" fn statement_get_parameter_schema<DriverType:
Driver>(
unsafe { statement_private_data::<DriverType>(statement) },
error
);
- let statement = &exported.0;
+ let statement = check_err!(exported.try_statement(), error);
let schema_value = check_err!(statement.get_parameter_schema(), error);
let schema_value: FFI_ArrowSchema =
check_err!(schema_value.try_into(), error);
@@ -1935,30 +1994,29 @@ unsafe extern "C" fn error_get_detail(
return default;
}
- match error.as_ref() {
- None => default,
- Some(error) => {
- let detail_count = error_get_detail_count(error);
- if index >= detail_count {
- return default;
- }
- let index = index as usize; // Cannot overflow since index >= 0
and index < detail_count
+ if let Some(error) = error.as_ref() {
+ let detail_count = error_get_detail_count(error);
+ if index >= detail_count {
+ return default;
+ }
+ let index = index as usize; // Cannot overflow since index >= 0 and
index < detail_count
- if error.private_data.is_null() {
- return default;
- }
- let private_data = error.private_data as *const ErrorPrivateData;
+ if error.private_data.is_null() {
+ return default;
+ }
+ let private_data = error.private_data as *const ErrorPrivateData;
- let key = (&(*private_data).keys)[index].as_ptr();
- let value = (&(*private_data).values)[index].as_ptr();
- let value_length = (&(*private_data).values)[index].len();
+ let key = (&(*private_data).keys)[index].as_ptr();
+ let value = (&(*private_data).values)[index].as_ptr();
+ let value_length = (&(*private_data).values)[index].len();
- FFI_AdbcErrorDetail {
- key,
- value,
- value_length,
- }
+ FFI_AdbcErrorDetail {
+ key,
+ value,
+ value_length,
}
+ } else {
+ default
}
}
