Resolve contradictory requirements by conversion of ordering sensitive aggregators

[mustafasrepo]

Which issue does this PR close?

Closes #.

Rationale for this change

Some of the seemingly conflicting ordering requirements can be resolved by reversing the aggregation function if possible.
For instance, consider query below

SELECT country, FIRST_VALUE(amount ORDER BY amount ASC) AS fv1,
  LAST_VALUE(amount ORDER BY amount DESC) AS fv2,
  ARRAY_AGG(amount ORDER BY amount DESC) AS amounts
  FROM sales_global
  GROUP BY country

requirements amount ASC by FIRST_VALUE, and amount DESC by LAST_VALUE are conflicting. However, FIRST_VALUE(amount ORDER BY amount ASC) AS fv1 is equivalent to LAST_VALUE(amount ORDER BY amount DESC) AS fv1, in terms of behavior. Hence by converting query above to its equivalent form below

SELECT country, LAST_VALUE(amount ORDER BY amount DESC) AS fv1,
  LAST_VALUE(amount ORDER BY amount DESC) AS fv2,
  ARRAY_AGG(amount ORDER BY amount DESC) AS amounts
  FROM sales_global
  GROUP BY country

we can resolve conflicting requirement, and successfully run equivalent version.

As another use case assume that, sales_global table is ordered by ts ASC. Query below,

SELECT country, LAST_VALUE(amount ORDER BY ts DESC) AS lv1
  FROM sales_global
  GROUP BY country

It produces following plan

ProjectionExec: expr=[country@0 as country, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts DESC NULLS FIRST]@1 as lv1]
  AggregateExec: mode=Single, gby=[country@0 as country], aggr=[LAST_VALUE(sales_global.amount)]
    SortExec: expr=[ts@1 DESC]
      MemoryExec: partitions=1, partition_sizes=[1], output_ordering=[ts ASC]

However, if we were to treat above query in its equivalent for like below

SELECT country, FIRST_VALUE(amount ORDER BY ts ASC) AS lv1
  FROM sales_global
  GROUP BY country

we can produce following plan

ProjectionExec: expr=[country@0 as country, FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as lv1]
  AggregateExec: mode=Single, gby=[country@0 as country], aggr=[LAST_VALUE(sales_global.amount)]
    MemoryExec: partitions=1, partition_sizes=[1], output_ordering=[ts ASC]

What changes are included in this PR?

This PR analyzes reverse ordering requirements, both to resolve conflicting ordering requirements and also remove away SortExecs from the physical plan, if it is possible.

Are these changes tested?

Yes new tests are added to groupby.slt file.

Are there any user-facing changes?

[mustafasrepo]

Since ordering requirement column a is inside the group by expressions. Each group consists of single value for column a. Hence previous result, and this result are both correct. However, with the changes in this PR, result changes.

[mustafasrepo]

Similar things with above test apply to this test case also.

[mustafasrepo]

reverse_order_bys is moved under sort_expr.rs to be able to use it in src/physical_plan/aggregates/mod.rs. With this change scope of the function is increased.

[alamb]

I plan to review this PR carefully tomorrow

[alamb]

in terms of behavior. Hence by converting query above to its equivalent form below

SELECT country, LAST_VALUE(amount ORDER BY amount DESC) AS fv1,
  LAST_VALUE(amount ORDER BY amount DESC) AS fv2,
  ARRAY_AGG(amount ORDER BY amount ASC) AS amounts
  FROM sales_global
  GROUP BY country

Did you mean like this (change LAST_VALUE to FIRST_VALUE and switch the order):

SELECT country, LAST_VALUE(amount ORDER BY amount DESC) AS fv1,
  FIRST_VALUE(amount ORDER BY amount ASC) AS fv2,
  ARRAY_AGG(amount ORDER BY amount ASC) AS amounts
  FROM sales_global
  GROUP BY country

[alamb]

Thanks @mustafasrepo -- I have a question about the plans but I think it is just my own misunderstanding

[alamb]

I don't understand this plan -- if the sort is by amount1 DESC then isn't

I am expecting that these are the same:

  FIRST_VALUE(amount ORDER BY amount ASC) AS fv1,
  LAST_VALUE(amount ORDER BY amount DESC) AS fv2

So when the orderby is rewritten to be ORDER BY amount1 DESC I expect to see that

  LAST_VALUE(amount ORDER BY amount DESC) AS fv1, -- this got switched
  LAST_VALUE(amount ORDER BY amount DESC) AS fv2

However, the query seems to get the right answer, so I must be missing something

[mustafasrepo]

AggregateExpr has a fn name(&self) -> &str method. This method returns string representation of the expression. This is given during initialization from outside(It is not calculated according to function and arguments). During reversal we use the name as is. However, it is misleading, as you observed (even though behind the scene appropriate version is working). However, I couldn't find an easy fix for this problem

[alamb]

🤔 I wonder if during rewrite we could rewrite the value of name() somehow

[mustafasrepo]

We can replace FIRST_VALUE with LAST_VALUE vice versa inside the string. This is not a nice solution, but this would solve our problem, what do you think about it?

[mustafasrepo]

I think, we can rewrite name similar to the version that is given during initialization. This way we may not overfit on string convention. I will experiment with rewriting the value of name(), and let you know about my observations.

[mustafasrepo]

@alamb I have changed the name implementation. It is now calculated inside the aggregator, it is not given from outside during initialization. The display of the AggregateExec changed from
--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[ARRAY_AGG(sales_global.amount), FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)] to
--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[ARRAY_AGG(sales_global.amount), LAST_VALUE(amount@1), LAST_VALUE(amount@1)]

Since argument expressions are Arc<dyn PhysicalExpr>, output display of the argument is changed from sales_global.amount to amount@1 for FIRST_VALUE AND LAST_VALUE.

This change, fortunately lead me to recognize a bug during order-sensitive aggregation when partition number is greater than 1.
In multi partition cases, it is not guaranteed that Final or FinalPartitioned stages in the AggregateExec to receive data in-order.
Hence I constrained order-sensitive aggregation to single partition cases.
I will add support for order-sensitive aggregation at multiple partitions, in following PRs.

[ozankabak]

So we have two options:

1. Simple string replacement of FIRST_VALUE to LAST_VALUE (and vice versa) and order direction when reversal happens.
2. Generating name with the new mechanism @mustafasrepo experimented with.

The advantage of (1) is that it preserves a uniform display format. Its disadvantage is that this is a specialized solution to FIRST_VALUE/LAST_VALUE and may not generalize to other functions in the future.

The advantage of (2) is that is generalizes, but it has the drawback that we end up with an intermixed format; i.e. we have entries like ARRAY_AGG(sales_global.amount) along with entries like LAST_VALUE(amount@1).

Let's choose one for now and create an issue to fix this in a more general fashion in the future. I do not have a preference at this point -- @alamb, which one do you prefer?

[alamb]

I don't have any preference -- they both sound fine to me

[ozankabak]

This LGTM, feel free to merge