From c794884abcc54b2b8f5ffca24f9a9d01674ceb5c Mon Sep 17 00:00:00 2001 From: kiwimeng Date: Mon, 19 Oct 2026 20:19:57 +0800 Subject: [PATCH] feat(parquet): apply name mapping to files without field IDs --- src/iceberg/parquet/parquet_reader.cc | 20 ++- src/iceberg/parquet/parquet_schema_util.cc | 55 +++++++ .../parquet/parquet_schema_util_internal.h | 11 ++ src/iceberg/test/parquet_schema_test.cc | 153 ++++++++++++++++++ src/iceberg/test/parquet_test.cc | 130 +++++++++++++++ 5 files changed, 361 insertions(+), 8 deletions(-) diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 3f15ab7a55..f21090e508 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -59,19 +59,23 @@ Result> OpenInputStream( } Result BuildProjection(::parquet::arrow::FileReader* reader, - const Schema& read_schema) { + const Schema& read_schema, + const NameMapping* name_mapping) { auto metadata = reader->parquet_reader()->metadata(); - if (!HasFieldIds(metadata->schema()->schema_root())) { - // TODO(gangwu): apply name mapping to Parquet schema - return NotImplemented("Applying name mapping to Parquet schema is not implemented"); - } - ::parquet::arrow::SchemaManifest schema_manifest; ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make( metadata->schema(), metadata->key_value_metadata(), reader->properties(), &schema_manifest)); + if (!HasFieldIds(metadata->schema()->schema_root())) { + if (name_mapping == nullptr) { + return InvalidSchema( + "Parquet file schema has no field IDs and no name mapping provided"); + } + ApplyNameMapping(*name_mapping, &schema_manifest); + } + // Leverage SchemaManifest to project the schema ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, schema_manifest)); return projection; @@ -280,7 +284,6 @@ struct ReadContext { // 1. Make the memory pool configurable // 2. Catch ParquetException and convert to Status/Result // 3. Add utility to convert Arrow Status/Result to Iceberg Status/Result -// 4. Check field ids and apply name mapping if needed class ParquetReader::Impl { public: // Open the Parquet reader with the given options @@ -312,7 +315,8 @@ class ParquetReader::Impl { arrow_reader_properties)); // Project read schema onto the Parquet file schema - ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_)); + ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_, + options.name_mapping.get())); metadata_context_ = {.file_path = options.path, .first_row_id = options.first_row_id, .data_sequence_number = options.data_sequence_number}; diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 05f038c133..6a15659e86 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -26,6 +26,7 @@ #include "iceberg/constants.h" #include "iceberg/metadata_columns.h" +#include "iceberg/name_mapping.h" #include "iceberg/parquet/parquet_schema_util_internal.h" #include "iceberg/result.h" #include "iceberg/schema_internal.h" @@ -716,6 +717,52 @@ void CollectColumnIds(const FieldProjection& field_projection, } } +// Forward declaration +void ApplyNameMappingToField(const NameMapping& mapping, std::vector& names, + ::parquet::arrow::SchemaField* parquet_field); + +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 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())})); + } + + 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; + } +} + } // namespace Result Project(const Schema& expected_schema, @@ -753,6 +800,14 @@ bool HasFieldIds(const ::parquet::schema::NodePtr& node) { return false; } +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); + } +} + Result> ToParquetSchema( const Schema& iceberg_schema) { ::parquet::schema::NodeVector fields; diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h index 2eafd013b5..6612de0569 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -26,6 +26,7 @@ #include "iceberg/schema.h" #include "iceberg/schema_util.h" +#include "iceberg/type_fwd.h" namespace iceberg::parquet { @@ -63,6 +64,16 @@ std::vector SelectedColumnIndices(const SchemaProjection& projection); /// \return True if the Parquet schema has field IDs, false otherwise. 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. +/// +/// \param mapping The name mapping to apply. +/// \param parquet_schema The Parquet schema to update in place. +void ApplyNameMapping(const NameMapping& mapping, + ::parquet::arrow::SchemaManifest* parquet_schema); + /// \brief Convert an Iceberg schema to a Parquet schema. Result> ToParquetSchema( const Schema& iceberg_schema); diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index 4663b184d4..4cf36da1af 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -31,6 +31,7 @@ #include #include "iceberg/metadata_columns.h" +#include "iceberg/name_mapping.h" #include "iceberg/parquet/parquet_schema_util_internal.h" #include "iceberg/schema.h" #include "iceberg/test/matchers.h" @@ -203,6 +204,158 @@ TEST(HasFieldIdsTest, GroupNode) { } // NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks) +TEST(ParquetNameMappingTest, ApplyNameMapping) { + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/1, "id", int64()), + SchemaField::MakeOptional(/*field_id=*/2, "name", string()), + }); + + auto parquet_schema = + MakeGroupNode("iceberg_schema", {MakeInt64Node("id"), MakeStringNode("name")}); + auto schema_manifest = MakeSchemaManifest(parquet_schema); + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"name"}, .field_id = 2}, + }); + + ApplyNameMapping(*mapping, &schema_manifest); + EXPECT_EQ(schema_manifest.schema_fields[0].field->metadata()->Get( + std::string(kParquetFieldIdKey)), + "1"); + 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()); + + const auto& projection = *projection_result; + ASSERT_EQ(projection.fields.size(), 2); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_PROJECTED_FIELD(projection.fields[1], 1); + ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0, 1})); +} + +TEST(ParquetNameMappingTest, ApplyNameMappingWithAlias) { + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/2, "full_name", string()), + }); + + auto parquet_schema = MakeGroupNode("iceberg_schema", {MakeStringNode("user_name")}); + auto schema_manifest = MakeSchemaManifest(parquet_schema); + auto mapping = NameMapping::Make({ + MappedField{.names = {"full_name", "user_name"}, .field_id = 2}, + }); + + 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(), 1); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); +} + +TEST(ParquetNameMappingTest, ApplyNameMappingToNestedFields) { + Schema expected_schema({ + SchemaField::MakeOptional( + /*field_id=*/1, "profile", + std::make_shared(std::vector{ + SchemaField::MakeOptional(/*field_id=*/101, "name", string()), + })), + 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()))), + }); + + auto parquet_schema = MakeGroupNode( + "iceberg_schema", + { + MakeGroupNode("profile", {MakeStringNode("name")}), + MakeListNode("numbers", MakeInt32Node("item")), + MakeMapNode("counts", + MakeStringNode("key", /*field_id=*/-1, /*optional=*/false), + MakeInt32Node("value")), + }); + auto mapping = NameMapping::Make({ + MappedField{.names = {"profile"}, + .field_id = 1, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"name"}, .field_id = 101}, + })}, + MappedField{.names = {"numbers"}, + .field_id = 2, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"element"}, .field_id = 201}, + })}, + MappedField{.names = {"counts"}, + .field_id = 3, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"key"}, .field_id = 301}, + MappedField{.names = {"value"}, .field_id = 302}, + })}, + }); + + 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(), 3); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_EQ(projection.fields[0].children.size(), 1); + ASSERT_PROJECTED_FIELD(projection.fields[0].children[0], 0); + ASSERT_PROJECTED_FIELD(projection.fields[1], 1); + ASSERT_EQ(projection.fields[1].children.size(), 1); + ASSERT_PROJECTED_FIELD(projection.fields[1].children[0], 0); + ASSERT_PROJECTED_FIELD(projection.fields[2], 2); + ASSERT_EQ(projection.fields[2].children.size(), 2); + ASSERT_PROJECTED_FIELD(projection.fields[2].children[0], 0); + ASSERT_PROJECTED_FIELD(projection.fields[2].children[1], 1); + ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0, 1, 2, 3})); +} + +TEST(ParquetNameMappingTest, ApplyNameMappingWithUnmappedFields) { + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/1, "id", int64()), + SchemaField::MakeOptional(/*field_id=*/2, "extra", string()), + SchemaField::MakeOptional(/*field_id=*/3, "unassigned", string()), + }); + + auto parquet_schema = MakeGroupNode( + "iceberg_schema", + {MakeInt64Node("id"), MakeStringNode("extra"), MakeStringNode("unassigned")}); + auto schema_manifest = MakeSchemaManifest(parquet_schema); + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"unassigned"}, .field_id = std::nullopt}, + }); + + ApplyNameMapping(*mapping, &schema_manifest); + for (size_t i = 1; i < schema_manifest.schema_fields.size(); ++i) { + const auto& metadata = schema_manifest.schema_fields[i].field->metadata(); + EXPECT_TRUE(metadata == nullptr || + metadata->FindKey(std::string(kParquetFieldIdKey)) < 0); + } + + auto projection_result = Project(expected_schema, schema_manifest); + ASSERT_THAT(projection_result, IsOk()); + + const auto& projection = *projection_result; + ASSERT_EQ(projection.fields.size(), 3); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_PROJECTED_NULL_FIELD(projection.fields[1]); + ASSERT_PROJECTED_NULL_FIELD(projection.fields[2]); + ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0})); +} + 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 6dedba443e..91c4579910 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -47,6 +47,7 @@ #include "iceberg/file_reader.h" #include "iceberg/file_writer.h" #include "iceberg/metadata_columns.h" +#include "iceberg/name_mapping.h" #include "iceberg/parquet/parquet_register.h" #include "iceberg/result.h" #include "iceberg/schema.h" @@ -201,6 +202,20 @@ class ParquetReaderTest : public TempFileTestBase { .properties = std::move(writer_properties)})); } + void CreateParquetFileWithoutFieldIds(const std::shared_ptr<::arrow::DataType>& type, + std::string_view json) { + auto array = ::arrow::json::ArrayFromJSONString(type, json).ValueOrDie(); + auto batch = ::arrow::RecordBatch::FromStructArray(array).ValueOrDie(); + auto table = ::arrow::Table::FromRecordBatches({batch}).ValueOrDie(); + 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()) + .ok()); + ASSERT_TRUE(outfile->Close().ok()); + } + void CreateListParquetFile() { auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "id", int32()), @@ -443,6 +458,121 @@ TEST_F(ParquetReaderTest, ReadTwoFields) { ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); } +TEST_F(ParquetReaderTest, ReadWithoutFieldIdsOrNameMapping) { + 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())}); + + 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")); +} + +TEST_F(ParquetReaderTest, ReadWithNameMapping) { + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( + ::arrow::struct_({::arrow::field("id", ::arrow::int32()), + ::arrow::field("name", ::arrow::utf8())}), + R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])")); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(2, "full_name", string()), + SchemaField::MakeOptional(1, "id", int64()), + SchemaField::MakeOptional(3, "extra", string()), + }); + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"name", "full_name"}, .field_id = 2}, + }); + + 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"([["Foo", 1, null], ["Bar", 2, null], ["Baz", 3, null]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadWithFieldIdsAndNameMapping) { + CreateSimpleParquetFile(); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "name", string()), + }); + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 2}, + MappedField{.names = {"name"}, .field_id = 1}, + }); + + 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, "Foo"], [2, "Bar"], [3, "Baz"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadNestedFieldsWithNameMapping) { + auto value_type = ::arrow::struct_({::arrow::field("value", ::arrow::int32())}); + auto type = ::arrow::struct_({ + ::arrow::field("profile", value_type), + ::arrow::field("numbers", ::arrow::list(::arrow::field("item", value_type))), + ::arrow::field("counts", ::arrow::map(::arrow::utf8(), value_type)), + }); + const std::string json = R"([ + [[1], [[2], null], [["a", [3]], ["b", null]]], + [null, [], []], + [[null], null, null] + ])"; + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds(type, json)); + + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "profile", + std::make_shared(std::vector{ + SchemaField::MakeOptional(101, "value", int64()), + })), + SchemaField::MakeOptional(2, "numbers", + std::make_shared(SchemaField::MakeOptional( + 201, "element", + std::make_shared(std::vector{ + SchemaField::MakeOptional(202, "value", int64()), + })))), + SchemaField::MakeOptional( + 3, "counts", + std::make_shared( + SchemaField::MakeRequired(301, "key", string()), + SchemaField::MakeOptional( + 302, "value", + std::make_shared(std::vector{ + SchemaField::MakeOptional(303, "value", int64()), + })))), + }); + ICEBERG_UNWRAP_OR_FAIL(auto unique_mapping, CreateMapping(*schema)); + std::shared_ptr mapping = std::move(unique_mapping); + + for (bool use_large_list : {false, true}) { + SCOPED_TRACE(use_large_list); + ReaderProperties properties; + properties.Set(ReaderProperties::kArrowUseLargeList, use_large_list); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, + {.path = temp_parquet_file_, + .io = file_io_, + .projection = schema, + .name_mapping = mapping, + .properties = std::move(properties)})); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, json)); + 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