Skip to content

branch-4.2: [fix](local shuffle) Prevent row loss at parallel-to-serial pipeline boundaries #67177 #68191

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
924060929 wants to merge 1 commit into from
Closed

branch-4.2: [fix](local shuffle) Prevent row loss at parallel-to-serial pipeline boundaries #67177 #68191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions be/src/exec/operator/hashjoin_build_sink.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,12 +133,16 @@ class HashJoinBuildSinkOperatorX MOCK_REMOVE(final)
._should_build_hash_table;
}

DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
DataDistribution required_data_distribution(RuntimeState* state) const override {
if (_join_op == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
return {TLocalPartitionType::NOOP};
} else if (_is_broadcast_join) {
return _child->is_serial_operator() ? DataDistribution(TLocalPartitionType::PASS_TO_ONE)
: DataDistribution(TLocalPartitionType::NOOP);
if (!_child->is_serial_operator()) {
return {TLocalPartitionType::NOOP};
}
return state->enable_share_hash_table_for_broadcast_join()
? DataDistribution(TLocalPartitionType::PASS_TO_ONE)
: DataDistribution(TLocalPartitionType::BROADCAST);
}
return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE ||
_join_distribution == TJoinDistributionType::COLOCATE
Expand Down
88 changes: 46 additions & 42 deletions be/src/exec/pipeline/pipeline_fragment_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -716,6 +716,7 @@ Status PipelineFragmentContext::_build_pipelines(ObjectPool* pool, const Descrip

Status PipelineFragmentContext::_create_deferred_local_exchangers() {
for (auto& info : _deferred_exchangers) {
const int source_count = cast_set<int>(info.shared_state->source_deps.size());
// DANGER ZONE — do not "fix" this line without reading the history.
//
// sender_count seeds Exchanger::_running_sink_operators, which the source side
Expand Down Expand Up @@ -748,34 +749,29 @@ Status PipelineFragmentContext::_create_deferred_local_exchangers() {
switch (info.partition_type) {
case TLocalPartitionType::LOCAL_EXECUTION_HASH_SHUFFLE:
case TLocalPartitionType::GLOBAL_EXECUTION_HASH_SHUFFLE:
info.shared_state->exchanger = ShuffleExchanger::create_unique(
sender_count, _num_instances, info.num_partitions, info.free_blocks_limit,
info.partition_type);
info.shared_state->exchanger =
ShuffleExchanger::create_unique(sender_count, source_count, info.num_partitions,
info.free_blocks_limit, info.partition_type);
break;
case TLocalPartitionType::BUCKET_HASH_SHUFFLE:
info.shared_state->exchanger = BucketShuffleExchanger::create_unique(
sender_count, _num_instances, info.num_partitions, info.free_blocks_limit);
sender_count, source_count, info.num_partitions, info.free_blocks_limit);
break;
case TLocalPartitionType::PASSTHROUGH:
info.shared_state->exchanger = PassthroughExchanger::create_unique(
sender_count, _num_instances, info.free_blocks_limit);
sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::BROADCAST:
info.shared_state->exchanger = BroadcastExchanger::create_unique(
sender_count, _num_instances, info.free_blocks_limit);
sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::PASS_TO_ONE:
if (_runtime_state->enable_share_hash_table_for_broadcast_join()) {
info.shared_state->exchanger = PassToOneExchanger::create_unique(
sender_count, _num_instances, info.free_blocks_limit);
} else {
info.shared_state->exchanger = BroadcastExchanger::create_unique(
sender_count, _num_instances, info.free_blocks_limit);
}
info.shared_state->exchanger = PassToOneExchanger::create_unique(
sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::ADAPTIVE_PASSTHROUGH:
info.shared_state->exchanger = AdaptivePassthroughExchanger::create_unique(
sender_count, _num_instances, info.free_blocks_limit);
sender_count, source_count, info.free_blocks_limit);
break;
case TLocalPartitionType::NOOP:
case TLocalPartitionType::LOCAL_MERGE_SORT:
Expand Down Expand Up @@ -852,11 +848,17 @@ void PipelineFragmentContext::_propagate_local_exchange_num_tasks() {
if (pit != id_to_pipe.end()) {
auto& pipe = pit->second;
const auto& ops = pipe->operators();
const bool le_source =
!ops.empty() && dynamic_cast<LocalExchangeSourceOperatorX*>(ops.front().get());
auto* le_source =
!ops.empty() ? dynamic_cast<LocalExchangeSourceOperatorX*>(ops.front().get())
: nullptr;
const bool serial_source = !ops.empty() && ops.front()->is_serial_operator();
if (le_source) {
pipe->set_num_tasks(_num_instances);
// PASS_TO_ONE is the explicit N-to-one boundary. Its upstream pipeline
// keeps all active tasks, while only fragment instance 0 creates the
// downstream serial pipeline task.
if (le_source->exchange_type() != TLocalPartitionType::PASS_TO_ONE) {
pipe->set_num_tasks(_num_instances);
}
} else if (!serial_source) {
int target = pipe->num_tasks();
const auto up_it = _dag.find(id);
Expand Down Expand Up @@ -1069,22 +1071,12 @@ Status PipelineFragmentContext::_add_local_exchange_impl(
: 0);
break;
case TLocalPartitionType::PASS_TO_ONE:
if (_runtime_state->enable_share_hash_table_for_broadcast_join()) {
// If shared hash table is enabled for BJ, hash table will be built by only one task
shared_state->exchanger = PassToOneExchanger::create_unique(
cur_pipe->num_tasks(), _num_instances,
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
? cast_set<int>(_runtime_state->query_options()
.local_exchange_free_blocks_limit)
: 0);
} else {
shared_state->exchanger = BroadcastExchanger::create_unique(
cur_pipe->num_tasks(), _num_instances,
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
? cast_set<int>(_runtime_state->query_options()
.local_exchange_free_blocks_limit)
: 0);
}
shared_state->exchanger = PassToOneExchanger::create_unique(
cur_pipe->num_tasks(), _num_instances,
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
? cast_set<int>(
_runtime_state->query_options().local_exchange_free_blocks_limit)
: 0);
break;
case TLocalPartitionType::ADAPTIVE_PASSTHROUGH:
shared_state->exchanger = AdaptivePassthroughExchanger::create_unique(
Expand Down Expand Up @@ -1987,9 +1979,12 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
}
case TPlanNodeType::LOCAL_EXCHANGE_NODE: {
op = std::make_shared<LocalExchangeSourceOperatorX>(pool, tnode, next_operator_id(), descs);
// The downstream pipeline (containing LocalExchangeSource) must have
// _num_instances tasks — matching BE-native _inherit_pipeline_properties
// which sets pipe_with_source.set_num_tasks(_num_instances).
const auto partition_type = tnode.local_exchange_node.partition_type;
const bool pass_to_one = partition_type == TLocalPartitionType::PASS_TO_ONE;
// Except at an explicit PASS_TO_ONE boundary, the downstream pipeline
// (containing LocalExchangeSource) must have _num_instances tasks. This
// matches BE-native _inherit_pipeline_properties, which sets
// pipe_with_source.set_num_tasks(_num_instances).
// Without this, when the parent pipeline was reduced by a serial operator
// (e.g., serial Exchange with use_serial_exchange=true, or UNPARTITIONED
// Exchange), the downstream inherits the reduced num_tasks via
Expand All @@ -1998,14 +1993,23 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
// sink round-robins to all channels and crashes on uninitialized ones.
RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances));
// Restore downstream pipeline's num_tasks (mirroring _inherit_pipeline_properties:
// downstream keeps _num_instances, upstream gets the serial/reduced count)
cur_pipe->set_num_tasks(_num_instances);
// downstream keeps _num_instances, upstream gets the serial/reduced count).
// PASS_TO_ONE is the explicit parallel-to-serial boundary: its downstream
// pipeline must keep the serial parent's single active task, while the upstream
// pipeline is expanded below so every fragment instance keeps an active receiver.
if (!pass_to_one) {
cur_pipe->set_num_tasks(_num_instances);
}
const int downstream_num_tasks = cur_pipe->num_tasks();

const auto downstream_pipeline_id = cur_pipe->id();
if (!_dag.contains(downstream_pipeline_id)) {
_dag.insert({downstream_pipeline_id, {}});
}
cur_pipe = add_pipeline(cur_pipe);
if (pass_to_one) {
cur_pipe->set_num_tasks(_num_instances);
}
// If this local exchange was inserted because of a serial scan (is_serial_operator),
// the upstream pipeline (cur_pipe) should have num_tasks=1 (only 1 scan task).
// We set this now so the exchanger is created with the correct sender count.
Expand All @@ -2017,7 +2021,6 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
_dag[downstream_pipeline_id].push_back(cur_pipe->id());
int num_partitions = 0;
std::map<int, int> shuffle_id_to_instance_idx;
auto partition_type = tnode.local_exchange_node.partition_type;
switch (partition_type) {
case TLocalPartitionType::BUCKET_HASH_SHUFFLE:
num_partitions = _params.num_buckets;
Expand Down Expand Up @@ -2057,9 +2060,10 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
? cast_set<int>(
_runtime_state->query_options().local_exchange_free_blocks_limit)
: 0;
auto shared_state = LocalExchangeSharedState::create_shared(_num_instances);
shared_state->create_source_dependencies(_num_instances, local_exchange_id,
local_exchange_id, "LOCAL_EXCHANGE_OPERATOR");
const int source_count = downstream_num_tasks;
auto shared_state = LocalExchangeSharedState::create_shared(source_count);
shared_state->create_source_dependencies(source_count, local_exchange_id, local_exchange_id,
"LOCAL_EXCHANGE_OPERATOR");
shared_state->create_sink_dependency(sink_id, local_exchange_id, "LOCAL_EXCHANGE_SINK");
_op_id_to_shared_state.insert({local_exchange_id, {shared_state, shared_state->sink_deps}});
// Defer exchanger creation: sender count depends on final upstream num_tasks
Expand Down
26 changes: 25 additions & 1 deletion be/test/exec/operator/hashjoin_build_sink_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,30 @@ TEST_F(HashJoinBuildSinkTest, Sink) {
run_test_block(test_block);
}

TEST_F(HashJoinBuildSinkTest, BroadcastJoinRequiredDataDistribution) {
auto tnode = _helper.create_test_plan_node(TJoinOp::INNER_JOIN, {TPrimitiveType::INT}, {false},
{false});
tnode.hash_join_node.__set_is_broadcast_join(true);
auto [probe_operator, sink_operator] = _helper.create_operators(tnode);
ASSERT_TRUE(probe_operator);
ASSERT_TRUE(sink_operator);

EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
.distribution_type,
TLocalPartitionType::NOOP);

sink_operator->child()->set_serial_operator();
_helper.runtime_state->_enable_share_hash_table_for_broadcast_join = true;
EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
.distribution_type,
TLocalPartitionType::PASS_TO_ONE);

_helper.runtime_state->_enable_share_hash_table_for_broadcast_join = false;
EXPECT_EQ(sink_operator->required_data_distribution(_helper.runtime_state.get())
.distribution_type,
TLocalPartitionType::BROADCAST);
}

TEST_F(HashJoinBuildSinkTest, Terminate) {
auto test_block = [&](TJoinOp::type op_type, const std::vector<TPrimitiveType::type>& key_types,
const std::vector<bool>& left_nullables,
Expand Down Expand Up @@ -358,4 +382,4 @@ TEST_F(HashJoinBuildSinkTest, Terminate) {
run_test_block(test_block);
}

} // namespace doris
} // namespace doris
69 changes: 63 additions & 6 deletions be/test/exec/pipeline/local_exchanger_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,12 +29,14 @@
#include "exec/exchange/local_exchange_sink_operator.h"
#include "exec/exchange/local_exchange_source_operator.h"
#include "exec/pipeline/dependency.h"
#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/pipeline/thrift_builder.h"
#include "exprs/vslot_ref.h"
#include "runtime/descriptor_helper.h"

namespace doris {

class LocalExchangerTest : public testing::Test {
class LocalExchangerTest : public testing::TestWithParam<int> {
public:
LocalExchangerTest() = default;
~LocalExchangerTest() override = default;
Expand Down Expand Up @@ -526,9 +528,60 @@ TEST_F(LocalExchangerTest, PassthroughExchanger) {
}
}

TEST_F(LocalExchangerTest, PassToOneExchanger) {
TEST_F(LocalExchangerTest, FePlannedPassToOneUsesOneDownstreamSource) {
constexpr int num_instances = 4;
_query_options.__set_enable_share_hash_table_for_broadcast_join(false);
TPipelineFragmentParams params;
auto context = std::make_shared<PipelineFragmentContext>(
_query_id, params, _query_ctx, ExecEnv::GetInstance(), [](RuntimeState*, Status*) {});
context->_num_instances = num_instances;
context->_total_instances = num_instances;
context->_runtime_state = RuntimeState::create_unique(_query_id, _fragment_id, _query_options,
_query_ctx->query_globals,
ExecEnv::GetInstance(), _query_ctx.get());

auto downstream_pipe = context->add_pipeline();
downstream_pipe->set_num_tasks(1);
auto upstream_pipe = downstream_pipe;

TLocalExchangeNode local_exchange_node;
local_exchange_node.__set_partition_type(TLocalPartitionType::PASS_TO_ONE);
TPlanNode tnode;
tnode.__set_node_type(TPlanNodeType::LOCAL_EXCHANGE_NODE);
tnode.__set_node_id(0);
tnode.__set_num_children(1);
tnode.__set_local_exchange_node(local_exchange_node);
tnode.__set_row_tuples({0});

ObjectPool pool;
TDescriptorTableBuilder desc_builder;
TTupleDescriptorBuilder().build(&desc_builder);
DescriptorTbl* descs = nullptr;
ASSERT_TRUE(DescriptorTbl::create(&pool, desc_builder.desc_tbl(), &descs).ok());
OperatorPtr op;
OperatorPtr cache_op;
ASSERT_TRUE(context->_create_operator(&pool, tnode, *descs, op, upstream_pipe,
/*parent_idx=*/-1, /*child_idx=*/0,
/*followed_by_shuffled_operator=*/false,
/*require_bucket_distribution=*/false, cache_op)
.ok());
ASSERT_EQ(context->_deferred_exchangers.size(), 1);
EXPECT_EQ(downstream_pipe->num_tasks(), 1);
EXPECT_EQ(upstream_pipe->num_tasks(), num_instances);

auto shared_state = context->_deferred_exchangers.front().shared_state;
EXPECT_EQ(shared_state->source_deps.size(), 1);
EXPECT_EQ(shared_state->mem_counters.size(), 1);
ASSERT_TRUE(context->_create_deferred_local_exchangers().ok());
ASSERT_NE(shared_state->exchanger, nullptr);
EXPECT_EQ(shared_state->exchanger->get_type(), TLocalPartitionType::PASS_TO_ONE);
EXPECT_EQ(shared_state->exchanger->_num_senders, num_instances);
EXPECT_EQ(shared_state->exchanger->_num_sources, 1);
}

TEST_P(LocalExchangerTest, PassToOneExchanger) {
int num_sink = 4;
int num_sources = 4;
int num_sources = GetParam();
int free_block_limit = 0;

const auto expect_block_bytes = 128;
Expand All @@ -549,6 +602,8 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
shared_state->create_source_dependencies(num_sources, 0, 0, "TEST");

auto* exchanger = (PassToOneExchanger*)shared_state->exchanger.get();
EXPECT_EQ(exchanger->_num_senders, num_sink);
EXPECT_EQ(exchanger->_num_sources, num_sources);
for (size_t i = 0; i < num_sink; i++) {
auto* compute_hash_value_timer =
ADD_TIMER(profile, "ComputeHashValueTime" + std::to_string(i));
Expand Down Expand Up @@ -579,10 +634,9 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
"MemoryUsage" + std::to_string(i), TUnit::BYTES, "", 1);
shared_state->mem_counters[i] = _local_states[i]->_memory_used_counter;
}

{
// Enqueue `num_blocks` blocks with 10 rows for each data queue.
for (size_t i = 0; i < num_sources; i++) {
// Enqueue `num_blocks` blocks with 10 rows from every sender.
for (size_t i = 0; i < num_sink; i++) {
for (size_t j = 0; j < num_blocks; j++) {
Block in_block;
DataTypePtr int_type = std::make_shared<DataTypeInt32>();
Expand Down Expand Up @@ -735,6 +789,8 @@ TEST_F(LocalExchangerTest, PassToOneExchanger) {
}
}

INSTANTIATE_TEST_SUITE_P(SourceCardinality, LocalExchangerTest, testing::Values(4, 1));

TEST_F(LocalExchangerTest, BroadcastExchanger) {
int num_sink = 4;
int num_sources = 4;
Expand Down Expand Up @@ -1386,4 +1442,5 @@ TEST_F(LocalExchangerTest, ShuffleExchangerRestoreOutputBlockOnAddRowsError) {
EXPECT_EQ(output_block.rows(), 1);
EXPECT_NO_THROW(output_block.check_number_of_rows());
}

} // namespace doris
Loading
Loading