You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "milenkovicm (via GitHub)" <gi...@apache.org> on 2026/05/23 20:22:30 UTC
[PR] WIP feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
milenkovicm opened a new pull request, #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752
## Overview
This commit implements **dynamic join strategy selection** in Ballista's Adaptive Query Execution (AQE) pipeline. Instead of choosing a join strategy at planning time (when data sizes are unknown), the plan now inserts a placeholder `DynamicJoinSelectionExec` node that defers the decision until runtime statistics are available.
```text
Before (planning time)
HashJoinExec / SortMergeJoinExec
↑ chosen by DataFusion optimizer (no runtime stats)
After (AQE flow)
Step 1 – plan creation:
DelayJoinSelectionRule → DynamicJoinSelectionExec (placeholder)
Step 2 – stage completes, stats available:
SelectJoinRule → CollectLeft | LateCollectLeft | Hash | Sort | Repartition
↑ decided using actual byte / row counts
```
The change means Ballista can now **broadcast small tables** or **swap join sides** based on real data sizes observed during execution, not just estimates from the logical plan.
Introduced planner rules consider few configuration values to decide join type:
- `datafusion.optimizer.hash_join_single_partition_threshold_rows`, to select broadcast join or not
- `datafusion.optimizer.hash_join_single_partition_threshold`, to select broadcast join or not
- `datafusion.optimizer.prefer_hash_join` to select actual preferred join implementation
with dynamic join selection some TPCH jobs have less stages created
<img width="1697" height="634" alt="Screenshot 2026-05-23 at 20 51 03" src="https://github.com/user-attachments/assets/bc9ff52b-3ec2-4879-a223-8ef13b99d379" />
compared to static scheduling
<img width="1707" height="630" alt="Screenshot 2026-05-23 at 20 55 20" src="https://github.com/user-attachments/assets/dfb037ee-e3b7-4bfe-a077-e97832f7c4d3" />
## TODO
- [ ] configuration to disable rule `ballista.planner.adaptive.join`
- [ ] test with `sort join`
- [ ] broadcast join gives bad results for TPCH 2, 13, 16
- [ ] decision should we stick with `ballista.optimizer.broadcast_join_threshold_bytes` or go with datafusion configuration values
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4601028944
thanks @metegenez and @martin-g for reviews, will merge this and will have few follow up on this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4586103950
Thanks @metegenez for your review. Will include your suggestions
I've found another issue running TPCDS Q72 when this rule makes join single partitioned, will address that first
I guess as a follow up it needs to get filter propagation, it might help out
One big problem I have with AQE is understanding why plan changes, there is need for plan flight recorder or something like that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [WIP] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3300885371
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,395 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ todo!(
+ "this should not be executed, it should have been replaced with optimizer rule"
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (ChildrenState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::LateCollectLeft),
+ //(ChildrenState::Repartitioned, PartitionMode::CollectLeft) =>
+ // self
+ // .to_hash_join(PartitionMode::Partitioned)
+ // .map(JoinSelectionAction::Hash),
+ (ChildrenState::Unknown, PartitionMode::Partitioned) => Ok(
+ JoinSelectionAction::Repartition(Arc::new(self.to_partitioned())),
+ ),
+ //
+ //
+ (_, PartitionMode::Auto) => internal_err!("this case should not be possible"),
+ }
+ }
+
+ pub(crate) fn to_hash_join(
+ &self,
+ partition_mode: PartitionMode,
+ ) -> Result<Arc<HashJoinExec>> {
+ let hash_join_exec = HashJoinExecBuilder::new(
+ self.left.clone(),
+ self.right.clone(),
+ self.on.clone(),
+ self.join_type,
+ )
+ .with_filter(self.filter.clone())
+ .with_projection(self.projection.clone())
+ .with_partition_mode(partition_mode)
+ .with_null_equality(self.null_equality)
+ .with_null_aware(false)
+ .build()?;
+
+ Ok(Arc::new(hash_join_exec))
+ }
+
+ pub fn to_sort_merge_join(&self) -> Result<Arc<SortMergeJoinExec>> {
+ //let null_equals_null = self.null_equality == NullEquality::NullEqualsNull;
+ let sort_options = vec![SortOptions::default(); self.on.len()];
+ Ok(Arc::new(SortMergeJoinExec::try_new(
+ Arc::clone(&self.left),
+ Arc::clone(&self.right),
+ self.on.clone(),
+ self.filter.clone(),
+ self.join_type,
+ sort_options,
+ self.null_equality,
+ )?))
+ }
+
+ fn supports_collect_by_thresholds(
+ plan: &dyn ExecutionPlan,
+ threshold_byte_size: usize,
+ threshold_num_rows: usize,
+ ) -> bool {
+ // Currently we do not trust the 0 value from stats, due to stats collection might have bug
+ // TODO check the logic in datasource::get_statistics_with_limit()
+ let Ok(stats) = plan.partition_statistics(None) else {
+ return false;
+ };
+
+ if let Some(byte_size) = stats.total_byte_size.get_value() {
+ *byte_size != 0 && *byte_size < threshold_byte_size
+ } else if let Some(num_rows) = stats.num_rows.get_value() {
+ *num_rows != 0 && *num_rows < threshold_num_rows
+ } else {
+ false
+ }
+ }
+
+ pub(crate) fn to_partitioned(&self) -> Self {
+ Self {
+ left: self.left.clone(),
+ right: self.right.clone(),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: self.properties.clone(),
+ selection_state: ChildrenState::Repartitioned,
+ }
+ }
+
+ pub(crate) fn from_hash_join(
+ hash_join: &HashJoinExec,
+ ) -> Result<Arc<DynamicJoinSelectionExec>> {
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(hash_join.left()),
+ right: Arc::clone(hash_join.right()),
+ on: hash_join.on().to_vec(),
+ filter: hash_join.filter().cloned(),
+ join_type: *hash_join.join_type(),
+ projection: hash_join.projection.as_deref().map(|p| p.to_vec()),
+ null_equality: hash_join.null_equality,
+ properties: Arc::clone(hash_join.properties()),
+ selection_state: ChildrenState::Unknown,
+ }))
+ }
+
+ pub(crate) fn children_resolved(&self) -> bool {
Review Comment:
We iterate plan twice with two methods, merge them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [WIP] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3302125324
##########
ballista/core/src/extension.rs:
##########
@@ -767,11 +767,11 @@ impl SessionConfigHelperExt for SessionConfig {
// has been disabled until fixed
.set_u64(
"datafusion.optimizer.hash_join_single_partition_threshold",
- 0,
+ 10 * 1024 * 1024,
Review Comment:
Revert values before merging
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,395 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ todo!(
Review Comment:
Don't panic here, return error
##########
ballista/scheduler/src/state/aqe/planner.rs:
##########
@@ -52,13 +56,16 @@ pub struct AdaptivePlanner {
/// Generates the next stage ID.
/// Stage IDs are incremental and unique for each job.
stage_id_generator: usize,
+ /// plan id generator
+ #[allow(dead_code)] // TODO: keep it for now until we refactor it
Review Comment:
Remove it maybe?
##########
ballista/scheduler/src/state/execution_stage.rs:
##########
@@ -1061,14 +1061,19 @@ impl StageOutput {
pub fn partition_locations(
mut self,
output_partitions: usize,
+ is_broadcast: bool,
Review Comment:
Break it into two methods
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,395 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ todo!(
+ "this should not be executed, it should have been replaced with optimizer rule"
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (ChildrenState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::LateCollectLeft),
+ //(ChildrenState::Repartitioned, PartitionMode::CollectLeft) =>
+ // self
+ // .to_hash_join(PartitionMode::Partitioned)
+ // .map(JoinSelectionAction::Hash),
+ (ChildrenState::Unknown, PartitionMode::Partitioned) => Ok(
+ JoinSelectionAction::Repartition(Arc::new(self.to_partitioned())),
+ ),
+ //
+ //
+ (_, PartitionMode::Auto) => internal_err!("this case should not be possible"),
+ }
+ }
+
+ pub(crate) fn to_hash_join(
+ &self,
+ partition_mode: PartitionMode,
+ ) -> Result<Arc<HashJoinExec>> {
+ let hash_join_exec = HashJoinExecBuilder::new(
+ self.left.clone(),
+ self.right.clone(),
+ self.on.clone(),
+ self.join_type,
+ )
+ .with_filter(self.filter.clone())
+ .with_projection(self.projection.clone())
+ .with_partition_mode(partition_mode)
+ .with_null_equality(self.null_equality)
+ .with_null_aware(false)
+ .build()?;
+
+ Ok(Arc::new(hash_join_exec))
+ }
+
+ pub fn to_sort_merge_join(&self) -> Result<Arc<SortMergeJoinExec>> {
+ //let null_equals_null = self.null_equality == NullEquality::NullEqualsNull;
+ let sort_options = vec![SortOptions::default(); self.on.len()];
+ Ok(Arc::new(SortMergeJoinExec::try_new(
+ Arc::clone(&self.left),
+ Arc::clone(&self.right),
+ self.on.clone(),
+ self.filter.clone(),
+ self.join_type,
+ sort_options,
+ self.null_equality,
+ )?))
+ }
+
+ fn supports_collect_by_thresholds(
+ plan: &dyn ExecutionPlan,
+ threshold_byte_size: usize,
+ threshold_num_rows: usize,
+ ) -> bool {
+ // Currently we do not trust the 0 value from stats, due to stats collection might have bug
+ // TODO check the logic in datasource::get_statistics_with_limit()
+ let Ok(stats) = plan.partition_statistics(None) else {
+ return false;
+ };
+
+ if let Some(byte_size) = stats.total_byte_size.get_value() {
+ *byte_size != 0 && *byte_size < threshold_byte_size
+ } else if let Some(num_rows) = stats.num_rows.get_value() {
+ *num_rows != 0 && *num_rows < threshold_num_rows
+ } else {
+ false
+ }
+ }
+
+ pub(crate) fn to_partitioned(&self) -> Self {
+ Self {
+ left: self.left.clone(),
+ right: self.right.clone(),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: self.properties.clone(),
+ selection_state: ChildrenState::Repartitioned,
+ }
+ }
+
+ pub(crate) fn from_hash_join(
+ hash_join: &HashJoinExec,
+ ) -> Result<Arc<DynamicJoinSelectionExec>> {
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(hash_join.left()),
+ right: Arc::clone(hash_join.right()),
+ on: hash_join.on().to_vec(),
+ filter: hash_join.filter().cloned(),
+ join_type: *hash_join.join_type(),
+ projection: hash_join.projection.as_deref().map(|p| p.to_vec()),
+ null_equality: hash_join.null_equality,
+ properties: Arc::clone(hash_join.properties()),
+ selection_state: ChildrenState::Unknown,
+ }))
+ }
+
+ pub(crate) fn children_resolved(&self) -> bool {
+ fn has_dynamic_join(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ plan.as_any()
+ .downcast_ref::<DynamicJoinSelectionExec>()
+ .is_some()
+ || plan.children().iter().any(|c| has_dynamic_join(c))
+ }
+
+ fn has_unresolved_exchange(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ if let Some(exchange) = plan.as_any().downcast_ref::<ExchangeExec>()
+ && !exchange.shuffle_created()
Review Comment:
We can stop on first resolved node, I believe
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,395 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ todo!(
+ "this should not be executed, it should have been replaced with optimizer rule"
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (ChildrenState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::LateCollectLeft),
+ //(ChildrenState::Repartitioned, PartitionMode::CollectLeft) =>
+ // self
+ // .to_hash_join(PartitionMode::Partitioned)
+ // .map(JoinSelectionAction::Hash),
+ (ChildrenState::Unknown, PartitionMode::Partitioned) => Ok(
+ JoinSelectionAction::Repartition(Arc::new(self.to_partitioned())),
+ ),
+ //
+ //
+ (_, PartitionMode::Auto) => internal_err!("this case should not be possible"),
+ }
+ }
+
+ pub(crate) fn to_hash_join(
+ &self,
+ partition_mode: PartitionMode,
+ ) -> Result<Arc<HashJoinExec>> {
+ let hash_join_exec = HashJoinExecBuilder::new(
+ self.left.clone(),
+ self.right.clone(),
+ self.on.clone(),
+ self.join_type,
+ )
+ .with_filter(self.filter.clone())
+ .with_projection(self.projection.clone())
+ .with_partition_mode(partition_mode)
+ .with_null_equality(self.null_equality)
+ .with_null_aware(false)
+ .build()?;
+
+ Ok(Arc::new(hash_join_exec))
+ }
+
+ pub fn to_sort_merge_join(&self) -> Result<Arc<SortMergeJoinExec>> {
+ //let null_equals_null = self.null_equality == NullEquality::NullEqualsNull;
+ let sort_options = vec![SortOptions::default(); self.on.len()];
+ Ok(Arc::new(SortMergeJoinExec::try_new(
+ Arc::clone(&self.left),
+ Arc::clone(&self.right),
+ self.on.clone(),
+ self.filter.clone(),
+ self.join_type,
+ sort_options,
+ self.null_equality,
+ )?))
+ }
+
+ fn supports_collect_by_thresholds(
Review Comment:
Mention this is taken from datafusion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4571734553
@metegenez im not sure if broadcast addition may break anything in coalesce side, when you get chance to review please let me know if i made something stupid
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "metegenez (via GitHub)" <gi...@apache.org>.
metegenez commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3329109527
##########
ballista/scheduler/src/state/aqe/test/join_selection.rs:
##########
@@ -0,0 +1,402 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::{
+ assert_plan,
+ state::aqe::{planner::AdaptivePlanner, test::mock_partitions_with_statistics},
+};
+use datafusion::{
+ arrow::{
+ array::{Int32Array, RecordBatch},
+ datatypes::{DataType, Field, Schema},
+ },
+ datasource::MemTable,
+ execution::{SessionStateBuilder, config::SessionConfig, context::SessionContext},
+};
+use std::sync::Arc;
+
+fn make_table(schema: Arc<Schema>) -> Arc<MemTable> {
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0])),
+ Arc::new(Int32Array::from(vec![
+ 10, 20, 30, 40, 50, 60, 70, 80, 90, 100,
+ ])),
+ ],
+ )
+ .unwrap();
+
+ Arc::new(
+ MemTable::try_new(
+ schema,
+ vec![
+ vec![batch.clone()],
+ vec![batch.clone()],
+ vec![batch.clone()],
+ vec![batch],
+ ],
+ )
+ .unwrap(),
+ )
+}
+
+fn make_ctx(prefer_hash_join: bool) -> SessionContext {
+ let config = SessionConfig::new()
+ // .set_u64(
Review Comment:
Is this comment-out intentional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4530159628
Note to myself: check #1643
#1643 mentions pushing down
```
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
```
And doing broadcast join, this is true for aqe implementation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "martin-g (via GitHub)" <gi...@apache.org>.
martin-g commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3317917581
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,528 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
Review Comment:
```suggestion
threshold_collect_left_join_bytes,
```
?!
##########
ballista/scheduler/src/state/aqe/execution_plan/exchange.rs:
##########
@@ -98,8 +103,38 @@ impl ExchangeExec {
plan_id,
Arc::new(AtomicI64::new(-1)),
Arc::new(Mutex::new(None)),
+ false,
)
}
+ /// new broadcast exchange
+ pub fn new_broadcast(
+ input: Arc<dyn ExecutionPlan>,
+ partitioning: Option<Partitioning>,
+ plan_id: usize,
+ ) -> Self {
+ Self::new_with_details(
+ input,
+ partitioning,
+ plan_id,
+ Arc::new(AtomicI64::new(-1)),
+ Arc::new(Mutex::new(None)),
+ true,
+ )
+ }
+
+ pub fn to_broadcast(&self, plan_id: usize) -> Self {
+ Self {
+ input: self.input.clone(),
+ properties: self.properties.clone(),
Review Comment:
Should this update the partitioning to `Partitioning::UnknownPartitioning(1)` ?
My AI friend says that Broadcast exchange should use 1 partition.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,528 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (ChildrenState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::LateCollectLeft),
+ //(ChildrenState::Repartitioned, PartitionMode::CollectLeft) =>
+ // self
+ // .to_hash_join(PartitionMode::Partitioned)
+ // .map(JoinSelectionAction::Hash),
+ (ChildrenState::Unknown, PartitionMode::Partitioned) => Ok(
+ JoinSelectionAction::Repartition(Arc::new(self.to_partitioned())),
+ ),
+ //
+ //
+ (_, PartitionMode::Auto) => internal_err!("this case should not be possible"),
+ }
+ }
+
+ pub(crate) fn to_hash_join(
+ &self,
+ partition_mode: PartitionMode,
+ ) -> Result<Arc<HashJoinExec>> {
+ let hash_join_exec = HashJoinExecBuilder::new(
+ self.left.clone(),
+ self.right.clone(),
+ self.on.clone(),
+ self.join_type,
+ )
+ .with_filter(self.filter.clone())
+ .with_projection(self.projection.clone())
+ .with_partition_mode(partition_mode)
+ .with_null_equality(self.null_equality)
+ .with_null_aware(false)
+ .build()?;
+
+ Ok(Arc::new(hash_join_exec))
+ }
+
+ pub fn to_sort_merge_join(&self) -> Result<Arc<SortMergeJoinExec>> {
+ //let null_equals_null = self.null_equality == NullEquality::NullEqualsNull;
+ let sort_options = vec![SortOptions::default(); self.on.len()];
+ Ok(Arc::new(SortMergeJoinExec::try_new(
+ Arc::clone(&self.left),
+ Arc::clone(&self.right),
+ self.on.clone(),
+ self.filter.clone(),
+ self.join_type,
+ sort_options,
+ self.null_equality,
+ )?))
+ }
+ // this method has been taken from datafusion
+ // join selection optimizer rule
+ fn supports_collect_by_thresholds(
+ plan: &dyn ExecutionPlan,
+ threshold_byte_size: usize,
+ threshold_num_rows: usize,
+ ) -> bool {
+ // Currently we do not trust the 0 value from stats, due to stats collection might have bug
+ // TODO check the logic in datasource::get_statistics_with_limit()
+ let Ok(stats) = plan.partition_statistics(None) else {
+ return false;
+ };
+
+ if let Some(byte_size) = stats.total_byte_size.get_value() {
+ *byte_size != 0 && *byte_size < threshold_byte_size
+ } else if let Some(num_rows) = stats.num_rows.get_value() {
+ *num_rows != 0 && *num_rows < threshold_num_rows
+ } else {
+ false
+ }
+ }
+
+ pub(crate) fn to_partitioned(&self) -> Self {
+ Self {
+ left: self.left.clone(),
+ right: self.right.clone(),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: self.properties.clone(),
+ selection_state: ChildrenState::Repartitioned,
+ }
+ }
+
+ pub(crate) fn from_hash_join(
+ hash_join: &HashJoinExec,
Review Comment:
should this preserve HashJoinExec's `null_aware` ?
`to_hash_join()` above always sets it to `false`, so it will be lost if it is still needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4582832561
# Q72 (AQE) ~ 65 sec
<img width="1484" height="348" alt="image" src="https://github.com/user-attachments/assets/f9dcac80-fc46-408b-b6f8-f3546cb604e0" />
```
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Job sO09M9B physical plan:
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
AdaptiveDatafusionExec: is_final=true, plan_id=2, stage_id=15, stage_resolved=true
SortPreservingMergeExec: [total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], fetch=100
ExchangeExec: partitioning=None, plan_id=19, stage_id=14, stage_resolved=true
SortExec: TopK(fetch=100), expr=[total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq, sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END)@3 as no_promo, sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END)@4 as promo, count(Int64(1))@5 as total_cnt]
AggregateExec: mode=FinalPartitioned, gby=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
ExchangeExec: partitioning=Hash([i_item_desc@0, w_warehouse_name@1, d_week_seq@2], 128), plan_id=18, stage_id=13, stage_resolved=true
AggregateExec: mode=Partial, gby=[i_item_desc@1 as i_item_desc, w_warehouse_name@0 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Left, on=[(cs_item_sk@0, cr_item_sk@0), (cs_order_number@1, cr_order_number@1)], projection=[w_warehouse_name@2, i_item_desc@3, d_week_seq@4, p_promo_sk@5]
ExchangeExec: partitioning=None, plan_id=17, stage_id=11, stage_resolved=true, broadcast=true
ProjectionExec: expr=[cs_item_sk@1 as cs_item_sk, cs_order_number@2 as cs_order_number, w_warehouse_name@3 as w_warehouse_name, i_item_desc@4 as i_item_desc, d_week_seq@5 as d_week_seq, p_promo_sk@0 as p_promo_sk]
HashJoinExec: mode=CollectLeft, join_type=Right, on=[(p_promo_sk@0, cs_promo_sk@1)], projection=[p_promo_sk@0, cs_item_sk@1, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@6]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/promotion.parquet]]}, projection=[p_promo_sk], file_type=parquet
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_ship_date_sk@0, d_date_sk@0)], filter=d_date@1 > d_date@0 + IntervalMonthDayNano { months: 0, days: 5, nanoseconds: 0 }, projection=[cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@7]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, inv_date_sk@4), (d_week_seq@1, d_week_seq@8)], projection=[cs_ship_date_sk@2, cs_item_sk@3, cs_promo_sk@4, cs_order_number@5, w_warehouse_name@7, i_item_desc@8, d_date@9, d_week_seq@10]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_week_seq], file_type=parquet, predicate=DynamicFilter [ empty ]
ExchangeExec: partitioning=None, plan_id=14, stage_id=10, stage_resolved=true, broadcast=true
ProjectionExec: expr=[cs_ship_date_sk@2 as cs_ship_date_sk, cs_item_sk@3 as cs_item_sk, cs_promo_sk@4 as cs_promo_sk, cs_order_number@5 as cs_order_number, inv_date_sk@6 as inv_date_sk, w_warehouse_name@7 as w_warehouse_name, i_item_desc@8 as i_item_desc, d_date@0 as d_date, d_week_seq@1 as d_week_seq]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, cs_sold_date_sk@0)], projection=[d_date@1, d_week_seq@2, cs_ship_date_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
FilterExec: d_year@3 = 1999, projection=[d_date_sk@0, d_date@1, d_week_seq@2]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date, d_week_seq, d_year], file_type=parquet, predicate=d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999, pruning_predicate=d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 199
9 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1
AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1, required_guarantees=[d_year in (1999)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(hd_demo_sk@0, cs_bill_hdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
ExchangeExec: partitioning=None, plan_id=13, stage_id=9, stage_resolved=true, broadcast=true
FilterExec: hd_buy_potential@1 = 501-1000, projection=[hd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/household_demographics.parquet]]}, projection=[hd_demo_sk, hd_buy_potential], file_type=parquet, predicate=hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000, pruning_predicate=hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND
hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2
!= row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1, required_guara
ntees=[hd_buy_potential in (501-1000)]
ExchangeExec: partitioning=Hash([cs_bill_hdemo_sk@2], 128), plan_id=11, stage_id=8, stage_resolved=true
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cd_demo_sk@0, cs_bill_cdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_bill_hdemo_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
ExchangeExec: partitioning=None, plan_id=10, stage_id=7, stage_resolved=true, broadcast=true
FilterExec: cd_marital_status@1 = S, projection=[cd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/customer_demographics.parquet]]}, projection=[cd_demo_sk, cd_marital_status], file_type=parquet, predicate=cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S, pruning_predicate=cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_coun
t@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1
AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1, required_guarantees=[cd_marital_status in (S)]
ExchangeExec: partitioning=Hash([cs_bill_cdemo_sk@2], 128), plan_id=8, stage_id=6, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@0 as cs_sold_date_sk, cs_ship_date_sk@1 as cs_ship_date_sk, cs_bill_cdemo_sk@2 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@3 as cs_bill_hdemo_sk, cs_item_sk@4 as cs_item_sk, cs_promo_sk@5 as cs_promo_sk, cs_order_number@6 as cs_order_number, inv_date_sk@7 as inv_date_sk, w_warehouse_name@8 as w_warehouse_name, i_item_desc@10 as i_item_desc]
SortMergeJoinExec: join_type=Inner, on=[(cs_item_sk@4, i_item_sk@0)]
ExchangeExec: partitioning=Hash([cs_item_sk@4], 128), plan_id=6, stage_id=4, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@1 as cs_sold_date_sk, cs_ship_date_sk@2 as cs_ship_date_sk, cs_bill_cdemo_sk@3 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@4 as cs_bill_hdemo_sk, cs_item_sk@5 as cs_item_sk, cs_promo_sk@6 as cs_promo_sk, cs_order_number@7 as cs_order_number, inv_date_sk@8 as inv_date_sk, w_warehouse_name@0 as w_warehouse_name]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(w_warehouse_sk@0, inv_warehouse_sk@8)], projection=[w_warehouse_name@1, cs_sold_date_sk@2, cs_ship_date_sk@3, cs_bill_cdemo_sk@4, cs_bill_hdemo_sk@5, cs_item_sk@6, cs_promo_sk@7, cs_order_number@8, inv_date_sk@9]
ExchangeExec: partitioning=None, plan_id=5, stage_id=3, stage_resolved=true, broadcast=true
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/warehouse.parquet]]}, projection=[w_warehouse_sk, w_warehouse_name], file_type=parquet
ExchangeExec: partitioning=Hash([inv_warehouse_sk@8], 128), plan_id=3, stage_id=2, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@0 as cs_sold_date_sk, cs_ship_date_sk@1 as cs_ship_date_sk, cs_bill_cdemo_sk@2 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@3 as cs_bill_hdemo_sk, cs_item_sk@4 as cs_item_sk, cs_promo_sk@5 as cs_promo_sk, cs_order_number@6 as cs_order_number, inv_date_sk@8 as inv_date_sk, inv_warehouse_sk@10 as inv_warehouse_sk]
SortMergeJoinExec: join_type=Inner, on=[(cs_item_sk@4, inv_item_sk@1)], filter=inv_quantity_on_hand@1 < cs_quantity@0
SortExec: expr=[cs_item_sk@4 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([cs_item_sk@4], 128), plan_id=0, stage_id=0, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:0..7953492], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:7953492..15906984], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:15906984..23860476], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:23860476..31813968], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:31813968..39767460], ...]}, projection=[cs_sold_date_sk, cs_ship_date_sk, cs_bill_cdemo_sk, cs_bill_hdemo_sk, cs_item_sk, cs_promo_sk, cs_order_number, cs_quantity], file_type=parquet
SortExec: expr=[inv_item_sk@1 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([inv_item_sk@1], 128), plan_id=1, stage_id=1, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:0..3954771], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:3954771..7909542], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:7909542..11864313], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:11864313..15819084], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:15819084..19773855], ...]}, projection=[inv_date_sk, inv_item_sk, inv_warehouse_sk, inv_quantity_on_hand], file_type=parquet
SortExec: expr=[i_item_sk@0 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([i_item_sk@0], 128), plan_id=7, stage_id=5, stage_resolved=true
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/item.parquet]]}, projection=[i_item_sk, i_item_desc], file_type=parquet
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ExchangeExec: partitioning=Hash([cr_item_sk@0, cr_order_number@1], 128), plan_id=16, stage_id=12, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:0..781584], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:781584..1563168], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:1563168..2344752], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:2344752..3126336], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:3126336..3907920], ...]}, projection=[cr_item_sk, cr_order_number], file_type=parquet
```
SET datafusion.optimizer.hash_join_single_partition_threshold=7485760"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3315649933
##########
ballista/scheduler/src/state/aqe/planner.rs:
##########
@@ -118,16 +127,54 @@ impl AdaptivePlanner {
///
/// # Returns
/// A new instance of `AdaptivePlanner` or an error if the initialization fails.
- pub fn try_new(
+ #[cfg(test)]
+ pub fn try_from_plan(
session_config: &SessionConfig,
plan: Arc<dyn ExecutionPlan>,
job_name: String,
) -> common::Result<Self> {
+ let plan_id_generator = Arc::new(AtomicUsize::new(0));
Self::try_new_with_optimizers(
session_config,
plan,
+ plan_id_generator.clone(),
+ job_name,
+ Self::default_optimizers(plan_id_generator),
+ )
+ }
+
+ /// Creates a new `AdaptivePlanner` with default physical optimizer rules.
+ ///
+ /// # Arguments
+ ///
+ /// * `ctx` - The session context
+ /// * `logical_plan` - The logical plan for the job.
+ /// * `job_name` - The name of the job.
+ ///
+ /// # Returns
+ /// A new instance of `AdaptivePlanner` or an error if the initialization fails.
+ pub async fn try_new(
+ ctx: &SessionContext,
+ logical_plan: &LogicalPlan,
+ job_name: String,
+ ) -> common::Result<Self> {
+ let state = SessionStateBuilder::new_from_existing(ctx.state())
Review Comment:
Extract rules to `plan preparation` rule set
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [WIP] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4557643930
## TPCH (SF10) AQE Enabled
```
Query 1 took 0.990 s and returned 4 rows
Query 2 took 0.514 s and returned 100 rows
Query 3 took 1.196 s and returned 10 rows
Query 4 took 0.775 s and returned 5 rows
Query 5 took 3.186 s and returned 5 rows
Query 6 took 0.261 s and returned 1 rows
Query 7 took 2.256 s and returned 4 rows
Query 8 took 0.722 s and returned 2 rows
Query 9 took 1.493 s and returned 168 rows
Query 10 took 1.246 s and returned 20 rows
Query 11 took 0.622 s and returned 0 rows
Query 12 took 0.821 s and returned 2 rows
Query 13 took 1.334 s and returned 46 rows
Query 14 took 0.401 s and returned 1 rows
Query 15 took 0.435 s and returned 0 rows
Query 16 took 0.320 s and returned 27840 rows
Query 17 took 3.115 s and returned 1 rows
Query 18 took 5.478 s and returned 100 rows
Query 19 took 0.831 s and returned 1 rows
Query 20 took 0.764 s and returned 1804 rows
Query 21 took 3.321 s and returned 100 rows
Query 22 took 0.549 s and returned 7 rows
Total time: 30.632 s
```
<img width="1711" height="419" alt="image" src="https://github.com/user-attachments/assets/a438589b-d7d0-4cab-ab74-8ef8d2cac974" />
## TPCH (SF10) AQE Disabled
```
Query 1 took 1.002 s and returned 4 rows
Query 2 took 1.096 s and returned 100 rows
Query 3 took 1.481 s and returned 10 rows
Query 4 took 0.764 s and returned 5 rows
Query 5 took 3.157 s and returned 5 rows
Query 6 took 0.267 s and returned 1 rows
Query 7 took 4.247 s and returned 4 rows
Query 8 took 4.588 s and returned 2 rows
Query 9 took 6.123 s and returned 168 rows
Query 10 took 1.326 s and returned 20 rows
Query 11 took 0.861 s and returned 0 rows
Query 12 took 0.992 s and returned 2 rows
Query 13 took 1.264 s and returned 46 rows
Query 14 took 0.423 s and returned 1 rows
Query 15 took 0.439 s and returned 0 rows
Query 16 took 0.314 s and returned 27840 rows
Query 17 took 4.774 s and returned 1 rows
Query 18 took 3.708 s and returned 100 rows
Query 19 took 0.859 s and returned 1 rows
Query 20 took 0.864 s and returned 1804 rows
Query 21 took 7.489 s and returned 100 rows
Query 22 took 0.534 s and returned 7 rows
Total time: 46.572 s
```
<img width="1709" height="415" alt="image" src="https://github.com/user-attachments/assets/718800e4-742a-4e05-816e-428bdf9b810c" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4597180155
After spending the long weekend with this PR, key takeout
Setting
```
datafusion.optimizer.hash_join_single_partition_threshold=1048576
```
Works, but not all stages have actual byte size reported, missing chance for further broadcast optimization
Setting
```
datafusion.optimizer.hash_join_single_partition_threshold_rows=100_000
```
At some point gets tucked with wrong row number reported, making a stage injecting billions or rows single partitioned, that stage makes job execution into 200sec range
Possible follow ups
- find the reason for wrong statistics
- when join is not collect left, we make it partitioned and run both stages at the same time. We should change this and make educated guess to run one of them first hoping it will produce change which could be folded into broadcast
Will do code clean up merge this pr and follow up with improvements as it's definitely improving things
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4582856165
## Q72 (60sec) Broadcast 10MB
```
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Job XM6lgrA physical plan:
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
AdaptiveDatafusionExec: is_final=true, plan_id=2, stage_id=15, stage_resolved=true
SortPreservingMergeExec: [total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], fetch=100
ExchangeExec: partitioning=None, plan_id=19, stage_id=14, stage_resolved=true
SortExec: TopK(fetch=100), expr=[total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq, sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END)@3 as no_promo, sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END)@4 as promo, count(Int64(1))@5 as total_cnt]
AggregateExec: mode=FinalPartitioned, gby=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
ExchangeExec: partitioning=Hash([i_item_desc@0, w_warehouse_name@1, d_week_seq@2], 128), plan_id=18, stage_id=13, stage_resolved=true
AggregateExec: mode=Partial, gby=[i_item_desc@1 as i_item_desc, w_warehouse_name@0 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Left, on=[(cs_item_sk@0, cr_item_sk@0), (cs_order_number@1, cr_order_number@1)], projection=[w_warehouse_name@2, i_item_desc@3, d_week_seq@4, p_promo_sk@5]
ExchangeExec: partitioning=None, plan_id=17, stage_id=11, stage_resolved=true, broadcast=true
ProjectionExec: expr=[cs_item_sk@1 as cs_item_sk, cs_order_number@2 as cs_order_number, w_warehouse_name@3 as w_warehouse_name, i_item_desc@4 as i_item_desc, d_week_seq@5 as d_week_seq, p_promo_sk@0 as p_promo_sk]
HashJoinExec: mode=CollectLeft, join_type=Right, on=[(p_promo_sk@0, cs_promo_sk@1)], projection=[p_promo_sk@0, cs_item_sk@1, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@6]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/promotion.parquet]]}, projection=[p_promo_sk], file_type=parquet
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_ship_date_sk@0, d_date_sk@0)], filter=d_date@1 > d_date@0 + IntervalMonthDayNano { months: 0, days: 5, nanoseconds: 0 }, projection=[cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@7]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, inv_date_sk@4), (d_week_seq@1, d_week_seq@8)], projection=[cs_ship_date_sk@2, cs_item_sk@3, cs_promo_sk@4, cs_order_number@5, w_warehouse_name@7, i_item_desc@8, d_date@9, d_week_seq@10]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_week_seq], file_type=parquet, predicate=DynamicFilter [ empty ]
ExchangeExec: partitioning=None, plan_id=14, stage_id=10, stage_resolved=true, broadcast=true
ProjectionExec: expr=[cs_ship_date_sk@2 as cs_ship_date_sk, cs_item_sk@3 as cs_item_sk, cs_promo_sk@4 as cs_promo_sk, cs_order_number@5 as cs_order_number, inv_date_sk@6 as inv_date_sk, w_warehouse_name@7 as w_warehouse_name, i_item_desc@8 as i_item_desc, d_date@0 as d_date, d_week_seq@1 as d_week_seq]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, cs_sold_date_sk@0)], projection=[d_date@1, d_week_seq@2, cs_ship_date_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
FilterExec: d_year@3 = 1999, projection=[d_date_sk@0, d_date@1, d_week_seq@2]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date, d_week_seq, d_year], file_type=parquet, predicate=d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999, pruning_predicate=d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 199
9 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1
AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1, required_guarantees=[d_year in (1999)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(hd_demo_sk@0, cs_bill_hdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
ExchangeExec: partitioning=None, plan_id=13, stage_id=9, stage_resolved=true, broadcast=true
FilterExec: hd_buy_potential@1 = 501-1000, projection=[hd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/household_demographics.parquet]]}, projection=[hd_demo_sk, hd_buy_potential], file_type=parquet, predicate=hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000, pruning_predicate=hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND
hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2
!= row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1, required_guara
ntees=[hd_buy_potential in (501-1000)]
ExchangeExec: partitioning=Hash([cs_bill_hdemo_sk@2], 128), plan_id=11, stage_id=8, stage_resolved=true
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cd_demo_sk@0, cs_bill_cdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_bill_hdemo_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
ExchangeExec: partitioning=None, plan_id=10, stage_id=7, stage_resolved=true, broadcast=true
FilterExec: cd_marital_status@1 = S, projection=[cd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/customer_demographics.parquet]]}, projection=[cd_demo_sk, cd_marital_status], file_type=parquet, predicate=cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S, pruning_predicate=cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_coun
t@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1
AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1, required_guarantees=[cd_marital_status in (S)]
ExchangeExec: partitioning=Hash([cs_bill_cdemo_sk@2], 128), plan_id=8, stage_id=6, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@0 as cs_sold_date_sk, cs_ship_date_sk@1 as cs_ship_date_sk, cs_bill_cdemo_sk@2 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@3 as cs_bill_hdemo_sk, cs_item_sk@4 as cs_item_sk, cs_promo_sk@5 as cs_promo_sk, cs_order_number@6 as cs_order_number, inv_date_sk@7 as inv_date_sk, w_warehouse_name@8 as w_warehouse_name, i_item_desc@10 as i_item_desc]
SortMergeJoinExec: join_type=Inner, on=[(cs_item_sk@4, i_item_sk@0)]
ExchangeExec: partitioning=Hash([cs_item_sk@4], 128), plan_id=6, stage_id=4, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@1 as cs_sold_date_sk, cs_ship_date_sk@2 as cs_ship_date_sk, cs_bill_cdemo_sk@3 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@4 as cs_bill_hdemo_sk, cs_item_sk@5 as cs_item_sk, cs_promo_sk@6 as cs_promo_sk, cs_order_number@7 as cs_order_number, inv_date_sk@8 as inv_date_sk, w_warehouse_name@0 as w_warehouse_name]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(w_warehouse_sk@0, inv_warehouse_sk@8)], projection=[w_warehouse_name@1, cs_sold_date_sk@2, cs_ship_date_sk@3, cs_bill_cdemo_sk@4, cs_bill_hdemo_sk@5, cs_item_sk@6, cs_promo_sk@7, cs_order_number@8, inv_date_sk@9]
ExchangeExec: partitioning=None, plan_id=5, stage_id=3, stage_resolved=true, broadcast=true
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/warehouse.parquet]]}, projection=[w_warehouse_sk, w_warehouse_name], file_type=parquet
ExchangeExec: partitioning=Hash([inv_warehouse_sk@8], 128), plan_id=3, stage_id=2, stage_resolved=true
ProjectionExec: expr=[cs_sold_date_sk@0 as cs_sold_date_sk, cs_ship_date_sk@1 as cs_ship_date_sk, cs_bill_cdemo_sk@2 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@3 as cs_bill_hdemo_sk, cs_item_sk@4 as cs_item_sk, cs_promo_sk@5 as cs_promo_sk, cs_order_number@6 as cs_order_number, inv_date_sk@8 as inv_date_sk, inv_warehouse_sk@10 as inv_warehouse_sk]
SortMergeJoinExec: join_type=Inner, on=[(cs_item_sk@4, inv_item_sk@1)], filter=inv_quantity_on_hand@1 < cs_quantity@0
SortExec: expr=[cs_item_sk@4 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([cs_item_sk@4], 128), plan_id=0, stage_id=0, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:0..7953492], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:7953492..15906984], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:15906984..23860476], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:23860476..31813968], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_sales.parquet:31813968..39767460], ...]}, projection=[cs_sold_date_sk, cs_ship_date_sk, cs_bill_cdemo_sk, cs_bill_hdemo_sk, cs_item_sk, cs_promo_sk, cs_order_number, cs_quantity], file_type=parquet
SortExec: expr=[inv_item_sk@1 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([inv_item_sk@1], 128), plan_id=1, stage_id=1, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:0..3954771], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:3954771..7909542], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:7909542..11864313], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:11864313..15819084], [Users/marko/TMP/tpcds_data/10g_parquet/inventory.parquet:15819084..19773855], ...]}, projection=[inv_date_sk, inv_item_sk, inv_warehouse_sk, inv_quantity_on_hand], file_type=parquet
SortExec: expr=[i_item_sk@0 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([i_item_sk@0], 128), plan_id=7, stage_id=5, stage_resolved=true
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/item.parquet]]}, projection=[i_item_sk, i_item_desc], file_type=parquet
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/10g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ExchangeExec: partitioning=Hash([cr_item_sk@0, cr_order_number@1], 128), plan_id=16, stage_id=12, stage_resolved=true
DataSourceExec: file_groups={128 groups: [[Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:0..781584], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:781584..1563168], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:1563168..2344752], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:2344752..3126336], [Users/marko/TMP/tpcds_data/10g_parquet/catalog_returns.parquet:3126336..3907920], ...]}, projection=[cr_item_sk, cr_order_number], file_type=parquet
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm merged PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3293527619
##########
ballista/scheduler/src/state/aqe/mod.rs:
##########
@@ -142,22 +141,37 @@ impl AdaptiveExecutionGraph {
) -> ballista_core::error::Result<Self> {
let session_id = ctx.session_id();
- // TODO: this is to be changed, we do not run default optimizers comming with the state at this
+ // TODO: this is to be changed, we do not run default optimizers coming with the state at this
// point.
- let plan = ctx.state().create_physical_plan(logical_plan).await?;
- // TODO: explain plan will probably return wrong explanation
- // do we handle it now or just ignore it?
- let physical_plan = handle_explain_plan(job_id, ctx, logical_plan, plan).await?;
+ // this state does not have any of the physical optimizers build in by default
Review Comment:
Not needed, remove
##########
ballista/scheduler/src/state/aqe/mod.rs:
##########
@@ -207,7 +221,7 @@ impl AdaptiveExecutionGraph {
failed_stage_attempts: HashMap::new(),
session_config,
logical_plan,
- physical_plan,
+ //physical_plan,
Review Comment:
Remove
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,395 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ todo!(
+ "this should not be executed, it should have been replaced with optimizer rule"
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (ChildrenState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
+ //
+ //
+ (ChildrenState::Repartitioned, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::LateCollectLeft),
+ //(ChildrenState::Repartitioned, PartitionMode::CollectLeft) =>
+ // self
+ // .to_hash_join(PartitionMode::Partitioned)
+ // .map(JoinSelectionAction::Hash),
+ (ChildrenState::Unknown, PartitionMode::Partitioned) => Ok(
+ JoinSelectionAction::Repartition(Arc::new(self.to_partitioned())),
+ ),
+ //
+ //
+ (_, PartitionMode::Auto) => internal_err!("this case should not be possible"),
+ }
+ }
+
+ pub(crate) fn to_hash_join(
+ &self,
+ partition_mode: PartitionMode,
+ ) -> Result<Arc<HashJoinExec>> {
+ let hash_join_exec = HashJoinExecBuilder::new(
+ self.left.clone(),
+ self.right.clone(),
+ self.on.clone(),
+ self.join_type,
+ )
+ .with_filter(self.filter.clone())
+ .with_projection(self.projection.clone())
+ .with_partition_mode(partition_mode)
+ .with_null_equality(self.null_equality)
+ .with_null_aware(false)
+ .build()?;
+
+ Ok(Arc::new(hash_join_exec))
+ }
+
+ pub fn to_sort_merge_join(&self) -> Result<Arc<SortMergeJoinExec>> {
+ //let null_equals_null = self.null_equality == NullEquality::NullEqualsNull;
+ let sort_options = vec![SortOptions::default(); self.on.len()];
+ Ok(Arc::new(SortMergeJoinExec::try_new(
+ Arc::clone(&self.left),
+ Arc::clone(&self.right),
+ self.on.clone(),
+ self.filter.clone(),
+ self.join_type,
+ sort_options,
+ self.null_equality,
+ )?))
+ }
+
+ fn supports_collect_by_thresholds(
+ plan: &dyn ExecutionPlan,
+ threshold_byte_size: usize,
+ threshold_num_rows: usize,
+ ) -> bool {
+ // Currently we do not trust the 0 value from stats, due to stats collection might have bug
+ // TODO check the logic in datasource::get_statistics_with_limit()
+ let Ok(stats) = plan.partition_statistics(None) else {
+ return false;
+ };
+
+ if let Some(byte_size) = stats.total_byte_size.get_value() {
+ *byte_size != 0 && *byte_size < threshold_byte_size
+ } else if let Some(num_rows) = stats.num_rows.get_value() {
+ *num_rows != 0 && *num_rows < threshold_num_rows
+ } else {
+ false
+ }
+ }
+
+ pub(crate) fn to_partitioned(&self) -> Self {
+ Self {
+ left: self.left.clone(),
+ right: self.right.clone(),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: self.properties.clone(),
+ selection_state: ChildrenState::Repartitioned,
+ }
+ }
+
+ pub(crate) fn from_hash_join(
+ hash_join: &HashJoinExec,
+ ) -> Result<Arc<DynamicJoinSelectionExec>> {
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(hash_join.left()),
+ right: Arc::clone(hash_join.right()),
+ on: hash_join.on().to_vec(),
+ filter: hash_join.filter().cloned(),
+ join_type: *hash_join.join_type(),
+ projection: hash_join.projection.as_deref().map(|p| p.to_vec()),
+ null_equality: hash_join.null_equality,
+ properties: Arc::clone(hash_join.properties()),
+ selection_state: ChildrenState::Unknown,
+ }))
+ }
+
+ pub(crate) fn children_are_ready(&self) -> bool {
Review Comment:
Horrible method name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3329330092
##########
ballista/scheduler/src/state/aqe/test/join_selection.rs:
##########
@@ -0,0 +1,402 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::{
+ assert_plan,
+ state::aqe::{planner::AdaptivePlanner, test::mock_partitions_with_statistics},
+};
+use datafusion::{
+ arrow::{
+ array::{Int32Array, RecordBatch},
+ datatypes::{DataType, Field, Schema},
+ },
+ datasource::MemTable,
+ execution::{SessionStateBuilder, config::SessionConfig, context::SessionContext},
+};
+use std::sync::Arc;
+
+fn make_table(schema: Arc<Schema>) -> Arc<MemTable> {
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0])),
+ Arc::new(Int32Array::from(vec![
+ 10, 20, 30, 40, 50, 60, 70, 80, 90, 100,
+ ])),
+ ],
+ )
+ .unwrap();
+
+ Arc::new(
+ MemTable::try_new(
+ schema,
+ vec![
+ vec![batch.clone()],
+ vec![batch.clone()],
+ vec![batch.clone()],
+ vec![batch],
+ ],
+ )
+ .unwrap(),
+ )
+}
+
+fn make_ctx(prefer_hash_join: bool) -> SessionContext {
+ let config = SessionConfig::new()
+ // .set_u64(
Review Comment:
yes, i believe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "metegenez (via GitHub)" <gi...@apache.org>.
metegenez commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4584555716
Also, when we add bailout code i give,
```
=== Stage 7 === (final join chain → ShuffleWriter)
HashJoinExec: mode=CollectLeft on (r_regionkey, n_regionkey) ← region BROADCAST
[output_rows=156.7K, build_input_rows=3, join_time=13.8ms]
FilterExec: r_name = EUROPE → 1 row
HashJoinExec: mode=CollectLeft on (n_nationkey, s_nationkey) ← nation BROADCAST
[output_rows=831.6K, build_input_rows=75, build_mem=2.65K, join_time=48ms]
DataSourceExec: nation (75 rows)
HashJoinExec: mode=Partitioned on (ps_suppkey, s_suppkey) ← big⋈big HASH
[output_rows=839.8K, build_mem=76.25M, join_time=108ms]
ShuffleReaderExec: stage 5, coalesce: 3 of 16 ← COALESCE 16→3
ShuffleReaderExec: stage 6, coalesce: 3 of 16 ← COALESCE 16→3
```
they can work together/enable together as well. Nice work @milenkovicm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [WIP] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3301062383
##########
ballista/scheduler/src/state/aqe/mod.rs:
##########
@@ -122,8 +121,8 @@ pub(crate) struct AdaptiveExecutionGraph {
session_config: Arc<SessionConfig>,
/// Logical plan as a human-readable string, captured at submission time.
logical_plan: Option<String>,
- /// Physical plan, captured at submission time.
- physical_plan: Arc<dyn ExecutionPlan>,
+ // /// Physical plan, captured at submission time.
Review Comment:
To be removed
##########
ballista/scheduler/src/state/aqe/optimizer_rule/join_selection.rs:
##########
@@ -0,0 +1,644 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use ballista_core::config::BallistaConfig;
+use datafusion::{
+ catalog::memory::DataSourceExec,
+ common::tree_node::{Transformed, TreeNode},
+ error::DataFusionError,
+ physical_optimizer::PhysicalOptimizerRule,
+ physical_plan::{
+ ExecutionPlan,
+ joins::{HashJoinExec, PartitionMode, SortMergeJoinExec},
+ },
+};
+use std::sync::{Arc, atomic::AtomicUsize};
+
+use crate::state::aqe::execution_plan::{
+ DynamicJoinSelectionExec, ExchangeExec, JoinSelectionAction,
+};
+
+#[derive(Debug, Default)]
+pub struct DelayJoinSelectionRule {}
+
+impl PhysicalOptimizerRule for DelayJoinSelectionRule {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &datafusion::config::ConfigOptions,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ let bc = config
+ .extensions
+ .get::<BallistaConfig>()
+ .cloned()
+ .unwrap_or_default();
+ if bc.adaptive_join_enabled() {
+ let result = plan.transform_up(|node| {
+ if let Some(hj) = node.as_any().downcast_ref::<HashJoinExec>() {
+ let dynamic = DynamicJoinSelectionExec::from_hash_join(hj)?;
+ Ok(Transformed::yes(dynamic as Arc<dyn ExecutionPlan>))
+ } else if let Some(smj) =
+ node.as_any().downcast_ref::<SortMergeJoinExec>()
+ {
+ let dynamic = DynamicJoinSelectionExec::from_sort_join(smj)?;
+ Ok(Transformed::yes(dynamic as Arc<dyn ExecutionPlan>))
+ } else {
+ Ok(Transformed::no(node))
+ }
+ })?;
+ Ok(result.data)
+ } else {
+ return Ok(plan);
+ }
+ }
+
+ fn name(&self) -> &str {
+ "DelayJoinSelectionRule"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+#[derive(Debug, Default)]
+pub struct SelectJoinRule {
+ plan_id_generator: Arc<AtomicUsize>,
+}
+
+impl SelectJoinRule {
+ pub(crate) fn new(plan_id_generator: Arc<AtomicUsize>) -> Self {
+ SelectJoinRule { plan_id_generator }
+ }
+
+ fn plan_id(&self) -> usize {
+ self.plan_id_generator
+ .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+ }
+
+ pub fn supports_swap_join_order(
+ left: &dyn ExecutionPlan,
+ right: &dyn ExecutionPlan,
+ ) -> datafusion::common::Result<bool> {
+ // Get the left and right table's total bytes
+ // If both the left and right tables contain total_byte_size statistics,
+ // use `total_byte_size` to determine `should_swap_join_order`, else use `num_rows`
+ let left_stats = left.partition_statistics(None)?;
+ let right_stats = right.partition_statistics(None)?;
+ // First compare `total_byte_size` of left and right side,
+ // if information in this field is insufficient fallback to the `num_rows`
+ match (
+ left_stats.total_byte_size.get_value(),
+ right_stats.total_byte_size.get_value(),
+ ) {
+ (Some(l), Some(r)) => Ok(l > r),
+ _ => match (
+ left_stats.num_rows.get_value(),
+ right_stats.num_rows.get_value(),
+ ) {
+ (Some(l), Some(r)) => Ok(l > r),
+ _ => Ok(false),
+ },
+ }
+ }
+}
+
+impl PhysicalOptimizerRule for SelectJoinRule {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &datafusion::config::ConfigOptions,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ let result = plan.transform_up(|node| {
Review Comment:
Should not get triggered if disabled in configuration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [WIP] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4552853344
## Q72 (SF1) Fails
```
AdaptiveDatafusionExec: is_final=false, plan_id=2, stage_id=pending, stage_resolved=false
SortPreservingMergeExec: [total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], fetch=100
SortExec: TopK(fetch=100), expr=[total_cnt@5 DESC, i_item_desc@0 ASC NULLS LAST, w_warehouse_name@1 ASC NULLS LAST, d_week_seq@2 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq, sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END)@3 as no_promo, sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END)@4 as promo, count(Int64(1))@5 as total_cnt]
AggregateExec: mode=FinalPartitioned, gby=[i_item_desc@0 as i_item_desc, w_warehouse_name@1 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
ExchangeExec: partitioning=Hash([i_item_desc@0, w_warehouse_name@1, d_week_seq@2], 14), plan_id=4, stage_id=3, stage_resolved=false
AggregateExec: mode=Partial, gby=[i_item_desc@1 as i_item_desc, w_warehouse_name@0 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Left, on=[(cs_item_sk@0, cr_item_sk@0), (cs_order_number@1, cr_order_number@1)], projection=[w_warehouse_name@2, i_item_desc@3, d_week_seq@4, p_promo_sk@5]
ProjectionExec: expr=[cs_item_sk@1 as cs_item_sk, cs_order_number@2 as cs_order_number, w_warehouse_name@3 as w_warehouse_name, i_item_desc@4 as i_item_desc, d_week_seq@5 as d_week_seq, p_promo_sk@0 as p_promo_sk]
HashJoinExec: mode=CollectLeft, join_type=Right, on=[(p_promo_sk@0, cs_promo_sk@1)], projection=[p_promo_sk@0, cs_item_sk@1, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@6]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/promotion.parquet]]}, projection=[p_promo_sk], file_type=parquet
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_ship_date_sk@0, d_date_sk@0)], filter=d_date@1 > d_date@0 + IntervalMonthDayNano { months: 0, days: 5, nanoseconds: 0 }, projection=[cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@7]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(inv_date_sk@4, d_date_sk@0), (d_week_seq@8, d_week_seq@1)], projection=[cs_ship_date_sk@0, cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@5, i_item_desc@6, d_date@7, d_week_seq@8]
ProjectionExec: expr=[cs_ship_date_sk@2 as cs_ship_date_sk, cs_item_sk@3 as cs_item_sk, cs_promo_sk@4 as cs_promo_sk, cs_order_number@5 as cs_order_number, inv_date_sk@6 as inv_date_sk, w_warehouse_name@7 as w_warehouse_name, i_item_desc@8 as i_item_desc, d_date@0 as d_date, d_week_seq@1 as d_week_seq]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, cs_sold_date_sk@0)], projection=[d_date@1, d_week_seq@2, cs_ship_date_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
FilterExec: d_year@3 = 1999, projection=[d_date_sk@0, d_date@1, d_week_seq@2]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date, d_week_seq, d_year], file_type=parquet, predicate=d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999, pruning_predicate=d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1, required_guarantees=[d_year in (1999)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(hd_demo_sk@0, cs_bill_hdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
FilterExec: hd_buy_potential@1 = 501-1000, projection=[hd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/household_demographics.parquet]]}, projection=[hd_demo_sk, hd_buy_potential], file_type=parquet, predicate=hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000, pruning_predicate=hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1, required_guarantees=[hd_buy_potential in (501-1000)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_bill_cdemo_sk@2, cd_demo_sk@0)], projection=[cs_sold_date_sk@0, cs_ship_date_sk@1, cs_bill_hdemo_sk@3, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(i_item_sk@0, cs_item_sk@4)], projection=[cs_sold_date_sk@2, cs_ship_date_sk@3, cs_bill_cdemo_sk@4, cs_bill_hdemo_sk@5, cs_item_sk@6, cs_promo_sk@7, cs_order_number@8, inv_date_sk@9, w_warehouse_name@10, i_item_desc@1]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/item.parquet]]}, projection=[i_item_sk, i_item_desc], file_type=parquet, predicate=DynamicFilter [ empty ]
ExchangeExec: partitioning=None, plan_id=3, stage_id=2, stage_resolved=true, broadcast=true
ProjectionExec: expr=[cs_sold_date_sk@1 as cs_sold_date_sk, cs_ship_date_sk@2 as cs_ship_date_sk, cs_bill_cdemo_sk@3 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@4 as cs_bill_hdemo_sk, cs_item_sk@5 as cs_item_sk, cs_promo_sk@6 as cs_promo_sk, cs_order_number@7 as cs_order_number, inv_date_sk@8 as inv_date_sk, w_warehouse_name@0 as w_warehouse_name]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(w_warehouse_sk@0, inv_warehouse_sk@8)], projection=[w_warehouse_name@1, cs_sold_date_sk@2, cs_ship_date_sk@3, cs_bill_cdemo_sk@4, cs_bill_hdemo_sk@5, cs_item_sk@6, cs_promo_sk@7, cs_order_number@8, inv_date_sk@9]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/warehouse.parquet]]}, projection=[w_warehouse_sk, w_warehouse_name], file_type=parquet
ProjectionExec: expr=[cs_sold_date_sk@0 as cs_sold_date_sk, cs_ship_date_sk@1 as cs_ship_date_sk, cs_bill_cdemo_sk@2 as cs_bill_cdemo_sk, cs_bill_hdemo_sk@3 as cs_bill_hdemo_sk, cs_item_sk@4 as cs_item_sk, cs_promo_sk@5 as cs_promo_sk, cs_order_number@6 as cs_order_number, inv_date_sk@8 as inv_date_sk, inv_warehouse_sk@10 as inv_warehouse_sk]
SortMergeJoinExec: join_type=Inner, on=[(cs_item_sk@4, inv_item_sk@1)], filter=inv_quantity_on_hand@1 < cs_quantity@0
SortExec: expr=[cs_item_sk@4 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([cs_item_sk@4], 14), plan_id=0, stage_id=0, stage_resolved=true
DataSourceExec: file_groups={14 groups: [[Users/marko/TMP/tpcds_data/1g_parquet/catalog_sales.parquet:0..7244714], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_sales.parquet:7244714..14489428], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_sales.parquet:14489428..21734142], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_sales.parquet:21734142..28978856], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_sales.parquet:28978856..36223570], ...]}, projection=[cs_sold_date_sk, cs_ship_date_sk, cs_bill_cdemo_sk, cs_bill_hdemo_sk, cs_item_sk, cs_promo_sk, cs_order_number, cs_quantity], file_type=parquet
SortExec: expr=[inv_item_sk@1 ASC], preserve_partitioning=[true]
ExchangeExec: partitioning=Hash([inv_item_sk@1], 14), plan_id=1, stage_id=1, stage_resolved=true
DataSourceExec: file_groups={14 groups: [[Users/marko/TMP/tpcds_data/1g_parquet/inventory.parquet:0..2725031], [Users/marko/TMP/tpcds_data/1g_parquet/inventory.parquet:2725031..5450062], [Users/marko/TMP/tpcds_data/1g_parquet/inventory.parquet:5450062..8175093], [Users/marko/TMP/tpcds_data/1g_parquet/inventory.parquet:8175093..10900124], [Users/marko/TMP/tpcds_data/1g_parquet/inventory.parquet:10900124..13625155], ...]}, projection=[inv_date_sk, inv_item_sk, inv_warehouse_sk, inv_quantity_on_hand], file_type=parquet
FilterExec: cd_marital_status@1 = S, projection=[cd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/customer_demographics.parquet]]}, projection=[cd_demo_sk, cd_marital_status], file_type=parquet, predicate=cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND DynamicFilter [ empty ], pruning_predicate=cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1, required_guarantees=[cd_marital_status in (S)]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_week_seq], file_type=parquet, predicate=DynamicFilter [ empty ]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date], file_type=parquet, predicate=DynamicFilter [ empty ]
DataSourceExec: file_groups={14 groups: [[Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:0..806517], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:806517..1613034], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:1613034..2419551], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:2419551..3226068], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:3226068..4032585], ...]}, projection=[cr_item_sk, cr_order_number], file_type=parquet
```
stage 3 fails with (all running tasks): `DataFusionError(Shared(Shared(Shared(Shared(ArrowError(OffsetOverflowError(2147506937), Some(\"\")))))))\n")), None)`
```
SortShuffleWriterExec: partitioning=Hash([i_item_desc@0, w_warehouse_name@1, d_week_seq@2], 14)
AggregateExec: mode=Partial, gby=[i_item_desc@1 as i_item_desc, w_warehouse_name@0 as w_warehouse_name, d_week_seq@2 as d_week_seq], aggr=[sum(CASE WHEN promotion.p_promo_sk IS NULL THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN promotion.p_promo_sk IS NOT NULL THEN Int64(1) ELSE Int64(0) END), count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Left, on=[(cs_item_sk@0, cr_item_sk@0), (cs_order_number@1, cr_order_number@1)], projection=[w_warehouse_name@2, i_item_desc@3, d_week_seq@4, p_promo_sk@5]
ProjectionExec: expr=[cs_item_sk@1 as cs_item_sk, cs_order_number@2 as cs_order_number, w_warehouse_name@3 as w_warehouse_name, i_item_desc@4 as i_item_desc, d_week_seq@5 as d_week_seq, p_promo_sk@0 as p_promo_sk]
HashJoinExec: mode=CollectLeft, join_type=Right, on=[(p_promo_sk@0, cs_promo_sk@1)], projection=[p_promo_sk@0, cs_item_sk@1, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@6]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/promotion.parquet]]}, projection=[p_promo_sk], file_type=parquet
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_ship_date_sk@0, d_date_sk@0)], filter=d_date@1 > d_date@0 + IntervalMonthDayNano { months: 0, days: 5, nanoseconds: 0 }, projection=[cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@4, i_item_desc@5, d_week_seq@7]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(inv_date_sk@4, d_date_sk@0), (d_week_seq@8, d_week_seq@1)], projection=[cs_ship_date_sk@0, cs_item_sk@1, cs_promo_sk@2, cs_order_number@3, w_warehouse_name@5, i_item_desc@6, d_date@7, d_week_seq@8]
ProjectionExec: expr=[cs_ship_date_sk@2 as cs_ship_date_sk, cs_item_sk@3 as cs_item_sk, cs_promo_sk@4 as cs_promo_sk, cs_order_number@5 as cs_order_number, inv_date_sk@6 as inv_date_sk, w_warehouse_name@7 as w_warehouse_name, i_item_desc@8 as i_item_desc, d_date@0 as d_date, d_week_seq@1 as d_week_seq]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, cs_sold_date_sk@0)], projection=[d_date@1, d_week_seq@2, cs_ship_date_sk@4, cs_item_sk@5, cs_promo_sk@6, cs_order_number@7, inv_date_sk@8, w_warehouse_name@9, i_item_desc@10]
FilterExec: d_year@3 = 1999, projection=[d_date_sk@0, d_date@1, d_week_seq@2]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date, d_week_seq, d_year], file_type=parquet, predicate=d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999 AND d_year@6 = 1999, pruning_predicate=d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1 AND d_year_null_count@2 != row_count@3 AND d_year_min@0 <= 1999 AND 1999 <= d_year_max@1, required_guarantees=[d_year in (1999)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(hd_demo_sk@0, cs_bill_hdemo_sk@2)], projection=[cs_sold_date_sk@1, cs_ship_date_sk@2, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
FilterExec: hd_buy_potential@1 = 501-1000, projection=[hd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/household_demographics.parquet]]}, projection=[hd_demo_sk, hd_buy_potential], file_type=parquet, predicate=hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000 AND hd_buy_potential@2 = 501-1000, pruning_predicate=hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1 AND hd_buy_potential_null_count@2 != row_count@3 AND hd_buy_potential_min@0 <= 501-1000 AND 501-1000 <= hd_buy_potential_max@1, required_guarantees=[hd_buy_potential in (501-1000)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(cs_bill_cdemo_sk@2, cd_demo_sk@0)], projection=[cs_sold_date_sk@0, cs_ship_date_sk@1, cs_bill_hdemo_sk@3, cs_item_sk@4, cs_promo_sk@5, cs_order_number@6, inv_date_sk@7, w_warehouse_name@8, i_item_desc@9]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(i_item_sk@0, cs_item_sk@4)], projection=[cs_sold_date_sk@2, cs_ship_date_sk@3, cs_bill_cdemo_sk@4, cs_bill_hdemo_sk@5, cs_item_sk@6, cs_promo_sk@7, cs_order_number@8, inv_date_sk@9, w_warehouse_name@10, i_item_desc@1]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/item.parquet]]}, projection=[i_item_sk, i_item_desc], file_type=parquet, predicate=DynamicFilter [ empty ]
ShuffleReaderExec: upstream_stage: 2, broadcast: true, upstream_partition_count: 14
FilterExec: cd_marital_status@1 = S, projection=[cd_demo_sk@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/customer_demographics.parquet]]}, projection=[cd_demo_sk, cd_marital_status], file_type=parquet, predicate=cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND cd_marital_status@2 = S AND DynamicFilter [ empty ], pruning_predicate=cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1 AND cd_marital_status_null_count@2 != row_count@3 AND cd_marital_status_min@0 <= S AND S <= cd_marital_status_max@1, required_guarantees=[cd_marital_status in (S)]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_week_seq], file_type=parquet, predicate=DynamicFilter [ empty ]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpcds_data/1g_parquet/date_dim.parquet]]}, projection=[d_date_sk, d_date], file_type=parquet, predicate=DynamicFilter [ empty ]
DataSourceExec: file_groups={14 groups: [[Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:0..806517], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:806517..1613034], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:1613034..2419551], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:2419551..3226068], [Users/marko/TMP/tpcds_data/1g_parquet/catalog_returns.parquet:3226068..4032585], ...]}, projection=[cr_item_sk, cr_order_number], file_type=parquet
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3313839243
##########
ballista/scheduler/src/state/aqe/planner.rs:
##########
@@ -99,7 +107,8 @@ impl AdaptivePlanner {
let plan = planner.optimize_physical_plan(plan, &session_state, |_, _| {})?;
Ok(Self {
- stage_id_generator: 0,
+ stage_id_generator: 0, // FIXME: compatibility issue with static where stages start from 1
Review Comment:
For follow up once this get merged
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,537 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functional tests for [`CoalescePartitionsRule`]: drive a query through
+//! `AdaptivePlanner`, finalize the upstream stage with synthetic per-partition
+//! byte stats, and snapshot the displayed plan tree so the rule's effect on
+//! the leaf `ExchangeExec` is visible at the `coalesce=K of M` field.
+//!
+//! Each test uses small synthetic byte sizes paired with a small
+//! `coalesce_target_partition_bytes` so the bin-pack outcome is hand-traceable
+//! against `split_size_list_by_target_size`.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
Review Comment:
This action gets triggered when both sides get repartitioned, does it make sense to change from partitioned to collect left join at that point ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3323465973
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,537 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum JoinInputState {
+ /// All inputs has been repartitioned
+ /// which means this join can be resolved
+ Repartitioned,
+ /// State on join inputs is unknown
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: JoinInputState,
+ pub null_aware: bool,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ null_aware: self.null_aware,
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, JoinInputState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, JoinInputState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (JoinInputState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ (JoinInputState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ (JoinInputState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
Review Comment:
i find `if after match` easier to look at and understand
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4529134363
Partitions get messed up in the last stage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4528967200
### Q2 (Static Plan)
```text
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Job 0buY2Mw physical plan:
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
SortPreservingMergeExec: [s_acctbal@0 DESC, n_name@2 ASC NULLS LAST, s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], fetch=100
SortExec: TopK(fetch=100), expr=[s_acctbal@0 DESC, n_name@2 ASC NULLS LAST, s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[s_acctbal@5 as s_acctbal, s_name@2 as s_name, n_name@7 as n_name, p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_address@3 as s_address, s_phone@4 as s_phone, s_comment@6 as s_comment]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(p_partkey@0, ps_partkey@1), (ps_supplycost@7, min(partsupp.ps_supplycost)@0)], projection=[p_partkey@0, p_mfgr@1, s_name@2, s_address@3, s_phone@4, s_acctbal@5, s_comment@6, n_name@8]
CoalescePartitionsExec
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0, n_regionkey@9)], projection=[p_partkey@1, p_mfgr@2, s_name@3, s_address@4, s_phone@5, s_acctbal@6, s_comment@7, ps_supplycost@8, n_name@9]
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/region/part-0.parquet]]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)]
ProjectionExec: expr=[p_partkey@2 as p_partkey, p_mfgr@3 as p_mfgr, s_name@4 as s_name, s_address@5 as s_address, s_phone@6 as s_phone, s_acctbal@7 as s_acctbal, s_comment@8 as s_comment, ps_supplycost@9 as ps_supplycost, n_name@0 as n_name, n_regionkey@1 as n_regionkey]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@4)], projection=[n_name@1, n_regionkey@2, p_partkey@3, p_mfgr@4, s_name@5, s_address@6, s_phone@8, s_acctbal@9, s_comment@10, ps_supplycost@11]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/nation/part-0.parquet]]}, projection=[n_nationkey, n_name, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ]
ProjectionExec: expr=[p_partkey@6 as p_partkey, p_mfgr@7 as p_mfgr, s_name@0 as s_name, s_address@1 as s_address, s_nationkey@2 as s_nationkey, s_phone@3 as s_phone, s_acctbal@4 as s_acctbal, s_comment@5 as s_comment, ps_supplycost@8 as ps_supplycost]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_suppkey@0, ps_suppkey@2)], projection=[s_name@1, s_address@2, s_nationkey@3, s_phone@4, s_acctbal@5, s_comment@6, p_partkey@7, p_mfgr@8, ps_supplycost@10]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/supplier/part-0.parquet]]}, projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment], file_type=parquet, predicate=DynamicFilter [ empty ]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(p_partkey@0, ps_partkey@0)], projection=[p_partkey@0, p_mfgr@1, ps_suppkey@3, ps_supplycost@4]
CoalescePartitionsExec
FilterExec: p_size@3 = 15 AND p_type@2 LIKE %BRASS, projection=[p_partkey@0, p_mfgr@1]
DataSourceExec: file_groups={7 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-0.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-1.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-10.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-11.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-12.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-13.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-2.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-3.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-4.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-5.parquet], ...]}, projection=[p_partkey, p_mfgr, p_type, p_size], file_type=parquet, predicate=p_size@5 = 15 AND p_type@4 LIKE %BRASS, pruning_predicate=p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1, required_guarantees=[p_size in (15)]
DataSourceExec: file_groups={8 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-0.parquet:0..2385925, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:0..1749648], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:1749648..2367980, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-10.parquet:0..2354492, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:0..1162749], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:1162749..2356504, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-12.parquet:0..2355805, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:0..586013], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:586013..2352367, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:0..2369219], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:2369219..2369944, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-3.parq
uet:0..2369786, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-4.parquet:0..1765062], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ProjectionExec: expr=[min(partsupp.ps_supplycost)@1 as min(partsupp.ps_supplycost), ps_partkey@0 as ps_partkey]
AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)]
RepartitionExec: partitioning=Hash([ps_partkey@0], 8), input_partitions=8
AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0, n_regionkey@2)], projection=[ps_partkey@1, ps_supplycost@2]
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/region/part-0.parquet]]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)]
ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as ps_supplycost, n_regionkey@0 as n_regionkey]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@2)], projection=[n_regionkey@1, ps_partkey@2, ps_supplycost@3]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/nation/part-0.parquet]]}, projection=[n_nationkey, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ]
ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as ps_supplycost, s_nationkey@0 as s_nationkey]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_suppkey@0, ps_suppkey@1)], projection=[s_nationkey@1, ps_partkey@2, ps_supplycost@4]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/supplier/part-0.parquet]]}, projection=[s_suppkey, s_nationkey], file_type=parquet, predicate=DynamicFilter [ empty ]
DataSourceExec: file_groups={8 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-0.parquet:0..2385925, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:0..1749648], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:1749648..2367980, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-10.parquet:0..2354492, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:0..1162749], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:1162749..2356504, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-12.parquet:0..2355805, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:0..586013], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:586013..2352367, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:0..2369219], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:2369219..2369944, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-3.
parquet:0..2369786, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-4.parquet:0..1765062], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet, predicate=DynamicFilter [ empty ]
```
### Q2 (AQE)
```text
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Job V3vvoKt physical plan:
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
AdaptiveDatafusionExec: is_final=true, plan_id=3, stage_id=6, stage_resolved=true
SortExec: TopK(fetch=100), expr=[s_acctbal@0 DESC, n_name@2 ASC NULLS LAST, s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], preserve_partitioning=[false]
ProjectionExec: expr=[s_acctbal@5 as s_acctbal, s_name@2 as s_name, n_name@7 as n_name, p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_address@3 as s_address, s_phone@4 as s_phone, s_comment@6 as s_comment]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(p_partkey@0, ps_partkey@1), (ps_supplycost@7, min(partsupp.ps_supplycost)@0)], projection=[p_partkey@0, p_mfgr@1, s_name@2, s_address@3, s_phone@4, s_acctbal@5, s_comment@6, n_name@8]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0, n_regionkey@9)], projection=[p_partkey@1, p_mfgr@2, s_name@3, s_address@4, s_phone@5, s_acctbal@6, s_comment@7, ps_supplycost@8, n_name@9]
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/region/part-0.parquet]]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND
EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)]
ProjectionExec: expr=[p_partkey@2 as p_partkey, p_mfgr@3 as p_mfgr, s_name@4 as s_name, s_address@5 as s_address, s_phone@6 as s_phone, s_acctbal@7 as s_acctbal, s_comment@8 as s_comment, ps_supplycost@9 as ps_supplycost, n_name@0 as n_name, n_regionkey@1 as n_regionkey]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@4)], projection=[n_name@1, n_regionkey@2, p_partkey@3, p_mfgr@4, s_name@5, s_address@6, s_phone@8, s_acctbal@9, s_comment@10, ps_supplycost@11]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/nation/part-0.parquet]]}, projection=[n_nationkey, n_name, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@3 as s_name, s_address@4 as s_address, s_nationkey@5 as s_nationkey, s_phone@6 as s_phone, s_acctbal@7 as s_acctbal, s_comment@8 as s_comment, ps_supplycost@2 as ps_supplycost]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(ps_suppkey@2, s_suppkey@0)], projection=[p_partkey@0, p_mfgr@1, ps_supplycost@3, s_name@5, s_address@6, s_nationkey@7, s_phone@8, s_acctbal@9, s_comment@10]
ExchangeExec: partitioning=None, plan_id=4, stage_id=3, stage_resolved=true, broadcast=true
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(p_partkey@0, ps_partkey@0)], projection=[p_partkey@0, p_mfgr@1, ps_suppkey@3, ps_supplycost@4]
ExchangeExec: partitioning=None, plan_id=0, stage_id=0, stage_resolved=true, broadcast=true
FilterExec: p_size@3 = 15 AND p_type@2 LIKE %BRASS, projection=[p_partkey@0, p_mfgr@1]
DataSourceExec: file_groups={7 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-0.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-1.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-10.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-11.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-12.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-13.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-2.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-3.parquet], [Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-4.parquet, Users/marko/TMP/tpch_data/tpch-data-sf1/part/part-5.parquet], ...]}, projection=[p_partkey, p_mfgr, p_type, p_size], file_type=parquet, predicate=p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size
@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS AND p_size@5 = 15 AND p_type@4 LIKE %BRASS, pruning_predicate=p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1 AND p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1, required_guarantees=[p_size in (15)]
DataSourceExec: file_groups={8 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-0.parquet:0..2385925, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:0..1749648], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:1749648..2367980, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-10.parquet:0..2354492, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:0..1162749], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:1162749..2356504, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-12.parquet:0..2355805, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:0..586013], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:586013..2352367, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:0..2369219], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:2369219..2369944, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-3.parq
uet:0..2369786, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-4.parquet:0..1765062], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/supplier/part-0.parquet]]}, projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ExchangeExec: partitioning=None, plan_id=5, stage_id=5, stage_resolved=true, broadcast=true
ProjectionExec: expr=[min(partsupp.ps_supplycost)@1 as min(partsupp.ps_supplycost), ps_partkey@0 as ps_partkey]
AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)]
ExchangeExec: partitioning=Hash([ps_partkey@0], 8), plan_id=6, stage_id=4, stage_resolved=true
AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0, n_regionkey@2)], projection=[ps_partkey@1, ps_supplycost@2]
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/region/part-0.parquet]]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE AND r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <=
EUROPE AND EUROPE <= r_name_max@1 AND r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)]
ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as ps_supplycost, n_regionkey@0 as n_regionkey]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@2)], projection=[n_regionkey@1, ps_partkey@2, ps_supplycost@3]
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/nation/part-0.parquet]]}, projection=[n_nationkey, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ]
ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as ps_supplycost, s_nationkey@0 as s_nationkey]
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(s_suppkey@0, ps_suppkey@1)], projection=[s_nationkey@1, ps_partkey@2, ps_supplycost@4]
ExchangeExec: partitioning=Hash([s_suppkey@0], 8), plan_id=2, stage_id=2, stage_resolved=true
DataSourceExec: file_groups={1 group: [[Users/marko/TMP/tpch_data/tpch-data-sf1/supplier/part-0.parquet]]}, projection=[s_suppkey, s_nationkey], file_type=parquet
ExchangeExec: partitioning=Hash([ps_suppkey@1], 8), plan_id=1, stage_id=1, stage_resolved=true
DataSourceExec: file_groups={8 groups: [[Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-0.parquet:0..2385925, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:0..1749648], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-1.parquet:1749648..2367980, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-10.parquet:0..2354492, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:0..1162749], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-11.parquet:1162749..2356504, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-12.parquet:0..2355805, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:0..586013], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-13.parquet:586013..2352367, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:0..2369219], [Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-2.parquet:2369219..2369944, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/par
t-3.parquet:0..2369786, Users/marko/TMP/tpch_data/tpch-data-sf1/partsupp/part-4.parquet:0..1765062], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3318149330
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,528 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChildrenState {
+ Repartitioned,
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: ChildrenState,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, ChildrenState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_rows,
Review Comment:
good catch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3318133726
##########
ballista/scheduler/src/state/aqe/execution_plan/exchange.rs:
##########
@@ -98,8 +103,38 @@ impl ExchangeExec {
plan_id,
Arc::new(AtomicI64::new(-1)),
Arc::new(Mutex::new(None)),
+ false,
)
}
+ /// new broadcast exchange
+ pub fn new_broadcast(
+ input: Arc<dyn ExecutionPlan>,
+ partitioning: Option<Partitioning>,
+ plan_id: usize,
+ ) -> Self {
+ Self::new_with_details(
+ input,
+ partitioning,
+ plan_id,
+ Arc::new(AtomicI64::new(-1)),
+ Arc::new(Mutex::new(None)),
+ true,
+ )
+ }
+
+ pub fn to_broadcast(&self, plan_id: usize) -> Self {
+ Self {
+ input: self.input.clone(),
+ properties: self.properties.clone(),
Review Comment:
make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "sandugood (via GitHub)" <gi...@apache.org>.
sandugood commented on code in PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#discussion_r3323398109
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,537 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum JoinInputState {
+ /// All inputs has been repartitioned
+ /// which means this join can be resolved
+ Repartitioned,
+ /// State on join inputs is unknown
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: JoinInputState,
+ pub null_aware: bool,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
Review Comment:
nit: I think using simply `Arc<>` is suitable, since you've brought it into scope (guess it can be changed everywhere in the `impl`)
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -0,0 +1,537 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::{
+ arrow::compute::SortOptions,
+ common::{JoinType, NullEquality, Result, exec_err, internal_err},
+ physical_expr_common::physical_expr::fmt_sql,
+ physical_plan::{
+ DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+ joins::{
+ HashJoinExec, HashJoinExecBuilder, JoinOn, PartitionMode, SortMergeJoinExec,
+ utils::JoinFilter,
+ },
+ },
+};
+use std::sync::Arc;
+
+use crate::state::aqe::execution_plan::ExchangeExec;
+
+/// has children of this join been
+/// repartitioned
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum JoinInputState {
+ /// All inputs has been repartitioned
+ /// which means this join can be resolved
+ Repartitioned,
+ /// State on join inputs is unknown
+ Unknown,
+}
+
+// SortMergeJoinExec::try_new(left, right, on, filter, join_type, sort_options, null_equality )
+// HashJoinExec::try_new (left, right, on, filter, join_type, projection, partition_mode, null_equality, null_aware )
+#[derive(Debug)]
+pub struct DynamicJoinSelectionExec {
+ pub left: Arc<dyn ExecutionPlan>,
+ pub right: Arc<dyn ExecutionPlan>,
+ pub on: JoinOn,
+ pub filter: Option<JoinFilter>,
+ pub join_type: JoinType,
+ pub projection: Option<Vec<usize>>,
+ pub null_equality: NullEquality,
+ pub properties: Arc<PlanProperties>,
+ pub selection_state: JoinInputState,
+ pub null_aware: bool,
+}
+
+impl ExecutionPlan for DynamicJoinSelectionExec {
+ fn name(&self) -> &str {
+ "DynamicJoinSelectionExec"
+ }
+
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn properties(&self) -> &std::sync::Arc<datafusion::physical_plan::PlanProperties> {
+ &self.properties
+ }
+
+ fn children(&self) -> Vec<&std::sync::Arc<dyn ExecutionPlan>> {
+ vec![&self.left, &self.right]
+ }
+
+ fn with_new_children(
+ self: std::sync::Arc<Self>,
+ children: Vec<std::sync::Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<std::sync::Arc<dyn ExecutionPlan>> {
+ if children.len() != 2 {
+ return internal_err!(
+ "DynamicJoinSelectionExec expects 2 children, got {}",
+ children.len()
+ );
+ }
+ Ok(Arc::new(DynamicJoinSelectionExec {
+ left: Arc::clone(&children[0]),
+ right: Arc::clone(&children[1]),
+ on: self.on.clone(),
+ filter: self.filter.clone(),
+ join_type: self.join_type,
+ projection: self.projection.clone(),
+ null_equality: self.null_equality,
+ properties: Arc::clone(&self.properties),
+ selection_state: self.selection_state.clone(),
+ null_aware: self.null_aware,
+ }))
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: std::sync::Arc<datafusion::execution::TaskContext>,
+ ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
+ exec_err!(
+ "Operator should not be executed; it should have been replaced by an optimizer rule."
+ )
+ }
+}
+
+impl DisplayAs for DynamicJoinSelectionExec {
+ fn fmt_as(
+ &self,
+ t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| format!("({c1}, {c2})"))
+ .collect::<Vec<String>>()
+ .join(", ");
+ let display_null_equality =
+ if self.null_equality == NullEquality::NullEqualsNull {
+ ", NullsEqual: true"
+ } else {
+ ""
+ };
+ write!(
+ f,
+ "{}: join_type={:?}, on=[{}]{}{}",
+ Self::static_name(),
+ self.join_type,
+ on,
+ self.filter.as_ref().map_or_else(
+ || "".to_string(),
+ |f| format!(", filter={}", f.expression())
+ ),
+ display_null_equality,
+ )?;
+
+ write!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, JoinInputState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ DisplayFormatType::TreeRender => {
+ let on = self
+ .on
+ .iter()
+ .map(|(c1, c2)| {
+ format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+ })
+ .collect::<Vec<String>>()
+ .join(", ");
+
+ if self.join_type != JoinType::Inner {
+ writeln!(f, "join_type={:?}", self.join_type)?;
+ }
+ writeln!(f, "on={on}")?;
+
+ if self.null_equality == NullEquality::NullEqualsNull {
+ writeln!(f, "NullsEqual: true")?;
+ }
+
+ writeln!(
+ f,
+ " repartitioned={}",
+ matches!(self.selection_state, JoinInputState::Repartitioned)
+ )?;
+
+ Ok(())
+ }
+ }
+ }
+}
+
+pub enum JoinSelectionAction {
+ Repartition(Arc<DynamicJoinSelectionExec>),
+ CollectLeft(Arc<HashJoinExec>),
+ LateCollectLeft(Arc<HashJoinExec>),
+ Hash(Arc<HashJoinExec>),
+ Sort(Arc<SortMergeJoinExec>),
+}
+
+impl DynamicJoinSelectionExec {
+ // this is required, as we do not want to implement
+ // ExecutionPlan::required_input_distribution as other
+ // datafusion planners are going to add repartition
+ pub(crate) fn _required_input_distribution(&self) -> Vec<Distribution> {
+ let (left_expr, right_expr) = self
+ .on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ vec![
+ Distribution::HashPartitioned(left_expr),
+ Distribution::HashPartitioned(right_expr),
+ ]
+ }
+
+ pub(crate) fn to_actual_join(
+ &self,
+ config: &datafusion::config::ConfigOptions,
+ ) -> Result<JoinSelectionAction> {
+ let prefer_hash_join = config.optimizer.prefer_hash_join;
+ let threshold_collect_left_join_bytes =
+ config.optimizer.hash_join_single_partition_threshold;
+ let threshold_collect_left_join_rows =
+ config.optimizer.hash_join_single_partition_threshold_rows;
+
+ let partition_mode = if Self::supports_collect_by_thresholds(
+ self.left.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) || Self::supports_collect_by_thresholds(
+ self.right.as_ref(),
+ threshold_collect_left_join_bytes,
+ threshold_collect_left_join_rows,
+ ) {
+ PartitionMode::CollectLeft
+ } else {
+ PartitionMode::Partitioned
+ };
+ match (&self.selection_state, partition_mode) {
+ (JoinInputState::Unknown, PartitionMode::CollectLeft) => self
+ .to_hash_join(PartitionMode::CollectLeft)
+ .map(JoinSelectionAction::CollectLeft),
+ (JoinInputState::Repartitioned, PartitionMode::Partitioned)
+ if prefer_hash_join =>
+ {
+ self.to_hash_join(PartitionMode::Partitioned)
+ .map(JoinSelectionAction::Hash)
+ }
+ (JoinInputState::Repartitioned, PartitionMode::Partitioned) => {
+ self.to_sort_merge_join().map(JoinSelectionAction::Sort)
+ }
Review Comment:
Maybe we can collapse it into one branch
```suggestion
(JoinInputState::Repartitioned, PartitionMode::Partitioned) => {
if prefer_hash_join {
self.to_hash_join(PartitionMode::Partitioned)
.map(JoinSelectionAction::Hash)
} else {
self.to_sort_merge_join().map(JoinSelectionAction::Sort)
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(aqe): delay join decision & introduce broadcast join in AQE [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #1752:
URL: https://github.com/apache/datafusion-ballista/pull/1752#issuecomment-4586109683
Empty rule needs some work, there are some changes in #1723 which I hope will help
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org