Merge branch 'spiceai' into lukim/skip-constraint-validation-on-overwrite-v2

Author: lukekim

Cargo.lock

@@ -50,9 +50,50 @@ datafusion-table-providers = { path = "core" }
 duckdb = "1.3.2" # Forked to add support for duckdb_scan_arrow, pending: https://github.com/duckdb/duckdb-rs/pull/488
 adbc_core = { version = "0.20.0" }
 adbc_driver_manager = { version = "0.20.0" }
+parquet = "56"
+rusqlite = "0.37"
+tokio-rusqlite = "0.7.0"
+async-stream = "0.3.5"
+async-trait = "0.1.80"
+bigdecimal = "0.4.5"
+byteorder = "1.5.0"
+chrono = "0.4"
+futures = "0.3.30"
+itertools = "0.14.0"
+num-bigint = "0.4"
+secrecy = "0.10.3"
+serde = { version = "1.0.209", features = ["derive"] }
+serde_json = "1.0.124"
+snafu = "0.8.3"
+time = "0.3.36"
+tokio = { version = "1.46", features = ["macros", "fs"] }
+tracing = "0.1.40"
+url = "2.5.1"
+rand = "0.9"
+regex = "1"
+uuid = "1.9.1"
+prost = "0.14.1"
+tonic = { version = "0.13", features = ["tls-native-roots", "tls-webpki-roots"] }
+bytes = "1.7.1"
+base64 = "0.22.1"
+dashmap = "6.1.0"
+fallible-iterator = "0.3.0"
+fundu = "2.0.1"
+geo-types = "0.7.13"
+sha2 = "0.10"
+hickory-resolver = "0.25.2"
+sea-query = { git = "https://github.com/spiceai/sea-query.git", rev = "213b6b876068f58159ebdd5852604a021afaebf9", features = [
+  "backend-sqlite",
+  "backend-postgres",
+  "postgres-array",
+  "with-rust_decimal",
+  "with-bigdecimal",
+  "with-time",
+  "with-chrono",
+] }
 
 [patch.crates-io]
-datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "007c82d039220478f406d648417d28c0d40a7987" } # spiceai-50
+datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "afcbd1ce99703f1322d176fc1b99745f647234d0" } # spiceai-50
 duckdb = { git = "https://github.com/spiceai/duckdb-rs.git", rev = "dd02045c3aa77895723e873222cbe30f5c8f77a9" }                            # spiceai-1.3.2
 
 datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "41d08054e37e9cdca9b760c16fc2b97a21893af2" }               # spiceai-50
@@ -71,3 +112,5 @@ arrow-ipc = { git = "https://github.com/spiceai/arrow-rs.git", rev = "9f9c372ff5
 arrow-ord = { git = "https://github.com/spiceai/arrow-rs.git", rev = "9f9c372ff5744488226462e3ea7d94ff47909833" }    # spiceai-56
 arrow-schema = { git = "https://github.com/spiceai/arrow-rs.git", rev = "9f9c372ff5744488226462e3ea7d94ff47909833" } # spiceai-56
 parquet = { git = "https://github.com/spiceai/arrow-rs.git", rev = "9f9c372ff5744488226462e3ea7d94ff47909833" }      # spiceai-56
+
+rusqlite = { git = "https://github.com/spiceai/rusqlite.git", rev = "3d1f5f6f6d6d062676210d095df45eafa6e19fd8" } # spiceai-v0.37.0

Cargo.toml

@@ -18,18 +18,18 @@ arrow-flight = { workspace = true, optional = true, features = [
 arrow-schema = { workspace = true, optional = true, features = ["serde"] }
 arrow-json = { workspace = true }
 arrow-odbc = { workspace = true, optional = true }
-async-stream = { version = "0.3.5", optional = true }
-async-trait = "0.1.80"
-base64 = { version = "0.22.1", optional = true }
+async-stream = { workspace = true, optional = true }
+async-trait = { workspace = true }
+base64 = { workspace = true, optional = true }
 bb8 = { version = "0.9.0", optional = true }
 bb8-postgres = { version = "0.9.0", optional = true }
-bigdecimal = "0.4.5"
-byteorder = "1.5.0"
-bytes = { version = "1.7.1", optional = true }
+bigdecimal = { workspace = true }
+byteorder = { workspace = true }
+bytes = { workspace = true, optional = true }
 byte-unit = { version = "5.1.4", optional = true }
-chrono = "0.4"
+chrono = { workspace = true }
 clickhouse = { version = "0.13.3", optional = true }
-dashmap = "6.1.0"
+dashmap = { workspace = true }
 datafusion = { workspace = true, default-features = false }
 datafusion-expr = { workspace = true, optional = true }
 datafusion-federation = { workspace = true, features = ["sql"], optional = true }
@@ -45,11 +45,11 @@ duckdb = { workspace = true, features = [
 ], optional = true }
 libduckdb-sys = { version = "=1.3.0", optional = true }
 dyn-clone = { version = "1.0.17", optional = true }
-fallible-iterator = "0.3.0"
-fundu = "2.0.1"
-futures = "0.3.30"
-geo-types = "0.7.13"
-itertools = "0.14.0"
+fallible-iterator = { workspace = true }
+fundu = { workspace = true }
+futures = { workspace = true }
+geo-types = { workspace = true }
+itertools = { workspace = true }
 mongodb = { version = "3.2.2", features = ["openssl-tls"], optional = true }
 mysql_async = { version = "0.36", features = [
   "native-tls-tls",
@@ -58,49 +58,39 @@ mysql_async = { version = "0.36", features = [
   "bigdecimal",
 ], optional = true }
 native-tls = { version = "0.2.11", optional = true }
-num-bigint = "0.4"
+num-bigint = { workspace = true }
 num-traits = { version = "0.2", optional = true }
 odbc-api = { version = "19.0", optional = true }
+once_cell = "1.21"
 pem = { version = "3.0.4", optional = true }
 postgres-native-tls = { version = "0.5.0", optional = true }
-prost = { version = "0.14.1", optional = true }
-rand = { version = "0.9" }
-regex = { version = "1" }
+prost = { workspace = true, optional = true }
+rand = { workspace = true }
+regex = { workspace = true }
 r2d2 = { version = "0.8", optional = true }
 r2d2_adbc = { version = "0.1.0", optional = true }
-rusqlite = { version = "0.37", optional = true }
+rusqlite = { workspace = true, optional = true }
 rust_decimal = { version = "1.38.0", features = ["db-postgres"], optional = true }
-sea-query = { version = "0.32", features = [
-  "backend-sqlite",
-  "backend-postgres",
-  "postgres-array",
-  "with-rust_decimal",
-  "with-bigdecimal",
-  "with-time",
-  "with-chrono",
-] }
-secrecy = "0.10.3"
-serde = { version = "1.0.209", features = ["derive"] }
-serde_json = "1.0.124"
-sha2 = "0.10"
-snafu = "0.8.3"
-time = "0.3.36"
-tokio = { version = "1.46", features = ["macros", "fs"] }
+sea-query = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sha2 = { workspace = true }
+snafu = { workspace = true }
+time = { workspace = true }
+tokio = { workspace = true }
 tokio-postgres = { version = "0.7.10", features = [
   "with-chrono-0_4",
   "with-uuid-1",
   "with-serde_json-1",
   "with-geo-types-0_7",
 ], optional = true }
-tokio-rusqlite = { version = "0.7.0", optional = true }
-tonic = { version = "0.13", optional = true, features = [
-  "tls-native-roots",
-  "tls-webpki-roots",
-] }
-tracing = "0.1.40"
-uuid = { version = "1.9.1", optional = true }
-hickory-resolver = "0.25.2"
-url = "2.5.1"
+tokio-rusqlite = { workspace = true, optional = true }
+tonic = { workspace = true, optional = true }
+tracing = { workspace = true }
+uuid = { workspace = true, optional = true }
+hickory-resolver = { workspace = true }
+url = { workspace = true }
 adbc_driver_manager = { workspace = true, optional = true }
 adbc_core = { workspace = true, optional = true }
 
@@ -109,9 +99,9 @@ anyhow = "1.0"
 bollard = "0.19"
 geozero = { version = "0.14.0", features = ["with-wkb"] }
 insta = { version = "1.43.2", features = ["filters"] }
-parquet = "56"
+parquet = { workspace = true }
 prost = { version = "=0.13.5" }
-rand = "0.9"
+rand = { workspace = true }
 reqwest = "0.12"
 rstest = "0.26.1"
 test-log = { version = "0.2", features = ["trace"] }
@@ -130,6 +120,7 @@ duckdb = [
   "dep:async-stream",
   "dep:arrow-schema",
   "dep:byte-unit",
+  "dep:datafusion-physical-expr",
 ]
 duckdb-federation = ["duckdb", "federation"]
 federation = ["dep:datafusion-federation"]
@@ -152,6 +143,7 @@ mongodb = [
   "dep:async-stream",
   "dep:arrow-schema",
   "dep:num-traits",
+  "dep:rust_decimal",
 ]
 mysql = ["dep:mysql_async", "dep:async-stream"]
 mysql-federation = ["mysql", "federation"]
@@ -173,6 +165,7 @@ postgres-federation = ["postgres", "federation"]
 sqlite = ["dep:rusqlite", "dep:tokio-rusqlite", "dep:arrow-schema"]
 sqlite-federation = ["sqlite", "federation"]
 sqlite-bundled = ["sqlite", "rusqlite/bundled"]
+sqlite-bundled-decimal = ["sqlite", "rusqlite/bundled-decimal"]
 adbc = [
   "dep:adbc_driver_manager",
   "dep:adbc_core",

core/Cargo.toml

@@ -435,7 +435,10 @@ impl TableProviderFactory for DuckDBTableProviderFactory {
             Mode::File => {
                 let read_pool = pool.clone();
 
-                read_pool.set_attached_databases(&self.attach_databases(&options))
+                read_pool
+                    .set_attached_databases(&self.attach_databases(&options))
+                    .context(DbConnectionPoolSnafu)
+                    .map_err(to_datafusion_error)?
             }
             Mode::Memory => pool.clone(),
         };

core/src/duckdb.rs

@@ -26,6 +26,7 @@ use futures::TryStreamExt;
 use std::collections::HashMap;
 use std::fmt::Display;
 use std::{any::Any, fmt, sync::Arc};
+use datafusion_physical_expr::EquivalenceProperties;
 
 pub struct DuckDBTable<T: 'static, P: 'static> {
     pub(crate) base_table: SqlTable<T, P>,
@@ -146,6 +147,7 @@ pub struct DuckSqlExec<T, P> {
     indexes: Vec<(ColumnReference, IndexType)>,
     optimized_sql: Option<String>,
     optimized_sql_schema: Option<SchemaRef>,
+    optimized_sql_properties: Option<PlanProperties>,
 }
 
 impl<T, P> Clone for DuckSqlExec<T, P> {
@@ -156,11 +158,12 @@ impl<T, P> Clone for DuckSqlExec<T, P> {
             indexes: self.indexes.clone(),
             optimized_sql: self.optimized_sql.clone(),
             optimized_sql_schema: self.optimized_sql_schema.clone(),
+            optimized_sql_properties: self.optimized_sql_properties.clone(),
         }
     }
 }
 
-impl<T, P> DuckSqlExec<T, P> {
+impl<T: 'static, P: 'static> DuckSqlExec<T, P> {
     #[allow(clippy::too_many_arguments)]
     fn new(
         projections: Option<&Vec<usize>>,
@@ -188,6 +191,7 @@ impl<T, P> DuckSqlExec<T, P> {
             indexes,
             optimized_sql: None,
             optimized_sql_schema: None,
+            optimized_sql_properties: None,
         })
     }
 
@@ -225,18 +229,26 @@ impl<T, P> DuckSqlExec<T, P> {
     ) -> Self {
         self.optimized_sql = Some(sql.into());
         self.optimized_sql_schema = new_schema;
+
+        if let Some(schema) = self.optimized_sql_schema.as_ref() {
+            let mut properties = self.base_exec.properties().clone();
+            let eq_properties = EquivalenceProperties::new(Arc::clone(schema));
+            properties.eq_properties = eq_properties;
+            self.optimized_sql_properties = Some(properties);
+        }
+
         self
     }
 }
 
-impl<T, P> std::fmt::Debug for DuckSqlExec<T, P> {
+impl<T: 'static, P: 'static> std::fmt::Debug for DuckSqlExec<T, P> {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         let sql = self.sql().unwrap_or_default();
         write!(f, "DuckSqlExec sql={sql}")
     }
 }
 
-impl<T, P> DisplayAs for DuckSqlExec<T, P> {
+impl<T: 'static, P: 'static> DisplayAs for DuckSqlExec<T, P> {
     fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter) -> std::fmt::Result {
         let sql = self.sql().unwrap_or_default();
         write!(f, "DuckSqlExec sql={sql}")
@@ -260,7 +272,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
     }
 
     fn properties(&self) -> &PlanProperties {
-        self.base_exec.properties()
+        self.optimized_sql_properties.as_ref().unwrap_or(self.base_exec.properties())
     }
 
     fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {