chore: update datafusion to 49 (#1285)

milenkovicm · web-flow · commit cdb5159594b5 · 2025-08-18T10:40:01.000-06:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -35,10 +35,10 @@ arrow-flight = { version = "55", features = ["flight-sql-experimental"] }
 clap = { version = "4.5", features = ["derive", "cargo"] }
 configure_me = { version = "0.4.0" }
 configure_me_codegen = { version = "0.4.4" }
-datafusion = "48.0.1"
-datafusion-cli = "48.0.1"
-datafusion-proto = "48.0.1"
-datafusion-proto-common = "48.0.1"
+datafusion = "49.0.1"
+datafusion-cli = "49.0.1"
+datafusion-proto = "49.0.1"
+datafusion-proto-common = "49.0.1"
 object_store = "0.12"
 prost = "0.13"
 prost-types = "0.13"
diff --git a/ballista-cli/Cargo.toml b/ballista-cli/Cargo.toml
@@ -18,7 +18,7 @@
 [package]
 name = "ballista-cli"
 description = "Command Line Client for Ballista distributed query engine."
-version = "48.0.0"
+version = "49.0.0"
 authors = ["Apache DataFusion <dev@datafusion.apache.org>"]
 edition = { workspace = true }
 rust-version = { workspace = true }
@@ -29,7 +29,7 @@ repository = "https://github.com/apache/datafusion-ballista"
 readme = "README.md"
 
 [dependencies]
-ballista = { path = "../ballista/client", version = "48.0.0", features = ["standalone"] }
+ballista = { path = "../ballista/client", version = "49.0.0", features = ["standalone"] }
 clap = { workspace = true, features = ["derive", "cargo"] }
 datafusion = { workspace = true }
 datafusion-cli = { workspace = true }
diff --git a/ballista/client/Cargo.toml b/ballista/client/Cargo.toml
@@ -19,7 +19,7 @@
 name = "ballista"
 description = "Ballista Distributed Compute"
 license = "Apache-2.0"
-version = "48.0.0"
+version = "49.0.0"
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
 readme = "README.md"
@@ -29,18 +29,18 @@ rust-version = { workspace = true }
 
 [dependencies]
 async-trait = { workspace = true }
-ballista-core = { path = "../core", version = "48.0.0" }
-ballista-executor = { path = "../executor", version = "48.0.0", optional = true }
-ballista-scheduler = { path = "../scheduler", version = "48.0.0", optional = true }
+ballista-core = { path = "../core", version = "49.0.0" }
+ballista-executor = { path = "../executor", version = "49.0.0", optional = true }
+ballista-scheduler = { path = "../scheduler", version = "49.0.0", optional = true }
 datafusion = { workspace = true }
 log = { workspace = true }
 
 tokio = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]
-ballista-executor = { path = "../executor", version = "48.0.0" }
-ballista-scheduler = { path = "../scheduler", version = "48.0.0" }
+ballista-executor = { path = "../executor", version = "49.0.0" }
+ballista-scheduler = { path = "../scheduler", version = "49.0.0" }
 ctor = { workspace = true }
 datafusion-proto = { workspace = true }
 env_logger = { workspace = true }
diff --git a/ballista/client/tests/context_checks.rs b/ballista/client/tests/context_checks.rs
@@ -498,7 +498,7 @@ mod supported {
         ctx: SessionContext,
     ) -> datafusion::error::Result<()> {
         let result = ctx
-            .sql("select name, value from information_schema.df_settings where name like 'datafusion.execution.parquet.schema_force_view_types' or name like 'datafusion.sql_parser.map_varchar_to_utf8view' order by name limit 2")
+            .sql("select name, value from information_schema.df_settings where name like 'datafusion.execution.parquet.schema_force_view_types' or name like 'datafusion.sql_parser.map_string_types_to_utf8view' order by name limit 2")
             .await?
             .collect()
             .await?;
@@ -508,7 +508,7 @@ mod supported {
             "| name                                                 | value |",
             "+------------------------------------------------------+-------+",
             "| datafusion.execution.parquet.schema_force_view_types | false |",
-            "| datafusion.sql_parser.map_varchar_to_utf8view        | false |",
+            "| datafusion.sql_parser.map_string_types_to_utf8view   | false |",
             "+------------------------------------------------------+-------+",
         ];
 
diff --git a/ballista/core/Cargo.toml b/ballista/core/Cargo.toml
@@ -19,7 +19,7 @@
 name = "ballista-core"
 description = "Ballista Distributed Compute"
 license = "Apache-2.0"
-version = "48.0.0"
+version = "49.0.0"
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
 readme = "README.md"
diff --git a/ballista/core/proto/datafusion.proto b/ballista/core/proto/datafusion.proto
@@ -98,6 +98,7 @@ message ListingTableScanNode {
     datafusion_common.ParquetFormat parquet = 11;
     datafusion_common.AvroFormat avro = 12;
     datafusion_common.NdJsonFormat json = 15;
+    datafusion_common.ArrowFormat arrow = 16;
   }
   repeated SortExprNodeCollection file_sort_order = 13;
 }
@@ -243,7 +244,7 @@ message JoinNode {
   datafusion_common.JoinConstraint join_constraint = 4;
   repeated LogicalExprNode left_join_key = 5;
   repeated LogicalExprNode right_join_key = 6;
-  bool null_equals_null = 7;
+  datafusion_common.NullEquality null_equality = 7;
   LogicalExprNode filter = 8;
 }
 
@@ -726,6 +727,7 @@ message PhysicalPlanNode {
     ParquetSinkExecNode parquet_sink = 29;
     UnnestExecNode unnest = 30;
     JsonScanExecNode json_scan = 31;
+    CooperativeExecNode cooperative = 32;
   }
 }
 
@@ -858,6 +860,7 @@ message PhysicalScalarUdfNode {
   optional bytes fun_definition = 3;
   datafusion_common.ArrowType return_type = 4;
   bool nullable = 5;
+  string return_field_name = 6;
 }
 
 message PhysicalAggregateExprNode {
@@ -869,6 +872,7 @@ message PhysicalAggregateExprNode {
   bool distinct = 3;
   bool ignore_nulls = 6;
   optional bytes fun_definition = 7;
+  string human_display = 8;
 }
 
 message PhysicalWindowExprNode {
@@ -1033,6 +1037,10 @@ message AvroScanExecNode {
   FileScanExecConf base_conf = 1;
 }
 
+message CooperativeExecNode {
+  PhysicalPlanNode input = 1;
+}
+
 enum PartitionMode {
   COLLECT_LEFT = 0;
   PARTITIONED = 1;
@@ -1045,7 +1053,7 @@ message HashJoinExecNode {
   repeated JoinOn on = 3;
   datafusion_common.JoinType join_type = 4;
   PartitionMode partition_mode = 6;
-  bool null_equals_null = 7;
+  datafusion_common.NullEquality null_equality = 7;
   JoinFilter filter = 8;
   repeated uint32 projection = 9;
 }
@@ -1061,7 +1069,7 @@ message SymmetricHashJoinExecNode {
   repeated JoinOn on = 3;
   datafusion_common.JoinType join_type = 4;
   StreamPartitionMode partition_mode = 6;
-  bool null_equals_null = 7;
+  datafusion_common.NullEquality null_equality = 7;
   JoinFilter filter = 8;
   repeated PhysicalSortExprNode left_sort_exprs = 9;
   repeated PhysicalSortExprNode right_sort_exprs = 10;
diff --git a/ballista/core/proto/datafusion_common.proto b/ballista/core/proto/datafusion_common.proto
@@ -55,6 +55,8 @@ message NdJsonFormat {
   JsonOptions options = 1;
 }
 
+message ArrowFormat {}
+
 
 message PrimaryKeyConstraint{
   repeated uint64 indices = 1;
@@ -85,13 +87,19 @@ enum JoinType {
   RIGHTSEMI = 6;
   RIGHTANTI = 7;
   LEFTMARK = 8;
+  RIGHTMARK = 9;
 }
 
 enum JoinConstraint {
   ON = 0;
   USING = 1;
 }
 
+enum NullEquality {
+  NULL_EQUALS_NOTHING = 0;
+  NULL_EQUALS_NULL = 1;
+}
+
 message AvroOptions {}
 message ArrowOptions {}
 
diff --git a/ballista/core/src/client.rs b/ballista/core/src/client.rs
@@ -232,7 +232,7 @@ impl Stream for FlightDataStream {
                             self.schema.clone(),
                             &self.dictionaries_by_id,
                         )
-                        .map_err(|e| DataFusionError::ArrowError(e, None))
+                        .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
                     });
                 Some(converted_chunk)
             }
diff --git a/ballista/core/src/error.rs b/ballista/core/src/error.rs
@@ -38,7 +38,7 @@ pub enum BallistaError {
     General(String),
     Internal(String),
     Configuration(String),
-    ArrowError(ArrowError),
+    ArrowError(Box<ArrowError>),
     DataFusionError(Box<DataFusionError>),
     SqlError(parser::ParserError),
     IoError(io::Error),
@@ -84,7 +84,7 @@ impl From<ArrowError> for BallistaError {
                     *e.downcast::<DataFusionError>().unwrap(),
                 ))
             }
-            other => BallistaError::ArrowError(other),
+            other => BallistaError::ArrowError(Box::new(other)),
         }
     }
 }
@@ -98,7 +98,7 @@ impl From<parser::ParserError> for BallistaError {
 impl From<DataFusionError> for BallistaError {
     fn from(e: DataFusionError) -> Self {
         match e {
-            DataFusionError::ArrowError(e, _) => Self::from(e),
+            DataFusionError::ArrowError(e, _) => Self::from(*e),
             _ => BallistaError::DataFusionError(Box::new(e)),
         }
     }
diff --git a/ballista/core/src/extension.rs b/ballista/core/src/extension.rs
@@ -395,7 +395,7 @@ impl SessionConfigHelperExt for SessionConfig {
                 false,
             )
             // same like previous comment
-            .set_bool("datafusion.sql_parser.map_varchar_to_utf8view", false)
+            .set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
     }
 }
 
diff --git a/ballista/core/src/serde/mod.rs b/ballista/core/src/serde/mod.rs
@@ -518,6 +518,7 @@ mod test {
             partition_by: vec![],
             file_type,
             options: Default::default(),
+            output_schema: Arc::new(DFSchema::empty()),
         });
 
         let codec = crate::serde::BallistaLogicalExtensionCodec::default();
diff --git a/ballista/core/src/serde/scheduler/to_proto.rs b/ballista/core/src/serde/scheduler/to_proto.rs
@@ -21,7 +21,7 @@ use std::convert::TryInto;
 
 use crate::error::BallistaError;
 
-use crate::serde::protobuf;
+use crate::serde::protobuf::{self};
 use datafusion_proto::protobuf as datafusion_protobuf;
 
 use crate::serde::scheduler::{
@@ -177,6 +177,11 @@ impl TryInto<protobuf::OperatorMetric> for &MetricValue {
                         .unwrap_or(0),
                 )),
             }),
+            // at the moment there there is no way to serialize custom metrics
+            // thus at the moment we can't support it
+            MetricValue::Custom { .. } => Err(BallistaError::General(String::from(
+                "Custom metrics values are not supported",
+            ))),
         }
     }
 }
diff --git a/ballista/executor/Cargo.toml b/ballista/executor/Cargo.toml
@@ -19,7 +19,7 @@
 name = "ballista-executor"
 description = "Ballista Distributed Compute - Executor"
 license = "Apache-2.0"
-version = "48.0.0"
+version = "49.0.0"
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
 readme = "README.md"
@@ -40,8 +40,9 @@ default = ["build-binary", "mimalloc"]
 arrow = { workspace = true }
 arrow-flight = { workspace = true }
 async-trait = { workspace = true }
-ballista-core = { path = "../core", version = "48.0.0" }
+ballista-core = { path = "../core", version = "49.0.0" }
 clap = { workspace = true, optional = true }
+configure_me = { workspace = true, optional = true }
 dashmap = { workspace = true }
 datafusion = { workspace = true }
 datafusion-proto = { workspace = true }
diff --git a/ballista/scheduler/Cargo.toml b/ballista/scheduler/Cargo.toml
@@ -19,7 +19,7 @@
 name = "ballista-scheduler"
 description = "Ballista Distributed Compute - Scheduler"
 license = "Apache-2.0"
-version = "48.0.0"
+version = "49.0.0"
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
 readme = "README.md"
@@ -47,7 +47,7 @@ rest-api = []
 arrow-flight = { workspace = true }
 async-trait = { workspace = true }
 axum = "0.7.7"
-ballista-core = { path = "../core", version = "48.0.0" }
+ballista-core = { path = "../core", version = "49.0.0" }
 clap = { workspace = true, optional = true }
 dashmap = { workspace = true }
 datafusion = { workspace = true }
diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml
@@ -18,9 +18,8 @@
 [package]
 name = "ballista-benchmarks"
 description = "Ballista Benchmarks"
-version = "48.0.0"
-edition = { workspace = true }
-rust-version = { workspace = true }
+version = "49.0.0"
+edition = "2021"
 authors = ["Apache DataFusion <dev@datafusion.apache.org>"]
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
@@ -33,7 +32,7 @@ default = ["mimalloc"]
 snmalloc = ["snmalloc-rs"]
 
 [dependencies]
-ballista = { path = "../ballista/client", version = "48.0.0" }
+ballista = { path = "../ballista/client", version = "49.0.0" }
 datafusion = { workspace = true }
 datafusion-proto = { workspace = true }
 env_logger = { workspace = true }
@@ -52,4 +51,4 @@ tokio = { version = "^1.44", features = [
 ] }
 
 [dev-dependencies]
-ballista-core = { path = "../ballista/core", version = "48.0.0" }
+ballista-core = { path = "../ballista/core", version = "49.0.0" }
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -18,7 +18,7 @@
 [package]
 name = "ballista-examples"
 description = "Ballista usage examples"
-version = "48.0.0"
+version = "49.0.0"
 homepage = "https://datafusion.apache.org/ballista/"
 repository = "https://github.com/apache/datafusion-ballista"
 authors = ["Apache DataFusion <dev@datafusion.apache.org>"]
@@ -34,10 +34,10 @@ path = "examples/standalone-sql.rs"
 required-features = ["ballista/standalone"]
 
 [dependencies]
-ballista = { path = "../ballista/client", version = "48.0.0" }
-ballista-core = { path = "../ballista/core", version = "48.0.0", default-features = false }
-ballista-executor = { path = "../ballista/executor", version = "48.0.0", default-features = false }
-ballista-scheduler = { path = "../ballista/scheduler", version = "48.0.0", default-features = false }
+ballista = { path = "../ballista/client", version = "49.0.0" }
+ballista-core = { path = "../ballista/core", version = "49.0.0", default-features = false }
+ballista-executor = { path = "../ballista/executor", version = "49.0.0", default-features = false }
+ballista-scheduler = { path = "../ballista/scheduler", version = "49.0.0", default-features = false }
 datafusion = { workspace = true }
 env_logger = { workspace = true }
 futures = { workspace = true }

Original file line number	Diff line number	Diff line change
`@@ -232,7 +232,7 @@ impl Stream for FlightDataStream {`
`232`	`232`	`self.schema.clone(),`
`233`	`233`	`&self.dictionaries_by_id,`
`234`	`234`	`)`
`235`		`- .map_err(\|e\| DataFusionError::ArrowError(e, None))`
	`235`	`+ .map_err(\|e\| DataFusionError::ArrowError(Box::new(e), None))`
`236`	`236`	`});`
`237`	`237`	`Some(converted_chunk)`
`238`	`238`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ pub enum BallistaError {`
`38`	`38`	`General(String),`
`39`	`39`	`Internal(String),`
`40`	`40`	`Configuration(String),`
`41`		`- ArrowError(ArrowError),`
	`41`	`+ ArrowError(Box<ArrowError>),`
`42`	`42`	`DataFusionError(Box<DataFusionError>),`
`43`	`43`	`SqlError(parser::ParserError),`
`44`	`44`	`IoError(io::Error),`
`@@ -84,7 +84,7 @@ impl From<ArrowError> for BallistaError {`
`84`	`84`	`*e.downcast::<DataFusionError>().unwrap(),`
`85`	`85`	`))`
`86`	`86`	`}`
`87`		`- other => BallistaError::ArrowError(other),`
	`87`	`+ other => BallistaError::ArrowError(Box::new(other)),`
`88`	`88`	`}`
`89`	`89`	`}`
`90`	`90`	`}`
`@@ -98,7 +98,7 @@ impl From<parser::ParserError> for BallistaError {`
`98`	`98`	`impl From<DataFusionError> for BallistaError {`
`99`	`99`	`fn from(e: DataFusionError) -> Self {`
`100`	`100`	`match e {`
`101`		`- DataFusionError::ArrowError(e, _) => Self::from(e),`
	`101`	`+ DataFusionError::ArrowError(e, _) => Self::from(*e),`
`102`	`102`	`_ => BallistaError::DataFusionError(Box::new(e)),`
`103`	`103`	`}`
`104`	`104`	`}`