draft: Bump datafusion to 50 and arrow to 56

Signed-off-by: Adam Gutglick <[email protected]>

Author: AdamGS

Cargo.lock

@@ -59,16 +59,16 @@ anyhow = "1.0.95"
 arbitrary = "1.3.2"
 arcref = "0.2.0"
 arrayref = "0.3.7"
-arrow-arith = "55.2.0"
-arrow-array = "55.2.0"
-arrow-buffer = "55.2.0"
-arrow-cast = "55.2.0"
-arrow-data = "55.2.0"
-arrow-ipc = "55.2.0"
-arrow-ord = "55.2.0"
-arrow-schema = "55.2.0"
-arrow-select = "55.2.0"
-arrow-string = "55.2.0"
+arrow-arith = "56"
+arrow-array = "56"
+arrow-buffer = "56"
+arrow-cast = "56"
+arrow-data = "56"
+arrow-ipc = "56"
+arrow-ord = "56"
+arrow-schema = "56"
+arrow-select = "56"
+arrow-string = "56"
 async-compat = "0.2.5"
 async-stream = "0.3.6"
 async-trait = "0.1.88"
@@ -88,15 +88,17 @@ crossbeam-deque = "0.8.6"
 crossbeam-queue = "0.3.12"
 crossterm = "0.29"
 dashmap = "6.1.0"
-datafusion = { version = "49", default-features = false }
-datafusion-catalog = { version = "49" }
-datafusion-common = { version = "49" }
-datafusion-common-runtime = { version = "49" }
-datafusion-datasource = { version = "49", default-features = false }
-datafusion-execution = { version = "49" }
-datafusion-expr = { version = "49" }
-datafusion-physical-expr = { version = "49" }
-datafusion-physical-plan = { version = "49" }
+datafusion = { version = "50", default-features = false, git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-catalog = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-common = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-common-runtime = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-datasource = { version = "50", default-features = false, git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-execution = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-expr = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-physical-expr = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-physical-expr-adapter = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-physical-expr-common = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
+datafusion-physical-plan = { version = "50", git = "https://github.com/apache/datafusion.git", branch = "branch-50" }
 dirs = "6.0.0"
 divan = { package = "codspeed-divan-compat", version = "3.0" }
 dyn-hash = "0.2.0"
@@ -137,7 +139,7 @@ opentelemetry = "0.30.0"
 opentelemetry-otlp = "0.30.0"
 opentelemetry_sdk = "0.30.0"
 parking_lot = { version = "0.12.3", features = ["nightly"] }
-parquet = "55.2.0"
+parquet = "56"
 paste = "1.0.15"
 pco = "0.4.4"
 pin-project = "1.1.5"
@@ -186,9 +188,8 @@ thiserror = "2.0.3"
 tokio = { version = "1.46" }
 tokio-stream = "0.1.17"
 tokio-util = { version = "0.7.16" }
-# replace these with releases
-tpchgen = { git = "https://github.com/clflushopt/tpchgen-rs.git", rev = "d849ff430cd52250f6891ed4d5e3adad77bb2698" }
-tpchgen-arrow = { git = "https://github.com/clflushopt/tpchgen-rs.git", rev = "d849ff430cd52250f6891ed4d5e3adad77bb2698" }
+tpchgen = { version = "2" }
+tpchgen-arrow = { version = "2" }
 tracing = { version = "0.1.41" }
 tracing-perfetto = "0.1.5"
 tracing-subscriber = "0.3.20"

Cargo.toml

@@ -29,6 +29,7 @@ datafusion = { workspace = true, features = [
     "parquet",
     "datetime_expressions",
     "nested_expressions",
+    "unicode_expressions"
 ] }
 datafusion-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }

bench-vortex/Cargo.toml

@@ -24,6 +24,8 @@ datafusion-datasource = { workspace = true, default-features = false }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-physical-expr = { workspace = true }
+datafusion-physical-expr-adapter = { workspace = true }
+datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }

vortex-datafusion/Cargo.toml

@@ -6,6 +6,7 @@ use std::sync::Arc;
 use arrow_schema::{DataType, Schema};
 use datafusion_expr::Operator as DFOperator;
 use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef};
+use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_plan::expressions as df_expr;
 use vortex::error::{VortexResult, vortex_bail, vortex_err};
 use vortex::expr::{BinaryExpr, ExprRef, LikeExpr, Operator, and, get_item, lit, root};
@@ -122,6 +123,11 @@ impl TryFromDataFusion<DFOperator> for Operator {
 }
 
 pub(crate) fn can_be_pushed_down(expr: &PhysicalExprRef, schema: &Schema) -> bool {
+    // We currently do not support pushdown of dynamic expressions in DF
+    if is_dynamic_physical_expr(expr) {
+        return false;
+    }
+
     let expr = expr.as_any();
     if let Some(binary) = expr.downcast_ref::<df_expr::BinaryExpr>() {
         can_binary_be_pushed_down(binary, schema)

vortex-datafusion/src/convert/exprs.rs

@@ -10,9 +10,9 @@ use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_datasource::{FileRange, PartitionedFile};
-use datafusion_physical_expr::schema_rewriter::PhysicalExprAdapterFactory;
 use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
 use datafusion_physical_expr::{PhysicalExprRef, split_conjunction};
+use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
 use futures::{FutureExt, StreamExt, TryStreamExt, stream};
 use object_store::ObjectStore;
 use object_store::path::Path;
@@ -215,9 +215,7 @@ impl FileOpener for VortexOpener {
                     ))))
                 })
                 .try_flatten()
-                .map(move |batch| {
-                    batch.and_then(|b| schema_mapping.map_batch(b).map_err(Into::into))
-                })
+                .map(move |batch| batch.and_then(|b| schema_mapping.map_batch(b)))
                 .boxed();
 
             Ok(stream)
@@ -256,15 +254,14 @@ fn byte_range_to_row_range(byte_range: Range<u64>, row_count: u64, total_size: u
 #[cfg(test)]
 mod tests {
     use chrono::Utc;
-    use datafusion::arrow;
     use datafusion::arrow::array::RecordBatch;
     use datafusion::arrow::datatypes::{DataType, Schema};
     use datafusion::arrow::util::pretty::print_batches;
     use datafusion::common::record_batch;
     use datafusion::datasource::schema_adapter::DefaultSchemaAdapterFactory;
     use datafusion::logical_expr::{col, lit};
     use datafusion::physical_expr::planner::logical2physical;
-    use datafusion::physical_expr::schema_rewriter::DefaultPhysicalExprAdapterFactory;
+    use datafusion::physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
     use datafusion::scalar::ScalarValue;
     use futures::stream::BoxStream;
     use itertools::Itertools;
@@ -335,7 +332,7 @@ mod tests {
     }
 
     async fn count_data(
-        mut stream: BoxStream<'static, Result<RecordBatch, ArrowError>>,
+        mut stream: BoxStream<'static, Result<RecordBatch, DataFusionError>>,
     ) -> anyhow::Result<(usize, usize)> {
         let mut batches = vec![];
 

vortex-datafusion/src/persistent/opener.rs

@@ -12,10 +12,10 @@ use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::file_stream::FileOpener;
 use datafusion_datasource::schema_adapter::{DefaultSchemaAdapterFactory, SchemaAdapterFactory};
-use datafusion_physical_expr::schema_rewriter::{
+use datafusion_physical_expr::{PhysicalExprRef, conjunction};
+use datafusion_physical_expr_adapter::{
     DefaultPhysicalExprAdapterFactory, PhysicalExprAdapterFactory,
 };
-use datafusion_physical_expr::{PhysicalExprRef, conjunction};
 use datafusion_physical_plan::filter_pushdown::{
     FilterPushdownPropagation, PushedDown, PushedDownPredicate,
 };

vortex-datafusion/src/persistent/source.rs

@@ -15,9 +15,7 @@
 
 use std::sync::Arc;
 
-use arrow_schema::{
-    DECIMAL128_MAX_PRECISION, DataType, Field, FieldRef, Fields, Schema, SchemaBuilder, SchemaRef,
-};
+use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, SchemaBuilder, SchemaRef};
 use vortex_error::{VortexExpect, VortexResult, vortex_bail, vortex_err};
 
 use crate::datetime::arrow::{make_arrow_temporal_dtype, make_temporal_ext_dtype};
@@ -61,8 +59,11 @@ impl TryFromArrowType<&DataType> for PType {
 impl TryFromArrowType<&DataType> for DecimalDType {
     fn try_from_arrow(value: &DataType) -> VortexResult<Self> {
         match value {
-            DataType::Decimal128(precision, scale) => Self::try_new(*precision, *scale),
-            DataType::Decimal256(precision, scale) => Self::try_new(*precision, *scale),
+            DataType::Decimal32(precision, scale)
+            | DataType::Decimal64(precision, scale)
+            | DataType::Decimal128(precision, scale)
+            | DataType::Decimal256(precision, scale) => Self::try_new(*precision, *scale),
+
             _ => Err(vortex_err!(
                 "Arrow datatype {:?} cannot be converted to DecimalDType",
                 value
@@ -108,7 +109,10 @@ impl FromArrowType<(&DataType, Nullability)> for DType {
 
         match data_type {
             DataType::Null => DType::Null,
-            DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
+            DataType::Decimal32(precision, scale)
+            | DataType::Decimal64(precision, scale)
+            | DataType::Decimal128(precision, scale)
+            | DataType::Decimal256(precision, scale) => {
                 DType::Decimal(DecimalDType::new(*precision, *scale), nullability)
             }
             DataType::Boolean => DType::Bool(nullability),
@@ -188,10 +192,17 @@ impl DType {
                 PType::F64 => DataType::Float64,
             },
             DType::Decimal(dt, _) => {
-                if dt.precision() > DECIMAL128_MAX_PRECISION {
-                    DataType::Decimal256(dt.precision(), dt.scale())
-                } else {
-                    DataType::Decimal128(dt.precision(), dt.scale())
+                let precision = dt.precision();
+                let scale = dt.scale();
+                match dt.precision() {
+                    // DECIMAL32_MAX_PRECISION
+                    0..=9 => DataType::Decimal32(precision, scale),
+                    // DECIMAL64_MAX_PRECISION
+                    10..=18 => DataType::Decimal64(precision, scale),
+                    // DECIMAL128_MAX_PRECISION
+                    19..=38 => DataType::Decimal128(precision, scale),
+                    // DECIMAL256_MAX_PRECISION
+                    39.. => DataType::Decimal256(precision, scale),
                 }
             }
             DType::Utf8(_) => DataType::Utf8View,