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 17a6ab7f63b0..9a5456b1c343 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 @@ -1460,6 +1461,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; @@ -1499,6 +1510,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) { @@ -1560,12 +1593,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; @@ -1600,14 +1653,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