feat: TopNRank optimization

[aditi-pandit]

Design doc : https://docs.google.com/document/d/1WQfNigR9bVrbM-PqY7F0mswcetN_tdNahzD9ENye-Q0/edit?usp=sharing

#9404

The refactoring in #11440 reduces the diffs in that part of the code.

e2e Presto PR (with changes in the Presto optimizer as well) prestodb/presto#24138

Latency for SF1K TPC-DS Q67 fell from 399s to 146s with this change.

(I also started working on a fuzzer in #12103 which I will enhance for the rank and dense_rank functions added here).

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	67f644d
🔍 Latest deploy log	https://app.netlify.com/sites/meta-velox/deploys/678068753a11b00008487921

[liujiayi771]

Could you also add a test case for the logic in fixTopRank.

[aditi-pandit]

@liujiayi771 : The fixTopRank logic is tested very thoroughly in the fewPartitions test. Have added a comment there.

[JkSelf]

@aditi-pandit
For TopNRowNumber, there is an issue similar to Window, which is that before TopNRowNumber, Spark will insert an OrderBy operator to sort the data as following.

So, do we need to make some abstractions in addInput here as well, to facilitate the addition of TopNStreamingRowNumber later on?

[aditi-pandit]

@aditi-pandit For TopNRowNumber, there is an issue similar to Window, which is that before TopNRowNumber, Spark will insert an OrderBy operator to sort the data as following.

So, do we need to make some abstractions in addInput here as well, to facilitate the addition of TopNStreamingRowNumber later on?

@JkSelf : TopNRowNumber is a somewhat streaming operator in its current implementation ... It uses HashTable internally to map the input row to a partition and each partition has an accumulator that maintains the ordered rows (as many required for limit) in a priority queue.

Window accumulates all the input rows and does a full sort of the input rows to demarcate into partitions and sort by order-by. So the preceding Sort was useful and we abstracted the streaming window.

With TopNRowNumber, doing a full sort and then making TopNRowNumber limit to only a partition at a time, the tradeoffs are different. Have you considered removing the global sort and checking if TopNRowNumber suffices ?

If we decide eventually that having a full streaming operator for topNRowNumber is useful, then it might be worth it to write a new operator itself (rather than enhance this current one). Offcourse, we can try to reuse some of the ranking logic pieces.

[JkSelf]

@aditi-pandit For TopNRowNumber, there is an issue similar to Window, which is that before TopNRowNumber, Spark will insert an OrderBy operator to sort the data as following.

So, do we need to make some abstractions in addInput here as well, to facilitate the addition of TopNStreamingRowNumber later on?

@JkSelf : TopNRowNumber is a streaming operator... It uses HashTable internally to map the input row to a partition and each partition has an accumulator that maintains the ordered rows in a priority queue.

So the OrderBy is wasteful. It's not required. Can you consider removing the OrderBy before TopNRowNumber ?

@aditi-pandit I see. Thanks for your explanations.