diff --git a/rust/Cargo.lock b/rust/Cargo.lock
index e8070b5dc3..0fc7638fb6 100644
--- a/rust/Cargo.lock
+++ b/rust/Cargo.lock
@@ -8,6 +8,7 @@ version = "0.22.0"
 dependencies = [
  "arrow-array",
  "arrow-schema",
+ "futures",
 ]
 
 [[package]]
@@ -22,6 +23,7 @@ dependencies = [
  "arrow-select",
  "datafusion",
  "datafusion-substrait",
+ "futures",
  "prost",
  "tokio",
 ]
@@ -40,7 +42,7 @@ dependencies = [
  "tempfile",
  "toml",
  "windows-registry",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
index 644769716d..2afbf9bc88 100644
--- a/rust/Cargo.toml
+++ b/rust/Cargo.toml
@@ -43,3 +43,4 @@ arrow-array = { version = ">=53.1.0, <58", default-features = false, features =
 arrow-buffer = { version = ">=53.1.0, <58", default-features = false }
 arrow-schema = { version = ">=53.1.0, <58", default-features = false }
 arrow-select = { version = ">=53.1.0, <58", default-features = false }
+futures = "0.3"
diff --git a/rust/core/Cargo.toml b/rust/core/Cargo.toml
index bd33bf57f5..ab17a68a91 100644
--- a/rust/core/Cargo.toml
+++ b/rust/core/Cargo.toml
@@ -36,6 +36,7 @@ default = []
 [dependencies]
 arrow-array.workspace = true
 arrow-schema.workspace = true
+futures.workspace = true
 
 [package.metadata.docs.rs]
 all-features = true
diff --git a/rust/core/src/lib.rs b/rust/core/src/lib.rs
index c8df8a9c18..36bbe10c46 100644
--- a/rust/core/src/lib.rs
+++ b/rust/core/src/lib.rs
@@ -31,26 +31,38 @@
 //! # Native Rust drivers
 //!
 //! Native Rust drivers will implement the abstract API consisting of the traits:
+//!
 //! - [Driver]
 //! - [Database]
 //! - [Connection]
 //! - [Statement]
 //!
+//! or async traits:
+//!
+//! - [crate::non_blocking::AsyncDriver]
+//! - [crate::non_blocking::AsyncDatabase]
+//! - [crate::non_blocking::AsyncConnection]
+//! - [crate::non_blocking::AsyncStatement]
+//!
+//! or async traits that doesn't require Send bound:
+//!
+//! - [crate::non_blocking::LocalAsyncDriver]
+//! - [crate::non_blocking::LocalAsyncDatabase]
+//! - [crate::non_blocking::LocalAsyncConnection]
+//! - [crate::non_blocking::LocalAsyncStatement]
+//!
 //! For drivers implemented in Rust, using these will be more efficient and
 //! safe, since it avoids the overhead of going through C FFI.
 
 pub mod constants;
 pub mod error;
+pub mod non_blocking;
 pub mod options;
 pub mod schemas;
+pub mod sync;
+pub mod syncify;
 
-use std::collections::HashSet;
-
-use arrow_array::{RecordBatch, RecordBatchReader};
-use arrow_schema::Schema;
-
-use error::Result;
-use options::{OptionConnection, OptionDatabase, OptionStatement, OptionValue};
+pub use sync::*;
 
 pub type LoadFlags = u32;
 
@@ -63,459 +75,17 @@ pub const LOAD_FLAG_DEFAULT: LoadFlags = LOAD_FLAG_SEARCH_ENV
     | LOAD_FLAG_SEARCH_SYSTEM
     | LOAD_FLAG_ALLOW_RELATIVE_PATHS;
 
-/// Ability to configure an object by setting/getting options.
-pub trait Optionable {
-    type Option: AsRef<str>;
-
-    /// Set a post-init option.
-    fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()>;
-
-    /// Get a string option value by key.
-    fn get_option_string(&self, key: Self::Option) -> Result<String>;
-
-    /// Get a bytes option value by key.
-    fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>>;
-
-    /// Get an integer option value by key.
-    fn get_option_int(&self, key: Self::Option) -> Result<i64>;
-
-    /// Get a float option value by key.
-    fn get_option_double(&self, key: Self::Option) -> Result<f64>;
-}
-
-/// A handle to an ADBC driver.
-pub trait Driver {
-    type DatabaseType: Database;
-
-    /// Allocate and initialize a new database without pre-init options.
-    fn new_database(&mut self) -> Result<Self::DatabaseType>;
-
-    /// Allocate and initialize a new database with pre-init options.
-    fn new_database_with_opts(
-        &mut self,
-        opts: impl IntoIterator<Item = (OptionDatabase, OptionValue)>,
-    ) -> Result<Self::DatabaseType>;
-}
-
-/// A handle to an ADBC database.
-///
-/// Databases hold state shared by multiple connections. This typically means
-/// configuration and caches. For in-memory databases, it provides a place to
-/// hold ownership of the in-memory database.
-///
-/// Databases must be kept alive as long as any connections exist.
-pub trait Database: Optionable<Option = OptionDatabase> {
-    type ConnectionType: Connection;
-
-    /// Allocate and initialize a new connection without pre-init options.
-    fn new_connection(&self) -> Result<Self::ConnectionType>;
-
-    /// Allocate and initialize a new connection with pre-init options.
-    fn new_connection_with_opts(
-        &self,
-        opts: impl IntoIterator<Item = (options::OptionConnection, OptionValue)>,
-    ) -> Result<Self::ConnectionType>;
-}
-
-/// A handle to an ADBC connection.
-///
-/// Connections provide methods for query execution, managing prepared
-/// statements, using transactions, and so on.
-///
-/// # Autocommit
-///
-/// Connections should start in autocommit mode. They can be moved out by
-/// setting [options::OptionConnection::AutoCommit] to "false". Turning off
-/// autocommit allows customizing the isolation level.
-pub trait Connection: Optionable<Option = OptionConnection> {
-    type StatementType: Statement;
-
-    /// Allocate and initialize a new statement.
-    fn new_statement(&mut self) -> Result<Self::StatementType>;
-
-    /// Cancel the in-progress operation on a connection.
-    fn cancel(&mut self) -> Result<()>;
-
-    /// Get metadata about the database/driver.
-    ///
-    /// # Arguments
-    ///
-    /// - `codes` - Requested metadata. If `None`, retrieve all available metadata.
-    ///
-    /// # Result
-    ///
-    /// The result is an Arrow dataset with the following schema:
-    ///
-    /// Field Name                  | Field Type
-    /// ----------------------------|------------------------
-    /// info_name                   | uint32 not null
-    /// info_value                  | INFO_SCHEMA
-    ///
-    /// INFO_SCHEMA is a dense union with members:
-    ///
-    /// Field Name (Type Code)      | Field Type
-    /// ----------------------------|------------------------
-    /// string_value (0)            | utf8
-    /// bool_value (1)              | bool
-    /// int64_value (2)             | int64
-    /// int32_bitmask (3)           | int32
-    /// string_list (4)             | list\<utf8\>
-    /// int32_to_int32_list_map (5) | map\<int32, list\<int32\>\>
-    fn get_info(
-        &self,
-        codes: Option<HashSet<options::InfoCode>>,
-    ) -> Result<impl RecordBatchReader + Send>;
-
-    /// Get a hierarchical view of all catalogs, database schemas, tables, and
-    /// columns.
-    ///
-    /// # Arguments
-    ///
-    /// - `depth` - The level of nesting to query.
-    /// - `catalog` - Only show tables in the given catalog. If `None`,
-    ///   do not filter by catalog. If an empty string, only show tables
-    ///   without a catalog.  May be a search pattern.
-    /// - `db_schema` - Only show tables in the given database schema. If
-    ///   `None`, do not filter by database schema. If an empty string, only show
-    ///   tables without a database schema. May be a search pattern.
-    /// - `table_name` - Only show tables with the given name. If `None`, do not
-    ///   filter by name. May be a search pattern.
-    /// - `table_type` - Only show tables matching one of the given table
-    ///   types. If `None`, show tables of any type. Valid table types can be fetched
-    ///   from [Connection::get_table_types].
-    /// - `column_name` - Only show columns with the given name. If
-    ///   `None`, do not filter by name.  May be a search pattern..
-    ///
-    /// # Result
-    ///
-    /// The result is an Arrow dataset with the following schema:
-    ///
-    /// | Field Name               | Field Type               |
-    /// |--------------------------|--------------------------|
-    /// | catalog_name             | utf8                     |
-    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> |
-    ///
-    /// DB_SCHEMA_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type              |
-    /// |--------------------------|-------------------------|
-    /// | db_schema_name           | utf8                    |
-    /// | db_schema_tables         | list\<TABLE_SCHEMA\>    |
-    ///
-    /// TABLE_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type                |
-    /// |--------------------------|---------------------------|
-    /// | table_name               | utf8 not null             |
-    /// | table_type               | utf8 not null             |
-    /// | table_columns            | list\<COLUMN_SCHEMA\>     |
-    /// | table_constraints        | list\<CONSTRAINT_SCHEMA\> |
-    ///
-    /// COLUMN_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type              | Comments |
-    /// |--------------------------|-------------------------|----------|
-    /// | column_name              | utf8 not null           |          |
-    /// | ordinal_position         | int32                   | (1)      |
-    /// | remarks                  | utf8                    | (2)      |
-    /// | xdbc_data_type           | int16                   | (3)      |
-    /// | xdbc_type_name           | utf8                    | (3)      |
-    /// | xdbc_column_size         | int32                   | (3)      |
-    /// | xdbc_decimal_digits      | int16                   | (3)      |
-    /// | xdbc_num_prec_radix      | int16                   | (3)      |
-    /// | xdbc_nullable            | int16                   | (3)      |
-    /// | xdbc_column_def          | utf8                    | (3)      |
-    /// | xdbc_sql_data_type       | int16                   | (3)      |
-    /// | xdbc_datetime_sub        | int16                   | (3)      |
-    /// | xdbc_char_octet_length   | int32                   | (3)      |
-    /// | xdbc_is_nullable         | utf8                    | (3)      |
-    /// | xdbc_scope_catalog       | utf8                    | (3)      |
-    /// | xdbc_scope_schema        | utf8                    | (3)      |
-    /// | xdbc_scope_table         | utf8                    | (3)      |
-    /// | xdbc_is_autoincrement    | bool                    | (3)      |
-    /// | xdbc_is_generatedcolumn  | bool                    | (3)      |
-    ///
-    /// 1. The column's ordinal position in the table (starting from 1).
-    /// 2. Database-specific description of the column.
-    /// 3. Optional value.  Should be null if not supported by the driver.
-    ///    `xdbc_` values are meant to provide JDBC/ODBC-compatible metadata
-    ///    in an agnostic manner.
-    ///
-    /// CONSTRAINT_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type              | Comments |
-    /// |--------------------------|-------------------------|----------|
-    /// | constraint_name          | utf8                    |          |
-    /// | constraint_type          | utf8 not null           | (1)      |
-    /// | constraint_column_names  | list\<utf8\> not null     | (2)      |
-    /// | constraint_column_usage  | list\<USAGE_SCHEMA\>      | (3)      |
-    ///
-    /// 1. One of `CHECK`, `FOREIGN KEY`, `PRIMARY KEY`, or `UNIQUE`.
-    /// 2. The columns on the current table that are constrained, in
-    ///    order.
-    /// 3. For `FOREIGN KEY` only, the referenced table and columns.
-    ///
-    /// USAGE_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type              |
-    /// |--------------------------|-------------------------|
-    /// | fk_catalog               | utf8                    |
-    /// | fk_db_schema             | utf8                    |
-    /// | fk_table                 | utf8 not null           |
-    /// | fk_column_name           | utf8 not null           |
-    ///
-    fn get_objects(
-        &self,
-        depth: options::ObjectDepth,
-        catalog: Option<&str>,
-        db_schema: Option<&str>,
-        table_name: Option<&str>,
-        table_type: Option<Vec<&str>>,
-        column_name: Option<&str>,
-    ) -> Result<impl RecordBatchReader + Send>;
-
-    /// Get the Arrow schema of a table.
-    ///
-    /// # Arguments
-    ///
-    /// - `catalog` - The catalog (or `None` if not applicable).
-    /// - `db_schema` - The database schema (or `None` if not applicable).
-    /// - `table_name` - The table name.
-    fn get_table_schema(
-        &self,
-        catalog: Option<&str>,
-        db_schema: Option<&str>,
-        table_name: &str,
-    ) -> Result<Schema>;
-
-    /// Get a list of table types in the database.
-    ///
-    /// # Result
-    ///
-    /// The result is an Arrow dataset with the following schema:
-    ///
-    /// Field Name     | Field Type
-    /// ---------------|--------------
-    /// table_type     | utf8 not null
-    fn get_table_types(&self) -> Result<impl RecordBatchReader + Send>;
-
-    /// Get the names of statistics specific to this driver.
-    ///
-    /// # Result
-    ///
-    /// The result is an Arrow dataset with the following schema:
-    ///
-    /// Field Name     | Field Type
-    /// ---------------|----------------
-    /// statistic_name | utf8 not null
-    /// statistic_key  | int16 not null
-    ///
-    /// # Since
-    /// ADBC API revision 1.1.0
-    fn get_statistic_names(&self) -> Result<impl RecordBatchReader + Send>;
-
-    /// Get statistics about the data distribution of table(s).
-    ///
-    /// # Arguments
-    ///
-    /// - `catalog` - The catalog (or `None` if not applicable). May be a search pattern.
-    /// - `db_schema` - The database schema (or `None` if not applicable). May be a search pattern
-    /// - `table_name` - The table name (or `None` if not applicable). May be a search pattern
-    /// - `approximate` - If false, request exact values of statistics, else
-    ///   allow for best-effort, approximate, or cached values. The database may
-    ///   return approximate values regardless, as indicated in the result.
-    ///   Requesting exact values may be expensive or unsupported.
-    ///
-    /// # Result
-    ///
-    /// The result is an Arrow dataset with the following schema:
-    ///
-    /// | Field Name               | Field Type                       |
-    /// |--------------------------|----------------------------------|
-    /// | catalog_name             | utf8                             |
-    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> not null|
-    ///
-    /// DB_SCHEMA_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type                        |
-    /// |--------------------------|-----------------------------------|
-    /// | db_schema_name           | utf8                              |
-    /// | db_schema_statistics     | list\<STATISTICS_SCHEMA\> not null|
-    ///
-    /// STATISTICS_SCHEMA is a Struct with fields:
-    ///
-    /// | Field Name               | Field Type                       | Comments |
-    /// |--------------------------|----------------------------------| -------- |
-    /// | table_name               | utf8 not null                    |          |
-    /// | column_name              | utf8                             | (1)      |
-    /// | statistic_key            | int16 not null                   | (2)      |
-    /// | statistic_value          | VALUE_SCHEMA not null            |          |
-    /// | statistic_is_approximate | bool not null                    | (3)      |
-    ///
-    /// 1. If null, then the statistic applies to the entire table.
-    /// 2. A dictionary-encoded statistic name (although we do not use the Arrow
-    ///    dictionary type). Values in [0, 1024) are reserved for ADBC.  Other
-    ///    values are for implementation-specific statistics.  For the definitions
-    ///    of predefined statistic types, see [options::Statistics]. To get
-    ///    driver-specific statistic names, use [Connection::get_statistic_names].
-    /// 3. If true, then the value is approximate or best-effort.
-    ///
-    /// VALUE_SCHEMA is a dense union with members:
-    ///
-    /// | Field Name               | Field Type                       |
-    /// |--------------------------|----------------------------------|
-    /// | int64                    | int64                            |
-    /// | uint64                   | uint64                           |
-    /// | float64                  | float64                          |
-    /// | binary                   | binary                           |
-    ///
-    /// # Since
-    ///
-    /// ADBC API revision 1.1.0
-    fn get_statistics(
-        &self,
-        catalog: Option<&str>,
-        db_schema: Option<&str>,
-        table_name: Option<&str>,
-        approximate: bool,
-    ) -> Result<impl RecordBatchReader + Send>;
-
-    /// Commit any pending transactions. Only used if autocommit is disabled.
-    ///
-    /// Behavior is undefined if this is mixed with SQL transaction statements.
-    fn commit(&mut self) -> Result<()>;
-
-    /// Roll back any pending transactions. Only used if autocommit is disabled.
-    ///
-    /// Behavior is undefined if this is mixed with SQL transaction statements.
-    fn rollback(&mut self) -> Result<()>;
-
-    /// Retrieve a given partition of data.
-    ///
-    /// A partition can be retrieved from [Statement::execute_partitions].
-    ///
-    /// # Arguments
-    ///
-    /// - `partition` - The partition descriptor.
-    fn read_partition(&self, partition: impl AsRef<[u8]>) -> Result<impl RecordBatchReader + Send>;
-}
-
-/// A handle to an ADBC statement.
-///
-/// A statement is a container for all state needed to execute a database query,
-/// such as the query itself, parameters for prepared statements, driver
-/// parameters, etc.
-///
-/// Statements may represent queries or prepared statements.
-///
-/// Statements may be used multiple times and can be reconfigured
-/// (e.g. they can be reused to execute multiple different queries).
-/// However, executing a statement (and changing certain other state)
-/// will invalidate result sets obtained prior to that execution.
-///
-/// Multiple statements may be created from a single connection.
-/// However, the driver may block or error if they are used concurrently
-/// (whether from a single thread or multiple threads).
-pub trait Statement: Optionable<Option = OptionStatement> {
-    /// Bind Arrow data. This can be used for bulk inserts or prepared
-    /// statements.
-    fn bind(&mut self, batch: RecordBatch) -> Result<()>;
-
-    /// Bind Arrow data. This can be used for bulk inserts or prepared
-    /// statements.
-    // TODO(alexandreyc): should we use a generic here instead of a trait object?
-    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567750972
-    fn bind_stream(&mut self, reader: Box<dyn RecordBatchReader + Send>) -> Result<()>;
-
-    /// Execute a statement and get the results.
-    ///
-    /// This invalidates any prior result sets.
-    // TODO(alexandreyc): is the Send bound absolutely necessary? same question
-    // for all methods that return an impl RecordBatchReader
-    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567748242
-    fn execute(&mut self) -> Result<impl RecordBatchReader + Send>;
-
-    /// Execute a statement that doesn’t have a result set and get the number
-    /// of affected rows.
-    ///
-    /// This invalidates any prior result sets.
-    ///
-    /// # Result
-    ///
-    /// Will return the number of rows affected. If the affected row count is
-    /// unknown or unsupported by the database, will return `None`.
-    fn execute_update(&mut self) -> Result<Option<i64>>;
-
-    /// Get the schema of the result set of a query without executing it.
-    ///
-    /// This invalidates any prior result sets.
-    ///
-    /// Depending on the driver, this may require first executing
-    /// [Statement::prepare].
-    ///
-    /// # Since
-    ///
-    /// ADBC API revision 1.1.0
-    fn execute_schema(&mut self) -> Result<Schema>;
-
-    /// Execute a statement and get the results as a partitioned result set.
-    fn execute_partitions(&mut self) -> Result<PartitionedResult>;
-
-    /// Get the schema for bound parameters.
-    ///
-    /// This retrieves an Arrow schema describing the number, names, and
-    /// types of the parameters in a parameterized statement. The fields
-    /// of the schema should be in order of the ordinal position of the
-    /// parameters; named parameters should appear only once.
-    ///
-    /// If the parameter does not have a name, or the name cannot be
-    /// determined, the name of the corresponding field in the schema will
-    /// be an empty string. If the type cannot be determined, the type of
-    /// the corresponding field will be NA (NullType).
-    ///
-    /// This should be called after [Statement::prepare].
-    fn get_parameter_schema(&self) -> Result<Schema>;
-
-    /// Turn this statement into a prepared statement to be executed multiple
-    /// times.
-    ///
-    /// This invalidates any prior result sets.
-    fn prepare(&mut self) -> Result<()>;
-
-    /// Set the SQL query to execute.
-    ///
-    /// The query can then be executed with [Statement::execute]. For queries
-    /// expected to be executed repeatedly, call [Statement::prepare] first.
-    fn set_sql_query(&mut self, query: impl AsRef<str>) -> Result<()>;
-
-    /// Set the Substrait plan to execute.
-    ///
-    /// The query can then be executed with [Statement::execute]. For queries
-    /// expected to be executed repeatedly, call [Statement::prepare] first.
-    fn set_substrait_plan(&mut self, plan: impl AsRef<[u8]>) -> Result<()>;
-
-    /// Cancel execution of an in-progress query.
-    ///
-    /// This can be called during [Statement::execute] (or similar), or while
-    /// consuming a result set returned from such.
-    ///
-    /// # Since
-    ///
-    /// ADBC API revision 1.1.0
-    fn cancel(&mut self) -> Result<()>;
-}
-
 /// Each data partition is described by an opaque byte array and can be
-/// retrieved with [Connection::read_partition].
+/// retrieved with [crate::Connection::read_partition].
 pub type Partitions = Vec<Vec<u8>>;
 
-/// A partitioned result set as returned by [Statement::execute_partitions].
+/// A partitioned result set as returned by [crate::Statement::execute_partitions].
 #[derive(Debug, PartialEq, Eq)]
 pub struct PartitionedResult {
     /// The result partitions.
     pub partitions: Partitions,
     /// The schema of the result set.
-    pub schema: Schema,
+    pub schema: arrow_schema::Schema,
     /// The number of rows affected if known, else -1.
     pub rows_affected: i64,
 }
diff --git a/rust/core/src/non_blocking.rs b/rust/core/src/non_blocking.rs
new file mode 100644
index 0000000000..f8972f0736
--- /dev/null
+++ b/rust/core/src/non_blocking.rs
@@ -0,0 +1,24 @@
+// 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.
+
+pub mod local;
+pub mod send;
+pub mod stream;
+
+pub use local::*;
+pub use send::*;
+pub use stream::*;
diff --git a/rust/core/src/non_blocking/local.rs b/rust/core/src/non_blocking/local.rs
new file mode 100644
index 0000000000..dcf18302b9
--- /dev/null
+++ b/rust/core/src/non_blocking/local.rs
@@ -0,0 +1,477 @@
+// 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 arrow_array::RecordBatch;
+use arrow_schema::Schema;
+
+use std::collections::HashSet;
+use std::pin::Pin;
+
+use crate::error::Result;
+use crate::options::*;
+
+/// Ability to configure an object by setting/getting options.
+#[allow(async_fn_in_trait)]
+pub trait LocalAsyncOptionable {
+    type Option: AsRef<str>;
+
+    /// Set a post-init option.
+    async fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()>;
+
+    /// Get a string option value by key.
+    async fn get_option_string(&self, key: Self::Option) -> Result<String>;
+
+    /// Get a bytes option value by key.
+    async fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>>;
+
+    /// Get an integer option value by key.
+    async fn get_option_int(&self, key: Self::Option) -> Result<i64>;
+
+    /// Get a float option value by key.
+    async fn get_option_double(&self, key: Self::Option) -> Result<f64>;
+}
+
+/// A handle to an async ADBC driver.
+#[allow(async_fn_in_trait)]
+pub trait LocalAsyncDriver {
+    type DatabaseType: LocalAsyncDatabase;
+
+    /// Allocate and initialize a new database without pre-init options.
+    async fn new_database(&mut self) -> Result<Self::DatabaseType>;
+
+    /// Allocate and initialize a new database with pre-init options.
+    async fn new_database_with_opts(
+        &mut self,
+        opts: Vec<(OptionDatabase, OptionValue)>,
+    ) -> Result<Self::DatabaseType>;
+}
+
+/// A handle to an async ADBC database.
+///
+/// Databases hold state shared by multiple connections. This typically means
+/// configuration and caches. For in-memory databases, it provides a place to
+/// hold ownership of the in-memory database.
+///
+/// Databases must be kept alive as long as any connections exist.
+#[allow(async_fn_in_trait)]
+pub trait LocalAsyncDatabase: LocalAsyncOptionable<Option = OptionDatabase> {
+    type ConnectionType: LocalAsyncConnection;
+
+    /// Allocate and initialize a new connection without pre-init options.
+    async fn new_connection(&self) -> Result<Self::ConnectionType>;
+
+    /// Allocate and initialize a new connection with pre-init options.
+    async fn new_connection_with_opts(
+        &self,
+        opts: Vec<(OptionConnection, OptionValue)>,
+    ) -> Result<Self::ConnectionType>;
+}
+
+/// A handle to an async ADBC connection.
+///
+/// Connections provide methods for query execution, managing prepared
+/// statements, using transactions, and so on.
+///
+/// # Autocommit
+///
+/// Connections should start in autocommit mode. They can be moved out by
+/// setting [crate::options::OptionConnection::AutoCommit] to "false". Turning off
+/// autocommit allows customizing the isolation level.
+#[allow(async_fn_in_trait)]
+pub trait LocalAsyncConnection: LocalAsyncOptionable<Option = OptionConnection> {
+    type StatementType: LocalAsyncStatement;
+
+    /// Allocate and initialize a new statement.
+    async fn new_statement(&mut self) -> Result<Self::StatementType>;
+
+    /// Cancel the in-progress operation on a connection.
+    async fn cancel(&mut self) -> Result<()>;
+
+    /// Get metadata about the database/driver.
+    ///
+    /// # Arguments
+    ///
+    /// - `codes` - Requested metadata. If `None`, retrieve all available metadata.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name                  | Field Type
+    /// ----------------------------|------------------------
+    /// info_name                   | uint32 not null
+    /// info_value                  | INFO_SCHEMA
+    ///
+    /// INFO_SCHEMA is a dense union with members:
+    ///
+    /// Field Name (Type Code)      | Field Type
+    /// ----------------------------|------------------------
+    /// string_value (0)            | utf8
+    /// bool_value (1)              | bool
+    /// int64_value (2)             | int64
+    /// int32_bitmask (3)           | int32
+    /// string_list (4)             | list\<utf8\>
+    /// int32_to_int32_list_map (5) | map\<int32, list\<int32\>\>
+    async fn get_info(
+        &self,
+        codes: Option<HashSet<InfoCode>>,
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Get a hierarchical view of all catalogs, database schemas, tables, and
+    /// columns.
+    ///
+    /// # Arguments
+    ///
+    /// - `depth` - The level of nesting to query.
+    /// - `catalog` - Only show tables in the given catalog. If `None`,
+    ///   do not filter by catalog. If an empty string, only show tables
+    ///   without a catalog.  May be a search pattern.
+    /// - `db_schema` - Only show tables in the given database schema. If
+    ///   `None`, do not filter by database schema. If an empty string, only show
+    ///   tables without a database schema. May be a search pattern.
+    /// - `table_name` - Only show tables with the given name. If `None`, do not
+    ///   filter by name. May be a search pattern.
+    /// - `table_type` - Only show tables matching one of the given table
+    ///   types. If `None`, show tables of any type. Valid table types can be fetched
+    ///   from [LocalAsyncConnection::get_table_types].
+    /// - `column_name` - Only show columns with the given name. If
+    ///   `None`, do not filter by name.  May be a search pattern..
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type               |
+    /// |--------------------------|--------------------------|
+    /// | catalog_name             | utf8                     |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> |
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | db_schema_name           | utf8                    |
+    /// | db_schema_tables         | list\<TABLE_SCHEMA\>    |
+    ///
+    /// TABLE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                |
+    /// |--------------------------|---------------------------|
+    /// | table_name               | utf8 not null             |
+    /// | table_type               | utf8 not null             |
+    /// | table_columns            | list\<COLUMN_SCHEMA\>     |
+    /// | table_constraints        | list\<CONSTRAINT_SCHEMA\> |
+    ///
+    /// COLUMN_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | column_name              | utf8 not null           |          |
+    /// | ordinal_position         | int32                   | (1)      |
+    /// | remarks                  | utf8                    | (2)      |
+    /// | xdbc_data_type           | int16                   | (3)      |
+    /// | xdbc_type_name           | utf8                    | (3)      |
+    /// | xdbc_column_size         | int32                   | (3)      |
+    /// | xdbc_decimal_digits      | int16                   | (3)      |
+    /// | xdbc_num_prec_radix      | int16                   | (3)      |
+    /// | xdbc_nullable            | int16                   | (3)      |
+    /// | xdbc_column_def          | utf8                    | (3)      |
+    /// | xdbc_sql_data_type       | int16                   | (3)      |
+    /// | xdbc_datetime_sub        | int16                   | (3)      |
+    /// | xdbc_char_octet_length   | int32                   | (3)      |
+    /// | xdbc_is_nullable         | utf8                    | (3)      |
+    /// | xdbc_scope_catalog       | utf8                    | (3)      |
+    /// | xdbc_scope_schema        | utf8                    | (3)      |
+    /// | xdbc_scope_table         | utf8                    | (3)      |
+    /// | xdbc_is_autoincrement    | bool                    | (3)      |
+    /// | xdbc_is_generatedcolumn  | bool                    | (3)      |
+    ///
+    /// 1. The column's ordinal position in the table (starting from 1).
+    /// 2. Database-specific description of the column.
+    /// 3. Optional value.  Should be null if not supported by the driver.
+    ///    `xdbc_` values are meant to provide JDBC/ODBC-compatible metadata
+    ///    in an agnostic manner.
+    ///
+    /// CONSTRAINT_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | constraint_name          | utf8                    |          |
+    /// | constraint_type          | utf8 not null           | (1)      |
+    /// | constraint_column_names  | list\<utf8\> not null     | (2)      |
+    /// | constraint_column_usage  | list\<USAGE_SCHEMA\>      | (3)      |
+    ///
+    /// 1. One of `CHECK`, `FOREIGN KEY`, `PRIMARY KEY`, or `UNIQUE`.
+    /// 2. The columns on the current table that are constrained, in
+    ///    order.
+    /// 3. For `FOREIGN KEY` only, the referenced table and columns.
+    ///
+    /// USAGE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | fk_catalog               | utf8                    |
+    /// | fk_db_schema             | utf8                    |
+    /// | fk_table                 | utf8 not null           |
+    /// | fk_column_name           | utf8 not null           |
+    async 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<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Get the Arrow schema of a table.
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable).
+    /// - `db_schema` - The database schema (or `None` if not applicable).
+    /// - `table_name` - The table name.
+    async fn get_table_schema(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: &str,
+    ) -> Result<Schema>;
+
+    /// Get a list of table types in the database.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|--------------
+    /// table_type     | utf8 not null
+    async fn get_table_types(&self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Get the names of statistics specific to this driver.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|----------------
+    /// statistic_name | utf8 not null
+    /// statistic_key  | int16 not null
+    ///
+    /// # Since
+    /// ADBC API revision 1.1.0
+    async fn get_statistic_names(&self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Get statistics about the data distribution of table(s).
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable). May be a search pattern.
+    /// - `db_schema` - The database schema (or `None` if not applicable). May be a search pattern
+    /// - `table_name` - The table name (or `None` if not applicable). May be a search pattern
+    /// - `approximate` - If false, request exact values of statistics, else
+    ///   allow for best-effort, approximate, or cached values. The database may
+    ///   return approximate values regardless, as indicated in the result.
+    ///   Requesting exact values may be expensive or unsupported.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | catalog_name             | utf8                             |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> not null|
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                        |
+    /// |--------------------------|-----------------------------------|
+    /// | db_schema_name           | utf8                              |
+    /// | db_schema_statistics     | list\<STATISTICS_SCHEMA\> not null|
+    ///
+    /// STATISTICS_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                       | Comments |
+    /// |--------------------------|----------------------------------| -------- |
+    /// | table_name               | utf8 not null                    |          |
+    /// | column_name              | utf8                             | (1)      |
+    /// | statistic_key            | int16 not null                   | (2)      |
+    /// | statistic_value          | VALUE_SCHEMA not null            |          |
+    /// | statistic_is_approximate | bool not null                    | (3)      |
+    ///
+    /// 1. If null, then the statistic applies to the entire table.
+    /// 2. A dictionary-encoded statistic name (although we do not use the Arrow
+    ///    dictionary type). Values in [0, 1024) are reserved for ADBC.  Other
+    ///    values are for implementation-specific statistics.  For the definitions
+    ///    of predefined statistic types, see [crate::options::Statistics]. To get
+    ///    driver-specific statistic names, use [LocalAsyncConnection::get_statistic_names].
+    /// 3. If true, then the value is approximate or best-effort.
+    ///
+    /// VALUE_SCHEMA is a dense union with members:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | int64                    | int64                            |
+    /// | uint64                   | uint64                           |
+    /// | float64                  | float64                          |
+    /// | binary                   | binary                           |
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    async fn get_statistics(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        approximate: bool,
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Commit any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    async fn commit(&mut self) -> Result<()>;
+
+    /// Roll back any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    async fn rollback(&mut self) -> Result<()>;
+
+    /// Retrieve a given partition of data.
+    ///
+    /// A partition can be retrieved from [LocalAsyncStatement::execute_partitions].
+    ///
+    /// # Arguments
+    ///
+    /// - `partition` - The partition descriptor.
+    async fn read_partition(
+        &self,
+        partition: &[u8],
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+}
+
+/// A handle to an async ADBC statement.
+///
+/// A statement is a container for all state needed to execute a database query,
+/// such as the query itself, parameters for prepared statements, driver
+/// parameters, etc.
+///
+/// Statements may represent queries or prepared statements.
+///
+/// Statements may be used multiple times and can be reconfigured
+/// (e.g. they can be reused to execute multiple different queries).
+/// However, executing a statement (and changing certain other state)
+/// will invalidate result sets obtained prior to that execution.
+///
+/// Multiple statements may be created from a single connection.
+/// However, the driver may block or error if they are used concurrently
+/// (whether from a single thread or multiple threads).
+#[allow(async_fn_in_trait)]
+pub trait LocalAsyncStatement: LocalAsyncOptionable<Option = OptionStatement> {
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    async fn bind(&mut self, batch: RecordBatch) -> Result<()>;
+
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    // TODO(alexandreyc): should we use a generic here instead of a trait object?
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567750972
+    async fn bind_stream(
+        &mut self,
+        reader: Pin<Box<dyn super::RecordBatchStream + Send>>,
+    ) -> Result<()>;
+
+    /// Execute a statement and get the results.
+    ///
+    /// This invalidates any prior result sets.
+    // TODO(alexandreyc): is the Send bound absolutely necessary? same question
+    // for all methods that return an impl RecordBatchReader
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567748242
+    async fn execute(&mut self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>>;
+
+    /// Execute a statement that doesn’t have a result set and get the number
+    /// of affected rows.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// # Result
+    ///
+    /// Will return the number of rows affected. If the affected row count is
+    /// unknown or unsupported by the database, will return `None`.
+    async fn execute_update(&mut self) -> Result<Option<i64>>;
+
+    /// Get the schema of the result set of a query without executing it.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// Depending on the driver, this may require first executing
+    /// [LocalAsyncStatement::prepare].
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    async fn execute_schema(&mut self) -> Result<Schema>;
+
+    /// Execute a statement and get the results as a partitioned result set.
+    async fn execute_partitions(&mut self) -> Result<crate::PartitionedResult>;
+
+    /// Get the schema for bound parameters.
+    ///
+    /// This retrieves an Arrow schema describing the number, names, and
+    /// types of the parameters in a parameterized statement. The fields
+    /// of the schema should be in order of the ordinal position of the
+    /// parameters; named parameters should appear only once.
+    ///
+    /// If the parameter does not have a name, or the name cannot be
+    /// determined, the name of the corresponding field in the schema will
+    /// be an empty string. If the type cannot be determined, the type of
+    /// the corresponding field will be NA (NullType).
+    ///
+    /// This should be called after [LocalAsyncStatement::prepare].
+    async fn get_parameter_schema(&self) -> Result<Schema>;
+
+    /// Turn this statement into a prepared statement to be executed multiple
+    /// times.
+    ///
+    /// This invalidates any prior result sets.
+    async fn prepare(&mut self) -> Result<()>;
+
+    /// Set the SQL query to execute.
+    ///
+    /// The query can then be executed with [LocalAsyncStatement::execute]. For queries
+    /// expected to be executed repeatedly, call [LocalAsyncStatement::prepare] first.
+    async fn set_sql_query(&mut self, query: &str) -> Result<()>;
+
+    /// Set the Substrait plan to execute.
+    ///
+    /// The query can then be executed with [LocalAsyncStatement::execute]. For queries
+    /// expected to be executed repeatedly, call [LocalAsyncStatement::prepare] first.
+    async fn set_substrait_plan(&mut self, plan: &[u8]) -> Result<()>;
+
+    /// Cancel execution of an in-progress query.
+    ///
+    /// This can be called during [LocalAsyncStatement::execute] (or similar), or while
+    /// consuming a result set returned from such.
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    async fn cancel(&mut self) -> Result<()>;
+}
diff --git a/rust/core/src/non_blocking/send.rs b/rust/core/src/non_blocking/send.rs
new file mode 100644
index 0000000000..adb0e46691
--- /dev/null
+++ b/rust/core/src/non_blocking/send.rs
@@ -0,0 +1,679 @@
+// 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 arrow_array::RecordBatch;
+use arrow_schema::Schema;
+
+use std::collections::HashSet;
+use std::pin::Pin;
+
+use crate::error::Result;
+use crate::options::*;
+
+/// Ability to configure an object by setting/getting options.
+pub trait AsyncOptionable: Send {
+    type Option: AsRef<str> + Send;
+
+    /// Set a post-init option.
+    fn set_option(
+        &mut self,
+        key: Self::Option,
+        value: OptionValue,
+    ) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Get a string option value by key.
+    fn get_option_string(
+        &self,
+        key: Self::Option,
+    ) -> impl ::core::future::Future<Output = Result<String>> + Send;
+
+    /// Get a bytes option value by key.
+    fn get_option_bytes(
+        &self,
+        key: Self::Option,
+    ) -> impl ::core::future::Future<Output = Result<Vec<u8>>> + Send;
+
+    /// Get an integer option value by key.
+    fn get_option_int(
+        &self,
+        key: Self::Option,
+    ) -> impl ::core::future::Future<Output = Result<i64>> + Send;
+
+    /// Get a float option value by key.
+    fn get_option_double(
+        &self,
+        key: Self::Option,
+    ) -> impl ::core::future::Future<Output = Result<f64>> + Send;
+}
+
+impl<T: AsyncOptionable> super::LocalAsyncOptionable for T {
+    type Option = <Self as AsyncOptionable>::Option;
+
+    async fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
+        <Self as AsyncOptionable>::set_option(self, key, value).await
+    }
+
+    async fn get_option_string(&self, key: Self::Option) -> Result<String> {
+        <Self as AsyncOptionable>::get_option_string(self, key).await
+    }
+
+    async fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
+        <Self as AsyncOptionable>::get_option_bytes(self, key).await
+    }
+
+    async fn get_option_int(&self, key: Self::Option) -> Result<i64> {
+        <Self as AsyncOptionable>::get_option_int(self, key).await
+    }
+
+    async fn get_option_double(&self, key: Self::Option) -> Result<f64> {
+        <Self as AsyncOptionable>::get_option_double(self, key).await
+    }
+}
+
+/// A handle to an async ADBC driver
+pub trait AsyncDriver: Send {
+    type DatabaseType: AsyncDatabase;
+
+    /// Allocate and initialize a new database without pre-init options.
+    fn new_database(
+        &mut self,
+    ) -> impl core::future::Future<Output = Result<Self::DatabaseType>> + Send;
+
+    /// Allocate and initialize a new database with pre-init options.
+    fn new_database_with_opts(
+        &mut self,
+        opts: Vec<(OptionDatabase, OptionValue)>,
+    ) -> impl core::future::Future<Output = Result<Self::DatabaseType>> + Send;
+}
+
+impl<T: AsyncDriver> super::LocalAsyncDriver for T {
+    type DatabaseType = <Self as AsyncDriver>::DatabaseType;
+
+    async fn new_database(&mut self) -> Result<Self::DatabaseType> {
+        <Self as AsyncDriver>::new_database(self).await
+    }
+
+    async fn new_database_with_opts(
+        &mut self,
+        opts: Vec<(OptionDatabase, OptionValue)>,
+    ) -> Result<Self::DatabaseType> {
+        <Self as AsyncDriver>::new_database_with_opts(self, opts).await
+    }
+}
+
+/// A handle to an async ADBC database.
+///
+/// Databases hold state shared by multiple connections. This typically means
+/// configuration and caches. For in-memory databases, it provides a place to
+/// hold ownership of the in-memory database.
+///
+/// Databases must be kept alive as long as any connections exist.
+pub trait AsyncDatabase: AsyncOptionable<Option = OptionDatabase> + Send {
+    type ConnectionType: AsyncConnection;
+
+    /// Allocate and initialize a new connection without pre-init options.
+    fn new_connection(
+        &self,
+    ) -> impl core::future::Future<Output = Result<Self::ConnectionType>> + Send;
+
+    /// Allocate and initialize a new connection with pre-init options.
+    fn new_connection_with_opts(
+        &self,
+        opts: Vec<(OptionConnection, OptionValue)>,
+    ) -> impl core::future::Future<Output = Result<Self::ConnectionType>> + Send;
+}
+
+impl<T: AsyncDatabase> super::LocalAsyncDatabase for T {
+    type ConnectionType = <Self as AsyncDatabase>::ConnectionType;
+
+    async fn new_connection(&self) -> Result<Self::ConnectionType> {
+        <Self as AsyncDatabase>::new_connection(self).await
+    }
+
+    async fn new_connection_with_opts(
+        &self,
+        opts: Vec<(OptionConnection, OptionValue)>,
+    ) -> Result<Self::ConnectionType> {
+        <Self as AsyncDatabase>::new_connection_with_opts(self, opts).await
+    }
+}
+
+/// A handle to an async ADBC connection.
+///
+/// Connections provide methods for query execution, managing prepared
+/// statements, using transactions, and so on.
+///
+/// # Autocommit
+///
+/// Connections should start in autocommit mode. They can be moved out by
+/// setting [crate::options::OptionConnection::AutoCommit] to "false". Turning off
+/// autocommit allows customizing the isolation level.
+pub trait AsyncConnection: AsyncOptionable<Option = OptionConnection> + Send {
+    type StatementType: AsyncStatement;
+
+    /// Allocate and initialize a new statement.
+    fn new_statement(
+        &mut self,
+    ) -> impl core::future::Future<Output = Result<Self::StatementType>> + Send;
+
+    /// Cancel the in-progress operation on a connection.
+    fn cancel(&mut self) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Get metadata about the database/driver.
+    ///
+    /// # Arguments
+    ///
+    /// - `codes` - Requested metadata. If `None`, retrieve all available metadata.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name                  | Field Type
+    /// ----------------------------|------------------------
+    /// info_name                   | uint32 not null
+    /// info_value                  | INFO_SCHEMA
+    ///
+    /// INFO_SCHEMA is a dense union with members:
+    ///
+    /// Field Name (Type Code)      | Field Type
+    /// ----------------------------|------------------------
+    /// string_value (0)            | utf8
+    /// bool_value (1)              | bool
+    /// int64_value (2)             | int64
+    /// int32_bitmask (3)           | int32
+    /// string_list (4)             | list\<utf8\>
+    /// int32_to_int32_list_map (5) | map\<int32, list\<int32\>\>
+    fn get_info(
+        &self,
+        codes: Option<HashSet<InfoCode>>,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Get a hierarchical view of all catalogs, database schemas, tables, and
+    /// columns.
+    ///
+    /// # Arguments
+    ///
+    /// - `depth` - The level of nesting to query.
+    /// - `catalog` - Only show tables in the given catalog. If `None`,
+    ///   do not filter by catalog. If an empty string, only show tables
+    ///   without a catalog.  May be a search pattern.
+    /// - `db_schema` - Only show tables in the given database schema. If
+    ///   `None`, do not filter by database schema. If an empty string, only show
+    ///   tables without a database schema. May be a search pattern.
+    /// - `table_name` - Only show tables with the given name. If `None`, do not
+    ///   filter by name. May be a search pattern.
+    /// - `table_type` - Only show tables matching one of the given table
+    ///   types. If `None`, show tables of any type. Valid table types can be fetched
+    ///   from [AsyncConnection::get_table_types].
+    /// - `column_name` - Only show columns with the given name. If
+    ///   `None`, do not filter by name.  May be a search pattern..
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type               |
+    /// |--------------------------|--------------------------|
+    /// | catalog_name             | utf8                     |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> |
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | db_schema_name           | utf8                    |
+    /// | db_schema_tables         | list\<TABLE_SCHEMA\>    |
+    ///
+    /// TABLE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                |
+    /// |--------------------------|---------------------------|
+    /// | table_name               | utf8 not null             |
+    /// | table_type               | utf8 not null             |
+    /// | table_columns            | list\<COLUMN_SCHEMA\>     |
+    /// | table_constraints        | list\<CONSTRAINT_SCHEMA\> |
+    ///
+    /// COLUMN_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | column_name              | utf8 not null           |          |
+    /// | ordinal_position         | int32                   | (1)      |
+    /// | remarks                  | utf8                    | (2)      |
+    /// | xdbc_data_type           | int16                   | (3)      |
+    /// | xdbc_type_name           | utf8                    | (3)      |
+    /// | xdbc_column_size         | int32                   | (3)      |
+    /// | xdbc_decimal_digits      | int16                   | (3)      |
+    /// | xdbc_num_prec_radix      | int16                   | (3)      |
+    /// | xdbc_nullable            | int16                   | (3)      |
+    /// | xdbc_column_def          | utf8                    | (3)      |
+    /// | xdbc_sql_data_type       | int16                   | (3)      |
+    /// | xdbc_datetime_sub        | int16                   | (3)      |
+    /// | xdbc_char_octet_length   | int32                   | (3)      |
+    /// | xdbc_is_nullable         | utf8                    | (3)      |
+    /// | xdbc_scope_catalog       | utf8                    | (3)      |
+    /// | xdbc_scope_schema        | utf8                    | (3)      |
+    /// | xdbc_scope_table         | utf8                    | (3)      |
+    /// | xdbc_is_autoincrement    | bool                    | (3)      |
+    /// | xdbc_is_generatedcolumn  | bool                    | (3)      |
+    ///
+    /// 1. The column's ordinal position in the table (starting from 1).
+    /// 2. Database-specific description of the column.
+    /// 3. Optional value.  Should be null if not supported by the driver.
+    ///    `xdbc_` values are meant to provide JDBC/ODBC-compatible metadata
+    ///    in an agnostic manner.
+    ///
+    /// CONSTRAINT_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | constraint_name          | utf8                    |          |
+    /// | constraint_type          | utf8 not null           | (1)      |
+    /// | constraint_column_names  | list\<utf8\> not null     | (2)      |
+    /// | constraint_column_usage  | list\<USAGE_SCHEMA\>      | (3)      |
+    ///
+    /// 1. One of `CHECK`, `FOREIGN KEY`, `PRIMARY KEY`, or `UNIQUE`.
+    /// 2. The columns on the current table that are constrained, in
+    ///    order.
+    /// 3. For `FOREIGN KEY` only, the referenced table and columns.
+    ///
+    /// USAGE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | fk_catalog               | utf8                    |
+    /// | fk_db_schema             | utf8                    |
+    /// | fk_table                 | utf8 not null           |
+    /// | fk_column_name           | utf8 not null           |
+    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>,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Get the Arrow schema of a table.
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable).
+    /// - `db_schema` - The database schema (or `None` if not applicable).
+    /// - `table_name` - The table name.
+    fn get_table_schema(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: &str,
+    ) -> impl core::future::Future<Output = Result<Schema>> + Send;
+
+    /// Get a list of table types in the database.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|--------------
+    /// table_type     | utf8 not null
+    fn get_table_types(
+        &self,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Get the names of statistics specific to this driver.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|----------------
+    /// statistic_name | utf8 not null
+    /// statistic_key  | int16 not null
+    ///
+    /// # Since
+    /// ADBC API revision 1.1.0
+    fn get_statistic_names(
+        &self,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Get statistics about the data distribution of table(s).
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable). May be a search pattern.
+    /// - `db_schema` - The database schema (or `None` if not applicable). May be a search pattern
+    /// - `table_name` - The table name (or `None` if not applicable). May be a search pattern
+    /// - `approximate` - If false, request exact values of statistics, else
+    ///   allow for best-effort, approximate, or cached values. The database may
+    ///   return approximate values regardless, as indicated in the result.
+    ///   Requesting exact values may be expensive or unsupported.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | catalog_name             | utf8                             |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> not null|
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                        |
+    /// |--------------------------|-----------------------------------|
+    /// | db_schema_name           | utf8                              |
+    /// | db_schema_statistics     | list\<STATISTICS_SCHEMA\> not null|
+    ///
+    /// STATISTICS_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                       | Comments |
+    /// |--------------------------|----------------------------------| -------- |
+    /// | table_name               | utf8 not null                    |          |
+    /// | column_name              | utf8                             | (1)      |
+    /// | statistic_key            | int16 not null                   | (2)      |
+    /// | statistic_value          | VALUE_SCHEMA not null            |          |
+    /// | statistic_is_approximate | bool not null                    | (3)      |
+    ///
+    /// 1. If null, then the statistic applies to the entire table.
+    /// 2. A dictionary-encoded statistic name (although we do not use the Arrow
+    ///    dictionary type). Values in [0, 1024) are reserved for ADBC.  Other
+    ///    values are for implementation-specific statistics.  For the definitions
+    ///    of predefined statistic types, see [crate::options::Statistics]. To get
+    ///    driver-specific statistic names, use [AsyncConnection::get_statistic_names].
+    /// 3. If true, then the value is approximate or best-effort.
+    ///
+    /// VALUE_SCHEMA is a dense union with members:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | int64                    | int64                            |
+    /// | uint64                   | uint64                           |
+    /// | float64                  | float64                          |
+    /// | binary                   | binary                           |
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn get_statistics(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        approximate: bool,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Commit any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    fn commit(&mut self) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Roll back any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    fn rollback(&mut self) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Retrieve a given partition of data.
+    ///
+    /// A partition can be retrieved from [AsyncStatement::execute_partitions].
+    ///
+    /// # Arguments
+    ///
+    /// - `partition` - The partition descriptor.
+    fn read_partition(
+        &self,
+        partition: &[u8],
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+}
+
+impl<T: AsyncConnection> super::LocalAsyncConnection for T {
+    type StatementType = <Self as AsyncConnection>::StatementType;
+
+    async fn new_statement(&mut self) -> Result<Self::StatementType> {
+        <Self as AsyncConnection>::new_statement(self).await
+    }
+
+    async fn cancel(&mut self) -> Result<()> {
+        <Self as AsyncConnection>::cancel(self).await
+    }
+
+    async fn get_info(
+        &self,
+        codes: Option<HashSet<InfoCode>>,
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::get_info(self, codes).await
+    }
+
+    async 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<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::get_objects(
+            self,
+            depth,
+            catalog,
+            db_schema,
+            table_name,
+            table_type,
+            column_name,
+        )
+        .await
+    }
+
+    async fn get_table_schema(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: &str,
+    ) -> Result<Schema> {
+        <Self as AsyncConnection>::get_table_schema(self, catalog, db_schema, table_name).await
+    }
+
+    async fn get_table_types(&self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::get_table_types(self).await
+    }
+
+    async fn get_statistic_names(&self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::get_statistic_names(self).await
+    }
+
+    async fn get_statistics(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        approximate: bool,
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::get_statistics(self, catalog, db_schema, table_name, approximate)
+            .await
+    }
+
+    async fn commit(&mut self) -> Result<()> {
+        <Self as AsyncConnection>::commit(self).await
+    }
+
+    async fn rollback(&mut self) -> Result<()> {
+        <Self as AsyncConnection>::rollback(self).await
+    }
+
+    async fn read_partition(
+        &self,
+        partition: &[u8],
+    ) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncConnection>::read_partition(self, partition).await
+    }
+}
+
+pub trait AsyncStatement: AsyncOptionable<Option = OptionStatement> + Send {
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    fn bind(&mut self, batch: RecordBatch)
+        -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    // TODO(alexandreyc): should we use a generic here instead of a trait object?
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567750972
+    fn bind_stream(
+        &mut self,
+        reader: Pin<Box<dyn super::RecordBatchStream + Send>>,
+    ) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Execute a statement and get the results.
+    ///
+    /// This invalidates any prior result sets.
+    // TODO(alexandreyc): is the Send bound absolutely necessary? same question
+    // for all methods that return an impl RecordBatchReader
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567748242
+    fn execute(
+        &mut self,
+    ) -> impl core::future::Future<Output = Result<Pin<Box<dyn super::RecordBatchStream + Send>>>> + Send;
+
+    /// Execute a statement that doesn’t have a result set and get the number
+    /// of affected rows.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// # Result
+    ///
+    /// Will return the number of rows affected. If the affected row count is
+    /// unknown or unsupported by the database, will return `None`.
+    fn execute_update(&mut self) -> impl core::future::Future<Output = Result<Option<i64>>> + Send;
+
+    /// Get the schema of the result set of a query without executing it.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// Depending on the driver, this may require first executing
+    /// [AsyncStatement::prepare].
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn execute_schema(&mut self) -> impl core::future::Future<Output = Result<Schema>> + Send;
+
+    /// Execute a statement and get the results as a partitioned result set.
+    fn execute_partitions(
+        &mut self,
+    ) -> impl ::core::future::Future<Output = Result<crate::PartitionedResult>> + Send;
+
+    /// Get the schema for bound parameters.
+    ///
+    /// This retrieves an Arrow schema describing the number, names, and
+    /// types of the parameters in a parameterized statement. The fields
+    /// of the schema should be in order of the ordinal position of the
+    /// parameters; named parameters should appear only once.
+    ///
+    /// If the parameter does not have a name, or the name cannot be
+    /// determined, the name of the corresponding field in the schema will
+    /// be an empty string. If the type cannot be determined, the type of
+    /// the corresponding field will be NA (NullType).
+    ///
+    /// This should be called after [AsyncStatement::prepare].
+    fn get_parameter_schema(&self) -> impl core::future::Future<Output = Result<Schema>> + Send;
+
+    /// Turn this statement into a prepared statement to be executed multiple
+    /// times.
+    ///
+    /// This invalidates any prior result sets.
+    fn prepare(&mut self) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Set the SQL query to execute.
+    ///
+    /// The query can then be executed with [AsyncStatement::execute]. For queries
+    /// expected to be executed repeatedly, call [AsyncStatement::prepare] first.
+    fn set_sql_query(
+        &mut self,
+        query: &str,
+    ) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Set the Substrait plan to execute.
+    ///
+    /// The query can then be executed with [AsyncStatement::execute]. For queries
+    /// expected to be executed repeatedly, call [AsyncStatement::prepare] first.
+    fn set_substrait_plan(
+        &mut self,
+        plan: &[u8],
+    ) -> impl core::future::Future<Output = Result<()>> + Send;
+
+    /// Cancel execution of an in-progress query.
+    ///
+    /// This can be called during [AsyncStatement::execute] (or similar), or while
+    /// consuming a result set returned from such.
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn cancel(&mut self) -> impl core::future::Future<Output = Result<()>> + Send;
+}
+
+impl<T: AsyncStatement> super::LocalAsyncStatement for T {
+    async fn bind(&mut self, batch: RecordBatch) -> Result<()> {
+        <Self as AsyncStatement>::bind(self, batch).await
+    }
+
+    async fn bind_stream(
+        &mut self,
+        reader: Pin<Box<dyn super::RecordBatchStream + Send>>,
+    ) -> Result<()> {
+        <Self as AsyncStatement>::bind_stream(self, reader).await
+    }
+
+    async fn execute(&mut self) -> Result<Pin<Box<dyn super::RecordBatchStream + Send>>> {
+        <Self as AsyncStatement>::execute(self).await
+    }
+
+    async fn execute_update(&mut self) -> Result<Option<i64>> {
+        <Self as AsyncStatement>::execute_update(self).await
+    }
+
+    async fn execute_schema(&mut self) -> Result<Schema> {
+        <Self as AsyncStatement>::execute_schema(self).await
+    }
+
+    async fn execute_partitions(&mut self) -> Result<crate::PartitionedResult> {
+        <Self as AsyncStatement>::execute_partitions(self).await
+    }
+
+    async fn get_parameter_schema(&self) -> Result<Schema> {
+        <Self as AsyncStatement>::get_parameter_schema(self).await
+    }
+
+    async fn prepare(&mut self) -> Result<()> {
+        <Self as AsyncStatement>::prepare(self).await
+    }
+
+    async fn set_sql_query(&mut self, query: &str) -> Result<()> {
+        <Self as AsyncStatement>::set_sql_query(self, query).await
+    }
+
+    async fn set_substrait_plan(&mut self, plan: &[u8]) -> Result<()> {
+        <Self as AsyncStatement>::set_substrait_plan(self, plan).await
+    }
+
+    async fn cancel(&mut self) -> Result<()> {
+        <Self as AsyncStatement>::cancel(self).await
+    }
+}
diff --git a/rust/core/src/non_blocking/stream.rs b/rust/core/src/non_blocking/stream.rs
new file mode 100644
index 0000000000..f4e592d826
--- /dev/null
+++ b/rust/core/src/non_blocking/stream.rs
@@ -0,0 +1,27 @@
+// 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.
+
+/// Trait for types that stream [arrow_array::RecordBatch]
+pub trait RecordBatchStream:
+    futures::Stream<Item = Result<arrow_array::RecordBatch, arrow_schema::ArrowError>>
+{
+    /// Returns the schema of this `RecordBatchStream`.
+    ///
+    /// Implementation of this trait should guarantee that all `RecordBatch`'s returned by this
+    /// stream should have the same schema as returned from this method.
+    fn schema(&self) -> arrow_schema::SchemaRef;
+}
diff --git a/rust/core/src/sync.rs b/rust/core/src/sync.rs
new file mode 100644
index 0000000000..77b3fe3c2d
--- /dev/null
+++ b/rust/core/src/sync.rs
@@ -0,0 +1,466 @@
+// 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 std::collections::HashSet;
+
+use arrow_array::{RecordBatch, RecordBatchReader};
+use arrow_schema::Schema;
+
+use crate::error::Result;
+use crate::options::{self, OptionConnection, OptionDatabase, OptionStatement, OptionValue};
+
+/// Ability to configure an object by setting/getting options.
+pub trait Optionable {
+    type Option: AsRef<str>;
+
+    /// Set a post-init option.
+    fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()>;
+
+    /// Get a string option value by key.
+    fn get_option_string(&self, key: Self::Option) -> Result<String>;
+
+    /// Get a bytes option value by key.
+    fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>>;
+
+    /// Get an integer option value by key.
+    fn get_option_int(&self, key: Self::Option) -> Result<i64>;
+
+    /// Get a float option value by key.
+    fn get_option_double(&self, key: Self::Option) -> Result<f64>;
+}
+
+/// A handle to an ADBC driver.
+pub trait Driver {
+    type DatabaseType: Database;
+
+    /// Allocate and initialize a new database without pre-init options.
+    fn new_database(&mut self) -> Result<Self::DatabaseType>;
+
+    /// Allocate and initialize a new database with pre-init options.
+    fn new_database_with_opts(
+        &mut self,
+        opts: impl IntoIterator<Item = (OptionDatabase, OptionValue)>,
+    ) -> Result<Self::DatabaseType>;
+}
+
+/// A handle to an ADBC database.
+///
+/// Databases hold state shared by multiple connections. This typically means
+/// configuration and caches. For in-memory databases, it provides a place to
+/// hold ownership of the in-memory database.
+///
+/// Databases must be kept alive as long as any connections exist.
+pub trait Database: Optionable<Option = OptionDatabase> {
+    type ConnectionType: Connection;
+
+    /// Allocate and initialize a new connection without pre-init options.
+    fn new_connection(&self) -> Result<Self::ConnectionType>;
+
+    /// Allocate and initialize a new connection with pre-init options.
+    fn new_connection_with_opts(
+        &self,
+        opts: impl IntoIterator<Item = (options::OptionConnection, OptionValue)>,
+    ) -> Result<Self::ConnectionType>;
+}
+
+/// A handle to an ADBC connection.
+///
+/// Connections provide methods for query execution, managing prepared
+/// statements, using transactions, and so on.
+///
+/// # Autocommit
+///
+/// Connections should start in autocommit mode. They can be moved out by
+/// setting [options::OptionConnection::AutoCommit] to "false". Turning off
+/// autocommit allows customizing the isolation level.
+pub trait Connection: Optionable<Option = OptionConnection> {
+    type StatementType: Statement;
+
+    /// Allocate and initialize a new statement.
+    fn new_statement(&mut self) -> Result<Self::StatementType>;
+
+    /// Cancel the in-progress operation on a connection.
+    fn cancel(&mut self) -> Result<()>;
+
+    /// Get metadata about the database/driver.
+    ///
+    /// # Arguments
+    ///
+    /// - `codes` - Requested metadata. If `None`, retrieve all available metadata.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name                  | Field Type
+    /// ----------------------------|------------------------
+    /// info_name                   | uint32 not null
+    /// info_value                  | INFO_SCHEMA
+    ///
+    /// INFO_SCHEMA is a dense union with members:
+    ///
+    /// Field Name (Type Code)      | Field Type
+    /// ----------------------------|------------------------
+    /// string_value (0)            | utf8
+    /// bool_value (1)              | bool
+    /// int64_value (2)             | int64
+    /// int32_bitmask (3)           | int32
+    /// string_list (4)             | list\<utf8\>
+    /// int32_to_int32_list_map (5) | map\<int32, list\<int32\>\>
+    fn get_info(
+        &self,
+        codes: Option<HashSet<options::InfoCode>>,
+    ) -> Result<impl RecordBatchReader + Send>;
+
+    /// Get a hierarchical view of all catalogs, database schemas, tables, and
+    /// columns.
+    ///
+    /// # Arguments
+    ///
+    /// - `depth` - The level of nesting to query.
+    /// - `catalog` - Only show tables in the given catalog. If `None`,
+    ///   do not filter by catalog. If an empty string, only show tables
+    ///   without a catalog.  May be a search pattern.
+    /// - `db_schema` - Only show tables in the given database schema. If
+    ///   `None`, do not filter by database schema. If an empty string, only show
+    ///   tables without a database schema. May be a search pattern.
+    /// - `table_name` - Only show tables with the given name. If `None`, do not
+    ///   filter by name. May be a search pattern.
+    /// - `table_type` - Only show tables matching one of the given table
+    ///   types. If `None`, show tables of any type. Valid table types can be fetched
+    ///   from [Connection::get_table_types].
+    /// - `column_name` - Only show columns with the given name. If
+    ///   `None`, do not filter by name.  May be a search pattern..
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type               |
+    /// |--------------------------|--------------------------|
+    /// | catalog_name             | utf8                     |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> |
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | db_schema_name           | utf8                    |
+    /// | db_schema_tables         | list\<TABLE_SCHEMA\>    |
+    ///
+    /// TABLE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                |
+    /// |--------------------------|---------------------------|
+    /// | table_name               | utf8 not null             |
+    /// | table_type               | utf8 not null             |
+    /// | table_columns            | list\<COLUMN_SCHEMA\>     |
+    /// | table_constraints        | list\<CONSTRAINT_SCHEMA\> |
+    ///
+    /// COLUMN_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | column_name              | utf8 not null           |          |
+    /// | ordinal_position         | int32                   | (1)      |
+    /// | remarks                  | utf8                    | (2)      |
+    /// | xdbc_data_type           | int16                   | (3)      |
+    /// | xdbc_type_name           | utf8                    | (3)      |
+    /// | xdbc_column_size         | int32                   | (3)      |
+    /// | xdbc_decimal_digits      | int16                   | (3)      |
+    /// | xdbc_num_prec_radix      | int16                   | (3)      |
+    /// | xdbc_nullable            | int16                   | (3)      |
+    /// | xdbc_column_def          | utf8                    | (3)      |
+    /// | xdbc_sql_data_type       | int16                   | (3)      |
+    /// | xdbc_datetime_sub        | int16                   | (3)      |
+    /// | xdbc_char_octet_length   | int32                   | (3)      |
+    /// | xdbc_is_nullable         | utf8                    | (3)      |
+    /// | xdbc_scope_catalog       | utf8                    | (3)      |
+    /// | xdbc_scope_schema        | utf8                    | (3)      |
+    /// | xdbc_scope_table         | utf8                    | (3)      |
+    /// | xdbc_is_autoincrement    | bool                    | (3)      |
+    /// | xdbc_is_generatedcolumn  | bool                    | (3)      |
+    ///
+    /// 1. The column's ordinal position in the table (starting from 1).
+    /// 2. Database-specific description of the column.
+    /// 3. Optional value.  Should be null if not supported by the driver.
+    ///    `xdbc_` values are meant to provide JDBC/ODBC-compatible metadata
+    ///    in an agnostic manner.
+    ///
+    /// CONSTRAINT_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              | Comments |
+    /// |--------------------------|-------------------------|----------|
+    /// | constraint_name          | utf8                    |          |
+    /// | constraint_type          | utf8 not null           | (1)      |
+    /// | constraint_column_names  | list\<utf8\> not null     | (2)      |
+    /// | constraint_column_usage  | list\<USAGE_SCHEMA\>      | (3)      |
+    ///
+    /// 1. One of `CHECK`, `FOREIGN KEY`, `PRIMARY KEY`, or `UNIQUE`.
+    /// 2. The columns on the current table that are constrained, in
+    ///    order.
+    /// 3. For `FOREIGN KEY` only, the referenced table and columns.
+    ///
+    /// USAGE_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type              |
+    /// |--------------------------|-------------------------|
+    /// | fk_catalog               | utf8                    |
+    /// | fk_db_schema             | utf8                    |
+    /// | fk_table                 | utf8 not null           |
+    /// | fk_column_name           | utf8 not null           |
+    ///
+    fn get_objects(
+        &self,
+        depth: options::ObjectDepth,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        table_type: Option<Vec<&str>>,
+        column_name: Option<&str>,
+    ) -> Result<impl RecordBatchReader + Send>;
+
+    /// Get the Arrow schema of a table.
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable).
+    /// - `db_schema` - The database schema (or `None` if not applicable).
+    /// - `table_name` - The table name.
+    fn get_table_schema(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: &str,
+    ) -> Result<Schema>;
+
+    /// Get a list of table types in the database.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|--------------
+    /// table_type     | utf8 not null
+    fn get_table_types(&self) -> Result<impl RecordBatchReader + Send>;
+
+    /// Get the names of statistics specific to this driver.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// Field Name     | Field Type
+    /// ---------------|----------------
+    /// statistic_name | utf8 not null
+    /// statistic_key  | int16 not null
+    ///
+    /// # Since
+    /// ADBC API revision 1.1.0
+    fn get_statistic_names(&self) -> Result<impl RecordBatchReader + Send>;
+
+    /// Get statistics about the data distribution of table(s).
+    ///
+    /// # Arguments
+    ///
+    /// - `catalog` - The catalog (or `None` if not applicable). May be a search pattern.
+    /// - `db_schema` - The database schema (or `None` if not applicable). May be a search pattern
+    /// - `table_name` - The table name (or `None` if not applicable). May be a search pattern
+    /// - `approximate` - If false, request exact values of statistics, else
+    ///   allow for best-effort, approximate, or cached values. The database may
+    ///   return approximate values regardless, as indicated in the result.
+    ///   Requesting exact values may be expensive or unsupported.
+    ///
+    /// # Result
+    ///
+    /// The result is an Arrow dataset with the following schema:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | catalog_name             | utf8                             |
+    /// | catalog_db_schemas       | list\<DB_SCHEMA_SCHEMA\> not null|
+    ///
+    /// DB_SCHEMA_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                        |
+    /// |--------------------------|-----------------------------------|
+    /// | db_schema_name           | utf8                              |
+    /// | db_schema_statistics     | list\<STATISTICS_SCHEMA\> not null|
+    ///
+    /// STATISTICS_SCHEMA is a Struct with fields:
+    ///
+    /// | Field Name               | Field Type                       | Comments |
+    /// |--------------------------|----------------------------------| -------- |
+    /// | table_name               | utf8 not null                    |          |
+    /// | column_name              | utf8                             | (1)      |
+    /// | statistic_key            | int16 not null                   | (2)      |
+    /// | statistic_value          | VALUE_SCHEMA not null            |          |
+    /// | statistic_is_approximate | bool not null                    | (3)      |
+    ///
+    /// 1. If null, then the statistic applies to the entire table.
+    /// 2. A dictionary-encoded statistic name (although we do not use the Arrow
+    ///    dictionary type). Values in [0, 1024) are reserved for ADBC.  Other
+    ///    values are for implementation-specific statistics.  For the definitions
+    ///    of predefined statistic types, see [options::Statistics]. To get
+    ///    driver-specific statistic names, use [Connection::get_statistic_names].
+    /// 3. If true, then the value is approximate or best-effort.
+    ///
+    /// VALUE_SCHEMA is a dense union with members:
+    ///
+    /// | Field Name               | Field Type                       |
+    /// |--------------------------|----------------------------------|
+    /// | int64                    | int64                            |
+    /// | uint64                   | uint64                           |
+    /// | float64                  | float64                          |
+    /// | binary                   | binary                           |
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn get_statistics(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        approximate: bool,
+    ) -> Result<impl RecordBatchReader + Send>;
+
+    /// Commit any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    fn commit(&mut self) -> Result<()>;
+
+    /// Roll back any pending transactions. Only used if autocommit is disabled.
+    ///
+    /// Behavior is undefined if this is mixed with SQL transaction statements.
+    fn rollback(&mut self) -> Result<()>;
+
+    /// Retrieve a given partition of data.
+    ///
+    /// A partition can be retrieved from [Statement::execute_partitions].
+    ///
+    /// # Arguments
+    ///
+    /// - `partition` - The partition descriptor.
+    fn read_partition(&self, partition: &[u8]) -> Result<impl RecordBatchReader + Send>;
+}
+
+/// A handle to an ADBC statement.
+///
+/// A statement is a container for all state needed to execute a database query,
+/// such as the query itself, parameters for prepared statements, driver
+/// parameters, etc.
+///
+/// Statements may represent queries or prepared statements.
+///
+/// Statements may be used multiple times and can be reconfigured
+/// (e.g. they can be reused to execute multiple different queries).
+/// However, executing a statement (and changing certain other state)
+/// will invalidate result sets obtained prior to that execution.
+///
+/// Multiple statements may be created from a single connection.
+/// However, the driver may block or error if they are used concurrently
+/// (whether from a single thread or multiple threads).
+pub trait Statement: Optionable<Option = OptionStatement> {
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    fn bind(&mut self, batch: RecordBatch) -> Result<()>;
+
+    /// Bind Arrow data. This can be used for bulk inserts or prepared
+    /// statements.
+    // TODO(alexandreyc): should we use a generic here instead of a trait object?
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567750972
+    fn bind_stream(&mut self, reader: Box<dyn RecordBatchReader + Send>) -> Result<()>;
+
+    /// Execute a statement and get the results.
+    ///
+    /// This invalidates any prior result sets.
+    // TODO(alexandreyc): is the Send bound absolutely necessary? same question
+    // for all methods that return an impl RecordBatchReader
+    // See: https://github.com/apache/arrow-adbc/pull/1725#discussion_r1567748242
+    fn execute(&mut self) -> Result<impl RecordBatchReader + Send>;
+
+    /// Execute a statement that doesn’t have a result set and get the number
+    /// of affected rows.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// # Result
+    ///
+    /// Will return the number of rows affected. If the affected row count is
+    /// unknown or unsupported by the database, will return `None`.
+    fn execute_update(&mut self) -> Result<Option<i64>>;
+
+    /// Get the schema of the result set of a query without executing it.
+    ///
+    /// This invalidates any prior result sets.
+    ///
+    /// Depending on the driver, this may require first executing
+    /// [Statement::prepare].
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn execute_schema(&mut self) -> Result<Schema>;
+
+    /// Execute a statement and get the results as a partitioned result set.
+    fn execute_partitions(&mut self) -> Result<crate::PartitionedResult>;
+
+    /// Get the schema for bound parameters.
+    ///
+    /// This retrieves an Arrow schema describing the number, names, and
+    /// types of the parameters in a parameterized statement. The fields
+    /// of the schema should be in order of the ordinal position of the
+    /// parameters; named parameters should appear only once.
+    ///
+    /// If the parameter does not have a name, or the name cannot be
+    /// determined, the name of the corresponding field in the schema will
+    /// be an empty string. If the type cannot be determined, the type of
+    /// the corresponding field will be NA (NullType).
+    ///
+    /// This should be called after [Statement::prepare].
+    fn get_parameter_schema(&self) -> Result<Schema>;
+
+    /// Turn this statement into a prepared statement to be executed multiple
+    /// times.
+    ///
+    /// This invalidates any prior result sets.
+    fn prepare(&mut self) -> Result<()>;
+
+    /// Set the SQL query to execute.
+    ///
+    /// The query can then be executed with [Statement::execute]. For queries
+    /// expected to be executed repeatedly, call [Statement::prepare] first.
+    fn set_sql_query(&mut self, query: impl AsRef<str>) -> Result<()>;
+
+    /// Set the Substrait plan to execute.
+    ///
+    /// The query can then be executed with [Statement::execute]. For queries
+    /// expected to be executed repeatedly, call [Statement::prepare] first.
+    fn set_substrait_plan(&mut self, plan: impl AsRef<[u8]>) -> Result<()>;
+
+    /// Cancel execution of an in-progress query.
+    ///
+    /// This can be called during [Statement::execute] (or similar), or while
+    /// consuming a result set returned from such.
+    ///
+    /// # Since
+    ///
+    /// ADBC API revision 1.1.0
+    fn cancel(&mut self) -> Result<()>;
+}
diff --git a/rust/core/src/syncify.rs b/rust/core/src/syncify.rs
new file mode 100644
index 0000000000..45236a7fb1
--- /dev/null
+++ b/rust/core/src/syncify.rs
@@ -0,0 +1,439 @@
+// 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 std::{collections::HashSet, fmt::Debug, future::Future, pin::Pin, sync::Arc};
+
+use arrow_array::{RecordBatch, RecordBatchReader};
+use arrow_schema::{Schema, SchemaRef};
+use futures::StreamExt;
+
+use crate::{
+    error::Result,
+    non_blocking::{
+        LocalAsyncConnection, LocalAsyncDatabase, LocalAsyncDriver, LocalAsyncOptionable,
+        LocalAsyncStatement,
+    },
+    options::{
+        InfoCode, ObjectDepth, OptionConnection, OptionDatabase, OptionStatement, OptionValue,
+    },
+    Connection, Database, Driver, Optionable, Statement,
+};
+
+pub trait AsyncExecutor: Sized + Send + Sync {
+    type Config: Default;
+    type Error: Debug;
+
+    fn new(config: Self::Config) -> std::result::Result<Self, Self::Error>;
+    fn block_on<F: Future>(&self, future: F) -> F::Output;
+}
+
+pub struct SyncRecordBatchStream<A> {
+    executor: Arc<A>,
+    inner: Pin<Box<dyn crate::non_blocking::RecordBatchStream + Send>>,
+}
+
+pub struct SyncDriverWrapper<A, D> {
+    inner: D,
+    executor: Arc<A>,
+}
+
+impl<A: AsyncExecutor> Iterator for SyncRecordBatchStream<A> {
+    type Item = std::result::Result<RecordBatch, arrow_schema::ArrowError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.executor.block_on(self.inner.next())
+    }
+}
+
+impl<A: AsyncExecutor> RecordBatchReader for SyncRecordBatchStream<A> {
+    fn schema(&self) -> SchemaRef {
+        self.inner.schema()
+    }
+}
+
+pub struct WrapperRecordBatchReader {
+    inner: Box<dyn RecordBatchReader + Send>,
+}
+
+impl futures::Stream for WrapperRecordBatchReader {
+    type Item = std::result::Result<RecordBatch, arrow_schema::ArrowError>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        std::task::Poll::Ready(self.get_mut().inner.next())
+    }
+}
+
+impl crate::non_blocking::RecordBatchStream for WrapperRecordBatchReader {
+    fn schema(&self) -> arrow_schema::SchemaRef {
+        self.inner.schema()
+    }
+}
+
+impl<A: AsyncExecutor, D: Default> Default for SyncDriverWrapper<A, D> {
+    fn default() -> Self {
+        Self {
+            inner: Default::default(),
+            executor: Arc::new(A::new(Default::default()).expect("failed to create executor")),
+        }
+    }
+}
+
+impl<A, D> Driver for SyncDriverWrapper<A, D>
+where
+    A: AsyncExecutor,
+    D: LocalAsyncDriver,
+    D::DatabaseType: LocalAsyncDatabase + LocalAsyncOptionable<Option = OptionDatabase>,
+    <D::DatabaseType as LocalAsyncDatabase>::ConnectionType:
+        LocalAsyncConnection + LocalAsyncOptionable<Option = OptionConnection>,
+    <<D::DatabaseType as LocalAsyncDatabase>::ConnectionType as LocalAsyncConnection>::StatementType: LocalAsyncOptionable<Option = OptionStatement>,
+{
+    type DatabaseType = SyncDatabaseWrapper<A, D::DatabaseType>;
+
+    fn new_database(&mut self) -> Result<Self::DatabaseType> {
+        let db = self.executor.block_on(self.inner.new_database())?;
+
+        Ok(SyncDatabaseWrapper { inner: db, executor: self.executor.clone() })
+    }
+
+    fn new_database_with_opts(
+        &mut self,
+        opts: impl IntoIterator<Item = (OptionDatabase, OptionValue)>,
+    ) -> Result<Self::DatabaseType> {
+        let db = self.executor.block_on(self.inner.new_database_with_opts(opts.into_iter().collect()))?;
+
+        Ok(SyncDatabaseWrapper { inner: db, executor: self.executor.clone() })
+    }
+}
+
+pub struct SyncDatabaseWrapper<A, DB> {
+    inner: DB,
+    executor: Arc<A>,
+}
+
+impl<A, DB> Optionable for SyncDatabaseWrapper<A, DB>
+where
+    A: AsyncExecutor,
+    DB: LocalAsyncOptionable<Option = OptionDatabase>,
+{
+    type Option = OptionDatabase;
+
+    fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
+        self.executor.block_on(self.inner.set_option(key, value))
+    }
+
+    fn get_option_string(&self, key: Self::Option) -> Result<String> {
+        self.executor.block_on(self.inner.get_option_string(key))
+    }
+
+    fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
+        self.executor.block_on(self.inner.get_option_bytes(key))
+    }
+
+    fn get_option_int(&self, key: Self::Option) -> Result<i64> {
+        self.executor.block_on(self.inner.get_option_int(key))
+    }
+
+    fn get_option_double(&self, key: Self::Option) -> Result<f64> {
+        self.executor.block_on(self.inner.get_option_double(key))
+    }
+}
+
+impl<A, DB> Database for SyncDatabaseWrapper<A, DB>
+where
+    A: AsyncExecutor,
+    DB: LocalAsyncDatabase + LocalAsyncOptionable<Option = OptionDatabase>,
+    DB::ConnectionType: LocalAsyncConnection + LocalAsyncOptionable<Option = OptionConnection>,
+    <DB::ConnectionType as LocalAsyncConnection>::StatementType:
+        LocalAsyncOptionable<Option = OptionStatement>,
+{
+    type ConnectionType = SyncConnectionWrapper<A, DB::ConnectionType>;
+
+    fn new_connection(&self) -> Result<Self::ConnectionType> {
+        let conn = self.executor.block_on(self.inner.new_connection())?;
+
+        Ok(SyncConnectionWrapper {
+            inner: conn,
+            executor: self.executor.clone(),
+        })
+    }
+
+    fn new_connection_with_opts(
+        &self,
+        opts: impl IntoIterator<Item = (OptionConnection, OptionValue)>,
+    ) -> Result<Self::ConnectionType> {
+        let conn = self.executor.block_on(
+            self.inner
+                .new_connection_with_opts(opts.into_iter().collect()),
+        )?;
+
+        Ok(SyncConnectionWrapper {
+            inner: conn,
+            executor: self.executor.clone(),
+        })
+    }
+}
+
+pub struct SyncConnectionWrapper<A, Conn> {
+    inner: Conn,
+    executor: Arc<A>,
+}
+
+impl<A, Conn> Optionable for SyncConnectionWrapper<A, Conn>
+where
+    A: AsyncExecutor,
+    Conn: LocalAsyncOptionable<Option = OptionConnection>,
+{
+    type Option = OptionConnection;
+
+    fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
+        self.executor.block_on(self.inner.set_option(key, value))
+    }
+
+    fn get_option_string(&self, key: Self::Option) -> Result<String> {
+        self.executor.block_on(self.inner.get_option_string(key))
+    }
+
+    fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
+        self.executor.block_on(self.inner.get_option_bytes(key))
+    }
+
+    fn get_option_int(&self, key: Self::Option) -> Result<i64> {
+        self.executor.block_on(self.inner.get_option_int(key))
+    }
+
+    fn get_option_double(&self, key: Self::Option) -> Result<f64> {
+        self.executor.block_on(self.inner.get_option_double(key))
+    }
+}
+
+impl<A, Conn> Connection for SyncConnectionWrapper<A, Conn>
+where
+    A: AsyncExecutor,
+    Conn: LocalAsyncConnection + LocalAsyncOptionable<Option = OptionConnection>,
+    Conn::StatementType: LocalAsyncOptionable<Option = OptionStatement>,
+{
+    type StatementType = SyncStatementWrapper<A, Conn::StatementType>;
+
+    fn new_statement(&mut self) -> Result<Self::StatementType> {
+        let stmt = self.executor.block_on(self.inner.new_statement())?;
+
+        Ok(SyncStatementWrapper {
+            inner: stmt,
+            executor: self.executor.clone(),
+        })
+    }
+
+    fn cancel(&mut self) -> Result<()> {
+        self.executor.block_on(self.inner.cancel())
+    }
+
+    fn get_info(&self, codes: Option<HashSet<InfoCode>>) -> Result<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.get_info(codes))?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    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<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.get_objects(
+            depth,
+            catalog,
+            db_schema,
+            table_name,
+            table_type,
+            column_name,
+        ))?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    fn get_table_schema(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: &str,
+    ) -> Result<Schema> {
+        self.executor
+            .block_on(self.inner.get_table_schema(catalog, db_schema, table_name))
+    }
+
+    fn get_table_types(&self) -> Result<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.get_table_types())?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    fn get_statistic_names(&self) -> Result<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.get_statistic_names())?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    fn get_statistics(
+        &self,
+        catalog: Option<&str>,
+        db_schema: Option<&str>,
+        table_name: Option<&str>,
+        approximate: bool,
+    ) -> Result<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.get_statistics(
+            catalog,
+            db_schema,
+            table_name,
+            approximate,
+        ))?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    fn commit(&mut self) -> Result<()> {
+        self.executor.block_on(self.inner.commit())
+    }
+
+    fn rollback(&mut self) -> Result<()> {
+        self.executor.block_on(self.inner.rollback())
+    }
+
+    fn read_partition(&self, partition: &[u8]) -> Result<impl RecordBatchReader + Send> {
+        let reader = self
+            .executor
+            .block_on(self.inner.read_partition(partition))?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+}
+
+pub struct SyncStatementWrapper<A, T> {
+    inner: T,
+    executor: Arc<A>,
+}
+
+impl<A, Stmt> Optionable for SyncStatementWrapper<A, Stmt>
+where
+    A: AsyncExecutor,
+    Stmt: LocalAsyncOptionable<Option = OptionStatement>,
+{
+    type Option = OptionStatement;
+
+    fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
+        self.executor.block_on(self.inner.set_option(key, value))
+    }
+
+    fn get_option_string(&self, key: Self::Option) -> Result<String> {
+        self.executor.block_on(self.inner.get_option_string(key))
+    }
+
+    fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
+        self.executor.block_on(self.inner.get_option_bytes(key))
+    }
+
+    fn get_option_int(&self, key: Self::Option) -> Result<i64> {
+        self.executor.block_on(self.inner.get_option_int(key))
+    }
+
+    fn get_option_double(&self, key: Self::Option) -> Result<f64> {
+        self.executor.block_on(self.inner.get_option_double(key))
+    }
+}
+
+impl<A, Stmt> Statement for SyncStatementWrapper<A, Stmt>
+where
+    A: AsyncExecutor,
+    Stmt: LocalAsyncStatement + LocalAsyncOptionable<Option = OptionStatement>,
+{
+    fn bind(&mut self, batch: RecordBatch) -> Result<()> {
+        self.executor.block_on(self.inner.bind(batch))
+    }
+
+    fn bind_stream(&mut self, reader: Box<dyn RecordBatchReader + Send>) -> Result<()> {
+        self.executor.block_on(
+            self.inner
+                .bind_stream(Box::pin(WrapperRecordBatchReader { inner: reader })),
+        )
+    }
+
+    fn execute(&mut self) -> Result<impl RecordBatchReader + Send> {
+        let reader = self.executor.block_on(self.inner.execute())?;
+
+        Ok(SyncRecordBatchStream {
+            executor: self.executor.clone(),
+            inner: reader,
+        })
+    }
+
+    fn execute_update(&mut self) -> Result<Option<i64>> {
+        self.executor.block_on(self.inner.execute_update())
+    }
+
+    fn execute_schema(&mut self) -> Result<Schema> {
+        self.executor.block_on(self.inner.execute_schema())
+    }
+
+    fn execute_partitions(&mut self) -> Result<crate::PartitionedResult> {
+        self.executor.block_on(self.inner.execute_partitions())
+    }
+
+    fn get_parameter_schema(&self) -> Result<Schema> {
+        self.executor.block_on(self.inner.get_parameter_schema())
+    }
+
+    fn prepare(&mut self) -> Result<()> {
+        self.executor.block_on(self.inner.prepare())
+    }
+
+    fn set_sql_query(&mut self, query: impl AsRef<str>) -> Result<()> {
+        self.executor
+            .block_on(self.inner.set_sql_query(query.as_ref()))
+    }
+
+    fn set_substrait_plan(&mut self, plan: impl AsRef<[u8]>) -> Result<()> {
+        self.executor
+            .block_on(self.inner.set_substrait_plan(plan.as_ref()))
+    }
+
+    fn cancel(&mut self) -> Result<()> {
+        self.executor.block_on(self.inner.cancel())
+    }
+}
diff --git a/rust/driver/datafusion/Cargo.toml b/rust/driver/datafusion/Cargo.toml
index 8b0a1311b9..dab0ccbb55 100644
--- a/rust/driver/datafusion/Cargo.toml
+++ b/rust/driver/datafusion/Cargo.toml
@@ -38,6 +38,7 @@ arrow-buffer = "56.2.0"
 arrow-schema = "56.2.0"
 datafusion = { version = "50.3.0", default-features = false }
 datafusion-substrait = { version = "50.3.0", default-features = false }
+futures.workspace = true
 tokio = { version = "1.48", features = ["rt-multi-thread"] }
 prost = "0.13.5"
 
diff --git a/rust/driver/datafusion/README.md b/rust/driver/datafusion/README.md
index ed5bb9d4db..4d1ea5d0b0 100644
--- a/rust/driver/datafusion/README.md
+++ b/rust/driver/datafusion/README.md
@@ -27,7 +27,7 @@
 
 ## Example Usage
 
-```
+```rust
 use adbc_core::driver_manager::ManagedDriver;
 use adbc_core::options::AdbcVersion;
 use adbc_core::{Connection, Database, Driver, Statement};
diff --git a/rust/driver/datafusion/src/lib.rs b/rust/driver/datafusion/src/lib.rs
index c08ba28c70..0010970ceb 100644
--- a/rust/driver/datafusion/src/lib.rs
+++ b/rust/driver/datafusion/src/lib.rs
@@ -14,10 +14,15 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
-
 #![allow(refining_impl_trait)]
 
-use adbc_core::constants;
+use adbc_core::non_blocking::{AsyncConnection, AsyncDatabase, AsyncDriver, AsyncStatement};
+use adbc_core::syncify::AsyncExecutor;
+use adbc_core::{
+    constants,
+    non_blocking::{AsyncOptionable, RecordBatchStream},
+    options,
+};
 use datafusion::dataframe::DataFrameWriteOptions;
 use datafusion::datasource::TableType;
 use datafusion::prelude::*;
@@ -26,16 +31,16 @@ use datafusion_substrait::substrait::proto::Plan;
 use prost::Message;
 use std::fmt::Debug;
 use std::future::Future;
+use std::pin::Pin;
 use std::sync::Arc;
-use std::vec::IntoIter;
 
 use arrow_array::builder::{
     BooleanBuilder, Int32Builder, Int64Builder, ListBuilder, MapBuilder, MapFieldNames,
     StringBuilder, UInt32Builder,
 };
 use arrow_array::{
-    ArrayRef, BooleanArray, Int16Array, Int32Array, ListArray, RecordBatch, RecordBatchReader,
-    StringArray, StructArray, UnionArray,
+    ArrayRef, BooleanArray, Int16Array, Int32Array, ListArray, RecordBatch, StringArray,
+    StructArray, UnionArray,
 };
 use arrow_buffer::{OffsetBuffer, ScalarBuffer};
 use arrow_schema::{ArrowError, DataType, Field, SchemaRef};
@@ -45,7 +50,7 @@ use adbc_core::{
     options::{
         InfoCode, ObjectDepth, OptionConnection, OptionDatabase, OptionStatement, OptionValue,
     },
-    schemas, Connection, Database, Driver, Optionable, Statement,
+    schemas,
 };
 
 pub enum Runtime {
@@ -89,22 +94,25 @@ impl SingleBatchReader {
     }
 }
 
-impl Iterator for SingleBatchReader {
+impl futures::Stream for SingleBatchReader {
     type Item = std::result::Result<RecordBatch, ArrowError>;
 
-    fn next(&mut self) -> Option<Self::Item> {
-        Ok(self.batch.take()).transpose()
+    fn poll_next(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        std::task::Poll::Ready(Ok(self.get_mut().batch.take()).transpose())
     }
 }
 
-impl RecordBatchReader for SingleBatchReader {
+impl RecordBatchStream for SingleBatchReader {
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
 }
 
 pub struct DataFusionReader {
-    batches: IntoIter<RecordBatch>,
+    reader: datafusion::execution::SendableRecordBatchStream,
     schema: SchemaRef,
 }
 
@@ -113,73 +121,71 @@ impl DataFusionReader {
         let schema = df.schema().as_arrow().clone();
 
         Self {
-            batches: df.collect().await.unwrap().into_iter(),
+            reader: df.execute_stream().await.unwrap(),
             schema: schema.into(),
         }
     }
 }
 
-impl Iterator for DataFusionReader {
+impl futures::Stream for DataFusionReader {
     type Item = std::result::Result<RecordBatch, ArrowError>;
 
-    fn next(&mut self) -> Option<Self::Item> {
-        self.batches.next().map(Ok)
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        match self.get_mut().reader.as_mut().poll_next(cx) {
+            std::task::Poll::Ready(item) => {
+                std::task::Poll::Ready(item.map(|batch| batch.map_err(|e| e.into())))
+            }
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
     }
 }
 
-impl RecordBatchReader for DataFusionReader {
+impl RecordBatchStream for DataFusionReader {
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
 }
 
 #[derive(Default)]
-pub struct DataFusionDriver {
-    handle: Option<tokio::runtime::Handle>,
-}
+pub struct DataFusionDriver {}
 
 impl DataFusionDriver {
-    pub fn new(handle: Option<tokio::runtime::Handle>) -> Self {
-        Self { handle }
+    pub fn new() -> Self {
+        Self {}
     }
 }
 
-impl Driver for DataFusionDriver {
+impl AsyncDriver for DataFusionDriver {
     type DatabaseType = DataFusionDatabase;
 
-    fn new_database(&mut self) -> Result<Self::DatabaseType> {
-        Ok(Self::DatabaseType {
-            handle: self.handle.clone(),
-        })
+    async fn new_database(&mut self) -> Result<Self::DatabaseType> {
+        Ok(Self::DatabaseType {})
     }
 
-    fn new_database_with_opts(
+    async fn new_database_with_opts(
         &mut self,
-        opts: impl IntoIterator<
-            Item = (
-                adbc_core::options::OptionDatabase,
-                adbc_core::options::OptionValue,
-            ),
-        >,
+        opts: Vec<(
+            adbc_core::options::OptionDatabase,
+            adbc_core::options::OptionValue,
+        )>,
     ) -> adbc_core::error::Result<Self::DatabaseType> {
-        let mut database = Self::DatabaseType {
-            handle: self.handle.clone(),
-        };
+        let mut database = Self::DatabaseType {};
         for (key, value) in opts {
-            database.set_option(key, value)?;
+            database.set_option(key, value).await?;
         }
         Ok(database)
     }
 }
 
-pub struct DataFusionDatabase {
-    handle: Option<tokio::runtime::Handle>,
-}
+pub struct DataFusionDatabase {}
 
-impl Optionable for DataFusionDatabase {
+impl AsyncOptionable for DataFusionDatabase {
     type Option = OptionDatabase;
 
-    fn set_option(
+    async fn set_option(
         &mut self,
         key: Self::Option,
         _value: adbc_core::options::OptionValue,
@@ -190,28 +196,28 @@ impl Optionable for DataFusionDatabase {
         ))
     }
 
-    fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
+    async fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
+    async fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
+    async fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
+    async fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
@@ -219,40 +225,25 @@ impl Optionable for DataFusionDatabase {
     }
 }
 
-impl Database for DataFusionDatabase {
+impl AsyncDatabase for DataFusionDatabase {
     type ConnectionType = DataFusionConnection;
 
-    fn new_connection(&self) -> Result<Self::ConnectionType> {
+    async fn new_connection(&self) -> Result<Self::ConnectionType> {
         let ctx = SessionContext::new();
 
-        let runtime = Runtime::new(self.handle.clone()).unwrap();
-
-        Ok(DataFusionConnection {
-            runtime: Arc::new(runtime),
-            ctx: Arc::new(ctx),
-        })
+        Ok(DataFusionConnection { ctx: Arc::new(ctx) })
     }
 
-    fn new_connection_with_opts(
+    async fn new_connection_with_opts(
         &self,
-        opts: impl IntoIterator<
-            Item = (
-                adbc_core::options::OptionConnection,
-                adbc_core::options::OptionValue,
-            ),
-        >,
+        opts: Vec<(options::OptionConnection, OptionValue)>,
     ) -> adbc_core::error::Result<Self::ConnectionType> {
         let ctx = SessionContext::new();
 
-        let runtime = Runtime::new(self.handle.clone()).unwrap();
-
-        let mut connection = DataFusionConnection {
-            runtime: Arc::new(runtime),
-            ctx: Arc::new(ctx),
-        };
+        let mut connection = DataFusionConnection { ctx: Arc::new(ctx) };
 
         for (key, value) in opts {
-            connection.set_option(key, value)?;
+            connection.set_option(key, value).await?;
         }
 
         Ok(connection)
@@ -260,14 +251,13 @@ impl Database for DataFusionDatabase {
 }
 
 pub struct DataFusionConnection {
-    runtime: Arc<Runtime>,
     ctx: Arc<SessionContext>,
 }
 
-impl Optionable for DataFusionConnection {
+impl AsyncOptionable for DataFusionConnection {
     type Option = OptionConnection;
 
-    fn set_option(
+    async fn set_option(
         &mut self,
         key: Self::Option,
         value: adbc_core::options::OptionValue,
@@ -275,10 +265,8 @@ impl Optionable for DataFusionConnection {
         match key.as_ref() {
             constants::ADBC_CONNECTION_OPTION_CURRENT_CATALOG => match value {
                 OptionValue::String(value) => {
-                    self.runtime.block_on(async {
-                        let query = format!("SET datafusion.catalog.default_catalog = {value}");
-                        self.ctx.sql(query.as_str()).await.unwrap();
-                    });
+                    let query = format!("SET datafusion.catalog.default_catalog = {value}");
+                    self.ctx.sql(query.as_str()).await.unwrap();
                     Ok(())
                 }
                 _ => Err(Error::with_message_and_status(
@@ -288,10 +276,8 @@ impl Optionable for DataFusionConnection {
             },
             constants::ADBC_CONNECTION_OPTION_CURRENT_DB_SCHEMA => match value {
                 OptionValue::String(value) => {
-                    self.runtime.block_on(async {
-                        let query = format!("SET datafusion.catalog.default_schema = {value}");
-                        self.ctx.sql(query.as_str()).await.unwrap();
-                    });
+                    let query = format!("SET datafusion.catalog.default_schema = {value}");
+                    self.ctx.sql(query.as_str()).await.unwrap();
                     Ok(())
                 }
                 _ => Err(Error::with_message_and_status(
@@ -306,7 +292,7 @@ impl Optionable for DataFusionConnection {
         }
     }
 
-    fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
+    async fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
         match key.as_ref() {
             constants::ADBC_CONNECTION_OPTION_CURRENT_CATALOG => Ok(self
                 .ctx
@@ -329,21 +315,21 @@ impl Optionable for DataFusionConnection {
         }
     }
 
-    fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
+    async fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
+    async fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
+    async fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
@@ -470,16 +456,15 @@ impl GetObjectsBuilder {
         }
     }
 
-    pub fn build(
+    pub async fn build(
         &mut self,
-        runtime: &Runtime,
         ctx: &SessionContext,
         depth: &ObjectDepth,
     ) -> Result<RecordBatch> {
         let mut catalogs = ctx.catalog_names();
         self.catalog_names.append(&mut catalogs);
 
-        self.catalog_names.iter().for_each(|cat| {
+        for cat in &self.catalog_names {
             let catalog_provider = ctx.catalog(cat).unwrap();
             let schema_names = catalog_provider.schema_names();
             self.catalog_db_schema_names
@@ -488,35 +473,33 @@ impl GetObjectsBuilder {
             self.catalog_db_schema_offsets
                 .push(self.catalog_db_schema_offsets.last().unwrap() + schema_names.len() as i32);
 
-            schema_names.iter().for_each(|schema| {
+            for schema in &schema_names {
                 let schema_provider = catalog_provider.schema(schema).unwrap();
                 let table_names = schema_provider.table_names();
                 self.table_names.append(&mut table_names.clone());
                 self.table_offsets
                     .push(self.table_offsets.last().unwrap() + table_names.len() as i32);
 
-                table_names.iter().for_each(|t| {
-                    runtime.block_on(async {
-                        let table_provider = schema_provider.table(t).await.unwrap().unwrap();
-                        let table_type = match table_provider.table_type() {
-                            TableType::Base => "Base",
-                            TableType::View => "View",
-                            TableType::Temporary => "Temporary",
-                        };
-                        self.table_types.push(table_type.to_string());
-
-                        let schema = table_provider.schema();
-                        let num_fields = schema.fields().len();
-
-                        schema.fields().iter().for_each(|f| {
-                            self.column_names.push(f.name().clone());
-                        });
-                        self.column_offsets
-                            .push(self.column_offsets.last().unwrap() + num_fields as i32);
+                for t in &table_names {
+                    let table_provider = schema_provider.table(t).await.unwrap().unwrap();
+                    let table_type = match table_provider.table_type() {
+                        TableType::Base => "Base",
+                        TableType::View => "View",
+                        TableType::Temporary => "Temporary",
+                    };
+                    self.table_types.push(table_type.to_string());
+
+                    let schema = table_provider.schema();
+                    let num_fields = schema.fields().len();
+
+                    schema.fields().iter().for_each(|f| {
+                        self.column_names.push(f.name().clone());
                     });
-                });
-            });
-        });
+                    self.column_offsets
+                        .push(self.column_offsets.last().unwrap() + num_fields as i32);
+                }
+            }
+        }
 
         //////////////////////////////////////////////////////
 
@@ -721,12 +704,11 @@ impl GetObjectsBuilder {
     }
 }
 
-impl Connection for DataFusionConnection {
+impl AsyncConnection for DataFusionConnection {
     type StatementType = DataFusionStatement;
 
-    fn new_statement(&mut self) -> adbc_core::error::Result<Self::StatementType> {
+    async fn new_statement(&mut self) -> adbc_core::error::Result<Self::StatementType> {
         Ok(DataFusionStatement {
-            runtime: self.runtime.clone(),
             ctx: self.ctx.clone(),
             sql_query: None,
             substrait_plan: None,
@@ -735,14 +717,14 @@ impl Connection for DataFusionConnection {
         })
     }
 
-    fn cancel(&mut self) -> adbc_core::error::Result<()> {
+    async fn cancel(&mut self) -> adbc_core::error::Result<()> {
         todo!()
     }
 
-    fn get_info(
+    async fn get_info(
         &self,
         codes: Option<std::collections::HashSet<adbc_core::options::InfoCode>>,
-    ) -> Result<impl RecordBatchReader + Send> {
+    ) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
         let mut get_info_builder = GetInfoBuilder::new();
 
         codes.unwrap().into_iter().for_each(|f| match f {
@@ -755,10 +737,10 @@ impl Connection for DataFusionConnection {
 
         let batch = get_info_builder.finish()?;
         let reader = SingleBatchReader::new(batch);
-        Ok(reader)
+        Ok(Box::pin(reader))
     }
 
-    fn get_objects(
+    async fn get_objects(
         &self,
         depth: adbc_core::options::ObjectDepth,
         _catalog: Option<&str>,
@@ -766,13 +748,13 @@ impl Connection for DataFusionConnection {
         _table_name: Option<&str>,
         _table_type: Option<Vec<&str>>,
         _column_name: Option<&str>,
-    ) -> Result<impl RecordBatchReader + Send> {
-        let batch = GetObjectsBuilder::new().build(&self.runtime, &self.ctx, &depth)?;
+    ) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
+        let batch = GetObjectsBuilder::new().build(&self.ctx, &depth).await?;
         let reader = SingleBatchReader::new(batch);
-        Ok(reader)
+        Ok(Box::pin(reader))
     }
 
-    fn get_table_schema(
+    async fn get_table_schema(
         &self,
         _catalog: Option<&str>,
         _db_schema: Option<&str>,
@@ -781,39 +763,41 @@ impl Connection for DataFusionConnection {
         todo!()
     }
 
-    fn get_table_types(&self) -> Result<SingleBatchReader> {
+    async fn get_table_types(&self) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
         todo!()
     }
 
-    fn get_statistic_names(&self) -> Result<SingleBatchReader> {
+    async fn get_statistic_names(&self) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
         todo!()
     }
 
-    fn get_statistics(
+    async fn get_statistics(
         &self,
         _catalog: Option<&str>,
         _db_schema: Option<&str>,
         _table_name: Option<&str>,
         _approximate: bool,
-    ) -> Result<SingleBatchReader> {
+    ) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
         todo!()
     }
 
-    fn commit(&mut self) -> adbc_core::error::Result<()> {
+    async fn commit(&mut self) -> adbc_core::error::Result<()> {
         todo!()
     }
 
-    fn rollback(&mut self) -> adbc_core::error::Result<()> {
+    async fn rollback(&mut self) -> adbc_core::error::Result<()> {
         todo!()
     }
 
-    fn read_partition(&self, _partition: impl AsRef<[u8]>) -> Result<SingleBatchReader> {
+    async fn read_partition(
+        &self,
+        _partition: &[u8],
+    ) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
         todo!()
     }
 }
 
 pub struct DataFusionStatement {
-    runtime: Arc<Runtime>,
     ctx: Arc<SessionContext>,
     sql_query: Option<String>,
     substrait_plan: Option<Plan>,
@@ -821,10 +805,10 @@ pub struct DataFusionStatement {
     ingest_target_table: Option<String>,
 }
 
-impl Optionable for DataFusionStatement {
+impl AsyncOptionable for DataFusionStatement {
     type Option = OptionStatement;
 
-    fn set_option(
+    async fn set_option(
         &mut self,
         key: Self::Option,
         value: adbc_core::options::OptionValue,
@@ -847,7 +831,7 @@ impl Optionable for DataFusionStatement {
         }
     }
 
-    fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
+    async fn get_option_string(&self, key: Self::Option) -> adbc_core::error::Result<String> {
         match key.as_ref() {
             constants::ADBC_INGEST_OPTION_TARGET_TABLE => {
                 let target_table = self.ingest_target_table.clone();
@@ -866,21 +850,21 @@ impl Optionable for DataFusionStatement {
         }
     }
 
-    fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
+    async fn get_option_bytes(&self, key: Self::Option) -> adbc_core::error::Result<Vec<u8>> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
+    async fn get_option_int(&self, key: Self::Option) -> adbc_core::error::Result<i64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
         ))
     }
 
-    fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
+    async fn get_option_double(&self, key: Self::Option) -> adbc_core::error::Result<f64> {
         Err(Error::with_message_and_status(
             format!("Unrecognized option: {key:?}"),
             Status::NotFound,
@@ -888,104 +872,119 @@ impl Optionable for DataFusionStatement {
     }
 }
 
-impl Statement for DataFusionStatement {
-    fn bind(&mut self, batch: arrow_array::RecordBatch) -> adbc_core::error::Result<()> {
+impl AsyncStatement for DataFusionStatement {
+    async fn bind(&mut self, batch: arrow_array::RecordBatch) -> adbc_core::error::Result<()> {
         self.bound_record_batch.replace(batch);
         Ok(())
     }
 
-    fn bind_stream(
+    async fn bind_stream(
         &mut self,
-        _reader: Box<dyn arrow_array::RecordBatchReader + Send>,
+        _reader: Pin<Box<dyn RecordBatchStream + Send>>,
     ) -> adbc_core::error::Result<()> {
         todo!()
     }
 
-    fn execute(&mut self) -> Result<impl RecordBatchReader + Send> {
-        self.runtime.block_on(async {
-            let df = if self.sql_query.is_some() {
-                self.ctx
-                    .sql(&self.sql_query.clone().unwrap())
+    async fn execute(&mut self) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
+        let df = if self.sql_query.is_some() {
+            self.ctx
+                .sql(&self.sql_query.clone().unwrap())
+                .await
+                .unwrap()
+        } else {
+            let plan =
+                from_substrait_plan(&self.ctx.state(), &self.substrait_plan.clone().unwrap())
                     .await
-                    .unwrap()
-            } else {
-                let plan =
-                    from_substrait_plan(&self.ctx.state(), &self.substrait_plan.clone().unwrap())
-                        .await
-                        .unwrap();
-                self.ctx.execute_logical_plan(plan).await.unwrap()
-            };
+                    .unwrap();
+            self.ctx.execute_logical_plan(plan).await.unwrap()
+        };
 
-            Ok(DataFusionReader::new(df).await)
-        })
+        Ok(Box::pin(DataFusionReader::new(df).await))
     }
 
-    fn execute_update(&mut self) -> adbc_core::error::Result<Option<i64>> {
+    async fn execute_update(&mut self) -> adbc_core::error::Result<Option<i64>> {
         if self.sql_query.is_some() {
-            self.runtime.block_on(async {
-                let _ = self
-                    .ctx
-                    .sql(&self.sql_query.clone().unwrap())
-                    .await
-                    .unwrap();
-            });
+            let _ = self
+                .ctx
+                .sql(&self.sql_query.clone().unwrap())
+                .await
+                .unwrap();
         } else if let Some(batch) = self.bound_record_batch.take() {
-            self.runtime.block_on(async {
-                let table = match self.ingest_target_table.clone() {
-                    Some(table) => table,
-                    None => todo!(),
-                };
-
-                self.ctx
-                    .read_batch(batch)
-                    .unwrap()
-                    .write_table(table.as_str(), DataFrameWriteOptions::new())
-                    .await
-                    .unwrap();
-            });
+            let table = match self.ingest_target_table.clone() {
+                Some(table) => table,
+                None => todo!(),
+            };
+
+            self.ctx
+                .read_batch(batch)
+                .unwrap()
+                .write_table(table.as_str(), DataFrameWriteOptions::new())
+                .await
+                .unwrap();
         }
 
         Ok(Some(0))
     }
 
-    fn execute_schema(&mut self) -> adbc_core::error::Result<arrow_schema::Schema> {
-        self.runtime.block_on(async {
-            let df = self
-                .ctx
-                .sql(&self.sql_query.clone().unwrap())
-                .await
-                .unwrap();
+    async fn execute_schema(&mut self) -> adbc_core::error::Result<arrow_schema::Schema> {
+        let df = self
+            .ctx
+            .sql(&self.sql_query.clone().unwrap())
+            .await
+            .unwrap();
 
-            Ok(df.schema().as_arrow().clone())
-        })
+        Ok(df.schema().as_arrow().clone())
     }
 
-    fn execute_partitions(&mut self) -> adbc_core::error::Result<adbc_core::PartitionedResult> {
+    async fn execute_partitions(
+        &mut self,
+    ) -> adbc_core::error::Result<adbc_core::PartitionedResult> {
         todo!()
     }
 
-    fn get_parameter_schema(&self) -> adbc_core::error::Result<arrow_schema::Schema> {
+    async fn get_parameter_schema(&self) -> adbc_core::error::Result<arrow_schema::Schema> {
         todo!()
     }
 
-    fn prepare(&mut self) -> adbc_core::error::Result<()> {
+    async fn prepare(&mut self) -> adbc_core::error::Result<()> {
         todo!()
     }
 
-    fn set_sql_query(&mut self, query: impl AsRef<str>) -> adbc_core::error::Result<()> {
-        self.sql_query = Some(query.as_ref().to_string());
+    async fn set_sql_query(&mut self, query: &str) -> adbc_core::error::Result<()> {
+        self.sql_query = Some(query.to_string());
         Ok(())
     }
 
-    fn set_substrait_plan(&mut self, plan: impl AsRef<[u8]>) -> adbc_core::error::Result<()> {
-        self.substrait_plan = Some(Plan::decode(plan.as_ref()).unwrap());
+    async fn set_substrait_plan(&mut self, plan: &[u8]) -> adbc_core::error::Result<()> {
+        self.substrait_plan = Some(Plan::decode(plan).unwrap());
         Ok(())
     }
 
-    fn cancel(&mut self) -> adbc_core::error::Result<()> {
+    async fn cancel(&mut self) -> adbc_core::error::Result<()> {
         todo!()
     }
 }
 
+pub struct TokioRuntime(tokio::runtime::Runtime);
+
+impl AsyncExecutor for TokioRuntime {
+    type Config = ();
+    type Error = std::io::Error;
+
+    fn new(_: Self::Config) -> std::result::Result<Self, Self::Error> {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .map(TokioRuntime)
+    }
+
+    fn block_on<F: Future>(&self, future: F) -> F::Output {
+        self.0.block_on(future)
+    }
+}
+
 #[cfg(feature = "ffi")]
-adbc_ffi::export_driver!(DataFusionDriverInit, DataFusionDriver);
+adbc_ffi::export_driver!(
+    DataFusionDriverInit,
+    adbc_core::syncify::SyncDriverWrapper::<TokioRuntime, DataFusionDriver>
+);
diff --git a/rust/driver/datafusion/tests/test_datafusion.rs b/rust/driver/datafusion/tests/test_datafusion.rs
index 9123a480a9..c5e9159d6a 100644
--- a/rust/driver/datafusion/tests/test_datafusion.rs
+++ b/rust/driver/datafusion/tests/test_datafusion.rs
@@ -15,7 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use adbc_core::{Connection, Database, Driver, Optionable, Statement};
+use std::pin::Pin;
+
+use adbc_core::non_blocking::RecordBatchStream;
+use adbc_core::non_blocking::{
+    AsyncConnection, AsyncDatabase, AsyncDriver, AsyncStatement, LocalAsyncOptionable,
+};
 use adbc_datafusion::{DataFusionConnection, DataFusionDriver};
 use arrow_array::RecordBatch;
 use datafusion::prelude::*;
@@ -24,176 +29,199 @@ use adbc_core::options::{OptionConnection, OptionStatement, OptionValue};
 use arrow_select::concat::concat_batches;
 use datafusion_substrait::logical_plan::producer::to_substrait_plan;
 use datafusion_substrait::substrait::proto::Plan;
+use futures::StreamExt;
 use prost::Message;
 
-fn get_connection(handle: Option<tokio::runtime::Handle>) -> DataFusionConnection {
-    let mut driver = DataFusionDriver::new(handle);
-    let database = driver.new_database().unwrap();
-    database.new_connection().unwrap()
+async fn read_all_record_batches(
+    mut reader: Pin<Box<dyn RecordBatchStream + Send>>,
+) -> adbc_core::error::Result<Vec<RecordBatch>> {
+    let mut batches = Vec::new();
+    while let Some(batch) = reader.next().await {
+        batches.push(batch?);
+    }
+    Ok(batches)
+}
+
+async fn get_connection() -> DataFusionConnection {
+    let mut driver = DataFusionDriver::default();
+    let database = driver.new_database().await.unwrap();
+    database.new_connection().await.unwrap()
 }
 
-fn get_objects(connection: &DataFusionConnection) -> RecordBatch {
-    let objects = connection.get_objects(
-        adbc_core::options::ObjectDepth::All,
-        None,
-        None,
-        None,
-        None,
-        None,
-    );
+async fn get_objects(connection: &DataFusionConnection) -> RecordBatch {
+    let objects = connection
+        .get_objects(
+            adbc_core::options::ObjectDepth::All,
+            None,
+            None,
+            None,
+            None,
+            None,
+        )
+        .await;
 
-    let batches: Vec<RecordBatch> = objects.unwrap().map(|b| b.unwrap()).collect();
+    let batches = read_all_record_batches(objects.unwrap()).await.unwrap();
 
     let schema = batches.first().unwrap().schema();
 
     concat_batches(&schema, &batches).unwrap()
 }
 
-fn execute_update(connection: &mut DataFusionConnection, query: &str) {
-    let mut statement = connection.new_statement().unwrap();
-    let _ = statement.set_sql_query(query);
-    let _ = statement.execute_update();
+async fn execute_update(connection: &mut DataFusionConnection, query: &str) {
+    let mut statement = connection.new_statement().await.unwrap();
+    let _ = statement.set_sql_query(query).await;
+    let _ = statement.execute_update().await;
 }
 
-fn execute_sql_query(connection: &mut DataFusionConnection, query: &str) -> RecordBatch {
-    let mut statement = connection.new_statement().unwrap();
-    let _ = statement.set_sql_query(query);
+async fn execute_sql_query(connection: &mut DataFusionConnection, query: &str) -> RecordBatch {
+    let mut statement = connection.new_statement().await.unwrap();
+    let _ = statement.set_sql_query(query).await;
 
-    let batches: Vec<RecordBatch> = statement.execute().unwrap().map(|b| b.unwrap()).collect();
+    let batches = read_all_record_batches(statement.execute().await.unwrap())
+        .await
+        .unwrap();
 
     let schema = batches.first().unwrap().schema();
 
     concat_batches(&schema, &batches).unwrap()
 }
 
-fn execute_substrait(connection: &mut DataFusionConnection, plan: Plan) -> RecordBatch {
-    let mut statement = connection.new_statement().unwrap();
+async fn execute_substrait(connection: &mut DataFusionConnection, plan: Plan) -> RecordBatch {
+    let mut statement = connection.new_statement().await.unwrap();
 
-    let _ = statement.set_substrait_plan(plan.encode_to_vec());
+    let _ = statement.set_substrait_plan(&plan.encode_to_vec()).await;
 
-    let batches: Vec<RecordBatch> = statement.execute().unwrap().map(|b| b.unwrap()).collect();
+    let batches = read_all_record_batches(statement.execute().await.unwrap())
+        .await
+        .unwrap();
 
     let schema = batches.first().unwrap().schema();
 
     concat_batches(&schema, &batches).unwrap()
 }
 
-#[test]
-fn test_connection_options() {
-    let mut connection = get_connection(None);
+#[tokio::test]
+async fn test_connection_options() {
+    let mut connection = get_connection().await;
 
     let current_catalog = connection
         .get_option_string(OptionConnection::CurrentCatalog)
+        .await
         .unwrap();
 
     assert_eq!(current_catalog, "datafusion");
 
-    let _ = connection.set_option(
-        OptionConnection::CurrentCatalog,
-        OptionValue::String("datafusion2".to_string()),
-    );
+    let _ = connection
+        .set_option(
+            OptionConnection::CurrentCatalog,
+            OptionValue::String("datafusion2".to_string()),
+        )
+        .await;
 
     let current_catalog = connection
         .get_option_string(OptionConnection::CurrentCatalog)
+        .await
         .unwrap();
 
     assert_eq!(current_catalog, "datafusion2");
 
     let current_schema = connection
         .get_option_string(OptionConnection::CurrentSchema)
+        .await
         .unwrap();
 
     assert_eq!(current_schema, "public");
 
-    let _ = connection.set_option(
-        OptionConnection::CurrentSchema,
-        OptionValue::String("public2".to_string()),
-    );
+    let _ = connection
+        .set_option(
+            OptionConnection::CurrentSchema,
+            OptionValue::String("public2".to_string()),
+        )
+        .await;
 
     let current_schema = connection
         .get_option_string(OptionConnection::CurrentSchema)
+        .await
         .unwrap();
 
     assert_eq!(current_schema, "public2");
 }
 
-#[test]
-fn test_get_objects_database() {
-    let mut connection = get_connection(None);
+#[tokio::test]
+async fn test_get_objects_database() {
+    let mut connection = get_connection().await;
 
-    let objects = get_objects(&connection);
+    let objects = get_objects(&connection).await;
 
     assert_eq!(objects.num_rows(), 1);
 
-    execute_update(&mut connection, "CREATE DATABASE another");
+    execute_update(&mut connection, "CREATE DATABASE another").await;
 
-    let objects = get_objects(&connection);
+    let objects = get_objects(&connection).await;
 
     assert_eq!(objects.num_rows(), 2);
 }
 
-#[test]
-fn test_execute_sql() {
-    let mut connection = get_connection(None);
+#[tokio::test]
+async fn test_execute_sql() {
+    let mut connection = get_connection().await;
 
-    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')");
+    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')").await;
 
-    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example");
+    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example").await;
 
     assert_eq!(batch.num_rows(), 3);
     assert_eq!(batch.num_columns(), 2);
 }
 
-#[test]
-fn test_ingest() {
-    let mut connection = get_connection(None);
+#[tokio::test]
+async fn test_ingest() {
+    let mut connection = get_connection().await;
 
-    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')");
+    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')").await;
 
-    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example");
+    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example").await;
 
     assert_eq!(batch.num_rows(), 3);
     assert_eq!(batch.num_columns(), 2);
 
-    let mut statement = connection.new_statement().unwrap();
+    let mut statement = connection.new_statement().await.unwrap();
 
-    let _ = statement.set_option(
-        OptionStatement::TargetTable,
-        OptionValue::String("example".to_string()),
-    );
-    let _ = statement.bind(batch);
+    let _ = statement
+        .set_option(
+            OptionStatement::TargetTable,
+            OptionValue::String("example".to_string()),
+        )
+        .await;
+    let _ = statement.bind(batch).await;
 
-    let _ = statement.execute_update();
+    let _ = statement.execute_update().await;
 
-    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example");
+    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example").await;
 
     assert_eq!(batch.num_rows(), 6);
 }
 
-#[test]
-fn test_execute_substrait() {
-    let mut connection = get_connection(None);
+#[tokio::test]
+async fn test_execute_substrait() {
+    let mut connection = get_connection().await;
 
-    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')");
+    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')").await;
 
     let ctx = SessionContext::new();
 
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()
-        .unwrap();
-
-    let plan = runtime.block_on(async {
-        let _ = ctx.sql(
-            "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')"
-        ).await;
+    let _ = ctx.sql(
+        "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')"
+    ).await;
 
-        let df = ctx.sql("SELECT c1, c2 FROM datafusion.public.example").await.unwrap();
+    let df = ctx
+        .sql("SELECT c1, c2 FROM datafusion.public.example")
+        .await
+        .unwrap();
 
-        to_substrait_plan(df.logical_plan(), &ctx.state()).unwrap()
-    });
+    let plan = to_substrait_plan(df.logical_plan(), &ctx.state()).unwrap();
 
-    let batch = execute_substrait(&mut connection, *plan);
+    let batch = execute_substrait(&mut connection, *plan).await;
 
     assert_eq!(batch.num_rows(), 3);
     assert_eq!(batch.num_columns(), 2);
@@ -201,11 +229,22 @@ fn test_execute_substrait() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_running_in_async() {
-    let mut connection = get_connection(Some(tokio::runtime::Handle::current()));
+    let mut connection = get_connection().await;
+
+    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')").await;
 
-    execute_update(&mut connection, "CREATE TABLE IF NOT EXISTS datafusion.public.example (c1 INT, c2 VARCHAR) AS VALUES(1,'HELLO'),(2,'DATAFUSION'),(3,'!')");
+    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example").await;
+
+    assert_eq!(batch.num_rows(), 3);
+    assert_eq!(batch.num_columns(), 2);
 
-    let batch = execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example");
+    let batch = tokio::spawn(async move {
+        let batch =
+            execute_sql_query(&mut connection, "SELECT * FROM datafusion.public.example").await;
+        batch
+    })
+    .await
+    .unwrap();
 
     assert_eq!(batch.num_rows(), 3);
     assert_eq!(batch.num_columns(), 2);
diff --git a/rust/driver/dummy/src/lib.rs b/rust/driver/dummy/src/lib.rs
index 4ab708afdf..d600e65e37 100644
--- a/rust/driver/dummy/src/lib.rs
+++ b/rust/driver/dummy/src/lib.rs
@@ -783,7 +783,7 @@ impl Connection for DummyConnection {
         Ok(reader)
     }
 
-    fn read_partition(&self, _partition: impl AsRef<[u8]>) -> Result<impl RecordBatchReader> {
+    fn read_partition(&self, _partition: &[u8]) -> Result<impl RecordBatchReader> {
         let batch = get_table_data();
         let reader = SingleBatchReader::new(batch);
         Ok(reader)
diff --git a/rust/driver/dummy/tests/driver_exporter_dummy.rs b/rust/driver/dummy/tests/driver_exporter_dummy.rs
index a5869c0555..3a6db400ca 100644
--- a/rust/driver/dummy/tests/driver_exporter_dummy.rs
+++ b/rust/driver/dummy/tests/driver_exporter_dummy.rs
@@ -30,8 +30,7 @@ use adbc_core::options::{
     AdbcVersion, InfoCode, IngestMode, IsolationLevel, ObjectDepth, OptionConnection,
     OptionDatabase, OptionStatement,
 };
-use adbc_core::Statement;
-use adbc_core::{schemas, Connection, Database, Driver, Optionable};
+use adbc_core::{schemas, Connection, Database, Driver, Optionable, Statement};
 use adbc_driver_manager::{ManagedConnection, ManagedDatabase, ManagedDriver, ManagedStatement};
 
 use adbc_dummy::{DummyConnection, DummyDatabase, DummyDriver, DummyStatement, SingleBatchReader};
diff --git a/rust/driver/snowflake/src/connection.rs b/rust/driver/snowflake/src/connection.rs
index cc0ba51d94..706924d900 100644
--- a/rust/driver/snowflake/src/connection.rs
+++ b/rust/driver/snowflake/src/connection.rs
@@ -24,7 +24,7 @@ use std::collections::HashSet;
 use adbc_core::{
     error::Result,
     options::{InfoCode, OptionConnection, OptionValue},
-    Optionable,
+    sync::Optionable,
 };
 use adbc_driver_manager::ManagedConnection;
 use arrow_array::RecordBatchReader;
@@ -63,7 +63,7 @@ impl Optionable for Connection {
     }
 }
 
-impl adbc_core::Connection for Connection {
+impl adbc_core::sync::Connection for Connection {
     type StatementType = Statement;
 
     fn new_statement(&mut self) -> Result<Self::StatementType> {
@@ -133,7 +133,7 @@ impl adbc_core::Connection for Connection {
         self.0.rollback()
     }
 
-    fn read_partition(&self, partition: impl AsRef<[u8]>) -> Result<impl RecordBatchReader + Send> {
+    fn read_partition(&self, partition: &[u8]) -> Result<impl RecordBatchReader + Send> {
         self.0.read_partition(partition)
     }
 }
diff --git a/rust/driver/snowflake/src/connection/builder.rs b/rust/driver/snowflake/src/connection/builder.rs
index fbc2b8cfba..6966f200a8 100644
--- a/rust/driver/snowflake/src/connection/builder.rs
+++ b/rust/driver/snowflake/src/connection/builder.rs
@@ -24,7 +24,7 @@ use std::fmt;
 use adbc_core::{
     error::Result,
     options::{OptionConnection, OptionValue},
-    Database as _,
+    sync::Database as _,
 };
 
 #[cfg(feature = "env")]
diff --git a/rust/driver/snowflake/src/database.rs b/rust/driver/snowflake/src/database.rs
index 48715a78e5..cac7ba4ca0 100644
--- a/rust/driver/snowflake/src/database.rs
+++ b/rust/driver/snowflake/src/database.rs
@@ -24,7 +24,7 @@ use std::{collections::HashSet, ffi::c_int, sync::Arc};
 use adbc_core::{
     error::{Error, Result, Status},
     options::{AdbcVersion, InfoCode, OptionConnection, OptionDatabase, OptionValue},
-    Connection as _, Database as _, Optionable,
+    sync::{Connection as _, Database as _, Optionable},
 };
 use adbc_driver_manager::ManagedDatabase;
 use arrow_array::{
@@ -165,7 +165,7 @@ impl Optionable for Database {
     }
 }
 
-impl adbc_core::Database for Database {
+impl adbc_core::sync::Database for Database {
     type ConnectionType = Connection;
 
     fn new_connection(&self) -> Result<Self::ConnectionType> {
diff --git a/rust/driver/snowflake/src/database/builder.rs b/rust/driver/snowflake/src/database/builder.rs
index 05dd921e9f..886266cbcb 100644
--- a/rust/driver/snowflake/src/database/builder.rs
+++ b/rust/driver/snowflake/src/database/builder.rs
@@ -27,7 +27,7 @@ use std::{fmt, path::PathBuf, str::FromStr, time::Duration};
 use adbc_core::{
     error::{Error, Result, Status},
     options::{OptionDatabase, OptionValue},
-    Driver as _,
+    sync::Driver as _,
 };
 use url::{Host, Url};
 
@@ -490,7 +490,7 @@ impl Builder {
     pub const CLIENT_STORE_TEMP_CREDS_ENV: &str =
         "ADBC_SNOWFLAKE_SQL_CLIENT_OPTION_STORE_TEMP_CREDS";
 
-    /// See [`Self::]
+    /// See [`Self::client_identity_provider`]
     pub const CLIENT_IDENTITY_PROVIDER_ENV: &str =
         "ADBC_SNOWFLAKE_SQL_CLIENT_OPTION_IDENTITY_PROVIDER";
 
diff --git a/rust/driver/snowflake/src/driver.rs b/rust/driver/snowflake/src/driver.rs
index 7467ae872c..324baacf18 100644
--- a/rust/driver/snowflake/src/driver.rs
+++ b/rust/driver/snowflake/src/driver.rs
@@ -153,7 +153,7 @@ impl Driver {
     }
 }
 
-impl adbc_core::Driver for Driver {
+impl adbc_core::sync::Driver for Driver {
     type DatabaseType = Database;
 
     fn new_database(&mut self) -> Result<Self::DatabaseType> {
diff --git a/rust/driver_manager/src/lib.rs b/rust/driver_manager/src/lib.rs
index 173b60d06d..310ef47413 100644
--- a/rust/driver_manager/src/lib.rs
+++ b/rust/driver_manager/src/lib.rs
@@ -50,7 +50,7 @@
 //! # use arrow_select::concat::concat_batches;
 //! # use adbc_core::{
 //! #     options::{AdbcVersion, OptionDatabase, OptionStatement},
-//! #     Connection, Database, Driver, Statement, Optionable
+//! #     sync::{Connection, Database, Driver, Statement, Optionable}
 //! # };
 //! # use adbc_driver_manager::ManagedDriver;
 //! # fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -1473,13 +1473,12 @@ impl Connection for ManagedConnection {
         Ok(reader)
     }
 
-    fn read_partition(&self, partition: impl AsRef<[u8]>) -> Result<impl RecordBatchReader> {
+    fn read_partition(&self, partition: &[u8]) -> Result<impl RecordBatchReader> {
         let mut stream = FFI_ArrowArrayStream::empty();
         let driver = self.ffi_driver();
         let mut connection = self.inner.connection.lock().unwrap();
         let mut error = adbc_ffi::FFI_AdbcError::with_driver(driver);
         let method = driver_method!(driver, ConnectionReadPartition);
-        let partition = partition.as_ref();
         let status = unsafe {
             method(
                 connection.deref_mut(),
diff --git a/rust/driver_manager/tests/driver_manager_sqlite.rs b/rust/driver_manager/tests/driver_manager_sqlite.rs
index b00f9a555a..64a6da00bd 100644
--- a/rust/driver_manager/tests/driver_manager_sqlite.rs
+++ b/rust/driver_manager/tests/driver_manager_sqlite.rs
@@ -17,9 +17,10 @@
 
 use arrow_schema::{Field, Schema};
 
+use adbc_core::error::Status;
 use adbc_core::options::{AdbcVersion, OptionConnection, OptionDatabase};
-use adbc_core::{error::Status, Driver, Optionable};
-use adbc_core::{Connection, Database, Statement, LOAD_FLAG_DEFAULT};
+use adbc_core::LOAD_FLAG_DEFAULT;
+use adbc_core::{Connection, Database, Driver, Optionable, Statement};
 use adbc_driver_manager::{ManagedDatabase, ManagedDriver};
 
 mod common;