Update DataFusion to latest

Cargo.toml

@@ -29,18 +29,18 @@ members = [
 resolver = "2"
 
 [workspace.dependencies]
-arrow = { version = "46.0.0" }
-arrow-flight = { version = "46.0.0", features = ["flight-sql-experimental"] }
-arrow-schema = { version = "46.0.0", default-features = false }
+arrow = { version = "47.0.0" }
+arrow-flight = { version = "47.0.0", features = ["flight-sql-experimental"] }
+arrow-schema = { version = "47.0.0", default-features = false }
 configure_me = { version = "0.4.0" }
 configure_me_codegen = { version = "0.4.4" }
-datafusion = "31.0.0"
-datafusion-cli = "31.0.0"
-datafusion-proto = "31.0.0"
+datafusion = { git = "https://github.com/apache/arrow-datafusion.git", rev="1cf808d0e081221cc31b5f92fc414de7c76cf5c4" }
+datafusion-cli = { git = "https://github.com/apache/arrow-datafusion.git", rev="1cf808d0e081221cc31b5f92fc414de7c76cf5c4" }
+datafusion-proto = { git = "https://github.com/apache/arrow-datafusion.git", rev="1cf808d0e081221cc31b5f92fc414de7c76cf5c4" }
 object_store = "0.7.0"
-sqlparser = "0.37.0"
-tonic = { version = "0.9" }
-tonic-build = { version = "0.9", default-features = false, features = [
+sqlparser = "0.38.0"
+tonic = { version = "0.10" }
+tonic-build = { version = "0.10", default-features = false, features = [
     "transport",
     "prost"
 ] }

ballista-cli/src/main.rs

@@ -23,6 +23,7 @@ use ballista_cli::{
     exec, print_format::PrintFormat, print_options::PrintOptions, BALLISTA_CLI_VERSION,
 };
 use clap::Parser;
+use datafusion_cli::print_options::MaxRows;
 use mimalloc::MiMalloc;
 
 #[global_allocator]
@@ -82,6 +83,13 @@ struct Args {
     #[clap(long, help = "Ballista scheduler port")]
     port: Option<u16>,
 
+    #[clap(
+        long,
+        help = "The max number of rows to display for 'Table' format\n[default: 40] [possible values: numbers(0/10/...), inf(no limit)]",
+        default_value = "40"
+    )]
+    maxrows: MaxRows,
+
     #[clap(
         short,
         long,
@@ -133,6 +141,7 @@ pub async fn main() -> Result<()> {
     let mut print_options = PrintOptions {
         format: args.format,
         quiet: args.quiet,
+        maxrows: args.maxrows,
     };
 
     let files = args.file;

ballista/cache/Cargo.toml

@@ -23,7 +23,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-trait = "0.1.64"
+async-trait = "0.1.73"
 futures = "0.3"
 hashbrown = "0.14"
 log = "0.4"

ballista/client/src/context.rs

@@ -357,7 +357,7 @@ impl BallistaContext {
         // the show tables、 show columns sql can not run at scheduler because the tables is store at client
         if is_show {
             let state = self.state.lock();
-            ctx = Arc::new(SessionContext::with_config(
+            ctx = Arc::new(SessionContext::new_with_config(
                 SessionConfig::new().with_information_schema(
                     state.config.default_with_information_schema(),
                 ),

ballista/core/Cargo.toml

@@ -46,7 +46,7 @@ simd = ["datafusion/simd"]
 [dependencies]
 ahash = { version = "0.8", default-features = false }
 arrow-flight = { workspace = true }
-async-trait = "0.1.41"
+async-trait = "0.1.73"
 ballista-cache = { path = "../cache", version = "0.11.0" }
 bytes = "1.0"
 chrono = { version = "0.4", default-features = false }
@@ -66,7 +66,7 @@ once_cell = "1.9.0"
 
 parking_lot = "0.12"
 parse_arg = "0.1.3"
-prost = "0.11"
+prost = "0.12"
 prost-types = "0.11"
 rand = "0.8"
 serde = { version = "1", features = ["derive"] }

ballista/core/src/serde/generated/ballista.rs

@@ -2025,7 +2025,9 @@ pub mod scheduler_grpc_server {
                             request: tonic::Request<super::PollWorkParams>,
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
-                            let fut = async move { (*inner).poll_work(request).await };
+                            let fut = async move {
+                                <T as SchedulerGrpc>::poll_work(&inner, request).await
+                            };
                             Box::pin(fut)
                         }
                     }
@@ -2070,7 +2072,8 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).register_executor(request).await
+                                <T as SchedulerGrpc>::register_executor(&inner, request)
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2116,7 +2119,11 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).heart_beat_from_executor(request).await
+                                <T as SchedulerGrpc>::heart_beat_from_executor(
+                                        &inner,
+                                        request,
+                                    )
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2162,7 +2169,8 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).update_task_status(request).await
+                                <T as SchedulerGrpc>::update_task_status(&inner, request)
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2208,7 +2216,8 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).get_file_metadata(request).await
+                                <T as SchedulerGrpc>::get_file_metadata(&inner, request)
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2254,7 +2263,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).create_session(request).await
+                                <T as SchedulerGrpc>::create_session(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2300,7 +2309,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).update_session(request).await
+                                <T as SchedulerGrpc>::update_session(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2346,7 +2355,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).remove_session(request).await
+                                <T as SchedulerGrpc>::remove_session(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2392,7 +2401,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).execute_query(request).await
+                                <T as SchedulerGrpc>::execute_query(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2438,7 +2447,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).get_job_status(request).await
+                                <T as SchedulerGrpc>::get_job_status(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2484,7 +2493,8 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).executor_stopped(request).await
+                                <T as SchedulerGrpc>::executor_stopped(&inner, request)
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2529,7 +2539,9 @@ pub mod scheduler_grpc_server {
                             request: tonic::Request<super::CancelJobParams>,
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
-                            let fut = async move { (*inner).cancel_job(request).await };
+                            let fut = async move {
+                                <T as SchedulerGrpc>::cancel_job(&inner, request).await
+                            };
                             Box::pin(fut)
                         }
                     }
@@ -2574,7 +2586,7 @@ pub mod scheduler_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).clean_job_data(request).await
+                                <T as SchedulerGrpc>::clean_job_data(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2782,7 +2794,9 @@ pub mod executor_grpc_server {
                             request: tonic::Request<super::LaunchTaskParams>,
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
-                            let fut = async move { (*inner).launch_task(request).await };
+                            let fut = async move {
+                                <T as ExecutorGrpc>::launch_task(&inner, request).await
+                            };
                             Box::pin(fut)
                         }
                     }
@@ -2827,7 +2841,8 @@ pub mod executor_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).launch_multi_task(request).await
+                                <T as ExecutorGrpc>::launch_multi_task(&inner, request)
+                                    .await
                             };
                             Box::pin(fut)
                         }
@@ -2873,7 +2888,7 @@ pub mod executor_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).stop_executor(request).await
+                                <T as ExecutorGrpc>::stop_executor(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2919,7 +2934,7 @@ pub mod executor_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).cancel_tasks(request).await
+                                <T as ExecutorGrpc>::cancel_tasks(&inner, request).await
                             };
                             Box::pin(fut)
                         }
@@ -2965,7 +2980,7 @@ pub mod executor_grpc_server {
                         ) -> Self::Future {
                             let inner = Arc::clone(&self.0);
                             let fut = async move {
-                                (*inner).remove_job_data(request).await
+                                <T as ExecutorGrpc>::remove_job_data(&inner, request).await
                             };
                             Box::pin(fut)
                         }

ballista/core/src/serde/scheduler/to_proto.rs

@@ -158,12 +158,18 @@ impl TryInto<protobuf::OperatorMetric> for &MetricValue {
             }),
             MetricValue::StartTimestamp(timestamp) => Ok(protobuf::OperatorMetric {
                 metric: Some(operator_metric::Metric::StartTimestamp(
-                    timestamp.value().map(|m| m.timestamp_nanos()).unwrap_or(0),
+                    timestamp
+                        .value()
+                        .map(|m| m.timestamp_nanos_opt().unwrap_or(0))
+                        .unwrap_or(0),
                 )),
             }),
             MetricValue::EndTimestamp(timestamp) => Ok(protobuf::OperatorMetric {
                 metric: Some(operator_metric::Metric::EndTimestamp(
-                    timestamp.value().map(|m| m.timestamp_nanos()).unwrap_or(0),
+                    timestamp
+                        .value()
+                        .map(|m| m.timestamp_nanos_opt().unwrap_or(0))
+                        .unwrap_or(0),
                 )),
             }),
         }

ballista/core/src/utils.rs

@@ -59,7 +59,7 @@ use tonic::transport::{Channel, Error, Server};
 
 /// Default session builder using the provided configuration
 pub fn default_session_builder(config: SessionConfig) -> SessionState {
-    SessionState::with_config_rt(
+    SessionState::new_with_config_rt(
         config,
         Arc::new(
             RuntimeEnv::new(with_object_store_registry(RuntimeConfig::default()))
@@ -244,7 +244,7 @@ pub fn create_df_ctx_with_ballista_query_planner<T: 'static + AsLogicalPlan>(
     let session_config = SessionConfig::new()
         .with_target_partitions(config.default_shuffle_partitions())
         .with_information_schema(true);
-    let mut session_state = SessionState::with_config_rt(
+    let mut session_state = SessionState::new_with_config_rt(
         session_config,
         Arc::new(
             RuntimeEnv::new(with_object_store_registry(RuntimeConfig::default()))
@@ -254,7 +254,7 @@ pub fn create_df_ctx_with_ballista_query_planner<T: 'static + AsLogicalPlan>(
     .with_query_planner(planner);
     session_state = session_state.with_session_id(session_id);
     // the SessionContext created here is the client side context, but the session_id is from server side.
-    SessionContext::with_state(session_state)
+    SessionContext::new_with_state(session_state)
 }
 
 pub struct BallistaQueryPlanner<T: AsLogicalPlan> {

ballista/executor/Cargo.toml

@@ -40,7 +40,7 @@ default = ["mimalloc"]
 anyhow = "1"
 arrow = { workspace = true }
 arrow-flight = { workspace = true }
-async-trait = "0.1.41"
+async-trait = "0.1.73"
 ballista-core = { path = "../core", version = "0.11.0", features = ["s3"] }
 chrono = { version = "0.4", default-features = false }
 configure_me = { workspace = true }

ballista/executor/src/executor_process.rs

@@ -25,6 +25,7 @@ use std::{env, io};
 
 use anyhow::{Context, Result};
 use arrow_flight::flight_service_server::FlightServiceServer;
+use datafusion::execution::cache::cache_manager::CacheManager;
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
 use log::{error, info, warn};
@@ -199,6 +200,7 @@ pub async fn start_executor_process(opt: Arc<ExecutorProcessConfig>) -> Result<(
         let cache_dir = opt.cache_dir.clone();
         let cache_capacity = opt.cache_capacity;
         let cache_io_concurrency = opt.cache_io_concurrency;
+        let cache_manager = CacheManager::default();
         let cache_layer =
             opt.data_cache_policy
                 .map(|data_cache_policy| match data_cache_policy {
@@ -221,6 +223,7 @@ pub async fn start_executor_process(opt: Arc<ExecutorProcessConfig>) -> Result<(
                 memory_pool: runtime.memory_pool.clone(),
                 disk_manager: runtime.disk_manager.clone(),
                 object_store_registry: registry,
+                cache_manager: Arc::new(cache_manager),
             }))
         } else {
             None

ballista/scheduler/Cargo.toml

@@ -45,9 +45,9 @@ sled = ["sled_package", "tokio-stream"]
 anyhow = "1"
 arrow-flight = { workspace = true }
 async-recursion = "1.0.0"
-async-trait = "0.1.41"
+async-trait = "0.1.73"
 ballista-core = { path = "../core", version = "0.11.0", features = ["s3"] }
-base64 = { version = "0.13", default-features = false }
+base64 = { version = "0.21", default-features = false }
 clap = { version = "3", features = ["derive", "cargo"] }
 configure_me = { workspace = true }
 dashmap = "5.4.0"
@@ -67,7 +67,7 @@ once_cell = { version = "1.16.0", optional = true }
 parking_lot = "0.12"
 parse_arg = "0.1.3"
 prometheus = { version = "0.13", features = ["process"], optional = true }
-prost = "0.11"
+prost = "0.12"
 prost-types = { version = "0.11.0" }
 rand = "0.8"
 serde = { version = "1", features = ["derive"] }