Upgrade DataFusion to 54 (#8044)

## Summary

This PR includes an upgrade of our DataFusion dependency/integration to
the upcoming 54 release. It aims to make the minimal amount of changes,
and implementing the new `Morselizer` API will be part of a future PR (I
have an old PR that was based on an earlier PoC, I'll try and pull stuff
from there when the time comes).

- Upstream release issue:
apache/datafusion#21080

Signed-off-by: Adam Gutglick <adam@spiraldb.com>

Author: AdamGS

Cargo.lock

@@ -99,17 +99,17 @@ anyhow = "1.0.97"
 arbitrary = "1.3.2"
 arc-swap = "1.9"
 arcref = "0.2.0"
-arrow-arith = "58"
-arrow-array = "58"
-arrow-buffer = "58"
-arrow-cast = "58"
-arrow-data = "58"
-arrow-ipc = "58"
-arrow-ord = "58"
-arrow-row = "58"
-arrow-schema = "58"
-arrow-select = "58"
-arrow-string = "58"
+arrow-arith = "58.3"
+arrow-array = "58.3"
+arrow-buffer = "58.3"
+arrow-cast = "58.3"
+arrow-data = "58.3"
+arrow-ipc = "58.3"
+arrow-ord = "58.3"
+arrow-row = "58.3"
+arrow-schema = "58.3"
+arrow-select = "58.3"
+arrow-string = "58.3"
 async-fs = "2.2.0"
 async-lock = "3.4"
 async-stream = "0.3.6"
@@ -137,20 +137,20 @@ cudarc = { version = "0.19.0", features = [
 ] }
 custom-labels = "0.4.4"
 dashmap = "6.1.0"
-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" }
+datafusion = { version = "54", default-features = false, features = ["sql"] }
+datafusion-catalog = { version = "54" }
+datafusion-common = { version = "54" }
+datafusion-common-runtime = { version = "54" }
+datafusion-datasource = { version = "54", default-features = false }
+datafusion-execution = { version = "54" }
+datafusion-expr = { version = "54" }
+datafusion-functions = { version = "54" }
+datafusion-physical-expr = { version = "54" }
+datafusion-physical-expr-adapter = { version = "54" }
+datafusion-physical-expr-common = { version = "54" }
+datafusion-physical-plan = { version = "54" }
+datafusion-pruning = { version = "54" }
+datafusion-sqllogictest = { version = "54" }
 dirs = "6.0.0"
 divan = { package = "codspeed-divan-compat", version = "4.0.4" }
 enum-iterator = "2.0.0"
@@ -164,7 +164,7 @@ get_dir = "0.5.0"
 glob = "0.3.2"
 goldenfile = "1"
 half = { version = "2.7.1", features = ["std", "num-traits"] }
-hashbrown = "0.17.0"
+hashbrown = "0.17.1"
 humansize = "2.1.3"
 indicatif = "0.18.0"
 insta = "1.43"
@@ -190,17 +190,17 @@ noodles-bgzf = "0.47.0"
 noodles-vcf = { version = "0.88.0", features = ["async"] }
 num-traits = "0.2.19"
 num_enum = { version = "0.7.3", default-features = false }
-object_store = { version = "0.13.1", default-features = false }
+object_store = { version = "0.13.2", default-features = false }
 once_cell = "1.21"
 oneshot = { version = "0.2.0", features = ["async"] }
 onpair = { version = "0.0.4" }
 opentelemetry = "0.32.0"
 opentelemetry-otlp = "0.32.0"
 opentelemetry_sdk = "0.32.0"
 parking_lot = { version = "0.12.3", features = ["nightly"] }
-parquet = "58"
-parquet-variant = "58"
-parquet-variant-compute = "58"
+parquet = "58.3"
+parquet-variant = "58.3"
+parquet-variant-compute = "58.3"
 paste = "1.0.15"
 pco = "1.0.1"
 pin-project-lite = "0.2.15"
@@ -253,7 +253,7 @@ tempfile = "3"
 termtree = { version = "1.0" }
 test-with = "0.16"
 thiserror = "2.0.3"
-tokio = { version = "1.48" }
+tokio = { version = "1.52" }
 tokio-stream = "0.1.17"
 tokio-util = "0.7.17"
 vortex-array-macros = { version = "0.1.0", path = "./vortex-array-macros" }
@@ -264,7 +264,7 @@ tracing = { version = "0.1.41", default-features = false }
 tracing-perfetto = "0.1.5"
 tracing-subscriber = "0.3"
 url = "2.5.7"
-uuid = { version = "1.21", features = ["js"] }
+uuid = { version = "1.23", features = ["js"] }
 wasm-bindgen-futures = "0.4.54"
 xshell = "0.2.6"
 zigzag = "0.1.0"

Cargo.toml

@@ -14,7 +14,7 @@ use datafusion::datasource::provider::DefaultTableFactory;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::execution::cache::DefaultListFilesCache;
 use datafusion::execution::cache::cache_manager::CacheManagerConfig;
-use datafusion::execution::cache::cache_unit::DefaultFileStatisticsCache;
+use datafusion::execution::cache::file_statistics_cache::DefaultFileStatisticsCache;
 use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::prelude::SessionConfig;
 use datafusion::prelude::SessionContext;
@@ -37,7 +37,7 @@ pub fn get_session_context() -> SessionContext {
     let file_static_cache = Arc::new(DefaultFileStatisticsCache::default());
     let list_file_cache = Arc::new(DefaultListFilesCache::default());
     let cache_config = CacheManagerConfig::default()
-        .with_files_statistics_cache(Some(file_static_cache))
+        .with_file_statistics_cache(Some(file_static_cache))
         .with_list_files_cache(Some(list_file_cache));
     rt_builder = rt_builder.with_cache_manager(cache_config);
 

benchmarks/datafusion-bench/src/lib.rs

@@ -278,7 +278,14 @@ async fn register_benchmark_tables<B: Benchmark + ?Sized>(
                     None => config.infer_schema(&session.state()).await?,
                 };
 
-                let listing_table = Arc::new(ListingTable::try_new(config)?);
+                let listing_table = Arc::new(
+                    ListingTable::try_new(config)?.with_cache(
+                        session
+                            .runtime_env()
+                            .cache_manager
+                            .get_file_statistic_cache(),
+                    ),
+                );
 
                 session.register_table(table.name, listing_table)?;
             }

benchmarks/datafusion-bench/src/main.rs

@@ -127,7 +127,6 @@ impl DefaultExpressionConvertor {
             let mut result = self.convert(source_expr.as_ref())?;
             for expr in field_names {
                 let field_name = expr
-                    .as_any()
                     .downcast_ref::<df_expr::Literal>()
                     .ok_or_else(|| exec_datafusion_err!("get_field field name must be a literal"))?
                     .value()
@@ -195,19 +194,19 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
     fn convert(&self, df: &dyn PhysicalExpr) -> DFResult<Expression> {
         // TODO(joe): Don't return an error when we have an unsupported node, bubble up "TRUE" as in keep
         //  for that node, up to any `and` or `or` node.
-        if let Some(binary_expr) = df.as_any().downcast_ref::<df_expr::BinaryExpr>() {
+        if let Some(binary_expr) = df.downcast_ref::<df_expr::BinaryExpr>() {
             let left = self.convert(binary_expr.left().as_ref())?;
             let right = self.convert(binary_expr.right().as_ref())?;
             let operator = try_operator_from_df(binary_expr.op())?;
 
             return Ok(Binary.new_expr(operator, [left, right]));
         }
 
-        if let Some(col_expr) = df.as_any().downcast_ref::<df_expr::Column>() {
+        if let Some(col_expr) = df.downcast_ref::<df_expr::Column>() {
             return Ok(get_item(col_expr.name().to_owned(), root()));
         }
 
-        if let Some(like) = df.as_any().downcast_ref::<df_expr::LikeExpr>() {
+        if let Some(like) = df.downcast_ref::<df_expr::LikeExpr>() {
             let child = self.convert(like.expr().as_ref())?;
             let pattern = self.convert(like.pattern().as_ref())?;
             return Ok(Like.new_expr(
@@ -219,42 +218,34 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
             ));
         }
 
-        if let Some(literal) = df.as_any().downcast_ref::<df_expr::Literal>() {
+        if let Some(literal) = df.downcast_ref::<df_expr::Literal>() {
             let value = Scalar::from_df(literal.value());
             return Ok(lit(value));
         }
 
-        if let Some(cast_expr) = df.as_any().downcast_ref::<df_expr::CastExpr>() {
-            let cast_dtype = DType::from_arrow((cast_expr.cast_type(), Nullability::Nullable));
+        if let Some(cast_expr) = df.downcast_ref::<df_expr::CastExpr>() {
+            let cast_dtype = DType::from_arrow(cast_expr.target_field().as_ref());
             let child = self.convert(cast_expr.expr().as_ref())?;
             return Ok(cast(child, cast_dtype));
         }
 
-        if let Some(cast_col_expr) = df.as_any().downcast_ref::<df_expr::CastColumnExpr>() {
-            let target = cast_col_expr.target_field();
-
-            let target_dtype = DType::from_arrow((target.data_type(), target.is_nullable().into()));
-            let child = self.convert(cast_col_expr.expr().as_ref())?;
-            return Ok(cast(child, target_dtype));
-        }
-
-        if let Some(is_null_expr) = df.as_any().downcast_ref::<df_expr::IsNullExpr>() {
+        if let Some(is_null_expr) = df.downcast_ref::<df_expr::IsNullExpr>() {
             let arg = self.convert(is_null_expr.arg().as_ref())?;
             return Ok(is_null(arg));
         }
 
-        if let Some(is_not_null_expr) = df.as_any().downcast_ref::<df_expr::IsNotNullExpr>() {
+        if let Some(is_not_null_expr) = df.downcast_ref::<df_expr::IsNotNullExpr>() {
             let arg = self.convert(is_not_null_expr.arg().as_ref())?;
             return Ok(is_not_null(arg));
         }
 
-        if let Some(in_list) = df.as_any().downcast_ref::<df_expr::InListExpr>() {
+        if let Some(in_list) = df.downcast_ref::<df_expr::InListExpr>() {
             let value = self.convert(in_list.expr().as_ref())?;
             let list_elements: Vec<_> = in_list
                 .list()
                 .iter()
                 .map(|e| {
-                    if let Some(lit) = e.as_any().downcast_ref::<df_expr::Literal>() {
+                    if let Some(lit) = e.downcast_ref::<df_expr::Literal>() {
                         Ok(Scalar::from_df(lit.value()))
                     } else {
                         Err(exec_datafusion_err!("Failed to cast sub-expression"))
@@ -272,11 +263,11 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
             return Ok(if in_list.negated() { not(expr) } else { expr });
         }
 
-        if let Some(scalar_fn) = df.as_any().downcast_ref::<ScalarFunctionExpr>() {
+        if let Some(scalar_fn) = df.downcast_ref::<ScalarFunctionExpr>() {
             return self.try_convert_scalar_function(scalar_fn);
         }
 
-        if let Some(case_expr) = df.as_any().downcast_ref::<df_expr::CaseExpr>() {
+        if let Some(case_expr) = df.downcast_ref::<df_expr::CaseExpr>() {
             return self.try_convert_case_expr(case_expr);
         }
 
@@ -297,7 +288,7 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
         for projection_expr in source_projection.iter() {
             let r = projection_expr.expr.apply(|node| {
                 // We only pull column children of scalar functions that we can't push into the scan.
-                if let Some(scalar_fn_expr) = node.as_any().downcast_ref::<ScalarFunctionExpr>()
+                if let Some(scalar_fn_expr) = node.downcast_ref::<ScalarFunctionExpr>()
                     && !can_scalar_fn_be_pushed_down(scalar_fn_expr)
                 {
                     scan_projection.extend(
@@ -312,7 +303,7 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
 
                 // DataFusion assumes different decimal types can be coerced.
                 // Vortex expects a perfect match so we don't push it down.
-                if let Some(binary_expr) = node.as_any().downcast_ref::<df_expr::BinaryExpr>()
+                if let Some(binary_expr) = node.downcast_ref::<df_expr::BinaryExpr>()
                     && binary_expr.op().is_numerical_operators()
                     && (is_decimal(&binary_expr.left().data_type(input_schema)?)
                         && is_decimal(&binary_expr.right().data_type(input_schema)?))
@@ -406,14 +397,13 @@ fn try_operator_from_df(value: &DFOperator) -> DFResult<Operator> {
     }
 }
 
-fn can_be_pushed_down_impl(df_expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
+fn can_be_pushed_down_impl(expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
     // We currently do not support pushdown of dynamic expressions in DF.
     // See issue: https://github.com/vortex-data/vortex/issues/4034
-    if is_dynamic_physical_expr(df_expr) {
+    if is_dynamic_physical_expr(expr) {
         return false;
     }
 
-    let expr = df_expr.as_any();
     if let Some(binary) = expr.downcast_ref::<df_expr::BinaryExpr>() {
         can_binary_be_pushed_down(binary, schema)
     } else if let Some(col) = expr.downcast_ref::<df_expr::Column>() {
@@ -429,9 +419,6 @@ fn can_be_pushed_down_impl(df_expr: &Arc<dyn PhysicalExpr>, schema: &Schema) ->
     } else if let Some(cast_expr) = expr.downcast_ref::<df_expr::CastExpr>() {
         // CastExpr child must be an expression type that convert() can handle
         is_convertible_expr(cast_expr.expr())
-    } else if let Some(cast_col_expr) = expr.downcast_ref::<df_expr::CastColumnExpr>() {
-        // CastColumnExpr child must be an expression type that convert() can handle
-        is_convertible_expr(cast_col_expr.expr())
     } else if let Some(is_null) = expr.downcast_ref::<df_expr::IsNullExpr>() {
         can_be_pushed_down_impl(is_null.arg(), schema)
     } else if let Some(is_not_null) = expr.downcast_ref::<df_expr::IsNotNullExpr>() {
@@ -447,17 +434,15 @@ fn can_be_pushed_down_impl(df_expr: &Arc<dyn PhysicalExpr>, schema: &Schema) ->
     } else if let Some(case_expr) = expr.downcast_ref::<df_expr::CaseExpr>() {
         can_case_be_pushed_down(case_expr, schema)
     } else {
-        tracing::debug!(%df_expr, "DataFusion expression can't be pushed down");
+        tracing::debug!(%expr, "DataFusion expression can't be pushed down");
         false
     }
 }
 
 /// Checks if an expression type is one that convert() can handle.
 /// This is less restrictive than can_be_pushed_down since it only checks
 /// expression types, not data type support.
-fn is_convertible_expr(df_expr: &Arc<dyn PhysicalExpr>) -> bool {
-    let expr = df_expr.as_any();
-
+fn is_convertible_expr(expr: &Arc<dyn PhysicalExpr>) -> bool {
     // Expression types that convert() handles
     expr.downcast_ref::<df_expr::BinaryExpr>().is_some()
         || expr.downcast_ref::<df_expr::Column>().is_some()
@@ -466,9 +451,6 @@ fn is_convertible_expr(df_expr: &Arc<dyn PhysicalExpr>) -> bool {
         || expr
             .downcast_ref::<df_expr::CastExpr>()
             .is_some_and(|e| is_convertible_expr(e.expr()))
-        || expr
-            .downcast_ref::<df_expr::CastColumnExpr>()
-            .is_some_and(|e| is_convertible_expr(e.expr()))
         || expr.downcast_ref::<df_expr::IsNullExpr>().is_some()
         || expr.downcast_ref::<df_expr::IsNotNullExpr>().is_some()
         || expr.downcast_ref::<df_expr::InListExpr>().is_some()
@@ -568,6 +550,8 @@ mod tests {
     use arrow_schema::Field;
     use arrow_schema::Schema;
     use arrow_schema::TimeUnit as ArrowTimeUnit;
+    use datafusion::arrow::array::AsArray;
+    use datafusion::arrow::datatypes::Int32Type;
     use datafusion_common::ScalarValue;
     use datafusion_expr::Operator as DFOperator;
     use datafusion_physical_expr::PhysicalExpr;
@@ -988,12 +972,7 @@ mod tests {
         let vortex_as_arrow = vortex_result.into_primitive().as_slice::<i32>().to_vec();
 
         // Convert DataFusion result to Vec for comparison
-        let df_as_arrow: Vec<i32> = df_array
-            .as_any()
-            .downcast_ref::<Int32Array>()
-            .unwrap()
-            .values()
-            .to_vec();
+        let df_as_arrow: Vec<i32> = df_array.as_primitive::<Int32Type>().values().to_vec();
 
         // Compare results
         // Expected: [0, 0, 50, 100, 100] for values [1, 5, 10, 15, 20]

vortex-datafusion/src/convert/exprs.rs

@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::any::Any;
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::sync::Arc;
@@ -299,10 +298,6 @@ impl FileFormatFactory for VortexFormatFactory {
     fn default(&self) -> Arc<dyn FileFormat> {
         Arc::new(VortexFormat::new(self.session.clone()))
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
 }
 
 impl VortexFormat {
@@ -330,10 +325,6 @@ impl VortexFormat {
 
 #[async_trait]
 impl FileFormat for VortexFormat {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn compression_type(&self) -> Option<FileCompressionType> {
         None
     }
@@ -594,7 +585,6 @@ impl FileFormat for VortexFormat {
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
         let mut source = file_scan_config
             .file_source()
-            .as_any()
             .downcast_ref::<VortexSource>()
             .cloned()
             .ok_or_else(|| internal_datafusion_err!("Expected VortexSource"))?;