zhuqi-lucas commented on code in PR #25688: URL: https://github.com/apache/datafusion/pull/25688#discussion_r4155684195
########## datafusion/physical-optimizer/src/analyzer.rs: ########## @@ -0,0 +1,153 @@ +// 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::{EnforceDistribution, EnforceSorting}; +use crate::output_requirements::OutputRequirements; + +// 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. The +/// default planner runs the whole analyzer phase before the +/// [`PhysicalOptimizer`](crate::optimizer::PhysicalOptimizer) phase, so every +/// optimizer rule can assume it receives a valid plan; see +/// [`PhysicalAnalyzerRule`] for details. +#[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 { + // All enforcement runs here, first, as the analyzer phase: it makes the + // plan *valid* (distribution first, then ordering) before any optimizer + // rule sees it, mirroring the logical Analyzer/Optimizer split. The + // optimizer rules that change these requirements (`JoinSelection`, + // `WindowTopN`, `FilterPushdown`) re-establish validity themselves, so + // enforcement is not repeated as an optimizer pass. The sort + // *optimizations* (`OptimizeSorts`) are not enforcement and stay in the + // optimizer phase. + let rules: Vec<Arc<dyn PhysicalAnalyzerRule + Send + Sync>> = vec![ + // Establish the output-requirement boundary first, so enforcement can + // see it (parallelize top-level scans below it, preserve the query's + // final ordering). The matching remove pass runs late in the optimizer. + Arc::new(OutputRequirements::new_add_mode()), + Arc::new(EnforceDistribution::new()), Review Comment: Thanks @alamb. So that a rule which disturbs only one of them can call the smallest repair it needs rather than a full pass: `JoinSelection` and `FilterPushdown` re-establish distribution only, since neither changes any operator ordering. On your related worry about the regression, I profiled it and the split is not where it goes. Across the two `enforce_distribution_requirements` calls per plan on tpch q21: phase 0 interleave normalisation 2.3 us, phase 1 join-key reordering 6.6 us, phase 2a distribution 292 us. Combining the two rules again would not recover that, since distribution and sorting are separate tree walks inside `EnsureRequirements` today as well. I will post the full attribution on the PR. ########## datafusion/sqllogictest/test_files/window_topn.slt: ########## @@ -1575,11 +1575,9 @@ logical_plan physical_plan 01)ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, row_number() PARTITION BY [t.c1] ORDER BY [t.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn] 02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [t.c1] ORDER BY [t.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [t.c1] ORDER BY [t.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] -03)----SortPreservingMergeExec: [c1@0 ASC NULLS LAST, c2@1 DESC] -04)------SortExec: expr=[c1@0 ASC NULLS LAST, c2@1 DESC], preserve_partitioning=[true] -05)--------PartitionedTopKExec: fn=row_number, fetch=1, partition=[c1@0], order=[c2@1 DESC] -06)----------RepartitionExec: partitioning=Hash([c1@0], 5), input_partitions=1 -07)------------DataSourceExec: partitions=1, partition_sizes=[1] +03)----PartitionedTopKExec: fn=row_number, fetch=1, partition=[c1@0], order=[c2@1 DESC] Review Comment: Thanks @alamb. It goes the other way, parallelism increases here. `SortPreservingMergeExec` outputs a single partition, so before this change `BoundedWindowAggExec` ran on one partition. Now it sits directly on the 5-way hash repartition and runs across all five. The `SortExec` and the merge became redundant rather than being dropped: `PartitionedTopKExec` carries `partition=[c1] order=[c2 DESC]`, so it already emits rows grouped by `c1` and ordered by `c2` descending, and hashing on `c1` keeps each group inside one partition, which is what `mode=[Sorted]` needs. It passes `SanityCheckPlan` and the query results are unchanged. -- 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]
