You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ag...@apache.org on 2023/12/12 14:56:57 UTC
(arrow-ballista) branch main updated: Set message sizes to 16MB in clients (#931)
This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new d0921c4d Set message sizes to 16MB in clients (#931)
d0921c4d is described below
commit d0921c4db82e2b9e57c662954b405c3b40f74f52
Author: Andy Grove <an...@gmail.com>
AuthorDate: Tue Dec 12 07:56:50 2023 -0700
Set message sizes to 16MB in clients (#931)
---
ballista/client/src/context.rs | 4 +++-
ballista/core/src/execution_plans/distributed_query.rs | 4 +++-
ballista/executor/src/executor_process.rs | 4 +++-
ballista/executor/src/executor_server.rs | 4 +++-
4 files changed, 12 insertions(+), 4 deletions(-)
diff --git a/ballista/client/src/context.rs b/ballista/client/src/context.rs
index 82ca1710..ea381559 100644
--- a/ballista/client/src/context.rs
+++ b/ballista/client/src/context.rs
@@ -100,7 +100,9 @@ impl BallistaContext {
let connection = create_grpc_client_connection(scheduler_url.clone())
.await
.map_err(|e| DataFusionError::Execution(format!("{e:?}")))?;
- let mut scheduler = SchedulerGrpcClient::new(connection);
+ let mut scheduler = SchedulerGrpcClient::new(connection)
+ .max_encoding_message_size(16 * 1024 * 1024)
+ .max_decoding_message_size(16 * 1024 * 1024);
let remote_session_id = scheduler
.create_session(CreateSessionParams {
diff --git a/ballista/core/src/execution_plans/distributed_query.rs b/ballista/core/src/execution_plans/distributed_query.rs
index 13511173..1818d03e 100644
--- a/ballista/core/src/execution_plans/distributed_query.rs
+++ b/ballista/core/src/execution_plans/distributed_query.rs
@@ -229,7 +229,9 @@ async fn execute_query(
.await
.map_err(|e| DataFusionError::Execution(format!("{e:?}")))?;
- let mut scheduler = SchedulerGrpcClient::new(connection);
+ let mut scheduler = SchedulerGrpcClient::new(connection)
+ .max_encoding_message_size(16 * 1024 * 1024)
+ .max_decoding_message_size(16 * 1024 * 1024);
let query_result = scheduler
.execute_query(query)
diff --git a/ballista/executor/src/executor_process.rs b/ballista/executor/src/executor_process.rs
index d3b56978..4e2f74c7 100644
--- a/ballista/executor/src/executor_process.rs
+++ b/ballista/executor/src/executor_process.rs
@@ -284,7 +284,9 @@ pub async fn start_executor_process(opt: Arc<ExecutorProcessConfig>) -> Result<(
}
}?;
- let mut scheduler = SchedulerGrpcClient::new(connection);
+ let mut scheduler = SchedulerGrpcClient::new(connection)
+ .max_encoding_message_size(16 * 1024 * 1024)
+ .max_decoding_message_size(16 * 1024 * 1024);
let default_codec: BallistaCodec<LogicalPlanNode, PhysicalPlanNode> =
BallistaCodec::default();
diff --git a/ballista/executor/src/executor_server.rs b/ballista/executor/src/executor_server.rs
index 90f888cc..dc5a08c9 100644
--- a/ballista/executor/src/executor_server.rs
+++ b/ballista/executor/src/executor_server.rs
@@ -237,7 +237,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> ExecutorServer<T,
} else {
let scheduler_url = format!("http://{scheduler_id}");
let connection = create_grpc_client_connection(scheduler_url).await?;
- let scheduler = SchedulerGrpcClient::new(connection);
+ let scheduler = SchedulerGrpcClient::new(connection)
+ .max_encoding_message_size(16 * 1024 * 1024)
+ .max_decoding_message_size(16 * 1024 * 1024);
{
self.schedulers