[VL] Flushable distinct agg caused correctness issue

[Yohahaha]

Backend

VL (Velox)

Bug description

TPC-DS q38 has distinct agg, when partial distinct is flushed, final result is incorrect.

flushed	unflushed

expect: [26056]
actual: [26131]

Spark version

None

Spark configurations

No response

System information

No response

Relevant logs

No response

[rui-mo]

It seems the aggregation before partial_count should not be flushable. Can we fully rely on whether the data being directly sent to exchange to decide whether flushable aggregation can be generated? cc @zhztheplayer

[Yohahaha]

When agg function is empty, we may not use flushable agg which may produce duplicate rows.

[zhztheplayer]

Thank you for reporting @Yohahaha . I think this is very likely similar to the possible corner case I mentioned in #4312 (comment)

if partial aggregation is already handing data that is partitioned by the exchange (aggregate) keys

We'd enhance FlushableAggregateRule to handle this case.

Probably our CI tests with result comparisons are not handling enough data to trigger flush. I'll take a look into that as well.

[zhztheplayer]

When agg function is empty, we may not use flushable agg which may produce duplicate rows.

Flushing on distinct aggregation can be useful, e.g., do flush on scan-side then do final distinct on reducers. The problem here is Spark has specialized logic on count(distinct), however if we think it generally it's about the aggregate key distribution. If there is an exchange on the exit side of aggregation, and the exchange would be enforced to redistribute the data populated in the same aggregate group but on different mappers, to the same reducer, the partial aggregation can be considered safe to flush.

[zhztheplayer]

Update: The issue should impact Spark version >= 3.3