alamb commented on code in PR #25688:
URL: https://github.com/apache/datafusion/pull/25688#discussion_r4092066476
##########
datafusion/core/src/execution/session_state.rs:
##########
@@ -259,6 +265,7 @@ impl Debug for SessionState {
ret.field("query_planners", &self.inner.query_planner)
.field("analyzer", &self.inner.analyzer)
.field("optimizer", &self.inner.optimizer)
+ .field("physical_analyzers", &self.inner.physical_analyzers)
Review Comment:
I know I am biased, but I really like the symmetry here with the `analyzer`
##########
datafusion/core/src/execution/session_state.rs:
##########
@@ -166,6 +169,9 @@ struct SessionStateInner {
type_planner: Option<Arc<dyn TypePlanner>>,
/// Responsible for optimizing a logical plan
optimizer: Optimizer,
+ /// Responsible for enforcing invariants on a physical execution plan
+ /// (distribution, ordering) before optimization
Review Comment:
❤️
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -3122,7 +2969,72 @@ impl DefaultPhysicalPlanner {
let optimizer_context = SessionOptimizerContext {
session: session_state,
};
- for optimizer in optimizers {
+
+ // Where the analyzer phase (enforcement) runs relative to the
optimizer
+ // rules. Enforcement makes the plan *valid* by satisfying the
+ // distribution and ordering requirements every operator declares.
+ //
+ // Ideally analyzers would run strictly first, but the built-in
pipeline
+ // cannot: `JoinSelection` is an optimizer that decides broadcast
Review Comment:
I think we should strive to get rid of this code (perhaps by refactoring the
JoinSelection pass to update the distribution itself (by calling a helper of
Enforce, for example) and actually put enforcement first.
I think longer term being able to easily adjust join orders and update the
plan as necessary is important for having better join optimizer support
So maybe we need to start with a PR to run JoinSelection after Enforcement,
and then this PR to pull enforcment to an analyzer rule that runs first
The other thing we could do temporarily is maybe put `JoinSelection` as an
analyzer rule (as strange as that is) and file a ticket to make it a real
analyzer rule
##########
datafusion/physical-optimizer/src/analyzer.rs:
##########
@@ -0,0 +1,113 @@
+// 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 make the plan *valid*: they enforce the invariants every
+/// operator declares (distribution, ordering) rather than making the plan
+/// faster, mirroring the logical layer's `Analyzer`/`Optimizer` split. They
run
+/// as a distinct phase relative to the
[`PhysicalOptimizer`](crate::optimizer::PhysicalOptimizer)
+/// rules; see [`PhysicalAnalyzerRule`] for how the default planner places that
+/// phase (it is not run strictly before every optimizer rule).
+#[derive(Clone, Debug)]
+pub struct PhysicalAnalyzer {
+ /// All rules to apply
+ pub rules: Vec<Arc<dyn PhysicalAnalyzerRule + Send + Sync>>,
+}
+
+impl Default for PhysicalAnalyzer {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl PhysicalAnalyzer {
+ /// Create a new analyzer using the recommended list of rules
+ pub fn new() -> Self {
+ let rules: Vec<Arc<dyn PhysicalAnalyzerRule + Send + Sync>> = vec
+/// rules; see [`PhysicalAnalyzerRule`] for how the default planner places that
+/// phase (it is not run strictly before every optimizer rule).
+#[derive(Clone, Debug)]
+pub struct PhysicalAnalyzer {
+ /// All rules to apply
+ pub rules: Vec<Arc<dyn PhysicalAnalyzerRule + Send + Sync>>,
+}
+
+impl Default for PhysicalAnalyzer {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl PhysicalAnalyzer {
+ /// Create a new analyzer using the recommended list of rules
+ pub fn new() -> Self {
+ let rules: Vec<Arc<dyn PhysicalAnalyzerRule + Send + Sync>> = vec![
+ // Ensures each input plan satisfies the distribution and ordering
+ // requirements declared by
`ExecutionPlan::required_input_distribution`
+ // and `ExecutionPlan::required_input_ordering`.
+ //
+ // If the requirements are already satisfied, this rule leaves the
plan
+ // unchanged. For example, it does not add sorting when the input
is a
+ // file scan whose existing order already satisfies the required
ordering.
+ // Otherwise, this rule inserts the necessary repartitioning and
sorting
+ // operators.
+ //
+ // This used to be implemented as two separate rules:
`EnforceDistribution`
Review Comment:
it isn't clear to me why the history of having 2 separate rules is important
for future readers to know (I think we could pare this one down)
##########
datafusion/session/src/physical_analyzer.rs:
##########
@@ -0,0 +1,104 @@
+// 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).
+///
+/// # Ordering
+///
+/// Analyzer rules run as their own phase, conceptually before the optimizer
+/// rules that assume a valid plan. Note that the built-in planner does not
+/// literally run every analyzer before every optimizer: to preserve the
+/// hand-tuned order of the default pipeline it runs the analyzer phase at the
+/// position `EnsureRequirements` historically occupied (see
+/// [`DefaultPhysicalPlanner::optimize_physical_plan`]), because some default
+/// optimizer rules (notably join selection) must run before enforcement. A
+/// custom rule should therefore not assume it sees the raw initial plan, only
+/// that requirement enforcement has not yet run when it does.
+///
+/// [`DefaultPhysicalPlanner::optimize_physical_plan`]:
https://docs.rs/datafusion/latest/datafusion/physical_planner/struct.DefaultPhysicalPlanner.html#method.optimize_physical_plan
+///
+/// [`PhysicalOptimizerRule`]: crate::physical_optimizer::PhysicalOptimizerRule
+/// [`ExecutionPlan::required_input_distribution`]:
datafusion_physical_plan::ExecutionPlan::required_input_distribution
+/// [`ExecutionPlan::required_input_ordering`]:
datafusion_physical_plan::ExecutionPlan::required_input_ordering
+pub trait PhysicalAnalyzerRule: Debug + std::any::Any {
+ /// Rewrite `plan` so that it satisfies the invariants this rule enforces.
+ ///
+ /// This is the primary method. Rules that need access to the statistics
+ /// registry should override
[`analyze_with_context`](Self::analyze_with_context)
+ /// instead.
+ fn analyze(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>>;
+
+ /// Rewrite `plan` with access to extended context (statistics registry,
etc.).
+ ///
+ /// The default implementation calls [`analyze`](Self::analyze) with the
+ /// config options from the context. This mirrors
+ /// [`PhysicalOptimizerRule::optimize_with_context`], so enforcement passes
+ /// keep the same statistics-registry access they had while they were
+ /// optimizer rules.
+ ///
+ /// [`PhysicalOptimizerRule::optimize_with_context`]:
crate::physical_optimizer::PhysicalOptimizerRule::optimize_with_context
+ fn analyze_with_context(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ context: &dyn PhysicalOptimizerContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ self.analyze(plan, context.config_options())
+ }
+
+ /// A human readable name for this analyzer rule.
+ fn name(&self) -> &str;
+
+ /// A flag to indicate whether the physical planner should validate that
+ /// the rule will not change the schema of the plan after the rewrite.
+ ///
+ /// This mirrors [`PhysicalOptimizerRule::schema_check`]; enforcement
passes
Review Comment:
FWIW I think schema_check should be true for all optimizer rules and only
analyzer rules should be alloed to change the schema, given the definition of
analyzer / optimizer we are adding. We could try and tighten this up as a
follow on issue / work -- no need to do it here
##########
datafusion/session/src/physical_analyzer.rs:
##########
@@ -0,0 +1,104 @@
+// 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).
+///
+/// # Ordering
+///
+/// Analyzer rules run as their own phase, conceptually before the optimizer
+/// rules that assume a valid plan. Note that the built-in planner does not
+/// literally run every analyzer before every optimizer: to preserve the
Review Comment:
As mentioned elsewhere I think we should change this
##########
datafusion/session/src/physical_analyzer.rs:
##########
@@ -0,0 +1,104 @@
+// 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
Review Comment:
I recommend making these doc links to ANalyzerRule and OptimizerRule (they
probably need to be links to the docs.rs page due to crate dependencies)
--
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]
