Add tests for Demonstrate EnforceSorting can remove a needed coalesce

[wiedld]

Which issue does this PR close?

• Demonstrates bug found for ParallelizeSorts, a subrule of EnforceSorting optimizer, should not remove necessary coalesce. #14691

Rationale for this change

Demonstrate the bug.
Afterwards we can continue the process of figuring out the proper fix.

What changes are included in this PR?

Adds a single test case, and associated methods.

Are these changes tested?

Yes

Are there any user-facing changes?

No

[alamb]

Thank you @wiedld

[alamb]

FYI @berkaysynnada

[berkaysynnada]

Did you intentionally put SinglePartitioned aggregate mode? The input is only 1 partition, so why don't you use AggregateMode::Single? I'm unsure if this is a valid case @wiedld @alamb

[berkaysynnada]

When I update the mode as Single, there seems no problem to me

[wiedld]

Thank you for taking the time @berkaysynnada .

The test case merged here is a minimal reproducer. The mode=SinglePartitioned is coming from the actual plans we have running and failing in prod: influxdata#58 (comment)

A simplified view of it is:

"SortExec: expr=[a@0 ASC], preserve_partitioning=[false]",
"  AggregateExec: mode=SinglePartitioned, gby=[a@0 as a1], aggr=[]",
"    CoalescePartitionsExec",
"      ProjectionExec: expr=[a@0 as a, b@1 as value]",
"        UnionExec",
"          DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet",
"          DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet"

The coalesce gets removed for the mode=SinglePartitioned. Do you think that the bug is elsewhere?

[berkaysynnada]

    /// Applies the entire logical aggregation operation in a single operator,
    /// as opposed to Partial / Final modes which apply the logical aggregation using
    /// two operators.
    ///
    /// This mode requires that the input is partitioned by group key (like
    /// FinalPartitioned)
    SinglePartitioned,

It means that when an AggregateExec is SinglePartitioned, then its input should be multi-partitioned (I should say the wording is a bit confusing -- SinglePartitioned means Single layer of aggregation, but its input is Partitioned). However, in your example plan, AggregateExec has a single partition (as it is above CoalescePartitions). So, what I was trying to say that your initial plan should cannot ever exists in general conditions. I'd also like to remind that; EnforceSorting can handle invalid plans and make them valid in terms of ordering conditions (and expects valid distribution conditions), and EnforceDistribution does the same for distribution conditions (and again expects valid ordering conditions). However, in this reproducer, you are giving an invalid plan to the EnforceSorting in terms of distribution.

[alamb]

SinglePartitioned means Single layer of aggregation, but its input is Partitioned). However, in your example plan, AggregateExec has a single partition (as it is above CoalescePartitions).

I see -- so you are saying that if the input to the AggregateExec has only a single partitition, then it should never be SinglePartitioned.

So you are saying that the input plan is not valid and thus the error is elsewhere (whatever generated this plan).

I will make a PR to clarify the documentation as I agree it is quite confusing

[alamb]

Proposed documentation PR:

• Minor: improve documentation of AggregateMode #14946

[wiedld]

However, in this reproducer, you are giving an invalid plan to the EnforceSorting in terms of distribution.

However, in your example plan, AggregateExec has a single partition (as it is above CoalescePartitions)

I think this means that the insertion of the coalesce in the first place (during the enforce distribution) is the bug. Since the input into the enforce distribution optimizer is:

     ...nodes...
        AggregateExec: mode=FinalPartitioned, gby=[time@0 as time], aggr=[sum(Value)]
          AggregateExec: mode=Partial, gby=[date_bin_wallclock(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 10000000000 }, time@0, 0) as time], aggr=[sum(Value)]
            SortExec: expr=[time@0 ASC NULLS LAST], preserve_partitioning=[false]
              ProjectionExec: expr=[time@1 as time, f@0 as Value]
                UnionExec. **multi-partitions**
                  ...multiple nodes...

Our Union outputs multiple partitions, and the AggregateExec;mode=Partial can take multiple partitions. Therefore the enforce distribution should not have inserted the coalesce?

I'll start by updating the reproducer test cases. Thank you.

[wiedld]

In a previous reproducer, I showed how the enforce distribution was inserting the coalesce.
See here: influxdata#58 (comment)

On the latest main, we no longer have the coalesce being inserted.
See here: #14949 (comment)

[wiedld]

I'm going to revert this PR, with this test case.

[berkaysynnada]

Our Union outputs multiple partitions, and the AggregateExec;mode=Partial can take multiple partitions. Therefore the enforce distribution should not have inserted the coalesce?

It might insert a coalesce, but then it has to update AggregateExec: mode=FinalPartitioned to AggregateExec: mode=Final

[alamb]

Reverted in

• Remove invalid bug reproducer. #14950