min_by/max_by returns wrong result in Window operation

[kagamiori]

Description

Window fuzzer found a bug of min_by(x, y, n) and max_by(x, y, n) when evaluated in the Window operator. The error is that for a given input row where y is NULL, Velox returns an empty array [] while Presto returns a NULL.

This error happens when there are at least two rows in a partition where the first row has non-null y and the second row has a NULL y, and the frame is RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW. What happens is that Window operator performs incremental aggregation for this partition. After the result for the first row is produced that is a non-null array, the is-null flag of the accumulator (i.e., isNull(group)) isn't reset to true before the second row is processed. Since y at the second row is NULL, MinMaxByNAggregate ignores this input row, so the accumulator remains empty (elements added by the previous row were already removed when the previous row's result was extracted). When the result of the second row is extracted, since isNull(group) is false and the accumulator has no element, an empty array is generated.

Error Reproduction

TEST_F(MinMaxByNTest, aaa) {
  auto data = makeRowVector({
          makeFlatVector<int32_t>({3, 3}),
          makeNullableFlatVector<int64_t>({2, std::nullopt}),
          makeFlatVector<int64_t>({1, 1}),
          makeFlatVector<bool>({false, false}),
          makeFlatVector<int64_t>({0, 1}),
      });

  auto plan = PlanBuilder()
                  .values({data})
                  .window({"min_by(c0, c1, c2) over (partition by c3 order by c4 asc)"})
                  .planNode();

  // Result should be two rows: {3, 2, 1, false, 0, [3]}, {3, null, 1, false, 1, null}.
  // Current wrong result is: {3, 2, 1, false, 0, [3]}, {3, null, 1, false, 1, []}.
  auto result = AssertQueryBuilder(plan).copyResults(pool());
}

Relevant logs

No response

[kagamiori]

cc @mbasmanova @aditi-pandit

[aditi-pandit]

@kagamiori : Are you interested in fixing this ? If not, I'll take a look.

[kagamiori]

@kagamiori : Are you interested in fixing this ? If not, I'll take a look.

Hi @aditi-pandit, please feel free to go ahead. Thanks!

[aditi-pandit]

@kagamiori : I'm on vacation this month with limited availability. It might be better for you to take a look if this is blocking you.

[kevinmingtarja]

Hi, I'm a first time contributor. If @kagamiori has other priorities, I'd be more than happy to take a look at this!

[kagamiori]

Hi, I'm a first time contributor. If @kagamiori has other priorities, I'd be more than happy to take a look at this!

Hi @kevinmingtarja, thank you for reaching out! This issue is a bit time-sensitive. For first time contributors, we'd suggest starting from the list of good first issues. Also, here is a document about code contribution that would be helpful.

[kevinmingtarja]

Got it. I'll take a look at other issues with less severity then! Thanks.

[kagamiori]

There are two bugs with min_by/max_by in Window operation, as explained below, one in MinMaxByNAggregate::extractValues() and the other in AggregateWindowFunction::incrementalAggregation().

1. MinMaxByNAggregate::extractValues() should return NULL when the accumulator is empty to follow Presto's behavior. https://github.com/prestodb/presto/blob/cb582bce0fd18f51d8862a8b2f53a134780f41aa/presto-main/src/main/java/com/facebook/presto/operator/aggregation/minmaxby/AbstractMinMaxByNAggregationFunction.java#L137-L140. This would solve the original error described above.
2. A new bug discovered during investigation is that when there are two peer rows in the same frame, Presto still calls Aggregate::extactValues() on the second row (https://github.com/prestodb/presto/blob/cb582bce0fd18f51d8862a8b2f53a134780f41aa/presto-main/src/main/java/com/facebook/presto/operator/window/AggregateWindowFunction.java#L80), while Velox only copies the result of the previous row (

   velox/velox/exec/AggregateWindow.cpp

   Line 305 in 7376fb2

   result->copy(aggregateResultVector_.get(), resultOffset + i, 0, 1);

   ). This distinction makes a difference in particular for min_by and max_by because the accumulator type of these two functions clears the existing content when Aggregate::extractValues() is called, so the second row in the same frame should get a different result from the first row.

Both bugs can be reproduced via the following unit test.

TEST_F(MinMaxByNTest, aaa) {
  // SELECT
  //  c0, c1, c2, c3, c4, 
  //  min_by(c0, c1, c2) over (partition by c3 order by c4 asc)
  // FROM (
  //  VALUES
  //      (4, 2, 1, false, 0),
  //      (3, 1, 1, false, 0),
  //      (2, 0, 1, false, 1),
  //      (1, null, 1, false, 2)
  // ) AS t(c0, c1, c2, c3, c4);
  auto data = makeRowVector({
      makeFlatVector<int32_t>({4, 3, 2, 1}),
      makeNullableFlatVector<int64_t>({2, 1, 0, std::nullopt}),
      makeFlatVector<int64_t>({1, 1, 1, 1}),
      makeFlatVector<bool>({false, false, false, false}),
      makeFlatVector<int64_t>({0, 0, 1, 2}),
  });

  auto plan =
      PlanBuilder()
          .values({data})
          .window({"min_by(c0, c1, c2) over (partition by c3 order by c4 asc)"})
          .planNode();

  // Result should be {4, 2, 1, false, 0, [3]}, {3, 1, 1, false, 0,
  // null}, {2, 0, 1, false, 1, [2]}, {1, null, 1, false, 2, null}.
  auto result = AssertQueryBuilder(plan).copyResults(pool());
  auto expected = makeRowVector({
      makeFlatVector<int32_t>({4, 3, 2, 1}),
      makeNullableFlatVector<int64_t>({2, 1, 0, std::nullopt}),
      makeFlatVector<int64_t>({1, 1, 1, 1}),
      makeFlatVector<bool>({false, false, false, false}),
      makeFlatVector<int64_t>({0, 0, 1, 2}),
      makeNullableArrayVector<int32_t>(
          {{std::vector<std::optional<int32_t>>(1, 3)},
           std::nullopt,
           {std::vector<std::optional<int32_t>>(1, 2)},
           std::nullopt}),
  });
  velox::test::assertEqualVectors(expected, result);
}

[aditi-pandit]

Great analysis @kagamiori. Thanks !

[kagamiori]

Another bug in min/max(x, n) is found during the testing of the fix #8296.

TEST_F(MinMaxNTest, aaa) {
  // SELECT
  //  c0, c1, c2, c3,
  //  max(c0, c1) over (partition by c2 order by c3 asc)
  // FROM (
  //  VALUES
  //      (1, 10, false, 0),
  //      (2, 10, false, 1)
  // ) AS t(c0, c1, c2, c3)
  auto data = makeRowVector({
      makeFlatVector<int64_t>({1, 2}),
      makeFlatVector<int64_t>({10, 10}),
      makeFlatVector<bool>({false, false}),
      makeFlatVector<int64_t>({0, 1}),
  });

  auto plan =
      PlanBuilder()
          .values({data})
          .window({"max(c0, c1) over (partition by c2 order by c3 asc)"})
          .planNode();

  auto result = AssertQueryBuilder(plan).copyResults(pool());
  auto expected = makeRowVector({
      makeFlatVector<int64_t>({1, 2}),
      makeFlatVector<int64_t>({10, 10}),
      makeFlatVector<bool>({false, false}),
      makeFlatVector<int64_t>({0, 1}),
      makeArrayVector<int64_t>({{1}, {2, 1}}),
  });
  // Expected result: {1, 10, false, 0, [1]}, {2, 10, false, 1, [2, 1]}.
  // Current wrong result: {1, 10, false, 0, [1]}, {2, 10, false, 1, [2]}.
  facebook::velox::test::assertEqualVectors(expected, result);
}

The root cause is that for min/max(x, n), when aggregation result is extracted from accumulator, the accumulator should retain its content (https://github.com/prestodb/presto/blob/bed052afeef85c728d0c99237be1047f60e24839/presto-main/src/main/java/com/facebook/presto/operator/aggregation/AbstractMinMaxNAggregationFunction.java#L180). In the current Velox implementation, accumulator content is cleared after the content is extracted (

velox/velox/functions/prestosql/aggregates/MinMaxAggregates.cpp

Line 604 in 7504f7f

topValues.pop();

).

One interesting thing is that this behavior of min/max(x, n) is exactly the opposite to the behavior of min_by/max_by(x, y, n) in Presto that clears the content in accumulator after extraction (https://github.com/prestodb/presto/blob/cb582bce0fd18f51d8862a8b2f53a134780f41aa/presto-main/src/main/java/com/facebook/presto/operator/aggregation/minmaxby/AbstractMinMaxByNAggregationFunction.java#L153). I wonder whether this difference is intended or a bug in Presto.

cc @mbasmanova