Skip to content

Preserve a few settings to partition export #1990

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

Open
arthurpassos wants to merge 2 commits into
base: antalya-26.3
Choose a base branch
from
+25 −0
Open

Preserve a few settings to partition export #1990

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b53014e
preserve a few settings to partition export
arthurpassos Jun 30, 2026
a51472b
add output_format_parquet_row_group_size_bytes and remove default values
arthurpassos Jul 1, 2026
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: 13 additions & 0 deletions src/Storages/ExportReplicatedMergeTreePartitionManifest.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,10 @@ struct ExportReplicatedMergeTreePartitionManifest
bool write_full_path_in_iceberg_metadata = false;
bool allow_lossy_cast = false;
String iceberg_metadata_json;
String parquet_compression_method;
UInt64 output_format_compression_level;
UInt64 parquet_row_group_size;
UInt64 parquet_row_group_size_bytes;

std::string toJsonString() const
{
Expand Down Expand Up @@ -208,6 +212,10 @@ struct ExportReplicatedMergeTreePartitionManifest
json.set("task_timeout_seconds", task_timeout_seconds);
json.set("write_full_path_in_iceberg_metadata", write_full_path_in_iceberg_metadata);
json.set("allow_lossy_cast", allow_lossy_cast);
json.set("parquet_compression_method", parquet_compression_method);
json.set("output_format_compression_level", output_format_compression_level);
json.set("parquet_row_group_size", parquet_row_group_size);
json.set("parquet_row_group_size_bytes", parquet_row_group_size_bytes);
std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
oss.exceptions(std::ios::failbit);
Poco::JSON::Stringifier::stringify(json, oss);
Expand Down Expand Up @@ -266,6 +274,11 @@ struct ExportReplicatedMergeTreePartitionManifest
/// on upgrade. New tasks always persist the initiator's actual choice.
manifest.allow_lossy_cast = json->has("allow_lossy_cast") ? json->getValue<bool>("allow_lossy_cast") : true;

manifest.parquet_compression_method = json->getValue<String>("parquet_compression_method");
manifest.output_format_compression_level = json->getValue<UInt64>("output_format_compression_level");
manifest.parquet_row_group_size = json->getValue<UInt64>("parquet_row_group_size");
manifest.parquet_row_group_size_bytes = json->getValue<UInt64>("parquet_row_group_size_bytes");

return manifest;
}
};
Expand Down
4 changes: 4 additions & 0 deletions src/Storages/MergeTree/ExportPartitionUtils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,10 @@ namespace ExportPartitionUtils
context_copy->setCurrentQueryId(manifest.query_id);
context_copy->setSetting("output_format_parallel_formatting", manifest.parallel_formatting);
context_copy->setSetting("output_format_parquet_parallel_encoding", manifest.parquet_parallel_encoding);
context_copy->setSetting("output_format_parquet_compression_method", manifest.parquet_compression_method);
context_copy->setSetting("output_format_compression_level", manifest.output_format_compression_level);
context_copy->setSetting("output_format_parquet_row_group_size", manifest.parquet_row_group_size);
context_copy->setSetting("output_format_parquet_row_group_size_bytes", manifest.parquet_row_group_size_bytes);
context_copy->setSetting("max_threads", manifest.max_threads);
context_copy->setSetting("export_merge_tree_part_file_already_exists_policy", String(magic_enum::enum_name(manifest.file_already_exists_policy)));
context_copy->setSetting("export_merge_tree_part_max_bytes_per_file", manifest.max_bytes_per_file);
Expand Down
8 changes: 8 additions & 0 deletions src/Storages/StorageReplicatedMergeTree.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,10 @@ namespace Setting
extern const SettingsUInt64 export_merge_tree_partition_task_timeout_seconds;
extern const SettingsBool output_format_parallel_formatting;
extern const SettingsBool output_format_parquet_parallel_encoding;
extern const SettingsParquetCompression output_format_parquet_compression_method;
extern const SettingsUInt64 output_format_compression_level;
extern const SettingsUInt64 output_format_parquet_row_group_size;
extern const SettingsUInt64 output_format_parquet_row_group_size_bytes;
extern const SettingsMaxThreads max_threads;
extern const SettingsMergeTreePartExportFileAlreadyExistsPolicy export_merge_tree_part_file_already_exists_policy;
extern const SettingsUInt64 export_merge_tree_part_max_bytes_per_file;
Expand Down Expand Up @@ -8505,6 +8509,10 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
manifest.max_threads = query_context->getSettingsRef()[Setting::max_threads];
manifest.parallel_formatting = query_context->getSettingsRef()[Setting::output_format_parallel_formatting];
manifest.parquet_parallel_encoding = query_context->getSettingsRef()[Setting::output_format_parquet_parallel_encoding];
manifest.parquet_compression_method = query_context->getSettingsRef()[Setting::output_format_parquet_compression_method].toString();
manifest.output_format_compression_level = query_context->getSettingsRef()[Setting::output_format_compression_level];
manifest.parquet_row_group_size = query_context->getSettingsRef()[Setting::output_format_parquet_row_group_size];

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 30, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Persist the Parquet row-group byte limit too

When users schedule EXPORT PARTITION with output_format_parquet_row_group_size_bytes (for example to avoid the default 512 MiB cap), this manifest only captures output_format_parquet_row_group_size; the background export context is rebuilt from the manifest, and getFormatSettings consumes both output_format_parquet_row_group_size and output_format_parquet_row_group_size_bytes (src/Formats/FormatFactory.cpp:202-203). Any worker replica will therefore fall back to its default byte limit and produce different row groups than the initiating query. Please persist and restore output_format_parquet_row_group_size_bytes alongside this field.

Useful? React with 👍 / 👎.

manifest.parquet_row_group_size_bytes = query_context->getSettingsRef()[Setting::output_format_parquet_row_group_size_bytes];
manifest.max_bytes_per_file = query_context->getSettingsRef()[Setting::export_merge_tree_part_max_bytes_per_file];
manifest.max_rows_per_file = query_context->getSettingsRef()[Setting::export_merge_tree_part_max_rows_per_file];

Expand Down