add unit test

Author: jiangtian

velox/exec/Window.h

@@ -67,6 +67,10 @@ class Window : public Operator {
   void reclaim(uint64_t targetBytes, memory::MemoryReclaimer::Stats& stats)
       override;
 
+  const HashStringAllocator* testingGetHashStringAllocator() const {
+    return &stringAllocator_;
+  }
+
   /// Runtime statistics holding total number of batches read from spilled data.
   /// 0 if no spilling occurred.
   static inline const std::string kWindowSpillReadNumBatches{

velox/exec/tests/WindowTest.cpp

@@ -26,6 +26,7 @@
 #include "velox/exec/tests/utils/PlanBuilder.h"
 #include "velox/exec/tests/utils/TempDirectoryPath.h"
 #include "velox/functions/prestosql/window/WindowFunctionsRegistration.h"
+#include "velox/functions/sparksql/aggregates/Register.h"
 
 using namespace facebook::velox::exec::test;
 
@@ -1036,5 +1037,67 @@ DEBUG_ONLY_TEST_F(WindowTest, releaseWindowBuildInTime) {
               "ORDER BY d");
 }
 
+class SparkWindowTest : public WindowTest {
+ public:
+  void SetUp() override {
+    WindowTest::SetUp();
+    functions::aggregate::sparksql::registerAggregateFunctions("");
+  }
+};
+
+DEBUG_ONLY_TEST_F(SparkWindowTest, destroyPreviousAccumulator) {
+  const auto size = 100;
+  auto input = makeRowVector(
+      {"d", "p0", "s"},
+      {
+          // Payload Data.
+          makeFlatVector<std::string>(size, [](auto row){ return std::string(1024, 'a'); }),
+          // Partition key.
+          makeFlatVector<int64_t>(size, [](auto row) { return row % 11; }),
+          // Sorting key.
+          makeFlatVector<int32_t>(size, [](auto row) { return row; }),
+      });
+
+  createDuckDbTable({input});
+
+  core::PlanNodeId windowId;
+  core::PlanNodeId orderById;
+  auto plan = PlanBuilder()
+                  .values(split(input, 10))
+                  .window({"last(d) over (partition by p0 order by s)"})
+                  .capturePlanNodeId(windowId)
+                  .orderBy({"d"}, false)
+                  .capturePlanNodeId(orderById)
+                  .planNode();
+
+  const HashStringAllocator* stringAllocator = nullptr;
+  uint64_t usedBytes = 0;
+  SCOPED_TESTVALUE_SET(
+      "facebook::velox::exec::Driver::runInternal::getOutput",
+      std::function<void(Operator*)>([&](exec::Operator* op) {
+        auto* windowOp = dynamic_cast<exec::Window*>(op);
+        if (windowOp == nullptr) {
+          return;
+        }
+        if (stringAllocator == nullptr) {
+          stringAllocator = windowOp->testingGetHashStringAllocator();
+        } else if (usedBytes == 0) {
+          // Record how many bytes have been used.
+          usedBytes = stringAllocator->currentBytes();
+        } else {
+          // Because we will destroy previous created accumulator and every string in input
+          // is of the same length, so here we check if the `usedBytes` is not changed.
+          ASSERT_EQ(usedBytes, stringAllocator->currentBytes());
+        }
+      }));
+
+  auto task =
+      AssertQueryBuilder(plan, duckDbQueryRunner_)
+          .config(core::QueryConfig::kPreferredOutputBatchBytes, "1024")
+          .assertResults(
+              "SELECT *, last(d) over (partition by p0 order by s) "
+              "FROM tmp ");
+}
+
 } // namespace
 } // namespace facebook::velox::exec