Fix the SortBuffer's noMoreInput called twice when enable smj (facebo…

…okincubator#10614)

Summary:
We encountered an exception while executing Q22 on a 1TB TPC-H dataset using sort merge join. The issue arises because the SortBuffer#noMoreInput() method is invoked multiple times. By eliminating this redundant check, Q22 executes successfully.

```
Caused by: org.apache.gluten.exception.GlutenException: Exception: VeloxRuntimeError
Error Source: RUNTIME
Error Code: INVALID_STATE
Retriable: False
Expression: !noMoreInput_
Context: Operator: OrderBy[1] 1
Function: noMoreInput
File: /mnt/DP_disk3/jk/projects/gluten/ep/build-velox/build/velox_ep/velox/exec/SortBuffer.cpp
Line: 101
Stack trace:
# 0  facebook::velox::VeloxException::VeloxException(char const*, unsigned long, char const*, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, bool, facebook::velox::VeloxException::Type, std::basic_string_view<char, std::char_traits<char> >)
# 1  void facebook::velox::detail::veloxCheckFail<facebook::velox::VeloxRuntimeError, facebook::velox::detail::CompileTimeEmptyString>(facebook::velox::detail::VeloxCheckFailArgs const&, facebook::velox::detail::CompileTimeEmptyString)
# 2  facebook::velox::exec::SortBuffer::noMoreInput()
# 3  facebook::velox::exec::OrderBy::noMoreInput()
# 4  facebook::velox::exec::Driver::runInternal(std::shared_ptr<facebook::velox::exec::Driver>&, std::shared_ptr<facebook::velox::exec::BlockingState>&, std::shared_ptr<facebook::velox::RowVector>&)
# 5  facebook::velox::exec::Driver::next(std::shared_ptr<facebook::velox::exec::BlockingState>&)
# 6  facebook::velox::exec::Task::next(folly::SemiFuture<folly::Unit>*)
```

Pull Request resolved: facebookincubator#10614

Reviewed By: tanjialiang

Differential Revision: D61219673

Pulled By: xiaoxmeng

fbshipit-source-id: f854cccac3a4ee990b30a5ac5910658b55ea3a42

velox/exec/MergeJoin.cpp

@@ -646,7 +646,13 @@ RowVectorPtr MergeJoin::getOutput() {
         // No rows survived the filter. Get more rows.
         continue;
       } else if (isAntiJoin(joinType_)) {
-        return filterOutputForAntiJoin(output);
+        output = filterOutputForAntiJoin(output);
+        if (output) {
+          return output;
+        }
+
+        // No rows survived the filter for anti join. Get more rows.
+        continue;
       } else {
         return output;
       }

velox/exec/tests/MergeJoinTest.cpp

@@ -770,6 +770,38 @@ TEST_F(MergeJoinTest, antiJoinWithFilter) {
           "SELECT t0 FROM t WHERE NOT exists (select 1 from u where t0 = u0 AND t.t0 > 2 ) ");
 }
 
+TEST_F(MergeJoinTest, antiJoinFailed) {
+  auto size = 1'00;
+  auto left = makeRowVector(
+      {"t0"}, {makeFlatVector<int64_t>(size, [](auto row) { return row; })});
+
+  auto right = makeRowVector(
+      {"u0"}, {makeFlatVector<int64_t>(size, [](auto row) { return row; })});
+
+  createDuckDbTable("t", {left});
+  createDuckDbTable("u", {right});
+
+  // Anti join.
+  auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
+  auto plan =
+      PlanBuilder(planNodeIdGenerator)
+          .values(split(left, 10))
+          .orderBy({"t0"}, false)
+          .mergeJoin(
+              {"t0"},
+              {"u0"},
+              PlanBuilder(planNodeIdGenerator).values({right}).planNode(),
+              "",
+              {"t0"},
+              core::JoinType::kAnti)
+          .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .config(core::QueryConfig::kMaxOutputBatchRows, "10")
+      .assertResults(
+          "SELECT t0 FROM t WHERE NOT exists (select 1 from u where t0 = u0) ");
+}
+
 TEST_F(MergeJoinTest, antiJoinWithTwoJoinKeys) {
   auto left = makeRowVector(
       {"a", "b"},