Copilot commented on code in PR #25688:
URL: https://github.com/apache/datafusion/pull/25688#discussion_r4091503403
##########
datafusion/session/src/session.rs:
##########
@@ -135,6 +137,17 @@ pub trait Session: Send + Sync {
&[]
}
+ /// Return the physical analyzer rules for this session.
+ ///
+ /// Analyzer rules run before the physical optimizer rules and make the
plan
+ /// *valid* (for example by enforcing the distribution and ordering
+ /// requirements every operator declares). The default implementation
+ /// returns **no rules**; any real session should override this method (for
+ /// example by returning `SessionState::physical_analyzers`).
+ fn physical_analyzers(&self) -> &[Arc<dyn PhysicalAnalyzerRule + Send +
Sync>] {
+ &[]
+ }
Review Comment:
This new default is empty, so existing `Session` adapters that only forward
`physical_optimizers` silently lose `EnsureRequirements`. In particular,
`ForeignSession` forwards the optimizer list across the FFI boundary but has no
`physical_analyzers` implementation; a planner invoked with that session can
therefore receive no enforcement and produce a plan missing required
repartitioning or sorting. Add the analyzer callback/conversion and delegate it
in the FFI adapter, or otherwise preserve the inner analyzer list.
##########
datafusion/physical-optimizer/src/analyzer.rs:
##########
@@ -0,0 +1,111 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Physical analyzer
+
+use std::sync::Arc;
+
+use crate::ensure_requirements::EnsureRequirements;
+
+// Re-export from this module for convenience.
+pub use datafusion_session::PhysicalAnalyzerRule;
+
+/// A rule-based physical analyzer.
+///
+/// Analyzer rules run before the
[`PhysicalOptimizer`](crate::optimizer::PhysicalOptimizer)
+/// rules and make the plan *valid*: they enforce the invariants every operator
+/// declares (distribution, ordering) rather than making the plan faster. This
+/// mirrors the logical layer's `Analyzer`/`Optimizer` split.
Review Comment:
The implementation does not run analyzer rules before the optimizer
pipeline: the planner delays this phase until the
`CombinePartialFinalAggregate` boundary to preserve the historical
`EnsureRequirements` position. This wording gives users of `PhysicalAnalyzer`
the wrong ordering contract.
##########
datafusion/session/src/physical_analyzer.rs:
##########
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Physical analyzer interfaces.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use datafusion_common::Result;
+use datafusion_common::config::ConfigOptions;
+use datafusion_physical_plan::ExecutionPlan;
+
+use crate::physical_optimizer::PhysicalOptimizerContext;
+
+/// A `PhysicalAnalyzerRule` transforms an [`ExecutionPlan`] to make the plan
+/// *valid* prior to the rest of the DataFusion physical optimization process.
+///
+/// `PhysicalAnalyzerRule`s are different from [`PhysicalOptimizerRule`]s: an
+/// optimizer rule must preserve the semantics of an already-valid plan while
+/// computing the same results in a more efficient way, whereas an analyzer
+/// rule is what *establishes* those semantics by satisfying the invariants
+/// every operator declares.
+///
+/// For example, an analyzer rule may repartition an [`ExecutionPlan`]'s input
+/// to match [`ExecutionPlan::required_input_distribution`] or insert a
+/// `SortExec` to match [`ExecutionPlan::required_input_ordering`].
+///
+/// This mirrors the logical layer's split between `AnalyzerRule` (make the
+/// plan valid) and `OptimizerRule` (make the plan faster). The physical
+/// planner runs all analyzer rules first, then the optimizer rules.
Review Comment:
This documents a phase ordering that the implementation does not provide:
the default planner deliberately invokes analyzers at the historical
`EnsureRequirements` boundary, after several optimizer rules, rather than
before every optimizer rule. This can mislead custom analyzer authors about
what plan state their rule will see; please document the boundary behavior here
(and in the corresponding `PhysicalAnalyzer` docs).
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -3122,7 +3125,62 @@ impl DefaultPhysicalPlanner {
let optimizer_context = SessionOptimizerContext {
session: session_state,
};
- for optimizer in optimizers {
+
+ // The analyzer phase (enforcement) makes the plan *valid* by
satisfying
+ // the distribution and ordering requirements every operator declares.
+ // It runs at the point in the historical rule order where
+ // `EnsureRequirements` used to sit: after the optimizer passes that
+ // *establish* those requirements and before the passes that assume
they
+ // already hold. `JoinSelection` is the load-bearing case: it decides
+ // broadcast (single-partition) vs partitioned joins, which changes the
+ // required input distribution, so enforcing before it repartitions the
+ // inputs of a join that is then turned into a single-partition
broadcast
+ // join, producing an invalid plan. In the default pipeline that
boundary
+ // is right before the rule named below. A custom optimizer list
without
+ // that rule runs the analyzers first (the phase-pure order), which is
+ // correct for any list whose passes do not depend on running before
+ // enforcement.
+ const ANALYZER_BOUNDARY_RULE: &str = "CombinePartialFinalAggregate";
+ let analyzer_boundary = optimizers
+ .iter()
+ .position(|o| o.name() == ANALYZER_BOUNDARY_RULE)
+ .unwrap_or(0);
Review Comment:
The analyzer placement is derived from a rule name that custom optimizer
lists can omit or rename, with the fallback running analyzers at index 0. For
example, a custom list containing `JoinSelection` and `SanityCheckPlan` but not
`CombinePartialFinalAggregate` runs `EnsureRequirements` before join selection;
the default pipeline explicitly requires join selection to run first because it
can change a join to single-partition/broadcast and therefore change its
requirements. That can leave the final plan invalid. The boundary needs to be
explicit or preserved independently of the presence of this particular rule.
This issue also appears on line 3498 of the same file.
--
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]
