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..f439c2158 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,30 @@ 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()) { + 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 +129,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 +156,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 +170,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..daff5e952 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()) { + 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