Skip to content
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

Jump to bottom

[Enhancement](hive-writer) Adjust table sink exchange rebalancer params. #33397

Merged
merged 3 commits into from
Apr 12, 2024
Merged

[Enhancement](hive-writer) Adjust table sink exchange rebalancer params. #33397

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
45e6cc2
Init commit.
kaka11chen Apr 8, 2024
5ed8c4f
Fix compiling issue.
kaka11chen Apr 9, 2024
90ad364
remote task address facility.
kaka11chen Apr 9, 2024
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
13 changes: 7 additions & 6 deletions be/src/common/config.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1181,14 +1181,15 @@ DEFINE_mString(ca_cert_file_paths,
"/etc/ssl/ca-bundle.pem");

/** Table sink configurations(currently contains only external table types) **/
// Minimum data processed to scale writers when non partition writing
// Minimum data processed to scale writers in exchange when non partition writing
DEFINE_mInt64(table_sink_non_partition_write_scaling_data_processed_threshold,
"125829120"); // 120MB
// Minimum data processed to start rebalancing in exchange when partition writing
DEFINE_mInt64(table_sink_partition_write_data_processed_threshold, "209715200"); // 200MB
"26214400"); // 25MB
// Minimum data processed to trigger skewed partition rebalancing in exchange when partition writing
DEFINE_mInt64(table_sink_partition_write_skewed_data_processed_rebalance_threshold,
"209715200"); // 200MB
DEFINE_mInt64(table_sink_partition_write_min_data_processed_rebalance_threshold,
"26214400"); // 25MB
// Minimum partition data processed to rebalance writers in exchange when partition writing
DEFINE_mInt64(table_sink_partition_write_min_partition_data_processed_rebalance_threshold,
"15728640"); // 15MB
// Maximum processed partition nums of per writer when partition writing
DEFINE_mInt32(table_sink_partition_write_max_partition_nums_per_writer, "128");

Expand Down
8 changes: 4 additions & 4 deletions be/src/common/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1265,12 +1265,12 @@ DECLARE_String(trino_connector_plugin_dir);
DECLARE_mString(ca_cert_file_paths);

/** Table sink configurations(currently contains only external table types) **/
// Minimum data processed to scale writers when non partition writing
// Minimum data processed to scale writers in exchange when non partition writing
DECLARE_mInt64(table_sink_non_partition_write_scaling_data_processed_threshold);
// Minimum data processed to start rebalancing in exchange when partition writing
DECLARE_mInt64(table_sink_partition_write_data_processed_threshold);
// Minimum data processed to trigger skewed partition rebalancing in exchange when partition writing
DECLARE_mInt64(table_sink_partition_write_skewed_data_processed_rebalance_threshold);
DECLARE_mInt64(table_sink_partition_write_min_data_processed_rebalance_threshold);
// Minimum partition data processed to rebalance writers in exchange when partition writing
DECLARE_mInt64(table_sink_partition_write_min_partition_data_processed_rebalance_threshold);
// Maximum processed partition nums of per writer when partition writing
DECLARE_mInt32(table_sink_partition_write_max_partition_nums_per_writer);

Expand Down
20 changes: 13 additions & 7 deletions be/src/pipeline/exec/exchange_sink_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,17 +257,23 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
_partitioner.reset(
new vectorized::Crc32HashPartitioner<LocalExchangeChannelIds>(_partition_count));
_partition_function.reset(new HashPartitionFunction(_partitioner.get()));
// const long MEGABYTE = 1024 * 1024;
// const long MIN_PARTITION_DATA_PROCESSED_REBALANCE_THRESHOLD = 10000 * MEGABYTE; // 1MB
// const long MIN_DATA_PROCESSED_REBALANCE_THRESHOLD = 50000 * MEGABYTE; // 50MB

// const long MIN_PARTITION_DATA_PROCESSED_REBALANCE_THRESHOLD = 1; // 1MB
// const long MIN_DATA_PROCESSED_REBALANCE_THRESHOLD = 1; // 50MB
scale_writer_partitioning_exchanger.reset(new vectorized::ScaleWriterPartitioningExchanger<
HashPartitionFunction>(
channels.size(), *_partition_function, _partition_count, channels.size(), 1,
config::table_sink_partition_write_data_processed_threshold,
config::table_sink_partition_write_skewed_data_processed_rebalance_threshold));
config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold /
state->task_num() ==
0
? config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold
: config::table_sink_partition_write_min_partition_data_processed_rebalance_threshold /
state->task_num(),
config::table_sink_partition_write_min_data_processed_rebalance_threshold /
state->task_num() ==
0
? config::table_sink_partition_write_min_data_processed_rebalance_threshold
: config::table_sink_partition_write_min_data_processed_rebalance_threshold /
state->task_num()));

RETURN_IF_ERROR(_partitioner->init(p._texprs));
RETURN_IF_ERROR(_partitioner->prepare(state, p._row_desc));
_profile->add_info_string("Partitioner",
Expand Down
1 change: 1 addition & 0 deletions be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -593,6 +593,7 @@ Status PipelineXFragmentContext::_build_pipeline_tasks(
init_runtime_state(task_runtime_state);
auto cur_task_id = _total_tasks++;
task_runtime_state->set_task_id(cur_task_id);
task_runtime_state->set_task_num(pipeline->num_tasks());
auto task = std::make_unique<PipelineXTask>(
pipeline, cur_task_id, get_task_runtime_state(cur_task_id), this,
pipeline_id_to_profile[pip_idx].get(), get_local_exchange_state(pipeline),
Expand Down
5 changes: 5 additions & 0 deletions be/src/runtime/runtime_state.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -624,6 +624,10 @@ class RuntimeState {

int task_id() const { return _task_id; }

void set_task_num(int task_num) { _task_num = task_num; }

int task_num() const { return _task_num; }

private:
Status create_error_log_file();

Expand Down Expand Up @@ -734,6 +738,7 @@ class RuntimeState {
std::vector<TErrorTabletInfo> _error_tablet_infos;
int _max_operator_id = 0;
int _task_id = -1;
int _task_num = 0;

std::vector<THivePartitionUpdate> _hive_partition_updates;

Expand Down
Loading