This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new aec0e8d0f feat(rust): add integration tests and some improvements
(#1883)
aec0e8d0f is described below
commit aec0e8d0fbf6a18c6bb66b7b5a149919edc3f420
Author: Alexandre Crayssac <[email protected]>
AuthorDate: Sat Jun 22 03:31:22 2024 +0200
feat(rust): add integration tests and some improvements (#1883)
This should be the last PR for the initial Rust implementation.
It includes:
- An integration test for the driver manager against the official SQLite
driver
- An integration test for the dummy driver against itself (comparing
output when using the Rust API and when using its exported version
through the driver manager)
- New info codes and options which have been added recently
- An enum for identifying statistics
Note that I also have an integration test for the driver manager against
the official PostgreSQL driver but it needs a running PostgreSQL
instance to be executed, so it cannot be easily integrated into the CI.
This test proved useful during development as it allowed me to catch
some bugs. The existence of this test is the reason why there is a
common module. Maybe we can add it in a follow-up PR?
---
rust/Cargo.lock | 74 ++-
rust/README.md | 29 +-
rust/core/src/ffi/constants.rs | 9 +-
rust/core/src/lib.rs | 3 +-
rust/core/src/options.rs | 122 +++-
rust/core/tests/common/mod.rs | 343 +++++++++++
rust/core/tests/driver_manager_sqlite.rs | 305 ++++++++++
rust/drivers/dummy/src/lib.rs | 7 +-
rust/drivers/dummy/tests/driver_exporter_dummy.rs | 655 ++++++++++++++++++++++
9 files changed, 1501 insertions(+), 46 deletions(-)
diff --git a/rust/Cargo.lock b/rust/Cargo.lock
index fcc708aa5..883ccf72f 100644
--- a/rust/Cargo.lock
+++ b/rust/Cargo.lock
@@ -4,7 +4,7 @@ version = 3
[[package]]
name = "adbc_core"
-version = "0.1.0"
+version = "0.13.0"
dependencies = [
"arrow",
"libloading",
@@ -13,7 +13,7 @@ dependencies = [
[[package]]
name = "adbc_dummy"
-version = "0.1.0"
+version = "0.13.0"
dependencies = [
"adbc_core",
"arrow",
@@ -230,15 +230,15 @@ dependencies = [
[[package]]
name = "autocfg"
-version = "1.2.0"
+version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "f1fdabc7756949593fe60f30ec81974b613357de856987752631dea1e3394c80"
+checksum = "0c4b4d0bd25bd0b74681c0ad21497610ce1b7c91b1022cd21c80c6fbdd9476b0"
[[package]]
name = "base64"
-version = "0.22.0"
+version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "9475866fec1451be56a3c2400fd081ff546538961565ccb5b7142cbd22bc7a51"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
[[package]]
name = "bitflags"
@@ -260,9 +260,9 @@ checksum =
"514de17de45fdb8dc022b1a7975556c53c86f9f0aa5f534b98977b171857c2c9"
[[package]]
name = "cc"
-version = "1.0.95"
+version = "1.0.98"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "d32a725bc159af97c3e629873bb9f88fb8cf8a4867175f76dc987815ea07c83b"
+checksum = "41c270e7540d725e65ac7f1b212ac8ce349719624d7bcff99f8e2e488e8cf03f"
[[package]]
name = "cfg-if"
@@ -316,9 +316,9 @@ checksum =
"7a81dae078cea95a014a339291cec439d2f232ebe854a9d672b796c6afafa9b7"
[[package]]
name = "getrandom"
-version = "0.2.14"
+version = "0.2.15"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "94b22e06ecb0110981051723910cbf0b5f5e09a2062dd7663334ee79a9d1286c"
+checksum = "c4567c8db10ae91089c99af84c68c38da3ec2f087c3f82960bcdbf3656b6f4d7"
dependencies = [
"cfg-if",
"libc",
@@ -338,9 +338,9 @@ dependencies = [
[[package]]
name = "hashbrown"
-version = "0.14.3"
+version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "290f1a1d9242c78d09ce40a5e87e7554ee637af1351968159f4952f028f75604"
+checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
[[package]]
name = "iana-time-zone"
@@ -440,9 +440,9 @@ dependencies = [
[[package]]
name = "libc"
-version = "0.2.153"
+version = "0.2.155"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "9c198f91728a82281a64e1f4f9eeb25d82cb32a5de251c6bd1b5154d63a8e7bd"
+checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c"
[[package]]
name = "libloading"
@@ -474,9 +474,9 @@ checksum =
"6c8640c5d730cb13ebd907d8d04b52f55ac9a2eec55b440c8892f40d56c76c1d"
[[package]]
name = "num"
-version = "0.4.2"
+version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "3135b08af27d103b0a51f2ae0f8632117b7b185ccf931445affa8df530576a41"
+checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23"
dependencies = [
"num-bigint",
"num-complex",
@@ -488,20 +488,19 @@ dependencies = [
[[package]]
name = "num-bigint"
-version = "0.4.4"
+version = "0.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "608e7659b5c3d7cba262d894801b9ec9d00de989e8a82bd4bef91d08da45cdc0"
+checksum = "c165a9ab64cf766f73521c0dd2cfdff64f488b8f0b3e621face3462d3db536d7"
dependencies = [
- "autocfg",
"num-integer",
"num-traits",
]
[[package]]
name = "num-complex"
-version = "0.4.5"
+version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "23c6602fda94a57c990fe0df199a035d83576b496aa29f4e634a8ac6004e68a6"
+checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495"
dependencies = [
"num-traits",
]
@@ -517,9 +516,9 @@ dependencies = [
[[package]]
name = "num-iter"
-version = "0.1.44"
+version = "0.1.45"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "d869c01cc0c455284163fd0092f1f93835385ccab5a98a0dcc497b2f8bf055a9"
+checksum = "1429034a0490724d0075ebb2bc9e875d6503c3cf69e235a8941aa757d83ef5bf"
dependencies = [
"autocfg",
"num-integer",
@@ -528,11 +527,10 @@ dependencies = [
[[package]]
name = "num-rational"
-version = "0.4.1"
+version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "0638a1c9d0a3c0914158145bc76cff373a75a627e6ecbfb71cbe6f453a5a19b0"
+checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824"
dependencies = [
- "autocfg",
"num-bigint",
"num-integer",
"num-traits",
@@ -540,9 +538,9 @@ dependencies = [
[[package]]
name = "num-traits"
-version = "0.2.18"
+version = "0.2.19"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "da0df0e5185db44f69b44f26786fe401b6c293d1907744beaa7fa62b2e5a517a"
+checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
dependencies = [
"autocfg",
"libm",
@@ -556,9 +554,9 @@ checksum =
"3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
[[package]]
name = "proc-macro2"
-version = "1.0.81"
+version = "1.0.83"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "3d1597b0c024618f09a9c3b8655b7e430397a36d23fdafec26d6965e9eec3eba"
+checksum = "0b33eb56c327dec362a9e55b3ad14f9d2f0904fb5a5b03b513ab5465399e9f43"
dependencies = [
"unicode-ident",
]
@@ -603,9 +601,9 @@ checksum =
"adad44e29e4c806119491a7f06f03de4d1af22c3a680dd47f1e6e179439d1f56"
[[package]]
name = "ryu"
-version = "1.0.17"
+version = "1.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "e86697c916019a8588c99b5fac3cead74ec0b4b819707a682fd4d23fa0ce1ba1"
+checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f"
[[package]]
name = "static_assertions"
@@ -615,9 +613,9 @@ checksum =
"a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
[[package]]
name = "syn"
-version = "2.0.60"
+version = "2.0.66"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "909518bc7b1c9b779f1bbf07f2929d35af9f0f37e47c6e9ef7f9dddc1e1821f3"
+checksum = "c42f3f41a2de00b01c0aaad383c5a45241efc8b2d1eda5661812fda5f3cdcff5"
dependencies = [
"proc-macro2",
"quote",
@@ -780,18 +778,18 @@ checksum =
"bec47e5bfd1bff0eeaf6d8b485cc1074891a197ab4225d504cb7a1ab88b02bf0"
[[package]]
name = "zerocopy"
-version = "0.7.32"
+version = "0.7.34"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "74d4d3961e53fa4c9a25a8637fc2bfaf2595b3d3ae34875568a5cf64787716be"
+checksum = "ae87e3fcd617500e5d106f0380cf7b77f3c6092aae37191433159dda23cfb087"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
-version = "0.7.32"
+version = "0.7.34"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "9ce1b18ccd8e73a9321186f97e46f9f04b778851177567b1975109d26a08d2a6"
+checksum = "15e934569e47891f7d9411f1a451d947a60e000ab3bd24fbb970f000387d1b3b"
dependencies = [
"proc-macro2",
"quote",
diff --git a/rust/README.md b/rust/README.md
index a62cd745e..66de62096 100644
--- a/rust/README.md
+++ b/rust/README.md
@@ -21,4 +21,31 @@
This is a Rust implementation of [Arrow Database Connectivity
(ADBC)](https://arrow.apache.org/adbc).
-It's still work in progress and should not be used in production.
+It currently provides:
+
+- An abstract Rust API to be implemented by vendor-specific drivers.
+- A driver manager which implements this same API, but dynamically loads
+ drivers internally and forwards calls appropriately using the C API.
+- A driver exporter that takes an implementation of the abstract API and
+ turns it into an object file that implements the C API.
+- A dummy driver implementation for testing and documentation purposes.
+
+## Development
+
+To run the integration tests you must:
+
+1. Install [SQLite](https://www.sqlite.org/) and have its dynamic library in
path.
+1. Build the official ADBC SQLite driver by following the
[documentation](../CONTRIBUTING.md).
+1. Place the resulting object file into your dynamic loader path or set
+ `LD_LIBRARY_PATH/DYLD_LIBRARY_PATH` appropriately.
+1. Run `cargo test --all-features --workspace`
+
+## Writing Drivers
+
+To write an ADBC driver in Rust you have to:
+
+1. Create a new library crate with `crate-type = ["lib", "cdylib"]`.
+1. Implement the abstract API which consists of the traits `Driver`,
`Database`, `Connection` and `Statement`.
+1. Export your driver to C with the macro `adbc_core::export_driver!`.
+
+The resulting object file can then be loaded by other languages trough their
own driver manager.
diff --git a/rust/core/src/ffi/constants.rs b/rust/core/src/ffi/constants.rs
index 754df1477..303c8d70a 100644
--- a/rust/core/src/ffi/constants.rs
+++ b/rust/core/src/ffi/constants.rs
@@ -41,6 +41,10 @@ pub const ADBC_VERSION_1_1_0: c_int = 1_001_000;
pub const ADBC_INFO_VENDOR_NAME: u32 = 0;
pub const ADBC_INFO_VENDOR_VERSION: u32 = 1;
pub const ADBC_INFO_VENDOR_ARROW_VERSION: u32 = 2;
+pub const ADBC_INFO_VENDOR_SQL: u32 = 3;
+pub const ADBC_INFO_VENDOR_SUBSTRAIT: u32 = 4;
+pub const ADBC_INFO_VENDOR_SUBSTRAIT_MIN_VERSION: u32 = 5;
+pub const ADBC_INFO_VENDOR_SUBSTRAIT_MAX_VERSION: u32 = 6;
pub const ADBC_INFO_DRIVER_NAME: u32 = 100;
pub const ADBC_INFO_DRIVER_VERSION: u32 = 101;
pub const ADBC_INFO_DRIVER_ARROW_VERSION: u32 = 102;
@@ -55,8 +59,11 @@ pub const ADBC_OBJECT_DEPTH_COLUMNS: c_int =
ADBC_OBJECT_DEPTH_ALL;
pub const ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA: i32 = i32::MIN;
pub const ADBC_INGEST_OPTION_TARGET_TABLE: &str = "adbc.ingest.target_table";
-pub const ADBC_INGEST_OPTION_MODE: &str = "adbc.ingest.mode";
+pub const ADBC_INGEST_OPTION_TARGET_CATALOG: &str =
"adbc.ingest.target_catalog";
+pub const ADBC_INGEST_OPTION_TARGET_DB_SCHEMA: &str =
"adbc.ingest.target_db_schema";
+pub const ADBC_INGEST_OPTION_TEMPORARY: &str = "adbc.ingest.temporary";
+pub const ADBC_INGEST_OPTION_MODE: &str = "adbc.ingest.mode";
pub const ADBC_INGEST_OPTION_MODE_CREATE: &str = "adbc.ingest.mode.create";
pub const ADBC_INGEST_OPTION_MODE_APPEND: &str = "adbc.ingest.mode.append";
pub const ADBC_INGEST_OPTION_MODE_REPLACE: &str = "adbc.ingest.mode.replace";
diff --git a/rust/core/src/lib.rs b/rust/core/src/lib.rs
index b24dabbb5..901f6a574 100644
--- a/rust/core/src/lib.rs
+++ b/rust/core/src/lib.rs
@@ -369,8 +369,7 @@ pub trait Connection: Optionable<Option = OptionConnection>
{
/// 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, TODO (change this when statistics
enum is added )
- /// see \ref adbc-table-statistics. To get
+ /// 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.
///
diff --git a/rust/core/src/options.rs b/rust/core/src/options.rs
index 3788a94ce..144e46aeb 100644
--- a/rust/core/src/options.rs
+++ b/rust/core/src/options.rs
@@ -138,6 +138,14 @@ pub enum InfoCode {
VendorVersion,
/// The database vendor/product Arrow library version (type: utf8).
VendorArrowVersion,
+ /// Indicates whether SQL queries are supported (type: bool).
+ VendorSql,
+ /// Indicates whether Substrait queries are supported (type: bool).
+ VendorSubstrait,
+ /// The minimum supported Substrait version, or null if Substrait is not
supported (type: utf8).
+ VendorSubstraitMinVersion,
+ /// The maximum supported Substrait version, or null if Substrait is not
supported (type: utf8).
+ VendorSubstraitMaxVersion,
/// The driver name (type: utf8).
DriverName,
/// The driver version (type: utf8).
@@ -150,7 +158,6 @@ pub enum InfoCode {
///
/// ADBC API revision 1.1.0
DriverAdbcVersion,
- // TODO(alexandreyc): add new codes (see
https://github.com/apache/arrow-adbc/commit/aa04aadccd319e6fa3abb07154fa8d87b58d5c21)
}
impl From<&InfoCode> for u32 {
@@ -159,6 +166,14 @@ impl From<&InfoCode> for u32 {
InfoCode::VendorName => constants::ADBC_INFO_VENDOR_NAME,
InfoCode::VendorVersion => constants::ADBC_INFO_VENDOR_VERSION,
InfoCode::VendorArrowVersion =>
constants::ADBC_INFO_VENDOR_ARROW_VERSION,
+ InfoCode::VendorSql => constants::ADBC_INFO_VENDOR_SQL,
+ InfoCode::VendorSubstrait => constants::ADBC_INFO_VENDOR_SUBSTRAIT,
+ InfoCode::VendorSubstraitMinVersion => {
+ constants::ADBC_INFO_VENDOR_SUBSTRAIT_MIN_VERSION
+ }
+ InfoCode::VendorSubstraitMaxVersion => {
+ constants::ADBC_INFO_VENDOR_SUBSTRAIT_MAX_VERSION
+ }
InfoCode::DriverName => constants::ADBC_INFO_DRIVER_NAME,
InfoCode::DriverVersion => constants::ADBC_INFO_DRIVER_VERSION,
InfoCode::DriverArrowVersion =>
constants::ADBC_INFO_DRIVER_ARROW_VERSION,
@@ -175,6 +190,14 @@ impl TryFrom<u32> for InfoCode {
constants::ADBC_INFO_VENDOR_NAME => Ok(InfoCode::VendorName),
constants::ADBC_INFO_VENDOR_VERSION => Ok(InfoCode::VendorVersion),
constants::ADBC_INFO_VENDOR_ARROW_VERSION =>
Ok(InfoCode::VendorArrowVersion),
+ constants::ADBC_INFO_VENDOR_SQL => Ok(InfoCode::VendorSql),
+ constants::ADBC_INFO_VENDOR_SUBSTRAIT =>
Ok(InfoCode::VendorSubstrait),
+ constants::ADBC_INFO_VENDOR_SUBSTRAIT_MIN_VERSION => {
+ Ok(InfoCode::VendorSubstraitMinVersion)
+ }
+ constants::ADBC_INFO_VENDOR_SUBSTRAIT_MAX_VERSION => {
+ Ok(InfoCode::VendorSubstraitMaxVersion)
+ }
constants::ADBC_INFO_DRIVER_NAME => Ok(InfoCode::DriverName),
constants::ADBC_INFO_DRIVER_VERSION => Ok(InfoCode::DriverVersion),
constants::ADBC_INFO_DRIVER_ARROW_VERSION =>
Ok(InfoCode::DriverArrowVersion),
@@ -336,6 +359,12 @@ pub enum OptionStatement {
IngestMode,
/// The name of the target table for a bulk insert.
TargetTable,
+ /// The catalog of the table for bulk insert.
+ TargetCatalog,
+ /// The schema of the table for bulk insert.
+ TargetDbSchema,
+ /// Use a temporary table for ingestion.
+ Temporary,
/// Whether query execution is nonblocking. By default, execution is
blocking.
///
/// When enabled,
[execute_partitions][crate::Statement::execute_partitions]
@@ -384,6 +413,9 @@ impl AsRef<str> for OptionStatement {
match self {
Self::IngestMode => constants::ADBC_INGEST_OPTION_MODE,
Self::TargetTable => constants::ADBC_INGEST_OPTION_TARGET_TABLE,
+ Self::TargetCatalog =>
constants::ADBC_INGEST_OPTION_TARGET_CATALOG,
+ Self::TargetDbSchema =>
constants::ADBC_INGEST_OPTION_TARGET_DB_SCHEMA,
+ Self::Temporary => constants::ADBC_INGEST_OPTION_TEMPORARY,
Self::Incremental => constants::ADBC_STATEMENT_OPTION_INCREMENTAL,
Self::Progress => constants::ADBC_STATEMENT_OPTION_PROGRESS,
Self::MaxProgress => constants::ADBC_STATEMENT_OPTION_MAX_PROGRESS,
@@ -397,6 +429,9 @@ impl From<&str> for OptionStatement {
match value {
constants::ADBC_INGEST_OPTION_MODE => Self::IngestMode,
constants::ADBC_INGEST_OPTION_TARGET_TABLE => Self::TargetTable,
+ constants::ADBC_INGEST_OPTION_TARGET_CATALOG =>
Self::TargetCatalog,
+ constants::ADBC_INGEST_OPTION_TARGET_DB_SCHEMA =>
Self::TargetDbSchema,
+ constants::ADBC_INGEST_OPTION_TEMPORARY => Self::Temporary,
constants::ADBC_STATEMENT_OPTION_INCREMENTAL => Self::Incremental,
constants::ADBC_STATEMENT_OPTION_PROGRESS => Self::Progress,
constants::ADBC_STATEMENT_OPTION_MAX_PROGRESS => Self::MaxProgress,
@@ -523,3 +558,88 @@ impl From<IngestMode> for OptionValue {
Self::String(value.into())
}
}
+
+/// Statistics about the data distribution.
+#[derive(Debug, Clone)]
+#[non_exhaustive]
+pub enum Statistics {
+ /// The average byte width statistic. The average size in bytes of a row in
+ /// the column. Value type is `float64`. For example, this is roughly the
+ /// average length of a string for a string column.
+ AverageByteWidth,
+ /// The distinct value count (NDV) statistic. The number of distinct values
+ /// in the column. Value type is `int64` (when not approximate) or
`float64`
+ /// (when approximate).
+ DistinctCount,
+ /// The max byte width statistic. The maximum size in bytes of a row in the
+ /// column. Value type is `int64` (when not approximate) or `float64`
(when approximate).
+ /// For example, this is the maximum length of a string for a string
column.
+ MaxByteWidth,
+ /// The max value statistic. Value type is column-dependent.
+ MaxValue,
+ /// The min value statistic. Value type is column-dependent.
+ MinValue,
+ /// The null count statistic. The number of values that are null in the
+ /// column. Value type is `int64` (when not approximate) or `float64`
(when approximate).
+ NullCount,
+ /// The row count statistic. The number of rows in the column or table.
+ /// Value type is `int64` (when not approximate) or `float64` (when
approximate).
+ RowCount,
+ /// Driver-specific statistics.
+ Other { key: i16, name: String },
+}
+
+impl TryFrom<i16> for Statistics {
+ type Error = Error;
+ fn try_from(value: i16) -> Result<Self, Self::Error> {
+ match value {
+ constants::ADBC_STATISTIC_AVERAGE_BYTE_WIDTH_KEY =>
Ok(Self::AverageByteWidth),
+ constants::ADBC_STATISTIC_DISTINCT_COUNT_KEY =>
Ok(Self::DistinctCount),
+ constants::ADBC_STATISTIC_MAX_BYTE_WIDTH_KEY =>
Ok(Self::MaxByteWidth),
+ constants::ADBC_STATISTIC_MAX_VALUE_KEY => Ok(Self::MaxValue),
+ constants::ADBC_STATISTIC_MIN_VALUE_KEY => Ok(Self::MinValue),
+ constants::ADBC_STATISTIC_NULL_COUNT_KEY => Ok(Self::NullCount),
+ constants::ADBC_STATISTIC_ROW_COUNT_KEY => Ok(Self::RowCount),
+ _ => Err(Error::with_message_and_status(
+ format!("Unknown standard statistic key: {}", value),
+ Status::InvalidArguments,
+ )),
+ }
+ }
+}
+
+impl From<Statistics> for i16 {
+ fn from(value: Statistics) -> Self {
+ match value {
+ Statistics::AverageByteWidth =>
constants::ADBC_STATISTIC_AVERAGE_BYTE_WIDTH_KEY,
+ Statistics::DistinctCount =>
constants::ADBC_STATISTIC_DISTINCT_COUNT_KEY,
+ Statistics::MaxByteWidth =>
constants::ADBC_STATISTIC_MAX_BYTE_WIDTH_KEY,
+ Statistics::MaxValue => constants::ADBC_STATISTIC_MAX_VALUE_KEY,
+ Statistics::MinValue => constants::ADBC_STATISTIC_MIN_VALUE_KEY,
+ Statistics::NullCount => constants::ADBC_STATISTIC_NULL_COUNT_KEY,
+ Statistics::RowCount => constants::ADBC_STATISTIC_ROW_COUNT_KEY,
+ Statistics::Other { key, name: _ } => key,
+ }
+ }
+}
+
+impl AsRef<str> for Statistics {
+ fn as_ref(&self) -> &str {
+ match self {
+ Statistics::AverageByteWidth =>
constants::ADBC_STATISTIC_AVERAGE_BYTE_WIDTH_NAME,
+ Statistics::DistinctCount =>
constants::ADBC_STATISTIC_DISTINCT_COUNT_NAME,
+ Statistics::MaxByteWidth =>
constants::ADBC_STATISTIC_MAX_BYTE_WIDTH_NAME,
+ Statistics::MaxValue => constants::ADBC_STATISTIC_MAX_VALUE_NAME,
+ Statistics::MinValue => constants::ADBC_STATISTIC_MIN_VALUE_NAME,
+ Statistics::NullCount => constants::ADBC_STATISTIC_NULL_COUNT_NAME,
+ Statistics::RowCount => constants::ADBC_STATISTIC_ROW_COUNT_NAME,
+ Statistics::Other { key: _, name } => name,
+ }
+ }
+}
+
+impl std::fmt::Display for Statistics {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}", self.as_ref())
+ }
+}
diff --git a/rust/core/tests/common/mod.rs b/rust/core/tests/common/mod.rs
new file mode 100644
index 000000000..3631579a3
--- /dev/null
+++ b/rust/core/tests/common/mod.rs
@@ -0,0 +1,343 @@
+// 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 std::ops::Deref;
+use std::sync::Arc;
+
+use adbc_core::driver_manager::{
+ ManagedConnection, ManagedDatabase, ManagedDriver, ManagedStatement,
+};
+use adbc_core::error::Status;
+use adbc_core::options::{
+ InfoCode, IngestMode, ObjectDepth, OptionConnection, OptionDatabase,
OptionStatement,
+};
+use adbc_core::schemas;
+use adbc_core::{Connection, Database, Driver, Optionable, Statement};
+
+use arrow::array::{as_string_array, Array, Float64Array, Int64Array,
StringArray};
+use arrow::compute::concat_batches;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use arrow::error::ArrowError;
+use arrow::record_batch::{RecordBatch, RecordBatchReader};
+
+pub struct SingleBatchReader {
+ batch: Option<RecordBatch>,
+ schema: SchemaRef,
+}
+
+impl SingleBatchReader {
+ pub fn new(batch: RecordBatch) -> Self {
+ let schema = batch.schema();
+ Self {
+ batch: Some(batch),
+ schema,
+ }
+ }
+}
+
+impl Iterator for SingleBatchReader {
+ // `RecordBatchReader` requires item to be wrapped within `Result`.
+ type Item = std::result::Result<RecordBatch, ArrowError>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ Ok(self.batch.take()).transpose()
+ }
+}
+
+impl RecordBatchReader for SingleBatchReader {
+ fn schema(&self) -> SchemaRef {
+ self.schema.clone()
+ }
+}
+
+pub fn sample_batch() -> RecordBatch {
+ let columns: Vec<Arc<dyn Array>> = vec![
+ Arc::new(Int64Array::from(vec![1, 2, 3, 4])),
+ Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0])),
+ Arc::new(StringArray::from(vec!["a", "b", "c", "d"])),
+ ];
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Float64, true),
+ Field::new("c", DataType::Utf8, true),
+ ]);
+ RecordBatch::try_new(Arc::new(schema), columns).unwrap()
+}
+
+pub fn concat_reader(reader: impl RecordBatchReader) -> RecordBatch {
+ let schema = reader.schema();
+ let batches: Vec<RecordBatch> = reader.map(|b| b.unwrap()).collect();
+ concat_batches(&schema, &batches).unwrap()
+}
+
+pub fn test_driver(driver: &mut ManagedDriver, uri: &str) {
+ let opts = [(OptionDatabase::Uri, uri.into())];
+ driver.new_database_with_opts(opts).unwrap();
+
+ // Unknown database option.
+ let opts = [(OptionDatabase::Other("unknown".into()), "".into())];
+ assert!(driver.new_database_with_opts(opts).is_err());
+}
+
+pub fn test_database(database: &mut ManagedDatabase) {
+ assert!(database.new_connection().is_ok());
+
+ let opts = [(OptionConnection::AutoCommit, "true".into())];
+ database.new_connection_with_opts(opts).unwrap();
+
+ // Unknown connection option.
+ let opts = [(OptionConnection::Other("unknown".into()), "".into())];
+ assert!(database.new_connection_with_opts(opts).is_err());
+}
+
+pub fn test_connection(connection: &mut ManagedConnection) {
+ assert!(connection
+ .set_option(OptionConnection::AutoCommit, "true".into())
+ .is_ok());
+
+ // Unknown connection option
+ assert!(connection
+ .set_option(OptionConnection::Other("unknown".into()), "".into())
+ .is_err());
+
+ assert!(connection.new_statement().is_ok());
+}
+
+pub fn test_connection_commit_rollback(connection: &mut ManagedConnection) {
+ let error = connection.commit().unwrap_err();
+ assert_eq!(error.status, Status::InvalidState);
+
+ let error = connection.rollback().unwrap_err();
+ assert_eq!(error.status, Status::InvalidState);
+
+ connection
+ .set_option(OptionConnection::AutoCommit, "false".into())
+ .unwrap();
+
+ connection.commit().unwrap();
+ connection.rollback().unwrap();
+}
+
+pub fn test_connection_read_partition(connection: &ManagedConnection) {
+ assert!(connection.read_partition(b"").is_err());
+}
+
+pub fn test_connection_get_table_types(connection: &ManagedConnection, actual:
&[&str]) {
+ let got = concat_reader(connection.get_table_types().unwrap());
+ assert_eq!(got.num_columns(), 1);
+ assert_eq!(got.schema(), *schemas::GET_TABLE_TYPES_SCHEMA.deref());
+
+ let got: Vec<Option<&str>> =
as_string_array(got.column(0)).iter().collect();
+ assert!(got.iter().all(|x| x.is_some()));
+
+ let got: HashSet<&str> = got.into_iter().map(|x| x.unwrap()).collect();
+ let actual: HashSet<&str> = actual.iter().copied().collect();
+ assert_eq!(got, actual);
+}
+
+pub fn test_connection_get_info(connection: &ManagedConnection,
actual_num_info: usize) {
+ let info = concat_reader(connection.get_info(None).unwrap());
+ assert_eq!(info.num_columns(), 2);
+ assert_eq!(info.num_rows(), actual_num_info);
+ assert_eq!(info.schema(), *schemas::GET_INFO_SCHEMA.deref());
+
+ let info = concat_reader(
+ connection
+ .get_info(Some(
+ [
+ InfoCode::VendorName,
+ InfoCode::DriverVersion,
+ InfoCode::DriverName,
+ InfoCode::VendorVersion,
+ ]
+ .into(),
+ ))
+ .unwrap(),
+ );
+ assert_eq!(info.num_columns(), 2);
+ assert_eq!(info.num_rows(), 4);
+ assert_eq!(info.schema(), *schemas::GET_INFO_SCHEMA.deref());
+}
+
+pub fn test_connection_get_objects(
+ connection: &ManagedConnection,
+ actual_num_catalog: usize,
+ actual_num_tables: usize,
+) {
+ let objects = concat_reader(
+ connection
+ .get_objects(ObjectDepth::All, None, None, None, None, None)
+ .unwrap(),
+ );
+ assert_eq!(objects.num_columns(), 2);
+ assert_eq!(objects.num_rows(), actual_num_catalog);
+
+ let objects = connection
+ .get_objects(
+ ObjectDepth::All,
+ None,
+ None,
+ None,
+ Some(vec!["table", "view"]),
+ None,
+ )
+ .unwrap();
+ let objects = concat_reader(objects);
+ assert_eq!(objects.num_columns(), 2);
+ assert_eq!(objects.num_rows(), actual_num_tables);
+
+ let objects = concat_reader(
+ connection
+ .get_objects(
+ ObjectDepth::All,
+ Some("my_catalog"),
+ Some("my_schema"),
+ Some("my_table"),
+ Some(vec!["table", "view"]),
+ Some("my_column"),
+ )
+ .unwrap(),
+ );
+ assert_eq!(objects.num_rows(), 0);
+ assert_eq!(objects.num_columns(), 2);
+}
+
+pub fn test_connection_get_table_schema(connection: &mut ManagedConnection) {
+ const TABLE_NAME: &str = "my_super_table";
+
+ connection
+ .set_option(OptionConnection::AutoCommit, "false".into())
+ .unwrap();
+
+ let mut statement = connection.new_statement().unwrap();
+ statement
+ .set_sql_query(&format!("create table {TABLE_NAME}(a bigint, b
bigint);"))
+ .unwrap();
+ statement.execute_update().unwrap();
+
+ let got = connection.get_table_schema(None, None, TABLE_NAME).unwrap();
+ let actual = Schema::new(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Int64, true),
+ ]);
+ assert_eq!(got, actual);
+
+ connection.rollback().unwrap();
+
+ assert!(connection
+ .get_table_schema(None, None, "nonexistent_table")
+ .is_err());
+}
+
+pub fn test_statement(statement: &mut ManagedStatement) {
+ statement
+ .set_option(OptionStatement::IngestMode, IngestMode::Create.into())
+ .unwrap();
+
+ statement
+ .set_option(
+ OptionStatement::Other("unknown".into()),
+ "unknown.value".into(),
+ )
+ .unwrap_err();
+}
+
+pub fn test_statement_prepare(statement: &mut ManagedStatement) {
+ let error = statement.prepare().unwrap_err();
+ assert_eq!(error.status, Status::InvalidState);
+
+ statement.set_sql_query("select 42").unwrap();
+ statement.prepare().unwrap();
+}
+
+pub fn test_statement_set_substrait_plan(statement: &mut ManagedStatement) {
+ let error = statement.set_substrait_plan(b"").unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+pub fn test_statement_execute(statement: &mut ManagedStatement) {
+ assert!(statement.execute().is_err());
+
+ statement.set_sql_query("select 42").unwrap();
+ let batch = concat_reader(statement.execute().unwrap());
+ assert_eq!(batch.num_rows(), 1);
+ assert_eq!(batch.num_columns(), 1);
+}
+
+pub fn test_statement_execute_update(connection: &mut ManagedConnection) {
+ let mut statement = connection.new_statement().unwrap();
+
+ let error = statement.execute_update().unwrap_err();
+ assert_eq!(error.status, Status::InvalidState);
+
+ connection
+ .set_option(OptionConnection::AutoCommit, "false".into())
+ .unwrap();
+
+ statement.set_sql_query("create table t(a int)").unwrap();
+ statement.execute_update().unwrap();
+
+ statement.set_sql_query("insert into t values(42)").unwrap();
+ statement.execute_update().unwrap();
+
+ connection.rollback().unwrap();
+}
+
+pub fn test_statement_execute_partitions(statement: &mut ManagedStatement) {
+ let error = statement.execute_partitions().unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+pub fn test_statement_bind(statement: &mut ManagedStatement) {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
true)]));
+ let columns: Vec<Arc<dyn Array>> = vec![Arc::new(Int64Array::from(vec![1,
2, 3]))];
+ let batch = RecordBatch::try_new(schema, columns).unwrap();
+ statement.bind(batch).unwrap();
+}
+
+pub fn test_statement_bind_stream(statement: &mut ManagedStatement) {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
true)]));
+ let columns: Vec<Arc<dyn Array>> = vec![Arc::new(Int64Array::from(vec![1,
2, 3]))];
+ let batch = RecordBatch::try_new(schema, columns).unwrap();
+ let reader = SingleBatchReader::new(batch);
+ statement.bind_stream(Box::new(reader)).unwrap();
+}
+
+pub fn test_ingestion_roundtrip(connection: &mut ManagedConnection) {
+ let mut statement = connection.new_statement().unwrap();
+ let batch = sample_batch();
+
+ connection
+ .set_option(OptionConnection::AutoCommit, "false".into())
+ .unwrap();
+
+ // Ingest
+ statement
+ .set_option(OptionStatement::TargetTable, "my_table".into())
+ .unwrap();
+
+ statement.bind(batch.clone()).unwrap();
+ statement.execute_update().unwrap();
+
+ // Read back
+ statement.set_sql_query("select * from my_table").unwrap();
+ let batch_got = concat_reader(statement.execute().unwrap());
+ assert_eq!(batch, batch_got);
+
+ connection.rollback().unwrap();
+}
diff --git a/rust/core/tests/driver_manager_sqlite.rs
b/rust/core/tests/driver_manager_sqlite.rs
new file mode 100644
index 000000000..db83de580
--- /dev/null
+++ b/rust/core/tests/driver_manager_sqlite.rs
@@ -0,0 +1,305 @@
+// 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::datatypes::{Field, Schema};
+
+use adbc_core::driver_manager::{ManagedDatabase, ManagedDriver};
+use adbc_core::options::{AdbcVersion, OptionConnection, OptionDatabase};
+use adbc_core::{error::Status, Driver, Optionable};
+use adbc_core::{Connection, Database, Statement};
+
+mod common;
+
+// By passing in ":memory:" for URI, we create a distinct temporary database
for
+// each test, preventing noisy neighbor issues on tests.
+const URI: &str = ":memory:";
+
+fn get_driver() -> ManagedDriver {
+ ManagedDriver::load_dynamic_from_name("adbc_driver_sqlite", None,
AdbcVersion::V100).unwrap()
+}
+
+fn get_database(driver: &mut ManagedDriver) -> ManagedDatabase {
+ let opts = [(OptionDatabase::Uri, URI.into())];
+ driver.new_database_with_opts(opts).unwrap()
+}
+
+#[test]
+fn test_driver() {
+ let mut driver = get_driver();
+ common::test_driver(&mut driver, URI);
+}
+
+#[test]
+fn test_database() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ common::test_database(&mut database);
+}
+
+#[test]
+fn test_database_get_option() {
+ let mut driver = get_driver();
+ let database = get_database(&mut driver);
+
+ let error = database
+ .get_option_bytes(OptionDatabase::Username)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = database
+ .get_option_string(OptionDatabase::Username)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = database
+ .get_option_int(OptionDatabase::Username)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = database
+ .get_option_double(OptionDatabase::Username)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+#[test]
+fn test_connection() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ common::test_connection(&mut connection);
+}
+
+#[test]
+fn test_connection_get_option() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+
+ let error = connection
+ .get_option_bytes(OptionConnection::AutoCommit)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = connection
+ .get_option_string(OptionConnection::AutoCommit)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = connection
+ .get_option_int(OptionConnection::AutoCommit)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+
+ let error = connection
+ .get_option_double(OptionConnection::AutoCommit)
+ .unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+#[test]
+fn test_connection_cancel() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+
+ let error = connection.cancel().unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+#[test]
+fn test_connection_commit_rollback() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ common::test_connection_commit_rollback(&mut connection);
+}
+
+#[test]
+fn test_connection_read_partition() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ common::test_connection_read_partition(&connection);
+}
+
+#[test]
+fn test_connection_get_table_types() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ common::test_connection_get_table_types(&connection, &["table", "view"]);
+}
+
+#[test]
+fn test_connection_get_info() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ common::test_connection_get_info(&connection, 5);
+}
+
+#[test]
+fn test_connection_get_objects() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ common::test_connection_get_objects(&connection, 1, 1);
+}
+
+#[test]
+fn test_connection_get_table_schema() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ common::test_connection_get_table_schema(&mut connection);
+}
+
+#[test]
+fn test_connection_get_statistic_names() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ assert!(connection.get_statistic_names().is_err());
+}
+
+#[test]
+fn test_connection_get_statistics() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let connection = database.new_connection().unwrap();
+ assert!(connection.get_statistics(None, None, None, false).is_err());
+}
+
+#[test]
+fn test_statement() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement(&mut statement);
+}
+
+#[test]
+fn test_statement_prepare() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_prepare(&mut statement);
+}
+
+#[test]
+fn test_statement_set_substrait_plan() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_set_substrait_plan(&mut statement);
+}
+
+#[test]
+fn test_statement_get_parameter_schema() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+
+ let error = statement.get_parameter_schema().unwrap_err();
+ assert_eq!(error.status, Status::InvalidState);
+
+ statement.set_sql_query("select 42").unwrap();
+ statement.prepare().unwrap();
+ let got = statement.get_parameter_schema().unwrap();
+ let fields: Vec<Field> = vec![];
+ let actual = Schema::new(fields);
+ assert_eq!(got, actual);
+}
+
+#[test]
+fn test_statement_execute() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_execute(&mut statement);
+}
+
+#[test]
+fn test_statement_execute_update() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ common::test_statement_execute_update(&mut connection);
+}
+
+#[test]
+fn test_statement_execute_schema() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+
+ let error = statement.execute_schema().unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+#[test]
+fn test_statement_execute_partitions() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_execute_partitions(&mut statement);
+}
+
+#[test]
+fn test_statement_cancel() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+
+ let error = statement.cancel().unwrap_err();
+ assert_eq!(error.status, Status::NotImplemented);
+}
+
+#[test]
+fn test_statement_bind() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_bind(&mut statement);
+}
+
+#[test]
+fn test_statement_bind_stream() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ let mut statement = connection.new_statement().unwrap();
+ common::test_statement_bind_stream(&mut statement);
+}
+
+#[test]
+fn test_ingestion_roundtrip() {
+ let mut driver = get_driver();
+ let mut database = get_database(&mut driver);
+ let mut connection = database.new_connection().unwrap();
+ common::test_ingestion_roundtrip(&mut connection);
+}
diff --git a/rust/drivers/dummy/src/lib.rs b/rust/drivers/dummy/src/lib.rs
index 1edd3b277..3591a4ed9 100644
--- a/rust/drivers/dummy/src/lib.rs
+++ b/rust/drivers/dummy/src/lib.rs
@@ -19,6 +19,7 @@ use std::collections::HashSet;
use std::sync::Arc;
use std::{collections::HashMap, fmt::Debug, hash::Hash};
+use adbc_core::options::Statistics;
use arrow::array::{
Array, ArrayRef, BinaryArray, BooleanArray, Float64Array, Int16Array,
Int32Array, Int64Array,
ListArray, MapArray, StringArray, StructArray, UInt32Array, UInt64Array,
UnionArray,
@@ -690,9 +691,9 @@ impl Connection for DummyConnection {
),
(
Arc::new(Field::new("statistic_key", DataType::Int16, false)),
- Arc::new(Int16Array::from(vec![
- constants::ADBC_STATISTIC_AVERAGE_BYTE_WIDTH_KEY,
- ])) as ArrayRef,
+ Arc::new(Int16Array::from(vec![Into::<i16>::into(
+ Statistics::AverageByteWidth,
+ )])) as ArrayRef,
),
(
Arc::new(Field::new(
diff --git a/rust/drivers/dummy/tests/driver_exporter_dummy.rs
b/rust/drivers/dummy/tests/driver_exporter_dummy.rs
new file mode 100644
index 000000000..318f822f9
--- /dev/null
+++ b/rust/drivers/dummy/tests/driver_exporter_dummy.rs
@@ -0,0 +1,655 @@
+// 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.
+
+/// This integration test compares the output of the dummy driver when it's
used
+/// directly using the Rust API (native) and trough the exported driver via the
+/// driver manager (exported). That allows us to test that data correctly
round-trip
+/// between C and Rust.
+use std::ops::Deref;
+use std::sync::Arc;
+
+use arrow::array::{Array, Float64Array, Int64Array, StringArray};
+use arrow::compute::concat_batches;
+use arrow::datatypes::{DataType, Field, Schema};
+use arrow::record_batch::{RecordBatch, RecordBatchReader};
+
+use adbc_core::driver_manager::{
+ ManagedConnection, ManagedDatabase, ManagedDriver, ManagedStatement,
+};
+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_dummy::{DummyConnection, DummyDatabase, DummyDriver, DummyStatement,
SingleBatchReader};
+
+const OPTION_STRING_LONG: &str =
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+const OPTION_BYTES_LONG: &[u8] =
b"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+
+pub fn concat_reader(reader: impl RecordBatchReader) -> RecordBatch {
+ let schema = reader.schema();
+ let batches: Vec<RecordBatch> = reader.map(|b| b.unwrap()).collect();
+ concat_batches(&schema, &batches).unwrap()
+}
+
+pub fn sample_batch() -> RecordBatch {
+ let columns: Vec<Arc<dyn Array>> = vec![
+ Arc::new(Int64Array::from(vec![1, 2, 3, 4])),
+ Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0])),
+ Arc::new(StringArray::from(vec!["a", "b", "c", "d"])),
+ ];
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Float64, true),
+ Field::new("c", DataType::Utf8, true),
+ ]);
+ RecordBatch::try_new(Arc::new(schema), columns).unwrap()
+}
+
+fn get_exported() -> (
+ ManagedDriver,
+ ManagedDatabase,
+ ManagedConnection,
+ ManagedStatement,
+) {
+ let mut driver = ManagedDriver::load_dynamic_from_name(
+ "adbc_dummy",
+ Some(b"DummyDriverInit"),
+ AdbcVersion::V110,
+ )
+ .unwrap();
+ let mut database = driver.new_database().unwrap();
+ let mut connection = database.new_connection().unwrap();
+ let statement = connection.new_statement().unwrap();
+ (driver, database, connection, statement)
+}
+
+fn get_native() -> (DummyDriver, DummyDatabase, DummyConnection,
DummyStatement) {
+ let mut driver = DummyDriver {};
+ let mut database = driver.new_database().unwrap();
+ let mut connection = database.new_connection().unwrap();
+ let statement = connection.new_statement().unwrap();
+ (driver, database, connection, statement)
+}
+
+// Database
+
+#[test]
+fn test_database_options() {
+ let (mut driver, _, _, _) = get_exported();
+
+ // Pre-init options.
+ let options = [
+ (OptionDatabase::Username, "Alice".into()),
+ (OptionDatabase::Password, 42.into()),
+ (OptionDatabase::Uri, std::f64::consts::PI.into()),
+ (OptionDatabase::Other("pre.bytes".into()), b"Hello".into()),
+ (
+ OptionDatabase::Other("pre.string.long".into()),
+ OPTION_STRING_LONG.into(),
+ ),
+ (
+ OptionDatabase::Other("pre.bytes.long".into()),
+ OPTION_BYTES_LONG.into(),
+ ),
+ ];
+
+ let mut database = driver.new_database_with_opts(options).unwrap();
+
+ let value = database
+ .get_option_string(OptionDatabase::Username)
+ .unwrap();
+ assert_eq!(value, "Alice");
+
+ let value = database.get_option_int(OptionDatabase::Password).unwrap();
+ assert_eq!(value, 42);
+
+ let value = database.get_option_double(OptionDatabase::Uri).unwrap();
+ assert_eq!(value, std::f64::consts::PI);
+
+ let value = database
+ .get_option_bytes(OptionDatabase::Other("pre.bytes".into()))
+ .unwrap();
+ assert_eq!(value, b"Hello");
+
+ let value = database
+ .get_option_string(OptionDatabase::Other("pre.string.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_STRING_LONG);
+
+ let value = database
+ .get_option_bytes(OptionDatabase::Other("pre.bytes.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_BYTES_LONG);
+
+ // Post-init options.
+ database
+ .set_option(OptionDatabase::Other("post.string".into()), "Bob".into())
+ .unwrap();
+ let value = database
+ .get_option_string(OptionDatabase::Other("post.string".into()))
+ .unwrap();
+ assert_eq!(value, "Bob");
+
+ database
+ .set_option(OptionDatabase::Other("post.int".into()), 1337.into())
+ .unwrap();
+ let value = database
+ .get_option_int(OptionDatabase::Other("post.int".into()))
+ .unwrap();
+ assert_eq!(value, 1337);
+
+ database
+ .set_option(OptionDatabase::Other("post.double".into()), 1.41.into())
+ .unwrap();
+ let value = database
+ .get_option_double(OptionDatabase::Other("post.double".into()))
+ .unwrap();
+ assert_eq!(value, 1.41);
+
+ database
+ .set_option(OptionDatabase::Other("post.bytes".into()), b"Bye".into())
+ .unwrap();
+ let value = database
+ .get_option_bytes(OptionDatabase::Other("post.bytes".into()))
+ .unwrap();
+ assert_eq!(value, b"Bye");
+
+ database
+ .set_option(
+ OptionDatabase::Other("post.string.long".into()),
+ OPTION_STRING_LONG.into(),
+ )
+ .unwrap();
+ let value = database
+ .get_option_string(OptionDatabase::Other("post.string.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_STRING_LONG);
+
+ database
+ .set_option(
+ OptionDatabase::Other("post.bytes.long".into()),
+ OPTION_BYTES_LONG.into(),
+ )
+ .unwrap();
+ let value = database
+ .get_option_bytes(OptionDatabase::Other("post.bytes.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_BYTES_LONG);
+}
+
+// Connection
+
+#[test]
+fn test_connection_options() {
+ let (_, mut database, _, _) = get_exported();
+
+ // Pre-init options
+ let options = [
+ (OptionConnection::CurrentCatalog, "Alice".into()),
+ (OptionConnection::AutoCommit, 42.into()),
+ (OptionConnection::CurrentSchema, std::f64::consts::PI.into()),
+ (
+ OptionConnection::IsolationLevel,
+ IsolationLevel::Linearizable.into(),
+ ),
+ (OptionConnection::Other("pre.bytes".into()), b"Hello".into()),
+ (OptionConnection::ReadOnly, OPTION_STRING_LONG.into()),
+ (
+ OptionConnection::Other("pre.bytes.long".into()),
+ OPTION_BYTES_LONG.into(),
+ ),
+ ];
+ let mut connection = database.new_connection_with_opts(options).unwrap();
+
+ let value = connection
+ .get_option_string(OptionConnection::CurrentCatalog)
+ .unwrap();
+ assert_eq!(value, "Alice");
+
+ let value = connection
+ .get_option_int(OptionConnection::AutoCommit)
+ .unwrap();
+ assert_eq!(value, 42);
+
+ let value = connection
+ .get_option_double(OptionConnection::CurrentSchema)
+ .unwrap();
+ assert_eq!(value, std::f64::consts::PI);
+
+ let value = connection
+ .get_option_string(OptionConnection::IsolationLevel)
+ .unwrap();
+ assert_eq!(value, Into::<String>::into(IsolationLevel::Linearizable));
+
+ let value = connection
+ .get_option_bytes(OptionConnection::Other("pre.bytes".into()))
+ .unwrap();
+ assert_eq!(value, b"Hello");
+
+ let value = connection
+ .get_option_string(OptionConnection::ReadOnly)
+ .unwrap();
+ assert_eq!(value, OPTION_STRING_LONG);
+
+ let value = connection
+ .get_option_bytes(OptionConnection::Other("pre.bytes.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_BYTES_LONG);
+
+ // Post-init options
+ connection
+ .set_option(OptionConnection::AutoCommit, "true".into())
+ .unwrap();
+ let value = connection
+ .get_option_string(OptionConnection::AutoCommit)
+ .unwrap();
+ assert_eq!(value, "true");
+
+ connection
+ .set_option(OptionConnection::CurrentCatalog, 1337.into())
+ .unwrap();
+ let value = connection
+ .get_option_int(OptionConnection::CurrentCatalog)
+ .unwrap();
+ assert_eq!(value, 1337);
+
+ connection
+ .set_option(OptionConnection::CurrentSchema, 1.41.into())
+ .unwrap();
+ let value = connection
+ .get_option_double(OptionConnection::CurrentSchema)
+ .unwrap();
+ assert_eq!(value, 1.41);
+
+ connection
+ .set_option(OptionConnection::Other("post.bytes".into()),
b"Bye".into())
+ .unwrap();
+ let value = connection
+ .get_option_bytes(OptionConnection::Other("post.bytes".into()))
+ .unwrap();
+ assert_eq!(value, b"Bye");
+
+ connection
+ .set_option(
+ OptionConnection::Other("post.string.long".into()),
+ OPTION_STRING_LONG.into(),
+ )
+ .unwrap();
+ let value = connection
+ .get_option_string(OptionConnection::Other("post.string.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_STRING_LONG);
+
+ connection
+ .set_option(
+ OptionConnection::Other("post.bytes.long".into()),
+ OPTION_BYTES_LONG.into(),
+ )
+ .unwrap();
+ let value = connection
+ .get_option_bytes(OptionConnection::Other("post.bytes.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_BYTES_LONG);
+}
+
+#[test]
+fn test_connection_get_table_types() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_table_types =
concat_reader(exported_connection.get_table_types().unwrap());
+ let native_table_types =
concat_reader(native_connection.get_table_types().unwrap());
+
+ assert_eq!(
+ exported_table_types.schema(),
+ *schemas::GET_TABLE_TYPES_SCHEMA.deref()
+ );
+ assert_eq!(exported_table_types, native_table_types);
+}
+
+#[test]
+fn test_connection_get_table_schema() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_schema = exported_connection
+ .get_table_schema(Some("default"), Some("default"), "default")
+ .unwrap();
+ let native_schema = native_connection
+ .get_table_schema(Some("default"), Some("default"), "default")
+ .unwrap();
+
+ assert_eq!(exported_schema, native_schema);
+}
+
+#[test]
+fn test_connection_get_info() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_info =
concat_reader(exported_connection.get_info(None).unwrap());
+ let native_info = concat_reader(native_connection.get_info(None).unwrap());
+ assert_eq!(exported_info.schema(), *schemas::GET_INFO_SCHEMA.deref());
+ assert_eq!(exported_info, native_info);
+
+ let exported_info = concat_reader(
+ exported_connection
+ .get_info(Some(
+ [InfoCode::DriverAdbcVersion, InfoCode::DriverName].into(),
+ ))
+ .unwrap(),
+ );
+ let native_info = concat_reader(
+ native_connection
+ .get_info(Some(
+ [InfoCode::DriverAdbcVersion, InfoCode::DriverName].into(),
+ ))
+ .unwrap(),
+ );
+ assert_eq!(exported_info.schema(), *schemas::GET_INFO_SCHEMA.deref());
+ assert_eq!(exported_info, native_info);
+}
+
+#[test]
+fn test_connection_cancel() {
+ let (_, _, mut exported_connection, _) = get_exported();
+ let (_, _, mut native_connection, _) = get_native();
+
+ let exported_error = exported_connection.cancel().unwrap_err();
+ let native_error = native_connection.cancel().unwrap_err();
+
+ assert_eq!(exported_error, native_error);
+}
+
+#[test]
+fn test_connection_commit_rollback() {
+ let (_, _, mut exported_connection, _) = get_exported();
+ let (_, _, mut native_connection, _) = get_native();
+
+ exported_connection.commit().unwrap();
+ exported_connection.rollback().unwrap();
+
+ native_connection.commit().unwrap();
+ native_connection.rollback().unwrap();
+}
+
+#[test]
+fn test_connection_get_statistic_names() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_names =
concat_reader(exported_connection.get_statistic_names().unwrap());
+ let native_names =
concat_reader(native_connection.get_statistic_names().unwrap());
+
+ assert_eq!(
+ exported_names.schema(),
+ *schemas::GET_STATISTIC_NAMES_SCHEMA.deref()
+ );
+ assert_eq!(exported_names, native_names);
+}
+
+#[test]
+fn test_connection_read_partition() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_partition =
concat_reader(exported_connection.read_partition(b"").unwrap());
+ let native_partition =
concat_reader(native_connection.read_partition(b"").unwrap());
+
+ assert_eq!(
+ exported_partition.schema(),
+ exported_connection
+ .get_table_schema(None, None, "default")
+ .unwrap()
+ .into()
+ );
+ assert_eq!(exported_partition, native_partition);
+}
+
+#[test]
+fn test_connection_get_statistics() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_statistics = concat_reader(
+ exported_connection
+ .get_statistics(None, None, None, false)
+ .unwrap(),
+ );
+ let native_statistics = concat_reader(
+ native_connection
+ .get_statistics(None, None, None, false)
+ .unwrap(),
+ );
+
+ assert_eq!(exported_statistics, native_statistics);
+ assert_eq!(
+ exported_statistics.schema(),
+ schemas::GET_STATISTICS_SCHEMA.clone(),
+ );
+}
+
+#[test]
+fn test_connection_get_objects() {
+ let (_, _, exported_connection, _) = get_exported();
+ let (_, _, native_connection, _) = get_native();
+
+ let exported_objects = concat_reader(
+ exported_connection
+ .get_objects(
+ ObjectDepth::All,
+ None,
+ None,
+ None,
+ Some(vec!["table", "view"]),
+ None,
+ )
+ .unwrap(),
+ );
+ let native_objects = concat_reader(
+ native_connection
+ .get_objects(ObjectDepth::All, None, None, None, None, None)
+ .unwrap(),
+ );
+
+ assert_eq!(exported_objects, native_objects);
+ assert_eq!(
+ exported_objects.schema(),
+ schemas::GET_OBJECTS_SCHEMA.clone(),
+ );
+}
+
+// Statement
+
+#[test]
+fn test_statement_options() {
+ let (_, _, _, mut statement) = get_exported();
+
+ statement
+ .set_option(OptionStatement::Incremental, "true".into())
+ .unwrap();
+ let value = statement
+ .get_option_string(OptionStatement::Incremental)
+ .unwrap();
+ assert_eq!(value, "true");
+
+ statement
+ .set_option(OptionStatement::TargetTable, 42.into())
+ .unwrap();
+ let value = statement
+ .get_option_int(OptionStatement::TargetTable)
+ .unwrap();
+ assert_eq!(value, 42);
+
+ statement
+ .set_option(OptionStatement::MaxProgress, std::f64::consts::PI.into())
+ .unwrap();
+ let value = statement
+ .get_option_double(OptionStatement::MaxProgress)
+ .unwrap();
+ assert_eq!(value, std::f64::consts::PI);
+
+ statement
+ .set_option(OptionStatement::Other("bytes".into()), b"Hello".into())
+ .unwrap();
+ let value = statement
+ .get_option_bytes(OptionStatement::Other("bytes".into()))
+ .unwrap();
+ assert_eq!(value, b"Hello");
+
+ statement
+ .set_option(OptionStatement::IngestMode,
IngestMode::CreateAppend.into())
+ .unwrap();
+ let value = statement
+ .get_option_string(OptionStatement::IngestMode)
+ .unwrap();
+ assert_eq!(value, Into::<String>::into(IngestMode::CreateAppend));
+
+ statement
+ .set_option(
+ OptionStatement::Other("bytes.long".into()),
+ OPTION_BYTES_LONG.into(),
+ )
+ .unwrap();
+ let value = statement
+ .get_option_bytes(OptionStatement::Other("bytes.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_BYTES_LONG);
+
+ statement
+ .set_option(
+ OptionStatement::Other("string.long".into()),
+ OPTION_STRING_LONG.into(),
+ )
+ .unwrap();
+ let value = statement
+ .get_option_string(OptionStatement::Other("string.long".into()))
+ .unwrap();
+ assert_eq!(value, OPTION_STRING_LONG);
+}
+
+#[test]
+fn test_statement_bind() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ let batch = sample_batch();
+
+ exported_statement.bind(batch.clone()).unwrap();
+ native_statement.bind(batch).unwrap();
+}
+
+#[test]
+fn test_statement_bind_stream() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ let batch = sample_batch();
+ let reader = Box::new(SingleBatchReader::new(batch));
+ exported_statement.bind_stream(reader).unwrap();
+
+ let batch = sample_batch();
+ let reader = Box::new(SingleBatchReader::new(batch));
+ native_statement.bind_stream(reader).unwrap();
+}
+
+#[test]
+fn test_statement_cancel() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ exported_statement.cancel().unwrap();
+ native_statement.cancel().unwrap();
+}
+
+#[test]
+fn test_statement_execute_query() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ let exported_data = concat_reader(exported_statement.execute().unwrap());
+ let native_data = concat_reader(native_statement.execute().unwrap());
+ assert_eq!(exported_data, native_data);
+
+ let exported_data = exported_statement.execute_update().unwrap();
+ let native_data = native_statement.execute_update().unwrap();
+ assert_eq!(exported_data, native_data);
+}
+
+#[test]
+fn test_statement_execute_schema() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ let exported_schema = exported_statement.execute_schema().unwrap();
+ let native_schema = native_statement.execute_schema().unwrap();
+ assert_eq!(exported_schema, native_schema);
+}
+
+#[test]
+fn test_statement_execute_partitions() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ let exported_result = exported_statement.execute_partitions().unwrap();
+ let native_result = native_statement.execute_partitions().unwrap();
+ assert_eq!(exported_result, native_result);
+}
+
+#[test]
+fn test_statement_prepare() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ exported_statement.prepare().unwrap();
+ native_statement.prepare().unwrap();
+}
+
+#[test]
+fn test_statement_set_sql_query() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ exported_statement
+ .set_sql_query("select * from table")
+ .unwrap();
+ native_statement
+ .set_sql_query("select * from table")
+ .unwrap();
+}
+
+#[test]
+fn test_statement_set_substrait_plan() {
+ let (_, _, _, mut exported_statement) = get_exported();
+ let (_, _, _, mut native_statement) = get_native();
+
+ exported_statement.set_substrait_plan(b"SCAN").unwrap();
+ native_statement.set_substrait_plan(b"SCAN").unwrap();
+}
+
+#[test]
+fn test_statement_get_parameter_schema() {
+ let (_, _, _, exported_statement) = get_exported();
+ let (_, _, _, native_statement) = get_native();
+
+ let exported_schema = exported_statement.get_parameter_schema().unwrap();
+ let native_schema = native_statement.get_parameter_schema().unwrap();
+ assert_eq!(exported_schema, native_schema);
+}
