Skip to content

feat(parquet): apply name mapping to files without field IDs #1044

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fairytale0828 wants to merge 1 commit into
base: main
Choose a base branch
from
+361 −8
Open

feat(parquet): apply name mapping to files without field IDs #1044

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions src/iceberg/parquet/parquet_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,19 +59,23 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenInputStream(
}

Result<SchemaProjection> 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(

@shangxinli shangxinli Oct 20, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java does not error here - ReadConf.java:85-88 falls back to addFallbackIds and assigns positional ids, so this file still reads there. Not a regression, but is closing that gap out of scope?

@fairytale0828 fairytale0828 Oct 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. I'd like to keep this PR scoped to name mapping. The no-mapping case currently matches the C++ Avro reader, which also returns InvalidSchema. I've opened #1046 to add the positional fallback to match Java. Does that work for you?

"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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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};
Expand Down
55 changes: 55 additions & 0 deletions src/iceberg/parquet/parquet_schema_util.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -716,6 +717,52 @@ void CollectColumnIds(const FieldProjection& field_projection,
}
}

// Forward declaration
void ApplyNameMappingToField(const NameMapping& mapping, std::vector<std::string>& names,
::parquet::arrow::SchemaField* parquet_field);

void ApplyNameMappingToChild(const NameMapping& mapping, std::vector<std::string>& 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<std::string>& 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<SchemaProjection> Project(const Schema& expected_schema,
Expand Down Expand Up @@ -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<std::string> names;
for (auto& parquet_field : parquet_schema->schema_fields) {
ApplyNameMappingToChild(mapping, names, parquet_field.field->name(), &parquet_field);
}
}

Result<std::shared_ptr<::parquet::SchemaDescriptor>> ToParquetSchema(
const Schema& iceberg_schema) {
::parquet::schema::NodeVector fields;
Expand Down
11 changes: 11 additions & 0 deletions src/iceberg/parquet/parquet_schema_util_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@

#include "iceberg/schema.h"
#include "iceberg/schema_util.h"
#include "iceberg/type_fwd.h"

namespace iceberg::parquet {

Expand Down Expand Up @@ -63,6 +64,16 @@ std::vector<int32_t> 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<std::shared_ptr<::parquet::SchemaDescriptor>> ToParquetSchema(
const Schema& iceberg_schema);
Expand Down
153 changes: 153 additions & 0 deletions src/iceberg/test/parquet_schema_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include <parquet/types.h>

#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"
Expand Down Expand Up @@ -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<int32_t>({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<StructType>(std::vector<SchemaField>{
SchemaField::MakeOptional(/*field_id=*/101, "name", string()),
})),
SchemaField::MakeOptional(
/*field_id=*/2, "numbers",
std::make_shared<ListType>(SchemaField::MakeOptional(
/*field_id=*/201, "element", int32()))),
SchemaField::MakeOptional(
/*field_id=*/3, "counts",
std::make_shared<MapType>(
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<int32_t>({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<int32_t>({0}));
}

TEST(ParquetSchemaProjectionTest, ProjectIdenticalSchemas) {
Schema expected_schema({
SchemaField::MakeRequired(/*field_id=*/1, "id", iceberg::int64()),
Expand Down
Loading
Loading