From 9970e492dfdf2b4bee487fef4b27fee66131531f Mon Sep 17 00:00:00 2001
From: xufei <xufeixw@mail.ustc.edu.cn>
Date: Tue, 27 Feb 2024 14:12:30 +0800
Subject: [PATCH] fix anti semi join  (#8792)

close pingcap/tiflash#8791
---
 dbms/src/Flash/tests/gtest_join_executor.cpp | 62 ++++++++++++++++++++
 dbms/src/Interpreters/Join.cpp               |  3 +
 2 files changed, 65 insertions(+)

diff --git a/dbms/src/Flash/tests/gtest_join_executor.cpp b/dbms/src/Flash/tests/gtest_join_executor.cpp
index d379c5e02ac..8e7b7a1fa12 100644
--- a/dbms/src/Flash/tests/gtest_join_executor.cpp
+++ b/dbms/src/Flash/tests/gtest_join_executor.cpp
@@ -1260,6 +1260,68 @@ try
 }
 CATCH
 
+TEST_F(JoinExecutorTestRunner, Issue8791)
+try
+{
+    auto build_key = toNullableVec<Int64>(
+        "id",
+        {
+            2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3,
+            4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5,
+            6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7,
+        });
+    auto build_col = toNullableVec<Int64>(
+        "build_value",
+        {
+            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10,
+            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10,
+            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10,
+            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10,
+        });
+    auto probe_key = toNullableVec<Int64>(
+        "id",
+        {
+            1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3,
+            4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 6,
+            6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7,
+        });
+    auto probe_col = toNullableVec<Int64>(
+        "probe_value",
+        {
+            20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20,
+            20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20,
+            20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20,
+            20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20, 20,
+        });
+    context.addMockTable(
+        "issue_8791",
+        "build_table",
+        {{"id", TiDB::TP::TypeLongLong}, {"build_value", TiDB::TP::TypeLongLong}},
+        {build_key, build_col});
+    context.addMockTable(
+        "issue_8791",
+        "probe_table",
+        {{"id", TiDB::TP::TypeLongLong}, {"probe_value", TiDB::TP::TypeLongLong}},
+        {probe_key, probe_col});
+
+    auto request = context.scan("issue_8791", "probe_table")
+                       .join(
+                           context.scan("issue_8791", "build_table"),
+                           tipb::JoinType::TypeAntiSemiJoin,
+                           {col("id")},
+                           {},
+                           {},
+                           {gt(col("probe_value"), col("build_value"))},
+                           {})
+                       .aggregation({Count(col("id"))}, {})
+                       .build(context);
+
+    context.context->setSetting("max_block_size", Field(static_cast<UInt64>(200)));
+    auto expected_columns = {toVec<UInt64>({1})};
+    ASSERT_COLUMNS_EQ_UR(expected_columns, executeStreams(request, 1));
+}
+CATCH
+
 TEST_F(JoinExecutorTestRunner, CrossJoinWithoutCondition)
 try
 {
diff --git a/dbms/src/Interpreters/Join.cpp b/dbms/src/Interpreters/Join.cpp
index 89afc5f4191..acb8485b656 100644
--- a/dbms/src/Interpreters/Join.cpp
+++ b/dbms/src/Interpreters/Join.cpp
@@ -1304,6 +1304,9 @@ Block Join::doJoinBlockHash(ProbeProcessInfo & probe_process_info) const
                 offsets_to_replicate->assign(
                     offsets_to_replicate->begin() + probe_process_info.start_row,
                     offsets_to_replicate->begin() + probe_process_info.end_row);
+                if (isAntiJoin(kind) && filter != nullptr)
+                    filter->assign(filter->begin() + probe_process_info.start_row,
+                                   filter->begin() + probe_process_info.end_row);
             }
         }
     }