Support right outer/semi/anti join in hash join v2

dbms/CMakeLists.txt

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

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

@@ -214,12 +214,12 @@ bool PhysicalJoinV2::isSupported(const tipb::Join & join)
     case Anti:
     case LeftOuterSemi:
     case LeftOuterAnti:
+    case RightOuter:
+    case RightSemi:
+    case RightAnti:
         if (!tiflash_join.getBuildJoinKeys().empty())
             return true;
         break;
-    //case RightOuter:
-    //case RightSemi:
-    //case RightAnti:
     default:
     }
     return false;

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

@@ -36,6 +36,7 @@ void PhysicalJoinV2Probe::buildPipelineExecGroupImpl(
         builder.appendTransformOp(
             std::make_unique<HashJoinV2ProbeTransformOp>(exec_context, log->identifier(), join_ptr, probe_index++));
     });
+    exec_context.addOneTimeFuture(join_ptr->getWaitProbeFinishFuture());
     join_ptr.reset();
 }
 } // namespace DB

dbms/src/Flash/tests/gtest_join_executor.cpp

@@ -3443,6 +3443,8 @@ try
                     Field(static_cast<UInt64>(threshold)));
                 ASSERT_COLUMNS_EQ_UR(ref, executeStreams(request, original_max_streams))
                     << "left_table_name = " << left_table_name << ", right_table_name = " << right_table_name;
+                if (cfg.enable_join_v2)
+                    break;
             }
             WRAP_FOR_JOIN_TEST_END
         }
@@ -3483,6 +3485,8 @@ try
                         << "left_table_name = " << left_table_name
                         << ", right_exchange_receiver_concurrency = " << exchange_concurrency
                         << ", join_probe_cache_columns_threshold = " << threshold;
+                if (cfg.enable_join_v2)
+                    break;
             }
             WRAP_FOR_JOIN_TEST_END
         }
@@ -4795,13 +4799,15 @@ try
         auto request = context.scan("right_semi_family", "t")
                            .join(context.scan("right_semi_family", "s"), type, {col("a")}, {}, {}, {}, {}, 0, false, 0)
                            .build(context);
+        WRAP_FOR_JOIN_TEST_BEGIN
         executeAndAssertColumnsEqual(request, res);
         auto request_column_prune
             = context.scan("right_semi_family", "t")
                   .join(context.scan("right_semi_family", "s"), type, {col("a")}, {}, {}, {}, {}, 0, false, 0)
                   .aggregation({Count(lit(static_cast<UInt64>(1)))}, {})
                   .build(context);
         ASSERT_COLUMNS_EQ_UR(genScalarCountResults(res), executeStreams(request_column_prune, 2));
+        WRAP_FOR_JOIN_TEST_END
     }
 
     /// One join key(t.a = s.a) + other condition(t.c < s.c).

dbms/src/Interpreters/JoinV2/HashJoin.cpp

@@ -133,6 +133,7 @@ HashJoin::HashJoin(
     , log(Logger::get(join_req_id))
     , has_other_condition(non_equal_conditions.other_cond_expr != nullptr)
     , output_columns(output_columns_)
+    , wait_probe_finished_future(std::make_shared<OneTimeNotifyFuture>(NotifyType::WAIT_ON_JOIN_PROBE_FINISH))
 {
     RUNTIME_ASSERT(key_names_left.size() == key_names_right.size());
     output_block = Block(output_columns);
@@ -198,19 +199,27 @@ void HashJoin::initRowLayoutAndHashJoinMethod()
         /// Move all raw required join key column to the end of the join key.
         Names new_key_names_left, new_key_names_right;
         BoolVec raw_required_key_flag(keys_size);
+        bool is_right_semi_join = isRightSemiFamily(kind);
+        auto check_key_required = [&](const String & name) -> bool {
+            // If it's right semi/anti join, the key is required if and only if it's needed for other condition
+            if (is_right_semi_join)
+                return required_columns_names_set_for_other_condition.contains(name);
+            else
+                return right_sample_block_pruned.has(name);
+        };
         for (size_t i = 0; i < keys_size; ++i)
         {
             bool is_raw_required = false;
             if (key_columns[i].column_ptr->valuesHaveFixedSize())
             {
-                if (right_sample_block_pruned.has(key_names_right[i]))
+                if (check_key_required(key_names_right[i]))
                     is_raw_required = true;
             }
             else
             {
                 if (canAsColumnString(key_columns[i].column_ptr)
                     && getStringCollatorKind(collators) == StringCollatorKind::StringBinary
-                    && right_sample_block_pruned.has(key_names_right[i]))
+                    && check_key_required(key_names_right[i]))
                 {
                     is_raw_required = true;
                 }
@@ -261,6 +270,7 @@ void HashJoin::initRowLayoutAndHashJoinMethod()
             if (c.column->valuesHaveFixedSize())
             {
                 row_layout.other_column_fixed_size += c.column->sizeOfValueIfFixed();
+                row_layout.other_column_for_other_condition_fixed_size += c.column->sizeOfValueIfFixed();
                 row_layout.other_column_indexes.push_back({i, true});
             }
             else
@@ -319,7 +329,7 @@ void HashJoin::initBuild(const Block & sample_block, size_t build_concurrency_)
     build_workers_data.resize(build_concurrency);
     for (size_t i = 0; i < build_concurrency; ++i)
         build_workers_data[i].key_getter = createHashJoinKeyGetter(method, collators);
-    for (size_t i = 0; i < JOIN_BUILD_PARTITION_COUNT + 1; ++i)
+    for (size_t i = 0; i < JOIN_BUILD_PARTITION_COUNT; ++i)
         multi_row_containers.emplace_back(std::make_unique<MultipleRowContainer>());
 
     build_initialized = true;
@@ -415,18 +425,28 @@ void HashJoin::initProbe(const Block & sample_block, size_t probe_concurrency_)
     active_probe_worker = probe_concurrency;
     probe_workers_data.resize(probe_concurrency);
 
+    if (needScanBuildSideAfterProbe())
+        join_build_scanner_after_probe = std::make_unique<JoinBuildScannerAfterProbe>(this);
+
     probe_initialized = true;
 }
 
 bool HashJoin::finishOneBuildRow(size_t stream_index)
 {
     auto & wd = build_workers_data[stream_index];
+    if (wd.non_joined_block.rows() > 0)
+    {
+        non_joined_blocks.insertNonFullBlock(std::move(wd.non_joined_block));
+        wd.non_joined_block = {};
+    }
     LOG_DEBUG(
         log,
-        "{} insert block to row containers cost {}ms, row count {}, padding size {}({:.2f}% of all size {})",
+        "{} insert block to row containers cost {}ms, row count {}, non-joined count {}, padding size {}({:.2f}% of "
+        "all size {})",
         stream_index,
         wd.build_time,
         wd.row_count,
+        wd.non_joined_row_count,
         wd.padding_size,
         100.0 * wd.padding_size / wd.all_size,
         wd.all_size);
@@ -455,11 +475,22 @@ bool HashJoin::finishOneProbe(size_t stream_index)
     if (active_probe_worker.fetch_sub(1) == 1)
     {
         FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_mpp_hash_probe);
+        wait_probe_finished_future->finish();
         return true;
     }
     return false;
 }
 
+bool HashJoin::isAllProbeFinished()
+{
+    if (active_probe_worker > 0)
+    {
+        setNotifyFuture(wait_probe_finished_future.get());
+        return false;
+    }
+    return true;
+}
+
 void HashJoin::workAfterBuildRowFinish()
 {
     size_t all_build_row_count = 0;
@@ -478,13 +509,15 @@ void HashJoin::workAfterBuildRowFinish()
         enable_tagged_pointer,
         false);
 
-    /// Conservative threshold: trigger late materialization when lm_row_size average >= 16 bytes.
-    constexpr size_t trigger_lm_row_size_threshold = 16;
     bool late_materialization = false;
     size_t avg_lm_row_size = 0;
-    if (has_other_condition
-        && row_layout.other_column_count_for_other_condition < row_layout.other_column_indexes.size())
+    if (shouldCheckLateMaterialization())
     {
+        // Calculate the average row size of late materialization rows.
+        // If the average row size is greater than or equal to the threshold, enable late materialization.
+        // Otherwise, disable it.
+        // Note: this is a conservative threshold, enable late materialization when lm_row_size average >= 16 bytes.
+        constexpr size_t trigger_lm_row_size_threshold = 16;
         size_t total_lm_row_size = 0;
         size_t total_lm_row_count = 0;
         for (size_t i = 0; i < build_concurrency; ++i)
@@ -556,41 +589,35 @@ void HashJoin::buildRowFromBlock(const Block & b, size_t stream_index)
 
     assertBlocksHaveEqualStructure(block, right_sample_block_pruned, "Join Build");
 
-    bool check_lm_row_size = has_other_condition
-        && row_layout.other_column_count_for_other_condition < row_layout.other_column_indexes.size();
-    insertBlockToRowContainers(
-        method,
-        needRecordNotInsertRows(kind),
+    JoinBuildHelper::insertBlockToRowContainers(
+        this,
         block,
         rows,
         key_columns,
         null_map,
-        row_layout,
-        multi_row_containers,
         build_workers_data[stream_index],
-        check_lm_row_size);
+        shouldCheckLateMaterialization());
 
     build_workers_data[stream_index].build_time += watch.elapsedMilliseconds();
 }
 
 bool HashJoin::buildPointerTable(size_t stream_index)
 {
     bool is_end;
+    size_t max_build_size = 2 * settings.max_block_size;
     switch (method)
     {
-#define M(METHOD)                                                                          \
-    case HashJoinKeyMethod::METHOD:                                                        \
-        using KeyGetterType##METHOD = HashJoinKeyGetterForType<HashJoinKeyMethod::METHOD>; \
-        if constexpr (KeyGetterType##METHOD::Type::joinKeyCompareHashFirst())              \
-            is_end = pointer_table.build<KeyGetterType##METHOD::HashValueType>(            \
-                build_workers_data[stream_index],                                          \
-                multi_row_containers,                                                      \
-                settings.max_block_size);                                                  \
-        else                                                                               \
-            is_end = pointer_table.build<void>(                                            \
-                build_workers_data[stream_index],                                          \
-                multi_row_containers,                                                      \
-                settings.max_block_size);                                                  \
+#define M(METHOD)                                                                                                    \
+    case HashJoinKeyMethod::METHOD:                                                                                  \
+        using KeyGetterType##METHOD = HashJoinKeyGetterForType<HashJoinKeyMethod::METHOD>;                           \
+        if constexpr (KeyGetterType##METHOD::Type::joinKeyCompareHashFirst())                                        \
+            is_end = pointer_table.build<KeyGetterType##METHOD::HashValueType>(                                      \
+                build_workers_data[stream_index],                                                                    \
+                multi_row_containers,                                                                                \
+                max_build_size);                                                                                     \
+        else                                                                                                         \
+            is_end                                                                                                   \
+                = pointer_table.build<void>(build_workers_data[stream_index], multi_row_containers, max_build_size); \
         break;
         APPLY_FOR_HASH_JOIN_VARIANTS(M)
 #undef M
@@ -665,6 +692,19 @@ Block HashJoin::probeLastResultBlock(size_t stream_index)
     return {};
 }
 
+bool HashJoin::needScanBuildSideAfterProbe() const
+{
+    return isRightOuterJoin(kind) || isRightSemiFamily(kind);
+}
+
+Block HashJoin::scanBuildSideAfterProbe(size_t stream_index)
+{
+    auto & wd = probe_workers_data[stream_index];
+    Stopwatch all_watch;
+    SCOPE_EXIT({ probe_workers_data[stream_index].scan_build_side_time += all_watch.elapsedFromLastTime(); });
+    return join_build_scanner_after_probe->scan(wd);
+}
+
 void HashJoin::removeUselessColumn(Block & block) const
 {
     const NameSet & probe_output_name_set = has_other_condition

dbms/src/Interpreters/JoinV2/HashJoin.h

@@ -19,8 +19,11 @@
 #include <Core/Block.h>
 #include <Flash/Coprocessor/DAGContext.h>
 #include <Flash/Coprocessor/JoinInterpreterHelper.h>
+#include <Flash/Pipeline/Schedule/Tasks/OneTimeNotifyFuture.h>
 #include <Interpreters/ExpressionActions.h>
+#include <Interpreters/JoinUtils.h>
 #include <Interpreters/JoinV2/HashJoinBuild.h>
+#include <Interpreters/JoinV2/HashJoinBuildScannerAfterProbe.h>
 #include <Interpreters/JoinV2/HashJoinKey.h>
 #include <Interpreters/JoinV2/HashJoinPointerTable.h>
 #include <Interpreters/JoinV2/HashJoinProbe.h>
@@ -54,13 +57,18 @@ class HashJoin
     bool finishOneBuildRow(size_t stream_index);
     /// Return true if it is the last probe worker.
     bool finishOneProbe(size_t stream_index);
+    /// Return true if all probe work has finished.
+    bool isAllProbeFinished();
 
     void buildRowFromBlock(const Block & block, size_t stream_index);
     bool buildPointerTable(size_t stream_index);
 
     Block probeBlock(JoinProbeContext & ctx, size_t stream_index);
     Block probeLastResultBlock(size_t stream_index);
 
+    bool needScanBuildSideAfterProbe() const;
+    Block scanBuildSideAfterProbe(size_t stream_index);
+
     void removeUselessColumn(Block & block) const;
     /// Block's schema must be all_sample_block_pruned.
     Block removeUselessColumnForOutput(const Block & block) const;
@@ -76,14 +84,25 @@ class HashJoin
 
     const JoinProfileInfoPtr & getProfileInfo() const { return profile_info; }
 
+    const OneTimeNotifyFuturePtr & getWaitProbeFinishFuture() { return wait_probe_finished_future; }
+
 private:
     void initRowLayoutAndHashJoinMethod();
 
     void workAfterBuildRowFinish();
 
+    bool shouldCheckLateMaterialization() const
+    {
+        bool is_any_semi_join = isSemiFamily(kind) || isLeftOuterSemiFamily(kind) || isRightSemiFamily(kind);
+        return has_other_condition && !is_any_semi_join
+            && row_layout.other_column_count_for_other_condition < row_layout.other_column_indexes.size();
+    }
+
 private:
+    friend JoinBuildHelper;
     friend JoinProbeHelper;
     friend SemiJoinProbeHelper;
+    friend JoinBuildScannerAfterProbe;
 
     static const DataTypePtr match_helper_type;
 
@@ -140,6 +159,8 @@ class HashJoin
 
     /// Row containers
     std::vector<std::unique_ptr<MultipleRowContainer>> multi_row_containers;
+    /// Non-joined blocks
+    NonJoinedBlocks non_joined_blocks;
 
     /// Build row phase
     size_t build_concurrency = 0;
@@ -152,8 +173,11 @@ class HashJoin
     size_t probe_concurrency = 0;
     std::vector<JoinProbeWorkerData> probe_workers_data;
     std::atomic<size_t> active_probe_worker = 0;
+    OneTimeNotifyFuturePtr wait_probe_finished_future;
     std::unique_ptr<JoinProbeHelper> join_probe_helper;
     std::unique_ptr<SemiJoinProbeHelper> semi_join_probe_helper;
+    /// Probe scan build side
+    std::unique_ptr<JoinBuildScannerAfterProbe> join_build_scanner_after_probe;
 
     const JoinProfileInfoPtr profile_info = std::make_shared<JoinProfileInfo>();