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: #10849

Reviewed By: gggrace14, tanjialiang

Differential Revision: D61834373

Pulled By: xiaoxmeng

fbshipit-source-id: bc5361f16eb25ad6867c556b1cd78583be494bd4

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(

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);

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();
 }