Merge pull request #13 from datafusion-contrib/12-bump-datafusion-v54

Bump DataFusion to 54.0.0 (#12)

Author: PierreZ

Cargo.toml

@@ -10,19 +10,25 @@ documentation = "https://docs.rs/datafusion-index-provider"
 keywords = ["datafusion", "index", "database", "query", "analytics"]
 categories = ["database", "data-structures"]
 readme = "README.md"
-rust-version = "1.75"
+rust-version = "1.88"
 
 [dependencies]
 async-trait = "0.1.89"
-datafusion = { version = "52.1.0", default-features = false }
-datafusion-common = { version = "52.1.0", default-features = false }
+datafusion = { version = "54.0.0", default-features = false }
+datafusion-common = { version = "54.0.0", default-features = false }
 futures = "0.3.32"
 futures-util = "0.3.32"
 futures-core = "0.3.32"
-log = "0.4.29"
+log = "0.4.33"
 
 [dev-dependencies]
-tokio = { version = "1.49.0", features = ["macros", "rt-multi-thread"] }
-env_logger = "0.11.9"
+tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread"] }
+env_logger = "0.11.11"
 parking_lot = "0.12.5"
-datafusion = { version = "52.1.0", default-features = false, features = ["sql"] }
+datafusion = { version = "54.0.0", default-features = false, features = ["sql"] }
+datafusion-sqllogictest = { version = "54.0.0", default-features = false }
+sqllogictest = "0.29"
+indicatif = "0.18"
+
+[lints.clippy]
+pedantic = { level = "warn", priority = -1 }

README.md

@@ -222,18 +222,21 @@ cargo test
 # Run with logging
 RUST_LOG=debug cargo test
 
-# Run integration tests only
-cargo test --test integration_tests
+# Run the SQL logic tests only
+cargo test --test sqllogictests
+
+# Regenerate the expected .slt results after an intended change
+SLT_COMPLETE=1 cargo test --test sqllogictests
 ```
 
 The test suite includes:
 - 27 unit tests covering execution plan generation and streaming
-- 36 integration tests covering query scenarios from simple to deeply nested, with both single and composite primary keys
+- SQL logic tests (`tests/slt/*.slt`, run via [sqllogictest](https://github.com/risinglightdb/sqllogictest-rs)) covering query scenarios from simple to deeply nested, with both single and composite primary keys
 
 ## Compatibility
 
-- **DataFusion**: 52.x
-- **Rust**: 1.75+ (MSRV)
+- **DataFusion**: 54.x
+- **Rust**: 1.88+ (MSRV)
 - **Arrow**: Compatible with DataFusion's Arrow version
 
 ## Architecture Details

examples/basic.rs

@@ -18,7 +18,7 @@
 //! Basic example demonstrating how to use `datafusion-index-provider`.
 //!
 //! This example creates an in-memory table with an age index and a department index,
-//! registers it as a DataFusion table, and runs queries that leverage index-based scans.
+//! registers it as a `DataFusion` table, and runs queries that leverage index-based scans.
 
 use std::any::Any;
 use std::collections::{BTreeMap, BTreeSet};
@@ -53,20 +53,20 @@ use datafusion_index_provider::{IndexedTableProvider, RecordFetcher, UnionMode};
 #[derive(Debug)]
 struct AgeIndex {
     /// age -> list of row ids
-    index: BTreeMap<i32, Vec<i32>>,
+    index: BTreeMap<i32, Vec<u64>>,
 }
 
 impl AgeIndex {
-    fn new(ages: &Int32Array, ids: &Int32Array) -> Self {
-        let mut index: BTreeMap<i32, Vec<i32>> = BTreeMap::new();
+    fn new(ages: &Int32Array, ids: &UInt64Array) -> Self {
+        let mut index: BTreeMap<i32, Vec<u64>> = BTreeMap::new();
         for i in 0..ages.len() {
             index.entry(ages.value(i)).or_default().push(ids.value(i));
         }
         Self { index }
     }
 
     fn matching_ids(&self, filters: &[Expr], limit: Option<usize>) -> Vec<u64> {
-        let mut ids: BTreeSet<i32> = BTreeSet::new();
+        let mut ids: BTreeSet<u64> = BTreeSet::new();
         for filter in filters {
             if let Expr::BinaryExpr(be) = filter {
                 if let (Expr::Column(c), Expr::Literal(ScalarValue::Int32(Some(v)), _)) =
@@ -82,7 +82,7 @@ impl AgeIndex {
                             }
                         }
                         Operator::Gt => {
-                            ids.extend(self.index.range((v + 1)..).flat_map(|(_, l)| l))
+                            ids.extend(self.index.range((v + 1)..).flat_map(|(_, l)| l));
                         }
                         Operator::GtEq => ids.extend(self.index.range(v..).flat_map(|(_, l)| l)),
                         Operator::Lt => ids.extend(self.index.range(..v).flat_map(|(_, l)| l)),
@@ -92,7 +92,7 @@ impl AgeIndex {
                 }
             }
         }
-        let mut result: Vec<u64> = ids.into_iter().map(|id| id as u64).collect();
+        let mut result: Vec<u64> = ids.into_iter().collect();
         if let Some(l) = limit {
             result.truncate(l);
         }
@@ -104,16 +104,16 @@ impl Index for AgeIndex {
     fn as_any(&self) -> &dyn Any {
         self
     }
-    fn name(&self) -> &str {
+    fn name(&self) -> &'static str {
         "age_index"
     }
     fn index_schema(&self) -> SchemaRef {
         create_index_schema([Field::new("id", DataType::UInt64, false)])
     }
-    fn table_name(&self) -> &str {
+    fn table_name(&self) -> &'static str {
         "employees"
     }
-    fn column_name(&self) -> &str {
+    fn column_name(&self) -> &'static str {
         "age"
     }
     fn scan(
@@ -168,7 +168,7 @@ impl RecordFetcher for InMemoryFetcher {
             .expect("expected UInt64Array for primary key column");
 
         // Convert 1-based ids to 0-based indices for arrow take
-        let indices = Int32Array::from_iter_values(ids.iter().flatten().map(|id| (id - 1) as i32));
+        let indices = UInt64Array::from_iter_values(ids.iter().flatten().map(|id| id - 1));
 
         let columns: Result<Vec<ArrayRef>> = self
             .batch
@@ -200,9 +200,6 @@ struct EmployeeTable {
 
 #[async_trait]
 impl TableProvider for EmployeeTable {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
@@ -269,20 +266,17 @@ async fn main() -> Result<()> {
         ],
     )?;
 
-    let ids = batch
-        .column(0)
-        .as_any()
-        .downcast_ref::<Int32Array>()
-        .unwrap();
     let ages = batch
         .column(2)
         .as_any()
         .downcast_ref::<Int32Array>()
-        .unwrap();
+        .expect("age column is an Int32Array");
+    // Row identifiers matching the index schema's UInt64 primary key column.
+    let pk_ids = UInt64Array::from(vec![1u64, 2, 3, 4, 5]);
 
     let provider = EmployeeTable {
         schema: schema.clone(),
-        age_index: Arc::new(AgeIndex::new(ages, ids)),
+        age_index: Arc::new(AgeIndex::new(ages, &pk_ids)),
         fetcher: Arc::new(InMemoryFetcher {
             batch: batch.clone(),
         }),

flake.lock

@@ -17,10 +17,10 @@
 
 #![warn(missing_docs)]
 
-//! # DataFusion Index Provider
+//! # `DataFusion` Index Provider
 //!
 //! This crate provides a comprehensive framework for adding index-based scanning capabilities
-//! to DataFusion [`datafusion::datasource::TableProvider`]s. It enables efficient query execution
+//! to `DataFusion` [`datafusion::datasource::TableProvider`]s. It enables efficient query execution
 //! by leveraging secondary indexes to reduce I/O and improve query performance through a
 //! sophisticated two-phase execution model.
 //!
@@ -38,7 +38,7 @@
 //!
 //! ### Index Management
 //! - [`physical_plan::Index`]: Trait representing a physical index that can be scanned to retrieve primary key values
-//! - [`provider::IndexedTableProvider`]: Extension of DataFusion's `TableProvider` with index discovery
+//! - [`provider::IndexedTableProvider`]: Extension of `DataFusion`'s `TableProvider` with index discovery
 //! - [`types::IndexFilter`]: Enum representing filter operations that can be pushed down to indexes
 //!
 //! ### Execution Engine  
@@ -57,7 +57,7 @@
 //! - **Default implementation**: Supports any predicate that references the index's column name
 //! - **Custom implementations**: Can implement sophisticated predicate analysis for complex index types
 //!
-//! ### IndexedTableProvider Filter Analysis
+//! ### `IndexedTableProvider` Filter Analysis
 //! The [`provider::IndexedTableProvider`] trait provides comprehensive filter analysis through
 //! `build_index_filter()`:
 //!
@@ -96,15 +96,15 @@
 //! The [`physical_plan::exec::fetch::RecordFetchExec`] generates
 //! optimized execution plans based on the [`types::IndexFilter`] structure:
 //!
-//! ### IndexFilter::Single - Direct Index Scan
+//! ### `IndexFilter::Single` - Direct Index Scan
 //!
 //! For simple conditions on a single indexed column:
 //! ```text
 //! RecordFetchExec
 //! └── IndexScanExec (target_index)
 //! ```
 //!
-//! ### IndexFilter::And - Index Intersection
+//! ### `IndexFilter::And` - Index Intersection
 //!
 //! For conjunctive conditions across multiple indexes, the system builds a left-deep tree
 //! of joins to intersect primary key values:
@@ -119,7 +119,7 @@
 //!         └── IndexScanExec (col_c_index)
 //! ```
 //!
-//! ### IndexFilter::Or - Union with Deduplication
+//! ### `IndexFilter::Or` - Union with Deduplication
 //!
 //! For disjunctive conditions, the system uses `UnionExec` followed by `AggregateExec`
 //! for automatic primary key deduplication:
@@ -138,9 +138,9 @@
 //! ## Implementation Guide
 //!
 //! - **Implement the Index Trait**: Create indexes that can scan and return primary key values
-//! - **Implement the RecordFetcher Trait**: Define how to fetch complete records using primary key values
-//! - **Implement IndexedTableProvider**: Expose available indexes and filter analysis capabilities
-//! - **Update TableProvider Implementation**: Integrate index-based execution into your scan method
+//! - **Implement the `RecordFetcher` Trait**: Define how to fetch complete records using primary key values
+//! - **Implement `IndexedTableProvider`**: Expose available indexes and filter analysis capabilities
+//! - **Update `TableProvider` Implementation**: Integrate index-based execution into your scan method
 //!
 //! ## Performance Characteristics
 //!