Support (left outer) (anti) semi join in hash join v2 (#10133)

ref #9060

Support (left outer) (anti) semi join in hash join v2

Signed-off-by: gengliqi <gengliqiii@gmail.com>

Author: gengliqi

dbms/CMakeLists.txt

@@ -113,6 +113,7 @@ check_then_add_sources_compile_flag (
     src/Interpreters/JoinV2/HashJoin.cpp
     src/Interpreters/JoinV2/HashJoinBuild.cpp
     src/Interpreters/JoinV2/HashJoinProbe.cpp
+    src/Interpreters/JoinV2/SemiJoinProbe.cpp
     src/IO/Compression/EncodingUtil.cpp
     src/Storages/DeltaMerge/BitmapFilter/BitmapFilter.cpp
     src/Storages/DeltaMerge/DMVersionFilterBlockInputStream.cpp

dbms/src/Flash/Planner/Plans/PhysicalJoinV2.cpp

@@ -210,15 +210,11 @@ bool PhysicalJoinV2::isSupported(const tipb::Join & join)
     {
     case Inner:
     case LeftOuter:
-        if (!tiflash_join.getBuildJoinKeys().empty())
-            return true;
-        break;
     case Semi:
     case Anti:
     case LeftOuterSemi:
     case LeftOuterAnti:
-        if (!tiflash_join.getBuildJoinKeys().empty() && join.other_conditions_size() == 0
-            && join.other_eq_conditions_from_in_size() == 0)
+        if (!tiflash_join.getBuildJoinKeys().empty())
             return true;
         break;
     //case RightOuter:

dbms/src/Interpreters/JoinV2/HashJoin.cpp

@@ -498,7 +498,10 @@ void HashJoin::workAfterBuildRowFinish()
     fiu_do_on(FailPoints::force_join_v2_probe_enable_lm, { late_materialization = true; });
     fiu_do_on(FailPoints::force_join_v2_probe_disable_lm, { late_materialization = false; });
 
-    join_probe_helper = std::make_unique<JoinProbeHelper>(this, late_materialization);
+    if (SemiJoinProbeHelper::isSupported(kind, has_other_condition))
+        semi_join_probe_helper = std::make_unique<SemiJoinProbeHelper>(this);
+    else
+        join_probe_helper = std::make_unique<JoinProbeHelper>(this, late_materialization);
 
     LOG_INFO(
         log,
@@ -626,6 +629,7 @@ Block HashJoin::probeBlock(JoinProbeContext & ctx, size_t stream_index)
         method,
         kind,
         has_other_condition,
+        !non_equal_conditions.other_eq_cond_from_in_name.empty(),
         key_names_left,
         non_equal_conditions.left_filter_column,
         probe_output_name_set,
@@ -636,7 +640,11 @@ Block HashJoin::probeBlock(JoinProbeContext & ctx, size_t stream_index)
     FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::random_join_prob_failpoint);
 
     auto & wd = probe_workers_data[stream_index];
-    Block res = join_probe_helper->probe(ctx, wd);
+    Block res;
+    if (semi_join_probe_helper)
+        res = semi_join_probe_helper->probe(ctx, wd);
+    else
+        res = join_probe_helper->probe(ctx, wd);
     if (ctx.isAllFinished())
         wd.probe_handle_rows += ctx.rows;
     return res;

dbms/src/Interpreters/JoinV2/HashJoin.h

@@ -26,6 +26,7 @@
 #include <Interpreters/JoinV2/HashJoinProbe.h>
 #include <Interpreters/JoinV2/HashJoinRowLayout.h>
 #include <Interpreters/JoinV2/HashJoinSettings.h>
+#include <Interpreters/JoinV2/SemiJoinProbe.h>
 
 
 namespace DB
@@ -82,6 +83,7 @@ class HashJoin
 
 private:
     friend JoinProbeHelper;
+    friend SemiJoinProbeHelper;
 
     static const DataTypePtr match_helper_type;
 
@@ -151,6 +153,7 @@ class HashJoin
     std::vector<JoinProbeWorkerData> probe_workers_data;
     std::atomic<size_t> active_probe_worker = 0;
     std::unique_ptr<JoinProbeHelper> join_probe_helper;
+    std::unique_ptr<SemiJoinProbeHelper> semi_join_probe_helper;
 
     const JoinProfileInfoPtr profile_info = std::make_shared<JoinProfileInfo>();
 

dbms/src/Interpreters/JoinV2/HashJoinKey.cpp

@@ -79,8 +79,7 @@ std::unique_ptr<void, std::function<void(void *)>> createHashJoinKeyGetter(
         using KeyGetterType##METHOD = typename HashJoinKeyGetterForType<HashJoinKeyMethod::METHOD>::Type; \
         return std::unique_ptr<void, std::function<void(void *)>>(                                        \
             static_cast<void *>(new KeyGetterType##METHOD(collators)),                                    \
-            [](void * ptr) { delete reinterpret_cast<KeyGetterType##METHOD *>(ptr); });                   \
-        break;
+            [](void * ptr) { delete reinterpret_cast<KeyGetterType##METHOD *>(ptr); });
         APPLY_FOR_HASH_JOIN_VARIANTS(M)
 #undef M
 

dbms/src/Interpreters/JoinV2/HashJoinProbe.cpp

@@ -21,6 +21,7 @@
 #include <Interpreters/JoinUtils.h>
 #include <Interpreters/JoinV2/HashJoin.h>
 #include <Interpreters/JoinV2/HashJoinProbe.h>
+#include <Interpreters/JoinV2/SemiJoinProbe.h>
 #include <Interpreters/NullableUtils.h>
 #include <Parsers/ASTTablesInSelectQuery.h>
 
@@ -39,7 +40,9 @@ bool JoinProbeContext::isProbeFinished() const
 {
     return current_row_idx >= rows
         // For prefetching
-        && prefetch_active_states == 0;
+        && prefetch_active_states == 0
+        // For (left outer) (anti) semi join with other conditions
+        && (semi_join_probe_list == nullptr || semi_join_probe_list->activeSlots() == 0);
 }
 
 bool JoinProbeContext::isAllFinished() const
@@ -71,6 +74,7 @@ void JoinProbeContext::prepareForHashProbe(
     HashJoinKeyMethod method,
     ASTTableJoin::Kind kind,
     bool has_other_condition,
+    bool has_other_eq_cond_from_in,
     const Names & key_names,
     const String & filter_column,
     const NameSet & probe_output_name_set,
@@ -123,6 +127,23 @@ void JoinProbeContext::prepareForHashProbe(
     {
         left_semi_match_res.clear();
         left_semi_match_res.resize_fill_zero(rows);
+        if (has_other_eq_cond_from_in)
+        {
+            left_semi_match_null_res.clear();
+            left_semi_match_null_res.resize_fill_zero(rows);
+        }
+    }
+    if ((kind == Semi || kind == Anti) && has_other_condition)
+    {
+        semi_selective_offsets.clear();
+        semi_selective_offsets.reserve(rows);
+    }
+
+    if (SemiJoinProbeHelper::isSupported(kind, has_other_condition))
+    {
+        if unlikely (!semi_join_probe_list)
+            semi_join_probe_list = createSemiJoinProbeList(method);
+        semi_join_probe_list->reset(rows);
     }
 
     is_prepared = true;
@@ -634,16 +655,17 @@ void JoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDa
     using Adder = JoinProbeAdder<kind, has_other_condition, late_materialization>;
 
     auto & key_getter = *static_cast<KeyGetterType *>(ctx.key_getter.get());
+    // Some columns in wd.result_block may remain empty due to late materialization for join with other conditions.
+    // But since all columns are cleared after handling other conditions, wd.result_block.rows() is always 0.
     size_t current_offset = wd.result_block.rows();
+    if constexpr (has_other_condition)
+        RUNTIME_CHECK(current_offset == 0);
     size_t idx = ctx.current_row_idx;
     RowPtr ptr = ctx.current_build_row_ptr;
     bool is_matched = ctx.current_row_is_matched;
     size_t collision = 0;
-    size_t key_offset = sizeof(RowPtr);
-    if constexpr (KeyGetterType::joinKeyCompareHashFirst())
-    {
-        key_offset += sizeof(HashValueType);
-    }
+    constexpr size_t key_offset
+        = sizeof(RowPtr) + (KeyGetterType::joinKeyCompareHashFirst() ? sizeof(HashValueType) : 0);
 
 #define NOT_MATCHED(not_matched)                                                     \
     if constexpr (Adder::need_not_matched)                                           \
@@ -782,8 +804,10 @@ struct ProbePrefetchState
     KeyType key{};
     union
     {
-        RowPtr ptr = nullptr;
-        std::atomic<RowPtr> * pointer_ptr;
+        /// Used when stage is FindHeader
+        std::atomic<RowPtr> * pointer_ptr = nullptr;
+        /// Used when stage is FindNext
+        RowPtr ptr;
     };
 };
 
@@ -801,24 +825,26 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
     using Adder = JoinProbeAdder<kind, has_other_condition, late_materialization>;
 
     auto & key_getter = *static_cast<KeyGetterType *>(ctx.key_getter.get());
-    if (!ctx.prefetch_states)
+    const size_t probe_prefetch_step = settings.probe_prefetch_step;
+    if unlikely (!ctx.prefetch_states)
     {
         ctx.prefetch_states = decltype(ctx.prefetch_states)(
-            static_cast<void *>(new ProbePrefetchState<KeyGetter>[settings.probe_prefetch_step]),
+            static_cast<void *>(new ProbePrefetchState<KeyGetter>[probe_prefetch_step]),
             [](void * ptr) { delete[] static_cast<ProbePrefetchState<KeyGetter> *>(ptr); });
     }
     auto * states = static_cast<ProbePrefetchState<KeyGetter> *>(ctx.prefetch_states.get());
 
     size_t idx = ctx.current_row_idx;
     size_t active_states = ctx.prefetch_active_states;
     size_t k = ctx.prefetch_iter;
+    // Some columns in wd.result_block may remain empty due to late materialization for join with other conditions.
+    // But since all columns are cleared after handling other conditions, wd.result_block.rows() is always 0.
     size_t current_offset = wd.result_block.rows();
+    if constexpr (has_other_condition)
+        RUNTIME_CHECK(current_offset == 0);
     size_t collision = 0;
-    size_t key_offset = sizeof(RowPtr);
-    if constexpr (KeyGetterType::joinKeyCompareHashFirst())
-    {
-        key_offset += sizeof(HashValueType);
-    }
+    constexpr size_t key_offset
+        = sizeof(RowPtr) + (KeyGetterType::joinKeyCompareHashFirst() ? sizeof(HashValueType) : 0);
 
 #define NOT_MATCHED(not_matched, idx)                                                \
     if constexpr (Adder::need_not_matched)                                           \
@@ -831,7 +857,6 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
         }                                                                            \
     }
 
-    const size_t probe_prefetch_step = settings.probe_prefetch_step;
     while (idx < ctx.rows || active_states > 0)
     {
         k = k == probe_prefetch_step ? 0 : k;
@@ -845,11 +870,10 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
             const auto & key2 = key_getter.deserializeJoinKey(ptr + key_offset);
             bool key_is_equal = joinKeyIsEqual(key_getter, state->key, key2, state->hash, ptr);
             collision += !key_is_equal;
+            if constexpr (Adder::need_not_matched)
+                state->is_matched |= key_is_equal;
             if (key_is_equal)
             {
-                if constexpr (Adder::need_not_matched)
-                    state->is_matched = true;
-
                 if constexpr (Adder::need_matched)
                 {
                     bool is_end = Adder::addMatched(
@@ -1026,6 +1050,17 @@ Block JoinProbeHelper::handleOtherConditions(
 
     non_equal_conditions.other_cond_expr->execute(exec_block);
 
+    SCOPE_EXIT({
+        RUNTIME_CHECK(wd.result_block.columns() == left_columns + right_columns);
+        /// Clear the data in result_block.
+        for (size_t i = 0; i < left_columns + right_columns; ++i)
+        {
+            auto column = wd.result_block.getByPosition(i).column->assumeMutable();
+            column->popBack(column->size());
+            wd.result_block.getByPosition(i).column = std::move(column);
+        }
+    });
+
     size_t rows = exec_block.rows();
     // Ensure BASE_OFFSETS is accessed within bound.
     // It must be true because max_block_size <= BASE_OFFSETS.size(HASH_JOIN_MAX_BLOCK_SIZE_UPPER_BOUND).
@@ -1212,17 +1247,6 @@ Block JoinProbeHelper::handleOtherConditions(
         }
     };
 
-    SCOPE_EXIT({
-        RUNTIME_CHECK(wd.result_block.columns() == left_columns + right_columns);
-        /// Clear the data in result_block.
-        for (size_t i = 0; i < left_columns + right_columns; ++i)
-        {
-            auto column = wd.result_block.getByPosition(i).column->assumeMutable();
-            column->popBack(column->size());
-            wd.result_block.getByPosition(i).column = std::move(column);
-        }
-    });
-
     size_t length = std::min(result_size, remaining_insert_size);
     fill_matched(0, length);
     if (result_size >= remaining_insert_size)

dbms/src/Interpreters/JoinV2/HashJoinProbe.h

@@ -23,6 +23,7 @@
 #include <Interpreters/JoinV2/HashJoinKey.h>
 #include <Interpreters/JoinV2/HashJoinPointerTable.h>
 #include <Interpreters/JoinV2/HashJoinSettings.h>
+#include <Interpreters/JoinV2/SemiJoinProbeList.h>
 #include <Parsers/ASTTablesInSelectQuery.h>
 #include <absl/base/optimization.h>
 
@@ -40,13 +41,19 @@ struct JoinProbeContext
     RowPtr current_build_row_ptr = nullptr;
     /// For left outer/(left outer) (anti) semi join without other conditions.
     bool current_row_is_matched = false;
-    /// For left outer/(left outer) (anti) semi join with other conditions.
+    /// For left outer with other conditions.
     IColumn::Filter rows_not_matched;
     /// < 0 means not_matched_offsets is not initialized.
     ssize_t not_matched_offsets_idx = -1;
     IColumn::Offsets not_matched_offsets;
     /// For left outer (anti) semi join.
     PaddedPODArray<Int8> left_semi_match_res;
+    /// For left outer (anti) semi join with other-eq-from-in conditions.
+    PaddedPODArray<UInt8> left_semi_match_null_res;
+    /// For (anti) semi join with other conditions.
+    IColumn::Offsets semi_selective_offsets;
+    /// For (left outer) (anti) semi join with other conditions.
+    std::unique_ptr<ISemiJoinProbeList> semi_join_probe_list;
 
     size_t prefetch_active_states = 0;
     size_t prefetch_iter = 0;
@@ -69,6 +76,7 @@ struct JoinProbeContext
         HashJoinKeyMethod method,
         ASTTableJoin::Kind kind,
         bool has_other_condition,
+        bool has_other_eq_cond_from_in,
         const Names & key_names,
         const String & filter_column,
         const NameSet & probe_output_name_set,