Fix null value check for kRange frame column (facebookincubator#11075)

Summary:
The columns in `WindowPartition::data_` are reordered to start with partition-by keys and order-by keys, whereas the columns in `WindowPartition::columns_` are in the same order as the Window operator. The variable `orderByColumn` contains the index of order-by key in the reordered columns, so this should be accessed from `data_` and not `columns_`. A unit test is added to ensure an exception is thrown when NULL values in order-by column and frame column do not match.

Pull Request resolved: facebookincubator#11075

Reviewed By: amitkdutta

Differential Revision: D63404094

Pulled By: kagamiori

fbshipit-source-id: cae3713c3c6c945001af8c2930ba08a34d574f67

velox/exec/WindowPartition.cpp

@@ -363,8 +363,11 @@ void WindowPartition::updateKRangeFrameBounds(
 
   vector_size_t start = 0;
   vector_size_t end;
+  // frameColumn is a column index into the original input rows, while
+  // orderByColumn is a column index into rows in data_ after the columns are
+  // reordered as per inputMapping_.
   RowColumn frameRowColumn = columns_[frameColumn];
-  RowColumn orderByRowColumn = columns_[inputMapping_[orderByColumn]];
+  RowColumn orderByRowColumn = data_->columnAt(orderByColumn);
   for (auto i = 0; i < numRows; i++) {
     auto currentRow = startRow + i;
     auto* partitionRow = partition_[currentRow];

velox/exec/tests/WindowTest.cpp

@@ -487,5 +487,41 @@ TEST_F(WindowTest, nagativeFrameArg) {
   }
 }
 
+DEBUG_ONLY_TEST_F(WindowTest, frameColumnNullCheck) {
+  auto makePlan = [&](const RowVectorPtr& input) {
+    return PlanBuilder()
+        .values({input})
+        .window(
+            {"sum(c0) OVER (PARTITION BY p0 ORDER BY s0 RANGE BETWEEN UNBOUNDED PRECEDING AND off0 FOLLOWING)"})
+        .planNode();
+  };
+
+  // Null values in order-by column 's0' and frame column 'off0' do not match,
+  // so exception is expected.
+  auto inputThrow = makeRowVector(
+      {"c0", "p0", "s0", "off0"},
+      {
+          makeNullableFlatVector<int64_t>({1, std::nullopt, 1, 2, 2}),
+          makeFlatVector<int64_t>({1, 2, 1, 2, 1}),
+          makeNullableFlatVector<int64_t>({1, 2, 3, std::nullopt, 5}),
+          makeNullableFlatVector<int64_t>({2, std::nullopt, 4, 5, 6}),
+      });
+  VELOX_ASSERT_THROW(
+      AssertQueryBuilder(makePlan(inputThrow)).copyResults(pool()), "");
+
+  // Null values in order-by column 's0' and frame column 'off0' match, so no
+  // exception should be thrown.
+  auto inputNoThrow = makeRowVector(
+      {"c0", "p0", "s0", "off0"},
+      {
+          makeNullableFlatVector<int64_t>({1, 1, 2, std::nullopt, 2}),
+          makeFlatVector<int64_t>({1, 1, 1, 2, 2}),
+          makeNullableFlatVector<int64_t>({1, std::nullopt, 2, 3, 5}),
+          makeNullableFlatVector<int64_t>({2, std::nullopt, 3, 4, 6}),
+      });
+  ASSERT_NO_THROW(
+      AssertQueryBuilder(makePlan(inputNoThrow)).copyResults(pool()));
+}
+
 } // namespace
 } // namespace facebook::velox::exec