You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by dh...@apache.org on 2023/11/28 14:35:03 UTC
(arrow-ballista) branch main updated: Dynamically optimize aggregate based on shuffle stats (#919)
This is an automated email from the ASF dual-hosted git repository.
dheres 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 e474e343 Dynamically optimize aggregate based on shuffle stats (#919)
e474e343 is described below
commit e474e343c683ffeedb6330ca284936cd25b1904b
Author: Daniël Heres <da...@gmail.com>
AuthorDate: Tue Nov 28 14:34:58 2023 +0000
Dynamically optimize aggregate based on shuffle stats (#919)
---
ballista/scheduler/src/state/execution_graph/execution_stage.rs | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/ballista/scheduler/src/state/execution_graph/execution_stage.rs b/ballista/scheduler/src/state/execution_graph/execution_stage.rs
index fcac54d5..f082fe43 100644
--- a/ballista/scheduler/src/state/execution_graph/execution_stage.rs
+++ b/ballista/scheduler/src/state/execution_graph/execution_stage.rs
@@ -22,6 +22,7 @@ use std::iter::FromIterator;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
+use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
use datafusion::physical_optimizer::join_selection::JoinSelection;
use datafusion::physical_optimizer::PhysicalOptimizerRule;
use datafusion::physical_plan::display::DisplayableExecutionPlan;
@@ -348,9 +349,13 @@ impl UnresolvedStage {
&input_locations,
)?;
- // Optimize join order based on new resolved statistics
+ // Optimize join order and statistics based on new resolved statistics
let optimize_join = JoinSelection::new();
- let plan = optimize_join.optimize(plan, SessionConfig::default().options())?;
+ let config = SessionConfig::default();
+ let plan = optimize_join.optimize(plan, config.options())?;
+ let optimize_aggregate = AggregateStatistics::new();
+ let plan =
+ optimize_aggregate.optimize(plan, SessionConfig::default().options())?;
Ok(ResolvedStage::new(
self.stage_id,