Combine Equivalence and Ordering equivalence to simplify state

[mustafasrepo]

Which issue does this PR close?

Closes #.

Rationale for this change

In the existing code base we have two different methods to keep track of equivalences in the Arc<dyn PhysicalPlan>.
Which are equivalence_properties and ordering_equivalence_properties.

As a background EquivalenceProperties keeps track of equivalent columns such as a=b=c, e=f.
OrderingEquivalenceProperties alternative orderings that table satisfies. such [a ASC, b ASC], and [c ASC, d ASC].
OrderingEquivalenceProperties keeps track of constant expressions also (e.g expression that are known to have a constant value. This can arise after filter, join, etc.).

Inherently, this information is coupled, as an example
Assume that

• existing table satisfies following orderings [a ASC, b ASC] and [c ASC, d ASC].
• table have equivalent columns a=e.
• It is known that f is constant.

If an operator requires ordering at its input [e ASC, f ASC, b ASC]. During the analysis for whether this requirement is satisfied by existing ordering, we need to consider all orderings, equivalences, and constants at the same time.

Following naive algorithm can be followed for this analysis (Please note that algorithm in this PR is more elaborate.)

• Remove constant expressions from the requirement (This converts requirement [e ASC, f ASC, b ASC] to [e ASC, b ASC])
• Rewrite requirement such that it uses only representative expression for each distinct equivalent group (This converts requirement [e ASC, b ASC] to [a ASC, b ASC]).
• Apply same procedures above each of the orderings inside the OrderingEquivalences (This converts ordering [a ASC, b ASC] to [a ASC, b ASC] and [c ASC, d ASC] to [c ASC, d ASC] no change ).
• Check whether normalized requirement [a ASC, b ASC] is satisfied by any of normalized orderings [a ASC, b ASC], [c ASC, d ASC].

As can be seen from the example above. Keeping track of these information separately, is a bit cumbersome.

Also for the user implementing new functionality is a bit hard, and existing APIs are a bit involved also. Such as ordering_satisfy, requirements_compatible, etc.

I think it is better to unify these information in a single struct so that

• We can expose better, and more friendly APIs from struct.
• Move utils, functions, to method calls
• Keep the invariants in the state (not relying on correct arguments).
• All of the implementations, algorithms resides in a single place, and logic is not scatterred in different files.

What changes are included in this PR?

This PR unifies EquivalenceProperties and OrderingEquivalenceProperties to single struct called EquivalenceProperties (equivalence now involves, exact equivalence, ordering equivalence, constants).
And all of the implementation that depend on this information is moved to method calls (such as projection, ordering_satisfy, etc.)

• Additional tests to show that better plans can be produced with new design. (As an example ordering_satisfy no longer just depends on single ordering, which is output ordering. Bu considers all of the valid orderings for the table. This enables additional optimizations. See new tests under window.slt as an example)

Are these changes tested?

Yes new tests are added.

Are there any user-facing changes?

api change ordering_equivalence_properties is removed from the ExecutionPlan and now equivalence_properties contains additional information.

[alamb]

Hi @alamb. Did you get any problems testing downstream with IOx? We have some follow-ons to this in the pipeline once this merges

I have two test failures downstream that I am looking into. I don't know if they are problems with this PR or something pre-existing that it exposed

Getting this PR reviewed / tested in my highest priority today. I will have an update in a few hours

[alamb]

One of the test failures internally looks like the following

The input looks like:

2023-11-02T15:58:06.601675Z TRACE log: Optimized physical plan by CombinePartialFinalAggregate:
OutputRequirementExec
  SortExec: expr=[time@1 ASC NULLS LAST]
    CoalescePartitionsExec
      ProjectionExec: expr=[cpu as iox::measurement, time@0 as time, (selector_last(sum_idle,time)@1).[value] as last, (selector_last(sum_system,time)@2).[value] as last_1]
        AggregateExec: mode=FinalPartitioned, gby=[time@0 as time], aggr=[selector_last(sum_idle,time), selector_last(sum_system,time)], ordering_mode=Sorted
          SortPreservingRepartitionExec: partitioning=Hash([time@0], 16), input_partitions=16, sort_exprs=time@0 ASC NULLS LAST
            AggregateExec: mode=Partial, gby=[date_bin(10000000000, time@0, 0) as time], aggr=[selector_last(sum_idle,time), selector_last(sum_system,time)]
              RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
                SortExec: expr=[time@0 ASC NULLS LAST]
                  CoalescePartitionsExec
                    ProjectionExec: expr=[time@0 as time, SUM(cpu.usage_idle)@1 as sum_idle, SUM(cpu.usage_system)@2 as sum_system]
                      AggregateExec: mode=FinalPartitioned, gby=[time@0 as time], aggr=[SUM(cpu.usage_idle), SUM(cpu.usage_system)]
                        RepartitionExec: partitioning=Hash([time@0], 16), input_partitions=16
                          AggregateExec: mode=Partial, gby=[date_bin(10000000000, time@0, 0) as time], aggr=[SUM(cpu.usage_idle), SUM(cpu.usage_system)]
                            RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
                              ProjectionExec: expr=[time@1 as time, usage_idle@2 as usage_idle, usage_system@3 as usage_system]
                                FilterExec: date_bin(10000000000, time@1, 0) <= 1698940686290451000 AND time@1 <= 1698940686290451000 AND cpu@0 = cpu-total
                                  ParquetExec: file_groups={1 group: [[2/8/0649f0e8b1abed092a356ec6181369fcf585431d1cc0694a0cc4ab45cf78b49d/0c5ac9b2-f6d4-4004-9036-15412da47647.parquet]]}, projection=[cpu, time, usage_idle, usage_system], predicate=date_bin(10000000000, time@2, 0) <= 1698940686290451000 AND time@2 <= 1698940686290451000 AND cpu@0 = cpu-total, pruning_predicate=time_min@0 <= 1698940686290451000 AND cpu_min@1 <= cpu-total AND cpu-total <= cpu_max@2

But then after EnforceSorting the SortPreservingMergeExec seems to have to sort exprs anymore:

2023-11-02T15:58:06.605925Z TRACE log: Optimized physical plan by EnforceSorting:
OutputRequirementExec
  SortPreservingMergeExec: [time@1 ASC NULLS LAST] 
    SortExec: expr=[time@1 ASC NULLS LAST]
      ProjectionExec: expr=[cpu as iox::measurement, time@0 as time, (selector_last(sum_idle,time)@1).[value] as last, (selector_last(sum_system,time)@2).[value] as last_1]
        AggregateExec: mode=FinalPartitioned, gby=[time@0 as time], aggr=[selector_last(sum_idle,time), selector_last(sum_system,time)]
    ----> SortPreservingRepartitionExec: partitioning=Hash([time@0], 16), input_partitions=16 
            AggregateExec: mode=Partial, gby=[date_bin(10000000000, time@0, 0) as time], aggr=[selector_last(sum_idle,time), selector_last(sum_system,time)]
              RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=16
                ProjectionExec: expr=[time@0 as time, SUM(cpu.usage_idle)@1 as sum_idle, SUM(cpu.usage_system)@2 as sum_system]
                  AggregateExec: mode=FinalPartitioned, gby=[time@0 as time], aggr=[SUM(cpu.usage_idle), SUM(cpu.usage_system)]
                    RepartitionExec: partitioning=Hash([time@0], 16), input_partitions=16
                      AggregateExec: mode=Partial, gby=[date_bin(10000000000, time@0, 0) as time], aggr=[SUM(cpu.usage_idle), SUM(cpu.usage_system)]
                        RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
                          ProjectionExec: expr=[time@1 as time, usage_idle@2 as usage_idle, usage_system@3 as usage_system]
                            FilterExec: date_bin(10000000000, time@1, 0) <= 1698940686290451000 AND time@1 <= 1698940686290451000 AND cpu@0 = cpu-total
                              ParquetExec: file_groups={1 group: [[2/8/0649f0e8b1abed092a356ec6181369fcf585431d1cc0694a0cc4ab45cf78b49d/0c5ac9b2-f6d4-4004-9036-15412da47647.parquet]]}, projection=[cpu, time, usage_idle, usage_system], predicate=date_bin(10000000000, time@2, 0) <= 1698940686290451000 AND time@2 <= 1698940686290451000 AND cpu@0 = cpu-total, pruning_predicate=time_min@0 <= 1698940686290451000 AND cpu_min@1 <= cpu-total AND cpu-total <= cpu_max@2

This then causes a failure during execution / streaming merge

Internal error: Sort expressions cannot be empty for streaming merge

[alamb]

Thank you again @mustafasrepo and @ozankabak -- I can't realistically review this entire PR in detail, but I did skim the whole thing and reviewed the test changes, comments, and parts about grouping.

I think there may be a bug with this implementation as I mentioned in my comment, However, given the size of this PR I think we should merge it to master and then debug / fix whatever issues we find as follow on PRs.

Thank you again -- I think this is a major improvement

[alamb]

FWIW in this kind of change (moving functions to other modules and renaming them) is something that we could probably do as individual PRs that would be quick to review as they would be mostly mechanical.

That would help make it easier to find the parts of a PR such as this one that needed more careful review

[ozankabak]

Agreed, we sent most of the drive by cleanups/refactors we did as separate small PRs, but this must have slipped through the cracks.

[alamb]

This change effectively removes the duplication between GroupByOrderMode and PartitionSearchMode that represent the same thing right?

I find PartitionSearchMode a confusing name as the term "partition" is pretty overloaded already (like each ExecutionPlan has input/output partitions, and WindowExec deals with partitions).

Also the fact it is in the windows module seems to be a mismatch given it is now used in the Aggregation logic

Maybe it could go into its own module 🤔 and be called SortOrderMode or something

Also, can we clarify what PartitionSearchMode::PartiallySorted means -- specifically, does it represent a prefix of the columns that is sorted?

This could all be done as follow on PRs

[alamb]

To make this easier to discover, perhaps it could be a method in ProjectionMapping such as ProjectionMapping::try_new

[ozankabak]

For now ProjectionMapping is just a type alias, but we plan to promote it to a full-fledged type in a follow on PR -- I agree that it makes sense to convert this into a method then.

[alamb]

FWIW I hacked up a version here and it worked great: #8033

[alamb]

This is another pattern that has appeared more than once (also in GroupBy) -- maybe it could be ProjectionMapping (which could also be made into a struct w/ a constructor)

[alamb]

BTW I made a PR to do this here -- #8033 and it worked out well. Once this PR is merged I'll rebase it and get it ready for review

[alamb]

Here is a similar one for EquivalenceClass -- #8034 that might be more of a conflict magnet so leaving it for a while

[ozankabak]

One of the test failures internally looks like the following

...

We will investigate tomorrow and include the fix in this PR if it is a quick fix. Thanks for battle-testing :)

[alamb]

FWIW I am running our internal tests against the code on this branch and they now pass. Thank you @mustafasrepo

Update: I misran the test -- I am now running again

Update: the internal test still fails

[alamb]

BTW I still think give the size of this PR we should merge it now and then fix the issue next week. Follow on ticket to serve as a placeholder

#8043

[mustafasrepo]

BTW I still think give the size of this PR we should merge it now and then fix the issue next week. Follow on ticket to serve as a placeholder

#8043

That would be great. I actually understood the reason for the bug, and working on it. However, I also think that it is better to merge this PR without waiting.

[alamb]

Let's merge this to keep the code flowing

[mustafasrepo]

Partially resolves issue #8064

[alamb]

While working on the upgrade in IOx, I also filed #8120 to maybe help avoid issues for others upgrading