diff --git a/docs/en/antalya/part_export.md b/docs/en/antalya/part_export.md index ef7a7e25f5f7..5cf8e2aa8b0e 100644 --- a/docs/en/antalya/part_export.md +++ b/docs/en/antalya/part_export.md @@ -53,10 +53,10 @@ Source and destination tables must support positional schema conversion. The fol - **Column types** may differ, as long as the source type is safely castable to the destination type. Set `export_merge_tree_part_allow_lossy_cast = 1` to also permit lossy casts. - **`Tuple` element names** may differ if either the source or destination declares the tuple without named elements: an unnamed `Tuple` (e.g. `Tuple(Int32, Int32)`) is matched against the destination by element position and type only, not by name. For example, exporting from `t Tuple(Int32, Int32)` to `t Tuple(x Int32, y Int32)` is allowed as long as element types match positionally. -The following must match between source and destination: +The following requirements apply to the source and destination: 1. **Column count** - source and destination must have the same number of columns by default. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode. -2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must be representable as an Iceberg partition spec and must match the destination partition fields and transforms. +2. **`PARTITION BY` expressions** - the whole part must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the part's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](/docs/en/antalya/partition_export.md#source-partition-key-compatibility). 3. **The position of every column backing the partition key** - it is not enough for the `PARTITION BY` expressions to be textually identical: every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. If such a column contains a named `Tuple`, its element names must also be declared in the same order (an unnamed `Tuple` on either side is exempt from this, per the allowance above). This comparison is recursive through nested tuples and through container types such as `Array` and `Map`. For example, `CREATE TABLE src (a Int32, b Int32) ... PARTITION BY a` and `CREATE TABLE dst (b Int32, a Int32) ... PARTITION BY a` both have the expression `PARTITION BY a`, but `a` is at position 0 in `src` and position 1 in `dst`. The export is rejected with a `BAD_ARGUMENTS` exception whose message includes `Cannot export to : partition key column 'a' is at position 0 in the source table, but the destination's column at that position is named 'b'`. diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 52b1e901b5d1..334caf82f972 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. Structural match: in case the source and destination are identical, the destination expression is a subset of the source expression or the destination expression can be entirely computed using only constants and the exact values guaranteed (pinned) by the source. +2. Dynamic proof: the destination expression is monotonic over the source partition min/max range. + ### 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__`. @@ -45,10 +52,10 @@ TO TABLE [destination_database.]destination_table ## Requirements -`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following must match: +`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following requirements apply: 1. **Column count** - source and destination must have the same number of columns by default. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode. -2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must match the destination partition fields and transforms. +2. **`PARTITION BY` expressions** - the whole source partition must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the partition's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](#source-partition-key-compatibility). 3. **Partition key column positions and layouts** - every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. Named `Tuple` elements within such a column must also be declared in the same order, including tuples nested inside `Array` or `Map`. This applies even if both tables' `PARTITION BY` expressions are textually identical. See [`EXPORT PART` requirements](/docs/en/antalya/part_export.md#requirements) for a worked example and the corresponding exception message. ## Settings diff --git a/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.cpp b/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.cpp index 3b6c9676c3b7..328840e95688 100644 --- a/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.cpp +++ b/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.cpp @@ -300,16 +300,7 @@ MatchedTrees::Matches matchTrees( } -struct PossiblyMonotonicChain -{ - const ActionsDAG::Node * input_node = nullptr; - std::vector non_const_arg_pos; - bool changes_order = false; - bool is_strict = true; -}; - -/// Build a chain of functions which may be monotonic. -static PossiblyMonotonicChain buildPossiblyMonitinicChain(const ActionsDAG::Node * node) +PossiblyMonotonicChain buildPossiblyMonotonicChain(const ActionsDAG::Node * node) { std::vector chain; @@ -362,8 +353,7 @@ static PossiblyMonotonicChain buildPossiblyMonitinicChain(const ActionsDAG::Node return {node, std::move(chain)}; } -/// Check whether all the function in chain are monotonic -static bool isMonotonicChain(const ActionsDAG::Node * node, PossiblyMonotonicChain & chain) +bool isMonotonicChain(const ActionsDAG::Node * node, PossiblyMonotonicChain & chain) { auto it = chain.non_const_arg_pos.begin(); while (node != chain.input_node) @@ -443,7 +433,7 @@ void applyActionsToSortDescription( if (output == output_to_skip) continue; - auto chain = buildPossiblyMonitinicChain(output); + auto chain = buildPossiblyMonotonicChain(output); if (!chain.input_node) break; diff --git a/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.h b/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.h index c9e14970fb20..9e30c712f1fb 100644 --- a/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.h +++ b/src/Processors/QueryPlan/Optimizations/actionsDAGUtils.h @@ -56,6 +56,22 @@ MatchedTrees::Matches matchTrees( bool check_monotonicity = true, size_t max_size_for_sets_from_tuple_to_compare = 0); +/// A path from a node down to an input, where every function on the path has a single non-constant argument. +/// `non_const_arg_pos` holds the position of that argument for every function on the path, top-down. +struct PossiblyMonotonicChain +{ + const ActionsDAG::Node * input_node = nullptr; + std::vector non_const_arg_pos; + bool changes_order = false; + bool is_strict = true; +}; + +/// Build a chain of functions which may be monotonic. `input_node` is nullptr if the node is not such a chain. +PossiblyMonotonicChain buildPossiblyMonotonicChain(const ActionsDAG::Node * node); + +/// Check whether all the function in chain are monotonic +bool isMonotonicChain(const ActionsDAG::Node * node, PossiblyMonotonicChain & chain); + /// Update SortDescription (inplace) by applying ActionsDAG. /// /// Assuming that sorting properties are fulfilled for inputs, calculate sorting properties for the outputs. diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index 753095900fa8..7857302b1261 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -250,6 +250,11 @@ struct ExportReplicatedMergeTreePartitionManifest std::optional parquet_row_group_size_bytes; std::optional schema_mismatch_mode; + /// 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; @@ -291,6 +296,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); if (schema_mismatch_mode) json.set("schema_mismatch_mode", String(magic_enum::enum_name(*schema_mismatch_mode))); std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM @@ -391,6 +398,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 dc83f80aee0f..3cd33e26db4f 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -7,6 +7,7 @@ #include "Storages/ExportReplicatedMergeTreePartitionManifest.h" #include "Storages/ExportReplicatedMergeTreePartitionTaskEntry.h" #include +#include #include #include #include @@ -23,11 +24,20 @@ #include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include #if USE_AVRO #include -#include +#include #endif namespace ProfileEvents @@ -83,6 +93,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 extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; } @@ -140,20 +153,51 @@ 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)) + minmax.merge(*part->getMinMaxIndex()); + + 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()->getMinMaxIndex()->getBlock(storage); + + const auto metadata_snapshot = storage.getInMemoryMetadataPtr(storage.getContext(), false); + const auto & partition_key = metadata_snapshot->getPartitionKey(); + const auto minmax_columns = MergeTreeData::getMinMaxColumns( + partition_key, storage.getSettings(), MergeTreePartMinMaxIndexColumns::PARTITION_KEY_ONLY); + + if (minmax.hyperrectangle.size() < minmax_columns.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_columns.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; + size_t i = 0; + for (const auto & [column_name, column_type] : minmax_columns) + { + auto column = column_type->createColumn(); + column->insert(minmax.hyperrectangle[i].left); + block.insert(ColumnWithTypeAndName(column->getPtr(), column_type, column_name)); + ++i; + } + + return block; } ContextPtr getContextCopyWithTaskSettings(const ContextPtr & context, const ExportReplicatedMergeTreePartitionManifest & manifest) @@ -206,7 +250,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 @@ -278,6 +327,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, @@ -332,7 +384,7 @@ namespace ExportPartitionUtils const auto source_metadata = source_storage.getInMemoryMetadataPtr(context, false); if (source_metadata->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( @@ -530,10 +582,185 @@ namespace ExportPartitionUtils ops.emplace_back(zkutil::makeSetRequest(last_exception_path, entry.toJsonString(), -1)); } +namespace +{ + /// Two types are interchangeable for partitioning only if their canonical names match. IDataType::equals + /// is too weak here: it deliberately treats DateTime and DateTime64 with different time zones as equal, + /// since they are interchangeable for INSERT, but a time zone changes what a temporal transform returns, + /// so the same expression over the two types can produce different partitions. + bool isSameTypeForPartitioning(const DataTypePtr & lhs, const DataTypePtr & rhs) + { + return lhs->getName() == rhs->getName(); + } + + /// The structural match is kind of permissive and is matching terms by name, not by type. + /// We also need to ensure types are the same if they are wrapped by functions. + bool castCannotBreakStructuralMatch( + const ActionsDAG::Node * destination_output, + const Names & minmax_column_names, + const DataTypes & minmax_column_types) + { + if (destination_output->type == ActionsDAG::ActionType::INPUT) + return true; + + for (const auto & required : ActionsDAG::cloneSubDAG({destination_output}, /*remove_aliases=*/ true).getRequiredColumns()) + { + const auto it = std::find(minmax_column_names.begin(), minmax_column_names.end(), required.name); + if (it == minmax_column_names.end()) + return false; + + if (!isSameTypeForPartitioning(minmax_column_types[static_cast(it - minmax_column_names.begin())], required.type)) + return false; + } + + return true; + } + + /// Dynamically verifies the destination expression maps to a single partition by checking its monotonicity over the source range. + void verifyOutputMapsToSinglePartition( + const ActionsDAG::Node * destination_output, + const Names & minmax_column_names, + const DataTypes & minmax_column_types, + const IMergeTreeDataPart::MinMaxIndex & minmax, + const String & partition_id, + const ContextPtr & context) + { + auto chain = buildPossiblyMonotonicChain(destination_output); + if (!chain.input_node) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: the destination partition expression '{}' is not a chain of functions " + "with known monotonicity over a single column, so it cannot be proven that the source partition " + "maps to a single destination partition.", destination_output->result_name); + + const auto & column = chain.input_node->result_name; + 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 (isNullableOrLowCardinalityNullable(source_type)) + 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; + + const auto & destination_type = chain.input_node->result_type; + + /// If the types are not the same, we need to check if the cast is monotonic + if (!isSameTypeForPartitioning(source_type, destination_type)) + { + 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()); + } + + if (!isMonotonicChain(destination_output, chain)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition '{}': the destination partition expression '{}' is not monotonic in " + "column '{}' (a hash such as icebergBucket never is), so its values at the endpoints of the " + "partition do not bound the rows in between.", + partition_id, destination_output->result_name, column); + + auto endpoints = source_type->createColumn(); + endpoints->insert(min_value); + endpoints->insert(max_value); + + Block block{{castColumn({std::move(endpoints), source_type, column}, destination_type), destination_type, column}}; + ExpressionActions(ActionsDAG::cloneSubDAG({destination_output}, /*remove_aliases=*/ true)).execute(block); + + const auto & result = *block.getByName(destination_output->result_name).column; + Field at_min; + Field at_max; + result.get(0, at_min); + result.get(1, at_max); + + if (at_min != at_max) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition '{}': the source partition might span multiple destination partitions " + "for expression '{}'. A source MergeTree partition must map to a single destination partition.", + partition_id, destination_output->result_name); + } + + /// A source partition is not split in the destination when every destination partition expression is + /// single-valued over it. That holds structurally when the expression is a deterministic function of the + /// source partition key, because rows agreeing on the source key then agree on it as well; the remaining + /// expressions have to be proven from the partition's min/max values. + void verifyPartitionKeyCompatibility( + const KeyDescription & source_key, + const KeyDescription & destination_key, + const MergeTreeData::DataPartsVector & parts, + const String & partition_id, + const ContextPtr & context) + { + /// An unpartitioned destination holds everything in a single partition. + if (destination_key.column_names.empty()) + return; + + const auto & destination_dag = destination_key.expression->getActionsDAG(); + const auto source_dag = ActionsDAG::cloneSubDAG( + source_key.expression->getActionsDAG().findInOutputs(source_key.column_names), /*remove_aliases=*/ true); + + /// ARRAY JOIN turns one row into many, which neither the tree matcher nor min/max models. + if (source_dag.hasArrayJoin() || destination_dag.hasArrayJoin()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: a partition key containing ARRAY JOIN is not supported."); + + /// Injective functions do not group rows, so the values they are applied to are what a destination + /// expression has to be a function of. + const auto irreducible_source_nodes = removeInjectiveFunctionsFromResultsRecursively(source_dag); + const auto matches = matchTrees(source_dag.getOutputs(), destination_dag); + + const auto minmax_columns = MergeTreeData::getMinMaxColumns( + source_key, parts.front()->storage.getSettings(), MergeTreePartMinMaxIndexColumns::PARTITION_KEY_ONLY); + const auto minmax_column_names = minmax_columns.getNames(); + const auto minmax_column_types = minmax_columns.getTypes(); + + /// Compute the global min/max index of the parts + IMergeTreeDataPart::MinMaxIndex minmax; + for (const auto & part : parts) + minmax.merge(*part->getMinMaxIndex()); + + /* + 1. If there is a structural match between the source and destination key, we accept it + 2. If there is not a structural match, we check if the destination expression maps to a single partition by checking its monotonicity over the source range. + */ + NodeMap visited; + for (const auto * destination_output : destination_dag.findInOutputs(destination_key.column_names)) + { + if (allOutputsDependsOnlyOnAllowedNodes(irreducible_source_nodes, matches, destination_output, visited) + && castCannotBreakStructuralMatch(destination_output, minmax_column_names, minmax_column_types)) + continue; + + verifyOutputMapsToSinglePartition( + destination_output, minmax_column_names, minmax_column_types, 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); @@ -567,89 +794,77 @@ 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); - }; + const auto spec_fields = partition_spec_json->getArray(Iceberg::f_fields); + const UInt32 spec_size = spec_fields ? static_cast(spec_fields->size()) : 0; + if (spec_size == 0) + return; - /// 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) + /// Rebuild the destination spec as a ClickHouse partition key, the way the Iceberg read path does in + /// ManifestFileIterator, so the same compatibility rule applies as for a plain object storage + /// destination and the transform arguments keep the order the writer will use. + const String partition_timezone = context->getSettingsRef()[Setting::iceberg_partition_timezone]; + auto partition_key_ast = make_intrusive(); + partition_key_ast->name = "tuple"; + partition_key_ast->arguments = make_intrusive(); + partition_key_ast->children.push_back(partition_key_ast->arguments); + + for (UInt32 i = 0; i < spec_size; ++i) { - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Cannot export partition to Iceberg table: the source MergeTree partition " - "key cannot be represented as an Iceberg partition spec: {}", e.message()); - } + const auto field = spec_fields->getObject(i); + const auto transform = field->getValue(Iceberg::f_transform); + const auto source_id = field->getValue(Iceberg::f_source_id); - 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 auto column_it = source_id_to_column_name.find(source_id); + if (column_it == source_id_to_column_name.end()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition to Iceberg table: destination partition spec refers to source_id " + "{}, which is not part of the current schema.", source_id); - 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) + auto transform_ast = Iceberg::getASTFromTransform(transform, column_it->second, partition_timezone); + if (!transform_ast) 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_it->second, transform); + + partition_key_ast->arguments->children.emplace_back(std::move(transform_ast)); } + + const auto destination_columns = ColumnsDescription::fromNamesAndTypes( + destination_metadata->getSampleBlockNonMaterialized().getNamesAndTypes()); + + verifyPartitionKeyCompatibility( + source_metadata->getPartitionKey(), + KeyDescription::getKeyFromAST(partition_key_ast, destination_columns, context), + 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) + { + verifyPartitionKeyCompatibility( + source_metadata->getPartitionKey(), destination_metadata->getPartitionKey(), parts, partition_id, context); + } + namespace { bool haveSameTupleElementLayout(const DataTypePtr & source_type, const DataTypePtr & destination_type) @@ -736,19 +951,6 @@ namespace ExportPartitionUtils } } - void assertPartitionKeyASTAreEqual( - const StorageMetadataPtr & source_metadata, - const StorageMetadataPtr & destination_metadata) - { - constexpr auto query_to_string = [] (const ASTPtr & ast) - { - return ast ? ast->formatWithSecretsOneLine() : ""; - }; - - if (query_to_string(source_metadata->getPartitionKeyAST()) != query_to_string(destination_metadata->getPartitionKeyAST())) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); - } - 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 652ec9223394..7605bd43ac4a 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, @@ -110,13 +104,28 @@ namespace ExportPartitionUtils 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 8581d733cd0d..3ebe3d9d3e84 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -7155,17 +7155,26 @@ void MergeTreeData::exportPartToTable( auto source_metadata_ptr = getInMemoryMetadataPtr(query_context, false); auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr(query_context, false); + 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_) { @@ -7200,26 +7209,29 @@ 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); - - if (!dest_storage->isDataLake()) - ExportPartitionUtils::assertPartitionKeyASTAreEqual(source_metadata_ptr, destination_metadata_ptr); - - 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()); + else + { + /// Plain (hive) object storage writes every row of the part to the one directory computed from + /// the destination PARTITION BY on the part's min row, so the source partition must map to a + /// single destination partition. Equivalent or finer source keys are accepted. + ExportPartitionUtils::verifyPlainPartitionCompatibility( + source_metadata_ptr, + destination_metadata_ptr, + {part}, + part->info.getPartitionId(), + query_context); + } if (part->getState() == MergeTreeDataPartState::Outdated && !allow_outdated_parts) throw Exception( diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 01e4a1d4ff6e..abd0cb896222 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #include @@ -694,9 +695,28 @@ SinkToStoragePtr StorageObjectStorage::import( std::string partition_key; + auto metadata_snapshot = getInMemoryMetadataPtr(local_context, false); + if (configuration->getPartitionStrategy()) { - const auto column_with_partition_key = configuration->getPartitionStrategy()->computePartitionKey(block_with_partition_values); + /// The values still carry the source table's types, but the partition key is rendered as text into the + /// object path and read back in this table's types, so it must be expressed in them first. A DateTime + /// in another time zone is the sharpest case: the epoch is the same, yet its text names another instant. + Block block_in_destination_types = block_with_partition_values; + const auto destination_sample = metadata_snapshot->getSampleBlock(); + for (auto & column : block_in_destination_types) + { + if (!destination_sample.has(column.name)) + continue; + + const auto & destination_type = destination_sample.getByName(column.name).type; + column.column = castColumn(column, destination_type); + /// castColumn is a no-op between types IDataType::equals considers equal, which includes DateTime + /// with different time zones, so relabel the column: serialization follows the type, not the values. + column.type = destination_type; + } + + const auto column_with_partition_key = configuration->getPartitionStrategy()->computePartitionKey(block_in_destination_types); if (!column_with_partition_key->empty()) { @@ -706,8 +726,6 @@ SinkToStoragePtr StorageObjectStorage::import( const auto base_path = configuration->getPathForWrite(partition_key, file_name).path; - auto metadata_snapshot = getInMemoryMetadataPtr(local_context, false); - return std::make_shared( base_path, /* transaction_id= */ file_name, /// not pretty, but the sink needs some sort of id to generate the commit file name. Using the source part name should be enough diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 233c8d6cc845..5f1c8293076e 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -247,6 +247,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; } @@ -8633,9 +8634,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & ExportPartitionUtils::verifyExportSchemaCastable( src_snapshot, destination_snapshot, dest_storage->getStorageID(), query_context); - if (!dest_storage->isDataLake()) - ExportPartitionUtils::assertPartitionKeyASTAreEqual(src_snapshot, destination_snapshot); - zkutil::ZooKeeperPtr zookeeper = getZooKeeperAndAssertNotReadonly(); const String partition_id = getPartitionIDFromQuery(command.partition, query_context); @@ -8753,6 +8751,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(); manifest.schema_mismatch_mode = query_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode].value; if (dest_storage->isDataLake()) @@ -8788,7 +8787,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); @@ -8802,6 +8805,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_merge_tree_part_to_iceberg/test.py b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py index 34590115d7de..6cc022eb593f 100644 --- a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py +++ b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py @@ -498,28 +498,6 @@ class RejectedPartExportCase(NamedTuple): ), id="same_partition_key_different_column_order_multi_column", ), - pytest.param( - RejectedPartExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(c, b, a)", - insert_values="(1, 2, 3, 'x')", - error_substrings=("partition field 0 mismatch",), - ), - id="multi_column_partition_key_order_mismatch", - ), - pytest.param( - RejectedPartExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(a, b)", - insert_values="(1, 2, 3, 'x')", - error_substrings=("partition scheme mismatch",), - ), - id="multi_column_partition_key_fewer_in_destination", - ), pytest.param( RejectedPartExportCase( src_columns="a Int32, b Int32, c Int32, val String", @@ -527,7 +505,7 @@ class RejectedPartExportCase(NamedTuple): dst_columns="a Int32, b Int32, c Int32, val String", dst_partition_by="(a, b, c)", insert_values="(1, 2, 3, 'x')", - error_substrings=("partition scheme mismatch",), + error_substrings=("column 'c', which is not part of the source MergeTree partition key",), ), id="multi_column_partition_key_more_in_destination", ), @@ -576,7 +554,10 @@ def test_export_part_partition_key_mismatch_variants_are_rejected(cluster, case) node.query(f"DROP TABLE IF EXISTS {iceberg}") -def test_export_part_multi_column_partition_key_success(cluster): +@pytest.mark.parametrize("dst_partition_by", ["(a, b, c)", "(c, b, a)", "(a, b)"]) +def test_export_part_multi_column_partition_key_success(cluster, dst_partition_by): + """The source key pins a, b and c, so any destination spec over those columns holds the whole + part in one partition, whatever order or subset of them it lists.""" node = cluster.instances["node1"] sfx = unique_suffix() mt = f"mt_multi_pkey_ok_{sfx}" @@ -584,7 +565,7 @@ def test_export_part_multi_column_partition_key_success(cluster): cols = "a Int32, b Int32, c Int32, val String" make_mt(node, mt, cols, "(a, b, c)") - make_iceberg_s3(node, iceberg, cols, "(a, b, c)") + make_iceberg_s3(node, iceberg, cols, dst_partition_by) node.query(f"INSERT INTO {mt} VALUES (1, 2, 3, 'x'), (1, 2, 3, 'y')") @@ -1181,9 +1162,10 @@ def test_export_part_tuple_subcolumn_partition_key_iceberg_rejected(cluster): f"SETTINGS allow_experimental_export_merge_tree_part = 1, " f"allow_experimental_insert_into_iceberg = 1" ) - assert "Unknown field to partition" in export_error, ( - f"Expected export validation to reject the tuple subcolumn partition key of {mt}, " - f"got: {export_error!r}" + assert "different Tuple element layout" in export_error, ( + f"The destination declares the elements of `t` in the opposite order, so the export " + f"of the tuple subcolumn partition key of {mt} has to be rejected, got: " + f"{export_error!r}" ) count = int(node.query(f"SELECT count() FROM {iceberg}").strip()) diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py index 29de23bc4d41..3ca7915e0811 100644 --- a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py +++ b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py @@ -370,32 +370,6 @@ class RejectedPartExportCase(NamedTuple): ), id="same_partition_key_different_column_order_multi_column", ), - pytest.param( - RejectedPartExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(c, b, a)", - insert_values="(1, 2, 3, 'x')", - error_substrings=( - "Tables have different partition key", - ), - ), - id="multi_column_partition_key_order_mismatch", - ), - pytest.param( - RejectedPartExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(a, b)", - insert_values="(1, 2, 3, 'x')", - error_substrings=( - "Tables have different partition key", - ), - ), - id="multi_column_partition_key_fewer_in_destination", - ), pytest.param( RejectedPartExportCase( src_columns="a Int32, b Int32, c Int32, val String", @@ -404,7 +378,7 @@ class RejectedPartExportCase(NamedTuple): dst_partition_by="(a, b, c)", insert_values="(1, 2, 3, 'x')", error_substrings=( - "Tables have different partition key", + "column 'c', which is not part of the source MergeTree partition key", ), ), id="multi_column_partition_key_more_in_destination", @@ -499,7 +473,15 @@ def test_export_part_partition_key_mismatch_variants_are_rejected(cluster, case) ) -def test_export_part_multi_column_partition_key_success(cluster): +@pytest.mark.parametrize( + "dst_partition_by", + ["(a, b, c)", "(c, b, a)", "(a, b)"], + ids=["same", "reordered", "coarser"], +) +def test_export_part_multi_column_partition_key_success(cluster, dst_partition_by): + """The source key pins every column the destination partitions by, so the destination may + also name them in another order or leave some out: each destination expression is still + single-valued over a source partition.""" skip_if_remote_database_disk_enabled(cluster) node = cluster.instances["node1"] @@ -518,7 +500,7 @@ def test_export_part_multi_column_partition_key_success(cluster): node.query(f""" CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') - PARTITION BY (a, b, c) + PARTITION BY {dst_partition_by} """) node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x'), (1, 2, 3, 'y')") @@ -621,38 +603,6 @@ def test_export_part_tuple_fields_reordered_for_partition_key_is_rejected( ) -def test_export_part_unnamed_tuple_partition_key_owner_matching_named_destination_is_allowed(cluster): - skip_if_remote_database_disk_enabled(cluster) - node = cluster.instances["node1"] - - postfix = str(uuid.uuid4()).replace("-", "_") - mt_table = f"unnamed_tuple_ok_mt_table_{postfix}" - s3_table = f"unnamed_tuple_ok_s3_table_{postfix}" - - node.query(f""" - CREATE TABLE {mt_table} (t Tuple(Int32, Int32), val String) - ENGINE = MergeTree() - PARTITION BY tupleElement(t, 1) - ORDER BY tuple() - SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 - """) - - node.query(f""" - CREATE TABLE {s3_table} (t Tuple(x Int32, y Int32), val String) - ENGINE = S3(s3_conn, filename='{s3_table}/{{_partition_id}}/{{_file}}', format=Parquet, partition_strategy='wildcard') - PARTITION BY tupleElement(t, 1) - """) - - node.query(f"INSERT INTO {mt_table} VALUES ((1, 99), 'x')") - - part_name = node.query( - f"SELECT name FROM system.parts WHERE database = currentDatabase() " - f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" - ).strip() - - node.query(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") - - def test_export_part_subcolumn_partition_key_different_subcolumn_is_rejected(cluster): skip_if_remote_database_disk_enabled(cluster) node = cluster.instances["node1"] @@ -687,16 +637,15 @@ def test_export_part_subcolumn_partition_key_different_subcolumn_is_rejected(clu ) assert ( "BAD_ARGUMENTS" in error - and "Tables have different partition key" + and "column 'a.c', which is not part of the source MergeTree partition key" in error ), ( f"Both tables declare `a` as the same Tuple(b Int32, c Int32) (so the column-cast " f"check passes and the owner-name-only `partition_key_owner_columns` contains " f"only `a`, so `verifyExportSchemaCastable` cannot distinguish `a.b` from " - f"`a.c`), but the source " - f"partitions by `a.b` and the destination by `a.c` — a genuinely different " - f"partition key that must be caught by the `PARTITION BY` AST comparison; " - f"got: {error!r}" + f"`a.c`), but the source partitions by `a.b` while the destination partitions by " + f"`a.c`, which the source key does not pin, so the compatibility gate has to " + f"reject it; got: {error!r}" ) 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 7c5025ef458e..13e9dbf833a3 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 @@ -739,7 +739,7 @@ def test_partition_transform_compatibility_accepted(cluster): supported transform when the MergeTree and Iceberg partition specs match. Cases covered: - 1. Compound identity (year, region) + 1. Compound identity (year, region), exported to a spec that lists the fields in reverse order 2. Year transform – toYearNumSinceEpoch(event_date) 3. Month transform – toMonthNumSinceEpoch(event_date) 4. truncate[4] – icebergTruncate(4, category) @@ -757,12 +757,13 @@ def check_accepted(mt, iceberg, description): ) return pid - # 1. Compound identity: (year, region) + # 1. Compound identity, with the destination listing the fields in the opposite order: the + # source key pins both columns, so the partition is single-valued for either field order. cols = "id Int64, year Int32, region String" t = f"mt_acc_1_{uid}"; i = f"iceberg_acc_1_{uid}" make_rmt(node, t, cols, "(year, region)") node.query(f"INSERT INTO {t} VALUES (1, 2023, 'EU')") - make_iceberg_s3(node, i, cols, "(year, region)") + make_iceberg_s3(node, i, cols, "(region, year)") pid = check_accepted(t, i, "compound identity (year, region)") wait_for_export_status(node, t, i, pid, "COMPLETED") count = int(node.query(f"SELECT count() FROM {i}").strip()) @@ -813,21 +814,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}", @@ -837,68 +842,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") - count = int(node.query(f"SELECT count() FROM {i}").strip()) - assert count == 0, f"[compound field order reversed] Expected 0 rows in destination, got {count}" - - # 2. Transform mismatch: MergeTree year-transform, Iceberg identity on same Date col + # 1. Transform mismatch: MergeTree year-transform, Iceberg identity on same Date col 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") - count = int(node.query(f"SELECT count() FROM {i}").strip()) - assert count == 0, f"[2-field MergeTree vs 1-field Iceberg] Expected 0 rows in destination, got {count}" + 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"] @@ -932,27 +926,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}" @@ -975,6 +973,286 @@ 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}, + # A hash is never monotonic, so min/max cannot prove anything about it, but an identity source key + # pins k within the partition and a bucket of a single value is a single bucket. + {"name": "bucket_from_identity_source", "columns": "id Int64, k Int64", "source_key": "k", + "dest_key": "icebergBucket(8, k)", "rows": "(1, 10), (2, 10)", "expect_ok": True, + "verify": [("k", "icebergBucket(8, k)")]}, + # The same bucket destination over a source key that does not pin k: nothing proves the rows of one + # source partition hash into the same bucket. + {"name": "bucket_needs_structural", "columns": "id Int64, k Int64", + "source_key": "intDiv(k, 100)", "dest_key": "icebergBucket(8, k)", + "rows": "(1, 10), (2, 20)", "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. @@ -1808,7 +2086,7 @@ class RejectedPartitionExportCase(NamedTuple): dst_columns="a Int32, b Int32, c Int32, val String", dst_partition_by="(a, b, c)", insert_values="(1, 2, 3, 'x')", - error_substrings=("partition scheme mismatch",), + error_substrings=("column 'c', which is not part of the source MergeTree partition key",), ), id="multi_column_partition_key_more_in_destination", ), @@ -2209,18 +2487,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)") @@ -2228,6 +2596,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}", @@ -2235,21 +2688,121 @@ 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}" + ) + + +def test_export_partition_column_timezone_mismatch_rejected(cluster): + """The same mismatch as above, but with the timezone carried by the column type instead of the + partition expression. Both sides read `toRelativeDayNum(event_time)`, so the terms are identical and + only the types differ - and DateTime types with different timezones compare equal, so the structural + match must not be decided by type equality alone. The part stays within one Tokyo day while spanning + two UTC days, so it maps to two destination partitions and must be rejected. + + `iceberg_partition_timezone` is deliberately left unset: setting it stamps a timezone onto the + destination term, which alone makes the terms differ and hides what this test covers.""" + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_coltz_{uid}" + iceberg_table = f"iceberg_coltz_{uid}" + + make_rmt(node, mt_table, "id Int64, event_time DateTime('Asia/Tokyo')", + "toRelativeDayNum(event_time)", replica_name="replica1") + # Both literals are 2024-03-05 in Tokyo (the column's timezone) but 2024-03-04 and 2024-03-05 in UTC. + node.query( + f"INSERT INTO {mt_table} VALUES (1, '2024-03-05 01:00:00'), (2, '2024-03-05 18: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}, + ) + assert "BAD_ARGUMENTS" in error, ( + f"Expected BAD_ARGUMENTS for a partition-column 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, @@ -2258,10 +2811,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 fdae5a075b08..48ae3acbd501 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 @@ -7,6 +7,7 @@ from helpers.cluster import ClickHouseCluster from helpers.export_partition_helpers import ( + first_partition_id, make_rmt, wait_for_exception_count, wait_for_export_status, @@ -1751,6 +1752,345 @@ def test_export_partition_all_failure_modes(cluster): ) +# ---- 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}" + + +def test_export_partition_column_timezone_rendered_in_destination_zone(cluster): + """A hive partition value lives as text in the object path and is read back in the destination + column's time zone, so the export has to spell it the way the destination would. Spelling it in the + source's zone names a different instant and the row reads back shifted by the offset between the + two zones. INSERT SELECT into an identical table is the reference behavior.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"tz_mt_{uid}" + s3_export = f"tz_export_s3_{uid}" + s3_insert = f"tz_insert_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, ts DateTime('UTC'))" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY toDate(ts) ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, '2024-03-05 15:00:00')") + for table in (s3_export, s3_insert): + node.query( + f"CREATE TABLE {table} (id UInt64, ts DateTime('Asia/Tokyo'))" + f" ENGINE = S3(s3_conn, filename='{table}', format=Parquet, partition_strategy='hive')" + f" PARTITION BY ts" + ) + + pid = first_partition_id(node, mt_table) + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {s3_export}") + wait_for_export_status(node, mt_table, s3_export, pid, "COMPLETED", timeout=90) + + node.query(f"INSERT INTO {s3_insert} SELECT * FROM {mt_table}") + + source_instant = node.query(f"SELECT toUnixTimestamp(ts) FROM {mt_table}").strip() + exported_instant = node.query(f"SELECT toUnixTimestamp(ts) FROM {s3_export}").strip() + inserted_instant = node.query(f"SELECT toUnixTimestamp(ts) FROM {s3_insert}").strip() + assert exported_instant == source_instant, ( + f"the exported row moved in time: source {source_instant}, destination {exported_instant}" + ) + assert inserted_instant == source_instant, ( + f"INSERT SELECT must not move it either: source {source_instant}," + f" destination {inserted_instant}" + ) + + # 2024-03-05 15:00:00 UTC is 2024-03-06 00:00:00 in Tokyo. + exported_directory = node.query( + f"SELECT DISTINCT extract(_path, 'ts=[^/]*') FROM {s3_export}" + ).strip() + inserted_directory = node.query( + f"SELECT DISTINCT extract(_path, 'ts=[^/]*') FROM {s3_insert}" + ).strip() + assert exported_directory == "ts=2024-03-06 00:00:00", ( + f"unexpected hive directory: {exported_directory!r}" + ) + assert inserted_directory == exported_directory, ( + f"export and INSERT SELECT disagree on the partition directory:" + f" {exported_directory!r} vs {inserted_directory!r}" + ) + + +def create_wildcard_destination(node, table, columns, partition_key): + """A wildcard destination, the only partition strategy that accepts an expression as its + partition key: the hive strategy allows storage columns only.""" + node.query( + f"CREATE TABLE {table} ({columns})" + f" ENGINE = S3(s3_conn, filename='{table}/{{_partition_id}}/{{_file}}.parquet'," + f" format=Parquet, partition_strategy='wildcard')" + f" PARTITION BY {partition_key}" + ) + + +def test_export_partition_dest_argument_order_rejected(cluster): + """The destination key intDiv(x, 100) has to be validated as written. This source part holds + x in [201, 350], which covers the destination partitions 2 and 3, so the export must be rejected. + Reading the arguments in the reverse order would validate intDiv(100, x) instead, which is 0 at + both endpoints and would silently write both destination partitions into one directory.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"argorder_mt_{uid}" + s3_table = f"argorder_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, x UInt64)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY intDiv(x, 1000) ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 201), (2, 350)") + create_wildcard_destination(node, s3_table, "id UInt64, x UInt64", "intDiv(x, 100)") + + 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}" + + +def test_export_partition_dest_finer_expression_single_partition_accepted(cluster): + """The same shape as the rejected case, with x in [100, 150]: the whole source partition maps to + the single destination partition 1, so it is accepted and every row lands in one directory. The + swapped-argument reading would refuse this one, since intDiv(100, 100) != intDiv(100, 150).""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"argorder_ok_mt_{uid}" + s3_table = f"argorder_ok_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, x UInt64)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY intDiv(x, 1000) ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 100), (2, 150)") + create_wildcard_destination(node, s3_table, "id UInt64, x UInt64", "intDiv(x, 100)") + + pid = first_partition_id(node, mt_table) + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {s3_table}") + wait_for_export_status(node, mt_table, s3_table, pid, "COMPLETED", timeout=90) + + # A wildcard destination cannot be read as a table, so read the objects it wrote. + exported = f"s3(s3_conn, filename='{s3_table}/**/*.parquet', format='Parquet', structure='id UInt64, x UInt64')" + src = node.query(f"SELECT id, x FROM {mt_table} ORDER BY id") + dst = node.query(f"SELECT id, x FROM {exported} ORDER BY id") + assert dst == src, f"destination rows differ from source:\nsrc={src!r}\ndst={dst!r}" + + directories = node.query( + f"SELECT DISTINCT extract(_path, '{s3_table}/[^/]*') FROM {exported}" + ).strip() + assert directories == f"{s3_table}/1", f"unexpected destination directories: {directories!r}" + + +def test_export_partition_dest_nested_expression_accepted(cluster): + """A destination key that wraps the source key in a coarser transform - toYYYYMM(toDate(ts)) over + a source keyed by toDate(ts) - is a function of the source key, so every source partition sits + inside one destination partition whatever the data is.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"nested_mt_{uid}" + s3_table = f"nested_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, ts DateTime)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY toDate(ts) ORDER BY tuple()" + ) + node.query( + f"INSERT INTO {mt_table} VALUES (1, '2024-03-05 01:00:00'), (2, '2024-03-05 20:00:00')" + ) + create_wildcard_destination(node, s3_table, "id UInt64, ts DateTime", "toYYYYMM(toDate(ts))") + + pid = first_partition_id(node, mt_table) + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {s3_table}") + wait_for_export_status(node, mt_table, s3_table, pid, "COMPLETED", timeout=90) + + exported = f"s3(s3_conn, filename='{s3_table}/**/*.parquet', format='Parquet', structure='id UInt64, ts DateTime')" + src = node.query(f"SELECT id, ts FROM {mt_table} ORDER BY id") + dst = node.query(f"SELECT id, ts FROM {exported} ORDER BY id") + assert dst == src, f"destination rows differ from source:\nsrc={src!r}\ndst={dst!r}" + + directories = node.query( + f"SELECT DISTINCT extract(_path, '{s3_table}/[^/]*') FROM {exported}" + ).strip() + assert directories == f"{s3_table}/202403", ( + f"unexpected destination directories: {directories!r}" + ) + + +def test_export_partition_dest_term_over_two_columns_rejected(cluster): + """A destination expression over two columns is only single-valued when the source key pins both. + This source pins b but only intDiv(a, 100), so a spans [10, 90] within one source partition and + intDiv(a + b, 100) takes both 0 and 1 there. Per-column min/max cannot bound such an expression, + so it is rejected; a source keyed by (a, b) would be accepted, since it pins both columns.""" + node = cluster.instances["replica1"] + + uid = str(uuid.uuid4()).replace("-", "_") + mt_table = f"twocol_mt_{uid}" + s3_table = f"twocol_s3_{uid}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, a UInt64, b UInt64)" + f" ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1')" + f" PARTITION BY (intDiv(a, 100), b) ORDER BY tuple()" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 10, 20), (2, 90, 20)") + create_wildcard_destination( + node, s3_table, "id UInt64, a UInt64, b UInt64", "intDiv(a + b, 100)" + ) + + 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}" class RejectedPartitionExportCase(NamedTuple): src_columns: str src_partition_by: str @@ -1783,32 +2123,6 @@ class RejectedPartitionExportCase(NamedTuple): ), id="same_partition_key_different_column_order_multi_column", ), - pytest.param( - RejectedPartitionExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(c, b, a)", - insert_values="(1, 2, 3, 'x')", - error_substrings=( - "Tables have different partition key", - ), - ), - id="multi_column_partition_key_order_mismatch", - ), - pytest.param( - RejectedPartitionExportCase( - src_columns="a Int32, b Int32, c Int32, val String", - src_partition_by="(a, b, c)", - dst_columns="a Int32, b Int32, c Int32, val String", - dst_partition_by="(a, b)", - insert_values="(1, 2, 3, 'x')", - error_substrings=( - "Tables have different partition key", - ), - ), - id="multi_column_partition_key_fewer_in_destination", - ), pytest.param( RejectedPartitionExportCase( src_columns="a Int32, b Int32, c Int32, val String", @@ -1817,7 +2131,7 @@ class RejectedPartitionExportCase(NamedTuple): dst_partition_by="(a, b, c)", insert_values="(1, 2, 3, 'x')", error_substrings=( - "Tables have different partition key", + "column 'c', which is not part of the source MergeTree partition key", ), ), id="multi_column_partition_key_more_in_destination", @@ -1866,7 +2180,15 @@ def test_export_partition_partition_key_mismatch_variants_are_rejected(cluster, assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" -def test_export_partition_multi_column_partition_key_success(cluster): +@pytest.mark.parametrize( + "dst_partition_by", + ["(a, b, c)", "(c, b, a)", "(a, b)"], + ids=["same", "reordered", "coarser"], +) +def test_export_partition_multi_column_partition_key_success(cluster, dst_partition_by): + """The source key pins every column the destination partitions by, so the destination may + also name them in another order or leave some out: each destination expression is still + single-valued over a source partition.""" skip_if_remote_database_disk_enabled(cluster) node = cluster.instances["replica1"] @@ -1884,7 +2206,7 @@ def test_export_partition_multi_column_partition_key_success(cluster): node.query(f""" CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') - PARTITION BY (a, b, c) + PARTITION BY {dst_partition_by} """) node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x'), (1, 2, 3, 'y')") 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"