You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/11/15 10:30:09 UTC
[GitHub] [arrow-datafusion] mingmwang opened a new pull request, #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
mingmwang opened a new pull request, #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #4139.
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
The PR covers below changes,
1. Rename Rule `HashBuildProbeOrder` to `JoinSelection`.
2. Expand the rule's capabilities and make a cost based decision to select which `PartitionMode` mode(`Partitioned`/`CollectLeft`) is optimal for `HashJoin`
3. Add a session level configuration `prefer_hash_join` to prefer `HashJoin` or `SortMergeJoin`, document SortMergeJoin support currently is experimental.
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
4. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029124355
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
Review Comment:
Sure, will change it.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1023841101
##########
datafusion/core/src/physical_plan/planner.rs:
##########
@@ -932,15 +933,41 @@ impl DefaultPhysicalPlanner {
if session_state.config.target_partitions > 1
&& session_state.config.repartition_joins
+ && !session_state.config.prefer_hash_join
{
- // Use hash partition by default to parallelize hash joins
+ // Use SortMergeJoin if hash join is not preferred
+ // Sort-Merge join support currently is experimental
+ if join_filter.is_some() {
+ // TODO SortMergeJoinExec need to support join filter
+ Err(DataFusionError::Plan("SortMergeJoinExec does not support join_filter now.".to_string()))
Review Comment:
Sure, will do.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029130930
##########
datafusion/core/src/physical_plan/joins/mod.rs:
##########
@@ -29,6 +29,9 @@ pub enum PartitionMode {
Partitioned,
/// Left side will collected into one partition
CollectLeft,
+ /// When set to Auto, DataFusion optimizer will decide which PartitionMode mode(Partitioned/CollectLeft) is optimal based on statistics.
+ /// It will also consider swapping the left and right inputs for the Join
Review Comment:
You are right. Especially when the mode is CollectLeft, some join types is unable to run CollectLeft mode.
https://github.com/apache/arrow-datafusion/blob/d5d2de3362649db85ad54161ee28d9374ed3437c/datafusion/core/src/physical_optimizer/join_selection.rs#L284-L286
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028797502
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
Review Comment:
Sure, will do.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028295969
##########
datafusion/core/src/execution/context.rs:
##########
@@ -1228,6 +1228,11 @@ pub struct SessionConfig {
pub collect_statistics: bool,
/// Should DataFusion optimizer run a top down process to reorder the join keys
pub top_down_join_key_reordering: bool,
+ /// Should DataFusion optimizer prefer HashJoin over SortMergeJoin.
+ /// HashJoin can work more efficently than SortMergeJoin but consumes more memory.
+ pub prefer_hash_join: bool,
Review Comment:
What would you think about moving these new settings into ConfigOptions (where they are visible via `SHOW` and automatically documented)?
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
Review Comment:
I like this design with `PartitionMode::Auto`
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
Review Comment:
is this TODO worth tracking with a ticket?
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
Review Comment:
the prestosql approach makes sense to me
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
+ let column_indices = filter
+ .column_indices()
+ .iter()
+ .map(|idx| {
+ let side = if matches!(idx.side, JoinSide::Left) {
+ JoinSide::Right
+ } else {
+ JoinSide::Left
+ };
+ ColumnIndex {
+ index: idx.index,
+ side,
+ }
+ })
+ .collect();
+
+ Some(JoinFilter::new(
+ filter.expression().clone(),
+ column_indices,
+ filter.schema().clone(),
+ ))
+ }
+ None => None,
+ }
+}
+
+impl PhysicalOptimizerRule for JoinSelection {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ session_config: &SessionConfig,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let collect_left_threshold = session_config.hash_join_collect_left_threshold;
+ plan.transform_up(&|plan| {
+ if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
+ match hash_join.partition_mode() {
+ PartitionMode::Auto => {
+ try_collect_left(hash_join, Some(collect_left_threshold))
+ .unwrap()
+ .or_else(|| Some(partitioned_hash_join(hash_join).unwrap()))
Review Comment:
Why does this code (and below) use `unwrap`? I think they should return an error rather than `panic`ing
##########
datafusion/core/src/execution/context.rs:
##########
@@ -1228,6 +1228,11 @@ pub struct SessionConfig {
pub collect_statistics: bool,
/// Should DataFusion optimizer run a top down process to reorder the join keys
pub top_down_join_key_reordering: bool,
+ /// Should DataFusion optimizer prefer HashJoin over SortMergeJoin.
+ /// HashJoin can work more efficently than SortMergeJoin but consumes more memory.
+ pub prefer_hash_join: bool,
+ /// The maximum estimated size in bytes for the left input a hash join will be collected into one partition
+ pub hash_join_collect_left_threshold: usize,
Review Comment:
I recommend calling this seetting something that doesn't have 'left' and 'right' as that can get confusing.
How about `hash_join_single_partition_threshold`?
##########
datafusion/core/src/physical_plan/joins/mod.rs:
##########
@@ -29,6 +29,9 @@ pub enum PartitionMode {
Partitioned,
/// Left side will collected into one partition
CollectLeft,
+ /// When set to Auto, DataFusion optimizer will decide which PartitionMode mode(Partitioned/CollectLeft) is optimal based on statistics.
+ /// It will also consider swapping the left and right inputs for the Join
Review Comment:
The optimizer also will swap the inputs for Partitioned and CollectLeft mode too, right? As written, this comment could be confusing and imply that inputs will only be swapped if the mode is set to Auto
##########
datafusion/core/src/execution/context.rs:
##########
@@ -1228,6 +1228,11 @@ pub struct SessionConfig {
pub collect_statistics: bool,
/// Should DataFusion optimizer run a top down process to reorder the join keys
pub top_down_join_key_reordering: bool,
+ /// Should DataFusion optimizer prefer HashJoin over SortMergeJoin.
+ /// HashJoin can work more efficently than SortMergeJoin but consumes more memory.
Review Comment:
```suggestion
/// HashJoin can work more efficiently than SortMergeJoin but consumes more memory. Defaults to true
```
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
Review Comment:
this comment seems incorrect for this module
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
Review Comment:
I am surprised we can swap the inputs to the various outer and semi join types without changing the query's output ๐ค
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
Review Comment:
Instead of `match filter` I think you can use `map` like:
```rust
filter.map(|filter| {
let column_indicies = ...
```
Not critical, I just figured I woudl point it out
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1030767000
##########
datafusion/core/tests/sql/joins.rs:
##########
@@ -1870,3 +1870,125 @@ async fn reduce_full_join_to_inner_join() -> Result<()> {
Ok(())
}
+
+#[tokio::test]
+async fn sort_merge_equijoin() -> Result<()> {
+ let ctx = create_sort_merge_join_context("t1_id", "t2_id")?;
+ let equivalent_sql = [
+ "SELECT t1_id, t1_name, t2_name FROM t1 JOIN t2 ON t1_id = t2_id ORDER BY t1_id",
+ "SELECT t1_id, t1_name, t2_name FROM t1 JOIN t2 ON t2_id = t1_id ORDER BY t1_id",
+ ];
+ let expected = vec![
+ "+-------+---------+---------+",
+ "| t1_id | t1_name | t2_name |",
+ "+-------+---------+---------+",
+ "| 11 | a | z |",
+ "| 22 | b | y |",
+ "| 44 | d | x |",
+ "+-------+---------+---------+",
+ ];
+ for sql in equivalent_sql.iter() {
+ let actual = execute_to_batches(&ctx, sql).await;
+ assert_batches_eq!(expected, &actual);
+ }
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn sort_merge_join_on_date32() -> Result<()> {
Review Comment:
๐
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028797376
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
Review Comment:
Actually I am not very sure about this, because our HashJoin implementation is quite different from PrestoSQL's. I think we need to do more benchmark 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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1030369663
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
Review Comment:
The `filter` is a ref here, can not call `map()` method, `map()` needs owned type.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb merged pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
alamb merged PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] isidentical commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
isidentical commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1325518946
I wasn't able to go through the last revision in detail but overall this looks great ๐ฏ A minor question of mine is still standing though (maybe let's do a follow up ticket on it): https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1023219190 (`collection_size_threshold` represents the size in both the number of rows and the number of bytes, which is a bit confusing because the distinction is only internal when one form of statistics is not available)
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] jackwener commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
jackwener commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1030978938
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
Review Comment:
Some databases choose to build one side, `TiDB` also is.
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,979 @@
+// 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.
+
+//! Select the proper PartitionMode and build side based on the avaliable statistics for hash join.
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::config::OPT_HASH_JOIN_SINGLE_PARTITION_THRESHOLD;
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
Review Comment:
we can put it into `JoinType`
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
mingmwang commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1316235523
There is some bug with HashJoin CollectLeft:
https://github.com/apache/arrow-datafusion/issues/4230
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] isidentical commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
isidentical commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1023215447
##########
datafusion/core/src/physical_plan/planner.rs:
##########
@@ -932,15 +933,41 @@ impl DefaultPhysicalPlanner {
if session_state.config.target_partitions > 1
&& session_state.config.repartition_joins
+ && !session_state.config.prefer_hash_join
{
- // Use hash partition by default to parallelize hash joins
+ // Use SortMergeJoin if hash join is not preferred
+ // Sort-Merge join support currently is experimental
+ if join_filter.is_some() {
+ // TODO SortMergeJoinExec need to support join filter
+ Err(DataFusionError::Plan("SortMergeJoinExec does not support join_filter now.".to_string()))
Review Comment:
```suggestion
Err(DataFusionError::NotImplemented("SortMergeJoinExec does not support join_filter now.".to_string()))
```
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -68,6 +75,19 @@ fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -
}
}
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: usize,
+) -> bool {
+ if let Some(size) = plan.statistics().total_byte_size {
+ size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count < collection_size_threshold
Review Comment:
Question: are we treating both bytes and rows equally here? Seems like `collection_size_threshold` has a unit of `bytes` compared to the `row_count` which specifies number of rows. I guess we can normalize it a bit if we want to pursue this (e.g. `collection_size_threshold / SOME_MAGIC_CONSTANT`) but otherwise it might be a bit off from a desirable scenerio.
##########
datafusion/core/src/physical_plan/joins/sort_merge_join.rs:
##########
@@ -348,7 +349,15 @@ impl ExecutionPlan for SortMergeJoinExec {
}
fn statistics(&self) -> Statistics {
- todo!()
+ // TODO stats: it is not possible in general to know the output size of joins
+ // There are some special cases though, for example:
+ // - `A LEFT JOIN B ON A.col=B.col` with `COUNT_DISTINCT(B.col)=COUNT(B.col)`
+ estimate_join_statistics(
Review Comment:
Really glad that this can also be used in other places ๐ฏ
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -150,66 +207,139 @@ fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
}
}
-impl PhysicalOptimizerRule for HashBuildProbeOrder {
+impl PhysicalOptimizerRule for JoinSelection {
fn optimize(
&self,
plan: Arc<dyn ExecutionPlan>,
session_config: &SessionConfig,
) -> Result<Arc<dyn ExecutionPlan>> {
- let plan = optimize_children(self, plan, session_config)?;
- if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
- let left = hash_join.left();
- let right = hash_join.right();
+ let collect_left_threshold = session_config.hash_join_collect_left_threshold;
+ plan.transform_up(&|plan| {
+ if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
+ if matches!(hash_join.partition_mode(), PartitionMode::Auto) {
+ try_collect_left(hash_join, collect_left_threshold)
+ .unwrap()
+ .or_else(|| Some(partitioned_hash_join(hash_join).unwrap()))
+ } else {
+ let left = hash_join.left();
+ let right = hash_join.right();
+ if should_swap_join_order(&**left, &**right)
+ && supports_swap(*hash_join.join_type())
+ {
+ Some(
+ swap_hash_join(
+ hash_join,
+ *hash_join.partition_mode(),
+ left,
+ right,
+ )
+ .unwrap(),
+ )
+ } else {
+ None
+ }
+ }
+ } else if let Some(cross_join) = plan.as_any().downcast_ref::<CrossJoinExec>()
+ {
+ let left = cross_join.left();
+ let right = cross_join.right();
+ if should_swap_join_order(&**left, &**right) {
+ let new_join =
+ CrossJoinExec::try_new(Arc::clone(right), Arc::clone(left))
+ .unwrap();
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )
+ .unwrap();
+ Some(Arc::new(proj))
+ } else {
+ None
+ }
+ } else {
+ None
+ }
+ })
+ }
+
+ fn name(&self) -> &str {
+ "[CBO]join_selection"
+ }
+}
+
+fn try_collect_left(
Review Comment:
Would it make sense to describe the logic here for each of the different scenarios (i was a bit lost till the end to figure out each state and how it should behave)?
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1315945094
I will try and review this carefully tomorrow
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028975418
##########
datafusion/core/src/execution/context.rs:
##########
@@ -1228,6 +1228,11 @@ pub struct SessionConfig {
pub collect_statistics: bool,
/// Should DataFusion optimizer run a top down process to reorder the join keys
pub top_down_join_key_reordering: bool,
+ /// Should DataFusion optimizer prefer HashJoin over SortMergeJoin.
+ /// HashJoin can work more efficently than SortMergeJoin but consumes more memory.
+ pub prefer_hash_join: bool,
+ /// The maximum estimated size in bytes for the left input a hash join will be collected into one partition
+ pub hash_join_collect_left_threshold: usize,
Review Comment:
Sure.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029117878
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
Review Comment:
https://github.com/apache/arrow-datafusion/issues/4323
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
mingmwang commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1315113209
@alamb @Dandandan @yahoNanJing @jackwener
Please help to take a look.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029103855
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
Review Comment:
I thought this was DataFusion specific problem. Because for HashJoin, DataFusion always choose to build the left side. If DataFusion can also support build right side, then there is no need to swap the Join, and can avoid this problem.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1321116490
As I wrote elsewhere,
I plan to review this and other join related PRs tomorrow. I apologize for the delays. The join work is really neat, but it is not a high priority at the moment in IOx so I have had to prioritize other work higher and do join related
I appreciate the help that @jackwener @mingmwang are giving each other in the review process. ๐
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028969487
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
+ let column_indices = filter
+ .column_indices()
+ .iter()
+ .map(|idx| {
+ let side = if matches!(idx.side, JoinSide::Left) {
+ JoinSide::Right
+ } else {
+ JoinSide::Left
+ };
+ ColumnIndex {
+ index: idx.index,
+ side,
+ }
+ })
+ .collect();
+
+ Some(JoinFilter::new(
+ filter.expression().clone(),
+ column_indices,
+ filter.schema().clone(),
+ ))
+ }
+ None => None,
+ }
+}
+
+impl PhysicalOptimizerRule for JoinSelection {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ session_config: &SessionConfig,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let collect_left_threshold = session_config.hash_join_collect_left_threshold;
+ plan.transform_up(&|plan| {
+ if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
+ match hash_join.partition_mode() {
+ PartitionMode::Auto => {
+ try_collect_left(hash_join, Some(collect_left_threshold))
+ .unwrap()
+ .or_else(|| Some(partitioned_hash_join(hash_join).unwrap()))
Review Comment:
I have another PR to address this. The interface of the Rewrite/transform Closures need to be changed.
https://github.com/apache/arrow-datafusion/pull/4318
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ursabot commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1326336405
Benchmark runs are scheduled for baseline = 22fdbcfd09f4e55057713892b5d632ab12761d5e and contender = 561be4fda2ed3abac038cf422cb7b03d443748de. 561be4fda2ed3abac038cf422cb7b03d443748de is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/d09577efe32747989f0e7570e179d0d8...39bad7052b3b473cb315c834fdcb91e3/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/f7075c7938054e3fa53e2219b5871700...d6afe4f2ffe14143a523094e1b45f8e9/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/d2eb0a15d4db4255863b9635486d2315...d3d76e633c554f5c93838b741efb0857/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/71a93c7475d944ed96b30e6c0d43e4b7...a138fe9c55c746529bd22d6647b21e31/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1326330487
> Look like we need fix conflict.
I am not sure what conflict you ran into @jackwener. I checked this branch out locally and merged master to this branch and reran all the tests and they passed.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1325494972
I will plan to merge this PR tomorrow unless anyone else would like more time to review
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] jackwener commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics
Posted by GitBox <gi...@apache.org>.
jackwener commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1027263477
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
Review Comment:
so careful! I also meet this problem in Doris.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1023840547
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -150,66 +207,139 @@ fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
}
}
-impl PhysicalOptimizerRule for HashBuildProbeOrder {
+impl PhysicalOptimizerRule for JoinSelection {
fn optimize(
&self,
plan: Arc<dyn ExecutionPlan>,
session_config: &SessionConfig,
) -> Result<Arc<dyn ExecutionPlan>> {
- let plan = optimize_children(self, plan, session_config)?;
- if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
- let left = hash_join.left();
- let right = hash_join.right();
+ let collect_left_threshold = session_config.hash_join_collect_left_threshold;
+ plan.transform_up(&|plan| {
+ if let Some(hash_join) = plan.as_any().downcast_ref::<HashJoinExec>() {
+ if matches!(hash_join.partition_mode(), PartitionMode::Auto) {
+ try_collect_left(hash_join, collect_left_threshold)
+ .unwrap()
+ .or_else(|| Some(partitioned_hash_join(hash_join).unwrap()))
+ } else {
+ let left = hash_join.left();
+ let right = hash_join.right();
+ if should_swap_join_order(&**left, &**right)
+ && supports_swap(*hash_join.join_type())
+ {
+ Some(
+ swap_hash_join(
+ hash_join,
+ *hash_join.partition_mode(),
+ left,
+ right,
+ )
+ .unwrap(),
+ )
+ } else {
+ None
+ }
+ }
+ } else if let Some(cross_join) = plan.as_any().downcast_ref::<CrossJoinExec>()
+ {
+ let left = cross_join.left();
+ let right = cross_join.right();
+ if should_swap_join_order(&**left, &**right) {
+ let new_join =
+ CrossJoinExec::try_new(Arc::clone(right), Arc::clone(left))
+ .unwrap();
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )
+ .unwrap();
+ Some(Arc::new(proj))
+ } else {
+ None
+ }
+ } else {
+ None
+ }
+ })
+ }
+
+ fn name(&self) -> &str {
+ "[CBO]join_selection"
+ }
+}
+
+fn try_collect_left(
Review Comment:
Sure, I will add more comments.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] jackwener commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
jackwener commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1326357627
It should be my mistake.๐ฅฒ
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#issuecomment-1325044071
@alamb Would you mind to take a look again ?
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1030369663
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
+ let proj = ProjectionExec::try_new(
+ swap_reverting_projection(&left.schema(), &right.schema()),
+ Arc::new(new_join),
+ )?;
+ Ok(Arc::new(proj))
+ }
+}
+
+/// When the order of the join is changed by the optimizer,
+/// the columns in the output should not be impacted.
+/// This helper creates the expressions that will allow to swap
+/// back the values from the original left as first columns and
+/// those on the right next
+fn swap_reverting_projection(
+ left_schema: &Schema,
+ right_schema: &Schema,
+) -> Vec<(Arc<dyn PhysicalExpr>, String)> {
+ let right_cols = right_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+ let right_len = right_cols.len();
+ let left_cols = left_schema.fields().iter().enumerate().map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), right_len + i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_owned(),
+ )
+ });
+
+ left_cols.chain(right_cols).collect()
+}
+
+/// Swaps join sides for filter column indices and produces new JoinFilter
+fn swap_join_filter(filter: &Option<JoinFilter>) -> Option<JoinFilter> {
+ match filter {
+ Some(filter) => {
Review Comment:
The `filter` is a ref here, can not call `map()` method, `map()` needs owned type.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1028796788
##########
datafusion/core/src/execution/context.rs:
##########
@@ -1228,6 +1228,11 @@ pub struct SessionConfig {
pub collect_statistics: bool,
/// Should DataFusion optimizer run a top down process to reorder the join keys
pub top_down_join_key_reordering: bool,
+ /// Should DataFusion optimizer prefer HashJoin over SortMergeJoin.
+ /// HashJoin can work more efficently than SortMergeJoin but consumes more memory.
+ pub prefer_hash_join: bool,
Review Comment:
Sure, will move the new added settings to ConfigOptions.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the available statistics
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1024131328
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -68,6 +75,19 @@ fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -
}
}
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: usize,
+) -> bool {
+ if let Some(size) = plan.statistics().total_byte_size {
+ size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count < collection_size_threshold
Review Comment:
My original thinking is making the threshold configuration represent both 1M in bytes or 1M number of rows.
Or maybe we can keep two different configuration items explicitly and document them clearly..
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029122197
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
Review Comment:
The rule has logic to add Projection to turn the output columns ordering back.
There is another PR which will valid the schema after physical rule plan rewriting :
https://github.com/apache/arrow-datafusion/pull/4316
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029108909
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
Review Comment:
This was coming from the original `hash_build_probe_order.rs`. I will fix it in this PR.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] jackwener commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics
Posted by GitBox <gi...@apache.org>.
jackwener commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1027263477
##########
datafusion/core/src/physical_optimizer/join_selection.rs:
##########
@@ -0,0 +1,972 @@
+// 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.
+
+//! Utilizing exact statistics from sources to avoid scanning data
+use std::sync::Arc;
+
+use arrow::datatypes::Schema;
+
+use crate::execution::context::SessionConfig;
+use crate::logical_expr::JoinType;
+use crate::physical_plan::expressions::Column;
+use crate::physical_plan::joins::{
+ utils::{ColumnIndex, JoinFilter, JoinSide},
+ CrossJoinExec, HashJoinExec, PartitionMode,
+};
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use super::optimizer::PhysicalOptimizerRule;
+use crate::error::Result;
+use crate::physical_plan::rewrite::TreeNodeRewritable;
+
+/// For hash join with the partition mode [PartitionMode::Auto], JoinSelection rule will make
+/// a cost based decision to select which PartitionMode mode(Partitioned/CollectLeft) is optimal
+/// based on the available statistics that the inputs have.
+/// If the statistics information is not available, the partition mode will fall back to [PartitionMode::Partitioned].
+///
+/// JoinSelection rule will also reorder the build and probe phase of the hash joins
+/// based on the avaliable statistics that the inputs have.
+/// The rule optimizes the order such that the left (build) side of the join is the smallest.
+/// If the statistics information is not available, the order stays the same as the original query.
+/// JoinSelection rule will also swap the left and right sides for cross join to keep the left side
+/// is the smallest.
+#[derive(Default)]
+pub struct JoinSelection {}
+
+impl JoinSelection {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+// TODO we need some performance test for Right Semi/Right Join swap to Left Semi/Left Join in case that the right side is smaller but not much smaller.
+// TODO In PrestoSQL, the optimizer flips join sides only if one side is much smaller than the other by more than SIZE_DIFFERENCE_THRESHOLD times, by default is is 8 times.
+fn should_swap_join_order(left: &dyn ExecutionPlan, right: &dyn ExecutionPlan) -> 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_size, right_size) = match (
+ left.statistics().total_byte_size,
+ right.statistics().total_byte_size,
+ ) {
+ (Some(l), Some(r)) => (Some(l), Some(r)),
+ _ => (left.statistics().num_rows, right.statistics().num_rows),
+ };
+
+ match (left_size, right_size) {
+ (Some(l), Some(r)) => l > r,
+ _ => false,
+ }
+}
+
+fn supports_collect_by_size(
+ plan: &dyn ExecutionPlan,
+ collection_size_threshold: 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()
+ if let Some(size) = plan.statistics().total_byte_size {
+ size != 0 && size < collection_size_threshold
+ } else if let Some(row_count) = plan.statistics().num_rows {
+ row_count != 0 && row_count < collection_size_threshold
+ } else {
+ false
+ }
+}
+
+fn supports_swap(join_type: JoinType) -> bool {
+ match join_type {
+ JoinType::Inner
+ | JoinType::Left
+ | JoinType::Right
+ | JoinType::Full
+ | JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti => true,
+ }
+}
+
+fn swap_join_type(join_type: JoinType) -> JoinType {
+ match join_type {
+ JoinType::Inner => JoinType::Inner,
+ JoinType::Full => JoinType::Full,
+ JoinType::Left => JoinType::Right,
+ JoinType::Right => JoinType::Left,
+ JoinType::LeftSemi => JoinType::RightSemi,
+ JoinType::RightSemi => JoinType::LeftSemi,
+ JoinType::LeftAnti => JoinType::RightAnti,
+ JoinType::RightAnti => JoinType::LeftAnti,
+ }
+}
+
+fn swap_hash_join(
+ hash_join: &HashJoinExec,
+ partition_mode: PartitionMode,
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let new_join = HashJoinExec::try_new(
+ Arc::clone(right),
+ Arc::clone(left),
+ hash_join
+ .on()
+ .iter()
+ .map(|(l, r)| (r.clone(), l.clone()))
+ .collect(),
+ swap_join_filter(hash_join.filter()),
+ &swap_join_type(*hash_join.join_type()),
+ partition_mode,
+ hash_join.null_equals_null(),
+ )?;
+ if matches!(
+ hash_join.join_type(),
+ JoinType::LeftSemi
+ | JoinType::RightSemi
+ | JoinType::LeftAnti
+ | JoinType::RightAnti
+ ) {
+ Ok(Arc::new(new_join))
+ } else {
+ // TODO avoid adding ProjectionExec again and again, only adding Final Projection
Review Comment:
very careful๐! I also meet this problem in Doris.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on a diff in pull request #4219: [CBO] JoinSelection Rule, select HashJoin Partition Mode based on the Join Type and available statistics, option for SortMergeJoin
Posted by GitBox <gi...@apache.org>.
mingmwang commented on code in PR #4219:
URL: https://github.com/apache/arrow-datafusion/pull/4219#discussion_r1029134146
##########
datafusion/core/src/physical_plan/joins/mod.rs:
##########
@@ -29,6 +29,9 @@ pub enum PartitionMode {
Partitioned,
/// Left side will collected into one partition
CollectLeft,
+ /// When set to Auto, DataFusion optimizer will decide which PartitionMode mode(Partitioned/CollectLeft) is optimal based on statistics.
+ /// It will also consider swapping the left and right inputs for the Join
Review Comment:
That's why in this PR I also modify the existing UTs in https://github.com/apache/arrow-datafusion/blob/d5d2de3362649db85ad54161ee28d9374ed3437c/datafusion/core/tests/sql/joins.rs
to make sure both the CollectLeft/Partitioned mode have enough test coverage.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org