Fix the SortBuffer's noMoreInput called twice

facebookincubator · Aug 6, 2024 · a5137b8 · a5137b8
1 parent c0fa8f2
commit a5137b8
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/velox/exec/MergeJoin.cpp b/velox/exec/MergeJoin.cpp
@@ -656,13 +656,20 @@ RowVectorPtr MergeJoin::getOutput() {
           for (const auto [channel, _] : filterInputToOutputChannel_) {
             filterInput_->childAt(channel).reset();
           }
+
           return output;
         }
 
         // No rows survived the filter. Get more rows.
         continue;
       } else if (isAntiJoin(joinType_)) {
-        return filterOutputForAntiJoin(output);
+        auto result = filterOutputForAntiJoin(output);
+        if (!result) {
+          // Make sure needsInput() method return true. Otherwise, Driver
+          // assumes the operator is finished.
+          input_ = nullptr;
+        }
+        return result;
       } else {
         return output;
       }