From 6c43dc5eef407d3cfb6b44249c01a870fa70c03e Mon Sep 17 00:00:00 2001
From: xiaoxmeng <xiaoxmeng@fb.com>
Date: Sun, 1 Sep 2024 11:25:21 -0700
Subject: [PATCH] Enable probe spill with dynamic filter replaced (#10849)

Summary:
Enable probe spill with dynamic filter replaced. Now the fuzzer test can pass 10hrs run and the
previous issue in table scan for lazy vector processing might have been fixed.

Pull Request resolved: https://github.com/facebookincubator/velox/pull/10849

Reviewed By: gggrace14, tanjialiang

Differential Revision: D61834373

Pulled By: xiaoxmeng

fbshipit-source-id: bc5361f16eb25ad6867c556b1cd78583be494bd4
---
 velox/exec/HashProbe.cpp          |  4 +---
 velox/exec/Operator.cpp           | 11 +++++++++++
 velox/exec/tests/HashJoinTest.cpp |  6 +-----
 3 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/velox/exec/HashProbe.cpp b/velox/exec/HashProbe.cpp
index efc3fe59c677..738e64c8f978 100644
--- a/velox/exec/HashProbe.cpp
+++ b/velox/exec/HashProbe.cpp
@@ -1654,9 +1654,7 @@ void HashProbe::ensureOutputFits() {
 }
 
 bool HashProbe::canReclaim() const {
-  // NOTE: we can't spill from a hash probe operator if it has generated dynamic
-  // filters.
-  return spillEnabled() && !hasGeneratedDynamicFilters_;
+  return spillEnabled();
 }
 
 void HashProbe::reclaim(
diff --git a/velox/exec/Operator.cpp b/velox/exec/Operator.cpp
index 0798785a031d..67afed0eafa3 100644
--- a/velox/exec/Operator.cpp
+++ b/velox/exec/Operator.cpp
@@ -657,6 +657,17 @@ uint64_t Operator::MemoryReclaimer::reclaim(
           memory::ScopedReclaimedBytesRecorder recoder(pool, &reclaimedBytes);
           op_->reclaim(targetBytes, stats);
         }
+        // NOTE: the parallel hash build is running at the background thread
+        // pool which won't stop during memory reclamation so the operator's
+        // memory usage might increase in such case. memory usage.
+        if (op_->operatorType() == "HashBuild") {
+          reclaimedBytes = std::max<int64_t>(0, reclaimedBytes);
+        }
+        VELOX_CHECK_GE(
+            reclaimedBytes,
+            0,
+            "Unexpected memory growth after reclaim from operator memory pool {}",
+            pool->name());
         return reclaimedBytes;
       },
       stats);
diff --git a/velox/exec/tests/HashJoinTest.cpp b/velox/exec/tests/HashJoinTest.cpp
index b76db56e1eff..f9c2402b65ef 100644
--- a/velox/exec/tests/HashJoinTest.cpp
+++ b/velox/exec/tests/HashJoinTest.cpp
@@ -8020,10 +8020,6 @@ DEBUG_ONLY_TEST_F(HashJoinTest, spillCheckOnLeftSemiFilterWithDynamicFilters) {
                        .values(buildVectors)
                        .project({"c0 AS u_c0", "c1 AS u_c1"})
                        .planNode();
-  auto keyOnlyBuildSide = PlanBuilder(planNodeIdGenerator, pool_.get())
-                              .values(keyOnlyBuildVectors)
-                              .project({"c0 AS u_c0"})
-                              .planNode();
 
   // Left semi join.
   core::PlanNodeId probeScanId;
@@ -8070,7 +8066,7 @@ DEBUG_ONLY_TEST_F(HashJoinTest, spillCheckOnLeftSemiFilterWithDynamicFilters) {
         // Verify spill hasn't triggered.
         auto taskStats = exec::toPlanStats(task->taskStats());
         auto& planStats = taskStats.at(joinNodeId);
-        ASSERT_EQ(planStats.spilledBytes, 0);
+        ASSERT_GT(planStats.spilledBytes, 0);
       })
       .run();
 }