Fix arbitrary() to always return the first non-null value in Window o…

…peration (facebookincubator#8640)

Summary:
Pull Request resolved: facebookincubator#8640

The implementation of arbitrary() intends to always return the first non-null value, but it
doesn't for complex-typed inputs when used in Window operation with incremental frames.
This is because ArbitraryFunction::addSingleGroupRawInput() still updates the accumulator
even if the accumulator already has a value. This diff fixes this issue by making
ArbitraryFunction::addSingleGroupRawInput() return immediately if accumulator already has
a value.

This diff fixes facebookincubator#8593.

Reviewed By: kgpai

Differential Revision: D53328253

fbshipit-source-id: 803261dce0ec1fc52187947b1f9316dfd814c3fd

velox/functions/prestosql/aggregates/ArbitraryAggregate.cpp

@@ -226,6 +226,11 @@ class NonNumericArbitrary : public exec::Aggregate {
       const SelectivityVector& rows,
       const std::vector<VectorPtr>& args,
       bool /*unused*/) override {
+    auto* accumulator = value<SingleValueAccumulator>(group);
+    if (accumulator->hasValue()) {
+      return;
+    }
+
     DecodedVector decoded(*args[0], rows, true);
     if (decoded.isConstantMapping() && decoded.isNullAt(0)) {
       // nothing to do; all values are nulls
@@ -234,7 +239,6 @@ class NonNumericArbitrary : public exec::Aggregate {
 
     const auto* indices = decoded.indices();
     const auto* baseVector = decoded.base();
-    auto* accumulator = value<SingleValueAccumulator>(group);
     // Find the first non-null value.
     rows.testSelected([&](vector_size_t i) {
       if (!decoded.isNullAt(i)) {

velox/functions/prestosql/aggregates/tests/ArbitraryTest.cpp

@@ -16,9 +16,11 @@
 
 #include "velox/exec/tests/utils/PlanBuilder.h"
 #include "velox/functions/lib/aggregates/tests/utils/AggregationTestBase.h"
+#include "velox/functions/lib/window/tests/WindowTestBase.h"
 
 using namespace facebook::velox::exec::test;
 using namespace facebook::velox::functions::aggregate::test;
+using namespace facebook::velox::window::test;
 
 namespace facebook::velox::aggregate::test {
 
@@ -367,5 +369,38 @@ TEST_F(ArbitraryTest, interval) {
   testAggregations({data}, {}, {"arbitrary(c2)"}, "SELECT null");
 }
 
+class ArbitraryWindowTest : public WindowTestBase {};
+
+TEST_F(ArbitraryWindowTest, basic) {
+  auto data = makeRowVector(
+      {makeFlatVector<int64_t>({1, 2, 3, 4, 5}),
+       makeArrayVector<double>({{1.0}, {2.0}, {3.0}, {4.0}, {5.0}}),
+       makeFlatVector<bool>({false, false, false, false, false})});
+
+  auto expected = makeRowVector(
+      {makeFlatVector<int64_t>({1, 2, 3, 4, 5}),
+       makeArrayVector<double>({{1.0}, {2.0}, {3.0}, {4.0}, {5.0}}),
+       makeFlatVector<bool>({false, false, false, false, false}),
+       makeFlatVector<int64_t>({1, 1, 1, 1, 1})});
+  window::test::WindowTestBase::testWindowFunction(
+      {data},
+      "arbitrary(c0)",
+      "partition by c2 order by c0",
+      "range between unbounded preceding and current row",
+      expected);
+
+  expected = makeRowVector(
+      {makeFlatVector<int64_t>({1, 2, 3, 4, 5}),
+       makeArrayVector<double>({{1.0}, {2.0}, {3.0}, {4.0}, {5.0}}),
+       makeFlatVector<bool>({false, false, false, false, false}),
+       makeArrayVector<double>({{1.0}, {1.0}, {1.0}, {1.0}, {1.0}})});
+  window::test::WindowTestBase::testWindowFunction(
+      {data},
+      "arbitrary(c1)",
+      "partition by c2 order by c0",
+      "range between unbounded preceding and current row",
+      expected);
+}
+
 } // namespace
 } // namespace facebook::velox::aggregate::test

velox/functions/prestosql/aggregates/tests/CMakeLists.txt

@@ -64,6 +64,7 @@ target_link_libraries(
   velox_file
   velox_functions_aggregates
   velox_functions_aggregates_test_lib
+  velox_functions_window_test_lib
   velox_functions_test_lib
   velox_functions_prestosql
   velox_functions_lib