feat: add Grouped equivalence properties

[xavlee]

Which issue does this PR relate to?

• Part of Supporting analytics over large amounts of pre‑partitioned data #24438.
• Builds on test: cover unsorted contiguous groups in one partition #24737 and refactor: separate aggregate group completion from input ordering #24697.

Rationale for this change

Lexicographical ordering is stronger than the guarantee required to recognize completed groups. A group is complete when all rows with the same grouping-key tuple occur in one contiguous run within a partition, even when the distinct tuples occur in an arbitrary order.

AAABBBCCCC -> ordered and grouped
CCCAAABBB  -> grouped but unordered
AAABBBAACC -> not grouped because A occurs in multiple runs

Representing this guarantee in expression equivalence analysis lets DataFusion normalize grouping expressions, derive grouping from existing orderings, project grouping through expressions, and invalidate it at operators that change the row sequence or combine partitions.

What changes are included in this PR?

Semantics

A grouping assertion applies to one complete expression tuple. Grouping on [a, b] means rows with the same pair of values are contiguous. It establishes no grouping assertion for [a] or [b] individually. [a, b] and [b, a] identify the same groups because changing the order of expressions does not change tuple equality.

Lexicographical ordering retains its prefix implication. An ordering on [a, b, c] establishes grouping for [a], [a, b], and [a, b, c], including equivalent permutations of those complete tuples. It does not establish grouping for [b].

Grouping assertions are correctness guarantees supplied by execution-plan producers. DataFusion does not inspect the rows to verify them. A projection retains an explicit tuple only when every expression in that tuple maps to the output. Operators that reorder rows or combine partitions clear explicit assertions whose guarantees no longer hold.

SortProperties::Grouped supports propagation for a single expression. Scalar functions retain it only for strictly order-preserving transformations, because a many-to-one transformation can map separate input runs to the same output value. For example, date_bin does not derive Grouped from a grouped timestamp expression. A producer can instead declare the complete derived tuple it guarantees, such as [bhandle, date_bin(timestamp)].

Public API Changes

• Add SortProperties::Grouped.
• Add the public GroupingEquivalenceClass type with construction, insertion, lookup, clearing, schema-rewriting, iteration, and display support.
• Add EquivalenceProperties::geq_class, normalized_geq_class, add_grouping, add_groupings, clear_groupings, and grouping_satisfy.
• Add FFI_SortProperties::Grouped and its conversions.
• Add TestMemoryExec::try_with_grouping_information for physical-plan tests and examples.

Execution-plan implementations attach grouping assertions through the existing PlanProperties::equivalence_properties. The aggregate runtime consumes this metadata in #24497; its GroupCompletionMode remains an internal implementation detail introduced by #24697.

Implementation

• Store complete grouping tuples alongside ordering, equality, and constant information in EquivalenceProperties.
• Normalize tuples through expression equivalences and constants.
• Derive grouping satisfaction from lexicographical ordering prefixes.
• Project only complete, representable tuples through ProjectionMapping.
• Propagate single-expression grouping through strictly order-preserving transformations.
• Maintain grouping metadata across schema replacement and clear it at row-reordering and partition-merging boundaries.
• Round-trip the new expression property through the Scalar UDF FFI bridge.

Stack

1. #24737 — test: cover unsorted contiguous groups in one partition
2. #24697 — refactor: separate aggregate group completion from input ordering
3. #24698 — feat: add grouped equivalence properties ← this PR
4. #24497 — feat: stream aggregates over grouped input

Are these changes tested?

Tests cover tuple semantics, normalization through equivalences and constants, ordering-derived grouping, projection mapping, strictly order-preserving expression propagation, schema replacement, row-reordering and partition-merging boundaries, display output, and FFI round trips.

Are there any user-facing changes?

Yes.

SortProperties is an exhaustive public enum, so adding Grouped requires downstream exhaustive matches to add an arm. Consumers that do not use grouping information can handle it conservatively with Unordered, for example:

SortProperties::Grouped | SortProperties::Unordered => {
    // No ordering guarantee
}

FFI_SortProperties gains the corresponding variant. Separately compiled FFI consumers should rebuild against the matching DataFusion version and handle Grouped when matching this enum.

GroupingEquivalenceClass and the new EquivalenceProperties methods are additive APIs. The migration guidance for the exhaustive enum additions will also be recorded in the version-specific upgrade guide before merge.

Review this layer

View only this PR layer

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.05028% with 150 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.44%. Comparing base (ebdb657) to head (c2fe167).

Files with missing lines	Patch %	Lines
datafusion/physical-expr/src/equivalence/mod.rs	58.97%	3 Missing and 61 partials ⚠️
...tafusion/physical-expr/src/equivalence/grouping.rs	82.17%	13 Missing and 5 partials ⚠️
datafusion/physical-plan/src/test.rs	60.46%	16 Missing and 1 partial ⚠️
datafusion/physical-plan/src/aggregates/mod.rs	82.19%	5 Missing and 8 partials ⚠️
...on/physical-expr/src/equivalence/properties/mod.rs	90.00%	7 Missing and 5 partials ⚠️
datafusion/physical-plan/src/repartition/mod.rs	78.57%	0 Missing and 6 partials ⚠️
...atafusion/physical-plan/src/coalesce_partitions.rs	84.61%	0 Missing and 4 partials ⚠️
datafusion/physical-expr/src/expressions/cast.rs	81.25%	0 Missing and 3 partials ⚠️
...n/physical-plan/src/sorts/sort_preserving_merge.rs	85.00%	0 Missing and 3 partials ⚠️
datafusion/functions/src/datetime/date_bin.rs	84.61%	0 Missing and 2 partials ⚠️
... and 6 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24698      +/-   ##
==========================================
- Coverage   82.45%   82.44%   -0.01%     
==========================================
  Files        1140     1141       +1     
  Lines      436434   437085     +651     
  Branches   436434   437085     +651     
==========================================
+ Hits       359840   360355     +515     
- Misses      54840    54879      +39     
- Partials    21754    21851      +97     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-expr v55.1.0 (current)
       Built [  33.973s] (current)
     Parsing datafusion-expr v55.1.0 (current)
      Parsed [   0.081s] (current)
    Building datafusion-expr v55.1.0 (baseline)
       Built [  29.628s] (baseline)
     Parsing datafusion-expr v55.1.0 (baseline)
      Parsed [   0.080s] (baseline)
    Checking datafusion-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   1.509s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  67.231s] datafusion-expr
    Building datafusion-expr-common v55.1.0 (current)
       Built [  20.119s] (current)
     Parsing datafusion-expr-common v55.1.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-expr-common v55.1.0 (baseline)
       Built [  20.235s] (baseline)
     Parsing datafusion-expr-common v55.1.0 (baseline)
      Parsed [   0.020s] (baseline)
    Checking datafusion-expr-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.232s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/enum_variant_added.ron

Failed in:
  variant SortProperties:Grouped in /home/runner/work/datafusion/datafusion/datafusion/expr-common/src/sort_properties.rs:42

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  41.338s] datafusion-expr-common
    Building datafusion-ffi v55.1.0 (current)
       Built [  55.686s] (current)
     Parsing datafusion-ffi v55.1.0 (current)
      Parsed [   0.063s] (current)
    Building datafusion-ffi v55.1.0 (baseline)
       Built [  55.719s] (baseline)
     Parsing datafusion-ffi v55.1.0 (baseline)
      Parsed [   0.065s] (baseline)
    Checking datafusion-ffi v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.254s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/enum_variant_added.ron

Failed in:
  variant FFI_SortProperties:Grouped in /home/runner/work/datafusion/datafusion/datafusion/ffi/src/expr/expr_properties.rs:70

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [ 113.226s] datafusion-ffi
    Building datafusion-functions v55.1.0 (current)
       Built [  32.536s] (current)
     Parsing datafusion-functions v55.1.0 (current)
      Parsed [   0.084s] (current)
    Building datafusion-functions v55.1.0 (baseline)
       Built [  32.183s] (baseline)
     Parsing datafusion-functions v55.1.0 (baseline)
      Parsed [   0.087s] (baseline)
    Checking datafusion-functions v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.422s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  66.740s] datafusion-functions
    Building datafusion-physical-expr v55.1.0 (current)
       Built [  29.783s] (current)
     Parsing datafusion-physical-expr v55.1.0 (current)
      Parsed [   0.049s] (current)
    Building datafusion-physical-expr v55.1.0 (baseline)
       Built [  30.008s] (baseline)
     Parsing datafusion-physical-expr v55.1.0 (baseline)
      Parsed [   0.050s] (baseline)
    Checking datafusion-physical-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.404s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  61.173s] datafusion-physical-expr
    Building datafusion-physical-optimizer v55.1.0 (current)
       Built [  42.331s] (current)
     Parsing datafusion-physical-optimizer v55.1.0 (current)
      Parsed [   0.022s] (current)
    Building datafusion-physical-optimizer v55.1.0 (baseline)
       Built [  42.414s] (baseline)
     Parsing datafusion-physical-optimizer v55.1.0 (baseline)
      Parsed [   0.021s] (baseline)
    Checking datafusion-physical-optimizer v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.112s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  86.164s] datafusion-physical-optimizer
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  39.614s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.170s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  39.604s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.174s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.648s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  81.584s] datafusion-physical-plan

[gene-bordegaray]

some nits but overall this is great work, thank you @xavlee