diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 360586b0e9bae..9db7e7c66b0f9 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1597,6 +1597,27 @@ config_namespace! { /// repartitioning to increase parallelism to leverage more CPU cores pub enable_round_robin_repartition: bool, default = true + /// Comma separated names of physical optimizer rules that may be + /// skipped when handed a plan they have already been seen to leave + /// untouched. Empty, the default, disables the optimization. + /// + /// Physical rules run as a fixed sequence with no fixpoint loop, so a + /// list holding the same rule several times runs it again on plans it + /// has already settled. A plan is remembered only after the rule ran + /// on it and returned that same plan, so a skip replays an observed + /// outcome rather than predicting one; a rule that has not yet + /// converged records nothing and keeps running. What is remembered is + /// scoped to one planning run, and plans are compared by rendered + /// form, since a rule that changes nothing still commonly rebuilds the + /// tree. Debug builds re-run a skipped rule and check it. + /// + /// Names are matched against what a rule reports as its name, which is + /// what `EXPLAIN VERBOSE` shows; an unmatched name is ignored. A name + /// stands for a behaviour, since every rule answering to it shares one + /// record: the built-in `OutputRequirements` names two instances that + /// do opposite things, so it must not be listed. + pub skip_unchanged_physical_rules: String, default = "".to_string() + /// When set to true, the optimizer will attempt to perform limit operations /// during aggregations, if possible pub enable_topk_aggregation: bool, default = true diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 77997c619e5ce..bc940e0d58416 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -2965,13 +2965,99 @@ impl DefaultPhysicalPlanner { let optimizer_context = SessionOptimizerContext { session: session_state, }; + // For each rule `skip_unchanged_physical_rules` names, the plans that + // rule has been *observed* to leave untouched, so a later pass handed + // one of them can return it instead of re-deriving it. + // + // A plan is recorded only after the rule has actually run on it and + // produced the same plan back, so nothing here is an assumption: a + // skip replays an outcome already seen. That matters because a rule + // is not required to reach its fixpoint in one pass, and the plan it + // returns is frequently not yet one. Keying on the plan the rule was + // *given* rather than on the plan it last *returned* is what keeps + // those two cases apart. + // + // Plans are keyed by rendered form rather than by pointer, because a + // rule that changes nothing still commonly rebuilds the tree and + // returns a fresh object. `HashSet` compares on collision, so + // two plans that hash alike are not confused for one another. + // + // Scoped to this call, which keeps the config out of the key: it + // cannot change midway through one optimization run. Rule instances + // are shared between queries, so this must not live on the rule. + let configured = &session_state + .config_options() + .optimizer + .skip_unchanged_physical_rules; + let mut fixpoints = (!configured.is_empty()).then(|| { + let names: HashSet<&str> = configured + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .collect(); + (names, HashMap::<&str, HashSet>::new()) + }); + for optimizer in optimizers { + // Rendered once per pass when the rule is named, and reused to + // record the outcome below. + let mut rendered_input = None; + if let Some((names, seen)) = fixpoints.as_ref() + && names.contains(optimizer.name()) + { + let before = displayable(new_plan.as_ref()).indent(true).to_string(); + if seen + .get(optimizer.name()) + .is_some_and(|plans| plans.contains(&before)) + { + // This rule has already run on this exact plan and left it + // alone, so running it again yields the same plan. Debug + // builds check that rather than trusting it. + #[cfg(debug_assertions)] + { + let rerun = optimizer + .optimize_with_context( + Arc::clone(&new_plan), + &optimizer_context, + ) + .map_err(|e| { + DataFusionError::Context( + optimizer.name().to_string(), + Box::new(e), + ) + })?; + debug_assert_eq!( + displayable(rerun.as_ref()).indent(true).to_string(), + before, + "PhysicalOptimizer rule '{}' is named in \ + datafusion.optimizer.skip_unchanged_physical_rules \ + but stopped leaving a plan it had left alone before \ + untouched, so it does not depend only on the plan \ + and the config", + optimizer.name(), + ); + } + observer(new_plan.as_ref(), optimizer.as_ref()); + continue; + } + rendered_input = Some(before); + } + let before_schema = new_plan.schema(); new_plan = optimizer .optimize_with_context(new_plan, &optimizer_context) .map_err(|e| { DataFusionError::Context(optimizer.name().to_string(), Box::new(e)) })?; + // Record a fixpoint only where the rule demonstrably produced + // the plan it was given. A rule still working towards its + // fixpoint records nothing, so its next pass is not skipped. + if let Some(before) = rendered_input + && let Some((_, seen)) = fixpoints.as_mut() + && displayable(new_plan.as_ref()).indent(true).to_string() == before + { + seen.entry(optimizer.name()).or_default().insert(before); + } // This only checks the schema in release build, and performs additional checks in debug mode. OptimizationInvariantChecker::new(optimizer) @@ -3368,7 +3454,7 @@ mod tests { use std::fmt::{self, Debug}; use std::mem::size_of_val; use std::ops::{BitAnd, Not}; - use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering}; use super::*; use crate::datasource::MemTable; @@ -3408,6 +3494,9 @@ mod tests { use datafusion_functions_aggregate::count::{count_all, count_udaf}; use datafusion_functions_aggregate::expr_fn::sum; use datafusion_physical_expr::EquivalenceProperties; + use datafusion_physical_optimizer::ensure_requirements::EnsureRequirements; + use datafusion_physical_optimizer::optimizer::PhysicalOptimizer; + use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion_physical_plan::{ChildrenPropertiesMode, ReplaceChildrenOptions}; use datafusion_session::QueryPlanner; @@ -3694,6 +3783,480 @@ mod tests { Ok(()) } + /// Counts its invocations and hands the plan back untouched, mimicking a + /// rule that finds nothing to do on an already-satisfied plan. + #[derive(Debug)] + struct CountingNoopRule { + calls: Arc, + } + + impl PhysicalOptimizerRule for CountingNoopRule { + fn optimize( + &self, + plan: Arc, + _config: &ConfigOptions, + ) -> Result> { + self.calls.fetch_add(1, AtomicOrdering::Relaxed); + Ok(plan) + } + + fn name(&self) -> &str { + "counting_noop_rule" + } + + fn schema_check(&self) -> bool { + true + } + } + + /// Stands in for the instrumentation wrappers downstream projects put + /// around every rule to time or trace it. Naming rules by name is what + /// lets such a wrapper keep working, since it already has to report the + /// name it wraps for `EXPLAIN VERBOSE` to stay readable. + #[derive(Debug)] + struct WrappingRule { + inner: Arc, + reports_inner_name: bool, + } + + impl PhysicalOptimizerRule for WrappingRule { + fn optimize( + &self, + plan: Arc, + config: &ConfigOptions, + ) -> Result> { + self.inner.optimize(plan, config) + } + + fn optimize_with_context( + &self, + plan: Arc, + context: &dyn PhysicalOptimizerContext, + ) -> Result> { + self.inner.optimize_with_context(plan, context) + } + + fn name(&self) -> &str { + if self.reports_inner_name { + self.inner.name() + } else { + "wrapping_rule" + } + } + + fn schema_check(&self) -> bool { + self.inner.schema_check() + } + } + + /// Expected invocations of a rule listed twice and named in the config. + /// Debug builds verify the idempotence claim by running a skipped rule + /// anyway and asserting it changed nothing, so the call still happens + /// there: what the skip saves in debug is nothing, and in release it is + /// the whole second pass. + const SKIPPED_CALLS: usize = if cfg!(debug_assertions) { 2 } else { 1 }; + + /// A context whose physical rule list is exactly `rules`, with the given + /// value for `skip_unchanged_physical_rules`. + fn session_with_rules( + skip_config: &str, + rules: Vec>, + ) -> SessionContext { + let mut config = SessionConfig::new(); + config.options_mut().optimizer.skip_unchanged_physical_rules = + skip_config.to_string(); + let state = SessionStateBuilder::new() + .with_config(config) + .with_default_features() + .with_physical_optimizer_rules(rules) + .build(); + SessionContext::new_with_state(state) + } + + /// Plans the same rule twice, which is the shape a custom rule list takes + /// when a rewrite between the two passes may or may not fire, and reports + /// how many times the rule was actually asked to optimize. + async fn run_repeated_rule(skip_config: &str) -> Result { + let calls = Arc::new(AtomicUsize::new(0)); + let rule = || { + Arc::new(CountingNoopRule { + calls: Arc::clone(&calls), + }) as Arc + }; + let ctx = session_with_rules(skip_config, vec![rule(), rule()]); + let logical_plan = LogicalPlanBuilder::empty(false).build()?; + ctx.state().create_physical_plan(&logical_plan).await?; + Ok(calls.load(AtomicOrdering::Relaxed)) + } + + /// A named rule is called once instead of twice: the second entry receives + /// the exact plan the first returned. + #[tokio::test] + async fn skip_unchanged_skips_the_repeated_pass() -> Result<()> { + assert_eq!( + run_repeated_rule("counting_noop_rule").await?, + SKIPPED_CALLS + ); + Ok(()) + } + + /// Off unless the rule is named, so an empty config behaves exactly as + /// before this feature existed, and a name that matches nothing, whether a + /// typo or a rule that is not in this list, is simply inert. + #[tokio::test] + async fn skip_unchanged_is_inert_unless_the_rule_is_named() -> Result<()> { + assert_eq!(run_repeated_rule("").await?, 2); + assert_eq!(run_repeated_rule("some_other_rule").await?, 2); + assert_eq!(run_repeated_rule("counting_noop_rul").await?, 2); + Ok(()) + } + + /// The config is a list, and reading it tolerates the spacing people + /// actually write. + #[tokio::test] + async fn skip_unchanged_reads_a_list_of_names() -> Result<()> { + for config in [ + "counting_noop_rule,some_other_rule", + "some_other_rule, counting_noop_rule", + " counting_noop_rule ,, ", + ] { + assert_eq!(run_repeated_rule(config).await?, SKIPPED_CALLS, "{config}"); + } + Ok(()) + } + + /// Replaces the plan with an equivalent new object, standing in for a + /// rewrite that fires only for some queries. + #[derive(Debug)] + struct RewritingRule { + name: &'static str, + rewrite: bool, + } + + impl PhysicalOptimizerRule for RewritingRule { + fn optimize( + &self, + plan: Arc, + _config: &ConfigOptions, + ) -> Result> { + if self.rewrite { + Ok(Arc::new(EmptyExec::new(plan.schema()))) + } else { + Ok(plan) + } + } + + fn name(&self) -> &str { + self.name + } + + fn schema_check(&self) -> bool { + true + } + } + + /// The shape this feature exists for: a chain that enforces requirements, + /// applies its own rewrites, and enforces again after each one. Only the + /// enforcement passes that follow a rewrite which actually fired have work + /// to do; the others receive the plan the previous enforcement produced. + /// + /// Here the first rewrite fires and the second does not, so of three + /// enforcement passes exactly two must run. + #[tokio::test] + async fn skip_unchanged_handles_an_interleaved_chain() -> Result<()> { + let calls = Arc::new(AtomicUsize::new(0)); + let enforce = || { + Arc::new(CountingNoopRule { + calls: Arc::clone(&calls), + }) as Arc + }; + let rewrite = |name, rewrite| { + Arc::new(RewritingRule { name, rewrite }) + as Arc + }; + + let ctx = session_with_rules( + "counting_noop_rule", + vec![ + enforce(), // runs: nothing memoized yet + rewrite("rewrite_that_fires", true), + enforce(), // runs: the plan changed + rewrite("rewrite_that_does_not", false), + enforce(), // skipped: plan is unchanged + ], + ); + let logical_plan = LogicalPlanBuilder::empty(false).build()?; + ctx.state().create_physical_plan(&logical_plan).await?; + + // Two enforcement passes have real work; the third is skipped (and in + // debug builds re-run by the self-check, which is why this counts + // against SKIPPED_CALLS rather than a literal). + assert_eq!(calls.load(AtomicOrdering::Relaxed), 2 + (SKIPPED_CALLS - 1)); + Ok(()) + } + + /// The memo is per optimization run, not per rule instance: planning a + /// second query must not let the first query's plan suppress a call. + #[tokio::test] + async fn skip_unchanged_does_not_leak_between_plans() -> Result<()> { + let calls = Arc::new(AtomicUsize::new(0)); + let ctx = session_with_rules( + "counting_noop_rule", + vec![Arc::new(CountingNoopRule { + calls: Arc::clone(&calls), + })], + ); + let logical_plan = LogicalPlanBuilder::empty(false).build()?; + ctx.state().create_physical_plan(&logical_plan).await?; + ctx.state().create_physical_plan(&logical_plan).await?; + // Once per plan; a rule-level memo would have suppressed the second. + // The rule is listed once here, so the debug self-check never fires. + assert_eq!(calls.load(AtomicOrdering::Relaxed), 2); + Ok(()) + } + + /// Rules are matched by the name they report, so a rule wrapped for timing + /// or tracing is reached through the name the wrapper passes through, with + /// no cooperation needed from the wrapper beyond what `EXPLAIN VERBOSE` + /// already requires of it. A wrapper that renames what it wraps is + /// addressed by its own name instead. + #[tokio::test] + async fn skip_unchanged_follows_the_name_a_wrapper_reports() -> Result<()> { + async fn wrapped_calls( + reports_inner_name: bool, + skip_config: &str, + ) -> Result { + let calls = Arc::new(AtomicUsize::new(0)); + let rule = || { + Arc::new(WrappingRule { + inner: Arc::new(CountingNoopRule { + calls: Arc::clone(&calls), + }), + reports_inner_name, + }) as Arc + }; + let ctx = session_with_rules(skip_config, vec![rule(), rule()]); + let logical_plan = LogicalPlanBuilder::empty(false).build()?; + ctx.state().create_physical_plan(&logical_plan).await?; + Ok(calls.load(AtomicOrdering::Relaxed)) + } + + assert_eq!( + wrapped_calls(true, "counting_noop_rule").await?, + SKIPPED_CALLS + ); + assert_eq!(wrapped_calls(false, "counting_noop_rule").await?, 2); + assert_eq!(wrapped_calls(false, "wrapping_rule").await?, SKIPPED_CALLS); + Ok(()) + } + + /// Changes the plan a fixed number of times and is a no-op after that, + /// standing in for a rule that needs several passes to converge. + /// `EnsureRequirements` is one: on real plans its distribution and sorting + /// phases can each still find work on a plan it produced itself. + #[derive(Debug)] + struct ConvergesAfter { + remaining: AtomicUsize, + calls: Arc, + } + + impl PhysicalOptimizerRule for ConvergesAfter { + fn optimize( + &self, + plan: Arc, + _config: &ConfigOptions, + ) -> Result> { + self.calls.fetch_add(1, AtomicOrdering::Relaxed); + if self.remaining.load(AtomicOrdering::Relaxed) == 0 { + return Ok(plan); + } + self.remaining.fetch_sub(1, AtomicOrdering::Relaxed); + Ok(Arc::new(CoalescePartitionsExec::new(plan))) + } + + fn name(&self) -> &str { + "counting_noop_rule" + } + + fn schema_check(&self) -> bool { + true + } + } + + /// The case that makes "skip what the rule last returned" wrong and this + /// design right: a rule still working towards its fixpoint must keep + /// running. Only a plan the rule has been seen to leave alone is recorded, + /// so the passes that still have work to do are never skipped, and the + /// plan comes out exactly as it does with the optimization off. + #[tokio::test] + async fn skip_unchanged_does_not_skip_a_rule_that_has_not_converged() -> Result<()> { + async fn run(skip_config: &str) -> Result<(usize, String)> { + let calls = Arc::new(AtomicUsize::new(0)); + // Shared across the five entries, as one rule instance repeated in + // a chain would be: it converges after two rewrites. + let rule = Arc::new(ConvergesAfter { + remaining: AtomicUsize::new(2), + calls: Arc::clone(&calls), + }) as Arc; + let ctx = session_with_rules( + skip_config, + (0..5).map(|_| Arc::clone(&rule)).collect(), + ); + let logical_plan = LogicalPlanBuilder::empty(false).build()?; + let plan = ctx.state().create_physical_plan(&logical_plan).await?; + Ok(( + calls.load(AtomicOrdering::Relaxed), + displayable(plan.as_ref()).indent(true).to_string(), + )) + } + + let (off_calls, off_plan) = run("").await?; + let (on_calls, on_plan) = run("counting_noop_rule").await?; + + // Five entries, all of which run with the optimization off. + assert_eq!(off_calls, 5); + // With it on, the two rewriting passes and the one that proves the + // fixpoint still run; only the last two are skipped. + assert_eq!(on_calls, 3 + 2 * (SKIPPED_CALLS - 1)); + // And the plan is unaffected, which is the point. + assert_eq!(on_plan, off_plan); + Ok(()) + } + + /// Queries chosen to reach the operators the built-in rules act on. + const PLAN_CORPUS: &[&str] = &[ + "SELECT a, sum(b) FROM t GROUP BY a ORDER BY a", + "SELECT count(*) FROM t", + "SELECT DISTINCT a FROM t", + "SELECT * FROM t ORDER BY b LIMIT 5", + "SELECT a FROM t WHERE b > 10 ORDER BY a LIMIT 3", + "SELECT t.a, u.d FROM t JOIN u ON t.a = u.c", + "SELECT a, row_number() OVER (PARTITION BY a ORDER BY b) FROM t", + "SELECT a, b FROM t UNION ALL SELECT c, d FROM u", + "SELECT a, sum(b) FROM t GROUP BY a HAVING sum(b) > 5 ORDER BY a LIMIT 2", + ]; + + /// Plans every query in [`PLAN_CORPUS`] through `rules`. + async fn corpus_plans( + skip_config: &str, + rules: Vec>, + ) -> Result> { + let ctx = session_with_rules(skip_config, rules); + ctx.sql("CREATE TABLE t(a INT, b INT) AS VALUES (1,10),(2,20),(1,30)") + .await? + .collect() + .await?; + ctx.sql("CREATE TABLE u(c INT, d INT) AS VALUES (1,100),(3,300)") + .await? + .collect() + .await?; + let mut plans = Vec::with_capacity(PLAN_CORPUS.len()); + for query in PLAN_CORPUS { + let plan = ctx.sql(query).await?.create_physical_plan().await?; + plans.push(displayable(plan.as_ref()).indent(true).to_string()); + } + Ok(plans) + } + + /// The built-in list with two further enforcement passes appended, as a + /// downstream list has after inserting rewrites of its own behind the + /// built-in enforcement. This is the shape that motivates the feature. + fn rules_with_trailing_enforcement() + -> Vec> { + let mut rules = PhysicalOptimizer::default().rules; + rules.push(Arc::new(EnsureRequirements::new())); + rules.push(Arc::new(EnsureRequirements::new())); + rules + } + + /// Turning the optimization on must change how often a rule runs and + /// nothing else, so every plan in the corpus has to come out identical. + #[tokio::test] + async fn skip_unchanged_leaves_the_plan_alone() -> Result<()> { + let skipped = + corpus_plans("EnsureRequirements", rules_with_trailing_enforcement()).await?; + let stock = corpus_plans("", rules_with_trailing_enforcement()).await?; + assert_eq!(skipped, stock); + Ok(()) + } + + /// Naming a rule in the config asserts that running it on its own output + /// arrives at the same plan. This checks that claim for every built-in + /// rule by running each one twice in place and requiring the corpus to + /// plan identically, and so records which of them may be named. + /// + /// All of them can, today. A rule that stops being idempotent breaks the + /// promise for anyone who named it, which is what this guards. + #[tokio::test] + async fn builtin_rules_are_idempotent() -> Result<()> { + let stock = PhysicalOptimizer::default().rules; + let baseline = corpus_plans("", stock.clone()).await?; + + for (position, rule) in stock.iter().enumerate() { + let mut doubled = stock.clone(); + doubled.insert(position + 1, Arc::clone(rule)); + assert_eq!( + corpus_plans("", doubled).await?, + baseline, + "running '{}' (position {position}) twice changed the plan, so it \ + is not idempotent and must not be named in \ + datafusion.optimizer.skip_unchanged_physical_rules", + rule.name(), + ); + } + Ok(()) + } + + /// A name stands for a behaviour, since every rule answering to it shares + /// one memo entry. The built-in chain breaks that for `OutputRequirements`, + /// whose two instances add and then remove the same requirements under one + /// name, so naming it in the config would let the first instance's output + /// suppress the second. Pinned here because the config documents it, and + /// because giving the two modes distinct names would make it safe. + /// + /// `ProjectionPushdown` repeats too, but both instances are the same rule + /// doing the same thing, which is the case the feature is built for. + #[tokio::test] + async fn builtin_chain_repeats_two_rule_names() -> Result<()> { + let stock = PhysicalOptimizer::default().rules; + let mut counts = HashMap::<&str, usize>::new(); + for rule in &stock { + *counts.entry(rule.name()).or_default() += 1; + } + let mut repeated: Vec<&str> = counts + .into_iter() + .filter(|(_, count)| *count > 1) + .map(|(name, _)| name) + .collect(); + repeated.sort_unstable(); + assert_eq!(repeated, ["OutputRequirements", "ProjectionPushdown"]); + Ok(()) + } + + /// `EXPLAIN VERBOSE` renders one plan snapshot per rule from the observer + /// callback, so a skipped rule still has to report the plan it would have + /// returned. Otherwise enabling the optimization would silently shorten + /// the explain output. + #[tokio::test] + async fn skip_unchanged_still_reports_every_rule_to_the_observer() -> Result<()> { + async fn explain_with(skip_config: &str) -> Result { + let rule = || { + Arc::new(CountingNoopRule { + calls: Arc::new(AtomicUsize::new(0)), + }) as Arc + }; + let ctx = session_with_rules(skip_config, vec![rule(), rule()]); + let batches = ctx.sql("EXPLAIN VERBOSE SELECT 1").await?.collect().await?; + Ok(arrow::util::pretty::pretty_format_batches(&batches)?.to_string()) + } + + let skipped = explain_with("counting_noop_rule").await?; + assert_eq!(skipped, explain_with("").await?); + // Both passes are reported, including the one that did not run. + assert_eq!(skipped.matches("counting_noop_rule").count(), 2); + Ok(()) + } + async fn aggregate_explain(logical_plan: &LogicalPlan) -> Result { let physical_plan = plan(logical_plan).await?; Ok(displayable(physical_plan.as_ref()).indent(true).to_string()) diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index b270eba99d7b0..b7fea64fbc568 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -339,6 +339,7 @@ datafusion.optimizer.repartition_joins true datafusion.optimizer.repartition_sorts true datafusion.optimizer.repartition_windows true datafusion.optimizer.skip_failed_rules false +datafusion.optimizer.skip_unchanged_physical_rules (empty) datafusion.optimizer.subset_repartition_threshold 4 datafusion.optimizer.top_down_join_key_reordering true datafusion.optimizer.use_statistics_registry false @@ -500,6 +501,7 @@ datafusion.optimizer.repartition_joins true Should DataFusion repartition data u datafusion.optimizer.repartition_sorts true Should DataFusion execute sorts in a per-partition fashion and merge afterwards instead of coalescing first and sorting globally. With this flag is enabled, plans in the form below ```text "SortExec: [a@0 ASC]", " CoalescePartitionsExec", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", ``` would turn into the plan below which performs better in multithreaded environments ```text "SortPreservingMergeExec: [a@0 ASC]", " SortExec: [a@0 ASC]", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", ``` datafusion.optimizer.repartition_windows true Should DataFusion repartition data using the partitions keys to execute window functions in parallel using the provided `target_partitions` level datafusion.optimizer.skip_failed_rules false When set to true, the logical plan optimizer will produce warning messages if any optimization rules produce errors and then proceed to the next rule. When set to false, any rules that produce errors will cause the query to fail +datafusion.optimizer.skip_unchanged_physical_rules (empty) Comma separated names of physical optimizer rules that may be skipped when handed a plan they have already been seen to leave untouched. Empty, the default, disables the optimization. Physical rules run as a fixed sequence with no fixpoint loop, so a list holding the same rule several times runs it again on plans it has already settled. A plan is remembered only after the rule ran on it and returned that same plan, so a skip replays an observed outcome rather than predicting one; a rule that has not yet converged records nothing and keeps running. What is remembered is scoped to one planning run, and plans are compared by rendered form, since a rule that changes nothing still commonly rebuilds the tree. Debug builds re-run a skipped rule and check it. Names are matched against what a rule reports as its name, which is what `EXPLAIN VERBOSE` shows; an unmatched name is ignored. A name stands for a behaviour, since every rule answering to it shares one record: the built-in `OutputRequirements` names two instances that do opposite things, so it must not be listed. datafusion.optimizer.subset_repartition_threshold 4 Partition count threshold for subset satisfaction optimization. When the current partition count is >= this threshold, DataFusion will skip repartitioning if the required partitioning expression is a subset of the current partition expression such as Hash(a) satisfies Hash(a, b). When the current partition count is < this threshold, DataFusion will repartition to increase parallelism even when subset satisfaction applies. Set to 0 to always repartition (disable subset satisfaction optimization). Set to a high value to always use subset satisfaction. Example (subset_repartition_threshold = 4): ```text Hash([a]) satisfies Hash([a, b]) because (Hash([a, b]) is subset of Hash([a]) If current partitions (3) < threshold (4), repartition: AggregateExec: mode=FinalPartitioned, gby=[a, b], aggr=[SUM(x)] RepartitionExec: partitioning=Hash([a, b], 8), input_partitions=3 AggregateExec: mode=Partial, gby=[a, b], aggr=[SUM(x)] DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 3) If current partitions (8) >= threshold (4), use subset satisfaction: AggregateExec: mode=SinglePartitioned, gby=[a, b], aggr=[SUM(x)] DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 8) ``` datafusion.optimizer.top_down_join_key_reordering true When set to true, the physical plan optimizer will run a top down process to reorder the join keys datafusion.optimizer.use_statistics_registry false (Deprecated) Ignored: the physical plan optimizer always consults the session's pluggable `StatisticsRegistry` (register providers on the `SessionState`; with none it is a no-op). diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 0085d4ac7c1fa..01bb94c6685e0 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -147,6 +147,7 @@ The following configuration settings are available: | datafusion.execution.hash_join_buffering_capacity | 0 | How many bytes to buffer in the probe side of hash joins while the build side is concurrently being built. Without this, hash joins will wait until the full materialization of the build side before polling the probe side. This is useful in scenarios where the query is not completely CPU bounded, allowing to do some early work concurrently and reducing the latency of the query. Note that when hash join buffering is enabled, the probe side will start eagerly polling data, not giving time for the producer side of dynamic filters to produce any meaningful predicate. Queries with dynamic filters might see performance degradation. Disabled by default, set to a number greater than 0 for enabling it. | | datafusion.optimizer.enable_distinct_aggregation_soft_limit | true | When set to true, the optimizer will push a limit operation into grouped aggregations which have no aggregate expressions, as a soft limit, emitting groups once the limit is reached, before all rows in the group are read. | | datafusion.optimizer.enable_round_robin_repartition | true | When set to true, the physical plan optimizer will try to add round robin repartitioning to increase parallelism to leverage more CPU cores | +| datafusion.optimizer.skip_unchanged_physical_rules | | Comma separated names of physical optimizer rules that may be skipped when handed a plan they have already been seen to leave untouched. Empty, the default, disables the optimization. Physical rules run as a fixed sequence with no fixpoint loop, so a list holding the same rule several times runs it again on plans it has already settled. A plan is remembered only after the rule ran on it and returned that same plan, so a skip replays an observed outcome rather than predicting one; a rule that has not yet converged records nothing and keeps running. What is remembered is scoped to one planning run, and plans are compared by rendered form, since a rule that changes nothing still commonly rebuilds the tree. Debug builds re-run a skipped rule and check it. Names are matched against what a rule reports as its name, which is what `EXPLAIN VERBOSE` shows; an unmatched name is ignored. A name stands for a behaviour, since every rule answering to it shares one record: the built-in `OutputRequirements` names two instances that do opposite things, so it must not be listed. | | datafusion.optimizer.enable_topk_aggregation | true | When set to true, the optimizer will attempt to perform limit operations during aggregations, if possible | | datafusion.optimizer.enable_window_limits | true | When set to true, the optimizer will attempt to push limit operations past window functions, if possible | | datafusion.optimizer.enable_window_topn | false | When set to true, the optimizer will replace Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a PartitionedTopKExec that maintains per-partition heaps, avoiding a full sort of the input. When the window partition key has low cardinality, enabling this optimization can improve performance. However, for high cardinality keys, it may cause regressions in both memory usage and runtime. |