diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 687029b9adc6..adc0bb7c74f9 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -24,10 +24,32 @@ 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__`. +#### 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 3f85ebc0e1fb..751f620fdb56 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -7,19 +7,31 @@ #include "Storages/ExportReplicatedMergeTreePartitionManifest.h" #include "Storages/ExportReplicatedMergeTreePartitionTaskEntry.h" #include +#include #include #include #include #include #include #include +#include #include #include #include +#include +#include +#include +#include +#include +#include #if USE_AVRO #include #include +#include +#include +#include +#include #endif namespace ProfileEvents @@ -75,6 +87,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 @@ -131,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); - - if (parts.empty()) + {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}, partition_id, lock); + + /// 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) @@ -316,7 +372,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); } const auto destination_commit_info = destination_storage->commitExportPartitionTransaction( @@ -514,10 +570,183 @@ namespace ExportPartitionUtils ops.emplace_back(zkutil::makeSetRequest(last_exception_path, entry.toJsonString(), -1)); } +namespace +{ + /// 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 PartitionTerm & other) const + { + return column == other.column && function == other.function && argument == other.argument; + } + }; + + /// 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; + if (!partition_key_ast) + return terms; + + auto parse_one = [&](const ASTPtr & element) + { + if (const auto * ident = element->as()) + { + terms.push_back({ident->name(), "", std::nullopt, std::nullopt}); + return; + } + + const auto * func = element->as(); + if (!func) + return; + + PartitionTerm term; + term.function = func->name; + 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()) + term.column = id->name(); + else if (const auto * lit = arg->as()) + { + if (lit->value.getType() == Field::Types::String) + term.time_zone = lit->value.safeGet(); + else + term.argument = lit->value.safeGet(); + } + } + } + terms.push_back(std::move(term)); + }; + + 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; + } + + std::optional> calculatePartitionColumnMinMax( + const MergeTreeData::DataPartsVector & parts, size_t slot) + { + 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 (!bounds) + { + bounds.emplace(range.left, range.right); + } + else + { + if (range.left < bounds->first) + bounds->first = range.left; + if (bounds->second < range.right) + bounds->second = range.right; + } + } + 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 + /// 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 PartitionTerm & 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 ColumnsWithTypeAndName arguments{{std::move(values_column), column_type, term.column}}; + const auto function = resolver->build(arguments); + + /// 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; + + return !endpointsMapToDifferentValues(*function, arguments, /*num_rows=*/ 2); + } +} + #if USE_AVRO +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 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 resolver = 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"}); + + return endpointsMapToDifferentValues(*resolver->build(arguments), arguments, num_rows); + } +} + 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); @@ -553,87 +782,265 @@ 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 = 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. + 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]; + + /// 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 = { + "toYearNumSinceEpoch", "toMonthNumSinceEpoch", "toRelativeDayNum", "toRelativeHourNum"}; + + auto column_explicit_time_zone = [](const DataTypePtr & type) -> String { - 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()); - } + if (const auto * tz = dynamic_cast(type.get()); tz && tz->hasExplicitTimeZone()) + return tz->getTimeZone().getTimeZone(); + return ""; + }; - 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; - - 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); - - 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) + for (UInt32 i = 0; i < actual_size; ++i) + { + 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). + /// 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) + { + /// 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; + if (transform_name != "identity") + { + 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; + + /// 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); + + const auto bounds = calculatePartitionColumnMinMax(parts, slot); + if (!bounds) 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: 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, + "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 (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 " + "single Iceberg partition.", partition_id, column, dest_transform); } } #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; + + /// 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()) + return; + + 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) + { + /// 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); + + 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, + "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 0bb8acb9bda4..4ecdfd9bfb05 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 @@ -29,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, @@ -99,13 +104,36 @@ 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 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..fd3f54ac6da7 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6692,25 +6692,29 @@ 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`."); + } + + ExportPartitionUtils::verifyExportSchemaCastable( + source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context); + + 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()) { - 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_) { @@ -6745,32 +6749,30 @@ 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_; 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); } - 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 1623611fec1b..215ad3451c69 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -8393,11 +8393,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(); @@ -8405,14 +8400,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); @@ -8564,7 +8551,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); @@ -8578,6 +8569,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_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index ad2deba8de19..2df26c38c61b 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 @@ -806,21 +806,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}", @@ -830,53 +834,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): @@ -964,6 +960,289 @@ 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. Accept cases are + verified end-to-end (data + metadata); reject cases must throw BAD_ARGUMENTS synchronously. + """ + node = cluster.instances["replica1"] + dt = "id Int64, event_time DateTime" + 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): + """ + 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. Accept cases are verified + end-to-end (data + metadata); reject cases must throw BAD_ARGUMENTS. + """ + node = cluster.instances["replica1"] + 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')" + + 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): + """ + 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_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 + 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. @@ -1642,18 +1921,107 @@ def _partition_scalar(partition, field): return value -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 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_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)") @@ -1662,27 +2030,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}" + ) + + +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}" - query_id = f"bucket_xform_{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, @@ -1691,10 +2150,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." ) 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 8d4589292e3c..ba61307466af 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, @@ -1747,3 +1748,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 431df7efbeb7..59f6ebedd979 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; 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"