You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "zhuqi-lucas (via GitHub)" <gi...@apache.org> on 2026/09/16 02:48:28 UTC
[PR] feat(physical-optimizer): let a rule opt into skipping an unchanged re-run [datafusion]
zhuqi-lucas opened a new pull request, #25356:
URL: https://github.com/apache/datafusion/pull/25356
## Which issue does this PR close?
Closes #25355.
## Rationale for this change
The logical optimizer iterates and stops on convergence (`max_passes` + `LogicalPlanSignature`). The physical optimizer runs its list once, which suits the default chain: every rule able to invalidate distribution or ordering requirements is deliberately ordered *before* the single `EnsureRequirements`, as the per-rule comments in `physical-optimizer/src/optimizer.rs` state.
Custom rule lists do not get that for free. A rewrite inserted after that point — a scan rewrite, a distributed-execution boundary, a view substitution — invalidates requirements again and needs its own enforcement pass behind it. Some of those passes then run on a plan that no preceding rule touched, and each is a full traversal recomputing requirements.
Physical rules already return their input `Arc` untouched when they have nothing to do, so pointer identity is an exact and allocation-free "nothing happened" signal — no hashing, no structural comparison.
## What changes are included in this PR?
- `PhysicalOptimizerRule::skip_if_unchanged()`, defaulted to `false`, for a rule to declare it is a pure function of the plan it is given.
- `datafusion.optimizer.skip_unchanged_physical_rules`, defaulted to `false`.
- When both agree, `optimize_physical_plan` remembers the plan each opted-in rule returned and skips the call when handed back that exact object. The memo is scoped to the optimization run and keyed by rule name: rule instances are shared between queries, so this must not live on the rule, and a rule listed twice is normally two instances rather than one.
- Debug builds run a skipped rule anyway and assert the result is unchanged, so a rule that declares purity without having it fails a test rather than a query. (Spark is the only engine I found that checks this — `RuleExecutor.checkBatchIdempotence` under `Utils.isTesting`; the engines relying on counters or an `Optional` instead have public incidents from non-idempotent rules, e.g. trinodb/trino#11559.)
- Three tests: the skip fires for a repeated rule, it stays inert unless both the rule and the config opt in, and the memo does not leak between plans.
Worth knowing for review: because of the debug self-check, the skip saves nothing in debug builds — it saves the pass in release. The tests encode that explicitly rather than hiding it.
## Are these changes tested?
Yes — the three tests above, plus the existing suites (`datafusion` lib 448, `datafusion-physical-optimizer` 37) pass in both debug and release.
## Are there any user-facing changes?
One new config option and one new trait method, both defaulted off, so existing rules and sessions behave exactly as before. The default rule list has no repeated rules, so it is unaffected either way; the benefit is for custom rule lists assembled via `with_physical_optimizer_rules`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(physical-optimizer): let a rule opt into skipping an unchanged re-run [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#issuecomment-5691333574
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion v55.1.0 (current)
Built [ 48.158s] (current)
Parsing datafusion v55.1.0 (current)
Parsed [ 0.029s] (current)
Building datafusion v55.1.0 (baseline)
Built [ 48.670s] (baseline)
Parsing datafusion v55.1.0 (baseline)
Parsed [ 0.030s] (baseline)
Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.724s] 223 checks: 223 pass, 31 skip
Summary no semver update required
Finished [ 99.435s] datafusion
Building datafusion-common v55.1.0 (current)
Built [ 27.317s] (current)
Parsing datafusion-common v55.1.0 (current)
Parsed [ 0.054s] (current)
Building datafusion-common v55.1.0 (baseline)
Built [ 27.549s] (baseline)
Parsing datafusion-common v55.1.0 (baseline)
Parsed [ 0.054s] (baseline)
Checking datafusion-common v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.815s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip
--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---
Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron
Failed in:
field OptimizerOptions.skip_unchanged_physical_rules in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:1584
Summary semver requires new major version: 1 major and 0 minor checks failed
Finished [ 57.271s] datafusion-common
Building datafusion-session v55.1.0 (current)
Built [ 32.827s] (current)
Parsing datafusion-session v55.1.0 (current)
Parsed [ 0.010s] (current)
Building datafusion-session v55.1.0 (baseline)
Built [ 32.372s] (baseline)
Parsing datafusion-session v55.1.0 (baseline)
Parsed [ 0.010s] (baseline)
Checking datafusion-session v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.217s] 223 checks: 223 pass, 31 skip
Summary no semver update required
Finished [ 66.532s] datafusion-session
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(core): skip a named rule handed a plan it has been seen to leave alone [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#discussion_r4062178461
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -2966,13 +3046,149 @@ 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<String>` 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, Vec<ProvenFixpoint>>::new())
+ });
+
+ if fixpoints.is_some() {
+ let available: HashSet<&str> =
+ optimizers.iter().map(|rule| rule.name()).collect();
+ let unmatched = unmatched_rule_names(configured, &available);
+ if !unmatched.is_empty() {
+ log::warn!(
+ "skip_unchanged_physical_rules names no rule in this chain: {}. \
+ Check the spelling against the names EXPLAIN VERBOSE prints; \
+ an unmatched name has no effect.",
+ unmatched.join(", ")
+ );
+ }
+ }
+
for optimizer in optimizers {
+ // `Some` when the rule is named and this plan is not already known
+ // to be one of its fixpoints: carries the input to record below.
+ let mut pending: Option<ProvenFixpoint> = None;
+ if let Some((names, seen)) = fixpoints.as_ref()
+ && names.contains(optimizer.name())
+ {
+ let known = seen.get(optimizer.name());
+
+ // The same object coming back around is the common case when
+ // the rules in between left the plan alone, and it settles
+ // identity without rendering anything.
+ let same_object = known.is_some_and(|entries| {
+ entries.iter().any(|(plan, _)| Arc::ptr_eq(plan, &new_plan))
+ });
+
+ // Otherwise the plan has to be rendered: a rule that changed
+ // nothing still commonly rebuilds the tree, so a different
+ // object can still be the same plan.
+ let fingerprint =
+ (!same_object).then(|| plan_fingerprint(new_plan.as_ref()));
+ let same_content = fingerprint.as_ref().is_some_and(|rendered| {
+ known.is_some_and(|entries| {
+ entries.iter().any(|(_, seen_fp)| seen_fp == rendered)
+ })
+ });
+
+ if same_object || same_content {
+ // 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.
+ //
+ // Off under `cfg(test)`: re-running the rule makes a skipped
+ // pass indistinguishable from one that ran, since both leave
+ // the same call count behind, which would leave every unit
+ // test below unable to tell the feature working from the
+ // feature absent. Integration tests compile the library
+ // without `cfg(test)` and so still exercise this.
+ #[cfg(all(debug_assertions, not(test)))]
Review Comment:
`#[cfg(all(debug_assertions, not(test)))]` self-check is never executed by any test: the comment relies on integration tests, none are added, and the option defaults to empty so slt/CI never reach it. It also makes debug and release call the rule a different number of times.
Simplest: delete the re-run — `builtin_rules_are_idempotent` already guards the built-ins, and the `cfg(test)` carve-out disappears with it:
```diff
- #[cfg(all(debug_assertions, not(test)))]
- {
- 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!(
- plan_fingerprint(rerun.as_ref()),
- plan_fingerprint(new_plan.as_ref()),
- "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;
```
Otherwise add a `core_integration` test with a stateful rule and `#[should_panic]`.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -193,6 +194,85 @@ impl PhysicalPlanner for DefaultPhysicalPlanner {
}
}
+/// Rendering used to tell one plan from another when deciding whether a rule
+/// has already been seen to leave a plan alone.
+///
+/// Verbose so that node detail is included, and with the schema appended so
+/// that two plans differing only in nullability are not taken for one. This is
+/// still not a structural equality: anything no node prints is invisible here.
+/// A plan a rule has been observed to return unchanged, kept alive alongside
+/// its fingerprint.
+///
+/// The `Arc` is held rather than a raw address so that pointer equality is a
+/// sound identity check: while this entry lives the node cannot be dropped, so
+/// its address cannot be handed to a different plan. That is what lets the
+/// lookup below answer from the pointer alone and skip rendering entirely.
+type ProvenFixpoint = (Arc<dyn ExecutionPlan>, String);
+
+/// Configured rule names that no rule in the chain answers to.
+///
+/// A name is matched against what a rule reports as its `name()`, so a typo or
+/// a rule since renamed upstream leaves the entry doing nothing at all. That is
+/// silent otherwise: the option keeps working, just never on the rule the user
+/// meant. Returned in the order configured so the warning reads back the way it
+/// was written.
+fn unmatched_rule_names<'a>(
+ configured: &'a str,
+ available: &HashSet<&str>,
+) -> Vec<&'a str> {
+ configured
+ .split(',')
+ .map(str::trim)
+ .filter(|name| !name.is_empty() && !available.contains(name))
+ .collect()
+}
+
+fn plan_fingerprint(plan: &dyn ExecutionPlan) -> String {
Review Comment:
Fingerprint omits statistics (`FileScanConfig::fmt_as` doesn't print them, `PlanProperties` doesn't carry them), but `join_selection` decides on them.
Trace: chain `[join_selection, <rule that refreshes scan stats>, join_selection]`, `skip_unchanged_physical_rules = "join_selection"`. Pass 1 sees `HashJoin(small, big)` → `should_swap_join_order` false → recorded. Stats rule inverts sizes, rendering unchanged. Pass 3 → fingerprint hit → skipped; with the option off it swaps.
Fix: name statistics explicitly in the config doc as invisible to the comparison, or include them in the fingerprint for leaf nodes. Doc fix now + follow-up for the rest is fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(physical-optimizer): skip a named rule handed back the plan it just produced [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#issuecomment-5692346305
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/25356?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `83.69565%` with `45 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 81.94%. Comparing base ([`18cbb87`](https://app.codecov.io/gh/apache/datafusion/commit/18cbb877ba9a5cae7fc82313e4569a982d202b83?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`21994a5`](https://app.codecov.io/gh/apache/datafusion/commit/21994a5915051139c63c73e51534b5940fb2c54b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/25356?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/core/src/physical\_planner.rs](https://app.codecov.io/gh/apache/datafusion/pull/25356?src=pr&el=tree&filepath=datafusion%2Fcore%2Fsrc%2Fphysical_planner.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9jb3JlL3NyYy9waHlzaWNhbF9wbGFubmVyLnJz) | 83.69% | [13 Missing and 32 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25356?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #25356 +/- ##
========================================
Coverage 81.94% 81.94%
========================================
Files 1135 1135
Lines 428167 428554 +387
Branches 428167 428554 +387
========================================
+ Hits 350845 351166 +321
- Misses 56375 56397 +22
- Partials 20947 20991 +44
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/25356?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(physical-optimizer): skip a named rule handed a plan it has been seen to leave alone [datafusion]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#discussion_r4032940682
##########
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<String>` 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<String>>::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))
Review Comment:
This memo uses `displayable(...).indent(true)` as the plan identity, but that renderer has `show_schema` disabled by default and does not include every execution-plan property. A custom rule can therefore return a plan with different schema or execution semantics while producing the same string; in release the later pass will be skipped and that transformation is silently lost. The debug self-check repeats the same incomplete comparison, so it cannot detect this case. Use a structural identity (or at least include all semantics needed by the rule) before making this opt-in cache available to arbitrary physical rules.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -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<AtomicUsize>,
+ }
+
+ impl PhysicalOptimizerRule for CountingNoopRule {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ 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<dyn PhysicalOptimizerRule + Send + Sync>,
+ reports_inner_name: bool,
+ }
+
+ impl PhysicalOptimizerRule for WrappingRule {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ self.inner.optimize(plan, config)
+ }
+
+ fn optimize_with_context(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ context: &dyn PhysicalOptimizerContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ 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<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,
+ ) -> 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<usize> {
+ let calls = Arc::new(AtomicUsize::new(0));
+ let rule = || {
+ Arc::new(CountingNoopRule {
+ calls: Arc::clone(&calls),
+ }) as Arc<dyn PhysicalOptimizerRule + Send + Sync>
+ };
+ 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<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ 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<dyn PhysicalOptimizerRule + Send + Sync>
+ };
+ let rewrite = |name, rewrite| {
+ Arc::new(RewritingRule { name, rewrite })
+ as Arc<dyn PhysicalOptimizerRule + Send + Sync>
+ };
+
+ 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<usize> {
+ 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<dyn PhysicalOptimizerRule + Send + Sync>
+ };
+ 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<AtomicUsize>,
+ }
+
+ impl PhysicalOptimizerRule for ConvergesAfter {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ 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<dyn PhysicalOptimizerRule + Send + Sync>;
+ 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<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,
+ ) -> Result<Vec<String>> {
+ 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<Arc<dyn PhysicalOptimizerRule + Send + Sync>> {
+ 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 \
Review Comment:
This assertion compares only the final plan after the entire optimizer chain has run. A later rule can normalize the extra rewrite from the duplicated rule, so a rule that changes its own output on the second invocation can still pass this test; it does not establish the idempotence claim in the test description (and the PR notes `EnsureRequirements` may require multiple passes). Capture the plan immediately after the duplicated rule, before subsequent rules run, to make this regression check meaningful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(core): skip a named rule handed a plan it has been seen to leave alone [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#discussion_r4063556671
##########
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()
Review Comment:
This seems to me like a workaround to a more fundamental problem (which is that PhysicalOptimizers have both enforcement and optimization duties). I think if we made that split explicit we would be in a much better place for optimizing the PhysicalOptimizer. See https://github.com/apache/datafusion/issues/25355#issuecomment-5762888518
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(core): skip a named rule handed a plan it has been seen to leave alone [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#issuecomment-5770152109
> Thank you for working on this @zhuqi-lucas and @jayzhan211
>
> IN my opinion this is adding complexity in the wrong place (see [#25355 (comment)](https://github.com/apache/datafusion/issues/25355#issuecomment-5762888518)).
Thank you @alamb and @jayzhan211 for review, mark this draft since we want to start the new design!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(core): skip a named rule handed a plan it has been seen to leave alone [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #25356:
URL: https://github.com/apache/datafusion/pull/25356#issuecomment-5762892813
I have an alternate proposal: `PhysicalAnalyzerRule`s: https://github.com/apache/datafusion/issues/25355#issuecomment-5762888518
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org