alamb commented on code in PR #17467: URL: https://github.com/apache/datafusion/pull/17467#discussion_r2333593307
########## datafusion/core/src/physical_planner/join_planner.rs: ########## @@ -0,0 +1,195 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::Arc; + +use crate::error::Result; +use crate::execution::context::SessionState; +use crate::physical_plan::joins::utils as join_utils; +use crate::physical_plan::joins::{ + CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec, +}; +use crate::physical_plan::ExecutionPlan; +use arrow::compute::SortOptions; +use datafusion_common::plan_err; +use datafusion_expr::JoinType; + +/// Build the appropriate join `ExecutionPlan` for the given join type, filter, and +/// configurations. +/// +/// For example, given an equi-join, the planner may execute it as a Nested Loop +/// Join, Hash Join, or another strategy. Configuration settings determine which +/// ExecutionPlan is used. +/// +/// # Strategy +/// - Step 1: Find all possible physical join types for the given join logical plan +/// - No join on keys and no filter => CrossJoin +/// - With equality? => HJ and SMJ (if with multiple partition) +/// TODO: The constraint on SMJ is added previously for optimization. Should +/// we remove it for configurability? +/// TODO: Allow NLJ for equal join for better configurability. +/// - Without equality? => NLJ +/// - Step 2: Filter the possible join types from step 1 according to the configuration +/// by checking if they're enabled by options like `datafusion.optimizer.enable_hash_join` +/// - Step 3: Choose one according to the built-in heuristics and also the preference +/// in the configuration, e.g. `datafusion.optimizer.prefer_hash_join` +pub(super) fn plan_join_exec( Review Comment: Maybe this could be called "plan_initial_join_exec" ########## datafusion/core/src/physical_planner/join_planner.rs: ########## @@ -0,0 +1,195 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::Arc; + +use crate::error::Result; +use crate::execution::context::SessionState; +use crate::physical_plan::joins::utils as join_utils; +use crate::physical_plan::joins::{ + CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec, +}; +use crate::physical_plan::ExecutionPlan; +use arrow::compute::SortOptions; +use datafusion_common::plan_err; +use datafusion_expr::JoinType; + +/// Build the appropriate join `ExecutionPlan` for the given join type, filter, and +/// configurations. +/// +/// For example, given an equi-join, the planner may execute it as a Nested Loop +/// Join, Hash Join, or another strategy. Configuration settings determine which +/// ExecutionPlan is used. +/// +/// # Strategy +/// - Step 1: Find all possible physical join types for the given join logical plan +/// - No join on keys and no filter => CrossJoin +/// - With equality? => HJ and SMJ (if with multiple partition) +/// TODO: The constraint on SMJ is added previously for optimization. Should +/// we remove it for configurability? +/// TODO: Allow NLJ for equal join for better configurability. +/// - Without equality? => NLJ +/// - Step 2: Filter the possible join types from step 1 according to the configuration +/// by checking if they're enabled by options like `datafusion.optimizer.enable_hash_join` +/// - Step 3: Choose one according to the built-in heuristics and also the preference +/// in the configuration, e.g. `datafusion.optimizer.prefer_hash_join` +pub(super) fn plan_join_exec( + session_state: &SessionState, + physical_left: Arc<dyn ExecutionPlan>, + physical_right: Arc<dyn ExecutionPlan>, + join_on: join_utils::JoinOn, + join_filter: Option<join_utils::JoinFilter>, + join_type: &JoinType, + null_equality: &datafusion_common::NullEquality, +) -> Result<Arc<dyn ExecutionPlan>> { + // Short-circuit: handle pure cross join (existing behavior) + if join_on.is_empty() && join_filter.is_none() && matches!(join_type, JoinType::Inner) + { + return Ok(Arc::new(CrossJoinExec::new(physical_left, physical_right))); + } + + // Step 1: Find possible join types for the given Logical Plan + // ---------------------------------------------------------------------- + + // Build the list of possible algorithms for this join + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum Algo { + Nlj, + Hj, + Smj, + } + + let cfg = &session_state.config_options().optimizer; + let can_smj = session_state.config().target_partitions() > 1 + && session_state.config().repartition_joins(); + + let mut possible: Vec<Algo> = Vec::new(); + if join_on.is_empty() { + possible.push(Algo::Nlj); + } else { + possible.push(Algo::Hj); + if can_smj { + possible.push(Algo::Smj); + } + } + + // Step 2: Filter the possible list according to enable flags from config + // ---------------------------------------------------------------------- + + // Filter by enable flags + let enabled_and_possible: Vec<Algo> = possible + .iter() + .copied() + .filter(|a| match a { + Algo::Nlj => cfg.enable_nested_loop_join, + Algo::Hj => cfg.enable_hash_join, + Algo::Smj => cfg.enable_sort_merge_join, + }) + .collect(); + + if enabled_and_possible.is_empty() { + return plan_err!( + "No enabled join algorithm is applicable for this join. Possible join types are {:?}. Try to enable them through configurations like `datafusion.optimizer.enable_hash_join`", &possible + ); + } + + // Step 3: Choose and plan the physical join type according to preference + // from the configuration, and also the built-in heuristics + // ---------------------------------------------------------------------- + + // Collect preferred algorithms + let mut preferred: Vec<Algo> = Vec::new(); + if cfg.prefer_hash_join { + preferred.push(Algo::Hj); + } + if cfg.prefer_sort_merge_join { + preferred.push(Algo::Smj); + } + if cfg.prefer_nested_loop_join { + preferred.push(Algo::Nlj); + } + + // Helper to pick by priority HJ > SMJ > NLJ Review Comment: this is an important detail (the relative priorities) of the different join algorithms that we should probably communicate to the user somehow -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
