zhuqi-lucas commented on code in PR #26757:
URL: https://github.com/apache/datafusion/pull/26757#discussion_r4430936718
##########
datafusion/session/src/physical_analyzer.rs:
##########
@@ -0,0 +1,114 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF)
under one or more contributor license agreements
+
+//! 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` runs before requirements are enforced. The
+/// analyzer phase runs the rules that must see the plan while each
+/// operator's [`ExecutionPlan::required_input_distribution`] and
+/// [`ExecutionPlan::required_input_ordering`] are only declared, then
+/// `EnsureRequirements`, which repartitions and sorts the inputs to meet them.
+/// Every [`PhysicalOptimizerRule`] after it receives a plan that meets its
+/// requirements and must return one.
+///
+/// A rule belongs in the analyzer when `EnsureRequirements` reads what the
+/// rule changes (a requirement, an ordering, or statistics), as with the join
+/// mode `join_selection` picks, or when the rule needs the plan before
+/// enforcement adds repartitions and sorts, as with `WindowTopN`. A rule that
+/// reads what enforcement produced, such as `LimitPushdown`, is a
+/// [`PhysicalOptimizerRule`]. Both phases hold
+/// performance rules; what differs is the plan each one receives. A rule is
+/// one or the other: moving it to the other phase means implementing the
+/// other trait.
Review Comment:
On #25688 the plan, following @alamb's suggestion there, was to keep today's
order here and move `aggregate_statistics`, `LimitedDistinctAggregation`,
`FilterPushdown` and `WindowTopN` into the optimizer in follow-ups, with
`join_selection` after its own design (#26481). Dropping their
`PhysicalOptimizerRule` impls turns each of those moves into another API break,
the double break you want to avoid. #25688 kept both impls, the analyzer one
delegating, so a move is just a list change. Could we keep that, and word this
paragraph as transitional?
##########
datafusion/core/tests/physical_optimizer/eliminate_constant_join.rs:
##########
@@ -108,8 +108,8 @@ async fn keeps_join_with_lambda_filter() -> Result<()> {
.create_physical_plan()
.await?;
- let optimized = EliminateConstantJoin::new()
- .optimize(Arc::clone(&join), &ConfigOptions::new())?;
+ let optimized =
+ EliminateConstantJoin::new().analyze(Arc::clone(&join),
&ConfigOptions::new())?;
Review Comment:
`metadata_session(false)` below filters `EliminateConstantJoin` out of
`physical_optimizers()`, which no longer holds it, so both sessions eliminate
the join and `keeps_metadata_of_constant_input` compares the rewrite with
itself (I printed both plans: neither has the join).
`proto/tests/cases/plans/misc.rs:673` ("No optimizations") and
`ffi/integration-tests/tests/cases/query_planner.rs:250` ("observe planning
alone") also clear only the optimizer list.
##########
datafusion/physical-optimizer/src/filter_pushdown.rs:
##########
@@ -367,56 +368,75 @@ use itertools::{Itertools, izip};
/// [`FilterExec`]: datafusion_physical_plan::filter::FilterExec
/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
-#[derive(Debug)]
-pub struct FilterPushdown {
- phase: FilterPushdownPhase,
- name: String,
-}
+#[derive(Debug, Default)]
+pub struct FilterPushdown {}
impl FilterPushdown {
- fn new_with_phase(phase: FilterPushdownPhase) -> Self {
- let name = match phase {
- FilterPushdownPhase::Pre => "FilterPushdown",
- FilterPushdownPhase::Post => "FilterPushdown(Post)",
- }
- .to_string();
- Self { phase, name }
+ /// Create a new [`FilterPushdown`] rule, which pushes static filters
+ /// ([`FilterPushdownPhase::Pre`]).
+ pub fn new() -> Self {
+ Self {}
}
+}
- /// Create a new [`FilterPushdown`] optimizer rule that runs in the
pre-optimization phase.
- /// See [`FilterPushdownPhase`] for more details.
+/// Pushes dynamic filters, such as those a TopK `SortExec` or a `HashJoinExec`
+/// creates, into the operators that can use them
+/// ([`FilterPushdownPhase::Post`]). It runs after the other optimizer rules,
so
+/// that later rewrites do not break the links between a filter's producer and
+/// its consumers. See [`FilterPushdown`] for how pushdown works.
+#[derive(Debug, Default)]
+pub struct DynamicFilterPushdown {}
+
+impl DynamicFilterPushdown {
Review Comment:
Per the API health guide, could `FilterPushdown::new_post_optimization` and
`OutputRequirements::new_remove_mode` stay as `#[deprecated]` shims returning
the new types? Both are still optimizer rules, so existing lists keep compiling.
##########
datafusion/ffi/src/session/mod.rs:
##########
@@ -153,6 +155,9 @@ pub(crate) struct FFI_SessionRef {
/// Utility to identify when FFI objects are accessed locally through
/// the foreign interface.
pub library_marker_id: extern "C" fn() -> usize,
+
+ /// The session's physical analyzer rules.
+ physical_analyzers: unsafe extern "C" fn(&Self) ->
SVec<FFI_PhysicalAnalyzerRule>,
Review Comment:
With this field the struct no longer matches DataFusion 55, but the comment
on `create_physical_plan` above still says that slot is kept for 55
compatibility.
--
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]
