diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 57a852be327ec..41406b45c4fa3 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -43,6 +43,7 @@ #include #if USE_AVRO +#include #include #include #include @@ -942,6 +943,8 @@ namespace if (spec_size == 0) return; + ChunkPartitioner::checkPartitionSpecSupportedForWrites(spec_fields); + /// 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. diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp index 6a006e5e95650..4c314a1a7cf57 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp @@ -133,7 +133,7 @@ std::vector mapTargetFieldsToSpecPositions( { for (size_t j = 0; j < entry_spec.size(); ++j) { - if (entry_spec[j].source_id == signature[i].column_id + if (!entry_spec[j].isMultiArg() && entry_spec[j].source_ids[0] == signature[i].column_id && Poco::toLower(entry_spec[j].transform_name) == signature[i].transform) { positions[i] = j; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp index 5a169f07d5636..bbc63ae17f8cb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp @@ -27,11 +27,30 @@ namespace ErrorCodes extern const int BAD_ARGUMENTS; } +void ChunkPartitioner::checkPartitionSpecSupportedForWrites(const Poco::JSON::Array::Ptr & partition_specification) +{ + if (!partition_specification) + return; + + for (size_t i = 0; i != partition_specification->size(); ++i) + { + /// Iceberg V3: multi-argument transforms use `source-ids` instead of `source-id`. + /// Writing with multi-arg transforms is not supported yet (hash semantics not finalized upstream). + if (partition_specification->getObject(static_cast(i))->has(Iceberg::f_source_ids)) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Multi-argument partition transforms (source-ids) are not supported for writes. " + "Multi-argument transform evaluation is not yet implemented"); + } +} + ChunkPartitioner::ChunkPartitioner( Poco::JSON::Array::Ptr partition_specification, Poco::JSON::Array::Ptr schema_fields, ContextPtr context, SharedHeader sample_block_) : sample_block(sample_block_) , max_partitions_count(context->getSettingsRef()[Setting::iceberg_insert_max_partitions]) { + checkPartitionSpecSupportedForWrites(partition_specification); + std::unordered_map id_to_column; { for (size_t i = 0; i < schema_fields->size(); ++i) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h index 9f65823374360..560c7499e1269 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h @@ -25,6 +25,10 @@ class ChunkPartitioner ContextPtr context, SharedHeader sample_block_); + /// Throws if the partition spec cannot be evaluated on write (e.g. multi-argument `source-ids` transforms). + /// Callers that write objects before constructing a `ChunkPartitioner` must call this first. + static void checkPartitionSpecSupportedForWrites(const Poco::JSON::Array::Ptr & partition_specification); + using PartitionKey = Row; struct PartitionKeyHasher { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp index e8bebbef824f7..5e2a46b9dfbac 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp @@ -227,6 +227,29 @@ static Plan getPlan( if (initial_metadata_object->getValue(Iceberg::f_format_version) != 2) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Compaction is supported only for format_version 2."); + /// `writeMetadataFiles` partitions by the default spec only after the rewritten data files are written, + /// so an unsupported spec must be rejected here, before any object is created. + { + const auto default_spec_id = initial_metadata_object->getValue(Iceberg::f_default_spec_id); + const auto partition_specs = initial_metadata_object->getArray(Iceberg::f_partition_specs); + Poco::JSON::Object::Ptr default_spec; + for (UInt32 i = 0; i < partition_specs->size(); ++i) + { + auto candidate = partition_specs->getObject(i); + if (candidate->getValue(Iceberg::f_spec_id) == default_spec_id) + { + default_spec = candidate; + break; + } + } + if (!default_spec) + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Iceberg metadata does not contain a partition spec entry matching default-spec-id {}", + default_spec_id); + ChunkPartitioner::checkPartitionSpecSupportedForWrites(default_spec->getArray(Iceberg::f_fields)); + } + auto current_schema_id = initial_metadata_object->getValue(Iceberg::f_current_schema_id); auto schemas = initial_metadata_object->getArray(Iceberg::f_schemas); Poco::JSON::Array::Ptr current_schema; @@ -568,6 +591,7 @@ static bool writeConsolidatedManifestFile( ResolvedPartitionSpec resolved; resolved.spec = spec; auto spec_fields = spec->getArray(f_fields); + ChunkPartitioner::checkPartitionSpecSupportedForWrites(spec_fields); /// Partition field names and the schema source-ids they transform. std::vector source_ids; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h index 5bbf755e7ee2d..62cd9f9913ebb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h @@ -127,6 +127,7 @@ DEFINE_ICEBERG_FIELD_ALIAS(order_id, order-id); DEFINE_ICEBERG_FIELD_ALIAS(default_sort_order_id, default-sort-order-id); DEFINE_ICEBERG_FIELD_ALIAS(sort_orders, sort-orders); DEFINE_ICEBERG_FIELD_ALIAS(source_id, source-id); +DEFINE_ICEBERG_FIELD_ALIAS(source_ids, source-ids); DEFINE_ICEBERG_FIELD_ALIAS(partition_transform, transform); DEFINE_ICEBERG_FIELD_ALIAS(partition_name, name); DEFINE_ICEBERG_FIELD_ALIAS(default_spec_id, default-spec-id); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 84d522be5d69b..5b667ba38e9bd 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -2073,8 +2073,14 @@ KeyDescription IcebergMetadata::getSortingKey(ContextPtr local_context, TableSta auto [schema, current_schema_id] = parseTableSchemaV2Method(metadata_object); auto result = getSortingKeyDescriptionFromMetadata(metadata_object, *persistent_components.schema_processor->getClickHouseTableSchemaById(current_schema_id), local_context); - auto sort_order_id = metadata_object->getValue(f_default_sort_order_id); - result.sort_order_id = sort_order_id; + /// Only attach the sort-order id when the parsed key is non-empty. When multi-arg or + /// unsupported transforms truncate the key to empty, pairing a stale sort_order_id with + /// an empty key would let `isDataSortedBySortingKey` incorrectly claim data is sorted. + if (result.definition_ast != nullptr) + { + auto sort_order_id = metadata_object->getValue(f_default_sort_order_id); + result.sort_order_id = sort_order_id; + } return result; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp index 201fdcd3a7eb0..b1e16d047ea61 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp @@ -9,6 +9,7 @@ #include #include +#include namespace DB::ErrorCodes @@ -99,8 +100,10 @@ void requireDirectReferencedDataFileForPuffinDeletionVector( static std::strong_ordering operator<=>(const PartitionSpecsEntry & lhs, const PartitionSpecsEntry & rhs) { - return std::tie(lhs.source_id, lhs.transform_name, lhs.partition_name) - <=> std::tie(rhs.source_id, rhs.transform_name, rhs.partition_name); + if (auto cmp = lhs.source_ids <=> rhs.source_ids; cmp != std::strong_ordering::equal) + return cmp; + return std::tie(lhs.transform_name, lhs.partition_name) + <=> std::tie(rhs.transform_name, rhs.partition_name); } template @@ -138,7 +141,8 @@ static String dumpPartitionSpecification(const PartitionSpecification & partitio { const auto & entry = partition_specification[i]; answer += fmt::format( - "(source id: {}, transform name: {}, partition name: {})", entry.source_id, entry.transform_name, entry.partition_name); + "(source ids: [{}], transform name: {}, partition name: {})", + fmt::join(entry.source_ids, ", "), entry.transform_name, entry.partition_name); if (i != partition_specification.size() - 1) answer += ", "; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h index 7b49fdd76d09c..d2d5a5f9eede4 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h @@ -61,10 +61,15 @@ String FileContentTypeToString(FileContentType type); struct PartitionSpecsEntry { - Int32 source_id; + /// For single-argument transforms (V1/V2 and single-arg V3) this holds one element. + /// For multi-argument V3 transforms (e.g. bucket over multiple columns) it holds multiple. + std::vector source_ids; String transform_name; String partition_name; Int32 tuple_index; + + /// Convenience: true when the transform references more than one source column (V3 multi-arg). + bool isMultiArg() const { return source_ids.size() > 1; } }; using PartitionSpecification = std::vector; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 43868071a14bc..bf2df59fef381 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -236,11 +236,63 @@ std::shared_ptr ManifestFileIterator::create( } PartitionSpecification partition_spec_vec; + /// Which spec-field positions made it into partition_key_ast (needed by the pruner to project + /// partition values, since the Avro tuple has one entry per spec field, not per kept field). + std::vector partition_key_spec_positions; for (size_t i = 0; i != partition_specification->size(); ++i) { auto partition_specification_field = partition_specification->getObject(static_cast(i)); - auto source_id = partition_specification_field->getValue(f_source_id); + /// Iceberg V3 spec: partition fields use either singular `source-id` (single-arg transforms) + /// or `source-ids` (multi-arg transforms, e.g. bucket over multiple columns). They are mutually exclusive. + bool has_source_id = partition_specification_field->has(f_source_id); + bool has_source_ids = partition_specification_field->has(f_source_ids); + + std::vector source_ids; + if (has_source_id && has_source_ids) + { + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Partition field in manifest '{}' has both 'source-id' and 'source-ids' — they are mutually exclusive per the Iceberg spec", + path_to_manifest_file_); + } + else if (has_source_ids) + { + auto source_ids_array = partition_specification_field->getArray(f_source_ids); + for (UInt32 idx = 0; idx < source_ids_array->size(); ++idx) + source_ids.push_back(source_ids_array->getElement(idx)); + } + else if (has_source_id) + { + source_ids.push_back(partition_specification_field->getValue(f_source_id)); + } + else + { + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Partition field in manifest '{}' has neither 'source-id' nor 'source-ids'", + path_to_manifest_file_); + } + + if (source_ids.empty()) + { + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Partition field in manifest '{}' has an empty 'source-ids' array", + path_to_manifest_file_); + } + + auto transform_name = partition_specification_field->getValue(f_partition_transform); + auto partition_name = partition_specification_field->getValue(f_partition_name); + partition_spec_vec.emplace_back(PartitionSpecsEntry{std::move(source_ids), transform_name, partition_name, static_cast(i)}); + + /// Multi-argument transforms (V3): we cannot evaluate the transform, so skip pruning for this field. + /// Per the Iceberg V3 spec: "all v3 readers are required to read tables with unknown transforms, + /// ignoring the unsupported partition fields when filtering." + if (partition_spec_vec.back().isMultiArg()) + continue; + + auto source_id = partition_spec_vec.back().source_ids[0]; /// NOTE: tricky part to support RENAME column in partition key. Instead of some name /// we use column internal number as it's name. auto numeric_column_name = DB::backQuote(DB::toString(source_id)); @@ -248,9 +300,6 @@ std::shared_ptr ManifestFileIterator::create( = schema_processor.tryGetFieldCharacteristics(manifest_schema_id, source_id); if (!manifest_file_column_characteristics.has_value()) continue; - auto transform_name = partition_specification_field->getValue(f_partition_transform); - auto partition_name = partition_specification_field->getValue(f_partition_name); - partition_spec_vec.emplace_back(source_id, transform_name, partition_name, static_cast(i)); auto partition_ast = getASTFromTransform(transform_name, numeric_column_name, context_->getSettingsRef()[Setting::iceberg_partition_timezone]); /// Unsupported partition key expression if (partition_ast == nullptr) @@ -262,6 +311,7 @@ std::shared_ptr ManifestFileIterator::create( /// against these input columns, which must contain each source column at most once. if (partition_columns_seen.insert(numeric_column_name).second) partition_columns_description.emplace_back(numeric_column_name, removeNullable(manifest_file_column_characteristics->type)); + partition_key_spec_positions.push_back(i); } std::optional partition_key_description; @@ -283,6 +333,7 @@ std::shared_ptr ManifestFileIterator::create( manifest_schema_id, std::make_shared(std::move(partition_spec_vec)), std::move(partition_key_description), + std::move(partition_key_spec_positions), total_rows, std::move(filter_dag_), table_snapshot_schema_id_, @@ -301,6 +352,7 @@ ManifestFileIterator::ManifestFileIterator( Int32 manifest_schema_id_, std::shared_ptr common_partition_specification_, std::optional partition_key_description_, + std::vector partition_key_spec_positions_, size_t total_rows_, std::shared_ptr filter_dag_, Int32 table_snapshot_schema_id_, @@ -315,6 +367,7 @@ ManifestFileIterator::ManifestFileIterator( , manifest_schema_id(manifest_schema_id_) , common_partition_specification(std::move(common_partition_specification_)) , partition_key_description(std::move(partition_key_description_)) + , partition_key_spec_positions(std::move(partition_key_spec_positions_)) , table_snapshot_schema_id(table_snapshot_schema_id_) , total_rows(total_rows_) , stop_flag(stop_flag_) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h index 91e387a0788b4..de67f89b67494 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h @@ -101,6 +101,9 @@ class ManifestFileIterator : public boost::noncopyable bool hasPartitionKey() const; const DB::KeyDescription & getPartitionKeyDescription() const; + /// Returns the spec-field positions (indices into the Avro partition tuple) that correspond + /// to the kept partition key columns. The pruner uses this to project partition values. + const std::vector & getPartitionKeySpecPositions() const { return partition_key_spec_positions; } /// Sums required per-file `record_count` for live entries of the given content type. /// Returns nullopt on negative `record_count` or Int64 overflow (fail closed). std::optional getRowsCountInAllFilesExcludingDeleted(FileContentType content) const; @@ -132,6 +135,7 @@ class ManifestFileIterator : public boost::noncopyable Int32 manifest_schema_id, std::shared_ptr common_partition_specification, std::optional partition_key_description, + std::vector partition_key_spec_positions, size_t total_rows, std::shared_ptr filter_dag, Int32 table_snapshot_schema_id, @@ -151,6 +155,8 @@ class ManifestFileIterator : public boost::noncopyable const Int32 manifest_schema_id; const std::shared_ptr common_partition_specification; const std::optional partition_key_description; + /// Indices (into the Avro partition tuple) of the spec fields that were kept for pruning. + const std::vector partition_key_spec_positions; const Int32 table_snapshot_schema_id; /// Iteration state diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp index 3a960114fdaad..6611afc1e6793 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp @@ -164,6 +164,7 @@ ManifestFilesPruner::ManifestFilesPruner( if (manifest_file.hasPartitionKey()) { partition_key = &manifest_file.getPartitionKeyDescription(); + partition_key_spec_positions = manifest_file.getPartitionKeySpecPositions(); ActionsDAGWithInversionPushDown inverted_dag(transformed_dag->getOutputs().front(), context, /* boolean_context */ true); partition_key_condition.emplace( inverted_dag, context, partition_key->column_names, partition_key->expression, true /* single_point */); @@ -408,7 +409,16 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( if (partition_key_condition.has_value()) { - Row wrapped_partition_value = partition_value; + /// Project: the Avro tuple has one entry per spec field, but the key only contains the + /// fields at the positions recorded in `partition_key_spec_positions`. Pick exactly those. + Row wrapped_partition_value; + wrapped_partition_value.reserve(partition_key_spec_positions.size()); + for (size_t pos : partition_key_spec_positions) + { + if (pos >= partition_value.size()) + return PruningReturnStatus::NOT_PRUNED; /// fail closed — cannot prune without the value + wrapped_partition_value.push_back(partition_value[pos]); + } wrapIntegerPartitionValuesAsDateTime64(wrapped_partition_value, partition_key->data_types); std::vector index_value(wrapped_partition_value.begin(), wrapped_partition_value.end()); @@ -426,7 +436,7 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( } bool can_be_true = partition_key_condition->mayBeTrueInRange( - partition_value.size(), index_value.data(), index_value.data(), partition_key->data_types); + index_value.size(), index_value.data(), index_value.data(), partition_key->data_types); if (!can_be_true) { @@ -452,7 +462,7 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( { for (const auto & partition_field : *entry->common_partition_specification) { - if (partition_field.source_id != column_id || partition_field.tuple_index < 0 + if (partition_field.isMultiArg() || partition_field.source_ids[0] != column_id || partition_field.tuple_index < 0 || static_cast(partition_field.tuple_index) >= partition_value.size()) continue; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h index 070a0b6fafc04..6dac40a8a0641 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h @@ -40,6 +40,8 @@ class ManifestFilesPruner Int32 current_schema_id; Int32 initial_schema_id; const DB::KeyDescription * partition_key; + /// Which positions in the Avro partition tuple correspond to the kept partition-key columns. + std::vector partition_key_spec_positions; std::optional partition_key_condition; std::unordered_map min_max_key_conditions; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 2ac34519693e1..81390e2a2cdc6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -761,6 +761,19 @@ void MetadataGenerator::generateDropColumnMetadata(const String & column_name) ErrorCodes::BAD_ARGUMENTS, "Cannot drop column '{}' (field id {}): it is referenced by the active sort order", column_name, dropped_field_id); + /// Also check multi-arg V3 sort fields + if (sf->has(Iceberg::f_source_ids)) + { + auto ids = sf->getArray(Iceberg::f_source_ids); + for (UInt32 k = 0; k < ids->size(); ++k) + { + if (ids->getElement(k) == dropped_field_id) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Cannot drop column '{}' (field id {}): it is referenced by the active sort order (multi-arg transform)", + column_name, dropped_field_id); + } + } } break; } @@ -788,6 +801,19 @@ void MetadataGenerator::generateDropColumnMetadata(const String & column_name) ErrorCodes::BAD_ARGUMENTS, "Cannot drop column '{}' (field id {}): it is referenced by the active partition spec", column_name, dropped_field_id); + /// Also check multi-arg V3 partition fields + if (pf->has(Iceberg::f_source_ids)) + { + auto ids = pf->getArray(Iceberg::f_source_ids); + for (UInt32 k = 0; k < ids->size(); ++k) + { + if (ids->getElement(k) == dropped_field_id) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Cannot drop column '{}' (field id {}): it is referenced by the active partition spec (multi-arg transform)", + column_name, dropped_field_id); + } + } } break; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 1d219b9ce925d..3155e86a5c2f9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -1725,6 +1726,17 @@ KeyDescription getSortingKeyDescriptionFromMetadata(Poco::JSON::Object::Ptr meta for (UInt32 field_index = 0; field_index < fields->size(); ++field_index) { auto field = fields->getObject(field_index); + + /// Iceberg V3: multi-argument transforms use `source-ids` instead of `source-id`. + /// We cannot evaluate multi-arg transforms. Data sorted by (f1, f2, f3) is only + /// guaranteed sorted by prefixes — never by a suffix — so stop at the first + /// unrepresentable field and return only the prefix before it. + if (field->has(f_source_ids)) + break; + + if (!field->has(f_source_id)) + break; + auto source_id = field->getValue(f_source_id); auto column_name = source_id_to_column_name[source_id]; int direction = field->getValue(f_direction) == "asc" ? 1 : -1; @@ -1779,6 +1791,28 @@ KeyDescription getSortingKeyDescriptionFromMetadata(Poco::JSON::Object::Ptr meta return KeyDescription::parse(order_by_str, column_description, {}, local_context, true); } +/// Format a multi-argument partition field for display in Iceberg/Spark style, e.g. "bucket(16, a, b)". +static String formatPartitionFieldDisplayMultiArg(const String & iceberg_transform_name, const std::vector & column_names) +{ + std::string name = Poco::toLower(iceberg_transform_name); + String columns_joined = fmt::format("{}", fmt::join(column_names, ", ")); + + if (name.starts_with("bucket") && name.back() == ']') + { + auto p = name.find('['); + if (p != std::string::npos) + return "bucket(" + name.substr(p + 1, name.size() - p - 2) + ", " + columns_joined + ")"; + } + if (name.starts_with("truncate") && name.back() == ']') + { + auto p = name.find('['); + if (p != std::string::npos) + return "truncate(" + name.substr(p + 1, name.size() - p - 2) + ", " + columns_joined + ")"; + } + /// Fallback for unknown multi-arg transforms: show as transform(col1, col2, ...) + return name + "(" + columns_joined + ")"; +} + /// Format one partition field for display in Iceberg/Spark style, e.g. "day(ts)" or "bucket(16, id)". static String formatPartitionFieldDisplay(const String & iceberg_transform_name, const String & column_name) { @@ -1840,12 +1874,32 @@ std::optional getPartitionKeyStringFromMetadata(Poco::JSON::Object::Ptr for (UInt32 i = 0; i < fields->size(); ++i) { auto field = fields->getObject(i); + auto iceberg_transform_name = field->getValue(f_transform); + + /// Iceberg V3 multi-argument transform: `source-ids` is an array of column IDs. + if (field->has(f_source_ids)) + { + auto source_ids_array = field->getArray(f_source_ids); + std::vector column_names; + for (UInt32 idx = 0; idx < source_ids_array->size(); ++idx) + { + auto sid = source_ids_array->getElement(idx); + auto it = source_id_to_column_name.find(sid); + if (it == source_id_to_column_name.end()) + return std::nullopt; + column_names.push_back(it->second); + } + part_exprs.push_back(formatPartitionFieldDisplayMultiArg(iceberg_transform_name, column_names)); + continue; + } + + if (!field->has(f_source_id)) + return std::nullopt; auto source_id = field->getValue(f_source_id); auto it = source_id_to_column_name.find(source_id); if (it == source_id_to_column_name.end()) return std::nullopt; String column_name = it->second; - auto iceberg_transform_name = field->getValue(f_transform); part_exprs.push_back(formatPartitionFieldDisplay(iceberg_transform_name, column_name)); } String result; @@ -1880,14 +1934,38 @@ std::optional getSortingKeyDisplayStringFromMetadata(Poco::JSON::Object: for (UInt32 j = 0; j < sort_fields->size(); ++j) { auto field = sort_fields->getObject(j); - auto source_id = field->getValue(f_source_id); - auto it = source_id_to_column_name.find(source_id); - if (it == source_id_to_column_name.end()) - return std::nullopt; - String column_name = it->second; - String direction = field->getValue(f_direction) == "asc" ? " asc" : " desc"; auto iceberg_transform_name = field->getValue(f_transform); - String expr = formatPartitionFieldDisplay(iceberg_transform_name, column_name); + String direction = field->getValue(f_direction) == "asc" ? " asc" : " desc"; + String expr; + + /// Iceberg V3 multi-argument transform + if (field->has(f_source_ids)) + { + auto source_ids_array = field->getArray(f_source_ids); + std::vector column_names; + for (UInt32 idx = 0; idx < source_ids_array->size(); ++idx) + { + auto sid = source_ids_array->getElement(idx); + auto it = source_id_to_column_name.find(sid); + if (it == source_id_to_column_name.end()) + return std::nullopt; + column_names.push_back(it->second); + } + expr = formatPartitionFieldDisplayMultiArg(iceberg_transform_name, column_names); + } + else if (field->has(f_source_id)) + { + auto source_id = field->getValue(f_source_id); + auto it = source_id_to_column_name.find(source_id); + if (it == source_id_to_column_name.end()) + return std::nullopt; + expr = formatPartitionFieldDisplay(iceberg_transform_name, it->second); + } + else + { + return std::nullopt; + } + if (!result.empty()) result += ", "; result += expr + direction; @@ -2125,7 +2203,7 @@ PartitionColumnValues getIdentityPartitionColumnValues( continue; const auto name_and_type - = schema_processor.tryGetFieldCharacteristics(manifest_file_entry.resolved_schema_id, partition_field.source_id); + = schema_processor.tryGetFieldCharacteristics(manifest_file_entry.resolved_schema_id, partition_field.source_ids[0]); if (!name_and_type.has_value()) continue; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp index 7453998c7bc80..e262df663178b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp @@ -793,6 +793,43 @@ TEST(IcebergMetadataGenerator, ModifyColumnWideningRecordsTheNewTypeInANewSchema } +/// --- Multi-argument transform (Iceberg V3 source-ids) tests --- + +/// Helper: make metadata with two columns (x: int, y: string) for multi-arg tests. +static Poco::JSON::Object::Ptr makeMetadataWithTwoColumns() +{ + auto metadata = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + metadata->set(f_format_version, 3); + metadata->set(f_current_schema_id, 0); + metadata->set(f_last_column_id, 2); + + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto schema = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema->set(f_schema_id, 0); + schema->set(f_type, "struct"); + auto fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + + auto field_x = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field_x->set(f_id, 1); + field_x->set(f_name, "x"); + field_x->set(f_required, true); + field_x->set(f_type, "int"); + fields->add(field_x); + + auto field_y = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field_y->set(f_id, 2); + field_y->set(f_name, "y"); + field_y->set(f_required, true); + field_y->set(f_type, "string"); + fields->add(field_y); + + schema->set(f_fields, fields); + schemas->add(schema); + metadata->set(f_schemas, schemas); + + return metadata; +} + TEST(IcebergMetadataGenerator, GetIcebergTypeNothingProducesUnknown) { Int32 iter = 0; @@ -887,6 +924,63 @@ TEST(IcebergMetadataGenerator, ModifyColumnAfterMovesFieldAfterNamedColumn) } +TEST(IcebergMetadataGenerator, DropColumnRejectsIfInMultiArgPartitionSpec) +{ + auto metadata = makeMetadataWithTwoColumns(); + + auto partition_specs = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto spec = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + spec->set(f_spec_id, static_cast(1)); + auto spec_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + + auto pf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + /// Multi-arg: source-ids instead of source-id + auto source_ids = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + source_ids->add(1); + source_ids->add(2); + pf->set(f_source_ids, source_ids); + pf->set("transform", "bucket[16]"); + pf->set("name", "xy_bucket"); + spec_fields->add(pf); + spec->set(f_fields, spec_fields); + partition_specs->add(spec); + metadata->set(f_partition_specs, partition_specs); + metadata->set(f_default_spec_id, static_cast(1)); + + /// Dropping either column that participates in the multi-arg transform should be rejected. + expectDropRejected(metadata, "x"); + expectDropRejected(metadata, "y"); +} + + +TEST(IcebergMetadataGenerator, DropColumnRejectsIfInMultiArgSortOrder) +{ + auto metadata = makeMetadataWithTwoColumns(); + + auto sort_orders = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto sort_order = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + sort_order->set(f_order_id, static_cast(1)); + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + + auto sf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + auto source_ids = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + source_ids->add(1); + source_ids->add(2); + sf->set(f_source_ids, source_ids); + sf->set("transform", "bucket[16]"); + sf->set("direction", "asc"); + sf->set("null-order", "nulls-first"); + sort_fields->add(sf); + sort_order->set(f_fields, sort_fields); + sort_orders->add(sort_order); + metadata->set(f_sort_orders, sort_orders); + metadata->set(f_default_sort_order_id, static_cast(1)); + + expectDropRejected(metadata, "x"); + expectDropRejected(metadata, "y"); +} + + TEST(IcebergMetadataGenerator, ModifyColumnTypeAndFirstDoesBoth) { auto metadata = makeMetadataWithGap(); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_multi_arg_transforms.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_multi_arg_transforms.cpp new file mode 100644 index 0000000000000..e7876e4af7bf4 --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_multi_arg_transforms.cpp @@ -0,0 +1,205 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace DB; +using namespace DB::Iceberg; + +namespace DB::ErrorCodes +{ +extern const int ICEBERG_SPECIFICATION_VIOLATION; +} + +namespace +{ + +/// Build metadata with three columns (a: int id=1, b: int id=2, c: int id=3) +/// and a configurable sort order. +Poco::JSON::Object::Ptr makeMetadataWithSortOrder(const Poco::JSON::Array::Ptr & sort_fields) +{ + auto metadata = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + metadata->set(f_format_version, 3); + metadata->set(f_current_schema_id, 0); + metadata->set(f_last_column_id, 3); + + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto schema = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema->set(f_schema_id, 0); + schema->set(f_type, "struct"); + + auto fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + for (Int32 id = 1; id <= 3; ++id) + { + auto field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field->set(f_id, id); + field->set(f_name, String(1, static_cast('a' + id - 1))); + field->set(f_required, true); + field->set(f_type, "int"); + fields->add(field); + } + schema->set(f_fields, fields); + schemas->add(schema); + metadata->set(f_schemas, schemas); + + auto sort_order = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + sort_order->set(f_order_id, static_cast(1)); + sort_order->set(f_fields, sort_fields); + + auto sort_orders = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_orders->add(sort_order); + metadata->set(f_sort_orders, sort_orders); + metadata->set(f_default_sort_order_id, static_cast(1)); + + return metadata; +} + +/// Make an identity sort field referencing a single column by source-id. +Poco::JSON::Object::Ptr makeIdentitySortField(Int64 source_id, const String & direction = "asc") +{ + auto sf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + sf->set(f_source_id, source_id); + sf->set("transform", "identity"); + sf->set("direction", direction); + sf->set("null-order", "nulls-first"); + return sf; +} + +/// Make a multi-arg sort field (e.g. bucket[16]) referencing multiple source-ids. +Poco::JSON::Object::Ptr makeMultiArgSortField( + const std::vector & source_ids, const String & transform = "bucket[16]", const String & direction = "asc") +{ + auto sf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + auto ids = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + for (auto id : source_ids) + ids->add(id); + sf->set(f_source_ids, ids); + sf->set("transform", transform); + sf->set("direction", direction); + sf->set("null-order", "nulls-first"); + return sf; +} + +NamesAndTypesList threeIntColumns() +{ + return NamesAndTypesList{ + NameAndTypePair("a", std::make_shared()), + NameAndTypePair("b", std::make_shared()), + NameAndTypePair("c", std::make_shared()), + }; +} + +} + +/// --- Sorting key prefix tests --- + +TEST(IcebergMultiArgTransforms, SortKeyMultiArgFirstFieldReturnsEmpty) +{ + /// Sort order: [bucket[16](a, b) ASC, c ASC] + /// The first field is multi-arg and unrepresentable, so the entire key must be empty + /// (cannot skip to c — that is not a valid prefix of the sort order). + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_fields->add(makeMultiArgSortField({1, 2})); + sort_fields->add(makeIdentitySortField(3)); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + EXPECT_EQ(key.definition_ast, nullptr); +} + + +TEST(IcebergMultiArgTransforms, SortKeyMultiArgLastFieldKeepsPrefix) +{ + /// Sort order: [c ASC, bucket[16](a, b) ASC] + /// The prefix before the multi-arg field is (c ASC) — that must be kept. + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_fields->add(makeIdentitySortField(3)); + sort_fields->add(makeMultiArgSortField({1, 2})); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + ASSERT_NE(key.definition_ast, nullptr); + EXPECT_EQ(key.column_names.size(), 1); + EXPECT_EQ(key.column_names[0], "c"); +} + + +TEST(IcebergMultiArgTransforms, SortKeyMultiArgMiddleFieldKeepsOnlyPrefix) +{ + /// Sort order: [a ASC, bucket[16](a, b) ASC, c ASC] + /// Only (a ASC) is a valid prefix. + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_fields->add(makeIdentitySortField(1)); + sort_fields->add(makeMultiArgSortField({1, 2})); + sort_fields->add(makeIdentitySortField(3)); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + ASSERT_NE(key.definition_ast, nullptr); + EXPECT_EQ(key.column_names.size(), 1); + EXPECT_EQ(key.column_names[0], "a"); +} + + +TEST(IcebergMultiArgTransforms, SortKeyTwoNormalFieldsPrefix) +{ + /// Sort order: [a ASC, c ASC, bucket[16](a, b) ASC] + /// Prefix is (a ASC, c ASC). + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_fields->add(makeIdentitySortField(1)); + sort_fields->add(makeIdentitySortField(3)); + sort_fields->add(makeMultiArgSortField({1, 2})); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + ASSERT_NE(key.definition_ast, nullptr); + EXPECT_EQ(key.column_names.size(), 2); + EXPECT_EQ(key.column_names[0], "a"); + EXPECT_EQ(key.column_names[1], "c"); +} + + +TEST(IcebergMultiArgTransforms, SortKeyAllNormalFieldsUnchanged) +{ + /// Sort order: [a ASC, b ASC, c ASC] — no multi-arg at all. + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + sort_fields->add(makeIdentitySortField(1)); + sort_fields->add(makeIdentitySortField(2)); + sort_fields->add(makeIdentitySortField(3)); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + ASSERT_NE(key.definition_ast, nullptr); + EXPECT_EQ(key.column_names.size(), 3); +} + + +/// --- Empty source-ids validation --- + +TEST(IcebergMultiArgTransforms, SortFieldMissingSourceIdBreaksSortKey) +{ + /// A sort field with neither source-id nor source-ids: break stops processing. + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto bad_field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + bad_field->set("transform", "identity"); + bad_field->set("direction", "asc"); + bad_field->set("null-order", "nulls-first"); + sort_fields->add(bad_field); + sort_fields->add(makeIdentitySortField(3)); + + auto metadata = makeMetadataWithSortOrder(sort_fields); + auto key = getSortingKeyDescriptionFromMetadata(metadata, threeIntColumns(), getContext().context); + /// The first field has no source-id/source-ids -> break -> empty key. + EXPECT_EQ(key.definition_ast, nullptr); +} + +#endif diff --git a/tests/integration/test_storage_iceberg_with_spark/configs/config.d/.gitignore b/tests/integration/test_storage_iceberg_with_spark/configs/config.d/.gitignore new file mode 100644 index 0000000000000..d387464fbb00f --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/configs/config.d/.gitignore @@ -0,0 +1 @@ +cache_*.xml diff --git a/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.reference b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.reference new file mode 100644 index 0000000000000..469f2fffb9183 --- /dev/null +++ b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.reference @@ -0,0 +1,6 @@ +Multi-argument partition transforms (source-ids) are not supported for writes +no orphaned data files +0 +2 +3 +4 diff --git a/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.sh b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.sh new file mode 100755 index 0000000000000..b3b7c9882a986 --- /dev/null +++ b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.sh @@ -0,0 +1,90 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +# Data-file compaction (`OPTIMIZE TABLE` with `allow_experimental_iceberg_compaction`) partitions +# rewritten files by the default partition spec only after it has written them. A multi-argument +# (`source-ids`) transform cannot be evaluated on write, so such a spec must be rejected before any +# data file is written; otherwise the rewritten files are left orphaned in the table directory. +# +# The test creates a table with position deletes (so compaction has work to do), switches the +# default spec to a multi-argument `bucket` transform by patching the latest metadata file, and +# checks that `OPTIMIZE` fails without adding any file under `data/`. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +TABLE="t_${CLICKHOUSE_DATABASE}_${RANDOM}" +TABLE_PATH="${USER_FILES_PATH}/${TABLE}/" + +${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS ${TABLE}" +${CLICKHOUSE_CLIENT} --query " + CREATE TABLE ${TABLE} (x Int64, y String) + ENGINE = IcebergLocal('${TABLE_PATH}', 'Parquet') +" + +${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg=1 --query \ + "INSERT INTO ${TABLE} SELECT number, toString(number) FROM numbers(5)" + +# Merge-on-read delete: writes a position-delete file, which makes data-file compaction non-trivial. +# Delete a single row: compaction currently accepts an overwrite snapshot only when its +# `added-position-deletes` equals `added-delete-files`. +${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg=1 --query \ + "ALTER TABLE ${TABLE} DELETE WHERE x = 1" + +# Make a multi-argument bucket over (x, y) the default partition spec. Existing files keep spec 0. +python3 - "${TABLE_PATH}" << 'EOF' +import glob +import json +import re +import sys + +files = glob.glob(sys.argv[1] + "metadata/*.metadata.json") +assert files, "no metadata files found" + +def version(path): + match = re.search(r"(\d+)[^/]*\.metadata\.json$", path) + return int(match.group(1)) if match else -1 + +latest = max(files, key=version) +with open(latest) as f: + metadata = json.load(f) + +field_ids = {field["name"]: field["id"] for schema in metadata["schemas"] for field in schema["fields"]} +metadata["partition-specs"].append({ + "spec-id": 1, + "fields": [{"source-ids": [field_ids["x"], field_ids["y"]], "transform": "bucket[16]", "name": "xy_bucket", "field-id": 1000}], +}) +metadata["default-spec-id"] = 1 +metadata["last-partition-id"] = max(metadata.get("last-partition-id", 999), 1000) +with open(latest, "w") as f: + json.dump(metadata, f) +EOF + +count_data_files() +{ + find "${TABLE_PATH}data" -type f | wc -l +} + +files_before=$(count_data_files) + +optimize_output=$(${CLICKHOUSE_CLIENT} --use_iceberg_metadata_files_cache=0 --allow_experimental_iceberg_compaction=1 \ + --query "OPTIMIZE TABLE ${TABLE}" 2>&1) +if ! echo "${optimize_output}" | grep -q -m1 'Multi-argument partition transforms (source-ids) are not supported for writes'; then + echo "FAIL: OPTIMIZE did not reject the multi-argument spec:" + echo "${optimize_output}" + exit 1 +fi +echo 'Multi-argument partition transforms (source-ids) are not supported for writes' + +files_after=$(count_data_files) +if [ "${files_before}" -eq "${files_after}" ]; then + echo "no orphaned data files" +else + echo "FAIL: data files before OPTIMIZE: ${files_before}, after: ${files_after}" +fi + +${CLICKHOUSE_CLIENT} --use_iceberg_metadata_files_cache=0 --query "SELECT x FROM ${TABLE} ORDER BY x" + +${CLICKHOUSE_CLIENT} --query "DROP TABLE ${TABLE}" +rm -rf "${TABLE_PATH}"