From 3f47047259d83cd6cf582a2130053a091c933b32 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Fri, 17 Jul 2026 16:59:22 -0300 Subject: [PATCH 1/8] vibe coded monotonicity check for mt -> iceberg exports --- docs/en/antalya/partition_export.md | 13 + .../MergeTree/ExportPartitionUtils.cpp | 319 ++++++++++++++---- src/Storages/MergeTree/ExportPartitionUtils.h | 17 +- src/Storages/MergeTree/MergeTreeData.cpp | 18 +- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 6 +- src/Storages/StorageReplicatedMergeTree.cpp | 6 +- .../test.py | 243 +++++++++++-- 7 files changed, 513 insertions(+), 109 deletions(-) diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 109971ac1e9e..5ef422426387 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -24,6 +24,19 @@ The manifest file produced by the commit contains a summary field `clickhouse.ex The Iceberg manifest files contain statistics about the data. Exporting a merge tree partition is a non ephemeral long running task, in which nodes can be turned off and turned on. This means the stats of individual files need to be persisted somewhere in order to produce the final manifest. This is implemented through sidecars. Each data file exported will contain a "sibling" sidecar file named `_clickhouse_export_part_sidecar.avro`. ClickHouse does not clean up these files, and they can be safely deleted once the data is comitted. +#### Source partition key compatibility + +Because the commit writes a single partition tuple per exported partition, every destination Iceberg partition field must be single-valued across the exported source partition. A source `PARTITION BY` field is accepted when either: + +- It structurally matches the destination transform on the same column. The functions with a direct Iceberg equivalent are `identity` (a bare column), `toYearNumSinceEpoch` (`year`), `toMonthNumSinceEpoch` (`month`), `toRelativeDayNum` (`day`), `toRelativeHourNum` (`hour`), `icebergTruncate` (`truncate`), and `icebergBucket` (`bucket`). +- Or the destination transform is proven constant over the exported partition's actual `[min, max]`. This accepts other equivalent or finer keys - for example `PARTITION BY toDate(ts)` or `toYYYYMM(ts)` or `toStartOfHour(ts)` into a destination partitioned by `day(ts)` / `month(ts)` / `hour(ts)`, a bare `Date` column into a `day` transform, or a source that adds extra partition columns on top of the destination's. + +The proof uses each part's min/max statistics, so it is data-dependent: a partition whose rows would span more than one destination partition (for example a monthly source partition exported into a daily destination that actually contains several days) is rejected with a `BAD_ARGUMENTS` error at `EXPORT` time. + +`bucket` is a hash and is not order-preserving, so it can only be matched structurally: the source must be partitioned by `icebergBucket(N, col)` with the same `N`. + +Lossy partition-column casts (allowed via `export_merge_tree_part_allow_lossy_cast`) are supported as long as the cast stays order-preserving over the partition's actual values; a partition whose values cross the destination type's overflow boundary is rejected. A `Nullable` partition column is only accepted through a structural match, because a `NULL` forms its own Iceberg partition. + ### On plain object storage exports: Each MergeTree part will become a separate file with the following name convention: `//_.`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `/commit__`. diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index ec8046f3e991..4efc8374840b 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -7,6 +7,7 @@ #include "Storages/ExportReplicatedMergeTreePartitionManifest.h" #include "Storages/ExportReplicatedMergeTreePartitionTaskEntry.h" #include +#include #include #include #include @@ -18,8 +19,18 @@ #include #if USE_AVRO +#include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include #endif namespace ProfileEvents @@ -75,6 +86,9 @@ namespace ErrorCodes namespace Setting { extern const SettingsBool export_merge_tree_part_allow_lossy_cast; +#if USE_AVRO + extern const SettingsTimezone iceberg_partition_timezone; +#endif } namespace FailPoints @@ -482,9 +496,145 @@ namespace ExportPartitionUtils } #if USE_AVRO +namespace +{ + /// One top-level term of the source MergeTree PARTITION BY, canonicalized for comparison against a + /// destination Iceberg partition field. `transform` holds the ClickHouse function name that + /// parseTransformAndArgument yields for the equivalent Iceberg transform ("toRelativeDayNum", + /// "identity", "icebergBucket", ...); std::nullopt marks a term that is not directly + /// Iceberg-representable and can therefore only satisfy a destination field via the dynamic proof. + struct SourcePartitionTerm + { + String column; + std::optional transform; + std::optional argument; + }; + + std::vector parseSourcePartitionTerms(const ASTPtr & partition_key_ast) + { + /// The same functions getPartitionField maps 1:1 to an Iceberg transform. + static const std::unordered_set iceberg_representable = { + "identity", "toYearNumSinceEpoch", "toMonthNumSinceEpoch", + "toRelativeDayNum", "toRelativeHourNum", "icebergBucket", "icebergTruncate"}; + + std::vector terms; + if (!partition_key_ast) + return terms; + + auto parse_one = [&](const ASTPtr & element) + { + if (const auto * ident = element->as()) + { + terms.push_back({ident->name(), std::optional("identity"), std::nullopt}); + return; + } + + const auto * func = element->as(); + if (!func) + return; + + std::optional column; + std::optional argument; + for (const auto & child : func->children) + { + const auto * expression_list = child->as(); + if (!expression_list) + continue; + for (const auto & arg : expression_list->children) + { + if (const auto * id = arg->as()) + column = id->name(); + else if (const auto * lit = arg->as(); lit && lit->value.getType() != Field::Types::String) + argument = lit->value.safeGet(); + } + } + if (!column) + return; + + std::optional transform; + if (iceberg_representable.contains(func->name)) + transform = func->name; + terms.push_back({*column, transform, argument}); + }; + + if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") + { + for (const auto & child : func->children) + if (const auto * expression_list = child->as()) + for (const auto & element : expression_list->children) + parse_one(element); + } + else + parse_one(partition_key_ast); + + return terms; + } + + /// Fold [min, max] of the partition-key column at `slot` across all parts of the partition. + /// Returns false if no part exposes an initialized min/max for that column. + bool foldPartitionColumnBounds( + const MergeTreeData::DataPartsVector & parts, size_t slot, Field & out_min, Field & out_max) + { + bool found = false; + for (const auto & part : parts) + { + if (!part->minmax_idx || !part->minmax_idx->initialized || slot >= part->minmax_idx->hyperrectangle.size()) + continue; + auto range = part->minmax_idx->hyperrectangle[slot]; + range.shrinkToIncludedIfPossible(); + if (!found) + { + out_min = range.left; + out_max = range.right; + found = true; + } + else + { + if (range.left < out_min) + out_min = range.left; + if (out_max < range.right) + out_max = range.right; + } + } + return found; + } + + /// Apply the destination Iceberg transform (rebuilt as a ClickHouse function, mirroring + /// ChunkPartitioner and the commit path) to a 2-row column holding [cast(min), cast(max)] and + /// return whether both rows map to the same value. + bool destinationTransformIsConstant( + const Iceberg::TransformAndArgument & transform, const ColumnWithTypeAndName & values, const ContextPtr & context) + { + auto function = FunctionFactory::instance().get(transform.transform_name, context); + const size_t num_rows = values.column->size(); + + ColumnsWithTypeAndName arguments; + if (transform.argument) + arguments.push_back({DataTypeUInt64().createColumnConst(num_rows, *transform.argument), + std::make_shared(), "width"}); + arguments.push_back(values); + if (transform.time_zone) + arguments.push_back({DataTypeString().createColumnConst(num_rows, *transform.time_zone), + std::make_shared(), "timezone"}); + + const auto result_type = function->getReturnType(arguments); + const auto result = function->build(arguments)->execute(arguments, result_type, num_rows, /*dry_run=*/ false); + + Field first; + Field second; + result->get(0, first); + result->get(1, second); + return first == second; + } +} + void verifyIcebergPartitionCompatibility( const Poco::JSON::Object::Ptr & metadata_object, - const ASTPtr & partition_key_ast) + const StorageMetadataPtr & source_metadata, + const StorageMetadataPtr & destination_metadata, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context) { const auto original_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); const auto partition_spec_id = metadata_object->getValue(Iceberg::f_default_spec_id); @@ -520,83 +670,138 @@ namespace ExportPartitionUtils if (!current_schema_json || !partition_spec_json) return; - /// Build column_name → Iceberg source-id from the destination schema (and the inverse). - std::unordered_map column_name_to_source_id; std::unordered_map source_id_to_column_name; { const auto schema_fields = current_schema_json->getArray(Iceberg::f_fields); for (size_t i = 0; i < schema_fields->size(); ++i) { auto f = schema_fields->getObject(static_cast(i)); - const auto col_name = f->getValue(Iceberg::f_name); - const auto source_id = f->getValue(Iceberg::f_id); - column_name_to_source_id[col_name] = source_id; - source_id_to_column_name[source_id] = col_name; + source_id_to_column_name[f->getValue(Iceberg::f_id)] = f->getValue(Iceberg::f_name); } } - auto source_id_to_name = [&](Int32 id) -> String { auto it = source_id_to_column_name.find(id); return it != source_id_to_column_name.end() ? it->second : fmt::format("", id); }; - /// Convert the MergeTree PARTITION BY AST into the equivalent Iceberg spec. - Poco::JSON::Array::Ptr expected_fields; - try + const auto actual_fields = partition_spec_json->getArray(Iceberg::f_fields); + const size_t actual_size = actual_fields ? actual_fields->size() : 0; + + const auto source_terms = parseSourcePartitionTerms(source_metadata->getPartitionKeyAST()); + + /// A partitioned source cannot be exported into an unpartitioned Iceberg table: there is no + /// destination partitioning to satisfy and the result would silently drop the source layout. + if (actual_size == 0) { - const auto expected_spec = Iceberg::getPartitionSpec( - partition_key_ast, column_name_to_source_id).first; - expected_fields = expected_spec->getArray(Iceberg::f_fields); + if (!source_terms.empty()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: partition scheme mismatch. " + "Source MergeTree is partitioned but the destination Iceberg table is unpartitioned."); + return; } - catch (const Exception & e) + + const auto & source_partition_key = source_metadata->getPartitionKey(); + const auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(source_partition_key); + const auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(source_partition_key); + const auto destination_sample = destination_metadata->getSampleBlockNonMaterialized(); + const String partition_timezone = context->getSettingsRef()[Setting::iceberg_partition_timezone]; + + for (UInt32 i = 0; i < actual_size; ++i) { - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition to Iceberg table: the source MergeTree partition " - "key cannot be represented as an Iceberg partition spec: {}", e.message()); - } + const auto af = actual_fields->getObject(i); + const auto dest_source_id = af->getValue(Iceberg::f_source_id); + const auto dest_transform = af->getValue(Iceberg::f_transform); + const String column = source_id_to_name(dest_source_id); + const auto dest_canonical = Iceberg::parseTransformAndArgument(dest_transform, ""); + + const std::optional dest_argument = dest_canonical && dest_canonical->argument + ? std::optional(static_cast(*dest_canonical->argument)) : std::nullopt; + + /// Fast path: the source already applies the matching transform on this column, so every + /// source partition is single-valued for this field by construction (covers bucket too). + bool matched_structurally = false; + for (const auto & term : source_terms) + { + if (term.column == column && term.transform && dest_canonical + && *term.transform == dest_canonical->transform_name && term.argument == dest_argument) + { + matched_structurally = true; + break; + } + } + if (matched_structurally) + continue; - const auto actual_fields = partition_spec_json->getArray(Iceberg::f_fields); - const size_t expected_size = expected_fields ? expected_fields->size() : 0; - const size_t actual_size = actual_fields ? actual_fields->size() : 0; + /// bucket is a hash: not order-preserving, so per-partition min/max cannot prove + /// single-valuedness. It (and any transform we cannot reconstruct) must match structurally. + if (!dest_canonical || dest_canonical->transform_name == "icebergBucket") + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: destination field on column '{}' uses " + "transform '{}', which requires the source MergeTree to be partitioned by the matching " + "function.", column, dest_transform); + + /// Dynamic proof: the destination transform must be constant across the partition's data. + const auto slot_it = std::find(minmax_column_names.begin(), minmax_column_names.end(), column); + if (slot_it == minmax_column_names.end()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: destination partitions by column '{}' " + "(transform '{}'), which is not part of the source MergeTree partition key.", + column, dest_transform); + const size_t slot = static_cast(slot_it - minmax_column_names.begin()); + const auto & source_type = minmax_column_types[slot]; + + /// A NULL value forms its own Iceberg partition, so a nullable column may split the source + /// partition; min/max cannot rule that out. Require a structural match for such columns. + if (source_type->isNullable()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: column '{}' is Nullable, so a NULL forms a " + "separate Iceberg partition; partition the source by the matching Iceberg transform.", + column); - if (expected_size != actual_size) - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition to Iceberg table: partition scheme mismatch. " - "Source MergeTree has {} partition field(s), destination Iceberg table has {}.", - expected_size, actual_size); + Field min_value; + Field max_value; + if (!foldPartitionColumnBounds(parts, slot, min_value, max_value)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: no min/max statistics available for column " + "'{}' in partition '{}'; cannot validate partitioning.", column, partition_id); - for (size_t i = 0; i < expected_size; ++i) - { - auto ef = expected_fields->getObject(static_cast(i)); - auto af = actual_fields->getObject(static_cast(i)); - - const auto expected_source_id = ef->getValue(Iceberg::f_source_id); - const auto actual_source_id = af->getValue(Iceberg::f_source_id); - const auto expected_transform = ef->getValue(Iceberg::f_transform); - const auto actual_transform = af->getValue(Iceberg::f_transform); - - /// Normalize both transform names through parseTransformAndArgument so that - /// equivalent aliases ("day"/"days", "hour"/"hours", "year"/"years", etc.) - /// produced by different writers (ClickHouse vs Spark/Trino) compare equal. - /// Comparison is on {function_name, argument}; time_zone is writer-specific - /// and not part of the partition spec identity. - const auto expected_canonical = Iceberg::parseTransformAndArgument(expected_transform, ""); - const auto actual_canonical = Iceberg::parseTransformAndArgument(actual_transform, ""); - const bool transforms_match = - (expected_canonical && actual_canonical) - ? (expected_canonical->transform_name == actual_canonical->transform_name - && expected_canonical->argument == actual_canonical->argument) - : (expected_transform == actual_transform); - - if (expected_source_id != actual_source_id || !transforms_match) + if (!destination_sample.has(column)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: destination column '{}' not found.", column); + const auto destination_type = destination_sample.getByName(column).type; + + /// The written value is transform(cast(source_col)). A value-preserving cast is + /// order-preserving, so the transform stays monotonic and the endpoints prove + /// single-valuedness. A lossy cast may wrap, so require it to be monotonic over this + /// partition's actual range; otherwise the partition genuinely spans several Iceberg + /// partitions and cannot be committed as one. + bool cast_is_monotonic = canBeSafelyCast(source_type, destination_type); + if (!cast_is_monotonic) + { + const auto cast_function + = createInternalCast({source_type, column}, destination_type, CastType::nonAccurate, {}, context); + cast_is_monotonic = cast_function->hasInformationAboutMonotonicity() + && cast_function->getMonotonicityForRange(*source_type, min_value, max_value).is_monotonic; + } + if (!cast_is_monotonic) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: values of column '{}' in partition '{}' cross " + "a non-monotonic cast boundary to the destination type {}, so the partition maps to " + "multiple Iceberg partitions.", column, partition_id, destination_type->getName()); + + auto values_column = source_type->createColumn(); + values_column->insert(min_value); + values_column->insert(max_value); + const ColumnWithTypeAndName cast_values{ + castColumn({std::move(values_column), source_type, column}, destination_type), destination_type, column}; + + const auto dest_transform_with_tz = Iceberg::parseTransformAndArgument(dest_transform, partition_timezone); + if (!destinationTransformIsConstant(*dest_transform_with_tz, cast_values, context)) throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition to Iceberg table: partition field {} mismatch. " - "Source MergeTree maps to column '{}' (source_id={}) transform='{}', " - "but destination Iceberg has column '{}' (source_id={}) transform='{}'.", - i, - source_id_to_name(expected_source_id), expected_source_id, expected_transform, - source_id_to_name(actual_source_id), actual_source_id, actual_transform); + "Cannot export partition to Iceberg table: partition '{}' spans multiple destination " + "partitions for column '{}' (transform '{}'). A source MergeTree partition must map to a " + "single Iceberg partition.", partition_id, column, dest_transform); } } #endif diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 0bb8acb9bda4..8580deb2f4a7 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/src/Storages/MergeTree/ExportPartitionUtils.h @@ -8,6 +8,7 @@ #include #include "Storages/IStorage.h" #include +#include #include #if USE_AVRO @@ -100,12 +101,20 @@ namespace ExportPartitionUtils const ContextPtr & context); #if USE_AVRO - /// Verifies the source MergeTree partition key matches the destination Iceberg - /// partition spec (source-ids and transforms in order). Throws BAD_ARGUMENTS on - /// mismatch. + /// Verifies the source MergeTree partition key is compatible with the destination Iceberg + /// partition spec: every destination partition field must be single-valued across the exported + /// source partition (which the commit path requires - it writes one partition tuple per export). + /// A field is proven either structurally (the source key already applies the matching transform + /// on that column) or dynamically, by checking the destination transform is constant over the + /// partition's actual [min, max] folded across `parts`. `bucket` is non-monotonic and can only be + /// matched structurally. Throws BAD_ARGUMENTS when a field cannot be proven. void verifyIcebergPartitionCompatibility( const Poco::JSON::Object::Ptr & metadata_object, - const ASTPtr & partition_key_ast); + const StorageMetadataPtr & source_metadata, + const StorageMetadataPtr & destination_metadata, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context); #endif } diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index 937c675fab2f..b45c0d4de5bc 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6700,6 +6700,12 @@ void MergeTreeData::exportPartToTable( auto source_metadata_ptr = getInMemoryMetadataPtr(); auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr(); + auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}); + + if (!part) + throw Exception(ErrorCodes::NO_SUCH_DATA_PART, "No such data part '{}' to export in table '{}'", + part_name, getStorageID().getFullTableName()); + std::string iceberg_metadata_json; if (dest_storage->isDataLake()) @@ -6745,7 +6751,11 @@ void MergeTreeData::exportPartToTable( ExportPartitionUtils::verifyIcebergPartitionCompatibility( metadata_object, - source_metadata_ptr->getPartitionKeyAST()); + source_metadata_ptr, + destination_metadata_ptr, + {part}, + part->info.getPartitionId(), + query_context); } #else (void)iceberg_metadata_json_; @@ -6765,12 +6775,6 @@ void MergeTreeData::exportPartToTable( throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); } - auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}); - - if (!part) - throw Exception(ErrorCodes::NO_SUCH_DATA_PART, "No such data part '{}' to export in table '{}'", - part_name, getStorageID().getFullTableName()); - if (part->getState() == MergeTreeDataPartState::Outdated && !allow_outdated_parts) throw Exception( ErrorCodes::BAD_ARGUMENTS, diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 8564eebda3ef..b069343527da 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -676,10 +676,10 @@ Poco::JSON::Object::Ptr getPartitionField( field = identifier->name(); } const auto * literal = expression_list_child->as(); - if (literal) - { + /// A String literal is an optional timezone argument (e.g. toRelativeDayNum(col, 'UTC')) and + /// does not affect the transform; only an integer literal is the bucket/truncate width. + if (literal && literal->value.getType() != Field::Types::String) param = literal->value.safeGet(); - } } } if (!field) diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index fe72425a36d5..c93960783f49 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -8565,7 +8565,11 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & ExportPartitionUtils::verifyIcebergPartitionCompatibility( metadata_object, - src_snapshot->getPartitionKeyAST()); + src_snapshot, + destination_snapshot, + parts, + partition_id, + query_context); std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM oss.exceptions(std::ios::failbit); diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 383739b6b0c7..5ec8494303f2 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -770,21 +770,25 @@ def check_accepted(mt, iceberg, description): def test_partition_transform_compatibility_rejected(cluster): """ - Verify that mismatched partition specs are rejected with BAD_ARGUMENTS. + Verify that partition specs that cannot be exported are rejected with BAD_ARGUMENTS. + + Acceptance is data-dependent: a source partition must map to a single Iceberg partition. The + mismatch cases below therefore use data that makes the source partition span several + destination partitions (a single-row partition would be trivially single-valued and accepted). Cases covered: - 1. Compound field order reversed: MergeTree (year, region) vs Iceberg (region, year) - 2. Transform mismatch on same column: year-transform vs identity - 3. Bucket count mismatch: bucket[8] vs bucket[16] - 4. Truncate width mismatch: truncate[4] vs truncate[8] - 5. Field-count mismatch: 2-field MergeTree vs 1-field Iceberg - 6. Unsupported MergeTree expression (intDiv — not an Iceberg transform) + 1. Transform mismatch on the same column: year-transform source vs identity destination, where + the year partition contains several distinct dates. + 2. Bucket count mismatch: bucket[8] vs bucket[16] (bucket is non-monotonic, always structural). + 3. Truncate width mismatch: truncate[4] source vs truncate[8] destination, with values sharing + the 4-char prefix but differing within the first 8 chars. + 4. Unsupported MergeTree expression (intDiv) vs identity, with one bucket spanning several years. + 5. Destination partitions by a column that is not in the source partition key. """ node = cluster.instances["replica1"] uid = unique_suffix() def assert_rejected(mt, iceberg, description): - # The compatibility check fires synchronously; any partition ID works here. pid = first_partition_id(node, mt) error = node.query_and_get_error( f"ALTER TABLE {mt} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg}", @@ -794,53 +798,45 @@ def assert_rejected(mt, iceberg, description): f"[{description}] Expected BAD_ARGUMENTS, got: {error!r}" ) - # 1. Compound field order reversed - cols = "id Int64, year Int32, region String" - t = f"mt_rej_1_{uid}"; i = f"iceberg_rej_1_{uid}" - make_rmt(node, t, cols, "(year, region)") - node.query(f"INSERT INTO {t} VALUES (1, 2020, 'EU')") - make_iceberg_s3(node, i, cols, "(region, year)") - assert_rejected(t, i, "compound field order reversed") - - # 2. Transform mismatch: MergeTree year-transform, Iceberg identity on same Date col + # 1. year-transform source vs identity destination: a year partition holds several dates. cols = "id Int64, event_date Date" - t = f"mt_rej_2_{uid}"; i = f"iceberg_rej_2_{uid}" + t = f"mt_rej_1_{uid}"; i = f"iceberg_rej_1_{uid}" make_rmt(node, t, cols, "toYearNumSinceEpoch(event_date)") - node.query(f"INSERT INTO {t} VALUES (1, '2020-01-01')") + node.query(f"INSERT INTO {t} VALUES (1, '2020-01-01'), (2, '2020-12-31')") make_iceberg_s3(node, i, cols, "event_date") # identity, not year-transform - assert_rejected(t, i, "year-transform vs identity on same column") + assert_rejected(t, i, "year-transform source vs identity destination") - # 3. Bucket count mismatch: bucket[8] vs bucket[16] + # 2. Bucket count mismatch: bucket[8] vs bucket[16] cols = "id Int64, user_id Int64" - t = f"mt_rej_3_{uid}"; i = f"iceberg_rej_3_{uid}" + t = f"mt_rej_2_{uid}"; i = f"iceberg_rej_2_{uid}" make_rmt(node, t, cols, "icebergBucket(8, user_id)") node.query(f"INSERT INTO {t} VALUES (1, 42)") make_iceberg_s3(node, i, cols, "icebergBucket(16, user_id)") assert_rejected(t, i, "bucket[8] vs bucket[16]") - # 4. Truncate width mismatch: truncate[4] vs truncate[8] + # 3. Truncate width mismatch: values share the 4-char prefix but differ within 8 chars. cols = "id Int64, category String" - t = f"mt_rej_4_{uid}"; i = f"iceberg_rej_4_{uid}" + t = f"mt_rej_3_{uid}"; i = f"iceberg_rej_3_{uid}" make_rmt(node, t, cols, "icebergTruncate(4, category)") - node.query(f"INSERT INTO {t} VALUES (1, 'clickhouse')") + node.query(f"INSERT INTO {t} VALUES (1, 'clickhouse'), (2, 'clickfmt')") make_iceberg_s3(node, i, cols, "icebergTruncate(8, category)") - assert_rejected(t, i, "truncate[4] vs truncate[8]") + assert_rejected(t, i, "truncate[4] source vs truncate[8] destination") - # 5. Field-count mismatch: MergeTree has 2 fields, Iceberg has 1 - cols = "id Int64, year Int32, region String" - t = f"mt_rej_5_{uid}"; i = f"iceberg_rej_5_{uid}" - make_rmt(node, t, cols, "(year, region)") - node.query(f"INSERT INTO {t} VALUES (1, 2020, 'EU')") + # 4. Unsupported MergeTree expression vs identity: one intDiv bucket spans several years. + cols = "id Int64, year Int32" + t = f"mt_rej_4_{uid}"; i = f"iceberg_rej_4_{uid}" + make_rmt(node, t, cols, "intDiv(year, 100)") + node.query(f"INSERT INTO {t} VALUES (1, 2000), (2, 2099)") make_iceberg_s3(node, i, cols, "year") - assert_rejected(t, i, "2-field MergeTree vs 1-field Iceberg") + assert_rejected(t, i, "intDiv source vs identity destination") - # 6. Unsupported MergeTree expression: intDiv(year, 100) is not an Iceberg transform + # 5. Destination partitions by a column absent from the source partition key. cols = "id Int64, year Int32" - t = f"mt_rej_6_{uid}"; i = f"iceberg_rej_6_{uid}" - make_rmt(node, t, cols, "intDiv(year, 100)") + t = f"mt_rej_5_{uid}"; i = f"iceberg_rej_5_{uid}" + make_rmt(node, t, cols, "year") node.query(f"INSERT INTO {t} VALUES (1, 2020)") - make_iceberg_s3(node, i, cols, "year") - assert_rejected(t, i, "unsupported MergeTree expression intDiv") + make_iceberg_s3(node, i, cols, "id") # identity on id, which the source does not partition by + assert_rejected(t, i, "destination partitions by a non-source-key column") def test_partition_key_compatibility_check(cluster): @@ -928,6 +924,179 @@ def test_partition_key_compatibility_check(cluster): ) +def test_partition_transform_equivalence_gate(cluster): + """ + The Iceberg partition-compatibility gate accepts a source partition key whose transform is + equivalent to (or finer than) the destination Iceberg transform when the exported partition is + provably single-valued for every destination field, and rejects it otherwise. The gate runs + synchronously at EXPORT time: acceptance means the ALTER does not throw, rejection means it + throws BAD_ARGUMENTS. + """ + node = cluster.instances["replica1"] + settings = {"allow_insert_into_iceberg": 1} + + def run_case(name, columns, source_key, dest_key, rows, *, expect_ok): + uid = unique_suffix() + mt_table = f"mt_{name}_{uid}" + iceberg_table = f"iceberg_{name}_{uid}" + make_rmt(node, mt_table, columns, source_key, replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES {rows}") + make_iceberg_s3(node, iceberg_table, columns, partition_by=dest_key) + pid = first_partition_id(node, mt_table) + statement = f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}" + if expect_ok: + node.query(statement, settings=settings) + else: + error = node.query_and_get_error(statement, settings=settings) + assert "BAD_ARGUMENTS" in error, f"{name}: expected BAD_ARGUMENTS, got: {error!r}" + + dt = "id Int64, event_time DateTime" + + # toDate -> day: rows within one day map to a single Iceberg day partition. + run_case("todate_day", dt, "toDate(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')", expect_ok=True) + + # toYYYYMM -> month: different days of the same month map to a single month partition. + run_case("toyyyymm_month", dt, "toYYYYMM(event_time)", "toMonthNumSinceEpoch(event_time)", + "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", expect_ok=True) + + # toStartOfHour -> hour. + run_case("startofhour_hour", dt, "toStartOfHour(event_time)", "toRelativeHourNum(event_time)", + "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:59:00')", expect_ok=True) + + # Finer source (day + country) into a day-partitioned destination: the extra column is allowed. + run_case("finer_day", "id Int64, event_time DateTime, country String", + "(toDate(event_time), country)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 01:00:00', 'US'), (2, '2024-03-05 20:00:00', 'US')", expect_ok=True) + + # Compound field order reversed: matching is by column, so (year, region) into (region, year) + # is accepted - the destination spec defines the written tuple order. + run_case("reversed_order", "id Int64, year Int32, region String", + "(year, region)", "(region, year)", "(1, 2020, 'EU')", expect_ok=True) + + # Superset source: a (year, region) source into a year-only destination is finer, so accepted. + run_case("superset", "id Int64, year Int32, region String", + "(year, region)", "year", "(1, 2020, 'EU')", expect_ok=True) + + # Coarser source: a month partition spans several days, so it cannot map to one day partition. + run_case("coarser_day", dt, "toYYYYMM(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", expect_ok=False) + + # bucket is non-monotonic: an identity source cannot satisfy a bucket destination via min/max. + run_case("bucket_needs_structural", "id Int64, k Int64", "k", "icebergBucket(8, k)", + "(1, 10), (2, 10)", expect_ok=False) + + +def test_export_partition_todate_source_matches_day_metadata(cluster): + """ + End-to-end: a source partitioned by toDate(event_time) exports into a day-partitioned Iceberg + table through the min/max refinement, and the day value written to the Iceberg metadata matches + the exported data. + """ + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_todate_{uid}" + iceberg_table = f"iceberg_todate_{uid}" + + make_rmt(node, mt_table, "id Int64, event_time DateTime", "toDate(event_time)", + replica_name="replica1") + node.query( + f"INSERT INTO {mt_table} VALUES " + f"(1, '2024-03-05 01:00:00'), (2, '2024-03-05 12:00:00'), (3, '2024-03-05 23:00:00')" + ) + make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime", + partition_by="toRelativeDayNum(event_time)") + + pid = first_partition_id(node, mt_table) + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 3, f"Expected 3 rows after export, got {count}" + + expected_day = int(node.query( + f"SELECT DISTINCT toRelativeDayNum(event_time) FROM {iceberg_table}" + ).strip()) + + query_id = f"todate_{uid}" + node.query( + f"SELECT * FROM {iceberg_table}", + query_id=query_id, + settings={"iceberg_metadata_log_level": "manifest_file_entry"}, + ) + entries = fetch_manifest_entries(node, query_id) + partitions = _data_file_partition_records(entries) + assert partitions, "No data-file partition records found in manifest entries" + meta_days = {int(_partition_scalar(p, "event_time")) for p in partitions} + assert meta_days == {expected_day}, ( + f"Metadata day {meta_days} must equal toRelativeDayNum {expected_day}." + ) + + +def test_export_partition_timezone_literal_partition_key(cluster): + """ + A timezone argument in the partition key (toRelativeDayNum(event_time, 'UTC')) must not break + Iceberg table creation or the export compatibility gate; previously it threw a `Bad get`. + """ + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_tz_{uid}" + iceberg_table = f"iceberg_tz_{uid}" + + make_rmt(node, mt_table, "id Int64, event_time DateTime", "toRelativeDayNum(event_time, 'UTC')", + replica_name="replica1") + node.query( + f"INSERT INTO {mt_table} VALUES (1, '2024-03-05 01:00:00'), (2, '2024-03-05 23:00:00')" + ) + # Creating the destination with the same timezone-qualified key exercises the getPartitionField fix. + make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime", + partition_by="toRelativeDayNum(event_time, 'UTC')") + + pid = first_partition_id(node, mt_table) + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + assert int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) == 2 + + +def test_export_partition_lossy_cast_dynamic_accept(cluster): + """ + A lossy Int64 -> Int32 partition-column cast is accepted by the dynamic proof when the + partition's values fit the destination type and map to a single Iceberg bucket. Source and + destination use different truncate widths, so the field is proven via min/max rather than a + structural match. + """ + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_lossy_{uid}" + iceberg_table = f"iceberg_lossy_{uid}" + + make_rmt(node, mt_table, "id Int64, val Int64", "icebergTruncate(10, val)", + replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, 100), (2, 109)") + make_iceberg_s3(node, iceberg_table, "id Int64, val Int32", + partition_by="icebergTruncate(1000000, val)") + + pid = first_partition_id(node, mt_table) + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={ + "allow_insert_into_iceberg": 1, + "export_merge_tree_part_allow_lossy_cast": 1, + }, + ) + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + assert int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) == 2 + + def test_export_data_files_are_not_cleaned_up_on_commit_failure(cluster): """ Verify that a commit failure does not delete the already-written data files. From 343c2d9162f6df7cf50f7794cc932f9f792fa39a Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Fri, 17 Jul 2026 17:36:31 -0300 Subject: [PATCH 2/8] add some more tests --- .../test.py | 117 ++++++++++++++++++ 1 file changed, 117 insertions(+) diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 5ec8494303f2..6c7f8d4d8cc8 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -987,6 +987,73 @@ def run_case(name, columns, source_key, dest_key, rows, *, expect_ok): "(1, 10), (2, 10)", expect_ok=False) +def test_partition_transform_granularity_matrix(cluster): + """ + Exercise the common ClickHouse temporal partition keys and the granularity relationships between + the source key and the destination Iceberg transform. Acceptance is data-dependent (a source + partition must be single-valued for every destination field), so a coarser source can still be + accepted when a particular partition does not actually repartition. + """ + node = cluster.instances["replica1"] + settings = {"allow_insert_into_iceberg": 1} + + def run_case(name, source_key, dest_key, rows, *, expect_ok): + uid = unique_suffix() + mt_table = f"mt_{name}_{uid}" + iceberg_table = f"iceberg_{name}_{uid}" + make_rmt(node, mt_table, "id Int64, event_time DateTime", source_key, replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES {rows}") + make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime", partition_by=dest_key) + pid = first_partition_id(node, mt_table) + statement = f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}" + if expect_ok: + node.query(statement, settings=settings) + else: + error = node.query_and_get_error(statement, settings=settings) + assert "BAD_ARGUMENTS" in error, f"{name}: expected BAD_ARGUMENTS, got: {error!r}" + + same_day = "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')" + same_month = "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')" + same_year = "(1, '2024-03-05 00:00:00'), (2, '2024-09-10 00:00:00')" + + # Common temporal keys at the same granularity as the destination transform. + run_case("startofmonth_month", "toStartOfMonth(event_time)", "toMonthNumSinceEpoch(event_time)", + same_month, expect_ok=True) + run_case("yyyymmdd_day", "toYYYYMMDD(event_time)", "toRelativeDayNum(event_time)", + same_day, expect_ok=True) + run_case("startofday_day", "toStartOfDay(event_time)", "toRelativeDayNum(event_time)", + same_day, expect_ok=True) + run_case("toyear_year", "toYear(event_time)", "toYearNumSinceEpoch(event_time)", + same_year, expect_ok=True) + run_case("startofyear_year", "toStartOfYear(event_time)", "toYearNumSinceEpoch(event_time)", + same_year, expect_ok=True) + + # Finer source into a coarser destination: a finer partition is contained in one coarser bucket. + run_case("day_into_month", "toDate(event_time)", "toMonthNumSinceEpoch(event_time)", + same_day, expect_ok=True) + run_case("day_into_year", "toDate(event_time)", "toYearNumSinceEpoch(event_time)", + same_day, expect_ok=True) + run_case("hour_into_day", "toStartOfHour(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:30:00')", expect_ok=True) + run_case("month_into_year", "toYYYYMM(event_time)", "toYearNumSinceEpoch(event_time)", + same_month, expect_ok=True) + + # Coarser source into a finer destination: the partition spans several destination buckets. + run_case("year_into_month", "toYear(event_time)", "toMonthNumSinceEpoch(event_time)", + "(1, '2020-01-15 00:00:00'), (2, '2020-06-15 00:00:00')", expect_ok=False) + run_case("year_into_day", "toYear(event_time)", "toRelativeDayNum(event_time)", + "(1, '2020-01-01 00:00:00'), (2, '2020-12-31 00:00:00')", expect_ok=False) + + # Same coarse source/fine destination pair as above, but this year partition holds a single day, + # so it does not repartition and is accepted - acceptance depends on the data, not the structure. + run_case("year_into_day_single_day", "toYear(event_time)", "toRelativeDayNum(event_time)", + same_day, expect_ok=True) + + # Weekly has no Iceberg equivalent: a week partition holding two days cannot map to one day. + run_case("week_into_day", "toMonday(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 00:00:00'), (2, '2024-03-07 00:00:00')", expect_ok=False) + + def test_export_partition_todate_source_matches_day_metadata(cluster): """ End-to-end: a source partitioned by toDate(event_time) exports into a day-partitioned Iceberg @@ -1037,6 +1104,56 @@ def test_export_partition_todate_source_matches_day_metadata(cluster): ) +def test_export_partition_day_source_into_year_metadata(cluster): + """ + End-to-end: a source partitioned by toDate(event_time) (finer) exports into a year-partitioned + Iceberg destination (coarser). The value written to the Iceberg metadata is the year computed by + the destination transform over the data, not the source day. + """ + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_day_year_{uid}" + iceberg_table = f"iceberg_day_year_{uid}" + + make_rmt(node, mt_table, "id Int64, event_time DateTime", "toDate(event_time)", + replica_name="replica1") + node.query( + f"INSERT INTO {mt_table} VALUES " + f"(1, '2024-03-05 01:00:00'), (2, '2024-03-05 12:00:00'), (3, '2024-03-05 23:00:00')" + ) + make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime", + partition_by="toYearNumSinceEpoch(event_time)") + + pid = first_partition_id(node, mt_table) + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 3, f"Expected 3 rows after export, got {count}" + + expected_year = int(node.query( + f"SELECT DISTINCT toYearNumSinceEpoch(event_time) FROM {iceberg_table}" + ).strip()) + + query_id = f"day_year_{uid}" + node.query( + f"SELECT * FROM {iceberg_table}", + query_id=query_id, + settings={"iceberg_metadata_log_level": "manifest_file_entry"}, + ) + entries = fetch_manifest_entries(node, query_id) + partitions = _data_file_partition_records(entries) + assert partitions, "No data-file partition records found in manifest entries" + meta_years = {int(_partition_scalar(p, "event_time")) for p in partitions} + assert meta_years == {expected_year}, ( + f"Metadata year {meta_years} must equal toYearNumSinceEpoch {expected_year}." + ) + + def test_export_partition_timezone_literal_partition_key(cluster): """ A timezone argument in the partition key (toRelativeDayNum(event_time, 'UTC')) must not break From 3f3af7c4fc6ebb9aece7ef0e2b14211280d97cd9 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Sat, 18 Jul 2026 14:34:35 -0300 Subject: [PATCH 3/8] make tests better --- .../test.py | 298 +++++++++++------- 1 file changed, 190 insertions(+), 108 deletions(-) diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 6c7f8d4d8cc8..601aee695222 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -928,63 +928,47 @@ def test_partition_transform_equivalence_gate(cluster): """ The Iceberg partition-compatibility gate accepts a source partition key whose transform is equivalent to (or finer than) the destination Iceberg transform when the exported partition is - provably single-valued for every destination field, and rejects it otherwise. The gate runs - synchronously at EXPORT time: acceptance means the ALTER does not throw, rejection means it - throws BAD_ARGUMENTS. + provably single-valued for every destination field, and rejects it otherwise. Accept cases are + verified end-to-end (data + metadata); reject cases must throw BAD_ARGUMENTS synchronously. """ node = cluster.instances["replica1"] - settings = {"allow_insert_into_iceberg": 1} - - def run_case(name, columns, source_key, dest_key, rows, *, expect_ok): - uid = unique_suffix() - mt_table = f"mt_{name}_{uid}" - iceberg_table = f"iceberg_{name}_{uid}" - make_rmt(node, mt_table, columns, source_key, replica_name="replica1") - node.query(f"INSERT INTO {mt_table} VALUES {rows}") - make_iceberg_s3(node, iceberg_table, columns, partition_by=dest_key) - pid = first_partition_id(node, mt_table) - statement = f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}" - if expect_ok: - node.query(statement, settings=settings) - else: - error = node.query_and_get_error(statement, settings=settings) - assert "BAD_ARGUMENTS" in error, f"{name}: expected BAD_ARGUMENTS, got: {error!r}" - dt = "id Int64, event_time DateTime" - - # toDate -> day: rows within one day map to a single Iceberg day partition. - run_case("todate_day", dt, "toDate(event_time)", "toRelativeDayNum(event_time)", - "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')", expect_ok=True) - - # toYYYYMM -> month: different days of the same month map to a single month partition. - run_case("toyyyymm_month", dt, "toYYYYMM(event_time)", "toMonthNumSinceEpoch(event_time)", - "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", expect_ok=True) - - # toStartOfHour -> hour. - run_case("startofhour_hour", dt, "toStartOfHour(event_time)", "toRelativeHourNum(event_time)", - "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:59:00')", expect_ok=True) - - # Finer source (day + country) into a day-partitioned destination: the extra column is allowed. - run_case("finer_day", "id Int64, event_time DateTime, country String", - "(toDate(event_time), country)", "toRelativeDayNum(event_time)", - "(1, '2024-03-05 01:00:00', 'US'), (2, '2024-03-05 20:00:00', 'US')", expect_ok=True) - - # Compound field order reversed: matching is by column, so (year, region) into (region, year) - # is accepted - the destination spec defines the written tuple order. - run_case("reversed_order", "id Int64, year Int32, region String", - "(year, region)", "(region, year)", "(1, 2020, 'EU')", expect_ok=True) - - # Superset source: a (year, region) source into a year-only destination is finer, so accepted. - run_case("superset", "id Int64, year Int32, region String", - "(year, region)", "year", "(1, 2020, 'EU')", expect_ok=True) - - # Coarser source: a month partition spans several days, so it cannot map to one day partition. - run_case("coarser_day", dt, "toYYYYMM(event_time)", "toRelativeDayNum(event_time)", - "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", expect_ok=False) - - # bucket is non-monotonic: an identity source cannot satisfy a bucket destination via min/max. - run_case("bucket_needs_structural", "id Int64, k Int64", "k", "icebergBucket(8, k)", - "(1, 10), (2, 10)", expect_ok=False) + yr = "id Int64, year Int32, region String" + + cases = [ + # toDate -> day: rows within one day map to a single Iceberg day partition. + {"name": "todate_day", "columns": dt, "source_key": "toDate(event_time)", + "dest_key": "toRelativeDayNum(event_time)", + "rows": "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')", "expect_ok": True}, + # toYYYYMM -> month: different days of the same month map to a single month partition. + {"name": "toyyyymm_month", "columns": dt, "source_key": "toYYYYMM(event_time)", + "dest_key": "toMonthNumSinceEpoch(event_time)", + "rows": "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", "expect_ok": True}, + # toStartOfHour -> hour. + {"name": "startofhour_hour", "columns": dt, "source_key": "toStartOfHour(event_time)", + "dest_key": "toRelativeHourNum(event_time)", + "rows": "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:59:00')", "expect_ok": True}, + # Finer source (day + country) into a day-partitioned destination: extra column allowed. + {"name": "finer_day", "columns": "id Int64, event_time DateTime, country String", + "source_key": "(toDate(event_time), country)", "dest_key": "toRelativeDayNum(event_time)", + "rows": "(1, '2024-03-05 01:00:00', 'US'), (2, '2024-03-05 20:00:00', 'US')", + "expect_ok": True}, + # Compound field order reversed: matching is by column; the destination defines tuple order. + {"name": "reversed_order", "columns": yr, "source_key": "(year, region)", + "dest_key": "(region, year)", "rows": "(1, 2020, 'EU')", "expect_ok": True, + "verify": [("region", "region"), ("year", "year")]}, + # Superset source: (year, region) into a year-only destination is finer, so accepted. + {"name": "superset", "columns": yr, "source_key": "(year, region)", "dest_key": "year", + "rows": "(1, 2020, 'EU')", "expect_ok": True, "verify": [("year", "year")]}, + # Coarser source: a month partition spans several days, so it cannot map to one day. + {"name": "coarser_day", "columns": dt, "source_key": "toYYYYMM(event_time)", + "dest_key": "toRelativeDayNum(event_time)", + "rows": "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')", "expect_ok": False}, + # bucket is non-monotonic: an identity source cannot satisfy a bucket destination. + {"name": "bucket_needs_structural", "columns": "id Int64, k Int64", "source_key": "k", + "dest_key": "icebergBucket(8, k)", "rows": "(1, 10), (2, 10)", "expect_ok": False}, + ] + run_partition_compat_cases(node, cases) def test_partition_transform_granularity_matrix(cluster): @@ -992,66 +976,75 @@ def test_partition_transform_granularity_matrix(cluster): Exercise the common ClickHouse temporal partition keys and the granularity relationships between the source key and the destination Iceberg transform. Acceptance is data-dependent (a source partition must be single-valued for every destination field), so a coarser source can still be - accepted when a particular partition does not actually repartition. + accepted when a particular partition does not actually repartition. Accept cases are verified + end-to-end (data + metadata); reject cases must throw BAD_ARGUMENTS. """ node = cluster.instances["replica1"] - settings = {"allow_insert_into_iceberg": 1} - - def run_case(name, source_key, dest_key, rows, *, expect_ok): - uid = unique_suffix() - mt_table = f"mt_{name}_{uid}" - iceberg_table = f"iceberg_{name}_{uid}" - make_rmt(node, mt_table, "id Int64, event_time DateTime", source_key, replica_name="replica1") - node.query(f"INSERT INTO {mt_table} VALUES {rows}") - make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime", partition_by=dest_key) - pid = first_partition_id(node, mt_table) - statement = f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}" - if expect_ok: - node.query(statement, settings=settings) - else: - error = node.query_and_get_error(statement, settings=settings) - assert "BAD_ARGUMENTS" in error, f"{name}: expected BAD_ARGUMENTS, got: {error!r}" - + dt = "id Int64, event_time DateTime" same_day = "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')" same_month = "(1, '2024-03-01 00:00:00'), (2, '2024-03-20 00:00:00')" same_year = "(1, '2024-03-05 00:00:00'), (2, '2024-09-10 00:00:00')" - # Common temporal keys at the same granularity as the destination transform. - run_case("startofmonth_month", "toStartOfMonth(event_time)", "toMonthNumSinceEpoch(event_time)", - same_month, expect_ok=True) - run_case("yyyymmdd_day", "toYYYYMMDD(event_time)", "toRelativeDayNum(event_time)", - same_day, expect_ok=True) - run_case("startofday_day", "toStartOfDay(event_time)", "toRelativeDayNum(event_time)", - same_day, expect_ok=True) - run_case("toyear_year", "toYear(event_time)", "toYearNumSinceEpoch(event_time)", - same_year, expect_ok=True) - run_case("startofyear_year", "toStartOfYear(event_time)", "toYearNumSinceEpoch(event_time)", - same_year, expect_ok=True) - - # Finer source into a coarser destination: a finer partition is contained in one coarser bucket. - run_case("day_into_month", "toDate(event_time)", "toMonthNumSinceEpoch(event_time)", - same_day, expect_ok=True) - run_case("day_into_year", "toDate(event_time)", "toYearNumSinceEpoch(event_time)", - same_day, expect_ok=True) - run_case("hour_into_day", "toStartOfHour(event_time)", "toRelativeDayNum(event_time)", - "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:30:00')", expect_ok=True) - run_case("month_into_year", "toYYYYMM(event_time)", "toYearNumSinceEpoch(event_time)", - same_month, expect_ok=True) - - # Coarser source into a finer destination: the partition spans several destination buckets. - run_case("year_into_month", "toYear(event_time)", "toMonthNumSinceEpoch(event_time)", - "(1, '2020-01-15 00:00:00'), (2, '2020-06-15 00:00:00')", expect_ok=False) - run_case("year_into_day", "toYear(event_time)", "toRelativeDayNum(event_time)", - "(1, '2020-01-01 00:00:00'), (2, '2020-12-31 00:00:00')", expect_ok=False) - - # Same coarse source/fine destination pair as above, but this year partition holds a single day, - # so it does not repartition and is accepted - acceptance depends on the data, not the structure. - run_case("year_into_day_single_day", "toYear(event_time)", "toRelativeDayNum(event_time)", - same_day, expect_ok=True) - - # Weekly has no Iceberg equivalent: a week partition holding two days cannot map to one day. - run_case("week_into_day", "toMonday(event_time)", "toRelativeDayNum(event_time)", - "(1, '2024-03-05 00:00:00'), (2, '2024-03-07 00:00:00')", expect_ok=False) + def case(name, source_key, dest_key, rows, expect_ok): + return {"name": name, "columns": dt, "source_key": source_key, "dest_key": dest_key, + "rows": rows, "expect_ok": expect_ok} + + cases = [ + # Common temporal keys at the same granularity as the destination transform. + case("startofmonth_month", "toStartOfMonth(event_time)", "toMonthNumSinceEpoch(event_time)", same_month, True), + case("yyyymmdd_day", "toYYYYMMDD(event_time)", "toRelativeDayNum(event_time)", same_day, True), + case("startofday_day", "toStartOfDay(event_time)", "toRelativeDayNum(event_time)", same_day, True), + case("toyear_year", "toYear(event_time)", "toYearNumSinceEpoch(event_time)", same_year, True), + case("startofyear_year", "toStartOfYear(event_time)", "toYearNumSinceEpoch(event_time)", same_year, True), + # Finer source into a coarser destination: a finer partition sits inside one coarser bucket. + case("day_into_month", "toDate(event_time)", "toMonthNumSinceEpoch(event_time)", same_day, True), + case("day_into_year", "toDate(event_time)", "toYearNumSinceEpoch(event_time)", same_day, True), + case("hour_into_day", "toStartOfHour(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 12:00:00'), (2, '2024-03-05 12:30:00')", True), + case("month_into_year", "toYYYYMM(event_time)", "toYearNumSinceEpoch(event_time)", same_month, True), + # Coarser source into a finer destination: the partition spans several destination buckets. + case("year_into_month", "toYear(event_time)", "toMonthNumSinceEpoch(event_time)", + "(1, '2020-01-15 00:00:00'), (2, '2020-06-15 00:00:00')", False), + case("year_into_day", "toYear(event_time)", "toRelativeDayNum(event_time)", + "(1, '2020-01-01 00:00:00'), (2, '2020-12-31 00:00:00')", False), + # Same coarse/fine pair, but this year partition holds a single day, so it does not + # repartition and is accepted - acceptance depends on the data, not the structure. + case("year_into_day_single_day", "toYear(event_time)", "toRelativeDayNum(event_time)", same_day, True), + # Weekly has no Iceberg equivalent: a week partition holding two days cannot map to one day. + case("week_into_day", "toMonday(event_time)", "toRelativeDayNum(event_time)", + "(1, '2024-03-05 00:00:00'), (2, '2024-03-07 00:00:00')", False), + ] + run_partition_compat_cases(node, cases) + + +def test_partition_multicolumn_subset(cluster): + """ + Destination partition columns must be a subset of the source partition-key columns. A wide + source whose partition key is a superset of the destination's is accepted (and its multi-column + data plus per-field metadata verified); a destination partitioning by a column absent from the + source partition key is rejected. + """ + node = cluster.instances["replica1"] + wide = "id Int64, event_time DateTime, region String, tenant Int32, v1 Float64, v2 String" + + cases = [ + # Destination partition columns {event_time, region} are a strict subset of the source's + # {event_time, region, tenant}: accepted, with multi-column data and per-field metadata. + {"name": "subset_ok", "columns": wide, + "source_key": "(toDate(event_time), region, tenant)", + "dest_key": "(toRelativeDayNum(event_time), region)", + "rows": "(1, '2024-03-05 01:00:00', 'US', 7, 1.5, 'a'), " + "(2, '2024-03-05 20:00:00', 'US', 7, 2.5, 'b')", + "expect_ok": True, + "verify": [("event_time", "toRelativeDayNum(event_time)"), ("region", "region")]}, + # Destination partitions by 'region', which is not in the source partition key: rejected. + {"name": "not_subset", "columns": "id Int64, event_time DateTime, region String", + "source_key": "toDate(event_time)", + "dest_key": "(toRelativeDayNum(event_time), region)", + "rows": "(1, '2024-03-05 01:00:00', 'US'), (2, '2024-03-05 20:00:00', 'EU')", + "expect_ok": False}, + ] + run_partition_compat_cases(node, cases) def test_export_partition_todate_source_matches_day_metadata(cluster): @@ -1869,6 +1862,95 @@ def _partition_scalar(partition, field): return value +def assert_iceberg_partition_metadata(node, iceberg_table, uid, fields): + """Assert every data-file partition record's field equals the single DISTINCT value of the + corresponding expression over the exported destination data. `fields` is a list of + (metadata_field_name, value_expr). String-normalized so integer transforms and identity + string/int fields compare uniformly.""" + query_id = f"verify_{uid}" + node.query( + f"SELECT * FROM {iceberg_table}", + query_id=query_id, + settings={"iceberg_metadata_log_level": "manifest_file_entry"}, + ) + entries = fetch_manifest_entries(node, query_id) + partitions = _data_file_partition_records(entries) + assert partitions, "No data-file partition records found in manifest entries" + for field_name, value_expr in fields: + expected = node.query( + f"SELECT DISTINCT toString({value_expr}) FROM {iceberg_table}" + ).strip() + got = {str(_partition_scalar(p, field_name)) for p in partitions} + assert got == {expected}, ( + f"metadata field {field_name!r} = {got}, expected {{{expected!r}}}" + ) + + +def run_partition_compat_cases(node, cases): + """Run partition-compatibility cases against the Iceberg export gate. + + Reject cases (``expect_ok=False``) are checked synchronously - the gate fires while scheduling, + so the ALTER throws immediately. Accept cases are dispatched together, then awaited, then their + data (full ordered row comparison against the exported source partition) and Iceberg partition + metadata are verified. Each case is a dict: name, columns, source_key, dest_key, rows, expect_ok, + and optional verify (list of (metadata_field_name, value_expr); defaults to + [("event_time", dest_key)]).""" + settings = {"allow_insert_into_iceberg": 1} + + def setup(case): + uid = unique_suffix() + mt_table = f"mt_{case['name']}_{uid}" + iceberg_table = f"iceberg_{case['name']}_{uid}" + make_rmt(node, mt_table, case["columns"], case["source_key"], replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES {case['rows']}") + make_iceberg_s3(node, iceberg_table, case["columns"], partition_by=case["dest_key"]) + pid = first_partition_id(node, mt_table) + return uid, mt_table, iceberg_table, pid + + for case in cases: + if case["expect_ok"]: + continue + _uid, mt_table, iceberg_table, pid = setup(case) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings=settings, + ) + assert "BAD_ARGUMENTS" in error, f"{case['name']}: expected BAD_ARGUMENTS, got: {error!r}" + + dispatched = [] + for case in cases: + if not case["expect_ok"]: + continue + uid, mt_table, iceberg_table, pid = setup(case) + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings=settings, + ) + dispatched.append((case, uid, mt_table, iceberg_table, pid)) + + for case, uid, mt_table, iceberg_table, pid in dispatched: + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + + for case, uid, mt_table, iceberg_table, pid in dispatched: + # Export is a positional cast into the destination schema, so verify the destination equals + # the source cast into the destination column types. Normalizing to the destination types + # tolerates legitimate Iceberg type promotion (e.g. DateTime is stored as a microsecond + # timestamp and returns as DateTime64(6)) while preserving destination precision, so a + # spurious sub-second value would still surface as a mismatch. + col_defs = node.query( + f"SELECT name, type FROM system.columns " + f"WHERE database = currentDatabase() AND table = '{iceberg_table}' ORDER BY position" + ).strip().split("\n") + projection = ", ".join( + f"CAST({name} AS {ctype})" for name, ctype in (c.split("\t") for c in col_defs) + ) + src = node.query(f"SELECT {projection} FROM {mt_table} ORDER BY id") + dst = node.query(f"SELECT {projection} FROM {iceberg_table} ORDER BY id") + assert src == dst, f"{case['name']}: destination rows differ from source" + fields = case.get("verify") or [("event_time", case["dest_key"])] + assert_iceberg_partition_metadata(node, iceberg_table, f"{case['name']}_{uid}", fields) + + def test_export_partition_bucket_transform_metadata_matches_data(cluster): """A bucket[N] partition column whose type changes Int64 -> String records the destination murmur(String) bucket in the Iceberg metadata, matching the exported From d385268cc04b7e31c0517a6689cd877020772a81 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Mon, 20 Jul 2026 09:48:56 -0300 Subject: [PATCH 4/8] more vibe coded changes --- docs/en/antalya/partition_export.md | 9 + .../MergeTree/ExportPartitionUtils.cpp | 260 ++++++++++++++++-- src/Storages/MergeTree/ExportPartitionUtils.h | 15 + src/Storages/MergeTree/MergeTreeData.cpp | 44 +-- src/Storages/StorageReplicatedMergeTree.cpp | 25 +- .../test.py | 148 ++++++++++ .../test_export_partition_iceberg.py | 49 +--- ...rge_tree_part_to_object_storage_simple.sql | 23 +- 8 files changed, 465 insertions(+), 108 deletions(-) diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 5ef422426387..b75c3085f683 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -41,6 +41,15 @@ Lossy partition-column casts (allowed via `export_merge_tree_part_allow_lossy_ca Each MergeTree part will become a separate file with the following name convention: `//_.`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `/commit__`. +#### Source partition key compatibility + +The export writes all rows of a part to the single directory computed from the destination `PARTITION BY` on the part's values, so - exactly like the Iceberg gate - each destination partition must be single-valued across the exported source part. A destination partition column is accepted when either: + +- The whole source and destination `PARTITION BY` are identical, or the source already partitions by the same expression on that column (this covers a source that adds extra partition columns on top of the destination's, in any order - for example `PARTITION BY (year, country)` into a destination partitioned by `year`). +- Or the destination expression is proven constant over the column's actual `[min, max]` in the part. This is data-dependent and accepts equivalent or finer source keys (for example `PARTITION BY toDate(ts)` into a destination partitioned by `toYYYYMM(ts)` when a part holds a single month). Only provably-monotonic single-argument functions and bare columns are proven this way. + +Otherwise the export is rejected with a `BAD_ARGUMENTS` error at `EXPORT` time: this includes a destination that partitions by a column absent from the source partition key, and a source partition whose rows would span more than one destination partition (for example a monthly source partition exported into a daily destination that actually contains several days). A `Nullable` partition column is only accepted through an exact match, because a `NULL` forms its own partition. Partition-column type differences follow the same lossy-cast gate as any other column (`export_merge_tree_part_allow_lossy_cast`). + ## Syntax ```sql diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 4efc8374840b..7c28ce895e0f 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -14,22 +14,22 @@ #include #include #include +#include #include #include #include - -#if USE_AVRO #include -#include -#include #include #include #include #include #include + +#if USE_AVRO +#include +#include #include #include -#include #include #endif @@ -495,46 +495,51 @@ namespace ExportPartitionUtils ops.emplace_back(zkutil::makeSetRequest(last_exception_path, entry.toJsonString(), -1)); } -#if USE_AVRO namespace { - /// One top-level term of the source MergeTree PARTITION BY, canonicalized for comparison against a - /// destination Iceberg partition field. `transform` holds the ClickHouse function name that - /// parseTransformAndArgument yields for the equivalent Iceberg transform ("toRelativeDayNum", - /// "identity", "icebergBucket", ...); std::nullopt marks a term that is not directly - /// Iceberg-representable and can therefore only satisfy a destination field via the dynamic proof. - struct SourcePartitionTerm + /// One top-level term of a MergeTree PARTITION BY: a column and, when the term is a function of a + /// single column, that function name and an optional integer argument. An empty function marks a + /// bare column (identity). + struct PartitionKeyTerm { String column; - std::optional transform; + String function; std::optional argument; + + bool operator==(const PartitionKeyTerm & other) const + { + return column == other.column && function == other.function && argument == other.argument; + } }; - std::vector parseSourcePartitionTerms(const ASTPtr & partition_key_ast) + /// Parse a PARTITION BY AST into per-column terms. Returns std::nullopt if any top-level term is + /// not a bare column or a function of exactly one column (nested/multi-column expressions), so the + /// caller can only satisfy such a key by an exact match. + std::optional> parsePartitionKeyTerms(const ASTPtr & partition_key_ast) { - /// The same functions getPartitionField maps 1:1 to an Iceberg transform. - static const std::unordered_set iceberg_representable = { - "identity", "toYearNumSinceEpoch", "toMonthNumSinceEpoch", - "toRelativeDayNum", "toRelativeHourNum", "icebergBucket", "icebergTruncate"}; - - std::vector terms; + std::vector terms; if (!partition_key_ast) return terms; + bool ok = true; auto parse_one = [&](const ASTPtr & element) { if (const auto * ident = element->as()) { - terms.push_back({ident->name(), std::optional("identity"), std::nullopt}); + terms.push_back({ident->name(), "", std::nullopt}); return; } const auto * func = element->as(); if (!func) + { + ok = false; return; + } std::optional column; std::optional argument; + size_t identifiers = 0; for (const auto & child : func->children) { const auto * expression_list = child->as(); @@ -543,18 +548,27 @@ namespace for (const auto & arg : expression_list->children) { if (const auto * id = arg->as()) + { column = id->name(); - else if (const auto * lit = arg->as(); lit && lit->value.getType() != Field::Types::String) - argument = lit->value.safeGet(); + ++identifiers; + } + else if (const auto * lit = arg->as()) + { + /// A string literal is an optional timezone argument and does not affect the + /// term; only an integer literal is the bucket/truncate width. + if (lit->value.getType() != Field::Types::String) + argument = lit->value.safeGet(); + } + else + ok = false; } } - if (!column) + if (identifiers != 1) + { + ok = false; return; - - std::optional transform; - if (iceberg_representable.contains(func->name)) - transform = func->name; - terms.push_back({*column, transform, argument}); + } + terms.push_back({*column, func->name, argument}); }; if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") @@ -567,6 +581,8 @@ namespace else parse_one(partition_key_ast); + if (!ok) + return std::nullopt; return terms; } @@ -599,6 +615,121 @@ namespace return found; } + /// Whether the destination partition term yields a single value across [min, max] of its source + /// column. A bare column is single-valued iff min == max. A single-argument function is + /// single-valued when it is provably monotonic over the range and maps both endpoints to the same + /// value. Functions carrying an integer argument (bucket/truncate and similar) are not order + /// preserving in general, so they can only be accepted by an exact structural match. + bool partitionTermIsConstantOverBounds( + const PartitionKeyTerm & term, const DataTypePtr & column_type, + const Field & min_value, const Field & max_value, const ContextPtr & context) + { + if (term.function.empty() || term.function == "identity") + return min_value == max_value; + + if (term.argument.has_value()) + return false; + + auto resolver = FunctionFactory::instance().get(term.function, context); + + auto values_column = column_type->createColumn(); + values_column->insert(min_value); + values_column->insert(max_value); + const ColumnWithTypeAndName values{std::move(values_column), column_type, term.column}; + + const ColumnsWithTypeAndName arguments{values}; + const auto result_type = resolver->getReturnType(arguments); + const auto function_base = resolver->build(arguments); + + if (!function_base->hasInformationAboutMonotonicity()) + return false; + if (!function_base->getMonotonicityForRange(*column_type, min_value, max_value).is_monotonic) + return false; + + const auto result = function_base->execute(arguments, result_type, 2, /*dry_run=*/ false); + Field first; + Field second; + result->get(0, first); + result->get(1, second); + return first == second; + } +} + +#if USE_AVRO +namespace +{ + /// One top-level term of the source MergeTree PARTITION BY, canonicalized for comparison against a + /// destination Iceberg partition field. `transform` holds the ClickHouse function name that + /// parseTransformAndArgument yields for the equivalent Iceberg transform ("toRelativeDayNum", + /// "identity", "icebergBucket", ...); std::nullopt marks a term that is not directly + /// Iceberg-representable and can therefore only satisfy a destination field via the dynamic proof. + struct SourcePartitionTerm + { + String column; + std::optional transform; + std::optional argument; + }; + + std::vector parseSourcePartitionTerms(const ASTPtr & partition_key_ast) + { + /// The same functions getPartitionField maps 1:1 to an Iceberg transform. + static const std::unordered_set iceberg_representable = { + "identity", "toYearNumSinceEpoch", "toMonthNumSinceEpoch", + "toRelativeDayNum", "toRelativeHourNum", "icebergBucket", "icebergTruncate"}; + + std::vector terms; + if (!partition_key_ast) + return terms; + + auto parse_one = [&](const ASTPtr & element) + { + if (const auto * ident = element->as()) + { + terms.push_back({ident->name(), std::optional("identity"), std::nullopt}); + return; + } + + const auto * func = element->as(); + if (!func) + return; + + std::optional column; + std::optional argument; + for (const auto & child : func->children) + { + const auto * expression_list = child->as(); + if (!expression_list) + continue; + for (const auto & arg : expression_list->children) + { + if (const auto * id = arg->as()) + column = id->name(); + else if (const auto * lit = arg->as(); lit && lit->value.getType() != Field::Types::String) + argument = lit->value.safeGet(); + } + } + if (!column) + return; + + std::optional transform; + if (iceberg_representable.contains(func->name)) + transform = func->name; + terms.push_back({*column, transform, argument}); + }; + + if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") + { + for (const auto & child : func->children) + if (const auto * expression_list = child->as()) + for (const auto & element : expression_list->children) + parse_one(element); + } + else + parse_one(partition_key_ast); + + return terms; + } + /// Apply the destination Iceberg transform (rebuilt as a ClickHouse function, mirroring /// ChunkPartitioner and the commit path) to a 2-row column holding [cast(min), cast(max)] and /// return whether both rows map to the same value. @@ -806,6 +937,77 @@ namespace } #endif + void verifyPlainPartitionCompatibility( + const StorageMetadataPtr & source_metadata, + const StorageMetadataPtr & destination_metadata, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context) + { + const auto source_key_ast = source_metadata->getPartitionKeyAST(); + const auto destination_key_ast = destination_metadata->getPartitionKeyAST(); + + auto ast_to_string = [](const ASTPtr & ast) { return ast ? ast->formatWithSecretsOneLine() : String{}; }; + + /// Fast path: identical partition keys are single-valued per source partition by construction. + /// This also preserves acceptance of non-monotonic keys (hashes, modulo) that the dynamic proof + /// below cannot reason about. + if (ast_to_string(source_key_ast) == ast_to_string(destination_key_ast)) + return; + + const auto destination_terms = parsePartitionKeyTerms(destination_key_ast); + if (!destination_terms) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: the destination partition key is not a supported per-column " + "expression and does not match the source partition key."); + + /// Unpartitioned destination: a single output partition trivially holds every source partition. + if (destination_terms->empty()) + return; + + const auto source_terms = parsePartitionKeyTerms(source_key_ast).value_or(std::vector{}); + + const auto & source_partition_key = source_metadata->getPartitionKey(); + const auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(source_partition_key); + const auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(source_partition_key); + + for (const auto & term : *destination_terms) + { + /// Fast path: the source already partitions by the same expression on this column, so every + /// source partition is single-valued for this destination term by construction. + if (std::find(source_terms.begin(), source_terms.end(), term) != source_terms.end()) + continue; + + const auto slot_it = std::find(minmax_column_names.begin(), minmax_column_names.end(), term.column); + if (slot_it == minmax_column_names.end()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: destination partitions by column '{}', which is not part of " + "the source MergeTree partition key.", term.column); + const size_t slot = static_cast(slot_it - minmax_column_names.begin()); + const auto & column_type = minmax_column_types[slot]; + + /// A NULL forms its own destination partition, so a nullable column may split the source + /// partition; min/max cannot rule that out. Require an exact match for such columns. + if (column_type->isNullable()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: partition column '{}' is Nullable, so a NULL forms a separate " + "destination partition. Partition the source by the matching expression.", term.column); + + Field min_value; + Field max_value; + if (!foldPartitionColumnBounds(parts, slot, min_value, max_value)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: no min/max statistics available for column '{}' in partition " + "'{}'; cannot validate partitioning.", term.column, partition_id); + + if (!partitionTermIsConstantOverBounds(term, column_type, min_value, max_value, context)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition '{}': the source partition spans multiple destination " + "partitions for column '{}'. A source MergeTree partition must map to a single " + "destination partition.", partition_id, term.column); + } + } + void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, const StorageMetadataPtr & destination_metadata, diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 8580deb2f4a7..7123c534e5bd 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/src/Storages/MergeTree/ExportPartitionUtils.h @@ -100,6 +100,21 @@ namespace ExportPartitionUtils const StorageID & destination_storage_id, const ContextPtr & context); + /// Verifies the source MergeTree partition key is compatible with a plain (hive) object storage + /// destination partition key. The hive write path evaluates the destination PARTITION BY on the + /// source part's minmax block and writes every part row to that single directory, so each source + /// partition must map to exactly one destination partition. A destination term is proven either by + /// an exact match (identical partition keys, or the same per-column expression) or dynamically, by + /// checking the destination expression is constant over the column's actual [min, max] folded + /// across `parts` (provably-monotonic single-argument functions and bare columns only). Throws + /// BAD_ARGUMENTS when a term cannot be proven single-valued. + void verifyPlainPartitionCompatibility( + const StorageMetadataPtr & source_metadata, + const StorageMetadataPtr & destination_metadata, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context); + #if USE_AVRO /// Verifies the source MergeTree partition key is compatible with the destination Iceberg /// partition spec: every destination partition field must be single-valued across the exported diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index b45c0d4de5bc..e5e5da0809e3 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6692,14 +6692,21 @@ void MergeTreeData::exportPartToTable( if (!dest_storage->supportsImport(query_context)) throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Destination storage {} does not support MergeTree parts or uses unsupported partitioning", dest_storage->getName()); - auto query_to_string = [] (const ASTPtr & ast) - { - return ast ? ast->formatWithSecretsOneLine() : ""; - }; - auto source_metadata_ptr = getInMemoryMetadataPtr(); auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr(); + if (dest_storage->isDataLake() && !query_context->getSettingsRef()[Setting::allow_insert_into_iceberg]) + { + throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, + "Iceberg writes are experimental. " + "To allow its usage, enable the setting `allow_insert_into_iceberg`."); + } + + /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`. Runs before the part + /// lookup so schema/type incompatibilities are reported even when the referenced part is absent. + ExportPartitionUtils::verifyExportSchemaCastable( + source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context); + auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}); if (!part) @@ -6710,13 +6717,6 @@ void MergeTreeData::exportPartToTable( if (dest_storage->isDataLake()) { - if (!query_context->getSettingsRef()[Setting::allow_insert_into_iceberg]) - { - throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, - "Iceberg writes are experimental. " - "To allow its usage, enable the setting `allow_insert_into_iceberg`."); - } - #if USE_AVRO if (iceberg_metadata_json_) { @@ -6762,17 +6762,17 @@ void MergeTreeData::exportPartToTable( throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Data lake export requires Avro support"); #endif } - - /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`. - ExportPartitionUtils::verifyExportSchemaCastable( - source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context); - - /// Iceberg partition compatibility is checked above; here we only need the - /// partition-key ASTs to match (partition-column types follow the lossy-cast gate). - if (!dest_storage->isDataLake()) + else { - if (query_to_string(source_metadata_ptr->getPartitionKeyAST()) != query_to_string(destination_metadata_ptr->getPartitionKeyAST())) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); + /// Plain (hive) object storage writes every row of the part to the one directory computed from + /// the destination PARTITION BY on the part's min row, so the source partition must map to a + /// single destination partition. Equivalent or finer source keys are accepted. + ExportPartitionUtils::verifyPlainPartitionCompatibility( + source_metadata_ptr, + destination_metadata_ptr, + {part}, + part->info.getPartitionId(), + query_context); } if (part->getState() == MergeTreeDataPartState::Outdated && !allow_outdated_parts) diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index c93960783f49..c1d488282ffb 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -8394,11 +8394,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & if (!dest_storage->supportsImport(query_context)) throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Destination storage {} does not support MergeTree parts or uses unsupported partitioning", dest_storage->getName()); - auto query_to_string = [] (const ASTPtr & ast) - { - return ast ? ast->formatWithSecretsOneLine() : ""; - }; - auto src_snapshot = getInMemoryMetadataPtr(); auto destination_snapshot = dest_storage->getInMemoryMetadataPtr(); @@ -8406,14 +8401,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & ExportPartitionUtils::verifyExportSchemaCastable( src_snapshot, destination_snapshot, dest_storage->getStorageID(), query_context); - /// Iceberg partition compatibility is checked below; here we only need the - /// partition-key ASTs to match (partition-column types follow the lossy-cast gate). - if (!dest_storage->isDataLake()) - { - if (query_to_string(src_snapshot->getPartitionKeyAST()) != query_to_string(destination_snapshot->getPartitionKeyAST())) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); - } - zkutil::ZooKeeperPtr zookeeper = getZooKeeperAndAssertNotReadonly(); const String partition_id = getPartitionIDFromQuery(command.partition, query_context); @@ -8583,6 +8570,18 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Data lake export requires Avro support"); #endif } + else + { + /// Plain (hive) object storage writes every row of each part to the one directory computed from + /// the destination PARTITION BY on the part's min row, so each source partition must map to a + /// single destination partition. Equivalent or finer source keys are accepted. + ExportPartitionUtils::verifyPlainPartitionCompatibility( + src_snapshot, + destination_snapshot, + parts, + partition_id, + query_context); + } ops.emplace_back(zkutil::makeCreateRequest( fs::path(partition_exports_path) / "metadata.json", diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py index 44f35925f1e7..612d28e45dec 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py @@ -6,6 +6,7 @@ from helpers.cluster import ClickHouseCluster from helpers.export_partition_helpers import ( + first_partition_id, wait_for_exception_count, wait_for_export_status, wait_for_export_to_start, @@ -1724,3 +1725,150 @@ def test_export_partition_all_failure_modes(cluster): f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}" f" SETTINGS export_merge_tree_partition_all_on_error = 'skip_conflicts'" ) + + +# ---- Partition-key compatibility gate (unified with the Iceberg gate) -------------------------- +# +# Plain (hive) object storage writes every row of a part to the single directory computed from the +# destination PARTITION BY, so each source partition must map to exactly one destination partition. +# The gate accepts equivalent or finer source keys (e.g. a source that adds partition columns on top +# of the destination's) and rejects source partitions that would span several destination partitions +# or that do not cover the destination partition column. Hive destinations partition by bare columns +# only, so these cases exercise the column-subset and single-value paths. + + +def _run_subset_accept(node, source_key): + """Export a source partitioned by *source_key* (a superset of the destination key ``year``) into a + hive destination partitioned by ``year``, then verify the full dataset, the hive directory layout, + and a round-trip back into MergeTree.""" + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"subset_mt_{uid}" + s3_table = f"subset_s3_{uid}" + roundtrip = f"subset_roundtrip_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, year UInt16, country String)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY {source_key} ORDER BY tuple()" + ) + node.query( + f"INSERT INTO {mt_table} VALUES (1, 2020, 'US'), (2, 2020, 'FR'), (3, 2021, 'US')" + ) + node.query( + f"CREATE TABLE {s3_table} (id UInt64, year UInt16, country String)" + f" ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive')" + f" PARTITION BY year" + ) + + partition_ids = node.query( + f"SELECT DISTINCT partition_id FROM system.parts" + f" WHERE database = currentDatabase() AND table = '{mt_table}' AND active" + ).strip().split("\n") + assert len(partition_ids) == 3, f"expected 3 source partitions, got {partition_ids}" + + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + for pid in partition_ids: + wait_for_export_status(node, mt_table, s3_table, pid, "COMPLETED", timeout=90) + + src = node.query(f"SELECT id, year, country FROM {mt_table} ORDER BY id") + dst = node.query(f"SELECT id, year, country FROM {s3_table} ORDER BY id") + assert dst == src, f"destination rows differ from source:\nsrc={src!r}\ndst={dst!r}" + + # The destination partitions by year only: rows land in the year= hive directory. + rows_2020 = node.query( + f"SELECT count() FROM s3(s3_conn, filename='{s3_table}/year=2020/*.parquet', format='Parquet')" + ).strip() + rows_2021 = node.query( + f"SELECT count() FROM s3(s3_conn, filename='{s3_table}/year=2021/*.parquet', format='Parquet')" + ).strip() + assert rows_2020 == "2", f"expected 2 rows under year=2020, got {rows_2020}" + assert rows_2021 == "1", f"expected 1 row under year=2021, got {rows_2021}" + + node.query( + f"CREATE TABLE {roundtrip} (id UInt64, year UInt16, country String)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{roundtrip}', 'replica1')" + f" PARTITION BY {source_key} ORDER BY tuple()" + ) + node.query(f"INSERT INTO {roundtrip} SELECT * FROM {s3_table}") + rt = node.query(f"SELECT id, year, country FROM {roundtrip} ORDER BY id") + assert rt == src, f"round-trip rows differ from source:\nsrc={src!r}\nrt={rt!r}" + + +def test_export_partition_multicolumn_subset_accepted(cluster): + """Source partitions by (year, country); destination by year only - a coarser key that is covered + by the source key, so every source partition has a single year and maps to exactly one destination + partition. Accepted (this was rejected as a partition-key mismatch before the plain gate was + unified with the Iceberg one).""" + node = cluster.instances["replica1"] + _run_subset_accept(node, "(year, country)") + + +def test_export_partition_subset_reversed_order_accepted(cluster): + """The subset match is order-independent: a source keyed by (country, year) still covers a + destination keyed by year.""" + node = cluster.instances["replica1"] + _run_subset_accept(node, "(country, year)") + + +def test_export_partition_coarser_source_rejected(cluster): + """Source partitions monthly (toYYYYMM(dt)); destination by the raw date. A single source part + holding two different days would map to two destination partitions, so the gate rejects the + export synchronously with BAD_ARGUMENTS and schedules nothing.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"coarser_mt_{uid}" + s3_table = f"coarser_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, dt Date)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY toYYYYMM(dt) ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, '2024-03-05'), (2, '2024-03-20')") + node.query( + f"CREATE TABLE {s3_table} (id UInt64, dt Date)" + f" ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive')" + f" PARTITION BY dt" + ) + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {s3_table}" + ) + assert "BAD_ARGUMENTS" in error, f"expected BAD_ARGUMENTS, got: {error!r}" + + scheduled = node.query( + f"SELECT count() FROM system.replicated_partition_exports" + f" WHERE source_table = '{mt_table}' AND destination_table = '{s3_table}'" + ).strip() + assert scheduled == "0", f"expected nothing scheduled after a synchronous reject, got {scheduled}" + + +def test_export_partition_dest_column_not_in_source_key_rejected(cluster): + """Destination partitions by a column that is not part of the source partition key; the gate + rejects the export synchronously with BAD_ARGUMENTS naming the uncovered column.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"nocover_mt_{uid}" + s3_table = f"nocover_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, year UInt16, country String)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY year ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 2020, 'US'), (2, 2020, 'FR')") + node.query( + f"CREATE TABLE {s3_table} (id UInt64, year UInt16, country String)" + f" ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive')" + f" PARTITION BY country" + ) + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {s3_table}" + ) + assert "BAD_ARGUMENTS" in error, f"expected BAD_ARGUMENTS, got: {error!r}" + assert "country" in error, f"expected the error to name column 'country', got: {error!r}" diff --git a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py index 94d4d6c1a017..b922bb895a83 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py @@ -586,16 +586,17 @@ def test_rejected_column_mismatch(export_cluster): def test_rejected_transform_mismatch(export_cluster): - """Spark years(dt) — RMT PARTITION BY dt (identity, not year-transform).""" + """Spark days(dt) destination — RMT PARTITION BY toStartOfMonth(dt): a month partition spans + several days, so it cannot map to a single Iceberg day partition.""" error = run_rejected( export_cluster, "rej_xform_mismatch", spark_ddl="CREATE TABLE {TABLE} (id BIGINT, dt DATE)" - " USING iceberg PARTITIONED BY (years(dt)) OPTIONS('format-version'='2')", + " USING iceberg PARTITIONED BY (days(dt)) OPTIONS('format-version'='2')", ch_schema="id Int64, dt Date", rmt_columns="id Int64, dt Date", - rmt_partition_by="dt", - insert_values="(1, '2021-06-01')", + rmt_partition_by="toStartOfMonth(dt)", + insert_values="(1, '2021-06-01'), (2, '2021-06-15')", ) assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}" @@ -616,47 +617,17 @@ def test_rejected_bucket_count_mismatch(export_cluster): def test_rejected_truncate_width_mismatch(export_cluster): - """Spark truncate(4, category) — RMT icebergTruncate(8, category): wrong width.""" + """Spark truncate(8, category) destination — RMT icebergTruncate(4, category): the coarser + width-4 source partition splits across several width-8 destination buckets.""" error = run_rejected( export_cluster, "rej_trunc_w", spark_ddl="CREATE TABLE {TABLE} (id BIGINT, category STRING)" - " USING iceberg PARTITIONED BY (truncate(4, category)) OPTIONS('format-version'='2')", + " USING iceberg PARTITIONED BY (truncate(8, category)) OPTIONS('format-version'='2')", ch_schema="id Int64, category String", rmt_columns="id Int64, category String", - rmt_partition_by="icebergTruncate(8, category)", - insert_values="(1, 'clickhouse')", - ) - assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}" - - -def test_rejected_field_count_mismatch(export_cluster): - """Spark 1-field identity(year) — RMT 2-field (year, region).""" - error = run_rejected( - export_cluster, - "rej_field_n", - spark_ddl="CREATE TABLE {TABLE} (id BIGINT, year INT, region STRING)" - " USING iceberg PARTITIONED BY (identity(year)) OPTIONS('format-version'='2')", - ch_schema="id Int64, year Int32, region String", - rmt_columns="id Int64, year Int32, region String", - rmt_partition_by="(year, region)", - insert_values="(1, 2024, 'EU')", - ) - assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}" - - -def test_rejected_compound_order_reversed(export_cluster): - """Spark (identity(year), identity(region)) — RMT (region, year): reversed order.""" - error = run_rejected( - export_cluster, - "rej_compound_rev", - spark_ddl="CREATE TABLE {TABLE} (id BIGINT, year INT, region STRING)" - " USING iceberg PARTITIONED BY (identity(year), identity(region))" - " OPTIONS('format-version'='2')", - ch_schema="id Int64, year Int32, region String", - rmt_columns="id Int64, year Int32, region String", - rmt_partition_by="(region, year)", - insert_values="(1, 2024, 'EU')", + rmt_partition_by="icebergTruncate(4, category)", + insert_values="(1, 'clickhouse'), (2, 'clickfast')", ) assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}" diff --git a/tests/queries/0_stateless/03572_export_merge_tree_part_to_object_storage_simple.sql b/tests/queries/0_stateless/03572_export_merge_tree_part_to_object_storage_simple.sql index d19254dc636f..c59cebc45c52 100644 --- a/tests/queries/0_stateless/03572_export_merge_tree_part_to_object_storage_simple.sql +++ b/tests/queries/0_stateless/03572_export_merge_tree_part_to_object_storage_simple.sql @@ -1,6 +1,6 @@ -- Tags: no-parallel, no-fasttest -DROP TABLE IF EXISTS 03572_mt_table, 03572_invalid_schema_table, 03572_ephemeral_mt_table, 03572_matching_ephemeral_s3_table, 03572_partition_type_mismatch_mt, 03572_partition_type_mismatch_s3, 03572_lossy_mt, 03572_lossy_s3, 03572_lossless_mt, 03572_lossless_s3; +DROP TABLE IF EXISTS 03572_mt_table, 03572_invalid_schema_table, 03572_ephemeral_mt_table, 03572_matching_ephemeral_s3_table, 03572_partition_type_mismatch_mt, 03572_partition_type_mismatch_s3, 03572_lossy_mt, 03572_lossy_s3, 03572_lossless_mt, 03572_lossless_s3, 03572_coarser_source_mt, 03572_finer_dest_s3; SET allow_experimental_export_merge_tree_part=1; @@ -8,9 +8,10 @@ CREATE TABLE 03572_mt_table (id UInt64, year UInt16) ENGINE = MergeTree() PARTIT INSERT INTO 03572_mt_table VALUES (1, 2020); --- Create a table with a different partition key and export a partition to it. It should throw --- on the partition-key AST mismatch (schema compat now follows INSERT SELECT positional semantics, --- so the column shape matches and the partition-key check is what fires). +-- Create a table partitioned by a column that is not part of the source partition key. The unified +-- plain-storage partition gate rejects it because the destination partition column is not covered by +-- the source partition key (schema compat follows INSERT SELECT positional semantics, so the column +-- shape matches and the partition-compatibility check is what fires). CREATE TABLE 03572_invalid_schema_table (id UInt64, x UInt16) ENGINE = S3(s3_conn, filename='03572_invalid_schema_table', format='Parquet', partition_strategy='hive') PARTITION BY x; ALTER TABLE 03572_mt_table EXPORT PART '2020_1_1_0' TO TABLE 03572_invalid_schema_table @@ -67,4 +68,16 @@ CREATE TABLE 03572_lossless_s3 (id Int64, year UInt16) ENGINE = S3(s3_conn, file ALTER TABLE 03572_lossless_mt EXPORT PART '2020_1_1_0' TO TABLE 03572_lossless_s3 SETTINGS allow_experimental_export_merge_tree_part = 1; -- {serverError NO_SUCH_DATA_PART} -DROP TABLE IF EXISTS 03572_mt_table, 03572_invalid_schema_table, 03572_ephemeral_mt_table, 03572_matching_ephemeral_s3_table, 03572_partition_type_mismatch_mt, 03572_partition_type_mismatch_s3, 03572_lossy_mt, 03572_lossy_s3, 03572_lossless_mt, 03572_lossless_s3; +-- Unified plain-storage partition gate: the destination partitioning must be single-valued within +-- each exported source part. The source is partitioned monthly (toYYYYMM(dt)) while the destination +-- is partitioned by the raw date, so a single source part holding two different days would map to two +-- destination partitions. The gate rejects it (the part exists, so the data-dependent check runs). +CREATE TABLE 03572_coarser_source_mt (id UInt64, dt Date) ENGINE = MergeTree() PARTITION BY toYYYYMM(dt) ORDER BY tuple(); +CREATE TABLE 03572_finer_dest_s3 (id UInt64, dt Date) ENGINE = S3(s3_conn, filename='03572_finer_dest_s3', format='Parquet', partition_strategy='hive') PARTITION BY dt; + +INSERT INTO 03572_coarser_source_mt VALUES (1, '2024-03-05'), (2, '2024-03-20'); + +ALTER TABLE 03572_coarser_source_mt EXPORT PART '202403_1_1_0' TO TABLE 03572_finer_dest_s3 +SETTINGS allow_experimental_export_merge_tree_part = 1; -- {serverError BAD_ARGUMENTS} + +DROP TABLE IF EXISTS 03572_mt_table, 03572_invalid_schema_table, 03572_ephemeral_mt_table, 03572_matching_ephemeral_s3_table, 03572_partition_type_mismatch_mt, 03572_partition_type_mismatch_s3, 03572_lossy_mt, 03572_lossy_s3, 03572_lossless_mt, 03572_lossless_s3, 03572_coarser_source_mt, 03572_finer_dest_s3; From 454e9fc156364399229546d3c9a12144d93fa2c1 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Mon, 20 Jul 2026 15:04:29 -0300 Subject: [PATCH 5/8] recompute dst key with correct set of parts, include tz --- .../MergeTree/ExportPartitionUtils.cpp | 124 +++++++++++++-- src/Storages/MergeTree/ExportPartitionUtils.h | 20 ++- src/Storages/MergeTree/MergeTreeData.cpp | 2 - .../test.py | 143 +++++++++++++++--- 4 files changed, 239 insertions(+), 50 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 7c28ce895e0f..79e42fe3e9df 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -31,6 +31,7 @@ #include #include #include +#include #endif namespace ProfileEvents @@ -145,20 +146,61 @@ namespace ExportPartitionUtils } Block getPartitionSourceBlockForIcebergCommit( - MergeTreeData & storage, const String & partition_id) + MergeTreeData & storage, const String & partition_id, const std::vector & exported_part_names) { auto lock = storage.readLockParts(); const auto parts = storage.getDataPartsVectorInPartitionForInternalUsage( - MergeTreeDataPartState::Active, partition_id, lock); + {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}, partition_id, lock); - if (parts.empty()) + /// Derive the representative only from the exact parts that were validated and exported (recorded + /// in the manifest), never from unrelated parts inserted/merged after scheduling: those could map + /// to a different Iceberg partition and stamp metadata that does not match the exported files. + const std::unordered_set exported(exported_part_names.begin(), exported_part_names.end()); + MergeTreeData::DataPartsVector exported_parts; + for (const auto & part : parts) + if (exported.contains(part->name) && part->minmax_idx && part->minmax_idx->initialized) + exported_parts.push_back(part); + + if (exported_parts.empty()) throw Exception(ErrorCodes::NO_SUCH_DATA_PART, - "Cannot find active part for partition_id '{}' to derive Iceberg partition " - "values. Edge case: the partition may have been dropped after export started, " - "or this replica has not yet received any part for this partition. " - "The commit will be retried.", + "Cannot find any of the exported parts for partition_id '{}' to derive Iceberg partition " + "values. They may have been merged and cleaned up before this commit, or are not present " + "on this replica. The commit will be retried.", partition_id); - return parts.front()->minmax_idx->getBlock(storage); + + /// The gate proved the exported parts map to a single Iceberg partition, so any exported part's + /// values yield the same tuple; fold the global min across them as a deterministic representative. + const auto metadata_snapshot = storage.getInMemoryMetadataPtr(); + const auto & partition_key = metadata_snapshot->getPartitionKey(); + const auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(partition_key); + const auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(partition_key); + + Block block; + for (size_t i = 0; i < minmax_column_types.size(); ++i) + { + Field min_value; + bool found = false; + for (const auto & part : exported_parts) + { + if (i >= part->minmax_idx->hyperrectangle.size()) + continue; + auto range = part->minmax_idx->hyperrectangle[i]; + range.shrinkToIncludedIfPossible(); + if (!found || range.left < min_value) + { + min_value = range.left; + found = true; + } + } + + auto column = minmax_column_types[i]->createColumn(); + if (found) + column->insert(min_value); + else + column->insertDefault(); + block.insert(ColumnWithTypeAndName(column->getPtr(), minmax_column_types[i], minmax_column_names[i])); + } + return block; } ContextPtr getContextCopyWithTaskSettings(const ContextPtr & context, const ExportReplicatedMergeTreePartitionManifest & manifest) @@ -323,7 +365,7 @@ namespace ExportPartitionUtils iceberg_args.metadata_json_string = manifest.iceberg_metadata_json; if (source_storage.getInMemoryMetadataPtr()->hasPartitionKey()) iceberg_args.partition_source_block = - getPartitionSourceBlockForIcebergCommit(source_storage, manifest.partition_id); + getPartitionSourceBlockForIcebergCommit(source_storage, manifest.partition_id, manifest.parts); } destination_storage->commitExportPartitionTransaction(manifest.transaction_id, manifest.partition_id, exported_paths, iceberg_args, context); @@ -668,6 +710,7 @@ namespace String column; std::optional transform; std::optional argument; + std::optional time_zone; }; std::vector parseSourcePartitionTerms(const ASTPtr & partition_key_ast) @@ -685,7 +728,7 @@ namespace { if (const auto * ident = element->as()) { - terms.push_back({ident->name(), std::optional("identity"), std::nullopt}); + terms.push_back({ident->name(), std::optional("identity"), std::nullopt, std::nullopt}); return; } @@ -695,6 +738,7 @@ namespace std::optional column; std::optional argument; + std::optional time_zone; for (const auto & child : func->children) { const auto * expression_list = child->as(); @@ -702,9 +746,12 @@ namespace continue; for (const auto & arg : expression_list->children) { + const auto * lit = arg->as(); if (const auto * id = arg->as()) column = id->name(); - else if (const auto * lit = arg->as(); lit && lit->value.getType() != Field::Types::String) + else if (lit && lit->value.getType() == Field::Types::String) + time_zone = lit->value.safeGet(); + else if (lit) argument = lit->value.safeGet(); } } @@ -714,7 +761,7 @@ namespace std::optional transform; if (iceberg_representable.contains(func->name)) transform = func->name; - terms.push_back({*column, transform, argument}); + terms.push_back({*column, transform, argument, time_zone}); }; if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") @@ -838,6 +885,18 @@ namespace const auto destination_sample = destination_metadata->getSampleBlockNonMaterialized(); const String partition_timezone = context->getSettingsRef()[Setting::iceberg_partition_timezone]; + /// year/month/day/hour depend on the timezone (for DateTime); the structural fast path is only + /// sound when the source and destination evaluate them in the same timezone. + static const std::unordered_set timezone_sensitive_transforms = { + "toYearNumSinceEpoch", "toMonthNumSinceEpoch", "toRelativeDayNum", "toRelativeHourNum"}; + + auto column_explicit_time_zone = [](const DataTypePtr & type) -> String + { + if (const auto * tz = dynamic_cast(type.get()); tz && tz->hasExplicitTimeZone()) + return tz->getTimeZone().getTimeZone(); + return ""; + }; + for (UInt32 i = 0; i < actual_size; ++i) { const auto af = actual_fields->getObject(i); @@ -851,15 +910,48 @@ namespace /// Fast path: the source already applies the matching transform on this column, so every /// source partition is single-valued for this field by construction (covers bucket too). + /// It is only taken when the transform semantics are provably identical. identity is + /// exempt: an identity source partition is already a single value, so it stays single-valued + /// under any cast. Every other transform is applied to the destination column type, so a + /// structural match requires identical types (icebergTruncate/icebergBucket are numeric on + /// integers but byte-wise on strings, so a pre-transform cast that changes the type changes + /// the result); year/month/day/hour on DateTime additionally require the same effective + /// timezone. Mismatches fall through to the dynamic proof, which evaluates the destination + /// transform on the real data. bool matched_structurally = false; for (const auto & term : source_terms) { - if (term.column == column && term.transform && dest_canonical - && *term.transform == dest_canonical->transform_name && term.argument == dest_argument) + if (term.column != column || !term.transform || !dest_canonical + || *term.transform != dest_canonical->transform_name || term.argument != dest_argument) + continue; + + const auto & transform_name = dest_canonical->transform_name; + if (transform_name != "identity") { - matched_structurally = true; - break; + const auto slot_it = std::find(minmax_column_names.begin(), minmax_column_names.end(), column); + if (slot_it == minmax_column_names.end() || !destination_sample.has(column)) + continue; + const auto & structural_source_type = minmax_column_types[slot_it - minmax_column_names.begin()]; + const auto & structural_dest_type = destination_sample.getByName(column).type; + if (!structural_source_type->equals(*structural_dest_type)) + continue; + + if (timezone_sensitive_transforms.contains(transform_name)) + { + const WhichDataType which(structural_source_type); + if (which.isDateTime() || which.isDateTime64()) + { + const String source_tz = term.time_zone.value_or(column_explicit_time_zone(structural_source_type)); + const String dest_tz = partition_timezone.empty() + ? column_explicit_time_zone(structural_dest_type) : partition_timezone; + if (source_tz != dest_tz) + continue; + } + } } + + matched_structurally = true; + break; } if (matched_structurally) continue; diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 7123c534e5bd..4ecdfd9bfb05 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/src/Storages/MergeTree/ExportPartitionUtils.h @@ -30,16 +30,20 @@ namespace ExportPartitionUtils ContextPtr getContextCopyWithTaskSettings(const ContextPtr & context, const ExportReplicatedMergeTreePartitionManifest & manifest); - /// Returns the representative source partition-key columns (the first active local part's - /// minmax block) for the given partition_id. The destination recomputes the Iceberg partition - /// tuple from this block by casting to its column types and applying the partition transform. + /// Returns the representative source partition-key columns (a folded global-min minmax block) for + /// the given partition_id, derived only from the exact parts that were validated and exported + /// (`exported_part_names`, looked up among Active and Outdated parts). The destination recomputes + /// the Iceberg partition tuple from this block by casting to its column types and applying the + /// partition transform. Restricting to the exported parts keeps the committed metadata consistent + /// with the exported files even if unrelated parts were inserted/merged into the partition after + /// scheduling. /// - /// Edge case: if the partition was dropped after export started, or this replica - /// has not yet received any part for this partition (extreme replication lag on a - /// recovery path), no active part will be found and the commit will fail. The task - /// will be retried on the next poll cycle or picked up by a different replica. + /// Edge case: if none of the exported parts are found (merged away and cleaned up before commit, + /// or not present on this replica), a NO_SUCH_DATA_PART exception is thrown; it is retryable, so + /// the commit is retried on the next poll cycle or picked up by a different replica, rather than + /// silently stamping metadata derived from unrelated parts. Block getPartitionSourceBlockForIcebergCommit( - MergeTreeData & storage, const String & partition_id); + MergeTreeData & storage, const String & partition_id, const std::vector & exported_part_names); void commit( const ExportReplicatedMergeTreePartitionManifest & manifest, diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index e5e5da0809e3..fd3f54ac6da7 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6702,8 +6702,6 @@ void MergeTreeData::exportPartToTable( "To allow its usage, enable the setting `allow_insert_into_iceberg`."); } - /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`. Runs before the part - /// lookup so schema/type incompatibilities are reported even when the referenced part is absent. ExportPartitionUtils::verifyExportSchemaCastable( source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context); diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 601aee695222..ee7261e2ae84 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -1951,18 +1951,18 @@ def setup(case): assert_iceberg_partition_metadata(node, iceberg_table, f"{case['name']}_{uid}", fields) -def test_export_partition_bucket_transform_metadata_matches_data(cluster): - """A bucket[N] partition column whose type changes Int64 -> String records the - destination murmur(String) bucket in the Iceberg metadata, matching the exported - data rather than the source hashLong bucket.""" +def test_export_partition_bucket_type_change_rejected(cluster): + """A bucket[N] partition column whose type changes (Int64 -> String) is rejected. The source + hashLong grouping differs from the destination murmur(String) grouping, so a single source bucket + can fan out across several destination buckets; bucket is not order-preserving, so this cannot be + proven dynamically and must be rejected. This previously slipped through the structural fast path, + which matched on transform name and width while ignoring the pre-transform cast.""" node = cluster.instances["replica1"] uid = unique_suffix() mt_table = f"mt_bucket_xform_{uid}" iceberg_table = f"iceberg_bucket_xform_{uid}" - # N=16, key=42 diverges: icebergBucket(16, 42::Int64)=14 (source/old hashLong) but - # icebergBucket(16, '42')=6 (destination/new murmur over the exported String). make_rmt(node, mt_table, "id Int64, key Int64", "icebergBucket(16, key)", replica_name="replica1") node.query(f"INSERT INTO {mt_table} VALUES (1, 42), (2, 42)") @@ -1971,27 +1971,118 @@ def test_export_partition_bucket_transform_metadata_matches_data(cluster): partition_by="icebergBucket(16, key)") pid = first_partition_id(node, mt_table) - node.query( + error = node.query_and_get_error( f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", settings={"allow_insert_into_iceberg": 1}, ) - wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + assert "BAD_ARGUMENTS" in error, ( + f"Expected BAD_ARGUMENTS for a type-changing bucket transform, got: {error!r}" + ) - count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) - assert count == 2, f"Expected 2 rows after export, got {count}" - string_bucket = int(node.query( - f"SELECT DISTINCT icebergBucket(16, key) FROM {iceberg_table}" - ).strip()) - long_bucket = int(node.query( - f"SELECT DISTINCT icebergBucket(16, toInt64(key)) FROM {iceberg_table}" - ).strip()) - assert string_bucket != long_bucket, ( - f"Test setup invalid: String and Int64 buckets coincide ({string_bucket}); " - f"pick a different N/key so the transform diverges." +def test_export_partition_truncate_type_change_rejected(cluster): + """icebergTruncate with the same width but a changed column type (Int64 -> String) is rejected. + Truncate is numeric on integers (120..129 -> 120) but byte-wise on strings ('120'..'129' stay + distinct), so one source truncate bucket can map to several destination buckets. The structural + fast path must not accept it on matching transform name and width; the dynamic proof rejects it + because the endpoints do not collapse to a single destination value.""" + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_trunc_xform_{uid}" + iceberg_table = f"iceberg_trunc_xform_{uid}" + + # 120 and 129 are one Int64 truncate[10] bucket (120) but two distinct string truncations. + make_rmt(node, mt_table, "id Int64, key Int64", "icebergTruncate(10, key)", + replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, 120), (2, 129)") + + make_iceberg_s3(node, iceberg_table, "id Int64, key String", + partition_by="icebergTruncate(10, key)") + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error, ( + f"Expected BAD_ARGUMENTS for a type-changing truncate transform, got: {error!r}" ) - query_id = f"bucket_xform_{uid}" + +def test_export_partition_timezone_mismatch_rejected(cluster): + """A source partitioned by day in one timezone must not be treated as structurally identical to a + destination day computed in another timezone. The source uses Asia/Tokyo (UTC+9) and the + destination UTC; the exported part spans a UTC-day boundary while staying within one Tokyo day, so + it maps to two destination partitions and must be rejected.""" + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_tzmismatch_{uid}" + iceberg_table = f"iceberg_tzmismatch_{uid}" + + make_rmt(node, mt_table, "id Int64, event_time DateTime('UTC')", + "toRelativeDayNum(event_time, 'Asia/Tokyo')", replica_name="replica1") + # Both instants are 2024-03-05 in Tokyo (UTC+9) but 2024-03-04 and 2024-03-05 in UTC. + node.query( + f"INSERT INTO {mt_table} VALUES (1, '2024-03-04 16:00:00'), (2, '2024-03-05 10:00:00')" + ) + + make_iceberg_s3(node, iceberg_table, "id Int64, event_time DateTime('UTC')", + partition_by="toRelativeDayNum(event_time)") + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1, "iceberg_partition_timezone": "UTC"}, + ) + assert "BAD_ARGUMENTS" in error, ( + f"Expected BAD_ARGUMENTS for a source/destination timezone mismatch, got: {error!r}" + ) + + +def test_export_partition_commit_uses_exported_parts_not_new_inserts(cluster): + """The deferred commit derives the Iceberg partition value only from the exact exported parts + recorded in the manifest, never from parts inserted/merged into the source partition after + scheduling. A month-partitioned source exports one day into a day-partitioned destination (a + data-dependent acceptance); while the commit is wedged, an earlier day is inserted and merged in, + so the only active part now spans both days with its min at the new day. The commit must still + stamp the exported day (the exported part is found among Outdated parts by name), not the merged-in + earlier day, so the metadata matches the exported data files.""" + node = cluster.instances["replica1"] + uid = unique_suffix() + mt_table = f"mt_commit_parts_{uid}" + iceberg_table = f"iceberg_commit_parts_{uid}" + + make_rmt(node, mt_table, "id Int64, event_date Date", "toYYYYMM(event_date)", replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, '2024-03-20'), (2, '2024-03-20')") + make_iceberg_s3(node, iceberg_table, "id Int64, event_date Date", + partition_by="toRelativeDayNum(event_date)") + + exported_day = int(node.query("SELECT toRelativeDayNum(toDate('2024-03-20'))").strip()) + injected_day = int(node.query("SELECT toRelativeDayNum(toDate('2024-03-05'))").strip()) + + node.query("SYSTEM ENABLE FAILPOINT export_partition_commit_always_throw") + try: + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '202403' TO TABLE {iceberg_table}" + f" SETTINGS allow_insert_into_iceberg = 1" + ) + # The commit is attempted only after every part is exported, so a non-zero exception count + # means the data files are written and the commit is now wedged by the failpoint. + wait_for_exception_count(node, mt_table, iceberg_table, "202403", min_exception_count=1, timeout=90) + + # Insert an earlier day into the same month partition and merge: the merged active part spans + # both days with min = the injected (earlier) day, while the exported part becomes Outdated. + node.query(f"INSERT INTO {mt_table} VALUES (3, '2024-03-05')") + node.query(f"OPTIMIZE TABLE {mt_table} PARTITION ID '202403' FINAL") + finally: + node.query("SYSTEM DISABLE FAILPOINT export_partition_commit_always_throw") + + wait_for_export_status(node, mt_table, iceberg_table, "202403", "COMPLETED", timeout=90) + + # The exported data files hold only 2024-03-20; the metadata day must match them. + query_id = f"commit_parts_{uid}" node.query( f"SELECT * FROM {iceberg_table}", query_id=query_id, @@ -2000,10 +2091,14 @@ def test_export_partition_bucket_transform_metadata_matches_data(cluster): entries = fetch_manifest_entries(node, query_id) partitions = _data_file_partition_records(entries) assert partitions, "No data-file partition records found in manifest entries" - meta_values = {int(_partition_scalar(p, "key")) for p in partitions} - assert meta_values == {string_bucket}, ( - f"Metadata bucket {meta_values} must equal the destination String bucket " - f"{string_bucket} (not the source Int64 bucket {long_bucket})." + meta_days = {int(_partition_scalar(p, "event_date")) for p in partitions} + assert meta_days == {exported_day}, ( + f"Metadata day {meta_days} must equal the exported day {exported_day} (2024-03-20), " + f"not the injected day {injected_day} (2024-03-05)." + ) + + assert int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) == 2, ( + "Only the two exported rows must be present in the destination." ) From e4a69a01071a8d67ca79ac18809cde15e8aea797 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Wed, 22 Jul 2026 09:30:21 -0300 Subject: [PATCH 6/8] some more vibe coded style changes --- .../MergeTree/ExportPartitionUtils.cpp | 86 +++++++++---------- 1 file changed, 40 insertions(+), 46 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 5cf5a76fc333..6b1ed7f878d4 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -661,33 +661,41 @@ namespace return terms; } - /// Fold [min, max] of the partition-key column at `slot` across all parts of the partition. - /// Returns false if no part exposes an initialized min/max for that column. - bool foldPartitionColumnBounds( - const MergeTreeData::DataPartsVector & parts, size_t slot, Field & out_min, Field & out_max) + std::optional> calculatePartitionColumnMinMax( + const MergeTreeData::DataPartsVector & parts, size_t slot) { - bool found = false; + std::optional> bounds; for (const auto & part : parts) { if (!part->minmax_idx || !part->minmax_idx->initialized || slot >= part->minmax_idx->hyperrectangle.size()) continue; auto range = part->minmax_idx->hyperrectangle[slot]; range.shrinkToIncludedIfPossible(); - if (!found) + if (!bounds) { - out_min = range.left; - out_max = range.right; - found = true; + bounds.emplace(range.left, range.right); } else { - if (range.left < out_min) - out_min = range.left; - if (out_max < range.right) - out_max = range.right; + if (range.left < bounds->first) + bounds->first = range.left; + if (bounds->second < range.right) + bounds->second = range.right; } } - return found; + return bounds; + } + + + bool endpointsMapToDifferentValues( + const IFunctionBase & function, const ColumnsWithTypeAndName & arguments, size_t num_rows) + { + const auto result = function.execute(arguments, function.getResultType(), num_rows, /*dry_run=*/ false); + Field first; + Field second; + result->get(0, first); + result->get(1, second); + return first != second; } /// Whether the destination partition term yields a single value across [min, max] of its source @@ -710,23 +718,15 @@ namespace auto values_column = column_type->createColumn(); values_column->insert(min_value); values_column->insert(max_value); - const ColumnWithTypeAndName values{std::move(values_column), column_type, term.column}; - - const ColumnsWithTypeAndName arguments{values}; - const auto result_type = resolver->getReturnType(arguments); - const auto function_base = resolver->build(arguments); + const ColumnsWithTypeAndName arguments{{std::move(values_column), column_type, term.column}}; + const auto function = resolver->build(arguments); - if (!function_base->hasInformationAboutMonotonicity()) - return false; - if (!function_base->getMonotonicityForRange(*column_type, min_value, max_value).is_monotonic) + /// Require provable monotonicity so the endpoint comparison covers all interior rows. + if (!function->hasInformationAboutMonotonicity() + || !function->getMonotonicityForRange(*column_type, min_value, max_value).is_monotonic) return false; - const auto result = function_base->execute(arguments, result_type, 2, /*dry_run=*/ false); - Field first; - Field second; - result->get(0, first); - result->get(1, second); - return first == second; + return !endpointsMapToDifferentValues(*function, arguments, /*num_rows=*/ 2); } } @@ -812,11 +812,12 @@ namespace /// Apply the destination Iceberg transform (rebuilt as a ClickHouse function, mirroring /// ChunkPartitioner and the commit path) to a 2-row column holding [cast(min), cast(max)] and - /// return whether both rows map to the same value. - bool destinationTransformIsConstant( + /// return whether the endpoints map to different values, i.e. the source partition would be split + /// across more than one destination partition. + bool wouldPartitionBeSplit( const Iceberg::TransformAndArgument & transform, const ColumnWithTypeAndName & values, const ContextPtr & context) { - auto function = FunctionFactory::instance().get(transform.transform_name, context); + auto resolver = FunctionFactory::instance().get(transform.transform_name, context); const size_t num_rows = values.column->size(); ColumnsWithTypeAndName arguments; @@ -828,14 +829,7 @@ namespace arguments.push_back({DataTypeString().createColumnConst(num_rows, *transform.time_zone), std::make_shared(), "timezone"}); - const auto result_type = function->getReturnType(arguments); - const auto result = function->build(arguments)->execute(arguments, result_type, num_rows, /*dry_run=*/ false); - - Field first; - Field second; - result->get(0, first); - result->get(1, second); - return first == second; + return endpointsMapToDifferentValues(*resolver->build(arguments), arguments, num_rows); } } @@ -1015,12 +1009,12 @@ namespace "separate Iceberg partition; partition the source by the matching Iceberg transform.", column); - Field min_value; - Field max_value; - if (!foldPartitionColumnBounds(parts, slot, min_value, max_value)) + const auto bounds = calculatePartitionColumnMinMax(parts, slot); + if (!bounds) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot export partition to Iceberg table: no min/max statistics available for column " "'{}' in partition '{}'; cannot validate partitioning.", column, partition_id); + const auto & [min_value, max_value] = *bounds; if (!destination_sample.has(column)) throw Exception(ErrorCodes::BAD_ARGUMENTS, @@ -1053,7 +1047,7 @@ namespace castColumn({std::move(values_column), source_type, column}, destination_type), destination_type, column}; const auto dest_transform_with_tz = Iceberg::parseTransformAndArgument(dest_transform, partition_timezone); - if (!destinationTransformIsConstant(*dest_transform_with_tz, cast_values, context)) + if (wouldPartitionBeSplit(*dest_transform_with_tz, cast_values, context)) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot export partition to Iceberg table: partition '{}' spans multiple destination " "partitions for column '{}' (transform '{}'). A source MergeTree partition must map to a " @@ -1118,12 +1112,12 @@ namespace "Cannot export partition: partition column '{}' is Nullable, so a NULL forms a separate " "destination partition. Partition the source by the matching expression.", term.column); - Field min_value; - Field max_value; - if (!foldPartitionColumnBounds(parts, slot, min_value, max_value)) + const auto bounds = calculatePartitionColumnMinMax(parts, slot); + if (!bounds) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot export partition: no min/max statistics available for column '{}' in partition " "'{}'; cannot validate partitioning.", term.column, partition_id); + const auto & [min_value, max_value] = *bounds; if (!partitionTermIsConstantOverBounds(term, column_type, min_value, max_value, context)) throw Exception(ErrorCodes::BAD_ARGUMENTS, From c60c97d425de41313541f6fbe216bab2426e0c1b Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Wed, 22 Jul 2026 09:44:32 -0300 Subject: [PATCH 7/8] add missing tests --- ...xport_part_hive_partition_subset.reference | 9 ++++ ...03572_export_part_hive_partition_subset.sh | 41 +++++++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 tests/queries/0_stateless/03572_export_part_hive_partition_subset.reference create mode 100755 tests/queries/0_stateless/03572_export_part_hive_partition_subset.sh diff --git a/tests/queries/0_stateless/03572_export_part_hive_partition_subset.reference b/tests/queries/0_stateless/03572_export_part_hive_partition_subset.reference new file mode 100644 index 000000000000..9e404f989566 --- /dev/null +++ b/tests/queries/0_stateless/03572_export_part_hive_partition_subset.reference @@ -0,0 +1,9 @@ +---- Export each source part into the coarser destination +---- Destination should hold all rows +1 2020 US +2 2020 FR +3 2021 US +---- Round-trip back into a MergeTree table (should match the source) +1 2020 US +2 2020 FR +3 2021 US diff --git a/tests/queries/0_stateless/03572_export_part_hive_partition_subset.sh b/tests/queries/0_stateless/03572_export_part_hive_partition_subset.sh new file mode 100755 index 000000000000..cf9f43684001 --- /dev/null +++ b/tests/queries/0_stateless/03572_export_part_hive_partition_subset.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +# Tags: replica, no-parallel, no-replicated-database, no-fasttest + +CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CURDIR"/../shell_config.sh + +rmt_table="rmt_table_${RANDOM}" +s3_table="s3_table_${RANDOM}" +rmt_table_roundtrip="rmt_table_roundtrip_${RANDOM}" + +query() { + $CLICKHOUSE_CLIENT --query "$1" +} + +query "DROP TABLE IF EXISTS $rmt_table, $s3_table, $rmt_table_roundtrip" + +# The source partitions by (year, country); the destination partitions by year only - a coarser key +# that is covered by the source partition key. Every source part has a single year, so it maps to +# exactly one destination partition and the unified plain-storage gate accepts the export even though +# the partition keys are not identical (this was rejected before the unification). +query "CREATE TABLE $rmt_table (id UInt64, year UInt16, country String) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{database}/$rmt_table', 'replica1') PARTITION BY (year, country) ORDER BY tuple()" +query "CREATE TABLE $s3_table (id UInt64, year UInt16, country String) ENGINE = S3(s3_conn, filename='$s3_table', format=Parquet, partition_strategy='hive') PARTITION BY year" + +query "INSERT INTO $rmt_table VALUES (1, 2020, 'US'), (2, 2020, 'FR'), (3, 2021, 'US')" + +echo "---- Export each source part into the coarser destination" +part_names=$(query "SELECT name FROM system.parts WHERE database = currentDatabase() AND table = '$rmt_table' AND active ORDER BY name") +for part in $part_names; do + query "ALTER TABLE $rmt_table EXPORT PART '$part' TO TABLE $s3_table SETTINGS allow_experimental_export_merge_tree_part = 1" +done + +echo "---- Destination should hold all rows" +query "SELECT * FROM $s3_table ORDER BY id" + +echo "---- Round-trip back into a MergeTree table (should match the source)" +query "CREATE TABLE $rmt_table_roundtrip (id UInt64, year UInt16, country String) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{database}/$rmt_table_roundtrip', 'replica1') PARTITION BY (year, country) ORDER BY tuple()" +query "INSERT INTO $rmt_table_roundtrip SELECT * FROM $s3_table" +query "SELECT * FROM $rmt_table_roundtrip ORDER BY id" + +query "DROP TABLE IF EXISTS $rmt_table, $s3_table, $rmt_table_roundtrip" From a9f51e14c8bb3c4c0b3bd8750a39756a4f124af1 Mon Sep 17 00:00:00 2001 From: Arthur Passos Date: Wed, 22 Jul 2026 14:15:24 -0300 Subject: [PATCH 8/8] code simplification --- .../MergeTree/ExportPartitionUtils.cpp | 164 +++++------------- 1 file changed, 39 insertions(+), 125 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 6b1ed7f878d4..751f620fdb56 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -572,49 +572,45 @@ namespace ExportPartitionUtils namespace { - /// One top-level term of a MergeTree PARTITION BY: a column and, when the term is a function of a - /// single column, that function name and an optional integer argument. An empty function marks a - /// bare column (identity). - struct PartitionKeyTerm + /// One top-level term of a PARTITION BY: a column and, when the term is a function of a single + /// column, its function name (empty for a bare column) and optional integer / timezone arguments. + /// The Iceberg gate canonicalizes the function name against the Iceberg transforms at comparison time. + struct PartitionTerm { String column; String function; std::optional argument; + std::optional time_zone; - bool operator==(const PartitionKeyTerm & other) const + bool operator==(const PartitionTerm & other) const { return column == other.column && function == other.function && argument == other.argument; } }; - /// Parse a PARTITION BY AST into per-column terms. Returns std::nullopt if any top-level term is - /// not a bare column or a function of exactly one column (nested/multi-column expressions), so the - /// caller can only satisfy such a key by an exact match. - std::optional> parsePartitionKeyTerms(const ASTPtr & partition_key_ast) + /// Parse a PARTITION BY AST into one term per top-level expression. A bare column becomes a term + /// with an empty function; a function keeps its single column argument (the last one seen) plus any + /// integer / timezone literal. Elements that are neither a column nor a function are skipped. + std::vector parsePartitionTerms(const ASTPtr & partition_key_ast) { - std::vector terms; + std::vector terms; if (!partition_key_ast) return terms; - bool ok = true; auto parse_one = [&](const ASTPtr & element) { if (const auto * ident = element->as()) { - terms.push_back({ident->name(), "", std::nullopt}); + terms.push_back({ident->name(), "", std::nullopt, std::nullopt}); return; } const auto * func = element->as(); if (!func) - { - ok = false; return; - } - std::optional column; - std::optional argument; - size_t identifiers = 0; + PartitionTerm term; + term.function = func->name; for (const auto & child : func->children) { const auto * expression_list = child->as(); @@ -623,27 +619,17 @@ namespace for (const auto & arg : expression_list->children) { if (const auto * id = arg->as()) - { - column = id->name(); - ++identifiers; - } + term.column = id->name(); else if (const auto * lit = arg->as()) { - /// A string literal is an optional timezone argument and does not affect the - /// term; only an integer literal is the bucket/truncate width. - if (lit->value.getType() != Field::Types::String) - argument = lit->value.safeGet(); + if (lit->value.getType() == Field::Types::String) + term.time_zone = lit->value.safeGet(); + else + term.argument = lit->value.safeGet(); } - else - ok = false; } } - if (identifiers != 1) - { - ok = false; - return; - } - terms.push_back({*column, func->name, argument}); + terms.push_back(std::move(term)); }; if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") @@ -656,8 +642,6 @@ namespace else parse_one(partition_key_ast); - if (!ok) - return std::nullopt; return terms; } @@ -704,7 +688,7 @@ namespace /// value. Functions carrying an integer argument (bucket/truncate and similar) are not order /// preserving in general, so they can only be accepted by an exact structural match. bool partitionTermIsConstantOverBounds( - const PartitionKeyTerm & term, const DataTypePtr & column_type, + const PartitionTerm & term, const DataTypePtr & column_type, const Field & min_value, const Field & max_value, const ContextPtr & context) { if (term.function.empty() || term.function == "identity") @@ -733,83 +717,6 @@ namespace #if USE_AVRO namespace { - /// One top-level term of the source MergeTree PARTITION BY, canonicalized for comparison against a - /// destination Iceberg partition field. `transform` holds the ClickHouse function name that - /// parseTransformAndArgument yields for the equivalent Iceberg transform ("toRelativeDayNum", - /// "identity", "icebergBucket", ...); std::nullopt marks a term that is not directly - /// Iceberg-representable and can therefore only satisfy a destination field via the dynamic proof. - struct SourcePartitionTerm - { - String column; - std::optional transform; - std::optional argument; - std::optional time_zone; - }; - - std::vector parseSourcePartitionTerms(const ASTPtr & partition_key_ast) - { - /// The same functions getPartitionField maps 1:1 to an Iceberg transform. - static const std::unordered_set iceberg_representable = { - "identity", "toYearNumSinceEpoch", "toMonthNumSinceEpoch", - "toRelativeDayNum", "toRelativeHourNum", "icebergBucket", "icebergTruncate"}; - - std::vector terms; - if (!partition_key_ast) - return terms; - - auto parse_one = [&](const ASTPtr & element) - { - if (const auto * ident = element->as()) - { - terms.push_back({ident->name(), std::optional("identity"), std::nullopt, std::nullopt}); - return; - } - - const auto * func = element->as(); - if (!func) - return; - - std::optional column; - std::optional argument; - std::optional time_zone; - for (const auto & child : func->children) - { - const auto * expression_list = child->as(); - if (!expression_list) - continue; - for (const auto & arg : expression_list->children) - { - const auto * lit = arg->as(); - if (const auto * id = arg->as()) - column = id->name(); - else if (lit && lit->value.getType() == Field::Types::String) - time_zone = lit->value.safeGet(); - else if (lit) - argument = lit->value.safeGet(); - } - } - if (!column) - return; - - std::optional transform; - if (iceberg_representable.contains(func->name)) - transform = func->name; - terms.push_back({*column, transform, argument, time_zone}); - }; - - if (const auto * func = partition_key_ast->as(); func && func->name == "tuple") - { - for (const auto & child : func->children) - if (const auto * expression_list = child->as()) - for (const auto & element : expression_list->children) - parse_one(element); - } - else - parse_one(partition_key_ast); - - return terms; - } - /// Apply the destination Iceberg transform (rebuilt as a ClickHouse function, mirroring /// ChunkPartitioner and the commit path) to a 2-row column holding [cast(min), cast(max)] and /// return whether the endpoints map to different values, i.e. the source partition would be split @@ -893,7 +800,7 @@ namespace const auto actual_fields = partition_spec_json->getArray(Iceberg::f_fields); const size_t actual_size = actual_fields ? actual_fields->size() : 0; - const auto source_terms = parseSourcePartitionTerms(source_metadata->getPartitionKeyAST()); + const auto source_terms = parsePartitionTerms(source_metadata->getPartitionKeyAST()); /// A partitioned source cannot be exported into an unpartitioned Iceberg table: there is no /// destination partitioning to satisfy and the result would silently drop the source layout. @@ -912,6 +819,11 @@ namespace const auto destination_sample = destination_metadata->getSampleBlockNonMaterialized(); const String partition_timezone = context->getSettingsRef()[Setting::iceberg_partition_timezone]; + /// ClickHouse functions that map 1:1 to an Iceberg transform; only these can match structurally. + static const std::unordered_set iceberg_representable = { + "identity", "toYearNumSinceEpoch", "toMonthNumSinceEpoch", + "toRelativeDayNum", "toRelativeHourNum", "icebergBucket", "icebergTruncate"}; + /// year/month/day/hour depend on the timezone (for DateTime); the structural fast path is only /// sound when the source and destination evaluate them in the same timezone. static const std::unordered_set timezone_sensitive_transforms = { @@ -948,8 +860,12 @@ namespace bool matched_structurally = false; for (const auto & term : source_terms) { - if (term.column != column || !term.transform || !dest_canonical - || *term.transform != dest_canonical->transform_name || term.argument != dest_argument) + /// A bare source column canonicalizes to "identity"; other functions match a destination + /// transform only when they are Iceberg-representable and their ClickHouse name equals + /// the one the transform maps to. + const String source_function = term.function.empty() ? "identity" : term.function; + if (term.column != column || !dest_canonical || !iceberg_representable.contains(source_function) + || source_function != dest_canonical->transform_name || term.argument != dest_argument) continue; const auto & transform_name = dest_canonical->transform_name; @@ -1074,23 +990,21 @@ namespace if (ast_to_string(source_key_ast) == ast_to_string(destination_key_ast)) return; - const auto destination_terms = parsePartitionKeyTerms(destination_key_ast); - if (!destination_terms) - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition: the destination partition key is not a supported per-column " - "expression and does not match the source partition key."); + /// A hive destination is always partitioned by bare columns (an expression key is rejected at + /// table creation), so every destination term is a single column here. + const auto destination_terms = parsePartitionTerms(destination_key_ast); /// Unpartitioned destination: a single output partition trivially holds every source partition. - if (destination_terms->empty()) + if (destination_terms.empty()) return; - const auto source_terms = parsePartitionKeyTerms(source_key_ast).value_or(std::vector{}); + const auto source_terms = parsePartitionTerms(source_key_ast); const auto & source_partition_key = source_metadata->getPartitionKey(); const auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(source_partition_key); const auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(source_partition_key); - for (const auto & term : *destination_terms) + for (const auto & term : destination_terms) { /// Fast path: the source already partitions by the same expression on this column, so every /// source partition is single-valued for this destination term by construction.