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