Upgrade datafusion to 30.0.0

Cargo.toml

@@ -24,25 +24,29 @@ members = [
     "ballista/executor",
     "ballista/scheduler",
     "benchmarks",
-    "examples",
+    "examples"
 ]
+resolver = "2"
 
 [workspace.dependencies]
-arrow = {version = "43.0.0"}
-arrow-flight = {version = "43.0.0", features = ["flight-sql-experimental"]}
-arrow-schema = {version = "43.0.0", default-features = false}
-configure_me = {version = "0.4.0"}
-configure_me_codegen = {version = "0.4.4"}
-datafusion = "28.0.0"
-datafusion-cli = "28.0.0"
-datafusion-proto = "28.0.0"
+arrow = { version = "45.0.0" }
+arrow-flight = { version = "45.0.0", features = ["flight-sql-experimental"] }
+arrow-schema = { version = "45.0.0", default-features = false }
+configure_me = { version = "0.4.0" }
+configure_me_codegen = { version = "0.4.4" }
+datafusion = "30.0.0"
+datafusion-cli = "30.0.0"
+datafusion-proto = "30.0.0"
 object_store = "0.6.1"
-sqlparser = "0.35.0"
-tonic = {version = "0.9"}
-tonic-build = {version = "0.9", default-features = false, features = ["transport", "prost"]}
+sqlparser = "0.36.1"
+tonic = { version = "0.9" }
+tonic-build = { version = "0.9", default-features = false, features = [
+    "transport",
+    "prost"
+] }
 tracing = "0.1.36"
 tracing-appender = "0.2.2"
-tracing-subscriber = {version = "0.3.15", features = ["env-filter"]}
+tracing-subscriber = { version = "0.3.15", features = ["env-filter"] }
 
 # cargo build --profile release-lto
 [profile.release-lto]

ballista/cache/src/backend/policy/lru/hashlink/linked_hash_map.rs

@@ -1977,10 +1977,7 @@ struct ValueLinks<K, V> {
 impl<K, V> Clone for ValueLinks<K, V> {
     #[inline]
     fn clone(&self) -> Self {
-        ValueLinks {
-            next: self.next,
-            prev: self.prev,
-        }
+        *self
     }
 }
 
@@ -1993,7 +1990,7 @@ struct FreeLink<K, V> {
 impl<K, V> Clone for FreeLink<K, V> {
     #[inline]
     fn clone(&self) -> Self {
-        FreeLink { next: self.next }
+        *self
     }
 }
 

ballista/client/src/context.rs

@@ -614,6 +614,8 @@ mod tests {
                         target_partitions: x.target_partitions,
                         file_sort_order: vec![],
                         infinite_source: false,
+                        insert_mode:
+                            datafusion::datasource::listing::ListingTableInsertMode::Error,
                     };
 
                     let table_paths = listing_table

ballista/core/src/plugin/mod.rs

@@ -40,6 +40,7 @@ pub trait Plugin {
 }
 
 /// The enum of Plugin
+#[repr(C)]
 #[derive(PartialEq, std::cmp::Eq, std::hash::Hash, Copy, Clone)]
 pub enum PluginEnum {
     /// UDF/UDAF plugin

ballista/core/src/utils.rs

@@ -36,7 +36,6 @@ use datafusion::logical_expr::{DdlStatement, LogicalPlan};
 use datafusion::physical_plan::aggregates::AggregateExec;
 use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
 use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
-use datafusion::physical_plan::common::batch_byte_size;
 use datafusion::physical_plan::empty::EmptyExec;
 use datafusion::physical_plan::filter::FilterExec;
 use datafusion::physical_plan::joins::HashJoinExec;
@@ -88,7 +87,7 @@ pub async fn write_stream_to_disk(
     while let Some(result) = stream.next().await {
         let batch = result?;
 
-        let batch_size_bytes: usize = batch_byte_size(&batch);
+        let batch_size_bytes: usize = batch.get_array_memory_size();
         num_batches += 1;
         num_rows += batch.num_rows();
         num_bytes += batch_size_bytes;

ballista/scheduler/src/flight_sql.rs

@@ -65,7 +65,6 @@ use datafusion::arrow::ipc::writer::{IpcDataGenerator, IpcWriteOptions};
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::common::DFSchemaRef;
 use datafusion::logical_expr::LogicalPlan;
-use datafusion::physical_plan::common::batch_byte_size;
 use datafusion::prelude::SessionContext;
 use datafusion_proto::protobuf::{LogicalPlanNode, PhysicalPlanNode};
 use prost::Message;
@@ -465,7 +464,7 @@ impl FlightSqlServiceImpl {
         data: &RecordBatch,
         name: &str,
     ) -> Result<Response<FlightInfo>, Status> {
-        let num_bytes = batch_byte_size(data) as i64;
+        let num_bytes = data.get_array_memory_size() as i64;
         let schema = data.schema();
         let num_rows = data.num_rows() as i64;
 

benchmarks/src/bin/tpch.rs

@@ -25,9 +25,8 @@ use ballista::prelude::{
 };
 use datafusion::arrow::array::*;
 use datafusion::arrow::util::display::array_value_to_string;
-use datafusion::datasource::file_format::csv::DEFAULT_CSV_EXTENSION;
-use datafusion::datasource::file_format::parquet::DEFAULT_PARQUET_EXTENSION;
-use datafusion::datasource::listing::ListingTableUrl;
+use datafusion::common::{DEFAULT_CSV_EXTENSION, DEFAULT_PARQUET_EXTENSION};
+use datafusion::datasource::listing::{ListingTableInsertMode, ListingTableUrl};
 use datafusion::datasource::{MemTable, TableProvider};
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::context::SessionState;
@@ -846,6 +845,7 @@ async fn get_table(
         table_partition_cols: vec![],
         file_sort_order: vec![],
         infinite_source: false,
+        insert_mode: ListingTableInsertMode::Error,
     };
 
     let url = ListingTableUrl::parse(path)?;