From 82edbc3b2c6a77922a40d5f467565abc48456c07 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Tue, 18 Aug 2026 13:22:41 +0800
Subject: [PATCH 01/10] perf: reuse the projected equivalence group when only
 child orderings change
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`replace_children_if_necessary` already short-circuits two cases: identical
child pointers, and identical child `PlanProperties` pointers. Inserting a
sort below a projection satisfies neither — the child is a new object, so its
properties pointer differs — yet the child's equivalence group is unchanged.
Sorting changes which orderings hold, not which expressions are equal.

`Recompute` therefore re-projects a group identical to the one the projection
already holds. `EquivalenceGroup::project` is a pure function of the group and
the mapping, so when both are unchanged the previous result can be handed back
instead.

`ProjectionExec::replace_children` now compares the old and new child
equivalence groups on the `Recompute` path and, when they match, reuses the
cached group and derives only the orderings. The check is local to
`ProjectionExec`: no new `ChildrenPropertiesMode` variant, and no change to the
shared path, so other operators are unaffected.

Supporting changes: `EquivalenceGroup` gains `PartialEq` (comparing `classes`,
since `map` is an index into them), and `EquivalenceProperties::project` splits
so `project_with_eq_group` can take an already-projected group.

Measured on a query over a 1191-line view with 38 SELECTs, 117 CASE
expressions and 7 joins across 11 tables. Ten warm samples per
configuration, same build flags, the only variable being this patch:

                      before (median)   after (median)
    EnforceSorting          197.8ms           90.1ms   -54%
    optimizer rules         338.1ms          214.6ms   -37%
    planning wall           420.6ms          299.0ms   -29%

The two ranges do not overlap (196.9-199.2 against 88.8-91.0). Logical rules
are unchanged, which is the control: the saving lands in the physical phase and
nowhere else. `EnforceDistribution` improves as well, since it rebuilds the
same projections.

The saving scales with projection count times expression size times rule
passes, so plans with narrow projections will see little.
---
 .../physical-expr/src/equivalence/class.rs    |  8 +++
 .../src/equivalence/properties/mod.rs         | 17 +++++
 datafusion/physical-plan/src/projection.rs    | 71 +++++++++++++++++--
 3 files changed, 90 insertions(+), 6 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 1f9a6a583cc44..1b0ccf6dfdfb4 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -309,6 +309,14 @@ pub struct EquivalenceGroup {
     classes: Vec<EquivalenceClass>,
 }
 
+impl PartialEq for EquivalenceGroup {
+    /// Compares the equivalence classes. `map` is an index into `classes`, so it
+    /// carries no information the classes themselves do not already have.
+    fn eq(&self, other: &Self) -> bool {
+        self.classes == other.classes
+    }
+}
+
 impl EquivalenceGroup {
     /// Creates an equivalence group from the given equivalence classes.
     pub fn new(classes: impl IntoIterator<Item = EquivalenceClass>) -> Self {
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index 54269e07f9309..a356743801e1e 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1169,6 +1169,23 @@ impl EquivalenceProperties {
     /// `output_schema`.
     pub fn project(&self, mapping: &ProjectionMapping, output_schema: SchemaRef) -> Self {
         let eq_group = self.eq_group.project(mapping);
+        self.project_with_eq_group(mapping, output_schema, eq_group)
+    }
+
+    /// Same as [`Self::project`], but takes an already-projected equivalence
+    /// group instead of computing one.
+    ///
+    /// [`EquivalenceGroup::project`] is a pure function of the group and the
+    /// mapping, so a caller that knows both are unchanged since the last
+    /// projection can hand back the previous result rather than recomputing an
+    /// identical one. Orderings are still derived here: they are precisely what
+    /// changes when a sort is introduced below this node.
+    pub fn project_with_eq_group(
+        &self,
+        mapping: &ProjectionMapping,
+        output_schema: SchemaRef,
+        eq_group: EquivalenceGroup,
+    ) -> Self {
         let orderings =
             self.projected_orderings(mapping, self.oeq_cache.normal_cls.clone());
         let normal_orderings = orderings
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index cf362cdee55d3..688f3b591796b 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -52,7 +52,7 @@ use datafusion_common::tree_node::{
 use datafusion_common::{DataFusionError, JoinSide, Result, internal_err, plan_err};
 use datafusion_execution::TaskContext;
 use datafusion_expr::ExpressionPlacement;
-use datafusion_physical_expr::equivalence::ProjectionMapping;
+use datafusion_physical_expr::equivalence::{EquivalenceGroup, ProjectionMapping};
 use datafusion_physical_expr::projection::Projector;
 use datafusion_physical_expr_common::physical_expr::{PhysicalExprRef, fmt_sql};
 use datafusion_physical_expr_common::sort_expr::{
@@ -194,6 +194,45 @@ impl ProjectionExec {
         })
     }
 
+    /// Like [`Self::try_from_projector`], but reuses `eq_group` as the output
+    /// equivalence group instead of projecting the input's group again.
+    ///
+    /// Only sound when the caller has established that the input's equivalence
+    /// group and this projection's mapping are both unchanged, since
+    /// [`EquivalenceGroup::project`] is a pure function of the two.
+    fn try_from_projector_reusing_eq_group(
+        projector: Projector,
+        input: Arc<dyn ExecutionPlan>,
+        eq_group: EquivalenceGroup,
+    ) -> Result<Self> {
+        let projection_mapping =
+            projector.projection().projection_mapping(&input.schema())?;
+        let input_eq_properties = input.equivalence_properties();
+        let eq_properties = input_eq_properties.project_with_eq_group(
+            &projection_mapping,
+            Arc::clone(projector.output_schema()),
+            eq_group,
+        );
+        // Partitioning is projected against the *input's* equivalence
+        // properties, matching `compute_properties`: the question is which
+        // input columns remain interchangeable, not which output ones do.
+        let output_partitioning = input
+            .output_partitioning()
+            .project(&projection_mapping, input_eq_properties);
+        let cache = PlanProperties::new(
+            eq_properties,
+            output_partitioning,
+            input.pipeline_behavior(),
+            input.boundedness(),
+        );
+        Ok(Self {
+            projector,
+            input,
+            metrics: ExecutionPlanMetricsSet::new(),
+            cache: Arc::new(cache),
+        })
+    }
+
     /// The projection expressions stored as tuples of (expression, output column name)
     pub fn expr(&self) -> &[ProjectionExpr] {
         self.projector.projection().as_ref()
@@ -352,11 +391,31 @@ impl ExecutionPlan for ProjectionExec {
                 metrics: ExecutionPlanMetricsSet::new(),
                 ..Self::clone(&*self)
             })),
-            ChildrenPropertiesMode::Recompute => ProjectionExec::try_from_projector(
-                self.projector.clone(),
-                children.swap_remove(0),
-            )
-            .map(|p| Arc::new(p) as _),
+            ChildrenPropertiesMode::Recompute => {
+                // `Keep` above requires the child's properties to be unchanged
+                // outright. A rule that introduces a sort below this projection
+                // does not qualify, yet the child's *equivalence group* is still
+                // identical: sorting changes which orderings hold, not which
+                // expressions are equal to one another. Projecting that group
+                // again would reproduce the group already cached here, so reuse
+                // it and derive only the orderings.
+                if self.input.equivalence_properties().eq_group()
+                    == children[0].equivalence_properties().eq_group()
+                {
+                    let eq_group = self.cache.equivalence_properties().eq_group().clone();
+                    return ProjectionExec::try_from_projector_reusing_eq_group(
+                        self.projector.clone(),
+                        children.swap_remove(0),
+                        eq_group,
+                    )
+                    .map(|p| Arc::new(p) as _);
+                }
+                ProjectionExec::try_from_projector(
+                    self.projector.clone(),
+                    children.swap_remove(0),
+                )
+                .map(|p| Arc::new(p) as _)
+            }
         }
     }
 

From b428ba2459b19dba05e5bce709b5dc7275e8963c Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 10:20:31 +0800
Subject: [PATCH 02/10] test: cover the equivalence-group reuse on child
 replacement

Adds unit tests for the fast path and the invariant it rests on.

`EquivalenceGroup`'s new `PartialEq`: equal classes compare equal however
they were arrived at (a bridged `a = b`, `b = c` matches a directly stated
one), member order within a class is immaterial, and differing or widened
classes compare unequal.

`EquivalenceProperties::project_with_eq_group`: handing back exactly the
group `project` would have computed reproduces it in full -- group,
orderings, constraints and schema all match. A second test passes an empty
group to confirm the argument is actually consumed rather than ignored.

`ProjectionExec::replace_children`: a sort below the projection changes the
orderings but not the equivalence group, and the resulting properties match
building the projection from scratch. A child that equates a different pair
must not inherit the cached group, which is the case that would be unsound;
inverting the guard to always reuse makes that test fail. `Keep` mode is
covered too, and a separate test pins the premise that sorting leaves the
equivalence group untouched, so a change in that behaviour fails there
first rather than silently weakening the fast path.
---
 .../physical-expr/src/equivalence/class.rs    |  66 +++++-
 .../src/equivalence/properties/mod.rs         | 106 ++++++++++
 datafusion/physical-plan/src/projection.rs    | 195 ++++++++++++++++++
 3 files changed, 366 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 1b0ccf6dfdfb4..268a7e9e60b16 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -932,7 +932,7 @@ mod tests {
     use super::*;
     use crate::equivalence::tests::create_test_params;
     use crate::expressions::{BinaryExpr, Column, binary, col, lit};
-    use arrow::datatypes::{DataType, Field, Schema};
+    use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 
     use datafusion_expr::Operator;
 
@@ -1253,4 +1253,68 @@ mod tests {
 
         Ok(())
     }
+
+    /// Builds a group from a list of equated column pairs.
+    fn group_of(schema: &SchemaRef, pairs: &[(&str, &str)]) -> Result<EquivalenceGroup> {
+        let mut group = EquivalenceGroup::default();
+        for (lhs, rhs) in pairs {
+            group.add_equal_conditions(col(lhs, schema)?, col(rhs, schema)?);
+        }
+        Ok(group)
+    }
+
+    fn abc_schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+            Field::new("c", DataType::Int32, false),
+        ]))
+    }
+
+    #[test]
+    fn test_equivalence_group_eq_compares_classes() -> Result<()> {
+        let schema = abc_schema();
+
+        // Two empty groups agree.
+        assert_eq!(group_of(&schema, &[])?, group_of(&schema, &[])?);
+        // The same class, built the same way.
+        assert_eq!(
+            group_of(&schema, &[("a", "b")])?,
+            group_of(&schema, &[("a", "b")])?
+        );
+        // A class is a set, so the order within a pair is immaterial.
+        assert_eq!(
+            group_of(&schema, &[("a", "b")])?,
+            group_of(&schema, &[("b", "a")])?
+        );
+        // A populated group is not an empty one.
+        assert_ne!(group_of(&schema, &[("a", "b")])?, group_of(&schema, &[])?);
+        // Equating a different pair yields a different group.
+        assert_ne!(
+            group_of(&schema, &[("a", "b")])?,
+            group_of(&schema, &[("a", "c")])?
+        );
+        // Widening a class yields a different group.
+        assert_ne!(
+            group_of(&schema, &[("a", "b")])?,
+            group_of(&schema, &[("a", "b"), ("b", "c")])?
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_equivalence_group_eq_ignores_the_map() -> Result<()> {
+        // `map` indexes into `classes`, so equal classes must imply equal
+        // groups no matter how the classes were arrived at. Bridging `a = b`
+        // and `b = c` into one class must match stating `a = c` and `a = b`.
+        let schema = abc_schema();
+        let bridged = group_of(&schema, &[("a", "b"), ("b", "c")])?;
+        let direct = group_of(&schema, &[("a", "c"), ("a", "b")])?;
+
+        assert_eq!(bridged.len(), 1, "expected a single bridged class");
+        assert_eq!(bridged, direct);
+
+        Ok(())
+    }
 }
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index a356743801e1e..673e1a76cb412 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1557,3 +1557,109 @@ fn get_expr_properties(
         expr.get_properties(&child_states)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use crate::equivalence::tests::create_test_params;
+    use crate::expressions::col;
+
+    use arrow::datatypes::{DataType, Field, Schema};
+
+    /// Renames `a`, `b`, `c` and `d` so the projection is non-trivial while
+    /// still carrying every ordering and the `a = c` class across.
+    fn renaming_mapping(
+        schema: &SchemaRef,
+        output_schema: &SchemaRef,
+    ) -> Result<ProjectionMapping> {
+        [
+            ("a", "a1", 0),
+            ("b", "b1", 1),
+            ("c", "c1", 2),
+            ("d", "d1", 3),
+        ]
+        .into_iter()
+        .map(|(source, target, index)| {
+            Ok((
+                col(source, schema)?,
+                vec![(col(target, output_schema)?, index)].into(),
+            ))
+        })
+        .collect::<Result<Vec<_>>>()
+        .map(|entries| entries.into_iter().collect())
+    }
+
+    fn renamed_schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![
+            Field::new("a1", DataType::Int32, true),
+            Field::new("b1", DataType::Int32, true),
+            Field::new("c1", DataType::Int32, true),
+            Field::new("d1", DataType::Int32, true),
+        ]))
+    }
+
+    #[test]
+    fn test_project_with_eq_group_matches_project() -> Result<()> {
+        let (schema, eq_properties) = create_test_params()?;
+        let output_schema = renamed_schema();
+        let mapping = renaming_mapping(&schema, &output_schema)?;
+
+        let baseline = eq_properties.project(&mapping, Arc::clone(&output_schema));
+        // Handing back exactly what `project` would have computed must
+        // reproduce it in full: this is the invariant the `ProjectionExec`
+        // fast path relies on.
+        let reused = eq_properties.project_with_eq_group(
+            &mapping,
+            Arc::clone(&output_schema),
+            eq_properties.eq_group().project(&mapping),
+        );
+
+        // Guard against a vacuous comparison.
+        assert!(
+            !baseline.eq_group().is_empty(),
+            "the projection dropped the equivalence class, nothing is being tested"
+        );
+        assert!(
+            !baseline.oeq_class().is_empty(),
+            "the projection dropped every ordering, nothing is being tested"
+        );
+
+        assert_eq!(baseline.eq_group(), reused.eq_group(), "equivalence group");
+        assert_eq!(baseline.oeq_class(), reused.oeq_class(), "orderings");
+        assert_eq!(baseline.constraints(), reused.constraints(), "constraints");
+        assert_eq!(baseline.schema(), reused.schema(), "schema");
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_project_with_eq_group_derives_orderings_from_the_caller_group() -> Result<()>
+    {
+        // `project_with_eq_group` must not silently ignore the group it is
+        // handed: orderings are normalized against it, so passing an empty
+        // group has to produce a different result than passing the real one.
+        let (schema, eq_properties) = create_test_params()?;
+        let output_schema = renamed_schema();
+        let mapping = renaming_mapping(&schema, &output_schema)?;
+
+        let with_real_group = eq_properties.project_with_eq_group(
+            &mapping,
+            Arc::clone(&output_schema),
+            eq_properties.eq_group().project(&mapping),
+        );
+        let with_empty_group = eq_properties.project_with_eq_group(
+            &mapping,
+            Arc::clone(&output_schema),
+            EquivalenceGroup::default(),
+        );
+
+        assert_ne!(
+            with_real_group.eq_group(),
+            with_empty_group.eq_group(),
+            "the supplied group was not used"
+        );
+
+        Ok(())
+    }
+}
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 688f3b591796b..041ce01d4d511 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -1495,6 +1495,8 @@ mod tests {
 
     use crate::common::collect;
     use crate::empty::EmptyExec;
+    use crate::filter::FilterExec;
+    use crate::sorts::sort::SortExec;
 
     use crate::filter_pushdown::PushedDown;
     use crate::statistics::{StatisticsArgs, StatisticsContext};
@@ -2274,4 +2276,197 @@ mod tests {
 
         Ok(())
     }
+
+    /// `EmptyExec(a, b, c)` under a filter that equates `lhs` and `rhs`, so the
+    /// child carries a non-trivial equivalence group.
+    fn filtered_source(lhs: &str, rhs: &str) -> Result<Arc<dyn ExecutionPlan>> {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+            Field::new("c", DataType::Int32, false),
+        ]));
+        let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+        let predicate = binary(
+            col(lhs, &schema)?,
+            Operator::Eq,
+            col(rhs, &schema)?,
+            &schema,
+        )?;
+        Ok(Arc::new(FilterExec::try_new(predicate, input)?))
+    }
+
+    /// `[a AS x, b AS y, c AS z]` against `filtered_source`'s schema.
+    fn renaming_exprs(schema: &SchemaRef) -> Result<Vec<ProjectionExpr>> {
+        [("a", "x"), ("b", "y"), ("c", "z")]
+            .into_iter()
+            .map(|(source, alias)| {
+                Ok(ProjectionExpr {
+                    expr: col(source, schema)?,
+                    alias: alias.to_string(),
+                })
+            })
+            .collect()
+    }
+
+    fn assert_same_properties(actual: &dyn ExecutionPlan, expected: &ProjectionExec) {
+        let actual_props = actual.properties().equivalence_properties();
+        let expected_props = expected.properties().equivalence_properties();
+        assert_eq!(
+            actual_props.eq_group(),
+            expected_props.eq_group(),
+            "equivalence group"
+        );
+        assert_eq!(
+            actual_props.oeq_class(),
+            expected_props.oeq_class(),
+            "orderings"
+        );
+        assert_eq!(
+            actual_props.constraints(),
+            expected_props.constraints(),
+            "constraints"
+        );
+        assert_eq!(actual_props.schema(), expected_props.schema(), "schema");
+        assert_eq!(
+            format!("{:?}", actual.properties().output_partitioning()),
+            format!("{:?}", expected.properties().output_partitioning()),
+            "partitioning"
+        );
+    }
+
+    #[test]
+    fn test_sort_below_changes_orderings_but_not_the_equivalence_group() -> Result<()> {
+        // The premise the fast path rests on. If a sort ever starts altering
+        // the equivalence group, reusing the cached group becomes unsound and
+        // this test is the one that should fail first.
+        let child = filtered_source("a", "b")?;
+        let ordering = LexOrdering::new(vec![PhysicalSortExpr::new_default(col(
+            "c",
+            &child.schema(),
+        )?)])
+        .expect("non-empty ordering");
+        let sorted = SortExec::new(ordering, Arc::clone(&child));
+
+        let child_props = child.properties().equivalence_properties();
+        let sorted_props = sorted.properties().equivalence_properties();
+
+        assert!(
+            !child_props.eq_group().is_empty(),
+            "the filter did not produce an equivalence class"
+        );
+        assert_eq!(
+            child_props.eq_group(),
+            sorted_props.eq_group(),
+            "sorting altered the equivalence group"
+        );
+        assert_ne!(
+            child_props.oeq_class(),
+            sorted_props.oeq_class(),
+            "sorting did not alter the orderings"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_replace_children_reuses_eq_group_when_only_orderings_change() -> Result<()> {
+        let child = filtered_source("a", "b")?;
+        let exprs = renaming_exprs(&child.schema())?;
+        let projection =
+            Arc::new(ProjectionExec::try_new(exprs.clone(), Arc::clone(&child))?);
+
+        // Sorting below the projection changes which orderings hold but leaves
+        // the equivalence group untouched -- the case the fast path targets.
+        let ordering = LexOrdering::new(vec![PhysicalSortExpr::new_default(col(
+            "c",
+            &child.schema(),
+        )?)])
+        .expect("non-empty ordering");
+        let sorted: Arc<dyn ExecutionPlan> =
+            Arc::new(SortExec::new(ordering, Arc::clone(&child)));
+
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&sorted)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
+
+        // Guard against vacuity: the group must be worth reusing, and the sort
+        // must genuinely have added an ordering the original did not have.
+        assert!(
+            !projection
+                .properties()
+                .equivalence_properties()
+                .eq_group()
+                .is_empty(),
+            "the projection carries no equivalence class, nothing is being reused"
+        );
+        assert!(
+            projection
+                .properties()
+                .equivalence_properties()
+                .oeq_class()
+                .is_empty(),
+            "the unsorted projection was already ordered"
+        );
+        assert!(
+            !replaced
+                .properties()
+                .equivalence_properties()
+                .oeq_class()
+                .is_empty(),
+            "the sort did not introduce an ordering"
+        );
+
+        // The fast path must agree with building the projection from scratch.
+        let expected = ProjectionExec::try_new(exprs, sorted)?;
+        assert_same_properties(replaced.as_ref(), &expected);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_replace_children_recomputes_when_eq_group_changes() -> Result<()> {
+        let child = filtered_source("a", "b")?;
+        let exprs = renaming_exprs(&child.schema())?;
+        let projection =
+            Arc::new(ProjectionExec::try_new(exprs.clone(), Arc::clone(&child))?);
+
+        // This child equates a different pair, so the cached group is stale and
+        // reusing it would be unsound: the guard has to fall through.
+        let other = filtered_source("a", "c")?;
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&other)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
+
+        assert_ne!(
+            replaced.properties().equivalence_properties().eq_group(),
+            projection.properties().equivalence_properties().eq_group(),
+            "the projection kept the previous child's equivalence group"
+        );
+
+        let expected = ProjectionExec::try_new(exprs, other)?;
+        assert_same_properties(replaced.as_ref(), &expected);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_replace_children_keep_mode_carries_properties_over() -> Result<()> {
+        // `Keep` is the caller's promise that the new child's properties match
+        // the old one's, so the cached properties must survive verbatim rather
+        // than being derived again.
+        let child = filtered_source("a", "b")?;
+        let exprs = renaming_exprs(&child.schema())?;
+        let projection = Arc::new(ProjectionExec::try_new(exprs, Arc::clone(&child))?);
+
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![filtered_source("a", "b")?],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Keep),
+        )?;
+
+        assert_same_properties(replaced.as_ref(), &projection);
+
+        Ok(())
+    }
 }

From df5df0cf1b6758989459c3b04b9d3c1884102663 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 12:03:13 +0800
Subject: [PATCH 03/10] review: compare equivalence groups as sets, and fix two
 weak tests

`EquivalenceGroup`'s new `PartialEq` compared `classes` positionally. `classes`
is a `Vec` but its order carries no meaning -- `remove_class_at_idx` uses
`swap_remove` -- so two groups describing exactly the same equalities could
compare unequal depending on how they were built. That is a trap for any caller
treating this as a semantic check, and it also made the projection fast path
needlessly conservative. It now compares the classes as the sets they are.

`test_project_with_eq_group_derives_orderings_from_the_caller_group` was
vacuous: it asserted on `eq_group()`, which stores the argument verbatim, so it
would have passed even if the group were ignored everywhere else. Asserting on
`oeq_class()` would not have helped either, since that is built from the
projected orderings and never consults the group. It now asserts on behaviour,
and picks the probe carefully: `projected_orderings` resolves `[a ASC]` to
`[c1 ASC]` by itself, so asking whether `c1` is ordered proves nothing. Asking
about `a1` does, because reaching that conclusion requires the supplied group to
say `a1 = c1`.

`assert_same_properties` compared partitioning through derived `Debug`, which
changes with any field addition. It now compares the partition count and the
explicit `Display` form.

Both fixes are covered: reverting the comparison to positional makes
`test_equivalence_group_eq_ignores_class_order` fail.
---
 .../physical-expr/src/equivalence/class.rs    | 36 +++++++++++++++++--
 .../src/equivalence/properties/mod.rs         | 25 ++++++++++---
 datafusion/physical-plan/src/projection.rs    | 14 ++++++--
 3 files changed, 66 insertions(+), 9 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 53a05817fff8e..de3dc9af9afef 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -310,10 +310,23 @@ pub struct EquivalenceGroup {
 }
 
 impl PartialEq for EquivalenceGroup {
-    /// Compares the equivalence classes. `map` is an index into `classes`, so it
-    /// carries no information the classes themselves do not already have.
+    /// Compares the equivalence classes as a set.
+    ///
+    /// `classes` is a `Vec`, but its order carries no meaning: `remove_class_at_idx`
+    /// uses `swap_remove`, so two groups describing exactly the same equalities can
+    /// hold their classes in different orders depending on how they were built. A
+    /// positional comparison would call those unequal, so this compares them as the
+    /// sets they are. The classes are distinct by construction, so equal lengths
+    /// plus containment in one direction is set equality.
+    ///
+    /// `map` is an index into `classes` and carries no information the classes do
+    /// not already have, so it takes no part in the comparison.
     fn eq(&self, other: &Self) -> bool {
-        self.classes == other.classes
+        self.classes.len() == other.classes.len()
+            && self
+                .classes
+                .iter()
+                .all(|class| other.classes.contains(class))
     }
 }
 
@@ -1268,9 +1281,26 @@ mod tests {
             Field::new("a", DataType::Int32, false),
             Field::new("b", DataType::Int32, false),
             Field::new("c", DataType::Int32, false),
+            Field::new("d", DataType::Int32, false),
         ]))
     }
 
+    #[test]
+    fn test_equivalence_group_eq_ignores_class_order() -> Result<()> {
+        // `classes` is a `Vec` and `remove_class_at_idx` uses `swap_remove`, so
+        // the order two groups hold their classes in depends on how they were
+        // built. Groups describing the same equalities must still compare equal,
+        // otherwise callers using this as a semantic check get false negatives.
+        let schema = abc_schema();
+        let ab_then_cd = group_of(&schema, &[("a", "b"), ("c", "d")])?;
+        let cd_then_ab = group_of(&schema, &[("c", "d"), ("a", "b")])?;
+
+        assert_eq!(ab_then_cd.len(), 2, "expected two disjoint classes");
+        assert_eq!(ab_then_cd, cd_then_ab);
+
+        Ok(())
+    }
+
     #[test]
     fn test_equivalence_group_eq_compares_classes() -> Result<()> {
         let schema = abc_schema();
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index 673e1a76cb412..5329b5a34adc0 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1565,6 +1565,7 @@ mod tests {
     use crate::equivalence::tests::create_test_params;
     use crate::expressions::col;
 
+    use arrow::compute::SortOptions;
     use arrow::datatypes::{DataType, Field, Schema};
 
     /// Renames `a`, `b`, `c` and `d` so the projection is non-trivial while
@@ -1654,10 +1655,26 @@ mod tests {
             EquivalenceGroup::default(),
         );
 
-        assert_ne!(
-            with_real_group.eq_group(),
-            with_empty_group.eq_group(),
-            "the supplied group was not used"
+        // Asserting on `eq_group()` would be vacuous, since the argument is
+        // stored there verbatim. Assert on behaviour instead.
+        //
+        // `projected_orderings` resolves `[a ASC]` to `[c1 ASC]` on its own, so
+        // asking about `c1` proves nothing. Asking about `a1` does: the stored
+        // ordering names `c1`, and concluding that `a1` is ordered too requires
+        // the supplied group to say `a1 = c1`.
+        let asc = SortOptions {
+            descending: false,
+            nulls_first: false,
+        };
+        let on_a1 = [PhysicalSortExpr::new(col("a1", &output_schema)?, asc)];
+
+        assert!(
+            with_real_group.ordering_satisfy(on_a1.clone())?,
+            "the supplied group was not used to answer ordering questions"
+        );
+        assert!(
+            !with_empty_group.ordering_satisfy(on_a1)?,
+            "an empty group still reported a1 as ordered, so the argument is ignored"
         );
 
         Ok(())
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 041ce01d4d511..97362fb3faf89 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -2327,9 +2327,19 @@ mod tests {
             "constraints"
         );
         assert_eq!(actual_props.schema(), expected_props.schema(), "schema");
+        // `Partitioning` has no `PartialEq`, so compare the partition count and
+        // the explicit `Display` form. Derived `Debug` would change with any
+        // field addition, making this brittle for no gain.
+        let actual_partitioning = actual.properties().output_partitioning();
+        let expected_partitioning = expected.properties().output_partitioning();
         assert_eq!(
-            format!("{:?}", actual.properties().output_partitioning()),
-            format!("{:?}", expected.properties().output_partitioning()),
+            actual_partitioning.partition_count(),
+            expected_partitioning.partition_count(),
+            "partition count"
+        );
+        assert_eq!(
+            actual_partitioning.to_string(),
+            expected_partitioning.to_string(),
             "partitioning"
         );
     }

From ee8ce6c055da14b0be5b8a3cf3ffee8eb881d12e Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 14:40:33 +0800
Subject: [PATCH 04/10] docs: spell out why the projection mapping is
 unchanged, and test it

The guard compares the old and new child equivalence groups, but
`EquivalenceGroup::project` is a pure function of *two* arguments. The
second, the projection mapping, was only implicitly unchanged. Document
why, and cover it.

The mapping comes from `projector.projection()`, carried over untouched, and
from the child's schema, which `ProjectionMapping::try_new` consults only for
field names and indices -- never for types or nullability. A child differing
only in nullability therefore keeps the same mapping. A child that renamed or
reordered those fields would change the group as well, since its members are
`Column`s carrying those names, so the comparison already rejects it; were one
to slip through, `try_new`'s name assertion errors out rather than letting a
stale group into the plan.

The new test swaps in a child differing only in nullability and asserts the
fast path agrees with `try_from_projector`, the path it replaces. It compares
against that rather than a freshly built projection deliberately:
`replace_children` carries the existing `Projector` over, so the output schema
stays as it was, while `try_new` derives a new one from the new child. That
difference belongs to `replace_children` and predates this change, so the
meaningful contract is that its two paths agree. Handing the fast path an
empty group makes the test fail.
---
 datafusion/physical-plan/src/projection.rs | 79 ++++++++++++++++++++--
 1 file changed, 73 insertions(+), 6 deletions(-)

diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 97362fb3faf89..e201dce32dd66 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -197,9 +197,19 @@ impl ProjectionExec {
     /// Like [`Self::try_from_projector`], but reuses `eq_group` as the output
     /// equivalence group instead of projecting the input's group again.
     ///
-    /// Only sound when the caller has established that the input's equivalence
-    /// group and this projection's mapping are both unchanged, since
-    /// [`EquivalenceGroup::project`] is a pure function of the two.
+    /// [`EquivalenceGroup::project`] is a pure function of the group and the
+    /// mapping, so reuse is sound exactly when both are unchanged.
+    ///
+    /// The caller establishes the first by comparing the old and new child
+    /// groups. The second holds because the mapping comes from
+    /// `projector.projection()`, carried over untouched, and from the child's
+    /// schema, which `ProjectionMapping::try_new` consults only for field names
+    /// and indices -- never for types or nullability. So a child differing only
+    /// in nullability keeps the same mapping. A child that renamed or reordered
+    /// those fields would change the group too, since its members are `Column`s
+    /// carrying those names, and the comparison above would reject it; were one
+    /// to slip through anyway, `try_new`'s name assertion errors out rather than
+    /// letting a stale group into the plan.
     fn try_from_projector_reusing_eq_group(
         projector: Projector,
         input: Arc<dyn ExecutionPlan>,
@@ -2280,10 +2290,20 @@ mod tests {
     /// `EmptyExec(a, b, c)` under a filter that equates `lhs` and `rhs`, so the
     /// child carries a non-trivial equivalence group.
     fn filtered_source(lhs: &str, rhs: &str) -> Result<Arc<dyn ExecutionPlan>> {
+        filtered_source_with_nullability(lhs, rhs, false)
+    }
+
+    /// As [`filtered_source`], but `nullable` varies the schema's nullability
+    /// while leaving field names and order alone.
+    fn filtered_source_with_nullability(
+        lhs: &str,
+        rhs: &str,
+        nullable: bool,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
         let schema = Arc::new(Schema::new(vec![
-            Field::new("a", DataType::Int32, false),
-            Field::new("b", DataType::Int32, false),
-            Field::new("c", DataType::Int32, false),
+            Field::new("a", DataType::Int32, nullable),
+            Field::new("b", DataType::Int32, nullable),
+            Field::new("c", DataType::Int32, nullable),
         ]));
         let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(Arc::clone(&schema)));
         let predicate = binary(
@@ -2434,6 +2454,53 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_replace_children_reuses_eq_group_across_a_nullability_change() -> Result<()> {
+        // Reuse is sound only if the projection mapping is unchanged as well.
+        // `ProjectionMapping::try_new` reads the child schema for field names
+        // and indices alone, so a child differing only in nullability keeps the
+        // same mapping and must still take the fast path.
+        //
+        // The comparison is against `try_from_projector`, the path this one
+        // replaces, rather than a freshly built projection: `replace_children`
+        // carries the existing `Projector` over, so the output schema stays as
+        // it was, while `try_new` would derive a new one from the new child.
+        // That difference is inherent to `replace_children` and not something
+        // this fast path introduces, so the meaningful contract is that the two
+        // `replace_children` paths agree.
+        let child = filtered_source("a", "b")?;
+        let exprs = renaming_exprs(&child.schema())?;
+        let projection = Arc::new(ProjectionExec::try_new(exprs, Arc::clone(&child))?);
+
+        let nullable_child = filtered_source_with_nullability("a", "b", true)?;
+        assert_ne!(
+            child.schema(),
+            nullable_child.schema(),
+            "the two children were meant to differ in nullability"
+        );
+        assert_eq!(
+            child.properties().equivalence_properties().eq_group(),
+            nullable_child
+                .properties()
+                .equivalence_properties()
+                .eq_group(),
+            "nullability moved the equivalence group, so the fast path is no longer under test"
+        );
+
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&nullable_child)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
+        let recomputed = ProjectionExec::try_from_projector(
+            projection.projector.clone(),
+            nullable_child,
+        )?;
+
+        assert_same_properties(replaced.as_ref(), &recomputed);
+
+        Ok(())
+    }
+
     #[test]
     fn test_replace_children_recomputes_when_eq_group_changes() -> Result<()> {
         let child = filtered_source("a", "b")?;

From 4013c26b12bdc78935fb83d0d4f1bc0f35e0ed3f Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 15:03:06 +0800
Subject: [PATCH 05/10] test: tighten nullability rather than loosen it in the
 reuse test

The test swapped in a child that made fields nullable. That is the one
direction `is_allowed_field_change` forbids a physical optimizer rule from
taking: it permits a field to become non-nullable, never the reverse. Testing
the fast path against an input the framework rules out proves less than it
appears to.

Swapping the other way exercises the same thing -- a child differing only in
nullability keeps the projection mapping identical, since
`ProjectionMapping::try_new` reads only field names and indices -- while
staying inside what a rule is allowed to do.

Worth recording why the cached output schema is not a hazard here. Both
`replace_children` paths carry the existing `Projector` over, so the output
schema does not track a child whose nullability changed. Since a rule may only
tighten, that cached schema can only ever be more conservative than the child,
never less, so it cannot claim non-null for data that carries nulls.
---
 datafusion/physical-plan/src/projection.rs | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index e201dce32dd66..bda22efc47c64 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -2461,6 +2461,9 @@ mod tests {
         // and indices alone, so a child differing only in nullability keeps the
         // same mapping and must still take the fast path.
         //
+        // The swap tightens nullability rather than loosening it, matching what
+        // `is_allowed_field_change` permits of a physical optimizer rule.
+        //
         // The comparison is against `try_from_projector`, the path this one
         // replaces, rather than a freshly built projection: `replace_children`
         // carries the existing `Projector` over, so the output schema stays as
@@ -2468,19 +2471,19 @@ mod tests {
         // That difference is inherent to `replace_children` and not something
         // this fast path introduces, so the meaningful contract is that the two
         // `replace_children` paths agree.
-        let child = filtered_source("a", "b")?;
+        let child = filtered_source_with_nullability("a", "b", true)?;
         let exprs = renaming_exprs(&child.schema())?;
         let projection = Arc::new(ProjectionExec::try_new(exprs, Arc::clone(&child))?);
 
-        let nullable_child = filtered_source_with_nullability("a", "b", true)?;
+        let tightened_child = filtered_source_with_nullability("a", "b", false)?;
         assert_ne!(
             child.schema(),
-            nullable_child.schema(),
+            tightened_child.schema(),
             "the two children were meant to differ in nullability"
         );
         assert_eq!(
             child.properties().equivalence_properties().eq_group(),
-            nullable_child
+            tightened_child
                 .properties()
                 .equivalence_properties()
                 .eq_group(),
@@ -2488,12 +2491,12 @@ mod tests {
         );
 
         let replaced = Arc::clone(&projection).replace_children(
-            vec![Arc::clone(&nullable_child)],
+            vec![Arc::clone(&tightened_child)],
             ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
         )?;
         let recomputed = ProjectionExec::try_from_projector(
             projection.projector.clone(),
-            nullable_child,
+            tightened_child,
         )?;
 
         assert_same_properties(replaced.as_ref(), &recomputed);

From 95cc5360b98bf7d506c9c004ccfe69917fda6c9e Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 15:40:21 +0800
Subject: [PATCH 06/10] perf: drop the set-wise group comparison that regressed
 clickbench planning

The equivalence-group comparison was changed to set semantics in review. That
turned an O(n) positional comparison into an O(n^2) scan on a hot path --
`ProjectionExec` consults it on every child replacement during physical
optimization -- and the benchmark bot caught it: `physical_plan_clickbench_all`
went from 129.7ms to 158.5ms against an unchanged 130.0ms baseline, with most
clickbench queries 1.22x to 1.31x slower. The run before that change showed no
regression at all.

Measured directly on the comparison, 20k iterations of two equal groups:

     classes   positional   set-wise   ratio
           8      121.9ms    267.5ms    2.2x
          16      204.2ms    899.3ms    4.4x
          32      404.1ms      3.23s    8.0x
          64      807.5ms     12.29s   15.2x

Positional is linear, set-wise quadratic, and the quadratic term costs far more
than the recomputation the fast path exists to avoid.

The review comment behind the change was about API shape rather than
correctness: a public `PartialEq` reads as semantic equality, and a positional
comparison of a `Vec` whose order is incidental is not that. Rather than pay
for set semantics, this drops `PartialEq` and exposes `has_same_classes`, whose
name claims only what it does. The doc records why it is positional and why the
one direction it can err in is harmless: a caller can be told the groups differ
when they match, never the reverse, so it only ever forfeits an optimization.
---
 .../physical-expr/src/equivalence/class.rs    | 85 +++++++++++--------
 .../src/equivalence/properties/mod.rs         |  5 +-
 datafusion/physical-plan/src/projection.rs    | 47 ++++++----
 3 files changed, 85 insertions(+), 52 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index de3dc9af9afef..6061c765e90a0 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -309,28 +309,34 @@ pub struct EquivalenceGroup {
     classes: Vec<EquivalenceClass>,
 }
 
-impl PartialEq for EquivalenceGroup {
-    /// Compares the equivalence classes as a set.
+impl EquivalenceGroup {
+    /// A cheap, deliberately conservative check that two groups hold the same
+    /// equivalence classes.
+    ///
+    /// This is not `PartialEq`, and the distinction is the point. `classes` is a
+    /// `Vec` whose order carries no meaning -- `remove_class_at_idx` uses
+    /// `swap_remove` -- so two groups describing exactly the same equalities can
+    /// hold their classes in different orders and this returns `false` for them.
+    /// Naming it `PartialEq` would invite callers to read it as semantic equality,
+    /// which it is not.
+    ///
+    /// The comparison is positional because it runs on a hot path:
+    /// `ProjectionExec` consults it every time a rule replaces its child. Set
+    /// semantics would mean scanning the other group per class, and the quadratic
+    /// blowup costs far more than the recomputation it is trying to avoid --
+    /// measured at 8x for 32 classes and 15x for 64.
     ///
-    /// `classes` is a `Vec`, but its order carries no meaning: `remove_class_at_idx`
-    /// uses `swap_remove`, so two groups describing exactly the same equalities can
-    /// hold their classes in different orders depending on how they were built. A
-    /// positional comparison would call those unequal, so this compares them as the
-    /// sets they are. The classes are distinct by construction, so equal lengths
-    /// plus containment in one direction is set equality.
+    /// Only the false direction is reachable: a group can be reported different
+    /// when it is not, never the same when it is not. Callers using this to skip
+    /// work must be built so that a `false` merely costs them that work, which is
+    /// exactly how the projection fast path uses it.
     ///
     /// `map` is an index into `classes` and carries no information the classes do
     /// not already have, so it takes no part in the comparison.
-    fn eq(&self, other: &Self) -> bool {
-        self.classes.len() == other.classes.len()
-            && self
-                .classes
-                .iter()
-                .all(|class| other.classes.contains(class))
+    pub fn has_same_classes(&self, other: &Self) -> bool {
+        self.classes == other.classes
     }
-}
 
-impl EquivalenceGroup {
     /// Creates an equivalence group from the given equivalence classes.
     pub fn new(classes: impl IntoIterator<Item = EquivalenceClass>) -> Self {
         classes.into_iter().collect::<Vec<_>>().into()
@@ -1286,55 +1292,62 @@ mod tests {
     }
 
     #[test]
-    fn test_equivalence_group_eq_ignores_class_order() -> Result<()> {
+    fn test_has_same_classes_is_conservative_about_class_order() -> Result<()> {
         // `classes` is a `Vec` and `remove_class_at_idx` uses `swap_remove`, so
         // the order two groups hold their classes in depends on how they were
-        // built. Groups describing the same equalities must still compare equal,
-        // otherwise callers using this as a semantic check get false negatives.
+        // built. This check is positional and reports such a pair as different.
+        //
+        // That is deliberate, and it is why this is not `PartialEq`: set
+        // semantics would mean scanning the other group per class, and the
+        // quadratic cost dwarfs the recomputation the caller is trying to skip.
+        // The error can only go this way -- different when they match, never the
+        // reverse -- so a caller only forfeits an optimization.
         let schema = abc_schema();
         let ab_then_cd = group_of(&schema, &[("a", "b"), ("c", "d")])?;
         let cd_then_ab = group_of(&schema, &[("c", "d"), ("a", "b")])?;
 
         assert_eq!(ab_then_cd.len(), 2, "expected two disjoint classes");
-        assert_eq!(ab_then_cd, cd_then_ab);
+        assert!(!ab_then_cd.has_same_classes(&cd_then_ab));
 
         Ok(())
     }
 
     #[test]
-    fn test_equivalence_group_eq_compares_classes() -> Result<()> {
+    fn test_has_same_classes_compares_classes() -> Result<()> {
         let schema = abc_schema();
 
         // Two empty groups agree.
-        assert_eq!(group_of(&schema, &[])?, group_of(&schema, &[])?);
+        assert!(group_of(&schema, &[])?.has_same_classes(&group_of(&schema, &[])?));
         // The same class, built the same way.
-        assert_eq!(
-            group_of(&schema, &[("a", "b")])?,
+        assert!(
             group_of(&schema, &[("a", "b")])?
+                .has_same_classes(&group_of(&schema, &[("a", "b")])?)
         );
         // A class is a set, so the order within a pair is immaterial.
-        assert_eq!(
-            group_of(&schema, &[("a", "b")])?,
-            group_of(&schema, &[("b", "a")])?
+        assert!(
+            group_of(&schema, &[("a", "b")])?
+                .has_same_classes(&group_of(&schema, &[("b", "a")])?)
         );
         // A populated group is not an empty one.
-        assert_ne!(group_of(&schema, &[("a", "b")])?, group_of(&schema, &[])?);
+        assert!(
+            !group_of(&schema, &[("a", "b")])?.has_same_classes(&group_of(&schema, &[])?)
+        );
         // Equating a different pair yields a different group.
-        assert_ne!(
-            group_of(&schema, &[("a", "b")])?,
-            group_of(&schema, &[("a", "c")])?
+        assert!(
+            !group_of(&schema, &[("a", "b")])?
+                .has_same_classes(&group_of(&schema, &[("a", "c")])?)
         );
         // Widening a class yields a different group.
-        assert_ne!(
-            group_of(&schema, &[("a", "b")])?,
-            group_of(&schema, &[("a", "b"), ("b", "c")])?
+        assert!(
+            !group_of(&schema, &[("a", "b")])?
+                .has_same_classes(&group_of(&schema, &[("a", "b"), ("b", "c")])?)
         );
 
         Ok(())
     }
 
     #[test]
-    fn test_equivalence_group_eq_ignores_the_map() -> Result<()> {
+    fn test_has_same_classes_ignores_the_map() -> Result<()> {
         // `map` indexes into `classes`, so equal classes must imply equal
         // groups no matter how the classes were arrived at. Bridging `a = b`
         // and `b = c` into one class must match stating `a = c` and `a = b`.
@@ -1343,7 +1356,7 @@ mod tests {
         let direct = group_of(&schema, &[("a", "c"), ("a", "b")])?;
 
         assert_eq!(bridged.len(), 1, "expected a single bridged class");
-        assert_eq!(bridged, direct);
+        assert!(bridged.has_same_classes(&direct));
 
         Ok(())
     }
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index 5329b5a34adc0..f207ddc37d027 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1626,7 +1626,10 @@ mod tests {
             "the projection dropped every ordering, nothing is being tested"
         );
 
-        assert_eq!(baseline.eq_group(), reused.eq_group(), "equivalence group");
+        assert!(
+            baseline.eq_group().has_same_classes(reused.eq_group()),
+            "equivalence group"
+        );
         assert_eq!(baseline.oeq_class(), reused.oeq_class(), "orderings");
         assert_eq!(baseline.constraints(), reused.constraints(), "constraints");
         assert_eq!(baseline.schema(), reused.schema(), "schema");
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index bda22efc47c64..c8164a2276d0e 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -409,8 +409,11 @@ impl ExecutionPlan for ProjectionExec {
                 // expressions are equal to one another. Projecting that group
                 // again would reproduce the group already cached here, so reuse
                 // it and derive only the orderings.
-                if self.input.equivalence_properties().eq_group()
-                    == children[0].equivalence_properties().eq_group()
+                if self
+                    .input
+                    .equivalence_properties()
+                    .eq_group()
+                    .has_same_classes(children[0].equivalence_properties().eq_group())
                 {
                     let eq_group = self.cache.equivalence_properties().eq_group().clone();
                     return ProjectionExec::try_from_projector_reusing_eq_group(
@@ -2331,10 +2334,13 @@ mod tests {
     fn assert_same_properties(actual: &dyn ExecutionPlan, expected: &ProjectionExec) {
         let actual_props = actual.properties().equivalence_properties();
         let expected_props = expected.properties().equivalence_properties();
-        assert_eq!(
+        assert!(
+            actual_props
+                .eq_group()
+                .has_same_classes(expected_props.eq_group()),
+            "equivalence group: {:?} vs {:?}",
             actual_props.eq_group(),
-            expected_props.eq_group(),
-            "equivalence group"
+            expected_props.eq_group()
         );
         assert_eq!(
             actual_props.oeq_class(),
@@ -2384,9 +2390,10 @@ mod tests {
             !child_props.eq_group().is_empty(),
             "the filter did not produce an equivalence class"
         );
-        assert_eq!(
-            child_props.eq_group(),
-            sorted_props.eq_group(),
+        assert!(
+            child_props
+                .eq_group()
+                .has_same_classes(sorted_props.eq_group()),
             "sorting altered the equivalence group"
         );
         assert_ne!(
@@ -2481,12 +2488,17 @@ mod tests {
             tightened_child.schema(),
             "the two children were meant to differ in nullability"
         );
-        assert_eq!(
-            child.properties().equivalence_properties().eq_group(),
-            tightened_child
+        assert!(
+            child
                 .properties()
                 .equivalence_properties()
-                .eq_group(),
+                .eq_group()
+                .has_same_classes(
+                    tightened_child
+                        .properties()
+                        .equivalence_properties()
+                        .eq_group()
+                ),
             "nullability moved the equivalence group, so the fast path is no longer under test"
         );
 
@@ -2519,9 +2531,14 @@ mod tests {
             ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
         )?;
 
-        assert_ne!(
-            replaced.properties().equivalence_properties().eq_group(),
-            projection.properties().equivalence_properties().eq_group(),
+        assert!(
+            !replaced
+                .properties()
+                .equivalence_properties()
+                .eq_group()
+                .has_same_classes(
+                    projection.properties().equivalence_properties().eq_group()
+                ),
             "the projection kept the previous child's equivalence group"
         );
 

From b0f4ea45374d2a43bece8f47dff28d992627ee90 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Thu, 20 Aug 2026 12:12:39 +0800
Subject: [PATCH 07/10] review: check the reuse precondition, and prove the
 tests exercise it

Three things from review.

`project_with_eq_group` is public and its precondition -- that `eq_group` is
`self.eq_group.project(mapping)` -- was unchecked, so a caller passing anything
else silently got equivalence properties that disagree with the projection. It
cannot be a hard error, since checking means performing the projection the
caller is skipping, so it is now a `debug_assert!`. `project` itself routes
around the check through a private unchecked entry point, since it builds the
group and satisfies the precondition by construction. A `should_panic` test
pins that the assertion is wired up.

The `replace_children` tests asserted the resulting properties were correct but
not that the fast path produced them, so deleting the fast path left them
passing. They now count reuses through a `cfg(test)` probe and assert the count:
one for the two cases that should reuse, zero for the case where the child's
equivalence group changed. Deleting the fast path now fails both of the first
two. The probe is thread local rather than a global counter, since the test
binary runs tests in parallel and a shared count would let one test observe
another's hits.

Dropped the measured multipliers from the `has_same_classes` doc. The reason the
comparison is positional does not depend on the exact numbers, and the numbers
would go stale; they live in the commit that introduced it instead.
---
 .../physical-expr/src/equivalence/class.rs    |  6 +-
 .../src/equivalence/properties/mod.rs         | 83 ++++++++++---------
 datafusion/physical-plan/src/projection.rs    | 70 +++++++++++++---
 3 files changed, 104 insertions(+), 55 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 6061c765e90a0..06f384ac2db03 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -322,9 +322,9 @@ impl EquivalenceGroup {
     ///
     /// The comparison is positional because it runs on a hot path:
     /// `ProjectionExec` consults it every time a rule replaces its child. Set
-    /// semantics would mean scanning the other group per class, and the quadratic
-    /// blowup costs far more than the recomputation it is trying to avoid --
-    /// measured at 8x for 32 classes and 15x for 64.
+    /// semantics would mean scanning the other group once per class, and that
+    /// quadratic term costs more than the recomputation the caller is trying to
+    /// skip, by a margin that widens with the number of classes.
     ///
     /// Only the false direction is reachable: a group can be reported different
     /// when it is not, never the same when it is not. Callers using this to skip
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index f207ddc37d027..b5944c4cd5698 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1169,7 +1169,10 @@ impl EquivalenceProperties {
     /// `output_schema`.
     pub fn project(&self, mapping: &ProjectionMapping, output_schema: SchemaRef) -> Self {
         let eq_group = self.eq_group.project(mapping);
-        self.project_with_eq_group(mapping, output_schema, eq_group)
+        // Built here, so it satisfies the precondition by construction; going
+        // through the checked entry point would reproject it under
+        // `debug_assertions` for nothing.
+        self.project_with_eq_group_unchecked(mapping, output_schema, eq_group)
     }
 
     /// Same as [`Self::project`], but takes an already-projected equivalence
@@ -1180,11 +1183,38 @@ impl EquivalenceProperties {
     /// projection can hand back the previous result rather than recomputing an
     /// identical one. Orderings are still derived here: they are precisely what
     /// changes when a sort is introduced below this node.
+    ///
+    /// # Preconditions
+    ///
+    /// `eq_group` must equal `self.eq_group.project(mapping)`. Anything else
+    /// yields equivalence properties that disagree with the projection, which
+    /// later rules will then reason from. The precondition cannot be enforced
+    /// cheaply -- checking it means performing the very projection the caller is
+    /// avoiding -- so it is asserted in debug builds only and left to the caller
+    /// in release.
     pub fn project_with_eq_group(
         &self,
         mapping: &ProjectionMapping,
         output_schema: SchemaRef,
         eq_group: EquivalenceGroup,
+    ) -> Self {
+        debug_assert!(
+            eq_group.has_same_classes(&self.eq_group.project(mapping)),
+            "project_with_eq_group was handed a group that is not the projection \
+             of this one: got {:?}, expected {:?}",
+            eq_group,
+            self.eq_group.project(mapping)
+        );
+        self.project_with_eq_group_unchecked(mapping, output_schema, eq_group)
+    }
+
+    /// [`Self::project_with_eq_group`] without the precondition check, for
+    /// callers that produced `eq_group` themselves.
+    fn project_with_eq_group_unchecked(
+        &self,
+        mapping: &ProjectionMapping,
+        output_schema: SchemaRef,
+        eq_group: EquivalenceGroup,
     ) -> Self {
         let orderings =
             self.projected_orderings(mapping, self.oeq_cache.normal_cls.clone());
@@ -1565,7 +1595,6 @@ mod tests {
     use crate::equivalence::tests::create_test_params;
     use crate::expressions::col;
 
-    use arrow::compute::SortOptions;
     use arrow::datatypes::{DataType, Field, Schema};
 
     /// Renames `a`, `b`, `c` and `d` so the projection is non-trivial while
@@ -1638,48 +1667,22 @@ mod tests {
     }
 
     #[test]
-    fn test_project_with_eq_group_derives_orderings_from_the_caller_group() -> Result<()>
-    {
-        // `project_with_eq_group` must not silently ignore the group it is
-        // handed: orderings are normalized against it, so passing an empty
-        // group has to produce a different result than passing the real one.
-        let (schema, eq_properties) = create_test_params()?;
+    #[cfg(debug_assertions)]
+    #[should_panic(expected = "not the projection of this one")]
+    fn test_project_with_eq_group_rejects_a_group_it_did_not_produce() {
+        // The precondition cannot be enforced cheaply in release -- checking it
+        // means performing the projection the caller is skipping -- so it is a
+        // debug assertion. This pins that the assertion is actually wired up,
+        // since a caller passing an unrelated group would otherwise get
+        // equivalence properties that silently disagree with the projection.
+        let (schema, eq_properties) = create_test_params().expect("test params");
         let output_schema = renamed_schema();
-        let mapping = renaming_mapping(&schema, &output_schema)?;
+        let mapping = renaming_mapping(&schema, &output_schema).expect("mapping");
 
-        let with_real_group = eq_properties.project_with_eq_group(
+        eq_properties.project_with_eq_group(
             &mapping,
-            Arc::clone(&output_schema),
-            eq_properties.eq_group().project(&mapping),
-        );
-        let with_empty_group = eq_properties.project_with_eq_group(
-            &mapping,
-            Arc::clone(&output_schema),
+            output_schema,
             EquivalenceGroup::default(),
         );
-
-        // Asserting on `eq_group()` would be vacuous, since the argument is
-        // stored there verbatim. Assert on behaviour instead.
-        //
-        // `projected_orderings` resolves `[a ASC]` to `[c1 ASC]` on its own, so
-        // asking about `c1` proves nothing. Asking about `a1` does: the stored
-        // ordering names `c1`, and concluding that `a1` is ordered too requires
-        // the supplied group to say `a1 = c1`.
-        let asc = SortOptions {
-            descending: false,
-            nulls_first: false,
-        };
-        let on_a1 = [PhysicalSortExpr::new(col("a1", &output_schema)?, asc)];
-
-        assert!(
-            with_real_group.ordering_satisfy(on_a1.clone())?,
-            "the supplied group was not used to answer ordering questions"
-        );
-        assert!(
-            !with_empty_group.ordering_satisfy(on_a1)?,
-            "an empty group still reported a1 as ordered, so the argument is ignored"
-        );
-
-        Ok(())
     }
 }
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index c8164a2276d0e..70e36b8f7836f 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -67,6 +67,31 @@ pub use datafusion_physical_expr::projection::{
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
+/// Counts how often [`ProjectionExec::replace_children`] reused a cached
+/// equivalence group instead of reprojecting it.
+///
+/// Thread local rather than a global counter: the test binary runs tests in
+/// parallel, and a shared count would let one test observe another's hits.
+#[cfg(test)]
+mod eq_group_reuse_probe {
+    use std::cell::Cell;
+
+    thread_local! {
+        static HITS: Cell<usize> = const { Cell::new(0) };
+    }
+
+    pub(super) fn record_hit() {
+        HITS.with(|h| h.set(h.get() + 1));
+    }
+
+    /// Runs `f` and reports how many reuses happened while it did.
+    pub(super) fn count<T>(f: impl FnOnce() -> T) -> (T, usize) {
+        let before = HITS.with(Cell::get);
+        let value = f();
+        (value, HITS.with(Cell::get) - before)
+    }
+}
+
 /// [`ExecutionPlan`] for a projection
 ///
 /// Computes a set of scalar value expressions for each input row, producing one
@@ -415,6 +440,8 @@ impl ExecutionPlan for ProjectionExec {
                     .eq_group()
                     .has_same_classes(children[0].equivalence_properties().eq_group())
                 {
+                    #[cfg(test)]
+                    eq_group_reuse_probe::record_hit();
                     let eq_group = self.cache.equivalence_properties().eq_group().clone();
                     return ProjectionExec::try_from_projector_reusing_eq_group(
                         self.projector.clone(),
@@ -2422,10 +2449,18 @@ mod tests {
         let sorted: Arc<dyn ExecutionPlan> =
             Arc::new(SortExec::new(ordering, Arc::clone(&child)));
 
-        let replaced = Arc::clone(&projection).replace_children(
-            vec![Arc::clone(&sorted)],
-            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-        )?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).replace_children(
+                vec![Arc::clone(&sorted)],
+                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+            )
+        });
+        let replaced = replaced?;
+        assert_eq!(
+            reuses, 1,
+            "expected the fast path to be taken exactly once; deleting it would \
+             otherwise leave this test passing"
+        );
 
         // Guard against vacuity: the group must be worth reusing, and the sort
         // must genuinely have added an ordering the original did not have.
@@ -2502,10 +2537,14 @@ mod tests {
             "nullability moved the equivalence group, so the fast path is no longer under test"
         );
 
-        let replaced = Arc::clone(&projection).replace_children(
-            vec![Arc::clone(&tightened_child)],
-            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-        )?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).replace_children(
+                vec![Arc::clone(&tightened_child)],
+                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+            )
+        });
+        let replaced = replaced?;
+        assert_eq!(reuses, 1, "expected the fast path to be taken");
         let recomputed = ProjectionExec::try_from_projector(
             projection.projector.clone(),
             tightened_child,
@@ -2526,10 +2565,17 @@ mod tests {
         // This child equates a different pair, so the cached group is stale and
         // reusing it would be unsound: the guard has to fall through.
         let other = filtered_source("a", "c")?;
-        let replaced = Arc::clone(&projection).replace_children(
-            vec![Arc::clone(&other)],
-            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-        )?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).replace_children(
+                vec![Arc::clone(&other)],
+                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+            )
+        });
+        let replaced = replaced?;
+        assert_eq!(
+            reuses, 0,
+            "the guard let a stale equivalence group through the fast path"
+        );
 
         assert!(
             !replaced

From 506d532b90a584142c85edbdd990a76227a35fe4 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Thu, 20 Aug 2026 23:30:01 +0800
Subject: [PATCH 08/10] review: fold the reuse path into compute_properties

The second constructor duplicated `compute_properties` -- partitioning,
pipeline behaviour, boundedness, the `PlanProperties` and `Self` literals were
all copied -- so a new field on `PlanProperties` would have had to be added
twice, and the two paths could drift apart silently.

They differed in exactly one expression: whether the equivalence group is
reprojected or handed back. That is now a parameter on `compute_properties`,
`try_from_projector` delegates with `None`, and there is a single body.
`replace_children` loses its early return and its second call site: the guard
just produces the `Option`.

No behaviour change. Deleting the fast path still fails the two tests that
assert it is taken.
---
 datafusion/physical-plan/src/projection.rs | 85 +++++++++-------------
 1 file changed, 34 insertions(+), 51 deletions(-)

diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 70e36b8f7836f..bc3f925829d5d 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -203,24 +203,12 @@ impl ProjectionExec {
         projector: Projector,
         input: Arc<dyn ExecutionPlan>,
     ) -> Result<Self> {
-        // Construct a map from the input expressions to the output expression of the Projection
-        let projection_mapping =
-            projector.projection().projection_mapping(&input.schema())?;
-        let cache = Self::compute_properties(
-            &input,
-            &projection_mapping,
-            Arc::clone(projector.output_schema()),
-        )?;
-        Ok(Self {
-            projector,
-            input,
-            metrics: ExecutionPlanMetricsSet::new(),
-            cache: Arc::new(cache),
-        })
+        Self::try_from_projector_with_eq_group(projector, input, None)
     }
 
-    /// Like [`Self::try_from_projector`], but reuses `eq_group` as the output
-    /// equivalence group instead of projecting the input's group again.
+    /// As [`Self::try_from_projector`], but `reused_eq_group` may carry an
+    /// already-projected equivalence group to install instead of projecting the
+    /// input's again.
     ///
     /// [`EquivalenceGroup::project`] is a pure function of the group and the
     /// mapping, so reuse is sound exactly when both are unchanged.
@@ -235,31 +223,20 @@ impl ProjectionExec {
     /// carrying those names, and the comparison above would reject it; were one
     /// to slip through anyway, `try_new`'s name assertion errors out rather than
     /// letting a stale group into the plan.
-    fn try_from_projector_reusing_eq_group(
+    fn try_from_projector_with_eq_group(
         projector: Projector,
         input: Arc<dyn ExecutionPlan>,
-        eq_group: EquivalenceGroup,
+        reused_eq_group: Option<EquivalenceGroup>,
     ) -> Result<Self> {
+        // Construct a map from the input expressions to the output expression of the Projection
         let projection_mapping =
             projector.projection().projection_mapping(&input.schema())?;
-        let input_eq_properties = input.equivalence_properties();
-        let eq_properties = input_eq_properties.project_with_eq_group(
+        let cache = Self::compute_properties(
+            &input,
             &projection_mapping,
             Arc::clone(projector.output_schema()),
-            eq_group,
-        );
-        // Partitioning is projected against the *input's* equivalence
-        // properties, matching `compute_properties`: the question is which
-        // input columns remain interchangeable, not which output ones do.
-        let output_partitioning = input
-            .output_partitioning()
-            .project(&projection_mapping, input_eq_properties);
-        let cache = PlanProperties::new(
-            eq_properties,
-            output_partitioning,
-            input.pipeline_behavior(),
-            input.boundedness(),
-        );
+            reused_eq_group,
+        )?;
         Ok(Self {
             projector,
             input,
@@ -288,10 +265,20 @@ impl ProjectionExec {
         input: &Arc<dyn ExecutionPlan>,
         projection_mapping: &ProjectionMapping,
         schema: SchemaRef,
+        reused_eq_group: Option<EquivalenceGroup>,
     ) -> Result<PlanProperties> {
-        // Calculate equivalence properties:
+        // Calculate equivalence properties. Whether the group is reprojected or
+        // handed back is the only thing reuse changes; everything below is
+        // common, so the two paths cannot drift apart.
         let input_eq_properties = input.equivalence_properties();
-        let eq_properties = input_eq_properties.project(projection_mapping, schema);
+        let eq_properties = match reused_eq_group {
+            Some(eq_group) => input_eq_properties.project_with_eq_group(
+                projection_mapping,
+                schema,
+                eq_group,
+            ),
+            None => input_eq_properties.project(projection_mapping, schema),
+        };
         // Calculate output partitioning, which needs to respect aliases:
         let output_partitioning = input
             .output_partitioning()
@@ -434,25 +421,21 @@ impl ExecutionPlan for ProjectionExec {
                 // expressions are equal to one another. Projecting that group
                 // again would reproduce the group already cached here, so reuse
                 // it and derive only the orderings.
-                if self
+                let child = children.swap_remove(0);
+                let reused_eq_group = self
                     .input
                     .equivalence_properties()
                     .eq_group()
-                    .has_same_classes(children[0].equivalence_properties().eq_group())
-                {
-                    #[cfg(test)]
-                    eq_group_reuse_probe::record_hit();
-                    let eq_group = self.cache.equivalence_properties().eq_group().clone();
-                    return ProjectionExec::try_from_projector_reusing_eq_group(
-                        self.projector.clone(),
-                        children.swap_remove(0),
-                        eq_group,
-                    )
-                    .map(|p| Arc::new(p) as _);
-                }
-                ProjectionExec::try_from_projector(
+                    .has_same_classes(child.equivalence_properties().eq_group())
+                    .then(|| {
+                        #[cfg(test)]
+                        eq_group_reuse_probe::record_hit();
+                        self.cache.equivalence_properties().eq_group().clone()
+                    });
+                ProjectionExec::try_from_projector_with_eq_group(
                     self.projector.clone(),
-                    children.swap_remove(0),
+                    child,
+                    reused_eq_group,
                 )
                 .map(|p| Arc::new(p) as _)
             }

From 392cb9311dd68ea9c3950a7530035e820e00ffce Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Fri, 21 Aug 2026 10:48:37 +0800
Subject: [PATCH 09/10] review: make the reuse decision inside
 EquivalenceProperties

`project_with_eq_group` took a group on trust and could only assert the
precondition in debug builds. `project_reusing` takes the properties the cached
projection was built from alongside the projection itself, compares the groups,
and falls back to a full projection when they differ. There is no longer a
precondition to violate, so the `debug_assert` and the unchecked entry point it
needed both go away.

One caveat stays and is documented: the caller must pass a `cached` that came
from projecting `previous` through this same mapping. `EquivalenceProperties`
cannot see the mapping's provenance. In practice it holds because
`ProjectionExec` carries its projector over untouched.

The reuse probe moves to `physical-expr` with the decision, since `cfg(test)`
does not reach across crates and the tests that assert the fast path is taken
have to live where the branch is. `physical-plan` keeps the tests that assert
the resulting properties are correct. Disabling the reuse still fails the
assertion that it happened.
---
 .../src/equivalence/properties/mod.rs         | 137 ++++++++++++------
 datafusion/physical-plan/src/projection.rs    | 113 +++++----------
 2 files changed, 125 insertions(+), 125 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index 7e6ec259aafe3..08c05efe0ccc0 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -194,6 +194,31 @@ impl OrderingEquivalenceCache {
     }
 }
 
+/// Counts how often [`EquivalenceProperties::project_reusing`] reused a cached
+/// equivalence group instead of reprojecting it.
+///
+/// Thread local rather than a global counter: the test binary runs tests in
+/// parallel, and a shared count would let one test observe another's hits.
+#[cfg(test)]
+mod eq_group_reuse_probe {
+    use std::cell::Cell;
+
+    thread_local! {
+        static HITS: Cell<usize> = const { Cell::new(0) };
+    }
+
+    pub(super) fn record_hit() {
+        HITS.with(|h| h.set(h.get() + 1));
+    }
+
+    /// Runs `f` and reports how many reuses happened while it did.
+    pub(super) fn count<T>(f: impl FnOnce() -> T) -> (T, usize) {
+        let before = HITS.with(Cell::get);
+        let value = f();
+        (value, HITS.with(Cell::get) - before)
+    }
+}
+
 impl EquivalenceProperties {
     /// Helper used by the ordering equivalence rule when considering whether
     /// an expression can replace an existing sort key without invalidating
@@ -1194,41 +1219,36 @@ impl EquivalenceProperties {
         self.project_with_eq_group_unchecked(mapping, output_schema, eq_group)
     }
 
-    /// Same as [`Self::project`], but takes an already-projected equivalence
-    /// group instead of computing one.
+    /// Projects `self`, reusing `cached`'s already-projected equivalence group
+    /// when `self`'s group is unchanged from `previous`'s.
     ///
     /// [`EquivalenceGroup::project`] is a pure function of the group and the
-    /// mapping, so a caller that knows both are unchanged since the last
-    /// projection can hand back the previous result rather than recomputing an
-    /// identical one. Orderings are still derived here: they are precisely what
-    /// changes when a sort is introduced below this node.
+    /// mapping, so when the group is unchanged the previous result can be handed
+    /// back rather than recomputed. Orderings are still derived here: they are
+    /// precisely what changes when a sort is introduced below this node.
     ///
-    /// # Preconditions
-    ///
-    /// `eq_group` must equal `self.eq_group.project(mapping)`. Anything else
-    /// yields equivalence properties that disagree with the projection, which
-    /// later rules will then reason from. The precondition cannot be enforced
-    /// cheaply -- checking it means performing the very projection the caller is
-    /// avoiding -- so it is asserted in debug builds only and left to the caller
-    /// in release.
-    pub fn project_with_eq_group(
+    /// Falls back to a full projection when the groups differ, so there is no
+    /// precondition to violate. The caller does still have to pass a `cached`
+    /// that came from projecting `previous` through this same `mapping`; that
+    /// part cannot be checked here, and in practice it holds because the caller
+    /// carries its projection over untouched.
+    pub fn project_reusing(
         &self,
         mapping: &ProjectionMapping,
         output_schema: SchemaRef,
-        eq_group: EquivalenceGroup,
+        previous: &EquivalenceProperties,
+        cached: &EquivalenceProperties,
     ) -> Self {
-        debug_assert!(
-            eq_group.has_same_classes(&self.eq_group.project(mapping)),
-            "project_with_eq_group was handed a group that is not the projection \
-             of this one: got {:?}, expected {:?}",
-            eq_group,
+        let eq_group = if self.eq_group.has_same_classes(&previous.eq_group) {
+            #[cfg(test)]
+            eq_group_reuse_probe::record_hit();
+            cached.eq_group.clone()
+        } else {
             self.eq_group.project(mapping)
-        );
+        };
         self.project_with_eq_group_unchecked(mapping, output_schema, eq_group)
     }
 
-    /// [`Self::project_with_eq_group`] without the precondition check, for
-    /// callers that produced `eq_group` themselves.
     fn project_with_eq_group_unchecked(
         &self,
         mapping: &ProjectionMapping,
@@ -1649,19 +1669,25 @@ mod tests {
     }
 
     #[test]
-    fn test_project_with_eq_group_matches_project() -> Result<()> {
+    fn test_project_reusing_matches_project() -> Result<()> {
         let (schema, eq_properties) = create_test_params()?;
         let output_schema = renamed_schema();
         let mapping = renaming_mapping(&schema, &output_schema)?;
 
         let baseline = eq_properties.project(&mapping, Arc::clone(&output_schema));
-        // Handing back exactly what `project` would have computed must
-        // reproduce it in full: this is the invariant the `ProjectionExec`
-        // fast path relies on.
-        let reused = eq_properties.project_with_eq_group(
-            &mapping,
-            Arc::clone(&output_schema),
-            eq_properties.eq_group().project(&mapping),
+
+        // Reusing the projection of an identical group must reproduce it exactly.
+        let (reused, hits) = eq_group_reuse_probe::count(|| {
+            eq_properties.project_reusing(
+                &mapping,
+                Arc::clone(&output_schema),
+                &eq_properties,
+                &baseline,
+            )
+        });
+        assert_eq!(
+            hits, 1,
+            "the reuse path was not taken, so this compares nothing"
         );
 
         // Guard against a vacuous comparison.
@@ -1686,22 +1712,41 @@ mod tests {
     }
 
     #[test]
-    #[cfg(debug_assertions)]
-    #[should_panic(expected = "not the projection of this one")]
-    fn test_project_with_eq_group_rejects_a_group_it_did_not_produce() {
-        // The precondition cannot be enforced cheaply in release -- checking it
-        // means performing the projection the caller is skipping -- so it is a
-        // debug assertion. This pins that the assertion is actually wired up,
-        // since a caller passing an unrelated group would otherwise get
-        // equivalence properties that silently disagree with the projection.
-        let (schema, eq_properties) = create_test_params().expect("test params");
+    fn test_project_reusing_falls_back_when_the_group_moved() -> Result<()> {
+        // A `previous` whose group differs must not have its projection carried
+        // over. There is no precondition to violate here: the fallback is what
+        // keeps a mismatched pair from producing properties that disagree with
+        // the projection.
+        let (schema, eq_properties) = create_test_params()?;
         let output_schema = renamed_schema();
-        let mapping = renaming_mapping(&schema, &output_schema).expect("mapping");
+        let mapping = renaming_mapping(&schema, &output_schema)?;
 
-        eq_properties.project_with_eq_group(
-            &mapping,
-            output_schema,
-            EquivalenceGroup::default(),
+        let baseline = eq_properties.project(&mapping, Arc::clone(&output_schema));
+        let unrelated = EquivalenceProperties::new(Arc::clone(&schema));
+        assert!(
+            !eq_properties
+                .eq_group()
+                .has_same_classes(unrelated.eq_group()),
+            "the two groups were meant to differ"
         );
+
+        let (recomputed, hits) = eq_group_reuse_probe::count(|| {
+            eq_properties.project_reusing(
+                &mapping,
+                Arc::clone(&output_schema),
+                &unrelated,
+                &baseline,
+            )
+        });
+        assert_eq!(hits, 0, "a mismatched group was carried over anyway");
+
+        // Falling back must land on exactly what `project` would have produced.
+        assert!(
+            baseline.eq_group().has_same_classes(recomputed.eq_group()),
+            "equivalence group"
+        );
+        assert_eq!(baseline.oeq_class(), recomputed.oeq_class(), "orderings");
+
+        Ok(())
     }
 }
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index bc3f925829d5d..b5b5159842796 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -52,7 +52,8 @@ use datafusion_common::tree_node::{
 use datafusion_common::{DataFusionError, JoinSide, Result, internal_err, plan_err};
 use datafusion_execution::TaskContext;
 use datafusion_expr::ExpressionPlacement;
-use datafusion_physical_expr::equivalence::{EquivalenceGroup, ProjectionMapping};
+use datafusion_physical_expr::EquivalenceProperties;
+use datafusion_physical_expr::equivalence::ProjectionMapping;
 use datafusion_physical_expr::projection::Projector;
 use datafusion_physical_expr_common::physical_expr::{PhysicalExprRef, fmt_sql};
 use datafusion_physical_expr_common::sort_expr::{
@@ -67,31 +68,6 @@ pub use datafusion_physical_expr::projection::{
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
-/// Counts how often [`ProjectionExec::replace_children`] reused a cached
-/// equivalence group instead of reprojecting it.
-///
-/// Thread local rather than a global counter: the test binary runs tests in
-/// parallel, and a shared count would let one test observe another's hits.
-#[cfg(test)]
-mod eq_group_reuse_probe {
-    use std::cell::Cell;
-
-    thread_local! {
-        static HITS: Cell<usize> = const { Cell::new(0) };
-    }
-
-    pub(super) fn record_hit() {
-        HITS.with(|h| h.set(h.get() + 1));
-    }
-
-    /// Runs `f` and reports how many reuses happened while it did.
-    pub(super) fn count<T>(f: impl FnOnce() -> T) -> (T, usize) {
-        let before = HITS.with(Cell::get);
-        let value = f();
-        (value, HITS.with(Cell::get) - before)
-    }
-}
-
 /// [`ExecutionPlan`] for a projection
 ///
 /// Computes a set of scalar value expressions for each input row, producing one
@@ -206,9 +182,10 @@ impl ProjectionExec {
         Self::try_from_projector_with_eq_group(projector, input, None)
     }
 
-    /// As [`Self::try_from_projector`], but `reused_eq_group` may carry an
-    /// already-projected equivalence group to install instead of projecting the
-    /// input's again.
+    /// As [`Self::try_from_projector`], but `reuse_from` may carry the previous
+    /// child's equivalence properties together with the projection they produced,
+    /// letting [`EquivalenceProperties::project_reusing`] skip reprojecting a
+    /// group that has not changed.
     ///
     /// [`EquivalenceGroup::project`] is a pure function of the group and the
     /// mapping, so reuse is sound exactly when both are unchanged.
@@ -226,7 +203,7 @@ impl ProjectionExec {
     fn try_from_projector_with_eq_group(
         projector: Projector,
         input: Arc<dyn ExecutionPlan>,
-        reused_eq_group: Option<EquivalenceGroup>,
+        reuse_from: Option<(&EquivalenceProperties, &EquivalenceProperties)>,
     ) -> Result<Self> {
         // Construct a map from the input expressions to the output expression of the Projection
         let projection_mapping =
@@ -235,7 +212,7 @@ impl ProjectionExec {
             &input,
             &projection_mapping,
             Arc::clone(projector.output_schema()),
-            reused_eq_group,
+            reuse_from,
         )?;
         Ok(Self {
             projector,
@@ -265,17 +242,18 @@ impl ProjectionExec {
         input: &Arc<dyn ExecutionPlan>,
         projection_mapping: &ProjectionMapping,
         schema: SchemaRef,
-        reused_eq_group: Option<EquivalenceGroup>,
+        reuse_from: Option<(&EquivalenceProperties, &EquivalenceProperties)>,
     ) -> Result<PlanProperties> {
         // Calculate equivalence properties. Whether the group is reprojected or
         // handed back is the only thing reuse changes; everything below is
         // common, so the two paths cannot drift apart.
         let input_eq_properties = input.equivalence_properties();
-        let eq_properties = match reused_eq_group {
-            Some(eq_group) => input_eq_properties.project_with_eq_group(
+        let eq_properties = match reuse_from {
+            Some((previous, cached)) => input_eq_properties.project_reusing(
                 projection_mapping,
                 schema,
-                eq_group,
+                previous,
+                cached,
             ),
             None => input_eq_properties.project(projection_mapping, schema),
         };
@@ -421,21 +399,17 @@ impl ExecutionPlan for ProjectionExec {
                 // expressions are equal to one another. Projecting that group
                 // again would reproduce the group already cached here, so reuse
                 // it and derive only the orderings.
-                let child = children.swap_remove(0);
-                let reused_eq_group = self
-                    .input
-                    .equivalence_properties()
-                    .eq_group()
-                    .has_same_classes(child.equivalence_properties().eq_group())
-                    .then(|| {
-                        #[cfg(test)]
-                        eq_group_reuse_probe::record_hit();
-                        self.cache.equivalence_properties().eq_group().clone()
-                    });
+                // Hand over what this projection was built from and what that
+                // produced; `project_reusing` decides whether the group can be
+                // carried over and falls back to a full projection otherwise.
+                let reuse_from = Some((
+                    self.input.equivalence_properties(),
+                    self.cache.equivalence_properties(),
+                ));
                 ProjectionExec::try_from_projector_with_eq_group(
                     self.projector.clone(),
-                    child,
-                    reused_eq_group,
+                    children.swap_remove(0),
+                    reuse_from,
                 )
                 .map(|p| Arc::new(p) as _)
             }
@@ -2432,18 +2406,10 @@ mod tests {
         let sorted: Arc<dyn ExecutionPlan> =
             Arc::new(SortExec::new(ordering, Arc::clone(&child)));
 
-        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
-            Arc::clone(&projection).replace_children(
-                vec![Arc::clone(&sorted)],
-                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-            )
-        });
-        let replaced = replaced?;
-        assert_eq!(
-            reuses, 1,
-            "expected the fast path to be taken exactly once; deleting it would \
-             otherwise leave this test passing"
-        );
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&sorted)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
 
         // Guard against vacuity: the group must be worth reusing, and the sort
         // must genuinely have added an ordering the original did not have.
@@ -2520,14 +2486,10 @@ mod tests {
             "nullability moved the equivalence group, so the fast path is no longer under test"
         );
 
-        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
-            Arc::clone(&projection).replace_children(
-                vec![Arc::clone(&tightened_child)],
-                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-            )
-        });
-        let replaced = replaced?;
-        assert_eq!(reuses, 1, "expected the fast path to be taken");
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&tightened_child)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
         let recomputed = ProjectionExec::try_from_projector(
             projection.projector.clone(),
             tightened_child,
@@ -2548,17 +2510,10 @@ mod tests {
         // This child equates a different pair, so the cached group is stale and
         // reusing it would be unsound: the guard has to fall through.
         let other = filtered_source("a", "c")?;
-        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
-            Arc::clone(&projection).replace_children(
-                vec![Arc::clone(&other)],
-                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
-            )
-        });
-        let replaced = replaced?;
-        assert_eq!(
-            reuses, 0,
-            "the guard let a stale equivalence group through the fast path"
-        );
+        let replaced = Arc::clone(&projection).replace_children(
+            vec![Arc::clone(&other)],
+            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+        )?;
 
         assert!(
             !replaced

From 73238635b29bb8086ddddd694b5bbd3c23f63b22 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Fri, 21 Aug 2026 11:19:49 +0800
Subject: [PATCH 10/10] docs: drop a link to a type no longer in scope

Swapping the `EquivalenceGroup` import for `EquivalenceProperties` left an
intra-doc link with nothing to resolve against, which `ci/scripts/rust_docs.sh`
rejects under `-D warnings`. The sentence does not need the link.
---
 datafusion/physical-plan/src/projection.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index b5b5159842796..1ec278eb9c377 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -187,7 +187,7 @@ impl ProjectionExec {
     /// letting [`EquivalenceProperties::project_reusing`] skip reprojecting a
     /// group that has not changed.
     ///
-    /// [`EquivalenceGroup::project`] is a pure function of the group and the
+    /// Projecting an equivalence group is a pure function of the group and the
     /// mapping, so reuse is sound exactly when both are unchanged.
     ///
     /// The caller establishes the first by comparing the old and new child