[Gluten-1.2] Backport facebookincubator#9025 and facebookincubator#10979 to fix the result mismatch in RowsStreamingWindowBuild (#499)

* Revert "Add RowsStreamingWindowBuild to avoid OOM in Window operator (9025)"

This reverts commit f34c9b1.

* Add RowsStreamingWindowBuild to avoid OOM in Window operator (facebookincubator#9025)

Summary:
Unlike `StreamingWindowBuild`, `RowLevelStreamingWindowBuild ` in this PR is capable of processing window functions as rows arrive within a single partition, without the need to wait for the entire partition to be ready. This approach can significantly reduce memory usage, especially when a single partition contains a large amount of data. It is particularly suited for optimizing `rank `and `row_number `functions, as well as aggregate window functions with a default frame.

The detailed discussions is [here](facebookincubator#8975). The design doc is [here](https://docs.google.com/document/d/17ONSJHK8XP5Lixm8XBl01RMNl4ntpixiVFe693ahw6k/edit?usp=sharing).

Pull Request resolved: facebookincubator#9025

Test Plan: Run through 10hrs fuzzer testing

Reviewed By: kagamiori

Differential Revision: D61473798

Pulled By: xiaoxmeng

fbshipit-source-id: 569a752770395330c48a3521bd5421eb89f5623d

* Fix error message

* Fix the result mismatch in RowsStreamingWindowBuild (facebookincubator#10979)

Summary:
For a Range frame, it is necessary to ensure that the peer is ready before commencing the window function computation

Pull Request resolved: facebookincubator#10979

Reviewed By: kagamiori

Differential Revision: D62622816

Pulled By: xiaoxmeng

fbshipit-source-id: 1a9911da416c867c9e295242a05d0f33fbc2e22d

---------

Co-authored-by: Jia Ke <[email protected]>

Author: ccat3z

velox/exec/AggregateWindow.cpp

@@ -410,6 +410,7 @@ void registerAggregateWindowFunction(const std::string& name) {
     exec::registerWindowFunction(
         name,
         std::move(signatures),
+        {exec::WindowFunction::ProcessMode::kRows, true},
         [name](
             const std::vector<exec::WindowFunctionArg>& args,
             const TypePtr& resultType,
@@ -426,8 +427,7 @@ void registerAggregateWindowFunction(const std::string& name) {
               pool,
               stringAllocator,
               config);
-        },
-        {exec::ProcessedUnit::kRows, true});
+        });
   }
 }
 } // namespace facebook::velox::exec

velox/exec/CMakeLists.txt

@@ -59,6 +59,7 @@ add_library(
   OutputBufferManager.cpp
   PartitionedOutput.cpp
   PartitionFunction.cpp
+  PartitionStreamingWindowBuild.cpp
   PlanNodeStats.cpp
   PrefixSort.cpp
   ProbeOperatorState.cpp
@@ -72,7 +73,6 @@ add_library(
   SpillFile.cpp
   Spiller.cpp
   StreamingAggregation.cpp
-  StreamingWindowBuild.cpp
   Strings.cpp
   TableScan.cpp
   TableWriteMerge.cpp

velox/exec/StreamingWindowBuild.cpp renamed to velox/exec/PartitionStreamingWindowBuild.cpp

@@ -14,24 +14,24 @@
  * limitations under the License.
  */
 
-#include "velox/exec/StreamingWindowBuild.h"
+#include "velox/exec/PartitionStreamingWindowBuild.h"
 
 namespace facebook::velox::exec {
 
-StreamingWindowBuild::StreamingWindowBuild(
+PartitionStreamingWindowBuild::PartitionStreamingWindowBuild(
     const std::shared_ptr<const core::WindowNode>& windowNode,
     velox::memory::MemoryPool* pool,
     const common::SpillConfig* spillConfig,
     tsan_atomic<bool>* nonReclaimableSection)
     : WindowBuild(windowNode, pool, spillConfig, nonReclaimableSection) {}
 
-void StreamingWindowBuild::buildNextPartition() {
+void PartitionStreamingWindowBuild::buildNextPartition() {
   partitionStartRows_.push_back(sortedRows_.size());
   sortedRows_.insert(sortedRows_.end(), inputRows_.begin(), inputRows_.end());
   inputRows_.clear();
 }
 
-void StreamingWindowBuild::addInput(RowVectorPtr input) {
+void PartitionStreamingWindowBuild::addInput(RowVectorPtr input) {
   for (auto i = 0; i < inputChannels_.size(); ++i) {
     decodedInputVectors_[i].decode(*input->childAt(inputChannels_[i]));
   }
@@ -53,14 +53,15 @@ void StreamingWindowBuild::addInput(RowVectorPtr input) {
   }
 }
 
-void StreamingWindowBuild::noMoreInput() {
+void PartitionStreamingWindowBuild::noMoreInput() {
   buildNextPartition();
 
   // Help for last partition related calculations.
   partitionStartRows_.push_back(sortedRows_.size());
 }
 
-std::shared_ptr<WindowPartition> StreamingWindowBuild::nextPartition() {
+std::shared_ptr<WindowPartition>
+PartitionStreamingWindowBuild::nextPartition() {
   VELOX_CHECK_GT(
       partitionStartRows_.size(), 0, "No window partitions available")
 
@@ -93,7 +94,7 @@ std::shared_ptr<WindowPartition> StreamingWindowBuild::nextPartition() {
       data_.get(), partition, inversedInputChannels_, sortKeyInfo_);
 }
 
-bool StreamingWindowBuild::hasNextPartition() {
+bool PartitionStreamingWindowBuild::hasNextPartition() {
   return partitionStartRows_.size() > 0 &&
       currentPartition_ < int(partitionStartRows_.size() - 2);
 }

velox/exec/StreamingWindowBuild.h renamed to velox/exec/PartitionStreamingWindowBuild.h

@@ -20,13 +20,13 @@
 
 namespace facebook::velox::exec {
 
-/// The StreamingWindowBuild is used when the input data is already sorted by
-/// {partition keys + order by keys}. The logic identifies partition changes
-/// when receiving input rows and splits out WindowPartitions for the Window
-/// operator to process.
-class StreamingWindowBuild : public WindowBuild {
+/// The PartitionStreamingWindowBuild is used when the input data is already
+/// sorted by {partition keys + order by keys}. The logic identifies partition
+/// changes when receiving input rows and splits out WindowPartitions for the
+/// Window operator to process.
+class PartitionStreamingWindowBuild : public WindowBuild {
  public:
-  StreamingWindowBuild(
+  PartitionStreamingWindowBuild(
       const std::shared_ptr<const core::WindowNode>& windowNode,
       velox::memory::MemoryPool* pool,
       const common::SpillConfig* spillConfig,
@@ -55,10 +55,6 @@ class StreamingWindowBuild : public WindowBuild {
         currentPartition_ == partitionStartRows_.size() - 2;
   }
 
-  std::string_view windowBuildType() const override {
-    return "StreamingWindowBuild";
-  }
-
  private:
   void buildNextPartition();
 

velox/exec/RowsStreamingWindowBuild.cpp

@@ -16,35 +16,54 @@
 
 #include "velox/exec/RowsStreamingWindowBuild.h"
 #include "velox/common/testutil/TestValue.h"
+#include "velox/exec/WindowFunction.h"
 
 namespace facebook::velox::exec {
 
+namespace {
+bool hasRangeFrame(const std::shared_ptr<const core::WindowNode>& windowNode) {
+  for (const auto& function : windowNode->windowFunctions()) {
+    if (function.frame.type == core::WindowNode::WindowType::kRange) {
+      return true;
+    }
+  }
+  return false;
+}
+} // namespace
+
 RowsStreamingWindowBuild::RowsStreamingWindowBuild(
     const std::shared_ptr<const core::WindowNode>& windowNode,
     velox::memory::MemoryPool* pool,
     const common::SpillConfig* spillConfig,
     tsan_atomic<bool>* nonReclaimableSection)
-    : WindowBuild(windowNode, pool, spillConfig, nonReclaimableSection) {}
+    : WindowBuild(windowNode, pool, spillConfig, nonReclaimableSection),
+      hasRangeFrame_(hasRangeFrame(windowNode)) {
+  velox::common::testutil::TestValue::adjust(
+      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      this);
+}
+
+void RowsStreamingWindowBuild::addPartitionInputs(bool finished) {
+  if (inputRows_.empty()) {
+    return;
+  }
 
-void RowsStreamingWindowBuild::buildNextInputOrPartition(bool isFinished) {
   if (windowPartitions_.size() <= inputPartition_) {
     windowPartitions_.push_back(std::make_shared<WindowPartition>(
         data_.get(), inversedInputChannels_, sortKeyInfo_));
   }
 
   windowPartitions_[inputPartition_]->addRows(inputRows_);
 
-  if (isFinished) {
+  if (finished) {
     windowPartitions_[inputPartition_]->setComplete();
-    inputPartition_++;
+    ++inputPartition_;
   }
 
   inputRows_.clear();
 }
 
 void RowsStreamingWindowBuild::addInput(RowVectorPtr input) {
-  velox::common::testutil::TestValue::adjust(
-      "facebook::velox::exec::RowsStreamingWindowBuild::addInput", this);
   for (auto i = 0; i < inputChannels_.size(); ++i) {
     decodedInputVectors_[i].decode(*input->childAt(inputChannels_[i]));
   }
@@ -58,11 +77,18 @@ void RowsStreamingWindowBuild::addInput(RowVectorPtr input) {
 
     if (previousRow_ != nullptr &&
         compareRowsWithKeys(previousRow_, newRow, partitionKeyInfo_)) {
-      buildNextInputOrPartition(true);
+      addPartitionInputs(true);
     }
 
     if (previousRow_ != nullptr && inputRows_.size() >= numRowsPerOutput_) {
-      buildNextInputOrPartition(false);
+      // Needs to wait the peer group ready for range frame.
+      if (hasRangeFrame_) {
+        if (compareRowsWithKeys(previousRow_, newRow, sortKeyInfo_)) {
+          addPartitionInputs(false);
+        }
+      } else {
+        addPartitionInputs(false);
+      }
     }
 
     inputRows_.push_back(newRow);
@@ -71,18 +97,17 @@ void RowsStreamingWindowBuild::addInput(RowVectorPtr input) {
 }
 
 void RowsStreamingWindowBuild::noMoreInput() {
-  buildNextInputOrPartition(true);
+  addPartitionInputs(true);
 }
 
 std::shared_ptr<WindowPartition> RowsStreamingWindowBuild::nextPartition() {
-  // The previous partition has already been set to nullptr by the
-  // Window.cpp#callResetPartition() method.
+  VELOX_CHECK(hasNextPartition());
   return windowPartitions_[++outputPartition_];
 }
 
 bool RowsStreamingWindowBuild::hasNextPartition() {
-  return windowPartitions_.size() > 0 &&
-      outputPartition_ <= int(windowPartitions_.size() - 2);
+  return !windowPartitions_.empty() &&
+      outputPartition_ + 2 <= windowPartitions_.size();
 }
 
 } // namespace facebook::velox::exec

velox/exec/RowsStreamingWindowBuild.h

@@ -20,13 +20,13 @@
 
 namespace facebook::velox::exec {
 
-/// Unlike StreamingWindowBuild, RowsStreamingWindowBuild is capable of
+/// Unlike PartitionStreamingWindowBuild, RowsStreamingWindowBuild is capable of
 /// processing window functions as rows arrive within a single partition,
-/// without the need to wait for the entire partition to be ready. This approach
-/// can significantly reduce memory usage, especially when a single partition
-/// contains a large amount of data. It is particularly suited for optimizing
-/// rank and row_number functions, as well as aggregate window functions with a
-/// default frame.
+/// without the need to wait for the entirewindow partition to be ready. This
+/// approach can significantly reduce memory usage, especially when a single
+/// partition contains a large amount of data. It is particularly suited for
+/// optimizing rank, dense_rank and row_number functions, as well as aggregate
+/// window functions with a default frame.
 class RowsStreamingWindowBuild : public WindowBuild {
  public:
   RowsStreamingWindowBuild(
@@ -54,30 +54,31 @@ class RowsStreamingWindowBuild : public WindowBuild {
   bool needsInput() override {
     // No partitions are available or the currentPartition is the last available
     // one, so can consume input rows.
-    return windowPartitions_.size() == 0 ||
+    return windowPartitions_.empty() ||
         outputPartition_ == windowPartitions_.size() - 1;
   }
 
-  std::string_view windowBuildType() const override {
-    return "RowsStreamingWindowBuild";
-  }
-
  private:
-  void buildNextInputOrPartition(bool isFinished);
+  // Adds input rows to the current partition, or creates a new partition if it
+  // does not exist.
+  void addPartitionInputs(bool finished);
+
+  // Sets to true if this window node has range frames.
+  const bool hasRangeFrame_;
 
-  // Holds input rows within the current partition.
+  // Points to the input rows in the current partition.
   std::vector<char*> inputRows_;
 
   // Used to compare rows based on partitionKeys.
   char* previousRow_ = nullptr;
 
-  // Current partition being output. Used to return the WidnowPartitions.
+  // Point to the current output partition if not -1.
   vector_size_t outputPartition_ = -1;
 
-  // Current partition when adding input. Used to construct WindowPartitions.
+  // Current input partition that receives inputs.
   vector_size_t inputPartition_ = 0;
 
-  // Holds all the WindowPartitions.
+  // Holds all the built window partitions.
   std::vector<std::shared_ptr<WindowPartition>> windowPartitions_;
 };
 

velox/exec/SortWindowBuild.h

@@ -55,10 +55,6 @@ class SortWindowBuild : public WindowBuild {
 
   std::shared_ptr<WindowPartition> nextPartition() override;
 
-  std::string_view windowBuildType() const override {
-    return "SortWindowBuild";
-  }
-
  private:
   void ensureInputFits(const RowVectorPtr& input);