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: [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]
