Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
3 changes: 2 additions & 1 deletion src/iceberg/file_reader.h
Original file line number Diff line number Diff line change
Expand Up @@ -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<class NameMapping> name_mapping;
/// \brief First row ID inherited by the data file, if assigned.
std::optional<int64_t> first_row_id;
Expand Down
96 changes: 31 additions & 65 deletions src/iceberg/name_mapping.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
#include "iceberg/name_mapping.h"

#include <format>
#include <sstream>

#include "iceberg/util/formatter_internal.h"
#include "iceberg/util/macros.h"
Expand All @@ -32,14 +31,14 @@ namespace {

// Helper function to join a list of field names with a dot
std::string JoinByDot(std::span<const std::string> 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
Expand Down Expand Up @@ -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<std::string> 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));
}
}
}
Expand All @@ -97,24 +96,30 @@ struct IndexByNameVisitor {

} // namespace

MappedFields::MappedFields(std::vector<MappedField> fields)
: fields_(std::move(fields)) {}
MappedFields::MappedFields(std::vector<MappedField> 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> MappedFields::Make(std::vector<MappedField> fields) {
return std::unique_ptr<MappedFields>(new MappedFields(std::move(fields)));
}

std::optional<MappedFieldConstRef> 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<int32_t> 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;
Expand All @@ -124,37 +129,19 @@ size_t MappedFields::Size() const { return fields_.size(); }

std::span<const MappedField> MappedFields::fields() const { return fields_; }

const std::unordered_map<std::string_view, int32_t>& 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<int32_t, MappedFieldConstRef>& 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<MappedFields> 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<MappedFieldConstRef> 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;
Expand All @@ -169,8 +156,7 @@ std::optional<MappedFieldConstRef> NameMapping::Find(
}

std::optional<MappedFieldConstRef> 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;
Expand All @@ -184,26 +170,6 @@ const MappedFields& NameMapping::AsMappedFields() const {
return *mapping_;
}

const std::unordered_map<int32_t, MappedFieldConstRef>& 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<std::string, MappedFieldConstRef>&
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> NameMapping::MakeEmpty() {
return std::unique_ptr<NameMapping>(new NameMapping(MappedFields::Make({})));
}
Expand Down
26 changes: 14 additions & 12 deletions src/iceberg/name_mapping.h
Original file line number Diff line number Diff line change
Expand Up @@ -56,13 +56,20 @@ struct ICEBERG_EXPORT MappedField {
using MappedFieldConstRef = std::reference_wrapper<const MappedField>;

/// \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.
/// \param[in] fields The list of field mappings.
/// \return A new MappedFields instance.
static std::unique_ptr<MappedFields> Make(std::vector<MappedField> 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.
Expand All @@ -84,18 +91,17 @@ class ICEBERG_EXPORT MappedFields {
private:
explicit MappedFields(std::vector<MappedField> fields);

const std::unordered_map<std::string_view, int32_t>& LazyNameToId() const;
const std::unordered_map<int32_t, MappedFieldConstRef>& LazyIdToField() const;

private:
std::vector<MappedField> fields_;

// Lazy-initialized mappings
mutable std::unordered_map<std::string_view, int32_t> name_to_id_;
mutable std::unordered_map<int32_t, MappedFieldConstRef> id_to_field_;
std::unordered_map<std::string_view, int32_t> name_to_id_;
std::unordered_map<int32_t, MappedFieldConstRef> 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.
Expand Down Expand Up @@ -124,15 +130,11 @@ class ICEBERG_EXPORT NameMapping {
private:
explicit NameMapping(std::unique_ptr<MappedFields> mapping);

const std::unordered_map<int32_t, MappedFieldConstRef>& LazyFieldsById() const;
const std::unordered_map<std::string, MappedFieldConstRef>& LazyFieldsByName() const;

private:
std::unique_ptr<MappedFields> mapping_;

// Lazy-initialized mappings
mutable std::unordered_map<int32_t, MappedFieldConstRef> fields_by_id_;
mutable std::unordered_map<std::string, MappedFieldConstRef> fields_by_name_;
std::unordered_map<int32_t, MappedFieldConstRef> fields_by_id_;
std::unordered_map<std::string, MappedFieldConstRef> fields_by_name_;
};

ICEBERG_EXPORT std::string ToString(const MappedField& field);
Expand Down
8 changes: 6 additions & 2 deletions src/iceberg/parquet/parquet_reader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -68,9 +69,12 @@ Result<SchemaProjection> 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<NameMapping> 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(
Expand Down
Loading