diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 687029b9adc6..5bbe09b632f8 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -24,6 +24,13 @@ 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 + +The source partition must not be split in the destination. This is validated at schedule time through two mechanisms: + +1. Identical expressions; or +2. Destination partition expression is monotonically increasing in the min/max column ranges and the destination expression columns are a subset of the source expression; + ### On plain object storage exports: Each MergeTree part will become a separate file with the following name convention: `//_.`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `/commit__`. diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index d58e9d534949..f58395bf98b0 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -249,6 +249,11 @@ struct ExportReplicatedMergeTreePartitionManifest std::optional parquet_row_group_size; std::optional parquet_row_group_size_bytes; + /// this is a controversial setting. As far as I can infer from the iceberg docs, the transforms are always UTC. + /// this setting allows to specify different timezones. Since it is already implemented, we must respect it. + /// At the same time, we don't allow transforms with timezones, so this is very weird. + std::optional iceberg_partition_timezone; + std::string toJsonString() const { Poco::JSON::Object json; @@ -290,6 +295,8 @@ struct ExportReplicatedMergeTreePartitionManifest json.set("parquet_row_group_size", *parquet_row_group_size); if (parquet_row_group_size_bytes) json.set("parquet_row_group_size_bytes", *parquet_row_group_size_bytes); + if (iceberg_partition_timezone) + json.set("iceberg_partition_timezone", *iceberg_partition_timezone); std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM oss.exceptions(std::ios::failbit); Poco::JSON::Stringifier::stringify(json, oss); @@ -378,6 +385,11 @@ struct ExportReplicatedMergeTreePartitionManifest manifest.parquet_row_group_size_bytes = json->getValue("parquet_row_group_size_bytes"); } + if (json->has("iceberg_partition_timezone")) + { + manifest.iceberg_partition_timezone = json->getValue("iceberg_partition_timezone"); + } + return manifest; } }; diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 3f85ebc0e1fb..02c245ca4776 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -7,15 +7,26 @@ #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 +#include +#include +#include #if USE_AVRO #include @@ -75,6 +86,9 @@ namespace ErrorCodes namespace Setting { extern const SettingsBool export_merge_tree_part_allow_lossy_cast; +#if USE_AVRO + extern const SettingsTimezone iceberg_partition_timezone; +#endif } namespace FailPoints @@ -131,20 +145,49 @@ namespace ExportPartitionUtils } Block getPartitionSourceBlockForIcebergCommit( - MergeTreeData & storage, const String & partition_id) + MergeTreeData & storage, const String & partition_id, const std::vector & exported_part_names) { auto lock = storage.readLockParts(); const auto parts = storage.getDataPartsVectorInPartitionForInternalUsage( - MergeTreeDataPartState::Active, partition_id, lock); + {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}, partition_id, lock); - if (parts.empty()) + /// Only look at the parts being exported. These parts are guaranteed to map to a single partition. + /// Parts that were later inserted shall be ignored + const std::unordered_set exported(exported_part_names.begin(), exported_part_names.end()); + IMergeTreeDataPart::MinMaxIndex minmax; + for (const auto & part : parts) + if (exported.contains(part->name) && part->minmax_idx) + minmax.merge(*part->minmax_idx); + + if (!minmax.initialized) 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); + + 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); + + if (minmax.hyperrectangle.size() != minmax_column_types.size()) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "Cannot derive Iceberg partition values: the exported parts of partition '{}' hold min/max " + "statistics for {} columns, but the partition key has {}.", + partition_id, minmax.hyperrectangle.size(), minmax_column_types.size()); + + /// When the query was scheduled, we validated that dst_expression(min) == dst_expression(max). + /// Therefore, we can use only the min value, no need for the max. + Block block; + for (size_t i = 0; i < minmax_column_types.size(); ++i) + { + auto column = minmax_column_types[i]->createColumn(); + column->insert(minmax.hyperrectangle[i].left); + block.insert(ColumnWithTypeAndName(column->getPtr(), minmax_column_types[i], minmax_column_names[i])); + } + + return block; } ContextPtr getContextCopyWithTaskSettings(const ContextPtr & context, const ExportReplicatedMergeTreePartitionManifest & manifest) @@ -191,7 +234,12 @@ namespace ExportPartitionUtils /// schema drifts to a lossy target between scheduling and execution. context_copy->setSetting("export_merge_tree_part_allow_lossy_cast", manifest.allow_lossy_cast); - return context_copy; + if (manifest.iceberg_partition_timezone) + { + context_copy->setSetting("iceberg_partition_timezone", *manifest.iceberg_partition_timezone); + } + + return context_copy; } /// Collect all the exported paths from the processed parts @@ -263,6 +311,9 @@ namespace ExportPartitionUtils auto context = Context::createCopy(context_in); context->setSetting("write_full_path_in_iceberg_metadata", manifest.write_full_path_in_iceberg_metadata); + if (manifest.iceberg_partition_timezone) + context->setSetting("iceberg_partition_timezone", *manifest.iceberg_partition_timezone); + /// Failpoint used by integration tests to force persistent commit failure and exercise /// the commit-attempts budget / FAILED state transition. fiu_do_on(FailPoints::export_partition_commit_always_throw, @@ -316,7 +367,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 +565,253 @@ namespace ExportPartitionUtils ops.emplace_back(zkutil::makeSetRequest(last_exception_path, entry.toJsonString(), -1)); } +namespace +{ + /// Terms of a given partition key expression. + /// E.g. for "PARTITION BY (toYYYYMM(ts), country)", the terms are "toYYYYMM(ts)" and "country". + 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 + && time_zone == other.time_zone; + } + }; + + 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; + /// A nested function such as toYYYYMM(toDate(ts)) is not unwrapped: it yields no column and + /// stays unmatched, as the monotonicity proof only handles a single function of a single column. + for (const auto & arg : expression_list->children) + { + if (const auto * id = arg->as()) + term.column = id->name(); + /// Integer literal is the bucket/truncate width; a string literal is a date transform's timezone. + /// In theory, the string literal could mean a different thing, but among the partition expressions iceberg supports + /// time_zone is the only option + 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; + } + + /// asserts that the source column maps to a single destination partition by checking the monotonicity on the min/max ranges + void verifyColumnMapsToSinglePartition( + const String & column, + const String & function_name, + std::optional argument, + const std::optional & time_zone, + const Names & minmax_column_names, + const DataTypes & minmax_column_types, + const Block & destination_sample, + const IMergeTreeDataPart::MinMaxIndex & minmax, + const String & partition_id, + const ContextPtr & context) + { + 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: the destination partition expression uses column '{}', which is " + "not part of the source MergeTree partition key.", column); + 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 destination 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: column '{}' is Nullable, so a NULL forms a separate destination " + "partition; partition the source by the matching destination partition expression.", column); + + if (!minmax.initialized || slot >= minmax.hyperrectangle.size()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: no min/max statistics available for column '{}' in partition " + "'{}'; cannot validate partitioning.", column, partition_id); + const auto & min_value = minmax.hyperrectangle[slot].left; + const auto & max_value = minmax.hyperrectangle[slot].right; + + if (!destination_sample.has(column)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: destination column '{}' not found.", column); + const auto destination_type = destination_sample.getByName(column).type; + + /// The written value is transform(cast(source)), and the endpoints only bound the interior rows if + /// the cast preserves order. Preserving every value is not enough: Int -> String loses nothing, yet + /// "10" sorts before "2", so an interior row can fall outside the endpoints. Without a cast the + /// order holds trivially; otherwise CAST must prove it over the partition's actual range. + const bool is_cast_needed = !source_type->equals(*destination_type); + if (is_cast_needed) + { + const auto cast_function + = createInternalCast({source_type, column}, destination_type, CastType::nonAccurate, {}, context); + if (!cast_function->hasInformationAboutMonotonicity() + || !cast_function->getMonotonicityForRange(*source_type, min_value, max_value).is_monotonic) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition '{}': values of column '{}' cross a non-monotonic cast boundary to " + "the destination type {}, so it spans multiple destination partitions.", + partition_id, column, destination_type->getName()); + } + + auto values_column = source_type->createColumn(); + values_column->insert(min_value); + values_column->insert(max_value); + const auto cast_column = castColumn({std::move(values_column), source_type, column}, destination_type); + + Field cast_min; + Field cast_max; + cast_column->get(0, cast_min); + cast_column->get(1, cast_max); + + /// A bare destination column is single-valued iff its two cast endpoints are equal; otherwise the + /// destination transform applied to [cast(min), cast(max)] must be monotonic (so the endpoints + /// bound every interior row; a hash such as bucket is not) and map both endpoints to one value. + bool spans_multiple_partitions; + if (function_name.empty() || function_name == "identity") + { + spans_multiple_partitions = cast_min != cast_max; + } + else + { + auto resolver = FunctionFactory::instance().get(function_name, context); + ColumnsWithTypeAndName arguments; + if (argument) + arguments.push_back({DataTypeUInt64().createColumnConst(2, *argument), std::make_shared(), "width"}); + arguments.push_back({cast_column, destination_type, column}); + if (time_zone) + arguments.push_back({DataTypeString().createColumnConst(2, *time_zone), std::make_shared(), "timezone"}); + + const auto function = resolver->build(arguments); + if (!function->hasInformationAboutMonotonicity() + || !function->getMonotonicityForRange(*destination_type, cast_min, cast_max).is_monotonic) + { + spans_multiple_partitions = true; + } + else + { + const auto result = function->execute(arguments, function->getResultType(), /*input_rows_count=*/ 2, /*dry_run=*/ false); + Field first; + Field second; + result->get(0, first); + result->get(1, second); + spans_multiple_partitions = first != second; + } + } + + if (spans_multiple_partitions) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition '{}': the source partition might span multiple destination partitions " + "for column '{}'. A source MergeTree partition must map to a single destination partition.", + partition_id, column); + } + + bool sourceHasMatchingPartitionTerm( + const PartitionTerm & term, + const std::vector & source_terms, + const Names & minmax_column_names, + const DataTypes & minmax_column_types, + const Block & destination_sample) + { + if (std::find(source_terms.begin(), source_terms.end(), term) == source_terms.end()) + return false; + + if (term.function.empty()) + return true; + + const auto it = std::find(minmax_column_names.begin(), minmax_column_names.end(), term.column); + return it != minmax_column_names.end() && destination_sample.has(term.column) + && minmax_column_types[it - minmax_column_names.begin()]->equals(*destination_sample.getByName(term.column).type); + } + + /// Asserts every destination partition term maps the whole source partition to a single destination + /// partition, either because the source already partitions by that term or because the min/max proof holds. + void verifyPartitionTermsCompatibility( + const std::vector & destination_terms, + const StorageMetadataPtr & source_metadata, + const StorageMetadataPtr & destination_metadata, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context) + { + const auto source_terms = parsePartitionTerms(source_metadata->getPartitionKeyAST()); + + 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(); + + /// The bounds of the whole partition are the union of its parts' bounds, so fold them once here + /// rather than per term. + IMergeTreeDataPart::MinMaxIndex minmax; + for (const auto & part : parts) + if (part->minmax_idx) + minmax.merge(*part->minmax_idx); + + for (const auto & term : destination_terms) + { + if (sourceHasMatchingPartitionTerm(term, source_terms, minmax_column_names, minmax_column_types, destination_sample)) + continue; + + verifyColumnMapsToSinglePartition(term.column, term.function, term.argument, term.time_zone, + minmax_column_names, minmax_column_types, destination_sample, minmax, partition_id, context); + } + } +} + #if USE_AVRO 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); @@ -551,89 +845,82 @@ namespace ExportPartitionUtils } if (!current_schema_json || !partition_spec_json) - return; + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: destination metadata is malformed, " + "current-schema-id '{}' or default-spec-id '{}' does not resolve to a schema/spec.", + original_schema_id, partition_spec_id); - /// 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 expected_spec = Iceberg::getPartitionSpec( - partition_key_ast, column_name_to_source_id).first; - expected_fields = expected_spec->getArray(Iceberg::f_fields); - } - catch (const Exception & e) - { - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition to Iceberg table: the source MergeTree partition " - "key cannot be represented as an Iceberg partition spec: {}", e.message()); - } - const auto 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; + const size_t actual_size = actual_fields ? actual_fields->size() : 0; + const String partition_timezone = context->getSettingsRef()[Setting::iceberg_partition_timezone]; + + /// Rebuild the destination spec as ClickHouse partition terms, mirroring ChunkPartitioner and the commit + /// path, so the same compatibility rule applies as for a plain object storage destination. An empty spec + /// yields no terms: an unpartitioned table has a single partition that every source part maps to. + std::vector destination_terms; + destination_terms.reserve(actual_size); + for (UInt32 i = 0; i < actual_size; ++i) + { + const auto af = actual_fields->getObject(i); + const auto dest_transform = af->getValue(Iceberg::f_transform); + const String column = source_id_to_name(af->getValue(Iceberg::f_source_id)); + const auto transform_and_argument = Iceberg::parseTransformAndArgument(dest_transform, partition_timezone); - 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) + if (!transform_and_argument) 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: destination field on column '{}' uses transform " + "'{}', which has no ClickHouse equivalent.", column, dest_transform); + + /// A bare source column parses to an empty function name, which is what Iceberg calls identity. + destination_terms.push_back({ + column, + transform_and_argument->transform_name == "identity" ? "" : transform_and_argument->transform_name, + transform_and_argument->argument + ? std::optional(static_cast(*transform_and_argument->argument)) : std::nullopt, + transform_and_argument->time_zone}); } + + verifyPartitionTermsCompatibility( + destination_terms, source_metadata, destination_metadata, parts, partition_id, context); } #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. + if (ast_to_string(source_key_ast) == ast_to_string(destination_key_ast)) + return; + + /// If the partition keys are not identical, we need to verify that the source partition maps to a single destination partition + verifyPartitionTermsCompatibility( + parsePartitionTerms(destination_key_ast), source_metadata, destination_metadata, parts, partition_id, context); + } + 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..e593f883a817 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,9 @@ 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. - /// - /// 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. + /// Get the min/max values from the partition expression columns 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, @@ -89,23 +83,34 @@ namespace ExportPartitionUtils const std::string & exception_message, const LoggerPtr & log); - /// Validates that source columns can be exported into the destination with the - /// same positional CAST matching as `INSERT INTO dest SELECT * FROM src`. Lossy - /// casts are rejected unless `export_merge_tree_part_allow_lossy_cast` is set. - /// Throws BAD_ARGUMENTS on any violation. void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, const StorageMetadataPtr & destination_metadata, const StorageID & destination_storage_id, const ContextPtr & context); + 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/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 1623611fec1b..32f7749594f5 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -235,6 +235,7 @@ namespace Setting extern const SettingsBool allow_insert_into_iceberg; extern const SettingsUInt64 iceberg_insert_max_bytes_in_data_file; extern const SettingsUInt64 iceberg_insert_max_rows_in_data_file; + extern const SettingsTimezone iceberg_partition_timezone; } @@ -8393,11 +8394,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & if (!dest_storage->supportsImport(query_context)) throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Destination storage {} does not support MergeTree parts or uses unsupported partitioning", dest_storage->getName()); - auto query_to_string = [] (const ASTPtr & ast) - { - return ast ? ast->formatWithSecretsOneLine() : ""; - }; - auto src_snapshot = getInMemoryMetadataPtr(); auto destination_snapshot = dest_storage->getInMemoryMetadataPtr(); @@ -8405,14 +8401,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & ExportPartitionUtils::verifyExportSchemaCastable( src_snapshot, destination_snapshot, dest_storage->getStorageID(), query_context); - /// Iceberg partition compatibility is checked below; here we only need the - /// partition-key ASTs to match (partition-column types follow the lossy-cast gate). - if (!dest_storage->isDataLake()) - { - if (query_to_string(src_snapshot->getPartitionKeyAST()) != query_to_string(destination_snapshot->getPartitionKeyAST())) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); - } - zkutil::ZooKeeperPtr zookeeper = getZooKeeperAndAssertNotReadonly(); const String partition_id = getPartitionIDFromQuery(command.partition, query_context); @@ -8530,6 +8518,7 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & manifest.filename_pattern = query_context->getSettingsRef()[Setting::export_merge_tree_part_filename_pattern].value; manifest.write_full_path_in_iceberg_metadata = query_context->getSettingsRef()[Setting::write_full_path_in_iceberg_metadata]; manifest.allow_lossy_cast = query_context->getSettingsRef()[Setting::export_merge_tree_part_allow_lossy_cast]; + manifest.iceberg_partition_timezone = query_context->getSettingsRef()[Setting::iceberg_partition_timezone].toString(); if (dest_storage->isDataLake()) { @@ -8564,7 +8553,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 +8571,15 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Data lake export requires Avro support"); #endif } + else + { + 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/helpers/export_partition_helpers.py b/tests/integration/helpers/export_partition_helpers.py index 46e73e8a04e6..04c9cb244757 100644 --- a/tests/integration/helpers/export_partition_helpers.py +++ b/tests/integration/helpers/export_partition_helpers.py @@ -137,15 +137,17 @@ def make_rmt( partition_by, replica_name="r1", order_by="tuple()", + extra_settings="", ): """Create a ReplicatedMergeTree table with block-number settings.""" + settings = f"{_BLOCK_SETTINGS}, {extra_settings}" if extra_settings else _BLOCK_SETTINGS node.query( f""" CREATE TABLE {name} ({columns}) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{name}', '{replica_name}') PARTITION BY {partition_by} ORDER BY {order_by} - SETTINGS {_BLOCK_SETTINGS} + SETTINGS {settings} """ ) 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..b0eb64494f45 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,64 +834,57 @@ 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): """ Verify that EXPORT PARTITION throws BAD_ARGUMENTS synchronously when the MergeTree partition key does not match the Iceberg table's partition spec, - and is accepted without error when the keys match. + and is accepted without error when the destination is satisfiable. Three cases: - 1. Column mismatch – MergeTree PARTITION BY year, Iceberg PARTITION BY id - 2. Count mismatch – MergeTree PARTITION BY year, Iceberg unpartitioned + 1. Column mismatch – MergeTree PARTITION BY year, Iceberg PARTITION BY id (must be rejected) + 2. Unpartitioned dst – MergeTree PARTITION BY year, Iceberg unpartitioned (accepted: the source is + flattened into the single empty Iceberg partition) 3. Matching keys – both PARTITION BY year (must be accepted) """ node = cluster.instances["replica1"] @@ -921,27 +918,31 @@ def test_partition_key_compatibility_check(cluster): f"Expected BAD_ARGUMENTS for partition column mismatch, got: {error!r}" ) - # --- Case 2: Iceberg unpartitioned but MergeTree PARTITION BY year --- - iceberg_count_mismatch = f"iceberg_count_mismatch_{uid}" + # --- Case 2: Iceberg unpartitioned, MergeTree PARTITION BY year --- + # An unpartitioned Iceberg table has a single (empty) partition, so a partitioned source is + # flattened into it and the export is accepted; the partition-column values survive as data. + iceberg_unpartitioned = f"iceberg_unpartitioned_{uid}" node.query( f""" - CREATE TABLE {iceberg_count_mismatch} + CREATE TABLE {iceberg_unpartitioned} (id Int64, year Int32) ENGINE = IcebergS3( - 'http://minio1:9001/root/data/{iceberg_count_mismatch}/', + 'http://minio1:9001/root/data/{iceberg_unpartitioned}/', 'minio', 'ClickHouse_Minio_P@ssw0rd' ) SETTINGS s3_retry_attempts = 3 """ ) - error = node.query_and_get_error( - f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {iceberg_count_mismatch}", + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {iceberg_unpartitioned}", settings={"allow_insert_into_iceberg": 1}, ) - assert "BAD_ARGUMENTS" in error, ( - f"Expected BAD_ARGUMENTS for partition count mismatch, got: {error!r}" - ) + wait_for_export_status(node, mt_table, iceberg_unpartitioned, "2020", "COMPLETED") + count = int(node.query(f"SELECT count() FROM {iceberg_unpartitioned}").strip()) + assert count == 2, f"Expected 2 rows in unpartitioned Iceberg table after export, got {count}" + result = node.query(f"SELECT id, year FROM {iceberg_unpartitioned} ORDER BY id").strip() + assert result == "1\t2020\n2\t2020", f"Unexpected data in unpartitioned Iceberg table:\n{result}" # --- Case 3: Matching partition keys (both PARTITION BY year) --- iceberg_match = f"iceberg_match_{uid}" @@ -964,6 +965,279 @@ 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}, + # Identical expressions on a Nullable column: accepted structurally. The min/max proof refuses + # Nullable (a NULL forms its own destination partition and the endpoints cannot rule it out), + # so this only passes because the source already groups by exactly this transform. DateTime64(6) + # round-trips through the Iceberg schema unchanged, which the structural type check requires. + {"name": "nullable_exact_day", "columns": "id Int64, event_time Nullable(DateTime64(6))", + "source_key": "toRelativeDayNum(event_time)", "dest_key": "toRelativeDayNum(event_time)", + "rows": "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')", + "source_settings": "allow_nullable_key = 1", "expect_ok": True}, + # Same, for identity, which is exempt from the structural type check. + {"name": "nullable_exact_identity", "columns": "id Int64, k Nullable(Int64)", + "source_key": "k", "dest_key": "k", "rows": "(1, 10), (2, 10)", + "source_settings": "allow_nullable_key = 1", "expect_ok": True, + "verify": [("k", "k")]}, + # A Nullable column without identical expressions falls to the min/max proof, which cannot see + # NULLs, so it is rejected. + {"name": "nullable_no_match", "columns": "id Int64, event_time Nullable(DateTime64(6))", + "source_key": "toYYYYMM(event_time)", "dest_key": "toRelativeDayNum(event_time)", + "rows": "(1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')", + "source_settings": "allow_nullable_key = 1", "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_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 +1916,108 @@ 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)]) and source_settings (extra MergeTree settings).""" + 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", + extra_settings=case.get("source_settings", "")) + 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)") @@ -1661,6 +2025,91 @@ def test_export_partition_bucket_transform_metadata_matches_data(cluster): make_iceberg_s3(node, iceberg_table, "id Int64, key String", partition_by="icebergBucket(16, 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 bucket transform, got: {error!r}" + ) + + +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_value_preserving_cast_not_order_preserving_rejected(cluster): + """Int64 -> String keeps every value, but not their order: 2 and 29 are the endpoints of the + source partition, yet the interior value 10 casts to a string that sorts outside them. The + endpoints truncate to '2' while 10 truncates to '1', so the partition spans two destination + buckets and must be rejected instead of being waved through as a lossless cast.""" + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_cast_order_{uid}" + iceberg_table = f"iceberg_cast_order_{uid}" + + make_rmt(node, mt_table, "id Int64, k Int64", "intDiv(k, 100)", + replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, 2), (2, 10), (3, 29)") + + make_iceberg_s3(node, iceberg_table, "id Int64, k String", + partition_by="icebergTruncate(1, k)") + + 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 non-order-preserving cast, got: {error!r}" + ) + + +def test_export_partition_order_preserving_cast_accepted(cluster): + """The same shape as the rejected case, but with all values sharing a digit count: Int64 -> + String is order-preserving over [20, 29], so the endpoints do bound the interior and the whole + source partition truncates to the single destination bucket '2'.""" + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_cast_order_ok_{uid}" + iceberg_table = f"iceberg_cast_order_ok_{uid}" + + make_rmt(node, mt_table, "id Int64, k Int64", "intDiv(k, 100)", + replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, 20), (2, 25), (3, 29)") + + make_iceberg_s3(node, iceberg_table, "id Int64, k String", + partition_by="icebergTruncate(1, k)") + pid = first_partition_id(node, mt_table) node.query( f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", @@ -1668,21 +2117,86 @@ def test_export_partition_bucket_transform_metadata_matches_data(cluster): ) wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") - count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) - assert count == 2, f"Expected 2 rows after export, got {count}" + src = node.query(f"SELECT id, toString(k) FROM {mt_table} ORDER BY id").strip() + dst = node.query(f"SELECT id, k FROM {iceberg_table} ORDER BY id").strip() + assert src == dst, f"destination rows differ from source:\n{src}\n---\n{dst}" - 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." + assert_iceberg_partition_metadata(node, iceberg_table, uid, [("k", "icebergTruncate(1, k)")]) + + +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}" ) - query_id = f"bucket_xform_{uid}" + +def test_export_partition_commit_uses_exported_parts_not_new_inserts(cluster): + """The deferred commit derives the Iceberg partition value only from the exact exported parts + recorded in the manifest, never from parts inserted/merged into the source partition after + scheduling. A month-partitioned source exports one day into a day-partitioned destination (a + data-dependent acceptance); while the commit is wedged, an earlier day is inserted and merged in, + so the only active part now spans both days with its min at the new day. The commit must still + stamp the exported day (the exported part is found among Outdated parts by name), not the merged-in + earlier day, so the metadata matches the exported data files.""" + node = cluster.instances["replica1"] + uid = unique_suffix() + mt_table = f"mt_commit_parts_{uid}" + iceberg_table = f"iceberg_commit_parts_{uid}" + + make_rmt(node, mt_table, "id Int64, event_date Date", "toYYYYMM(event_date)", replica_name="replica1") + node.query(f"INSERT INTO {mt_table} VALUES (1, '2024-03-20'), (2, '2024-03-20')") + make_iceberg_s3(node, iceberg_table, "id Int64, event_date Date", + partition_by="toRelativeDayNum(event_date)") + + exported_day = int(node.query("SELECT toRelativeDayNum(toDate('2024-03-20'))").strip()) + injected_day = int(node.query("SELECT toRelativeDayNum(toDate('2024-03-05'))").strip()) + + node.query("SYSTEM ENABLE FAILPOINT export_partition_commit_always_throw") + try: + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '202403' TO TABLE {iceberg_table}" + f" SETTINGS allow_insert_into_iceberg = 1" + ) + # The commit is attempted only after every part is exported, so a non-zero exception count + # means the data files are written and the commit is now wedged by the failpoint. + wait_for_exception_count(node, mt_table, iceberg_table, "202403", min_exception_count=1, timeout=90) + + # Insert an earlier day into the same month partition and merge: the merged active part spans + # both days with min = the injected (earlier) day, while the exported part becomes Outdated. + node.query(f"INSERT INTO {mt_table} VALUES (3, '2024-03-05')") + node.query(f"OPTIMIZE TABLE {mt_table} PARTITION ID '202403' FINAL") + finally: + node.query("SYSTEM DISABLE FAILPOINT export_partition_commit_always_throw") + + wait_for_export_status(node, mt_table, iceberg_table, "202403", "COMPLETED", timeout=90) + + # The exported data files hold only 2024-03-20; the metadata day must match them. + query_id = f"commit_parts_{uid}" node.query( f"SELECT * FROM {iceberg_table}", query_id=query_id, @@ -1691,10 +2205,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"