From 880b51fbbc4044e0fe5955bad8f2c22dd37f4feb Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Sun, 6 Sep 2026 16:40:09 +0200 Subject: [PATCH 1/4] Added logic to read source_ids and not throw an error --- .../DataLakes/Iceberg/ChunkPartitioner.cpp | 8 ++ .../DataLakes/Iceberg/Constant.h | 1 + .../DataLakes/Iceberg/ManifestFile.cpp | 10 +- .../DataLakes/Iceberg/ManifestFile.h | 7 +- .../Iceberg/ManifestFileIterator.cpp | 46 ++++++++- .../DataLakes/Iceberg/MetadataGenerator.cpp | 26 +++++ .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 93 ++++++++++++++++-- .../gtest_iceberg_metadata_generator.cpp | 95 +++++++++++++++++++ 8 files changed, 270 insertions(+), 16 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp index 924fe8842ee9..23372d6b51b3 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp @@ -49,6 +49,14 @@ ChunkPartitioner::ChunkPartitioner( FunctionOverloadResolverPtr transform; + /// 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_field->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"); + auto source_id = partition_specification_field->getValue(Iceberg::f_source_id); auto column_name = id_to_column[source_id]; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h index cd7c9f29d7e3..eff874094e47 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h @@ -117,6 +117,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/ManifestFile.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp index 201fdcd3a7eb..b1e16d047ea6 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 acf7a39b17e9..bb33a665816f 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h @@ -61,9 +61,14 @@ 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; + + /// 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 7856444feab0..2003f9b376e1 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -182,7 +182,48 @@ std::shared_ptr ManifestFileIterator::create( { 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_); + } + + 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}); + + /// 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)); @@ -190,9 +231,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); auto partition_ast = getASTFromTransform(transform_name, numeric_column_name, context_->getSettingsRef()[Setting::iceberg_partition_timezone]); /// Unsupported partition key expression if (partition_ast == nullptr) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 55ee1c99baf3..f908511d1ef8 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -526,6 +526,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; } @@ -553,6 +566,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 48f088cbe96e..c2ed5362a47d 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 @@ -1422,6 +1423,16 @@ 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, so skip these sort fields — this disables + /// the read-in-order optimization for such columns (safe: data is still correct). + if (field->has(f_source_ids)) + continue; + + if (!field->has(f_source_id)) + continue; + 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; @@ -1461,6 +1472,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) { @@ -1522,12 +1555,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; @@ -1562,14 +1615,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; 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 1f2c5bec0cb3..82cde2597ea1 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 @@ -491,4 +491,99 @@ TEST(IcebergMetadataGenerator, ModifyColumnWideningRecordsTheNewTypeInANewSchema EXPECT_EQ(stored_type.extract(), "long"); } + +/// --- 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, 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"); +} + #endif From df16563eff7cdf2d70167cce5ffba503f344d11c Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Mon, 5 Oct 2026 05:26:54 +0200 Subject: [PATCH 2/4] Address AI review comments --- .../DataLakes/Iceberg/IcebergMetadata.cpp | 10 +- .../Iceberg/ManifestFileIterator.cpp | 15 ++ .../DataLakes/Iceberg/ManifestFileIterator.h | 6 + .../Iceberg/ManifestFilesPruning.cpp | 16 +- .../DataLakes/Iceberg/ManifestFilesPruning.h | 2 + .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 9 +- .../gtest_iceberg_multi_arg_transforms.cpp | 205 ++++++++++++++++++ .../configs/config.d/.gitignore | 1 + 8 files changed, 256 insertions(+), 8 deletions(-) create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_multi_arg_transforms.cpp create mode 100644 tests/integration/test_storage_iceberg_with_spark/configs/config.d/.gitignore diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 7f7d7211c680..81ad5d33ce16 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1689,8 +1689,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/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 2003f9b376e1..a328bf462564 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -178,6 +178,9 @@ std::shared_ptr ManifestFileIterator::create( schema_processor.addIcebergTableSchema(schema_object, context_); 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)); @@ -213,6 +216,14 @@ std::shared_ptr ManifestFileIterator::create( 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}); @@ -238,6 +249,7 @@ std::shared_ptr ManifestFileIterator::create( partition_key_ast->as()->arguments->children.emplace_back(std::move(partition_ast)); 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; @@ -259,6 +271,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_)); @@ -276,6 +289,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_) @@ -289,6 +303,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_) , data_files_without_deleted(std::make_shared>()) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h index e109cd107890..9fcc8921dd2d 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h @@ -90,6 +90,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; @@ -121,6 +124,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); @@ -139,6 +143,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 483fb979bbbe..07bbd1a6df8d 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp @@ -130,6 +130,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 */); @@ -214,7 +215,18 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( if (partition_key_condition.has_value()) { const auto & partition_value = entry->parsed_entry->partition_key_value; - std::vector index_value(partition_value.begin(), partition_value.end()); + + /// 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. + std::vector index_value; + index_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 + index_value.push_back(FieldRef(partition_value[pos])); + } + for (size_t i = 0; i < index_value.size(); ++i) { auto & field = index_value[i]; @@ -230,7 +242,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) { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h index d04ced3796ee..e24e08e891d3 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/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index cb9bc7252b02..e8d65113a521 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -1475,13 +1475,14 @@ KeyDescription getSortingKeyDescriptionFromMetadata(Poco::JSON::Object::Ptr meta auto field = fields->getObject(field_index); /// Iceberg V3: multi-argument transforms use `source-ids` instead of `source-id`. - /// We cannot evaluate multi-arg transforms, so skip these sort fields — this disables - /// the read-in-order optimization for such columns (safe: data is still correct). + /// 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)) - continue; + break; if (!field->has(f_source_id)) - continue; + break; auto source_id = field->getValue(f_source_id); auto column_name = source_id_to_column_name[source_id]; 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 000000000000..e7876e4af7bf --- /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 000000000000..d387464fbb00 --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/configs/config.d/.gitignore @@ -0,0 +1 @@ +cache_*.xml From 2bea22995f36eaed430485039364f4da3715769a Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 14 Oct 2026 23:34:47 +0200 Subject: [PATCH 3/4] Fix iceberg metadata generator test --- .../gtest_iceberg_metadata_generator.cpp | 88 ++++++++++--------- 1 file changed, 47 insertions(+), 41 deletions(-) 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 f32a2c30cab1..ae701f77d493 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 @@ -637,6 +637,32 @@ static Poco::JSON::Object::Ptr makeMetadataWithTwoColumns() 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; @@ -703,26 +729,31 @@ TEST(IcebergMetadataGenerator, ModifyColumnAfterMovesFieldAfterNamedColumn) 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); - + for (Int32 i = 0; i < 3; ++i) + { + auto field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field->set(f_id, i + 1); + field->set(f_name, String(1, static_cast('a' + i))); + field->set(f_required, false); + field->set(f_type, "long"); + fields->add(field); + } schema->set(f_fields, fields); + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); schemas->add(schema); metadata->set(f_schemas, schemas); - return metadata; + MetadataGenerator gen(metadata); + + EXPECT_TRUE(gen.generateModifyColumnMetadata( + "a", makeNullable(std::make_shared()), getContext().context, + /* first */ false, /* after_column */ "b")); + + auto names = getCurrentFieldNames(metadata); + ASSERT_EQ(names.size(), 3u); + EXPECT_EQ(names[0], "b"); + EXPECT_EQ(names[1], "a"); + EXPECT_EQ(names[2], "c"); } @@ -780,31 +811,6 @@ TEST(IcebergMetadataGenerator, DropColumnRejectsIfInMultiArgSortOrder) expectDropRejected(metadata, "x"); expectDropRejected(metadata, "y"); - for (Int32 i = 0; i < 3; ++i) - { - auto field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); - field->set(f_id, i + 1); - field->set(f_name, String(1, static_cast('a' + i))); - field->set(f_required, false); - field->set(f_type, "long"); - fields->add(field); - } - schema->set(f_fields, fields); - auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); - schemas->add(schema); - metadata->set(f_schemas, schemas); - - MetadataGenerator gen(metadata); - - EXPECT_TRUE(gen.generateModifyColumnMetadata( - "a", makeNullable(std::make_shared()), getContext().context, - /* first */ false, /* after_column */ "b")); - - auto names = getCurrentFieldNames(metadata); - ASSERT_EQ(names.size(), 3u); - EXPECT_EQ(names[0], "b"); - EXPECT_EQ(names[1], "a"); - EXPECT_EQ(names[2], "c"); } From b63e0a2e41fcd831f806a4eb1f6c9729fa23fb80 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 15 Oct 2026 04:10:32 +0200 Subject: [PATCH 4/4] Make optimize reject multi source-id partition specs so ChunkPartitioner rejection can no longer leave orphaned files --- .../MergeTree/ExportPartitionUtils.cpp | 3 + .../DataLakes/Iceberg/ChunkPartitioner.cpp | 27 ++++-- .../DataLakes/Iceberg/ChunkPartitioner.h | 4 + .../DataLakes/Iceberg/Compaction.cpp | 24 ++++++ ...timize_multi_arg_spec_no_orphans.reference | 5 ++ ...berg_optimize_multi_arg_spec_no_orphans.sh | 83 +++++++++++++++++++ 6 files changed, 138 insertions(+), 8 deletions(-) create mode 100644 tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.reference create mode 100755 tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.sh diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 57a852be327e..41406b45c4fa 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/ChunkPartitioner.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.cpp index 23372d6b51b3..5a4628112951 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) @@ -49,14 +68,6 @@ ChunkPartitioner::ChunkPartitioner( FunctionOverloadResolverPtr transform; - /// 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_field->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"); - auto source_id = partition_specification_field->getValue(Iceberg::f_source_id); auto column_name = id_to_column[source_id]; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h index 7395ed0c8b9c..38ac02e69b51 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ChunkPartitioner.h @@ -24,6 +24,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 e3f046866040..1ee30bfd4708 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp @@ -206,6 +206,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; @@ -511,6 +534,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/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 000000000000..4286d961ce94 --- /dev/null +++ b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.reference @@ -0,0 +1,5 @@ +Multi-argument partition transforms (source-ids) are not supported for writes +no orphaned data files +0 +2 +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 000000000000..e2582ceaede5 --- /dev/null +++ b/tests/queries/0_stateless/05301_iceberg_optimize_multi_arg_spec_no_orphans.sh @@ -0,0 +1,83 @@ +#!/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. +${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg=1 --query \ + "ALTER TABLE ${TABLE} DELETE WHERE x IN (1, 3)" + +# 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) + +${CLICKHOUSE_CLIENT} --use_iceberg_metadata_files_cache=0 --allow_experimental_iceberg_compaction=1 \ + --query "OPTIMIZE TABLE ${TABLE}" 2>&1 \ + | grep -o -m1 '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}"