Upgrade to DataFusion 53 (#6740)

AdamGS · web-flow · commit 9b1f954f14a6 · 2026-03-26T14:08:20.000Z
Just a routine upgrade to DataFusion 53. - Tracking issue for the release: apache/datafusion#19692 This PR includes a perf regression that is fixed by #7176, keeping them as separate PRs so it doesn't get lost in all the noise here. ~Things left here~ All done! - [x] Either wait for clflushopt/tpchgen-rs#246 to merge and release, or change how we generate data in benchmarks (either by removing the dependency on `tpchgen-arrow` or using the CLI). - [x] Go through the 53 release changelog and see if there's anything new we should take advantage of or adjust. - [x] Finish the `object_store` upgrade --------- Signed-off-by: Adam Gutglick <adam@spiraldb.com>
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -88,16 +88,16 @@ arbitrary = "1.3.2"
 arc-swap = "1.8"
 arcref = "0.2.0"
 arrayref = "0.3.7"
-arrow-arith = "57.1"
-arrow-array = "57.1"
-arrow-buffer = "57.1"
-arrow-cast = "57.1"
-arrow-data = "57.1"
-arrow-ipc = "57.1"
-arrow-ord = "57.1"
-arrow-schema = "57.1"
-arrow-select = "57.1"
-arrow-string = "57.1"
+arrow-arith = "58"
+arrow-array = "58"
+arrow-buffer = "58"
+arrow-cast = "58"
+arrow-data = "58"
+arrow-ipc = "58"
+arrow-ord = "58"
+arrow-schema = "58"
+arrow-select = "58"
+arrow-string = "58"
 async-fs = "2.2.0"
 async-lock = "3.4"
 async-stream = "0.3.6"
@@ -125,20 +125,20 @@ cudarc = { version = "0.18.2", features = [
 custom-labels = "0.4.4"
 daachorse = "1.0.0"
 dashmap = "6.1.0"
-datafusion = { version = "52", default-features = false, features = ["sql"] }
-datafusion-catalog = { version = "52" }
-datafusion-common = { version = "52" }
-datafusion-common-runtime = { version = "52" }
-datafusion-datasource = { version = "52", default-features = false }
-datafusion-execution = { version = "52" }
-datafusion-expr = { version = "52" }
-datafusion-functions = { version = "52" }
-datafusion-physical-expr = { version = "52" }
-datafusion-physical-expr-adapter = { version = "52" }
-datafusion-physical-expr-common = { version = "52" }
-datafusion-physical-plan = { version = "52" }
-datafusion-pruning = { version = "52" }
-datafusion-sqllogictest = { version = "52" }
+datafusion = { version = "53", default-features = false, features = ["sql"] }
+datafusion-catalog = { version = "53" }
+datafusion-common = { version = "53" }
+datafusion-common-runtime = { version = "53" }
+datafusion-datasource = { version = "53", default-features = false }
+datafusion-execution = { version = "53" }
+datafusion-expr = { version = "53" }
+datafusion-functions = { version = "53" }
+datafusion-physical-expr = { version = "53" }
+datafusion-physical-expr-adapter = { version = "53" }
+datafusion-physical-expr-common = { version = "53" }
+datafusion-physical-plan = { version = "53" }
+datafusion-pruning = { version = "53" }
+datafusion-sqllogictest = { version = "53" }
 dirs = "6.0.0"
 divan = { package = "codspeed-divan-compat", version = "4.0.4" }
 enum-iterator = "2.0.0"
@@ -176,14 +176,14 @@ noodles-bgzf = "0.46.0"
 noodles-vcf = { version = "0.86.0", features = ["async"] }
 num-traits = "0.2.19"
 num_enum = { version = "0.7.3", default-features = false }
-object_store = { version = "0.12.4", default-features = false }
+object_store = { version = "0.13.1", default-features = false }
 once_cell = "1.21"
 oneshot = { version = "0.2.0", features = ["async"] }
 opentelemetry = "0.31.0"
 opentelemetry-otlp = "0.31.0"
 opentelemetry_sdk = "0.31.0"
 parking_lot = { version = "0.12.3", features = ["nightly"] }
-parquet = "57.1"
+parquet = "58"
 paste = "1.0.15"
 pco = "1.0.1"
 pin-project-lite = "0.2.15"
@@ -196,8 +196,8 @@ public-api = "0.51"
 pyo3 = { version = "0.28.0" }
 pyo3-bytes = "0.6"
 pyo3-log = "0.13.0"
-pyo3-object_store = "0.10.0"
-quote = "1.0.41"
+pyo3-object_store = "0.9.0"
+quote = "1.0.44"
 rand = "0.10.0"
 rand_distr = "0.6"
 ratatui = { version = "0.30", default-features = false }
@@ -224,7 +224,7 @@ sketches-ddsketch = "0.4.0"
 smol = "2.0.2"
 static_assertions = "1.1"
 strum = "0.28"
-syn = { version = "2.0.113", features = ["full"] }
+syn = { version = "2.0.117", features = ["full"] }
 sysinfo = "0.38.0"
 tabled = { version = "0.20.0", default-features = false }
 taffy = "0.9.0"
@@ -238,13 +238,14 @@ thiserror = "2.0.3"
 tokio = { version = "1.48" }
 tokio-stream = "0.1.17"
 tokio-util = "0.7.17"
-tpchgen = { version = "2.0.2" }
-tpchgen-arrow = { version = "2.0.2" }
+# Pull these into non-public crates to support DF 58
+tpchgen = { version = "2.0.2", git = "https://github.com/clflushopt/tpchgen-rs.git", rev = "438e9c2dbc25b2fff82c0efc08b3f13b5707874f" }
+tpchgen-arrow = { version = "2.0.2", git = "https://github.com/clflushopt/tpchgen-rs.git", rev = "438e9c2dbc25b2fff82c0efc08b3f13b5707874f" }
 tracing = { version = "0.1.41", default-features = false }
 tracing-perfetto = "0.1.5"
 tracing-subscriber = "0.3"
 url = "2.5.7"
-uuid = { version = "1.19", features = ["js"] }
+uuid = { version = "1.21", features = ["js"] }
 wasm-bindgen-futures = "0.4.54"
 xshell = "0.2.6"
 zigzag = "0.1.0"
diff --git a/vortex-bench/src/random_access/take.rs b/vortex-bench/src/random_access/take.rs
@@ -14,6 +14,7 @@ use itertools::Itertools;
 use parquet::arrow::ParquetRecordBatchStreamBuilder;
 use parquet::arrow::arrow_reader::ArrowReaderMetadata;
 use parquet::arrow::arrow_reader::ArrowReaderOptions;
+use parquet::file::metadata::PageIndexPolicy;
 use stream::StreamExt;
 use tokio::fs::File;
 use vortex::array::Canonical;
@@ -100,7 +101,7 @@ impl ParquetRandomAccessor {
     /// Open a Parquet file, parse the footer, and return a ready-to-use accessor.
     pub async fn open(path: PathBuf, name: impl Into<String>) -> anyhow::Result<Self> {
         let mut file = File::open(&path).await?;
-        let options = ArrowReaderOptions::new().with_page_index(true);
+        let options = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required);
         let arrow_metadata = ArrowReaderMetadata::load_async(&mut file, options).await?;
 
         let row_group_offsets = once(0)
diff --git a/vortex-cuda/src/pooled_read_at.rs b/vortex-cuda/src/pooled_read_at.rs
@@ -13,6 +13,7 @@ use object_store::GetOptions;
 use object_store::GetRange;
 use object_store::GetResultPayload;
 use object_store::ObjectStore;
+use object_store::ObjectStoreExt;
 use object_store::path::Path as ObjectPath;
 use vortex::array::buffer::BufferHandle;
 use vortex::buffer::Alignment;
diff --git a/vortex-datafusion/src/convert/exprs.rs b/vortex-datafusion/src/convert/exprs.rs
@@ -395,7 +395,8 @@ fn try_operator_from_df(value: &DFOperator) -> DFResult<Operator> {
         | DFOperator::AtQuestion
         | DFOperator::Question
         | DFOperator::QuestionAnd
-        | DFOperator::QuestionPipe => {
+        | DFOperator::QuestionPipe
+        | DFOperator::Colon => {
             tracing::debug!(operator = %value, "Can't pushdown binary_operator operator");
             Err(exec_datafusion_err!(
                 "Unsupported datafusion operator {value}"
diff --git a/vortex-datafusion/src/persistent/format.rs b/vortex-datafusion/src/persistent/format.rs
@@ -32,6 +32,7 @@ use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
 use datafusion_datasource::file_sink_config::FileSinkConfig;
 use datafusion_datasource::sink::DataSinkExec;
 use datafusion_datasource::source::DataSourceExec;
+use datafusion_execution::cache::cache_manager::CachedFileMetadataEntry;
 use datafusion_expr::dml::InsertOp;
 use datafusion_physical_expr::LexRequirement;
 use datafusion_physical_plan::ExecutionPlan;
@@ -251,16 +252,19 @@ impl FileFormat for VortexFormat {
                 let cache = file_metadata_cache.clone();
 
                 SpawnedTask::spawn(async move {
-                    // Check if we have cached metadata for this file
-                    if let Some(cached) = cache.get(&object)
-                        && let Some(cached_vortex) =
-                            cached.as_any().downcast_ref::<CachedVortexMetadata>()
+                    // Check if we have entry metadata for this file
+                    if let Some(entry) = cache.get(&object.location)
+                        && entry.is_valid_for(&object)
+                        && let Some(cached_vortex) = entry
+                            .file_metadata
+                            .as_any()
+                            .downcast_ref::<CachedVortexMetadata>()
                     {
                         let inferred_schema = cached_vortex.footer().dtype().to_arrow_schema()?;
                         return VortexResult::Ok((object.location, inferred_schema));
                     }
 
-                    // Not cached or invalid - open the file
+                    // Not entry or invalid - open the file
                     let reader = Arc::new(ObjectStoreReadAt::new(
                         store,
                         object.location.clone(),
@@ -276,7 +280,8 @@ impl FileFormat for VortexFormat {
 
                     // Cache the metadata
                     let cached_metadata = Arc::new(CachedVortexMetadata::new(&vxf));
-                    cache.put(&object, cached_metadata);
+                    let entry = CachedFileMetadataEntry::new(object.clone(), cached_metadata);
+                    cache.put(&object.location, entry);
 
                     let inferred_schema = vxf.dtype().to_arrow_schema()?;
                     VortexResult::Ok((object.location, inferred_schema))
@@ -310,24 +315,28 @@ impl FileFormat for VortexFormat {
         let file_metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();
 
         SpawnedTask::spawn(async move {
-            // Try to get cached metadata first
-            let cached_metadata = file_metadata_cache.get(&object).and_then(|cached| {
-                cached
-                    .as_any()
-                    .downcast_ref::<CachedVortexMetadata>()
-                    .map(|m| {
-                        (
-                            m.footer().dtype().clone(),
-                            m.footer().statistics().cloned(),
-                            m.footer().row_count(),
-                        )
-                    })
-            });
+            // Try to get entry metadata first
+            let cached_metadata = file_metadata_cache
+                .get(&object.location)
+                .filter(|entry| entry.is_valid_for(&object))
+                .and_then(|entry| {
+                    entry
+                        .file_metadata
+                        .as_any()
+                        .downcast_ref::<CachedVortexMetadata>()
+                        .map(|m| {
+                            (
+                                m.footer().dtype().clone(),
+                                m.footer().statistics().cloned(),
+                                m.footer().row_count(),
+                            )
+                        })
+                });
 
             let (dtype, file_stats, row_count) = match cached_metadata {
                 Some(metadata) => metadata,
                 None => {
-                    // Not cached - open the file
+                    // Not entry - open the file
                     let reader = Arc::new(ObjectStoreReadAt::new(
                         store,
                         object.location.clone(),
@@ -348,8 +357,9 @@ impl FileFormat for VortexFormat {
                         })?;
 
                     // Cache the metadata
-                    let cached = Arc::new(CachedVortexMetadata::new(&vxf));
-                    file_metadata_cache.put(&object, cached);
+                    let file_metadata = Arc::new(CachedVortexMetadata::new(&vxf));
+                    let entry = CachedFileMetadataEntry::new(object.clone(), file_metadata);
+                    file_metadata_cache.put(&object.location, entry);
 
                     (
                         vxf.dtype().clone(),
diff --git a/vortex-datafusion/src/persistent/opener.rs b/vortex-datafusion/src/persistent/opener.rs
@@ -185,8 +185,10 @@ impl FileOpener for VortexOpener {
                 .with_labels(labels);
 
             if let Some(file_metadata_cache) = file_metadata_cache
-                && let Some(file_metadata) = file_metadata_cache.get(&file.object_meta)
-                && let Some(vortex_metadata) = file_metadata
+                && let Some(entry) = file_metadata_cache.get(file.path())
+                && entry.is_valid_for(&file.object_meta)
+                && let Some(vortex_metadata) = entry
+                    .file_metadata
                     .as_any()
                     .downcast_ref::<CachedVortexMetadata>()
             {
@@ -216,7 +218,7 @@ impl FileOpener for VortexOpener {
             let expr_adapter = expr_adapter_factory.create(
                 Arc::clone(&unified_file_schema),
                 Arc::clone(&this_file_schema),
-            );
+            )?;
 
             let simplifier = PhysicalExprSimplifier::new(&this_file_schema);
 
diff --git a/vortex-io/src/object_store/read_at.rs b/vortex-io/src/object_store/read_at.rs
@@ -11,6 +11,7 @@ use object_store::GetOptions;
 use object_store::GetRange;
 use object_store::GetResultPayload;
 use object_store::ObjectStore;
+use object_store::ObjectStoreExt;
 use object_store::path::Path as ObjectPath;
 use vortex_array::buffer::BufferHandle;
 use vortex_buffer::Alignment;
diff --git a/vortex-io/src/object_store/write.rs b/vortex-io/src/object_store/write.rs
@@ -9,6 +9,7 @@ use futures::TryStreamExt;
 use futures::stream::FuturesUnordered;
 use object_store::MultipartUpload;
 use object_store::ObjectStore;
+use object_store::ObjectStoreExt;
 use object_store::PutPayload;
 use object_store::PutResult;
 use object_store::path::Path;
diff --git a/vortex-jni/src/file.rs b/vortex-jni/src/file.rs
@@ -15,6 +15,7 @@ use jni::objects::ReleaseMode;
 use jni::sys::jlong;
 use jni::sys::jobject;
 use object_store::ObjectStore;
+use object_store::ObjectStoreExt;
 use object_store::path::Path;
 use prost::Message;
 use url::Url;
diff --git a/vortex-sqllogictest/Cargo.toml b/vortex-sqllogictest/Cargo.toml
@@ -23,7 +23,7 @@ datafusion-sqllogictest = { workspace = true }
 futures.workspace = true
 indicatif.workspace = true
 rstest = { workspace = true }
-sqllogictest = "0.28"
+sqllogictest = "0.29.1"
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 vortex = { workspace = true, features = ["tokio"] }
