From eb60661748a506d3fe047df42877a8d2aec23e5d Mon Sep 17 00:00:00 2001 From: ChangRui-Ryan Date: Mon, 27 Jul 2026 11:39:37 +0800 Subject: [PATCH 1/3] executor: skip probe for empty hash joins --- dbms/src/DataStreams/HashJoinProbeExec.cpp | 3 + dbms/src/Flash/Pipeline/Exec/PipelineExec.cpp | 13 +- dbms/src/Flash/Pipeline/Exec/PipelineExec.h | 1 + .../Exec/tests/gtest_simple_operator.cpp | 77 +++++++ dbms/src/Flash/tests/gtest_join_executor.cpp | 211 ++++++++++++++++++ dbms/src/Interpreters/Join.cpp | 9 + dbms/src/Interpreters/Join.h | 11 + dbms/src/Interpreters/JoinV2/HashJoin.cpp | 8 + dbms/src/Interpreters/JoinV2/HashJoin.h | 9 + .../Operators/HashJoinProbeTransformOp.cpp | 9 +- dbms/src/Operators/HashJoinProbeTransformOp.h | 2 + .../Operators/HashJoinV2ProbeTransformOp.cpp | 7 + .../Operators/HashJoinV2ProbeTransformOp.h | 2 + dbms/src/Operators/HashProbeTransformExec.h | 1 + dbms/src/Operators/Operator.h | 5 + 15 files changed, 363 insertions(+), 5 deletions(-) diff --git a/dbms/src/DataStreams/HashJoinProbeExec.cpp b/dbms/src/DataStreams/HashJoinProbeExec.cpp index ffd7be1a396..151ad6f0d64 100644 --- a/dbms/src/DataStreams/HashJoinProbeExec.cpp +++ b/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(); diff --git a/dbms/src/Flash/Pipeline/Exec/PipelineExec.cpp b/dbms/src/Flash/Pipeline/Exec/PipelineExec.cpp index d2886c268c2..879c043760a 100644 --- a/dbms/src/Flash/Pipeline/Exec/PipelineExec.cpp +++ b/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); } diff --git a/dbms/src/Flash/Pipeline/Exec/PipelineExec.h b/dbms/src/Flash/Pipeline/Exec/PipelineExec.h index c14dc437406..b2d01a96c84 100644 --- a/dbms/src/Flash/Pipeline/Exec/PipelineExec.h +++ b/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; diff --git a/dbms/src/Flash/Pipeline/Exec/tests/gtest_simple_operator.cpp b/dbms/src/Flash/Pipeline/Exec/tests/gtest_simple_operator.cpp index cf738535f3b..8f1ffd9488e 100644 --- a/dbms/src/Flash/Pipeline/Exec/tests/gtest_simple_operator.cpp +++ b/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 & 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 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(); + auto source = std::make_unique(exec_context, lifecycle); + TransformOps transforms; + transforms.push_back(std::make_unique(exec_context)); + ResultHandler result_handler{[](const Block &) { + }}; + auto sink = std::make_unique(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 { diff --git a/dbms/src/Flash/tests/gtest_join_executor.cpp b/dbms/src/Flash/tests/gtest_join_executor.cpp index 3f5ddcc5143..85e1e351292 100644 --- a/dbms/src/Flash/tests/gtest_join_executor.cpp +++ b/dbms/src/Flash/tests/gtest_join_executor.cpp @@ -256,6 +256,217 @@ try } CATCH +TEST_F(JoinExecutorTestRunner, EmptyBuildInnerJoinSkipsProbe) +try +{ + context.addMockTable( + "empty_build_join", + "probe_table", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("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("a", {1, 2, 3})}); + context.addExchangeReceiver( + "null_key_build_receiver", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("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("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("a", {1, 2, 3})}); + context.addExchangeReceiver( + "null_key_build_semi_receiver", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("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, 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("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("a", {{}})}); + context.addExchangeReceiver( + "null_key_build_right_semi_receiver", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("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("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 { diff --git a/dbms/src/Interpreters/Join.cpp b/dbms/src/Interpreters/Join.cpp index 35c0ba8db93..4d36380d581 100644 --- a/dbms/src/Interpreters/Join.cpp +++ b/dbms/src/Interpreters/Join.cpp @@ -1802,6 +1802,8 @@ void Join::workAfterBuildFinish(size_t stream_index) has_build_data_in_memory = !original_blocks.empty(); } + + build_side_empty.store(!isSpilled() && getTotalRowCount() == 0, std::memory_order_release); } void Join::finalizeNullAwareSemiFamilyBuild() @@ -1984,6 +1986,13 @@ Block Join::joinBlock(ProbeProcessInfo & probe_process_info) const LOG_WARNING(log, "JoinBlock without non zero active_build_threads, return empty block"); return {}; } + + if unlikely (shouldSkipProbe()) + { + probe_process_info.updateEndRow(probe_process_info.block.rows()); + return output_block_after_finalize; + } + std::shared_lock lock(rwlock); Block block{}; diff --git a/dbms/src/Interpreters/Join.h b/dbms/src/Interpreters/Join.h index 7029e2b0631..124f1b07e5b 100644 --- a/dbms/src/Interpreters/Join.h +++ b/dbms/src/Interpreters/Join.h @@ -247,6 +247,16 @@ class Join ASTTableJoin::Kind getKind() const { return kind; } + /// Inner/Semi cannot produce rows without build entries. RightSemi has no matched build rows to output. + /// This is available after finalizeBuild and can be used to avoid reading the probe side. + bool shouldSkipProbe() const + { + const bool can_skip_probe = kind == ASTTableJoin::Kind::Inner || kind == ASTTableJoin::Kind::Semi + || kind == ASTTableJoin::Kind::RightSemi; + return can_skip_probe && build_finished.load(std::memory_order_acquire) + && build_side_empty.load(std::memory_order_acquire); + } + const Names & getLeftJoinKeys() const { return key_names_left; } void setInitActiveBuildThreads() @@ -436,6 +446,7 @@ class Join const LoggerPtr log; std::atomic total_input_build_rows{0}; + std::atomic_bool build_side_empty{false}; /** Protect state for concurrent use in insertFromBlock and joinBlock. * Note that these methods could be called simultaneously only while use of StorageJoin, diff --git a/dbms/src/Interpreters/JoinV2/HashJoin.cpp b/dbms/src/Interpreters/JoinV2/HashJoin.cpp index 305be3a2c99..226dba634f7 100644 --- a/dbms/src/Interpreters/JoinV2/HashJoin.cpp +++ b/dbms/src/Interpreters/JoinV2/HashJoin.cpp @@ -465,6 +465,7 @@ void HashJoin::workAfterBuildRowFinish() size_t all_build_row_count = 0; for (size_t i = 0; i < build_concurrency; ++i) all_build_row_count += build_workers_data[i].row_count; + build_side_empty.store(all_build_row_count == 0, std::memory_order_release); bool enable_tagged_pointer = settings.enable_tagged_pointer; for (size_t i = 0; i < build_concurrency; ++i) @@ -622,6 +623,13 @@ Block HashJoin::probeBlock(JoinProbeContext & ctx, size_t stream_index) Stopwatch all_watch; SCOPE_EXIT({ probe_workers_data[stream_index].probe_time += all_watch.elapsedFromLastTime(); }); + if unlikely (shouldSkipProbe()) + { + ctx.current_row_idx = ctx.rows; + probe_workers_data[stream_index].probe_handle_rows += ctx.rows; + return output_block_after_finalize; + } + const NameSet & probe_output_name_set = has_other_condition ? output_columns_names_set_for_other_condition_after_finalize : output_column_names_set_after_finalize; diff --git a/dbms/src/Interpreters/JoinV2/HashJoin.h b/dbms/src/Interpreters/JoinV2/HashJoin.h index e28292e6a61..082ad602b8b 100644 --- a/dbms/src/Interpreters/JoinV2/HashJoin.h +++ b/dbms/src/Interpreters/JoinV2/HashJoin.h @@ -74,6 +74,14 @@ class HashJoin size_t getBuildConcurrency() const { return build_concurrency; } size_t getProbeConcurrency() const { return probe_concurrency; } + /// Inner/Semi cannot produce rows if no build row is inserted into the hash table. + /// The probe pipeline is scheduled after the build pointer-table event, so build_side_empty is already published. + bool shouldSkipProbe() const + { + const bool can_skip_probe = kind == ASTTableJoin::Kind::Inner || kind == ASTTableJoin::Kind::Semi; + return can_skip_probe && build_side_empty.load(std::memory_order_acquire); + } + const JoinProfileInfoPtr & getProfileInfo() const { return profile_info; } private: @@ -145,6 +153,7 @@ class HashJoin size_t build_concurrency = 0; std::vector build_workers_data; std::atomic active_build_worker = 0; + std::atomic_bool build_side_empty{false}; HashJoinPointerTable pointer_table; diff --git a/dbms/src/Operators/HashJoinProbeTransformOp.cpp b/dbms/src/Operators/HashJoinProbeTransformOp.cpp index 21b37d39469..d6a2c629013 100644 --- a/dbms/src/Operators/HashJoinProbeTransformOp.cpp +++ b/dbms/src/Operators/HashJoinProbeTransformOp.cpp @@ -163,9 +163,12 @@ OperatorStatus HashJoinProbeTransformOp::tryOutputImpl(Block & block) { if (status == ProbeStatus::PROBE && probe_process_info.all_rows_joined_finish) { - if (auto ret = probe_transform->tryFillProcessInfoInProbeStage(probe_process_info); - ret != OperatorStatus::HAS_OUTPUT) - return ret; + if (!probe_transform->shouldSkipProbe()) + { + if (auto ret = probe_transform->tryFillProcessInfoInProbeStage(probe_process_info); + ret != OperatorStatus::HAS_OUTPUT) + return ret; + } } return onOutput(block); diff --git a/dbms/src/Operators/HashJoinProbeTransformOp.h b/dbms/src/Operators/HashJoinProbeTransformOp.h index db9c2bfcc4a..705d4a72081 100644 --- a/dbms/src/Operators/HashJoinProbeTransformOp.h +++ b/dbms/src/Operators/HashJoinProbeTransformOp.h @@ -33,6 +33,8 @@ class HashJoinProbeTransformOp : public TransformOp String getName() const override { return "HashJoinProbeTransformOp"; } + bool shouldSkipSource() const override { return origin_join->shouldSkipProbe(); } + protected: OperatorStatus transformImpl(Block & block) override; diff --git a/dbms/src/Operators/HashJoinV2ProbeTransformOp.cpp b/dbms/src/Operators/HashJoinV2ProbeTransformOp.cpp index efd8ce06bbc..57a65f33652 100644 --- a/dbms/src/Operators/HashJoinV2ProbeTransformOp.cpp +++ b/dbms/src/Operators/HashJoinV2ProbeTransformOp.cpp @@ -80,6 +80,13 @@ OperatorStatus HashJoinV2ProbeTransformOp::tryOutputImpl(Block & block) block = {}; return OperatorStatus::HAS_OUTPUT; } + if unlikely (probe_context.isAllFinished() && join_ptr->shouldSkipProbe()) + { + join_ptr->finishOneProbe(op_index); + probe_context.input_is_finished = true; + block = join_ptr->probeLastResultBlock(op_index); + return OperatorStatus::HAS_OUTPUT; + } if (probe_context.isAllFinished()) return OperatorStatus::NEED_INPUT; return onOutput(block); diff --git a/dbms/src/Operators/HashJoinV2ProbeTransformOp.h b/dbms/src/Operators/HashJoinV2ProbeTransformOp.h index b78872a6ae6..1889458aabd 100644 --- a/dbms/src/Operators/HashJoinV2ProbeTransformOp.h +++ b/dbms/src/Operators/HashJoinV2ProbeTransformOp.h @@ -31,6 +31,8 @@ class HashJoinV2ProbeTransformOp : public TransformOp String getName() const override { return "HashJoinV2ProbeTransformOp"; } + bool shouldSkipSource() const override { return join_ptr->shouldSkipProbe(); } + protected: OperatorStatus transformImpl(Block & block) override; diff --git a/dbms/src/Operators/HashProbeTransformExec.h b/dbms/src/Operators/HashProbeTransformExec.h index b4d04592477..3ddc0dfe834 100644 --- a/dbms/src/Operators/HashProbeTransformExec.h +++ b/dbms/src/Operators/HashProbeTransformExec.h @@ -68,6 +68,7 @@ class HashProbeTransformExec : public std::enable_shared_from_thisdispatchProbeBlock(block, partition_blocks_list, op_index); } bool finishOneProbe() { return join->finishOneProbe(op_index); } + bool shouldSkipProbe() const { return join->shouldSkipProbe(); } bool hasMarkedSpillData() const { return join->hasProbeSideMarkedSpillData(op_index); } bool isProbeFinishedForPipeline() const { return join->isProbeFinishedForPipeline(); } void finalizeProbe() { join->finalizeProbe(); } diff --git a/dbms/src/Operators/Operator.h b/dbms/src/Operators/Operator.h index 5312898848d..312215ac6d9 100644 --- a/dbms/src/Operators/Operator.h +++ b/dbms/src/Operators/Operator.h @@ -148,6 +148,11 @@ class TransformOp : public Operator OperatorStatus transform(Block & block); virtual OperatorStatus transformImpl(Block & block) = 0; + /// Return true only when this transform can finish the pipeline without source input. + /// PipelineExec uses it to avoid starting the source operator. The transform must be able to emit an EOF block + /// from tryOutputImpl so that downstream transforms and the sink can finish normally. + virtual bool shouldSkipSource() const { return false; } + virtual void transformHeaderImpl(Block & header_) = 0; void transformHeader(Block & header_) { From 072ff9c9a3a181a9e587d3a78a66fec1fe668be1 Mon Sep 17 00:00:00 2001 From: ChangRui-Ryan Date: Mon, 10 Aug 2026 11:52:56 +0800 Subject: [PATCH 2/3] test: cover empty hash join filtering and spill --- dbms/src/Flash/tests/gtest_join_executor.cpp | 43 ++++++++++++++++++++ dbms/src/Flash/tests/gtest_spill_join.cpp | 34 ++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/dbms/src/Flash/tests/gtest_join_executor.cpp b/dbms/src/Flash/tests/gtest_join_executor.cpp index 85e1e351292..ac6d87cad23 100644 --- a/dbms/src/Flash/tests/gtest_join_executor.cpp +++ b/dbms/src/Flash/tests/gtest_join_executor.cpp @@ -366,6 +366,49 @@ try } CATCH +TEST_F(JoinExecutorTestRunner, BuildRowsFilteredOutInnerAndSemiJoinSkipsProbe) +try +{ + context.addMockTable( + "filtered_build_join", + "probe_table", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("a", {1, 2, 3})}); + context.addMockTable( + "filtered_build_join", + "build_table", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("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(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 { diff --git a/dbms/src/Flash/tests/gtest_spill_join.cpp b/dbms/src/Flash/tests/gtest_spill_join.cpp index 99058189bf3..62cfca417ce 100644 --- a/dbms/src/Flash/tests/gtest_spill_join.cpp +++ b/dbms/src/Flash/tests/gtest_spill_join.cpp @@ -238,6 +238,40 @@ try } CATCH +TEST_F(SpillJoinTestRunner, EmptyHashTableAfterSpillStillReadsProbe) +try +{ + constexpr size_t build_rows = 8192; + /// All build keys are NULL, so the hash table has no entries. The payload makes the build side spill. + std::vector> null_keys(build_rows); + std::vector> payload(build_rows, String(128, 'x')); + + context.addMockTable( + "empty_hash_after_spill", + "probe_table", + {{"a", TiDB::TP::TypeLong}}, + {toNullableVec("a", {1, 2, 3})}); + context.addMockTable( + "empty_hash_after_spill", + "build_table", + {{"a", TiDB::TP::TypeLong}, {"payload", TiDB::TP::TypeString}}, + {toNullableVec("a", null_keys), toNullableVec("payload", payload)}); + + auto request + = context.scan("empty_hash_after_spill", "probe_table") + .join(context.scan("empty_hash_after_spill", "build_table"), tipb::JoinType::TypeInnerJoin, {col("a")}) + .build(context); + ColumnsWithTypeAndName empty_result; + + context.context->getSettingsRef().enable_hash_join_v2 = false; + context.context->setSetting("max_bytes_before_external_join", Field(static_cast(10000))); + + WRAP_FOR_SPILL_TEST_BEGIN + ASSERT_COLUMNS_EQ_UR(empty_result, executeStreams(request, 10)); + WRAP_FOR_SPILL_TEST_END +} +CATCH + TEST_F(SpillJoinTestRunner, ScanHashMapAfterProbeDataWithSpillEnabledAndSpillTriggered) try { From d1f4dd72178e57a1b86f061e7ceca31ac677fb61 Mon Sep 17 00:00:00 2001 From: ChangRui-Ryan Date: Mon, 17 Aug 2026 19:30:42 +0800 Subject: [PATCH 3/3] test: verify probe reads after V1 join spill --- dbms/src/Flash/tests/gtest_spill_join.cpp | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/dbms/src/Flash/tests/gtest_spill_join.cpp b/dbms/src/Flash/tests/gtest_spill_join.cpp index 62cfca417ce..1e2b3113a83 100644 --- a/dbms/src/Flash/tests/gtest_spill_join.cpp +++ b/dbms/src/Flash/tests/gtest_spill_join.cpp @@ -267,7 +267,15 @@ try context.context->setSetting("max_bytes_before_external_join", Field(static_cast(10000))); WRAP_FOR_SPILL_TEST_BEGIN - ASSERT_COLUMNS_EQ_UR(empty_result, executeStreams(request, 10)); + DAGContext dag_context(*request, "empty_hash_after_spill", 10); + ASSERT_COLUMNS_EQ_UR(empty_result, executeStreams(&dag_context)); + + const auto & join_execute_info = dag_context.getJoinExecuteInfoMap().at("Join_2"); + ASSERT_TRUE(join_execute_info.join_profile_info->is_spilled); + + // A spilled hash join must not skip its probe side even when no build row enters the hash table. + Expect expect{{"table_scan_0", {3, 10}}, {"table_scan_1", {8192, 10}}, {"Join_2", {0, 10}}}; + testForExecutionSummary(request, expect); WRAP_FOR_SPILL_TEST_END } CATCH