diff --git a/src/iceberg/avro/avro_direct_decoder.cc b/src/iceberg/avro/avro_direct_decoder.cc index 465bc4b24..e3d757a56 100644 --- a/src/iceberg/avro/avro_direct_decoder.cc +++ b/src/iceberg/avro/avro_direct_decoder.cc @@ -623,6 +623,10 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& d const auto& branch_node = avro_node->leafAt(branch_index); if (branch_node->type() == ::avro::AVRO_NULL) { + // A no-op for the binary decoder, but the resolving decoder used when the reader + // schema differs from the file's (for example after applying a name mapping) + // expects the null to be consumed. + decoder.decodeNull(); if (is_row_lineage) { return arrow::AppendInheritedRowLineageValue(projected_field.field_id(), metadata_context, array_builder); diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc index f333a38cb..032642367 100644 --- a/src/iceberg/avro/avro_reader.cc +++ b/src/iceberg/avro/avro_reader.cc @@ -312,18 +312,19 @@ class AvroReader::Impl { ICEBERG_RETURN_UNEXPECTED(has_id_visitor.Visit(file_schema)); if (has_id_visitor.HasNoIds()) { - // Apply field IDs based on name mapping if available - if (options.name_mapping) { - ICEBERG_ASSIGN_OR_RAISE( - auto new_root_node, - MakeAvroNodeWithFieldIds(file_schema.root(), *options.name_mapping)); - - // Update the file schema to use the new schema with field IDs - file_schema = ::avro::ValidSchema(new_root_node); - } else { - return InvalidSchema( - "Avro file schema has no field IDs and no name mapping provided"); + // Apply field IDs from the name mapping. Like Java's ProjectionDatumReader, fall + // back to the read schema's names when the table has no name mapping. + const NameMapping* name_mapping = options.name_mapping.get(); + std::unique_ptr fallback_mapping; + if (name_mapping == nullptr) { + ICEBERG_ASSIGN_OR_RAISE(fallback_mapping, CreateMapping(*read_schema_)); + name_mapping = fallback_mapping.get(); } + ICEBERG_ASSIGN_OR_RAISE(auto new_root_node, MakeAvroNodeWithFieldIds( + file_schema.root(), *name_mapping)); + + // Update the file schema to use the new schema with field IDs + file_schema = ::avro::ValidSchema(new_root_node); } else if (!has_id_visitor.AllHaveIds()) { return InvalidSchema("Not all fields in the Avro file schema have field IDs"); } diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 94d0a6116..feb10c956 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -538,6 +538,20 @@ Result GetFieldId(const ::avro::NodePtr& node, size_t field_idx) { return GetId(node, kFieldIdKey, field_idx); } +// Like GetId, but returns nullopt when the attribute is absent, for example on a field +// that a name mapping does not cover. +Result> FindId(const ::avro::NodePtr& node, + const std::string& attr_name, size_t field_idx) { + if (field_idx >= node->customAttributes()) { + return std::nullopt; + } + auto id_str = node->customAttributesAt(field_idx).getAttribute(attr_name); + if (!id_str.has_value()) { + return std::nullopt; + } + return StringUtils::ParseNumber(id_str.value()); +} + Status ValidateAvroSchemaEvolution(const Type& expected_type, const ::avro::NodePtr& avro_node) { if (avro_node->type() == ::avro::AVRO_NULL) { @@ -683,6 +697,69 @@ Status UnwrapUnion(const ::avro::NodePtr& node, ::avro::NodePtr* result) { return {}; } +// Lists and maps are matched to the file by the IDs of their element, key and value. +// Returns false when the Avro node leaves any of them without an ID, for example when a +// 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 node of the wrong shape returns true +// so that projection reports the mismatch. +Result HasNestedIds(const Type& expected_type, const ::avro::NodePtr& avro_node) { + static const std::string kFieldIdKey{kFieldIdProp}; + static const std::string kElementIdKey{kElementIdProp}; + static const std::string kKeyIdKey{kKeyIdProp}; + static const std::string kValueIdKey{kValueIdProp}; + if (expected_type.type_id() != TypeId::kList && + expected_type.type_id() != TypeId::kMap) { + return true; + } + + ::avro::NodePtr node; + ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node, &node)); + + if (expected_type.type_id() == TypeId::kList) { + if (node->type() != ::avro::AVRO_ARRAY || HasMapLogicalType(node) || + node->leaves() != 1) { + return true; + } + ICEBERG_ASSIGN_OR_RAISE(auto element_id, FindId(node, kElementIdKey, 0)); + if (!element_id.has_value()) { + return false; + } + const auto& list_type = internal::checked_cast(expected_type); + return HasNestedIds(*list_type.element().type(), node->leafAt(0)); + } + + const auto& map_type = internal::checked_cast(expected_type); + std::optional key_id; + std::optional value_id; + ::avro::NodePtr key_node; + ::avro::NodePtr value_node; + if (node->type() == ::avro::AVRO_MAP && node->leaves() == 2) { + ICEBERG_ASSIGN_OR_RAISE(key_id, FindId(node, kKeyIdKey, 0)); + ICEBERG_ASSIGN_OR_RAISE(value_id, FindId(node, kValueIdKey, 0)); + key_node = node->leafAt(0); + value_node = node->leafAt(1); + } else if (node->type() == ::avro::AVRO_ARRAY && HasMapLogicalType(node) && + node->leaves() == 1 && node->leafAt(0)->type() == ::avro::AVRO_RECORD && + node->leafAt(0)->leaves() == 2) { + const auto& entry_node = node->leafAt(0); + ICEBERG_ASSIGN_OR_RAISE(key_id, FindId(entry_node, kFieldIdKey, 0)); + ICEBERG_ASSIGN_OR_RAISE(value_id, FindId(entry_node, kFieldIdKey, 1)); + key_node = entry_node->leafAt(0); + value_node = entry_node->leafAt(1); + } else { + return true; + } + if (!key_id.has_value() || !value_id.has_value()) { + return false; + } + ICEBERG_ASSIGN_OR_RAISE(auto key_has_ids, + HasNestedIds(*map_type.key().type(), key_node)); + if (!key_has_ids) { + return false; + } + return HasNestedIds(*map_type.value().type(), value_node); +} + // Forward declaration Result ProjectNested(const Type& expected_type, const ::avro::NodePtr& avro_node, @@ -733,14 +810,19 @@ Result ProjectStruct(const StructType& struct_type, std::unordered_map node_info_map; node_info_map.reserve(avro_node->leaves()); + static const std::string kFieldIdKey{kFieldIdProp}; for (size_t i = 0; i < avro_node->leaves(); ++i) { - ICEBERG_ASSIGN_OR_RAISE(int32_t field_id, GetFieldId(avro_node, i)); - ::avro::NodePtr field_node = avro_node->leafAt(i); - if (const auto [iter, inserted] = node_info_map.emplace( - std::piecewise_construct, std::forward_as_tuple(field_id), - std::forward_as_tuple(i, field_node)); - !inserted) [[unlikely]] { - return InvalidSchema("Duplicate field id found in Avro schema: {}", field_id); + // Like Java's PruneColumns, a field without an ID (one that a name mapping does not + // cover) is not read. + ICEBERG_ASSIGN_OR_RAISE(auto field_id, FindId(avro_node, kFieldIdKey, i)); + if (field_id.has_value()) { + ::avro::NodePtr field_node = avro_node->leafAt(i); + if (const auto [iter, inserted] = node_info_map.emplace( + std::piecewise_construct, std::forward_as_tuple(*field_id), + std::forward_as_tuple(i, field_node)); + !inserted) [[unlikely]] { + return InvalidSchema("Duplicate field id found in Avro schema: {}", *field_id); + } } } @@ -751,7 +833,16 @@ Result ProjectStruct(const StructType& struct_type, int32_t field_id = expected_field.field_id(); FieldProjection child_projection; - if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) { + auto iter = node_info_map.find(field_id); + if (iter != node_info_map.cend()) { + ICEBERG_ASSIGN_OR_RAISE(auto has_nested_ids, HasNestedIds(*expected_field.type(), + iter->second.field_node)); + if (!has_nested_ids) { + iter = node_info_map.end(); + } + } + + if (iter != node_info_map.cend()) { ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(expected_field, iter->second.field_node, iter->second.local_index, prune_source)); @@ -926,38 +1017,35 @@ void CopyCustomAttributes(const ::avro::CustomAttributes& source, } } -Result FindMappedField(const NameMapping& mapping, - const std::vector& names) { - auto field_opt = mapping.Find(names); - if (!field_opt.has_value()) { - return InvalidSchema("Field '{}' not found in name mapping", names.back()); - } - - const MappedField& field = field_opt.value().get(); - if (!field.field_id.has_value()) { - return InvalidSchema("Field ID is missing for field '{}' in name mapping", - names.back()); +// Returns the ID the name mapping assigns to the field at `names`. Like Java's +// ApplyNameMapping, a field the mapping does not cover, or maps without an ID, is kept +// without an ID, so projection skips it. +std::optional FindMappedFieldId(const NameMapping& mapping, + const std::vector& names) { + auto field = mapping.Find(names); + if (!field.has_value()) { + return std::nullopt; } - - return field_opt.value(); + return field->get().field_id; } template Status ProcessChildNode(const ::avro::NodePtr& parent_node, size_t child_index, - const std::string& child_name, std::string_view fieldIdPropKy, + const std::string& child_name, std::string_view field_id_prop, const NameMapping& mapping, std::vector& names, NodeType& new_parent_node) { NamesGuard guard(names, child_name); - ICEBERG_ASSIGN_OR_RAISE(auto mapped_field, FindMappedField(mapping, names)); + auto field_id = FindMappedFieldId(mapping, names); ICEBERG_ASSIGN_OR_RAISE( auto new_child_node, MakeAvroNodeWithFieldIds(parent_node->leafAt(child_index), mapping, names)); ::avro::CustomAttributes attributes; - attributes.addAttribute(std::string(fieldIdPropKy), - std::to_string(*mapped_field.get().field_id), - /*addQuote=*/false); + if (field_id.has_value()) { + attributes.addAttribute(std::string(field_id_prop), std::to_string(*field_id), + /*addQuote=*/false); + } if (parent_node->customAttributes() > child_index) { CopyCustomAttributes(parent_node->customAttributesAt(child_index), attributes); } @@ -998,6 +1086,7 @@ Result<::avro::NodePtr> MakeArrayNodeWithFieldIds(const ::avro::NodePtr& origina } auto new_array_node = std::make_shared<::avro::NodeArray>(); + new_array_node->setLogicalType(original_node->logicalType()); // Check if this is a map represented as array if (HasMapLogicalType(original_node)) { @@ -1013,8 +1102,9 @@ Result<::avro::NodePtr> MakeArrayNodeWithFieldIds(const ::avro::NodePtr& origina return new_array_node; } - ICEBERG_RETURN_UNEXPECTED(ProcessChildNode(original_node, 0, "element", kElementIdProp, - mapping, names, new_array_node)); + ICEBERG_RETURN_UNEXPECTED( + ProcessChildNode(original_node, 0, std::string(ListType::kElementName), + kElementIdProp, mapping, names, new_array_node)); return new_array_node; } @@ -1026,11 +1116,32 @@ Result<::avro::NodePtr> MakeMapNodeWithFieldIds(const ::avro::NodePtr& original_ return InvalidSchema("Map type must have exactly two leaves"); } + // Like ToAvroNodeVisitor, keep the implicit string key that NodeMap creates and put + // both IDs in the map's single set of custom attributes. + ::avro::CustomAttributes attributes; + if (original_node->customAttributes() > 0) { + CopyCustomAttributes(original_node->customAttributesAt(0), attributes); + } + { + NamesGuard guard(names, std::string(MapType::kKeyName)); + if (auto key_id = FindMappedFieldId(mapping, names); key_id.has_value()) { + attributes.addAttribute(std::string(kKeyIdProp), std::to_string(*key_id), + /*addQuote=*/false); + } + } + + NamesGuard guard(names, std::string(MapType::kValueName)); + if (auto value_id = FindMappedFieldId(mapping, names); value_id.has_value()) { + attributes.addAttribute(std::string(kValueIdProp), std::to_string(*value_id), + /*addQuote=*/false); + } + ICEBERG_ASSIGN_OR_RAISE( + auto new_value_node, + MakeAvroNodeWithFieldIds(original_node->leafAt(1), mapping, names)); + auto new_map_node = std::make_shared<::avro::NodeMap>(); - ICEBERG_RETURN_UNEXPECTED(ProcessChildNode(original_node, 0, "key", kKeyIdProp, mapping, - names, new_map_node)); - ICEBERG_RETURN_UNEXPECTED(ProcessChildNode(original_node, 1, "value", kValueIdProp, - mapping, names, new_map_node)); + new_map_node->addLeaf(new_value_node); + new_map_node->addCustomAttributesForField(attributes); return new_map_node; } diff --git a/src/iceberg/avro/avro_schema_util_internal.h b/src/iceberg/avro/avro_schema_util_internal.h index 342b119a5..c9d43116d 100644 --- a/src/iceberg/avro/avro_schema_util_internal.h +++ b/src/iceberg/avro/avro_schema_util_internal.h @@ -101,9 +101,10 @@ class HasIdVisitor { return total_fields_ == fields_with_id_ && fields_with_id_ != 0; } - /// \brief Check if all fields in the visited schema have field IDs. - /// \return True if all fields have IDs, false otherwise. - bool HasNoIds() const { return total_fields_ == 0; } + /// \brief Check if no field in the visited schema has a field ID. + /// \return True if no field has an ID (including when there are no fields), false + /// otherwise. + bool HasNoIds() const { return fields_with_id_ == 0; } private: /// \brief Visit a record node to check for field IDs. diff --git a/src/iceberg/test/avro_schema_test.cc b/src/iceberg/test/avro_schema_test.cc index f147a607b..7dfee96f7 100644 --- a/src/iceberg/test/avro_schema_test.cc +++ b/src/iceberg/test/avro_schema_test.cc @@ -615,7 +615,23 @@ TEST(HasIdVisitorTest, ArrayWithoutElementId) { HasIdVisitor visitor; EXPECT_THAT(visitor.Visit(::avro::compileJsonSchemaFromString(schema_json)), IsOk()); - EXPECT_FALSE(visitor.HasNoIds()); + EXPECT_TRUE(visitor.HasNoIds()); + EXPECT_FALSE(visitor.AllHaveIds()); +} + +TEST(HasIdVisitorTest, RecordWithoutFieldIds) { + const std::string schema_json = R"({ + "type": "record", + "name": "test_record", + "fields": [ + {"name": "int_field", "type": "int"}, + {"name": "list_field", "type": {"type": "array", "items": "string"}} + ] + })"; + + HasIdVisitor visitor; + EXPECT_THAT(visitor.Visit(::avro::compileJsonSchemaFromString(schema_json)), IsOk()); + EXPECT_TRUE(visitor.HasNoIds()); EXPECT_FALSE(visitor.AllHaveIds()); } @@ -1813,17 +1829,86 @@ TEST_F(NameMappingAvroSchemaTest, ApplyNameMappingToMap) { ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(new_node, /*index=*/0, /*field_id=*/1)); ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(new_node, /*index=*/1, /*field_id=*/2)); - // Check map field structure and key-value field IDs + // Check map field structure and key-value field IDs. Like maps written by Iceberg, + // both IDs are in the map's single set of custom attributes. const auto& map_node = new_node->leafAt(1); EXPECT_EQ(map_node->type(), ::avro::AVRO_MAP); - ASSERT_GE(map_node->leaves(), 2); + ASSERT_EQ(map_node->leaves(), 2); EXPECT_EQ(map_node->leafAt(0)->type(), ::avro::AVRO_STRING); EXPECT_EQ(map_node->leafAt(1)->type(), ::avro::AVRO_STRING); - ASSERT_EQ(map_node->customAttributes(), 2); + ASSERT_EQ(map_node->customAttributes(), 1); ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(map_node, /*index=*/0, /*field_id=*/30, /*key=*/"key-id")); - ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(map_node, /*index=*/1, /*field_id=*/31, + ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(map_node, /*index=*/0, /*field_id=*/31, /*key=*/"value-id")); + + // The mapped schema can be projected. + Schema expected_schema({ + SchemaField::MakeRequired(/*field_id=*/1, "id", int32()), + SchemaField::MakeRequired( + /*field_id=*/2, "properties", + std::make_shared( + SchemaField::MakeRequired(/*field_id=*/30, "key", string()), + SchemaField::MakeRequired(/*field_id=*/31, "value", string()))), + }); + auto projection_result = Project(expected_schema, new_node, /*prune_source=*/false); + ASSERT_THAT(projection_result, IsOk()); + ASSERT_EQ(projection_result->fields.size(), 2); + ASSERT_EQ(projection_result->fields[1].kind, FieldProjection::Kind::kProjected); + ASSERT_EQ(std::get<1>(projection_result->fields[1].from), 1); + ASSERT_EQ(projection_result->fields[1].children.size(), 2); +} + +TEST_F(NameMappingAvroSchemaTest, ApplyNameMappingToArrayBackedMap) { + ::iceberg::avro::RegisterLogicalTypes(); + std::string avro_schema_json = R"({ + "type": "record", + "name": "test_record", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "counts", "type": { + "type": "array", + "items": { + "type": "record", + "name": "key_value", + "fields": [ + {"name": "key", "type": "int"}, + {"name": "value", "type": "long"} + ] + }, + "logicalType": "map" + }} + ] + })"; + auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json); + auto name_mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"counts"}, + .field_id = 2, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"key"}, .field_id = 20}, + MappedField{.names = {"value"}, .field_id = 21}, + })}, + }); + + ICEBERG_UNWRAP_OR_FAIL(auto new_node, + MakeAvroNodeWithFieldIds(avro_schema.root(), *name_mapping)); + const auto& array_node = new_node->leafAt(1); + EXPECT_TRUE(HasMapLogicalType(array_node)); + ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(array_node->leafAt(0), /*index=*/0, 20)); + ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(array_node->leafAt(0), /*index=*/1, 21)); + + Schema expected_schema({ + SchemaField::MakeRequired( + /*field_id=*/2, "counts", + std::make_shared( + SchemaField::MakeRequired(/*field_id=*/20, "key", int32()), + SchemaField::MakeRequired(/*field_id=*/21, "value", int64()))), + }); + auto projection_result = Project(expected_schema, new_node, /*prune_source=*/false); + ASSERT_THAT(projection_result, IsOk()); + ASSERT_EQ(projection_result->fields[0].kind, FieldProjection::Kind::kProjected); + ASSERT_EQ(std::get<1>(projection_result->fields[0].from), 1); } TEST_F(NameMappingAvroSchemaTest, ApplyNameMappingToComplexMap) { @@ -1921,25 +2006,65 @@ TEST_F(NameMappingAvroSchemaTest, ApplyNameMappingToUnion) { EXPECT_EQ(non_null_branch->type(), ::avro::AVRO_STRING); } -TEST_F(NameMappingAvroSchemaTest, MissingFieldIdError) { - // Create a name mapping with missing field ID +TEST_F(NameMappingAvroSchemaTest, UnmappedFieldsHaveNoFieldIds) { + // "name" is mapped without an ID, and "extra" and the list element are not mapped. std::vector fields; fields.emplace_back(MappedField{.names = {"id"}, .field_id = 1}); - fields.emplace_back(MappedField{.names = {"name"}}); // Missing field_id + fields.emplace_back(MappedField{.names = {"name"}}); + fields.emplace_back(MappedField{.names = {"tags"}, .field_id = 3}); auto name_mapping = NameMapping::Make(std::move(fields)); - // Create a simple Avro record schema std::string avro_schema_json = R"({ "type": "record", "name": "test_record", "fields": [ {"name": "id", "type": "int"}, - {"name": "name", "type": "string"} + {"name": "name", "type": "string"}, + {"name": "extra", "type": "long"}, + {"name": "tags", "type": {"type": "array", "items": "string"}} ] })"; auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json); - auto result = MakeAvroNodeWithFieldIds(avro_schema.root(), *name_mapping); - ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); + ICEBERG_UNWRAP_OR_FAIL(auto new_node, + MakeAvroNodeWithFieldIds(avro_schema.root(), *name_mapping)); + ASSERT_EQ(new_node->leaves(), 4); + ASSERT_EQ(new_node->customAttributes(), 4); + ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(new_node, /*index=*/0, /*field_id=*/1)); + EXPECT_FALSE(new_node->customAttributesAt(1).getAttribute("field-id").has_value()); + EXPECT_FALSE(new_node->customAttributesAt(2).getAttribute("field-id").has_value()); + ASSERT_NO_FATAL_FAILURE(CheckFieldIdAt(new_node, /*index=*/3, /*field_id=*/3)); + const auto& list_node = new_node->leafAt(3); + ASSERT_EQ(list_node->customAttributes(), 1); + EXPECT_FALSE(list_node->customAttributesAt(0).getAttribute("element-id").has_value()); + + // Fields without IDs are not read, and a list whose element has no ID is read as + // missing. + Schema expected_schema({ + SchemaField::MakeRequired(/*field_id=*/1, "id", int32()), + SchemaField::MakeOptional(/*field_id=*/2, "name", string()), + SchemaField::MakeOptional( + /*field_id=*/3, "tags", + std::make_shared( + SchemaField::MakeOptional(/*field_id=*/30, "element", string()))), + }); + auto projection_result = Project(expected_schema, new_node, /*prune_source=*/false); + ASSERT_THAT(projection_result, IsOk()); + const auto& projection = *projection_result; + ASSERT_EQ(projection.fields.size(), 3); + ASSERT_EQ(projection.fields[0].kind, FieldProjection::Kind::kProjected); + ASSERT_EQ(std::get<1>(projection.fields[0].from), 0); + ASSERT_EQ(projection.fields[1].kind, FieldProjection::Kind::kNull); + ASSERT_EQ(projection.fields[2].kind, FieldProjection::Kind::kNull); + + // A required list that cannot be read is still an error. + Schema required_schema({ + SchemaField::MakeRequired( + /*field_id=*/3, "tags", + std::make_shared( + SchemaField::MakeOptional(/*field_id=*/30, "element", string()))), + }); + EXPECT_THAT(Project(required_schema, new_node, /*prune_source=*/false), + HasErrorMessage("Missing required field with ID: 3")); } } // namespace iceberg::avro diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index a16116410..faf2bcbd2 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -36,8 +36,11 @@ #include #include #include +#include #include +#include #include +#include #include #include "iceberg/arrow/arrow_io_internal.h" @@ -48,6 +51,7 @@ #include "iceberg/expression/literal.h" #include "iceberg/file_reader.h" #include "iceberg/metadata_columns.h" +#include "iceberg/name_mapping.h" #include "iceberg/schema.h" #include "iceberg/schema_internal.h" #include "iceberg/test/matchers.h" @@ -220,6 +224,41 @@ class AvroReaderTest : public TempFileTestBase { .properties = std::move(properties)}); } + // Writes an Avro file whose schema has no Iceberg field IDs, as a non-Iceberg writer + // would. `json_records` use Avro's JSON encoding. + void CreateAvroFileWithoutFieldIds(const std::string& schema_json, + const std::vector& json_records) { + auto avro_schema = ::avro::compileJsonSchemaFromString(schema_json); + auto& io = internal::checked_cast(*file_io_); + auto output = io.fs()->OpenOutputStream(temp_avro_file_).ValueOrDie(); + ::avro::DataFileWriter<::avro::GenericDatum> writer( + std::make_unique(output, /*buffer_size=*/1024), avro_schema); + for (const auto& json : json_records) { + auto input = ::avro::memoryInputStream( + reinterpret_cast(json.data()), json.size()); + auto decoder = ::avro::jsonDecoder(avro_schema); + decoder->init(*input); + ::avro::GenericDatum datum(avro_schema); + ::avro::decode(*decoder, datum); + writer.write(datum); + } + writer.close(); + ASSERT_TRUE(output->Close().ok()); + } + + Result> OpenReader( + const std::shared_ptr& schema, + std::shared_ptr name_mapping = nullptr) { + ReaderProperties properties; + properties.Set(ReaderProperties::kAvroSkipDatum, skip_datum_); + return ReaderFactoryRegistry::Open(FileFormatType::kAvro, + {.path = temp_avro_file_, + .io = file_io_, + .projection = schema, + .name_mapping = std::move(name_mapping), + .properties = std::move(properties)}); + } + void VerifyNextBatch(Reader& reader, std::string_view expected_json) { // Boilerplate to get Arrow schema auto schema_result = reader.Schema(); @@ -923,6 +962,129 @@ TEST_P(AvroReaderParameterizedTest, SplitWithRowIdNotSupported) { HasErrorMessage("'_row_id' metadata column with split is not supported")); } +constexpr std::string_view kFlatSchemaWithoutIds = R"({ + "type": "record", + "name": "flat", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": ["null", "string"]}, + {"name": "extra", "type": "long"} + ] +})"; + +constexpr std::string_view kNestedSchemaWithoutIds = R"({ + "type": "record", + "name": "nested", + "fields": [ + {"name": "profile", "type": { + "type": "record", "name": "profile_t", "fields": [{"name": "age", "type": "int"}] + }}, + {"name": "tags", "type": {"type": "array", "items": "string"}}, + {"name": "counts", "type": {"type": "map", "values": "long"}} + ] +})"; + +TEST_P(AvroReaderParameterizedTest, ReadWithoutFieldIdsOrNameMapping) { + // Without a name mapping, fields are matched by the read schema's names. + ASSERT_NO_FATAL_FAILURE(CreateAvroFileWithoutFieldIds( + std::string(kFlatSchemaWithoutIds), + {R"({"id": 1, "name": {"string": "Alice"}, "extra": 10})", + R"({"id": 2, "name": null, "extra": 20})"})); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(2, "name", string()), + SchemaField::MakeRequired(7, "id", int32()), + SchemaField::MakeOptional(3, "full_name", string()), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, OpenReader(schema)); + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([["Alice", 1, null], [null, 2, null]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_P(AvroReaderParameterizedTest, ReadWithNameMappingWithoutFieldIds) { + ASSERT_NO_FATAL_FAILURE(CreateAvroFileWithoutFieldIds( + std::string(kFlatSchemaWithoutIds), + {R"({"id": 1, "name": {"string": "Alice"}, "extra": 10})", + R"({"id": 2, "name": null, "extra": 20})"})); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(2, "full_name", string()), + SchemaField::MakeRequired(1, "id", int32()), + }); + // "extra" is not mapped, so it is not read. + std::shared_ptr mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"name", "full_name"}, .field_id = 2}, + }); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, OpenReader(schema, mapping)); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([["Alice", 1], [null, 2]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_P(AvroReaderParameterizedTest, ReadNestedFieldsWithoutFieldIds) { + ASSERT_NO_FATAL_FAILURE(CreateAvroFileWithoutFieldIds( + std::string(kNestedSchemaWithoutIds), + {R"({"profile": {"age": 30}, "tags": ["a", "b"], "counts": {"x": 1}})", + R"({"profile": {"age": 40}, "tags": [], "counts": {}})"})); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "profile", + std::make_shared(std::vector{ + SchemaField::MakeRequired(10, "age", int32()), + })), + SchemaField::MakeRequired( + 2, "tags", + std::make_shared(SchemaField::MakeRequired(20, "element", string()))), + SchemaField::MakeRequired( + 3, "counts", + std::make_shared(SchemaField::MakeRequired(30, "key", string()), + SchemaField::MakeRequired(31, "value", int64()))), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, OpenReader(schema)); + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([[[30], ["a", "b"], [["x", 1]]], [[40], [], []]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_P(AvroReaderParameterizedTest, ReadContainersWithUnmappedChildrenAsMissing) { + ASSERT_NO_FATAL_FAILURE(CreateAvroFileWithoutFieldIds( + std::string(kNestedSchemaWithoutIds), + {R"({"profile": {"age": 30}, "tags": ["a", "b"], "counts": {"x": 1}})"})); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "profile", + std::make_shared(std::vector{ + SchemaField::MakeOptional(10, "age", int32()), + })), + SchemaField::MakeOptional( + 2, "tags", + std::make_shared(SchemaField::MakeOptional(20, "element", string()))), + SchemaField::MakeOptional( + 3, "counts", + std::make_shared(SchemaField::MakeRequired(30, "key", string()), + SchemaField::MakeOptional(31, "value", int64()))), + }); + // The list element and the map value are not mapped, so both containers are read as + // missing. + std::shared_ptr mapping = NameMapping::Make({ + MappedField{.names = {"profile"}, + .field_id = 1, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"age"}, .field_id = 10}, + })}, + MappedField{.names = {"tags"}, .field_id = 2}, + MappedField{.names = {"counts"}, + .field_id = 3, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"key"}, .field_id = 30}, + })}, + }); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, OpenReader(schema, mapping)); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[[30], null, null]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + INSTANTIATE_TEST_SUITE_P(DirectDecoderModes, AvroReaderParameterizedTest, ::testing::Bool(), [](const ::testing::TestParamInfo& info) {