Copilot commented on code in PR #17467: URL: https://github.com/apache/datafusion/pull/17467#discussion_r2346795989
########## datafusion/core/src/physical_planner/join_planner.rs: ########## @@ -0,0 +1,212 @@ +// 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::{config_err, 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 `datafusion.optimizer.join_method_priority` +pub(super) fn plan_initial_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 + // `join_method_priority` and built-in heuristics + // ---------------------------------------------------------------------- + + // Parse join method priority string into an ordered list of algorithms + let parse_priority = |s: &str| -> Result<Vec<Algo>> { + let mut out = Vec::new(); + let mut unknown = Vec::new(); + for token in s.split(',').map(|t| t.trim()).filter(|t| !t.is_empty()) { + match token { + "hj" | "hash_join" => out.push(Algo::Hj), + "smj" | "sort_merge_join" => out.push(Algo::Smj), + "nlj" | "nested_loop_join" => out.push(Algo::Nlj), + other => unknown.push(other.to_string()), + } + } + if !unknown.is_empty() { + let valid = "hj/hash_join, smj/sort_merge_join, nlj/nested_loop_join"; + return config_err!( + "Invalid join method(s) in datafusion.optimizer.join_method_priority: {}. Valid values: {}", + unknown.join(", "), + valid + ); + } + Ok(out) + }; + + // Backward compatibility: + // If `join_method_priority` is empty, honor legacy `prefer_hash_join` by + // setting the priority to a single entry accordingly. Otherwise, parse the Review Comment: The parsing logic is executed on every join operation. Since configuration values are typically stable during query execution, consider parsing and caching the priority list once when the configuration is set, rather than re-parsing the string for each join operation. ```suggestion // Use the cached, pre-parsed join method priority from the configuration. // Backward compatibility: // If `join_method_priority` is empty, honor legacy `prefer_hash_join` by // setting the priority to a single entry accordingly. Otherwise, use the cached, pre-parsed priority list from the configuration. ``` ########## datafusion/common/src/config.rs: ########## @@ -824,8 +824,35 @@ config_namespace! { /// When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin. /// HashJoin can work more efficiently than SortMergeJoin but consumes more memory + /// + /// Note: if `join_method_priority` is set, this configuration will be overridden + #[deprecated(since = "51.0.0", note = "Please use configuration option `join_method_priority` instead")] pub prefer_hash_join: bool, default = true + /// Comma-separated join priority (case-insensitive) selecting the first applicable and enabled (through configurations like `enable_hash_join`) join method. + /// + /// Valid join methods are: hj/hash_join, smj/sort_merge_join, nlj/nested_loop_join. + /// + /// It is allowed to specify any number of join methods in the configuration. If none of them is applicable, the default join planning heuristic will be used. + /// The default join planning policy is: for equi-joins Hash Join will be selected; for non equi-joins, Nested Loop Join will be chosen. + /// + /// Example usage: SET datafusion.optimizer.join_method_priority = 'hj, nlj' + /// + /// Note: If this option is not set (default empty string), the deprecated legacy option `prefer_hash_join` will be used. + pub join_method_priority: String, transform = str::to_lowercase, default = "".to_string() Review Comment: The `str::to_lowercase` transformation is applied on every access of this configuration value. Consider transforming and validating the input once during configuration parsing rather than on every use to avoid repeated string allocations. ```suggestion pub join_method_priority: String, default = "".to_string() ``` ########## datafusion/core/src/physical_planner/join_planner.rs: ########## @@ -0,0 +1,212 @@ +// 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::{config_err, 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 `datafusion.optimizer.join_method_priority` +pub(super) fn plan_initial_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 + // `join_method_priority` and built-in heuristics + // ---------------------------------------------------------------------- + + // Parse join method priority string into an ordered list of algorithms + let parse_priority = |s: &str| -> Result<Vec<Algo>> { + let mut out = Vec::new(); + let mut unknown = Vec::new(); + for token in s.split(',').map(|t| t.trim()).filter(|t| !t.is_empty()) { + match token { + "hj" | "hash_join" => out.push(Algo::Hj), + "smj" | "sort_merge_join" => out.push(Algo::Smj), + "nlj" | "nested_loop_join" => out.push(Algo::Nlj), Review Comment: [nitpick] The string matching for join method aliases could benefit from a centralized mapping. Consider defining a constant or function that maps all valid aliases to avoid duplication and make it easier to add new aliases in the future. -- 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]
