From 4909c63624174a11d33f9668515bb8c1bdd56122 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 10 Nov 2026 22:17:23 +0800 Subject: [PATCH 1/2] fix(parquet): harden reads of files without field IDs Follow-up to #1044. - Build NameMapping and MappedFields lookup indexes on construction. They were built lazily in mutable members, so readers sharing one mapping raced on the first lookup (and on every lookup of an empty mapping). Matches Java's NameMapping constructor. - Without a name mapping, match Parquet files without field IDs by the read schema's names, as Java's Avro and ORC readers do, instead of failing. This also covers files with no columns. - Read a list or map whose element, key or value has no ID as missing (initial default, null, or required-field error), like Java's PruneColumns, instead of failing the whole file. - When aliases map two sibling columns to one ID, keep it on the last column, like Java's reader, instead of failing with a duplicate ID. - Replace the recursive name-mapping helpers with a small applier that builds the lookup path incrementally and uses the shared element, key and value name constants. - Write the no-ID test files with non-compliant nested names so the nested name-mapping test actually covers "item" list elements. Co-Authored-By: Claude Opus 5.5 --- src/iceberg/file_reader.h | 3 +- src/iceberg/name_mapping.cc | 97 +++------ src/iceberg/name_mapping.h | 26 +-- src/iceberg/parquet/parquet_reader.cc | 8 +- src/iceberg/parquet/parquet_schema_util.cc | 193 +++++++++++++----- .../parquet/parquet_schema_util_internal.h | 3 +- src/iceberg/test/name_mapping_test.cc | 32 +++ src/iceberg/test/parquet_schema_test.cc | 100 +++++++++ src/iceberg/test/parquet_test.cc | 137 ++++++++++++- 9 files changed, 452 insertions(+), 147 deletions(-) diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index eded9191c..1e3107601 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -116,7 +116,8 @@ struct ICEBERG_EXPORT ReaderOptions { /// \brief Case sensitivity inherited from the scan when resolving field names. bool case_sensitive = true; /// \brief Name mapping for schema evolution compatibility. Used when reading files - /// that may have different field names than the current schema. + /// that may have different field names than the current schema. When it is null, the + /// Parquet reader matches files without field IDs by the projection's field names. std::shared_ptr name_mapping; /// \brief First row ID inherited by the data file, if assigned. std::optional first_row_id; diff --git a/src/iceberg/name_mapping.cc b/src/iceberg/name_mapping.cc index ff2d32507..54bb7a1bd 100644 --- a/src/iceberg/name_mapping.cc +++ b/src/iceberg/name_mapping.cc @@ -20,7 +20,6 @@ #include "iceberg/name_mapping.h" #include -#include #include "iceberg/util/formatter_internal.h" #include "iceberg/util/macros.h" @@ -32,14 +31,14 @@ namespace { // Helper function to join a list of field names with a dot std::string JoinByDot(std::span parts) { - std::stringstream ss; - for (size_t i = 0; i < parts.size(); ++i) { - if (i > 0) { - ss << "."; + std::string joined; + for (const auto& part : parts) { + if (!joined.empty()) { + joined.push_back('.'); } - ss << parts[i]; + joined.append(part); } - return ss.str(); + return joined; } // Helper class to recursively index MappedField by field id @@ -79,8 +78,8 @@ struct IndexByNameVisitor { for (const auto& [name, mapped_field] : nested_visitor.field_by_name) { for (const auto& prefix : field.names) { - std::vector parts = {prefix, name}; - field_by_name.emplace(JoinByDot(parts), std::cref(mapped_field)); + field_by_name.emplace(std::format("{}.{}", prefix, name), + std::cref(mapped_field)); } } } @@ -97,24 +96,31 @@ struct IndexByNameVisitor { } // namespace -MappedFields::MappedFields(std::vector fields) - : fields_(std::move(fields)) {} +MappedFields::MappedFields(std::vector fields) : fields_(std::move(fields)) { + for (const auto& field : fields_) { + if (!field.field_id.has_value()) { + continue; + } + id_to_field_.emplace(field.field_id.value(), std::cref(field)); + for (const auto& name : field.names) { + name_to_id_.emplace(name, field.field_id.value()); + } + } +} std::unique_ptr MappedFields::Make(std::vector fields) { return std::unique_ptr(new MappedFields(std::move(fields))); } std::optional MappedFields::Field(int32_t id) const { - const auto& id_to_field = LazyIdToField(); - if (auto it = id_to_field.find(id); it != id_to_field.cend()) { + if (auto it = id_to_field_.find(id); it != id_to_field_.cend()) { return it->second; } return std::nullopt; } std::optional MappedFields::Id(std::string_view name) const { - const auto& name_to_id = LazyNameToId(); - if (auto it = name_to_id.find(name); it != name_to_id.cend()) { + if (auto it = name_to_id_.find(name); it != name_to_id_.cend()) { return it->second; } return std::nullopt; @@ -124,37 +130,19 @@ size_t MappedFields::Size() const { return fields_.size(); } std::span MappedFields::fields() const { return fields_; } -const std::unordered_map& MappedFields::LazyNameToId() const { - if (name_to_id_.empty() && !fields_.empty()) { - for (const auto& field : fields_) { - for (const auto& name : field.names) { - if (field.field_id.has_value()) { - name_to_id_.emplace(name, field.field_id.value()); - } - } - } - } - return name_to_id_; -} - -const std::unordered_map& MappedFields::LazyIdToField() - const { - if (id_to_field_.empty() && !fields_.empty()) { - for (const auto& field : fields_) { - if (field.field_id.has_value()) { - id_to_field_.emplace(field.field_id.value(), std::cref(field)); - } - } - } - return id_to_field_; -} - NameMapping::NameMapping(std::unique_ptr mapping) - : mapping_(std::move(mapping)) {} + : mapping_(std::move(mapping)) { + IndexByIdVisitor id_visitor; + id_visitor.Visit(AsMappedFields()); + fields_by_id_ = std::move(id_visitor.field_by_id); + + IndexByNameVisitor name_visitor; + name_visitor.Visit(AsMappedFields()); + fields_by_name_ = std::move(name_visitor.field_by_name); +} std::optional NameMapping::Find(int32_t id) const { - const auto& fields_by_id = LazyFieldsById(); - if (auto iter = fields_by_id.find(id); iter != fields_by_id.cend()) { + if (auto iter = fields_by_id_.find(id); iter != fields_by_id_.cend()) { return iter->second; } return std::nullopt; @@ -169,8 +157,7 @@ std::optional NameMapping::Find( } std::optional NameMapping::Find(const std::string& name) const { - const auto& fields_by_name = LazyFieldsByName(); - if (auto iter = fields_by_name.find(name); iter != fields_by_name.cend()) { + if (auto iter = fields_by_name_.find(name); iter != fields_by_name_.cend()) { return iter->second; } return std::nullopt; @@ -184,26 +171,6 @@ const MappedFields& NameMapping::AsMappedFields() const { return *mapping_; } -const std::unordered_map& NameMapping::LazyFieldsById() - const { - if (fields_by_id_.empty()) { - IndexByIdVisitor visitor; - visitor.Visit(AsMappedFields()); - fields_by_id_ = std::move(visitor.field_by_id); - } - return fields_by_id_; -} - -const std::unordered_map& -NameMapping::LazyFieldsByName() const { - if (fields_by_name_.empty()) { - IndexByNameVisitor visitor; - visitor.Visit(AsMappedFields()); - fields_by_name_ = std::move(visitor.field_by_name); - } - return fields_by_name_; -} - std::unique_ptr NameMapping::MakeEmpty() { return std::unique_ptr(new NameMapping(MappedFields::Make({}))); } diff --git a/src/iceberg/name_mapping.h b/src/iceberg/name_mapping.h index 7fd140c0d..973518b41 100644 --- a/src/iceberg/name_mapping.h +++ b/src/iceberg/name_mapping.h @@ -56,6 +56,9 @@ struct ICEBERG_EXPORT MappedField { using MappedFieldConstRef = std::reference_wrapper; /// \brief A list of field mappings for child field of structs, maps, and lists. +/// +/// The lookup indexes are built on construction, so a const instance is safe to share +/// across threads. class ICEBERG_EXPORT MappedFields { public: /// \brief Create a new MappedFields instance. @@ -63,6 +66,10 @@ class ICEBERG_EXPORT MappedFields { /// \return A new MappedFields instance. static std::unique_ptr Make(std::vector fields); + // The indexes refer into `fields_`, so a copy would point at the source's fields. + MappedFields(const MappedFields&) = delete; + MappedFields& operator=(const MappedFields&) = delete; + /// \brief Get the field for a given field ID. /// \param[in] id The ID of the field. /// \return The field for the given field ID. @@ -84,18 +91,17 @@ class ICEBERG_EXPORT MappedFields { private: explicit MappedFields(std::vector fields); - const std::unordered_map& LazyNameToId() const; - const std::unordered_map& LazyIdToField() const; - private: std::vector fields_; - // Lazy-initialized mappings - mutable std::unordered_map name_to_id_; - mutable std::unordered_map id_to_field_; + std::unordered_map name_to_id_; + std::unordered_map id_to_field_; }; /// \brief Represents a mapping from external schema names to Iceberg type IDs. +/// +/// The lookup indexes are built on construction, so a const instance is safe to share +/// across threads, for example by the readers of a scan. class ICEBERG_EXPORT NameMapping { public: /// \brief Create a new NameMapping instance. @@ -124,15 +130,11 @@ class ICEBERG_EXPORT NameMapping { private: explicit NameMapping(std::unique_ptr mapping); - const std::unordered_map& LazyFieldsById() const; - const std::unordered_map& LazyFieldsByName() const; - private: std::unique_ptr mapping_; - // Lazy-initialized mappings - mutable std::unordered_map fields_by_id_; - mutable std::unordered_map fields_by_name_; + std::unordered_map fields_by_id_; + std::unordered_map fields_by_name_; }; ICEBERG_EXPORT std::string ToString(const MappedField& field); diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index b88dfd000..f59f524c5 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -42,6 +42,7 @@ #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/metadata_column_util_internal.h" #include "iceberg/expression/binder.h" +#include "iceberg/name_mapping.h" #include "iceberg/parquet/parquet_data_util_internal.h" #include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" #include "iceberg/parquet/parquet_register.h" @@ -68,9 +69,12 @@ Result BuildProjection(::parquet::arrow::FileReader* reader, auto metadata = reader->parquet_reader()->metadata(); if (!HasFieldIds(metadata->schema()->schema_root())) { + // Like Java's Avro and ORC readers, fall back to matching the read schema's names + // when the table has no name mapping. + std::unique_ptr fallback_mapping; if (name_mapping == nullptr) { - return InvalidSchema( - "Parquet file schema has no field IDs and no name mapping provided"); + ICEBERG_ASSIGN_OR_RAISE(fallback_mapping, CreateMapping(read_schema)); + name_mapping = fallback_mapping.get(); } ::parquet::arrow::SchemaManifest schema_manifest; ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make( diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index d16b8a1ae..0d3ed173a 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -293,6 +293,43 @@ void SelectAnchorColumnIfEmpty( } } +// Lists and maps are matched to the file by the field IDs of their element, key and +// value. Returns false when the Parquet field leaves any of them without an ID, for +// example when a partial name mapping does not cover them. Like Java's PruneColumns, +// the caller then reads the container as if it were missing from the file. A Parquet +// field of the wrong shape returns true so that projection reports the mismatch. +bool HasNestedFieldIds(const Type& expected_type, + const ::parquet::arrow::SchemaField& parquet_field) { + const auto arrow_type_id = parquet_field.field->type()->id(); + switch (expected_type.type_id()) { + case TypeId::kList: { + if ((arrow_type_id != ::arrow::Type::LIST && + arrow_type_id != ::arrow::Type::LARGE_LIST) || + parquet_field.children.size() != 1) { + return true; + } + const auto& element = parquet_field.children[0]; + const auto& list_type = internal::checked_cast(expected_type); + return GetFieldId(element).has_value() && + HasNestedFieldIds(*list_type.element().type(), element); + } + case TypeId::kMap: { + if (arrow_type_id != ::arrow::Type::MAP || parquet_field.children.size() != 1 || + parquet_field.children[0].children.size() != 2) { + return true; + } + const auto& key = parquet_field.children[0].children[0]; + const auto& value = parquet_field.children[0].children[1]; + const auto& map_type = internal::checked_cast(expected_type); + return GetFieldId(key).has_value() && GetFieldId(value).has_value() && + HasNestedFieldIds(*map_type.key().type(), key) && + HasNestedFieldIds(*map_type.value().type(), value); + } + default: + return true; + } +} + template Status ValidateGeospatialParameters(const IcebergType& expected, const ParquetType& actual, @@ -555,6 +592,23 @@ class ProjectionBuilder { return projection; } + // Project a field that cannot be read from the file. + static Result ProjectMissingField(const SchemaField& field) { + FieldProjection projection; + if (MetadataColumns::IsMetadataColumn(field.field_id())) { + projection.kind = FieldProjection::Kind::kMetadata; + } else if (field.initial_default() != nullptr) { + // Rows written before the field existed assume its `initial-default` value. + projection.kind = FieldProjection::Kind::kDefault; + projection.from = *field.initial_default(); + } else if (field.optional()) { + projection.kind = FieldProjection::Kind::kNull; + } else { + return InvalidSchema("Missing required field with id: {}", field.field_id()); + } + return projection; + } + Result ProjectStruct( const StructType& struct_type, const std::vector<::parquet::arrow::SchemaField>& parquet_fields) { @@ -584,24 +638,16 @@ class ProjectionBuilder { result.children.reserve(struct_type.fields().size()); for (const auto& field : struct_type.fields()) { - int32_t field_id = field.field_id(); FieldProjection child_projection; - if (auto iter = field_context_map.find(field_id); - iter != field_context_map.cend()) { + if (auto iter = field_context_map.find(field.field_id()); + iter != field_context_map.cend() && + HasNestedFieldIds(*field.type(), iter->second.parquet_field)) { const auto& parquet_field = iter->second.parquet_field; ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(field, parquet_field, iter->second.local_index)); - } else if (MetadataColumns::IsMetadataColumn(field_id)) { - child_projection.kind = FieldProjection::Kind::kMetadata; - } else if (field.initial_default() != nullptr) { - // Rows written before the field existed assume its `initial-default` value. - child_projection.kind = FieldProjection::Kind::kDefault; - child_projection.from = *field.initial_default(); - } else if (field.optional()) { - child_projection.kind = FieldProjection::Kind::kNull; } else { - return InvalidSchema("Missing required field with id: {}", field_id); + ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectMissingField(field)); } result.children.emplace_back(std::move(child_projection)); @@ -735,51 +781,89 @@ void CollectColumnIds(const FieldProjection& field_projection, } } -// Forward declaration -void ApplyNameMappingToField(const NameMapping& mapping, std::vector& names, - ::parquet::arrow::SchemaField* parquet_field); +// Assigns field IDs from a name mapping to a Parquet schema without field IDs, like +// Java's ApplyNameMapping. List and map children are looked up as "element", "key" and +// "value", whatever their physical names in the file. +class NameMappingApplier { + public: + explicit NameMappingApplier(const NameMapping& mapping) : mapping_(mapping) {} -void ApplyNameMappingToChild(const NameMapping& mapping, std::vector& names, - std::string name, ::parquet::arrow::SchemaField* child) { - names.push_back(std::move(name)); - ApplyNameMappingToField(mapping, names, child); - names.pop_back(); -} + void ApplyToStruct(std::vector<::parquet::arrow::SchemaField>& fields) { + struct MappedSibling { + size_t index; + std::shared_ptr<::arrow::Field> unmapped_field; + }; + std::unordered_map mapped_siblings; + for (size_t i = 0; i < fields.size(); ++i) { + auto unmapped_field = fields[i].field; + auto field_id = ApplyToChild(unmapped_field->name(), &fields[i]); + if (!field_id.has_value()) { + continue; + } + auto [iter, inserted] = mapped_siblings.try_emplace( + field_id.value(), MappedSibling{.index = i, .unmapped_field = unmapped_field}); + if (!inserted) { + // Aliases can map two sibling columns to the same ID. Java's reader keeps the + // last one, so drop the ID from the earlier column. + fields[iter->second.index].field = std::move(iter->second.unmapped_field); + iter->second = MappedSibling{.index = i, .unmapped_field = unmapped_field}; + } + } + } -void ApplyNameMappingToField(const NameMapping& mapping, std::vector& names, - ::parquet::arrow::SchemaField* parquet_field) { - if (auto mapped_field = mapping.Find(names); - mapped_field.has_value() && mapped_field->get().field_id.has_value()) { - parquet_field->field = - parquet_field->field->WithMergedMetadata(::arrow::key_value_metadata( - {std::string(kParquetFieldIdKey)}, - {std::to_string(mapped_field->get().field_id.value())})); + private: + // Returns the field ID assigned to `field`, if any. + std::optional ApplyToChild(std::string_view name, + ::parquet::arrow::SchemaField* field) { + const size_t parent_size = path_.size(); + if (!path_.empty()) { + path_.push_back('.'); + } + path_.append(name); + auto field_id = Apply(field); + path_.resize(parent_size); + return field_id; } - auto& children = parquet_field->children; - switch (parquet_field->field->type()->id()) { - case ::arrow::Type::STRUCT: - for (auto& child : children) { - ApplyNameMappingToChild(mapping, names, child.field->name(), &child); - } - break; - case ::arrow::Type::LIST: - case ::arrow::Type::LARGE_LIST: - if (children.size() == 1) { - ApplyNameMappingToChild(mapping, names, "element", &children[0]); - } - break; - case ::arrow::Type::MAP: - // The key and value sit under the entries struct, which the name path skips. - if (children.size() == 1 && children[0].children.size() == 2) { - ApplyNameMappingToChild(mapping, names, "key", &children[0].children[0]); - ApplyNameMappingToChild(mapping, names, "value", &children[0].children[1]); - } - break; - default: - break; + std::optional Apply(::parquet::arrow::SchemaField* parquet_field) { + std::optional field_id; + if (auto mapped_field = mapping_.Find(path_); mapped_field.has_value()) { + field_id = mapped_field->get().field_id; + } + if (field_id.has_value()) { + parquet_field->field = + parquet_field->field->WithMergedMetadata(::arrow::key_value_metadata( + {std::string(kParquetFieldIdKey)}, {std::to_string(field_id.value())})); + } + + auto& children = parquet_field->children; + switch (parquet_field->field->type()->id()) { + case ::arrow::Type::STRUCT: + ApplyToStruct(children); + break; + case ::arrow::Type::LIST: + case ::arrow::Type::LARGE_LIST: + if (children.size() == 1) { + ApplyToChild(ListType::kElementName, &children[0]); + } + break; + case ::arrow::Type::MAP: + // The key and value sit under the entries struct, which the name path skips. + if (children.size() == 1 && children[0].children.size() == 2) { + ApplyToChild(MapType::kKeyName, &children[0].children[0]); + ApplyToChild(MapType::kValueName, &children[0].children[1]); + } + break; + default: + break; + } + return field_id; } -} + + const NameMapping& mapping_; + // Dot-joined names of the field being visited, the key used by NameMapping::Find. + std::string path_; +}; } // namespace @@ -820,10 +904,7 @@ bool HasFieldIds(const ::parquet::schema::NodePtr& node) { void ApplyNameMapping(const NameMapping& mapping, ::parquet::arrow::SchemaManifest* parquet_schema) { - std::vector names; - for (auto& parquet_field : parquet_schema->schema_fields) { - ApplyNameMappingToChild(mapping, names, parquet_field.field->name(), &parquet_field); - } + NameMappingApplier(mapping).ApplyToStruct(parquet_schema->schema_fields); } Result> ToParquetSchema( diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h index 49c62ff09..dc90b3a78 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -74,7 +74,8 @@ bool HasFieldIds(const ::parquet::schema::NodePtr& root_node); /// \brief Assign field IDs from a name mapping to a Parquet schema without field IDs. /// /// List elements use "element"; map keys and values use "key" and "value". -/// Fields without a mapped ID remain unassigned. +/// Fields without a mapped ID remain unassigned. When aliases map several sibling +/// fields to the same ID, only the last of them keeps it, matching Java's reader. /// /// \param mapping The name mapping to apply. /// \param parquet_schema The Parquet schema to update in place. diff --git a/src/iceberg/test/name_mapping_test.cc b/src/iceberg/test/name_mapping_test.cc index 70f71560a..415d1cec2 100644 --- a/src/iceberg/test/name_mapping_test.cc +++ b/src/iceberg/test/name_mapping_test.cc @@ -19,8 +19,11 @@ #include "iceberg/name_mapping.h" +#include #include +#include #include +#include #include #include @@ -189,6 +192,35 @@ TEST_F(NameMappingTest, FindMethodsOnConstEmptyMapping) { EXPECT_FALSE(field_by_parts.has_value()); } +// Readers of a scan share one mapping, so concurrent lookups must not race. A data race +// here is reliably reported only under ThreadSanitizer. +TEST_F(NameMappingTest, ConcurrentFind) { + const std::shared_ptr mapping = MakeNameMapping(); + const std::shared_ptr empty_mapping = NameMapping::MakeEmpty(); + const auto& nested_fields = mapping->Find(3)->get().nested_mapping; + + std::vector threads; + std::atomic failures = 0; + for (int i = 0; i < 8; ++i) { + threads.emplace_back([&] { + for (int j = 0; j < 100; ++j) { + bool ok = mapping->Find("qux.hello").has_value() && + mapping->Find(5).has_value() && + nested_fields->Id("world") == std::optional(5) && + nested_fields->Field(4).has_value() && + !empty_mapping->Find("foo").has_value(); + if (!ok) { + ++failures; + } + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(failures, 0); +} + TEST_F(NameMappingTest, Equality) { auto mapping1 = MakeNameMapping(); auto mapping2 = MakeNameMapping(); diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index c9f503691..3af9d875e 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -356,6 +356,106 @@ TEST(ParquetNameMappingTest, ApplyNameMappingWithUnmappedFields) { ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0})); } +TEST(ParquetNameMappingTest, ApplyNameMappingKeepsLastAliasedSibling) { + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/2, "full_name", string()), + }); + + auto parquet_schema = MakeGroupNode( + "iceberg_schema", {MakeStringNode("name"), MakeStringNode("full_name")}); + auto schema_manifest = MakeSchemaManifest(parquet_schema); + auto mapping = NameMapping::Make({ + MappedField{.names = {"name", "full_name"}, .field_id = 2}, + }); + + ApplyNameMapping(*mapping, &schema_manifest); + const auto& first_metadata = schema_manifest.schema_fields[0].field->metadata(); + EXPECT_TRUE(first_metadata == nullptr || + first_metadata->FindKey(std::string(kParquetFieldIdKey)) < 0); + EXPECT_EQ(schema_manifest.schema_fields[1].field->metadata()->Get( + std::string(kParquetFieldIdKey)), + "2"); + + auto projection_result = Project(expected_schema, schema_manifest); + ASSERT_THAT(projection_result, IsOk()); + + // `from` indexes the selected columns, which only hold the "full_name" column. + const auto& projection = *projection_result; + ASSERT_EQ(projection.fields.size(), 1); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_EQ(SelectedColumnIndices(projection), std::vector({1})); +} + +TEST(ParquetNameMappingTest, ProjectContainersWithUnmappedChildrenAsMissing) { + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/1, "id", int64()), + SchemaField::MakeOptional( + /*field_id=*/2, "numbers", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/201, "element", int32()))), + SchemaField::MakeOptional( + /*field_id=*/3, "counts", + std::make_shared( + SchemaField::MakeRequired(/*field_id=*/301, "key", string()), + SchemaField::MakeOptional(/*field_id=*/302, "value", int32()))), + SchemaField::MakeOptional( + /*field_id=*/4, "nested", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/401, "element", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/402, "element", int32()))))), + }); + + auto parquet_schema = MakeGroupNode( + "iceberg_schema", + { + MakeInt64Node("id"), + MakeListNode("numbers", MakeInt32Node("element")), + MakeMapNode("counts", + MakeStringNode("key", /*field_id=*/-1, /*optional=*/false), + MakeInt32Node("value")), + MakeListNode("nested", MakeListNode("element", MakeInt32Node("element"))), + }); + // Each container is mapped, but some of its children are not. + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"numbers"}, .field_id = 2}, + MappedField{.names = {"counts"}, + .field_id = 3, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"value"}, .field_id = 302}, + })}, + MappedField{.names = {"nested"}, + .field_id = 4, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"element"}, .field_id = 401}, + })}, + }); + + auto schema_manifest = MakeSchemaManifest(parquet_schema); + ApplyNameMapping(*mapping, &schema_manifest); + auto projection_result = Project(expected_schema, schema_manifest); + ASSERT_THAT(projection_result, IsOk()); + + const auto& projection = *projection_result; + ASSERT_EQ(projection.fields.size(), 4); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_PROJECTED_NULL_FIELD(projection.fields[1]); + ASSERT_PROJECTED_NULL_FIELD(projection.fields[2]); + ASSERT_PROJECTED_NULL_FIELD(projection.fields[3]); + ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0})); + + // A required container that cannot be read is still an error. + Schema required_schema({ + SchemaField::MakeRequired( + /*field_id=*/2, "numbers", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/201, "element", int32()))), + }); + EXPECT_THAT(Project(required_schema, schema_manifest), + HasErrorMessage("Missing required field with id: 2")); +} + TEST(ParquetSchemaProjectionTest, ProjectIdenticalSchemas) { Schema expected_schema({ SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int64()), diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index c35ba4471..da997f205 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -276,12 +276,29 @@ class ParquetReaderTest : public TempFileTestBase { auto io = internal::checked_cast(*file_io_); auto outfile = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie(); - ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(), - outfile, table->num_rows()) + // Keep the Arrow names of list elements (such as "item") instead of the standard + // "element", so that reads exercise the name mapping's path normalization. + auto arrow_writer_props = ::parquet::ArrowWriterProperties::Builder() + .disable_compliant_nested_types() + ->build(); + ASSERT_TRUE(::parquet::arrow::WriteTable( + *table, ::arrow::default_memory_pool(), outfile, table->num_rows(), + ::parquet::default_writer_properties(), arrow_writer_props) .ok()); ASSERT_TRUE(outfile->Close().ok()); } + std::vector ParquetColumnPaths() { + auto& io = internal::checked_cast(*file_io_); + auto input_file = io.fs()->OpenInputFile(temp_parquet_file_).ValueOrDie(); + auto metadata = ::parquet::ReadMetaData(input_file); + std::vector paths; + for (int i = 0; i < metadata->schema()->num_columns(); ++i) { + paths.push_back(metadata->schema()->Column(i)->path()->ToDotString()); + } + return paths; + } + void CreateListParquetFile() { auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "id", int32()), @@ -572,16 +589,52 @@ TEST_F(ParquetReaderTest, ReadTwoFields) { } TEST_F(ParquetReaderTest, ReadWithoutFieldIdsOrNameMapping) { + // Without a name mapping, columns are matched by the read schema's names. ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( - ::arrow::struct_({::arrow::field("id", ::arrow::int32())}), R"([[1]])")); - auto schema = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "id", int32())}); + ::arrow::struct_({::arrow::field("id", ::arrow::int32()), + ::arrow::field("name", ::arrow::utf8())}), + R"([[1, "Foo"], [2, "Bar"]])")); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(2, "name", string()), + SchemaField::MakeOptional(7, "id", int64()), + SchemaField::MakeOptional(3, "full_name", string()), + }); - auto reader_result = ReaderFactoryRegistry::Open( - FileFormatType::kParquet, - {.path = temp_parquet_file_, .io = file_io_, .projection = schema}); - ASSERT_THAT(reader_result, IsError(ErrorKind::kInvalidSchema)); - EXPECT_THAT(reader_result, HasErrorMessage("no field IDs and no name mapping")); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, + ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + {.path = temp_parquet_file_, .io = file_io_, .projection = schema})); + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([["Foo", 1, null], ["Bar", 2, null]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadNestedFieldsWithoutFieldIdsOrNameMapping) { + auto type = ::arrow::struct_({ + ::arrow::field("numbers", ::arrow::list(::arrow::field("item", ::arrow::int32()))), + ::arrow::field("counts", ::arrow::map(::arrow::utf8(), ::arrow::int32())), + }); + const std::string json = R"([[[1, 2], [["a", 3]]], [null, null]])"; + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds(type, json)); + + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional( + 1, "numbers", + std::make_shared(SchemaField::MakeOptional(101, "element", int64()))), + SchemaField::MakeOptional( + 2, "counts", + std::make_shared(SchemaField::MakeRequired(201, "key", string()), + SchemaField::MakeOptional(202, "value", int64()))), + }); + + ICEBERG_UNWRAP_OR_FAIL( + auto reader, + ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + {.path = temp_parquet_file_, .io = file_io_, .projection = schema})); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, json)); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); } TEST_F(ParquetReaderTest, ReadWithNameMapping) { @@ -645,6 +698,7 @@ TEST_F(ParquetReaderTest, ReadNestedFieldsWithNameMapping) { [[null], null, null] ])"; ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds(type, json)); + EXPECT_THAT(ParquetColumnPaths(), ::testing::Contains("numbers.list.item.value")); auto schema = std::make_shared(std::vector{ SchemaField::MakeOptional(1, "profile", @@ -686,6 +740,69 @@ TEST_F(ParquetReaderTest, ReadNestedFieldsWithNameMapping) { } } +TEST_F(ParquetReaderTest, ReadListWithUnmappedElementAsMissing) { + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( + ::arrow::struct_({ + ::arrow::field("id", ::arrow::int32()), + ::arrow::field("tags", ::arrow::list(::arrow::utf8())), + ::arrow::field("counts", ::arrow::map(::arrow::utf8(), ::arrow::int32())), + }), + R"([[1, ["a"], [["x", 1]]], [2, null, null]])")); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "id", int32()), + SchemaField::MakeOptional(2, "tags", + std::make_shared( + SchemaField::MakeOptional(201, "element", string()))), + SchemaField::MakeOptional( + 3, "counts", + std::make_shared(SchemaField::MakeRequired(301, "key", string()), + SchemaField::MakeOptional(302, "value", int32()))), + }); + // The mapping covers the list and the map but not their children, so they are read + // as if they were missing from the file. + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"tags"}, .field_id = 2}, + MappedField{.names = {"counts"}, + .field_id = 3, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"key"}, .field_id = 301}, + })}, + }); + + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, + {.path = temp_parquet_file_, + .io = file_io_, + .projection = schema, + .name_mapping = std::move(mapping)})); + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([[1, null, null], [2, null, null]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadWithAliasMappedToTwoColumns) { + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( + ::arrow::struct_({::arrow::field("name", ::arrow::utf8()), + ::arrow::field("full_name", ::arrow::utf8())}), + R"([["a", "Alice"], ["b", "Bob"]])")); + auto schema = std::make_shared( + std::vector{SchemaField::MakeOptional(2, "full_name", string())}); + auto mapping = NameMapping::Make({ + MappedField{.names = {"name", "full_name"}, .field_id = 2}, + }); + + // Like Java, the last column mapped to the ID is read. + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, + {.path = temp_parquet_file_, + .io = file_io_, + .projection = schema, + .name_mapping = std::move(mapping)})); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([["Alice"], ["Bob"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + TEST_F(ParquetReaderTest, ReadMissingFieldsWithDefaults) { // The file contains only fields 1 and 2; the projected schema adds fields 3 and 4 // with initial-defaults, which are filled for all rows written before the columns From a471965a386ebbae7d1e43cec6b61ac841a955fa Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 10 Nov 2026 22:26:36 +0800 Subject: [PATCH 2/2] refactor: avoid continue in name mapping loops Co-Authored-By: Claude Opus 5.5 --- src/iceberg/name_mapping.cc | 11 +++++------ src/iceberg/parquet/parquet_schema_util.cc | 20 ++++++++++---------- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/src/iceberg/name_mapping.cc b/src/iceberg/name_mapping.cc index 54bb7a1bd..f439c2158 100644 --- a/src/iceberg/name_mapping.cc +++ b/src/iceberg/name_mapping.cc @@ -98,12 +98,11 @@ struct IndexByNameVisitor { MappedFields::MappedFields(std::vector fields) : fields_(std::move(fields)) { for (const auto& field : fields_) { - if (!field.field_id.has_value()) { - continue; - } - id_to_field_.emplace(field.field_id.value(), std::cref(field)); - for (const auto& name : field.names) { - name_to_id_.emplace(name, field.field_id.value()); + if (field.field_id.has_value()) { + id_to_field_.emplace(field.field_id.value(), std::cref(field)); + for (const auto& name : field.names) { + name_to_id_.emplace(name, field.field_id.value()); + } } } } diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 0d3ed173a..daff5e952 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -797,16 +797,16 @@ class NameMappingApplier { for (size_t i = 0; i < fields.size(); ++i) { auto unmapped_field = fields[i].field; auto field_id = ApplyToChild(unmapped_field->name(), &fields[i]); - if (!field_id.has_value()) { - continue; - } - auto [iter, inserted] = mapped_siblings.try_emplace( - field_id.value(), MappedSibling{.index = i, .unmapped_field = unmapped_field}); - if (!inserted) { - // Aliases can map two sibling columns to the same ID. Java's reader keeps the - // last one, so drop the ID from the earlier column. - fields[iter->second.index].field = std::move(iter->second.unmapped_field); - iter->second = MappedSibling{.index = i, .unmapped_field = unmapped_field}; + if (field_id.has_value()) { + auto [iter, inserted] = mapped_siblings.try_emplace( + field_id.value(), + MappedSibling{.index = i, .unmapped_field = unmapped_field}); + if (!inserted) { + // Aliases can map two sibling columns to the same ID. Java's reader keeps the + // last one, so drop the ID from the earlier column. + fields[iter->second.index].field = std::move(iter->second.unmapped_field); + iter->second = MappedSibling{.index = i, .unmapped_field = unmapped_field}; + } } } }