From 4c335bf45c14a899a09b21417ae31ee871ddf711 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 11:01:02 +0800
Subject: [PATCH 1/8] [X-3555] perf: reuse the projected equivalence group when
 only child orderings change

Port of apache/datafusion#24445, adapted to this branch's `with_new_children`
API (upstream has since split that into `replace_children` with an explicit
`ChildrenPropertiesMode`; the guard sits in the same spot either way).

`check_if_same_properties!` requires the child's properties to be unchanged
outright. A rule that inserts a sort below a projection does not qualify,
since the child is a new object with new properties. Its equivalence group,
however, is unchanged: sorting changes which orderings hold, not which
expressions are equal to one another. The projection therefore re-projects a
group identical to the one it already holds.

`EquivalenceGroup::project` is a pure function of the group and the mapping,
so when both are unchanged the previous result can be reused and only the
orderings derived. The cost is paid once per projection above the insertion
point on every pass of the rule, so it grows with projection count times
expression size times passes.

Measured on the stocks last-trade snapshot endpoint, ten warm samples per
configuration:

    EnforceSorting    197.8ms -> 90.1ms   -54%
    optimizer rules   338.1ms -> 214.6ms  -37%
    planning wall     420.6ms -> 299.0ms  -29%

Logical rules are unchanged across the two configurations, which acts as the
control: the saving lands in the physical phase and nowhere else.
---
 .../physical-expr/src/equivalence/class.rs    |  8 +++
 .../src/equivalence/properties/mod.rs         | 17 ++++++
 datafusion/physical-plan/src/projection.rs    | 58 ++++++++++++++++++-
 3 files changed, 82 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index d00a4a32278f0..ef2a84c148d7e 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 bb74cd1d9c7b3..0c19188e58f30 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1143,6 +1143,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 d49522dfd2989..f776b049b208c 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -48,7 +48,7 @@ use datafusion_common::tree_node::{
 use datafusion_common::{DataFusionError, JoinSide, Result, internal_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::{
@@ -162,6 +162,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()
@@ -317,6 +356,23 @@ impl ExecutionPlan for ProjectionExec {
         mut children: Vec<Arc<dyn ExecutionPlan>>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         check_if_same_properties!(self, children);
+        // The macro 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),

From 1e204a7418ea34b16ef3131a31bc9202b019d6ab Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 11:30:08 +0800
Subject: [PATCH 2/8] [X-3555] test: cover the equivalence-group reuse on child
 replacement

Ports the upstream tests from apache/datafusion#24445, adapted to this
branch's `with_new_children` API.

`EquivalenceGroup`'s new `PartialEq`: equal classes compare equal however
they were arrived at, 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. A second test
passes an empty group to confirm the argument is consumed rather than
ignored.

`ProjectionExec::with_new_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; inverting the guard to always reuse makes
that test fail. The `check_if_same_properties!` fast path is covered by
handing back the same child, since it compares properties by pointer. A
separate test pins the premise that sorting leaves the equivalence group
untouched, so a change there fails first rather than silently weakening the
reuse.
---
 .../physical-expr/src/equivalence/class.rs    |  66 +++++-
 .../src/equivalence/properties/mod.rs         | 106 ++++++++++
 datafusion/physical-plan/src/projection.rs    | 191 ++++++++++++++++++
 3 files changed, 362 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index ef2a84c148d7e..7006524dea9fc 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -920,7 +920,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;
 
@@ -1241,4 +1241,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 0c19188e58f30..fcaad2a545e2e 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1504,3 +1504,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 f776b049b208c..8a21855717e47 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -1227,6 +1227,9 @@ mod tests {
     use super::*;
 
     use crate::common::collect;
+    use crate::empty::EmptyExec;
+    use crate::filter::FilterExec;
+    use crate::sorts::sort::SortExec;
 
     use crate::filter_pushdown::PushedDown;
     use crate::test;
@@ -1856,4 +1859,192 @@ 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_with_new_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).with_new_children(vec![Arc::clone(&sorted)])?;
+
+        // 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_with_new_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).with_new_children(vec![Arc::clone(&other)])?;
+
+        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_with_new_children_keeps_properties_when_the_child_is_unchanged() -> Result<()>
+    {
+        // `check_if_same_properties!` compares the children's `PlanProperties`
+        // by pointer, so handing back the very same child takes the existing
+        // fast path ahead of the equivalence-group one. The cached properties
+        // must then 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).with_new_children(vec![child])?;
+
+        assert_same_properties(replaced.as_ref(), &projection);
+
+        Ok(())
+    }
 }

From a7074a59f0b580f7b6c570ffd23bdef495937559 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 12:03:13 +0800
Subject: [PATCH 3/8] [X-3555] Address the review: set-equality for groups, and
 two weak tests

`EquivalenceGroup`'s `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 7006524dea9fc..c37c6eff3436b 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))
     }
 }
 
@@ -1256,9 +1269,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 fcaad2a545e2e..b55f385fdbcce 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1512,6 +1512,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
@@ -1601,10 +1602,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 8a21855717e47..3abf7ba5f608c 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -1910,9 +1910,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 fe5d0be5cd8fe9f434cf01676e6d3789c4bedbe5 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 14:40:33 +0800
Subject: [PATCH 4/8] [X-3555] 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 | 78 ++++++++++++++++++++--
 1 file changed, 72 insertions(+), 6 deletions(-)

diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 3abf7ba5f608c..799672990d5af 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -165,9 +165,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>,
@@ -1863,10 +1873,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(
@@ -2015,6 +2035,52 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_with_new_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: `with_new_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 `with_new_children` and not something
+        // this fast path introduces, so the meaningful contract is that its two
+        // 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)
+            .with_new_children(vec![Arc::clone(&nullable_child)])?;
+        let recomputed = ProjectionExec::try_from_projector(
+            projection.projector.clone(),
+            nullable_child,
+        )?;
+
+        assert_same_properties(replaced.as_ref(), &recomputed);
+
+        Ok(())
+    }
+
     #[test]
     fn test_with_new_children_recomputes_when_eq_group_changes() -> Result<()> {
         let child = filtered_source("a", "b")?;

From f39f79e3cdce947e66f7790a06261855dd8564b3 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 15:12:01 +0800
Subject: [PATCH 5/8] [X-3555] 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
`with_new_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 799672990d5af..87c633bb192fa 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -2043,6 +2043,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: `with_new_children`
         // carries the existing `Projector` over, so the output schema stays as
@@ -2050,19 +2053,19 @@ mod tests {
         // That difference is inherent to `with_new_children` and not something
         // this fast path introduces, so the meaningful contract is that its two
         // 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(),
@@ -2070,10 +2073,10 @@ mod tests {
         );
 
         let replaced = Arc::clone(&projection)
-            .with_new_children(vec![Arc::clone(&nullable_child)])?;
+            .with_new_children(vec![Arc::clone(&tightened_child)])?;
         let recomputed = ProjectionExec::try_from_projector(
             projection.projector.clone(),
-            nullable_child,
+            tightened_child,
         )?;
 
         assert_same_properties(replaced.as_ref(), &recomputed);

From 3cd2132006bd1f77f6efbdf21821634175805e65 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Wed, 19 Aug 2026 15:40:21 +0800
Subject: [PATCH 6/8] [X-3555] 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 c37c6eff3436b..44a1749d5cac6 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()
@@ -1274,55 +1280,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`.
@@ -1331,7 +1344,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 b55f385fdbcce..09610f871f37e 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1573,7 +1573,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 87c633bb192fa..3dd2cc9264dc8 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -372,8 +372,11 @@ impl ExecutionPlan for ProjectionExec {
         // 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()
+        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(
@@ -1914,10 +1917,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(),
@@ -1967,9 +1973,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!(
@@ -2063,12 +2070,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"
         );
 
@@ -2097,9 +2109,14 @@ mod tests {
         let replaced =
             Arc::clone(&projection).with_new_children(vec![Arc::clone(&other)])?;
 
-        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 6a3e400b7992c9d373ef472a2581b03970460098 Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Thu, 20 Aug 2026 12:12:39 +0800
Subject: [PATCH 7/8] [X-3555] 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    | 55 ++++++++++--
 3 files changed, 95 insertions(+), 49 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 44a1749d5cac6..2e6253393b7df 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 09610f871f37e..bb1505b9a268a 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -1143,7 +1143,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
@@ -1154,11 +1157,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());
@@ -1512,7 +1542,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
@@ -1585,48 +1614,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 3dd2cc9264dc8..2d846fae8c2ae 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -63,6 +63,31 @@ pub use datafusion_physical_expr::projection::{
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
+/// Counts how often [`ProjectionExec::with_new_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
@@ -378,6 +403,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(),
@@ -2005,8 +2032,15 @@ mod tests {
         let sorted: Arc<dyn ExecutionPlan> =
             Arc::new(SortExec::new(ordering, Arc::clone(&child)));
 
-        let replaced =
-            Arc::clone(&projection).with_new_children(vec![Arc::clone(&sorted)])?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).with_new_children(vec![Arc::clone(&sorted)])
+        });
+        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.
@@ -2084,8 +2118,11 @@ mod tests {
             "nullability moved the equivalence group, so the fast path is no longer under test"
         );
 
-        let replaced = Arc::clone(&projection)
-            .with_new_children(vec![Arc::clone(&tightened_child)])?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).with_new_children(vec![Arc::clone(&tightened_child)])
+        });
+        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,
@@ -2106,8 +2143,14 @@ 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).with_new_children(vec![Arc::clone(&other)])?;
+        let (replaced, reuses) = eq_group_reuse_probe::count(|| {
+            Arc::clone(&projection).with_new_children(vec![Arc::clone(&other)])
+        });
+        let replaced = replaced?;
+        assert_eq!(
+            reuses, 0,
+            "the guard let a stale equivalence group through the fast path"
+        );
 
         assert!(
             !replaced

From 11b379dd2598be3b7f87c89703bcdc3badb8e4bc Mon Sep 17 00:00:00 2001
From: Qi Zhu <zhuqilucas@gmail.com>
Date: Fri, 21 Aug 2026 12:06:59 +0800
Subject: [PATCH 8/8] [X-3555] review: make the reuse decision inside
 EquivalenceProperties

Ports three upstream review rounds from apache/datafusion#24445 in one go,
since the second reworked what the first had just changed.

`compute_properties` takes the reuse as a parameter and there is one
constructor instead of two. The second duplicated partitioning, pipeline
behaviour, boundedness and both literals, so a new field on `PlanProperties`
would have had to be added twice.

`project_with_eq_group` took a group on trust and could only assert its
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, so there is no longer a
precondition to violate.

The reuse probe moves to `physical-expr` with the decision: `cfg(test)` does not
reach across crates, so the tests asserting the fast path is taken have to live
where the branch is. `physical-plan` keeps the tests asserting the resulting
properties are right. Disabling the reuse still fails the former.

Also drops an intra-doc link to a type no longer in scope, which
`ci/scripts/rust_docs.sh` rejects under `-D warnings`.
---
 .../src/equivalence/properties/mod.rs         | 138 +++++++++------
 datafusion/physical-plan/src/projection.rs    | 157 +++++-------------
 2 files changed, 133 insertions(+), 162 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index bb1505b9a268a..074bce91c0522 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 a
     /// cast-bearing expression can replace an existing sort key without
@@ -1149,41 +1174,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.
+    /// Projecting an equivalence group is a pure function of the group and the
+    /// 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,
@@ -1577,19 +1597,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.
@@ -1614,22 +1640,40 @@ 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");
+
+        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 2d846fae8c2ae..3ea9c33edc250 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -48,7 +48,8 @@ use datafusion_common::tree_node::{
 use datafusion_common::{DataFusionError, JoinSide, Result, internal_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::{
@@ -63,31 +64,6 @@ pub use datafusion_physical_expr::projection::{
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
-/// Counts how often [`ProjectionExec::with_new_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
@@ -171,63 +147,27 @@ 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.
-    ///
-    /// [`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(
+    /// 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.
+    fn try_from_projector_with_eq_group(
         projector: Projector,
         input: Arc<dyn ExecutionPlan>,
-        eq_group: 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 =
             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(),
-        );
+            reuse_from,
+        )?;
         Ok(Self {
             projector,
             input,
@@ -256,10 +196,21 @@ impl ProjectionExec {
         input: &Arc<dyn ExecutionPlan>,
         projection_mapping: &ProjectionMapping,
         schema: SchemaRef,
+        reuse_from: Option<(&EquivalenceProperties, &EquivalenceProperties)>,
     ) -> Result<PlanProperties> {
-        // Calculate equivalence properties:
+        // Calculate equivalence properties. Whether the group is reprojected or
+        // carried over 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 reuse_from {
+            Some((previous, cached)) => input_eq_properties.project_reusing(
+                projection_mapping,
+                schema,
+                previous,
+                cached,
+            ),
+            None => input_eq_properties.project(projection_mapping, schema),
+        };
         // Calculate output partitioning, which needs to respect aliases:
         let output_partitioning = input
             .output_partitioning()
@@ -397,25 +348,17 @@ impl ExecutionPlan for ProjectionExec {
         // 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()
-            .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(
+        // 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(),
             children.swap_remove(0),
+            reuse_from,
         )
         .map(|p| Arc::new(p) as _)
     }
@@ -2032,15 +1975,8 @@ 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).with_new_children(vec![Arc::clone(&sorted)])
-        });
-        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).with_new_children(vec![Arc::clone(&sorted)])?;
 
         // Guard against vacuity: the group must be worth reusing, and the sort
         // must genuinely have added an ordering the original did not have.
@@ -2118,11 +2054,8 @@ 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).with_new_children(vec![Arc::clone(&tightened_child)])
-        });
-        let replaced = replaced?;
-        assert_eq!(reuses, 1, "expected the fast path to be taken");
+        let replaced = Arc::clone(&projection)
+            .with_new_children(vec![Arc::clone(&tightened_child)])?;
         let recomputed = ProjectionExec::try_from_projector(
             projection.projector.clone(),
             tightened_child,
@@ -2143,14 +2076,8 @@ 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).with_new_children(vec![Arc::clone(&other)])
-        });
-        let replaced = replaced?;
-        assert_eq!(
-            reuses, 0,
-            "the guard let a stale equivalence group through the fast path"
-        );
+        let replaced =
+            Arc::clone(&projection).with_new_children(vec![Arc::clone(&other)])?;
 
         assert!(
             !replaced