executor: skip probe for empty hash joins

dbms/src/DataStreams/HashJoinProbeExec.cpp

@@ -128,6 +128,9 @@ PartitionBlock HashJoinProbeExec::getProbeBlock()
 
 Block HashJoinProbeExec::probe()
 {
+    if (probe_process_info.all_rows_joined_finish && join->shouldSkipProbe())
+        return {};
+
     if (probe_process_info.all_rows_joined_finish)
     {
         auto partition_block = getProbeBlock();

dbms/src/Flash/Pipeline/Exec/PipelineExec.cpp

@@ -91,9 +91,17 @@ PipelineExec::PipelineExec(
 void PipelineExec::executePrefix()
 {
     sink_op->operatePrefix();
+    bool skip_source = false;
     for (auto it = transform_ops.rbegin(); it != transform_ops.rend(); ++it) // NOLINT(modernize-loop-convert)
+    {
         (*it)->operatePrefix();
-    source_op->operatePrefix();
+        skip_source |= (*it)->shouldSkipSource();
+    }
+    if (!skip_source)
+    {
+        source_op->operatePrefix();
+        source_prefix_executed = true;
+    }
     FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::random_pipeline_model_execute_prefix_failpoint);
 }
 
@@ -102,7 +110,8 @@ void PipelineExec::executeSuffix()
     sink_op->operateSuffix();
     for (auto it = transform_ops.rbegin(); it != transform_ops.rend(); ++it) // NOLINT(modernize-loop-convert)
         (*it)->operateSuffix();
-    source_op->operateSuffix();
+    if (source_prefix_executed)
+        source_op->operateSuffix();
     FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::random_pipeline_model_execute_suffix_failpoint);
 }
 

dbms/src/Flash/Pipeline/Exec/PipelineExec.h

@@ -80,6 +80,7 @@ class PipelineExec : private boost::noncopyable
     TransformOps transform_ops;
     SinkOpPtr sink_op;
     bool has_pipeline_breaker_wait_time = false;
+    bool source_prefix_executed = false;
 
     // hold the operator which is ready for executing await.
     Operator * awaitable = nullptr;

dbms/src/Flash/Pipeline/Exec/tests/gtest_simple_operator.cpp

@@ -51,6 +51,60 @@ class SimpleGetResultSinkOp : public SinkOp
 private:
     ResultHandler result_handler;
 };
+
+struct SourceLifecycle
+{
+    size_t prefix_count = 0;
+    size_t read_count = 0;
+    size_t suffix_count = 0;
+};
+
+class LifecycleSourceOp : public SourceOp
+{
+public:
+    LifecycleSourceOp(PipelineExecutorContext & exec_context_, const std::shared_ptr<SourceLifecycle> & lifecycle_)
+        : SourceOp(exec_context_, "")
+        , lifecycle(lifecycle_)
+    {}
+
+    String getName() const override { return "LifecycleSourceOp"; }
+
+protected:
+    void operatePrefixImpl() override { ++lifecycle->prefix_count; }
+    void operateSuffixImpl() override { ++lifecycle->suffix_count; }
+
+    OperatorStatus readImpl(Block & block) override
+    {
+        ++lifecycle->read_count;
+        block = {};
+        return OperatorStatus::HAS_OUTPUT;
+    }
+
+private:
+    std::shared_ptr<SourceLifecycle> lifecycle;
+};
+
+class SkipSourceTransformOp : public TransformOp
+{
+public:
+    explicit SkipSourceTransformOp(PipelineExecutorContext & exec_context_)
+        : TransformOp(exec_context_, "")
+    {}
+
+    String getName() const override { return "SkipSourceTransformOp"; }
+    bool shouldSkipSource() const override { return true; }
+
+protected:
+    OperatorStatus transformImpl(Block &) override { return OperatorStatus::HAS_OUTPUT; }
+
+    OperatorStatus tryOutputImpl(Block & block) override
+    {
+        block = {};
+        return OperatorStatus::HAS_OUTPUT;
+    }
+
+    void transformHeaderImpl(Block &) override {}
+};
 } // namespace
 
 class SimpleOperatorTestRunner : public DB::tests::ExecutorTest
@@ -144,6 +198,29 @@ try
 }
 CATCH
 
+TEST_F(SimpleOperatorTestRunner, SkipSourceWhenTransformCanFinish)
+try
+{
+    PipelineExecutorContext exec_context;
+    auto lifecycle = std::make_shared<SourceLifecycle>();
+    auto source = std::make_unique<LifecycleSourceOp>(exec_context, lifecycle);
+    TransformOps transforms;
+    transforms.push_back(std::make_unique<SkipSourceTransformOp>(exec_context));
+    ResultHandler result_handler{[](const Block &) {
+    }};
+    auto sink = std::make_unique<SimpleGetResultSinkOp>(exec_context, "", std::move(result_handler));
+    PipelineExec pipeline(std::move(source), std::move(transforms), std::move(sink), false);
+
+    pipeline.executePrefix();
+    ASSERT_EQ(pipeline.execute(), OperatorStatus::FINISHED);
+    pipeline.executeSuffix();
+
+    ASSERT_EQ(lifecycle->prefix_count, 0);
+    ASSERT_EQ(lifecycle->read_count, 0);
+    ASSERT_EQ(lifecycle->suffix_count, 0);
+}
+CATCH
+
 TEST_F(SimpleOperatorTestRunner, Filter)
 try
 {

dbms/src/Flash/tests/gtest_join_executor.cpp

@@ -256,6 +256,260 @@ try
 }
 CATCH
 
+TEST_F(JoinExecutorTestRunner, EmptyBuildInnerJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "empty_build_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addExchangeReceiver("empty_build_receiver", {{"a", TiDB::TP::TypeLong}});
+
+    auto request = context.scan("empty_build_join", "probe_table")
+                       .join(context.receive("empty_build_receiver"), tipb::JoinType::TypeInnerJoin, {col("a")})
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {0, 10}}, {"exchange_receiver_1", {0, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, BuildWithOnlyNullKeysInnerJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "null_key_build_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addExchangeReceiver(
+        "null_key_build_receiver",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {{}})});
+
+    auto request = context.scan("null_key_build_join", "probe_table")
+                       .join(context.receive("null_key_build_receiver"), tipb::JoinType::TypeInnerJoin, {col("a")})
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {0, 10}}, {"exchange_receiver_1", {1, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, EmptyBuildSemiJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "empty_build_semi_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addExchangeReceiver("empty_build_semi_receiver", {{"a", TiDB::TP::TypeLong}});
+
+    auto request = context.scan("empty_build_semi_join", "probe_table")
+                       .join(context.receive("empty_build_semi_receiver"), tipb::JoinType::TypeSemiJoin, {col("a")})
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {0, 10}}, {"exchange_receiver_1", {0, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, BuildWithOnlyNullKeysSemiJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "null_key_build_semi_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addExchangeReceiver(
+        "null_key_build_semi_receiver",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {{}})});
+
+    auto request = context.scan("null_key_build_semi_join", "probe_table")
+                       .join(context.receive("null_key_build_semi_receiver"), tipb::JoinType::TypeSemiJoin, {col("a")})
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {0, 10}}, {"exchange_receiver_1", {1, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, BuildRowsFilteredOutInnerAndSemiJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "filtered_build_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addMockTable(
+        "filtered_build_join",
+        "build_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+
+    /// Keep non-empty build input, but filter every build row before it reaches the join.
+    for (const auto join_type : {tipb::JoinType::TypeInnerJoin, tipb::JoinType::TypeSemiJoin})
+    {
+        auto request = context.scan("filtered_build_join", "probe_table")
+                           .join(
+                               context.scan("filtered_build_join", "build_table")
+                                   .filter(gt(col("a"), lit(Field(static_cast<Int64>(100))))),
+                               join_type,
+                               {col("a")})
+                           .build(context);
+
+        WRAP_FOR_TEST_BEGIN
+        WRAP_FOR_JOIN_TEST_BEGIN
+        if (!enable_pipeline && cfg.enable_join_v2)
+            continue;
+
+        executeAndAssertColumnsEqual(request, {});
+        Expect expect{
+            {"table_scan_0", {0, 10}},
+            {"table_scan_1", {3, 10}},
+            {"selection_2", {0, 10}},
+            {"Join_3", {0, 10}}};
+        testForExecutionSummary(request, expect);
+        WRAP_FOR_JOIN_TEST_END
+        WRAP_FOR_TEST_END
+    }
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, EmptyBuildRightSemiJoinSkipsProbe)
+try
+{
+    context.addMockTable("empty_build_right_semi_join", "build_table", {{"a", TiDB::TP::TypeLong}});
+    context.addExchangeReceiver(
+        "empty_build_right_semi_receiver",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+
+    auto request = context.scan("empty_build_right_semi_join", "build_table")
+                       .join(
+                           context.receive("empty_build_right_semi_receiver"),
+                           tipb::JoinType::TypeSemiJoin,
+                           {col("a")},
+                           {},
+                           {},
+                           {},
+                           {},
+                           0,
+                           false,
+                           0)
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {0, 10}}, {"exchange_receiver_1", {0, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, BuildWithOnlyNullKeysRightSemiJoinSkipsProbe)
+try
+{
+    context.addMockTable(
+        "null_key_build_right_semi_join",
+        "build_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {{}})});
+    context.addExchangeReceiver(
+        "null_key_build_right_semi_receiver",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+
+    auto request = context.scan("null_key_build_right_semi_join", "build_table")
+                       .join(
+                           context.receive("null_key_build_right_semi_receiver"),
+                           tipb::JoinType::TypeSemiJoin,
+                           {col("a")},
+                           {},
+                           {},
+                           {},
+                           {},
+                           0,
+                           false,
+                           0)
+                       .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {1, 10}}, {"exchange_receiver_1", {0, 10}}, {"Join_2", {0, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
+TEST_F(JoinExecutorTestRunner, EmptyBuildAntiSemiJoinStillReadsProbe)
+try
+{
+    context.addMockTable(
+        "empty_build_anti_semi_join",
+        "probe_table",
+        {{"a", TiDB::TP::TypeLong}},
+        {toNullableVec<Int32>("a", {1, 2, 3})});
+    context.addExchangeReceiver("empty_build_anti_semi_receiver", {{"a", TiDB::TP::TypeLong}});
+
+    auto request
+        = context.scan("empty_build_anti_semi_join", "probe_table")
+              .join(context.receive("empty_build_anti_semi_receiver"), tipb::JoinType::TypeAntiSemiJoin, {col("a")})
+              .build(context);
+
+    WRAP_FOR_TEST_BEGIN
+    WRAP_FOR_JOIN_TEST_BEGIN
+    if (!enable_pipeline && cfg.enable_join_v2)
+        continue;
+
+    Expect expect{{"table_scan_0", {3, 10}}, {"exchange_receiver_1", {0, 10}}, {"Join_2", {3, 10}}};
+    testForExecutionSummary(request, expect);
+    WRAP_FOR_JOIN_TEST_END
+    WRAP_FOR_TEST_END
+}
+CATCH
+
 TEST_F(JoinExecutorTestRunner, MultiJoin)
 try
 {