Incorrect result of min_by/max_by(x, y, n) in Window operation

[kagamiori]

Hi community, I noticed min_by/max_by(x, y, n) produces incorrect results when it is used in Window operation. As an example, for the query below with min_by(x, y, n). Since the window frame is UNBOUNDED PRECEDING to CURRENT ROW, when the second row is processed, the function should aggregate both the first and second input rows and hence there should be two values in the result array, i.e., [1, 2]. However, the current result of the second row is [2] only.

SELECT
    MIN_BY(c0, c0, c1) OVER (
        PARTITION BY
            c2
        ORDER BY
            c3 ASC
        RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    )
FROM (
    VALUES
        (1, 10, FALSE, 0),
        (2, 10, FALSE, 1)
) AS t(c0, c1, c2, c3)

This error is caused by AbstractMinMaxByNAggregationFunction::output() breaking the assumption for "expanding frame" in AggregateWindowFunction::processRow(). AggregateWindowFunction::processRow() has an optimization branch for "same or expanding frame" of the previously computed frame, where it only add additional input rows of the new frame to accumulator.

presto/presto-main/src/main/java/com/facebook/presto/operator/window/AggregateWindowFunction.java

Lines 67 to 71 in d0b658e

	else if ((frameStart == currentStart) && (frameEnd >= currentEnd)) {
	// same or expanding frame
	accumulate(currentEnd + 1, frameEnd);
	currentEnd = frameEnd;
	}

This draws an implicit assumption that the content of the accumulator of the previous frame remains in the accumulator when processing the current frame. This, however, is not true with min_by because AbstractMinMaxByNAggregationFunction::output() clears the accumulator.

presto/presto-main/src/main/java/com/facebook/presto/operator/aggregation/minmaxby/AbstractMinMaxByNAggregationFunction.java

Line 147 in cb582bc

heap.popAll(reversedBlockBuilder);

Your Environment

• Presto version used: 0.286

Expected Behavior

Expected result is

[1]
[1,2]

Current Behavior

Current result is

[1]
[2]

Possible Solution

Add heap.addAll(reversedBlockBuilder); to the end of AbstractMinMaxByNAggregationFunction::output() before out.closeEntry().

Steps to Reproduce

Screenshots (if appropriate)

Context

[kagamiori]

cc @mbasmanova

[mbasmanova]

CC: @aditi-pandit @tdcmeehan @kaikalur @amitkdutta

[mbasmanova]

CC: @arhimondr