From ad7f86b5cd0c4468e5350da8834ff21b22961277 Mon Sep 17 00:00:00 2001
From: PHILO-HE <feilong.he@intel.com>
Date: Wed, 11 Dec 2024 10:48:37 +0800
Subject: [PATCH] Refactor the code

---
 velox/exec/WindowPartition.cpp | 106 +++++++++++++++++----------------
 1 file changed, 55 insertions(+), 51 deletions(-)

diff --git a/velox/exec/WindowPartition.cpp b/velox/exec/WindowPartition.cpp
index 309f92d96eba..edd5438b2b9f 100644
--- a/velox/exec/WindowPartition.cpp
+++ b/velox/exec/WindowPartition.cpp
@@ -365,8 +365,6 @@ void WindowPartition::updateKRangeFrameBounds(
 
   vector_size_t start = 0;
   vector_size_t end;
-  // vector_size_t lastFoundIndex = isPreceding ? 0 : -1;
-  vector_size_t lastFoundIndex = 0;
   // frameColumn is a column index into the original input rows, while
   // orderByColumn and mappedFrameColumn are column indices into rows in data_
   // after the columns are reordered as per inputMapping_.
@@ -409,59 +407,65 @@ void WindowPartition::updateKRangeFrameBounds(
             mappedFrameColumn,
             flags) == 0) {
       rawFrameBounds[i] = rawPeerBounds[i];
-    } else {
-      // If the search is for a preceding bound then rows between
-      // [0, currentRow] are examined. For following bounds, rows between
-      // [currentRow, numRows()) are checked.
-      if (isPreceding) {
-        // start =
-        //     lastFoundIndex == -1 ? 0 : std::min(lastFoundIndex, numRows - 1);
-        if (i == 0) {
-          start = 0;
-        } else {
-          auto compareResult = data_->compare(
-              partition_[currentRow - 1],
-              partition_[currentRow],
-              mappedFrameColumn,
-              mappedFrameColumn,
-              flags);
-          if (compareResult <= 0) {
-            start = lastFoundIndex == -1 ? 0 : std::min(lastFoundIndex, i);
-          } else {
-            start = 0;
-          }
-        }
-        end = currentRow + 1;
-      } else {
-        // start = lastFoundIndex == -1 ? currentRow
-        //                              : std::min(lastFoundIndex, numRows - 1);
-        if (i == 0) {
-          start = currentRow;
-        } else {
-          auto compareResult = data_->compare(
-              partition_[currentRow - 1],
-              partition_[currentRow],
-              mappedFrameColumn,
-              mappedFrameColumn,
-              flags);
-          if (compareResult <= 0) {
-            start = lastFoundIndex == -1 ? currentRow : lastFoundIndex;
-          } else {
-            start = currentRow;
-          }
-        }
-        end = partition_.size();
+      continue;
+    }
+
+    auto compareWithLastRow = [&](vector_size_t currentRow) {
+      // For the first row, not to apply the sliding pattern.
+      if (currentRow == startRow) {
+        // For the first row, just returns a arbitrary positive value
+        // to indicate not to use sliding search pattern.
+        return 1;
       }
-      rawFrameBounds[i] = searchFrameValue(
-          firstMatch,
-          start,
-          end,
-          currentRow,
-          orderByColumn,
+      return data_->compare(
+          partition_[currentRow - 1],
+          partition_[currentRow],
+          mappedFrameColumn,
           mappedFrameColumn,
           flags);
-      lastFoundIndex = rawFrameBounds[i];
+    };
+
+    // If the search is for a preceding bound then rows between
+    // [0, currentRow] are examined. For following bounds, rows between
+    // [currentRow, numRows()) are checked. If the sliding search pattern
+    // is applicable, we also use the last row's search result to determine
+    // the start search bound for current row.
+
+    if (compareWithLastRow(currentRow) == 0) {
+      // As same as last row's searching result.
+      rawFrameBounds[i] = rawFrameBounds[i - 1];
+      continue;
     }
+
+    if (isPreceding) {
+      // It's the first row or it doesn't match the sliding search pattern.
+      if (compareWithLastRow(currentRow) > 0) {
+        start = 0;
+      } else {
+        start = rawFrameBounds[i - 1] == -1
+            ? 0
+            : std::min(rawFrameBounds[i - 1], i);
+      }
+      end = currentRow + 1;
+    } else {
+      // It's the first row or it doesn't match the sliding search pattern.
+      if (compareWithLastRow(currentRow) > 0) {
+        start = currentRow;
+      } else {
+        start =
+            rawFrameBounds[i - 1] == -1 ? currentRow : rawFrameBounds[i - 1];
+      }
+      end = partition_.size();
+    }
+
+    rawFrameBounds[i] = searchFrameValue(
+        firstMatch,
+        start,
+        end,
+        currentRow,
+        orderByColumn,
+        mappedFrameColumn,
+        flags);
   }
 }