From f4be233cbf10d53e2b0c663244318db22c030a35 Mon Sep 17 00:00:00 2001 From: ump45nose <52391318+ump45nose@users.noreply.github.com> Date: Sat, 24 Oct 2026 09:13:41 +0800 Subject: [PATCH 1/2] feat(avro,parquet): read the full file schema when no projection is set When ReaderOptions::projection is not set, the Avro and Parquet readers now derive the read schema from the file schema instead of failing. Closes #514 --- src/iceberg/avro/avro_reader.cc | 13 +- src/iceberg/avro/avro_schema_util.cc | 160 ++++++++++++++++++ src/iceberg/avro/avro_schema_util_internal.h | 11 ++ src/iceberg/file_reader.h | 3 +- src/iceberg/parquet/parquet_reader.cc | 35 ++-- src/iceberg/parquet/parquet_schema_util.cc | 77 +++++++++ .../parquet/parquet_schema_util_internal.h | 12 ++ src/iceberg/test/avro_schema_test.cc | 103 +++++++++++ src/iceberg/test/avro_test.cc | 21 +++ src/iceberg/test/parquet_schema_test.cc | 99 +++++++++++ src/iceberg/test/parquet_test.cc | 47 +++++ 11 files changed, 557 insertions(+), 24 deletions(-) diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc index f333a38cbc..3e4c99a0db 100644 --- a/src/iceberg/avro/avro_reader.cc +++ b/src/iceberg/avro/avro_reader.cc @@ -284,13 +284,7 @@ std::shared_ptr<::arrow::DataType> StorageTypeForBuilder( class AvroReader::Impl { public: Status Open(const ReaderOptions& options) { - // TODO(gangwu): perhaps adding a ReaderOptions::Validate() method - if (options.projection == nullptr) { - return InvalidArgument("Projected schema is required by Avro reader"); - } - batch_size_ = options.properties.Get(ReaderProperties::kBatchSize); - read_schema_ = options.projection; // Open the input stream and adapt to the avro interface. ICEBERG_ASSIGN_OR_RAISE( @@ -328,6 +322,13 @@ class AvroReader::Impl { return InvalidSchema("Not all fields in the Avro file schema have field IDs"); } + if (options.projection != nullptr) { + read_schema_ = options.projection; + } else { + // Read all fields of the file schema when no projection is given. + ICEBERG_ASSIGN_OR_RAISE(read_schema_, FromAvroSchema(file_schema.root())); + } + // Project the read schema on top of the file schema. ICEBERG_ASSIGN_OR_RAISE(projection_, Project(*read_schema_, file_schema.root(), /*prune_source=*/false)); diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 94d0a6116b..fcd8f7756b 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -903,6 +903,166 @@ Result Project(const Schema& expected_schema, namespace { +Result> FromAvroNode(const ::avro::NodePtr& node); + +std::optional GetIcebergFieldName(const ::avro::NodePtr& node, + size_t field_idx) { + if (field_idx >= node->customAttributes()) { + return std::nullopt; + } + auto name = node->customAttributesAt(field_idx).getAttribute( + std::string(kIcebergFieldNameProp)); + if (name.has_value() && name->size() >= 2 && name->front() == '"' && + name->back() == '"') { + return name->substr(1, name->size() - 2); + } + return name; +} + +// Converts an Avro field node to an Iceberg field. A union with null is read as an +// optional field and a bare null node as an optional field of unknown type. +Result FromAvroField(int32_t field_id, std::string_view name, + const ::avro::NodePtr& node) { + if (node->type() == ::avro::AVRO_NULL) { + return SchemaField::MakeOptional(field_id, name, iceberg::unknown()); + } + ::avro::NodePtr value_node; + ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(node, &value_node)); + ICEBERG_ASSIGN_OR_RAISE(auto type, FromAvroNode(value_node)); + bool optional = node->type() == ::avro::AVRO_UNION; + return SchemaField(field_id, name, std::move(type), optional); +} + +Result> FromAvroRecordFields(const ::avro::NodePtr& node) { + if (node->type() != ::avro::AVRO_RECORD) { + return InvalidSchema("Expected AVRO_RECORD type, but got {}", ToString(node)); + } + std::vector fields; + fields.reserve(node->leaves()); + for (size_t i = 0; i < node->leaves(); ++i) { + ICEBERG_ASSIGN_OR_RAISE(int32_t field_id, GetFieldId(node, i)); + std::string name = GetIcebergFieldName(node, i).value_or(node->nameAt(i)); + ICEBERG_ASSIGN_OR_RAISE(auto field, FromAvroField(field_id, name, node->leafAt(i))); + fields.emplace_back(std::move(field)); + } + return fields; +} + +Result> FromAvroTimestamp(const ::avro::NodePtr& node, bool nanos) { + bool adjust_to_utc = GetAdjustToUtc(node).value_or("false") == "true"; + if (nanos) { + return adjust_to_utc ? std::shared_ptr(iceberg::timestamptz_ns()) + : std::shared_ptr(iceberg::timestamp_ns()); + } + return adjust_to_utc ? std::shared_ptr(iceberg::timestamp_tz()) + : std::shared_ptr(iceberg::timestamp()); +} + +Result> FromAvroNode(const ::avro::NodePtr& node) { + switch (node->type()) { + case ::avro::AVRO_RECORD: { + ICEBERG_ASSIGN_OR_RAISE(auto fields, FromAvroRecordFields(node)); + return iceberg::struct_(std::move(fields)); + } + case ::avro::AVRO_ARRAY: { + if (node->leaves() != 1) { + return InvalidSchema("Array type must have exactly one node, got {}", + node->leaves()); + } + if (HasMapLogicalType(node)) { + const auto& entry_node = node->leafAt(0); + ICEBERG_ASSIGN_OR_RAISE(auto entry_fields, FromAvroRecordFields(entry_node)); + if (entry_fields.size() != 2) { + return InvalidSchema( + "Array-backed map type must have a record node with two fields"); + } + SchemaField key(entry_fields[0].field_id(), MapType::kKeyName, + entry_fields[0].type(), entry_fields[0].optional()); + SchemaField value(entry_fields[1].field_id(), MapType::kValueName, + entry_fields[1].type(), entry_fields[1].optional()); + return iceberg::map(std::move(key), std::move(value)); + } + ICEBERG_ASSIGN_OR_RAISE(int32_t element_id, GetElementId(node)); + ICEBERG_ASSIGN_OR_RAISE( + auto element, + FromAvroField(element_id, ListType::kElementName, node->leafAt(0))); + return iceberg::list(std::move(element)); + } + case ::avro::AVRO_MAP: { + if (node->leaves() != 2) { + return InvalidSchema("Map type must have exactly two nodes, got {}", + node->leaves()); + } + ICEBERG_ASSIGN_OR_RAISE(int32_t key_id, GetKeyId(node)); + ICEBERG_ASSIGN_OR_RAISE(int32_t value_id, GetValueId(node)); + auto key = SchemaField::MakeRequired(key_id, MapType::kKeyName, iceberg::string()); + ICEBERG_ASSIGN_OR_RAISE( + auto value, FromAvroField(value_id, MapType::kValueName, node->leafAt(1))); + return iceberg::map(std::move(key), std::move(value)); + } + case ::avro::AVRO_BOOL: + return iceberg::boolean(); + case ::avro::AVRO_INT: + if (HasLogicalType(node, ::avro::LogicalType::DATE)) { + return iceberg::date(); + } + if (HasLogicalType(node, ::avro::LogicalType::NONE)) { + return iceberg::int32(); + } + break; + case ::avro::AVRO_LONG: + switch (node->logicalType().type()) { + case ::avro::LogicalType::NONE: + return iceberg::int64(); + case ::avro::LogicalType::TIME_MICROS: + return iceberg::time(); + case ::avro::LogicalType::TIMESTAMP_MICROS: + return FromAvroTimestamp(node, /*nanos=*/false); + case ::avro::LogicalType::TIMESTAMP_NANOS: + return FromAvroTimestamp(node, /*nanos=*/true); + default: + break; + } + break; + case ::avro::AVRO_FLOAT: + return iceberg::float32(); + case ::avro::AVRO_DOUBLE: + return iceberg::float64(); + case ::avro::AVRO_STRING: + return iceberg::string(); + case ::avro::AVRO_BYTES: + if (HasLogicalType(node, ::avro::LogicalType::NONE)) { + return iceberg::binary(); + } + break; + case ::avro::AVRO_FIXED: + if (HasLogicalType(node, ::avro::LogicalType::DECIMAL)) { + const auto logical_type = node->logicalType(); + return iceberg::decimal(static_cast(logical_type.precision()), + static_cast(logical_type.scale())); + } + if (HasLogicalType(node, ::avro::LogicalType::UUID) && node->fixedSize() == 16) { + return iceberg::uuid(); + } + if (HasLogicalType(node, ::avro::LogicalType::NONE)) { + return iceberg::fixed(static_cast(node->fixedSize())); + } + break; + default: + break; + } + return NotSupported("Cannot convert Avro type {} to an Iceberg type", ToString(node)); +} + +} // namespace + +Result> FromAvroSchema(const ::avro::NodePtr& avro_node) { + ICEBERG_ASSIGN_OR_RAISE(auto fields, FromAvroRecordFields(avro_node)); + return std::make_unique(std::move(fields)); +} + +namespace { + class NamesGuard { public: NamesGuard(std::vector& names, const std::string& name) : names_(names) { diff --git a/src/iceberg/avro/avro_schema_util_internal.h b/src/iceberg/avro/avro_schema_util_internal.h index 342b119a53..546a10302a 100644 --- a/src/iceberg/avro/avro_schema_util_internal.h +++ b/src/iceberg/avro/avro_schema_util_internal.h @@ -146,6 +146,17 @@ class HasIdVisitor { Result Project(const Schema& expected_schema, const ::avro::NodePtr& avro_node, bool prune_source); +/// \brief Convert an Avro record schema annotated with Iceberg field IDs to an +/// Iceberg Schema. +/// +/// Every field must carry a field ID (and every array or map an element, key or +/// value ID). A union with null becomes an optional field. +/// +/// \param avro_node The Avro record node to convert. +/// \return The converted Iceberg Schema, or an error if the Avro schema cannot be +/// represented in Iceberg. +Result> FromAvroSchema(const ::avro::NodePtr& avro_node); + std::string ToString(const ::avro::NodePtr& node); std::string ToString(const ::avro::LogicalType& logical_type); std::string ToString(const ::avro::LogicalType::Type& logical_type); diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index e08b2df494..4da41dc033 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -102,7 +102,8 @@ struct ICEBERG_EXPORT ReaderOptions { std::optional split; /// \brief FileIO instance to open the file. std::shared_ptr io; - /// \brief The projection schema to read from the file. This field is required. + /// \brief The projection schema to read from the file. If not set, all fields of + /// the file schema are read. std::shared_ptr projection; /// \brief The filter to apply to the data. Reader implementations may ignore this if /// the file format does not support filtering. diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 549a3d3aed..c2bbb35b8f 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -60,27 +60,25 @@ Result> OpenInputStream( return arrow::OpenArrowInputStream(options.io, options.path, options.length); } -Result BuildProjection(::parquet::arrow::FileReader* reader, - const Schema& read_schema, - const NameMapping* name_mapping) { +// SchemaManifest holds pointers into its own fields, so it is filled in place. +Status MakeSchemaManifest(::parquet::arrow::FileReader* reader, + const NameMapping* name_mapping, + ::parquet::arrow::SchemaManifest* schema_manifest) { auto metadata = reader->parquet_reader()->metadata(); - ::parquet::arrow::SchemaManifest schema_manifest; ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make( metadata->schema(), metadata->key_value_metadata(), reader->properties(), - &schema_manifest)); + 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); + ApplyNameMapping(*name_mapping, schema_manifest); } - // Leverage SchemaManifest to project the schema - ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, schema_manifest)); - return projection; + return {}; } // forward declaration to unblock cycle dependence. @@ -277,12 +275,6 @@ class ParquetReader::Impl { public: // Open the Parquet reader with the given options Status Open(const ReaderOptions& options) { - if (options.projection == nullptr) { - return InvalidArgument("Projected schema is required by Parquet reader"); - } - - read_schema_ = options.projection; - // Prepare reader properties ::parquet::ReaderProperties reader_properties(pool_); ::parquet::ArrowReaderProperties arrow_reader_properties; @@ -303,9 +295,18 @@ class ParquetReader::Impl { arrow_reader_properties)); ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options)); + ::parquet::arrow::SchemaManifest schema_manifest; + ICEBERG_RETURN_UNEXPECTED( + MakeSchemaManifest(reader_.get(), options.name_mapping.get(), &schema_manifest)); + if (options.projection != nullptr) { + read_schema_ = options.projection; + } else { + // Read all fields of the file schema when no projection is given. + ICEBERG_ASSIGN_OR_RAISE(read_schema_, FromParquetSchema(schema_manifest)); + } + // Project read schema onto the Parquet file schema - ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_, - options.name_mapping.get())); + ICEBERG_ASSIGN_OR_RAISE(projection_, Project(*read_schema_, schema_manifest)); 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 2a230fe4cc..1aca81677b 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -17,6 +17,7 @@ * under the License. */ +#include #include #include #include @@ -24,6 +25,7 @@ #include #include +#include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/constants.h" #include "iceberg/metadata_columns.h" #include "iceberg/name_mapping.h" @@ -785,6 +787,60 @@ void ApplyNameMappingToField(const NameMapping& mapping, std::vector ToArrowFieldWithIds( + const ::parquet::arrow::SchemaField& parquet_field) { + if (!GetFieldId(parquet_field).has_value()) { + return nullptr; + } + + const auto& field = parquet_field.field; + const auto& children = parquet_field.children; + switch (field->type()->id()) { + case ::arrow::Type::STRUCT: { + ::arrow::FieldVector child_fields; + child_fields.reserve(children.size()); + for (const auto& child : children) { + if (auto child_field = ToArrowFieldWithIds(child); child_field != nullptr) { + child_fields.push_back(std::move(child_field)); + } + } + return field->WithType(::arrow::struct_(std::move(child_fields))); + } + case ::arrow::Type::LIST: + case ::arrow::Type::LARGE_LIST: { + if (children.size() != 1) { + return nullptr; + } + auto element_field = ToArrowFieldWithIds(children[0]); + if (element_field == nullptr) { + return nullptr; + } + return field->WithType( + ::arrow::list(element_field->WithName(std::string(ListType::kElementName)))); + } + case ::arrow::Type::MAP: { + if (children.size() != 1 || children[0].children.size() != 2) { + return nullptr; + } + auto key_field = ToArrowFieldWithIds(children[0].children[0]); + auto value_field = ToArrowFieldWithIds(children[0].children[1]); + if (key_field == nullptr || value_field == nullptr) { + return nullptr; + } + return field->WithType(std::make_shared<::arrow::MapType>( + key_field->WithName(std::string(MapType::kKeyName)), + value_field->WithName(std::string(MapType::kValueName)))); + } + default: + return field; + } +} + } // namespace Result Project(const Schema& expected_schema, @@ -830,6 +886,27 @@ void ApplyNameMapping(const NameMapping& mapping, } } +Result> FromParquetSchema( + const ::parquet::arrow::SchemaManifest& parquet_schema) { + ::arrow::FieldVector fields; + fields.reserve(parquet_schema.schema_fields.size()); + for (const auto& parquet_field : parquet_schema.schema_fields) { + if (auto field = ToArrowFieldWithIds(parquet_field); field != nullptr) { + fields.push_back(std::move(field)); + } + } + + ArrowSchema arrow_schema; + if (auto status = + ::arrow::ExportSchema(::arrow::Schema(std::move(fields)), &arrow_schema); + !status.ok()) { + return InvalidSchema("Failed to export Parquet file schema to Arrow: {}", + status.message()); + } + internal::ArrowSchemaGuard guard(&arrow_schema); + return FromArrowSchema(arrow_schema, /*schema_id=*/std::nullopt); +} + 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 6612de0569..d2a6aa62c3 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -74,6 +74,18 @@ bool HasFieldIds(const ::parquet::schema::NodePtr& root_node); void ApplyNameMapping(const NameMapping& mapping, ::parquet::arrow::SchemaManifest* parquet_schema); +/// \brief Convert a Parquet schema to an Iceberg schema. +/// +/// Fields without a field ID (including lists and maps whose element, key or value +/// has no field ID) cannot be read by ID and are left out of the result. Apply a +/// name mapping first for files written without field IDs. +/// +/// \param parquet_schema The Parquet schema to convert. +/// \return The converted Iceberg schema, or an error if a Parquet type cannot be +/// represented in Iceberg. +Result> FromParquetSchema( + const ::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/avro_schema_test.cc b/src/iceberg/test/avro_schema_test.cc index f147a607bc..9d4cd451e1 100644 --- a/src/iceberg/test/avro_schema_test.cc +++ b/src/iceberg/test/avro_schema_test.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include #include "iceberg/avro/avro_register.h" @@ -30,6 +31,7 @@ #include "iceberg/name_mapping.h" #include "iceberg/schema.h" #include "iceberg/test/matchers.h" +#include "iceberg/util/macros.h" namespace iceberg::avro { @@ -1942,4 +1944,105 @@ TEST_F(NameMappingAvroSchemaTest, MissingFieldIdError) { auto result = MakeAvroNodeWithFieldIds(avro_schema.root(), *name_mapping); ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); } +namespace { + +// Converts an Iceberg schema to Avro, serializes it to JSON and parses it back, as an +// Avro file schema would be, before converting it back to Iceberg. +Result> RoundTripThroughAvro(const Schema& schema) { + ::avro::NodePtr node; + ICEBERG_RETURN_UNEXPECTED(ToAvroNodeVisitor{}.Visit(schema, &node)); + auto json = ::avro::ValidSchema(node).toJson(/*prettyPrint=*/false); + auto parsed = ::avro::compileJsonSchemaFromString(json); + return FromAvroSchema(parsed.root()); +} + +} // namespace + +TEST(FromAvroSchemaTest, PrimitiveTypes) { + Schema schema({ + SchemaField::MakeRequired(1, "bool_col", iceberg::boolean()), + SchemaField::MakeOptional(2, "int_col", iceberg::int32()), + SchemaField::MakeRequired(3, "long_col", iceberg::int64()), + SchemaField::MakeOptional(4, "float_col", iceberg::float32()), + SchemaField::MakeRequired(5, "double_col", iceberg::float64()), + SchemaField::MakeRequired(6, "decimal_col", iceberg::decimal(10, 2)), + SchemaField::MakeRequired(7, "date_col", iceberg::date()), + SchemaField::MakeRequired(8, "time_col", iceberg::time()), + SchemaField::MakeRequired(9, "ts_col", iceberg::timestamp()), + SchemaField::MakeRequired(10, "ts_tz_col", iceberg::timestamp_tz()), + SchemaField::MakeRequired(11, "ts_ns_col", iceberg::timestamp_ns()), + SchemaField::MakeRequired(12, "ts_tz_ns_col", iceberg::timestamptz_ns()), + SchemaField::MakeOptional(13, "string_col", iceberg::string()), + SchemaField::MakeRequired(14, "uuid_col", iceberg::uuid()), + SchemaField::MakeRequired(15, "fixed_col", iceberg::fixed(20)), + SchemaField::MakeOptional(16, "binary_col", iceberg::binary()), + SchemaField::MakeOptional(17, "unknown_col", iceberg::unknown()), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto result, RoundTripThroughAvro(schema)); + EXPECT_EQ(*result, schema); +} + +TEST(FromAvroSchemaTest, NestedTypes) { + Schema schema({ + SchemaField::MakeRequired( + 1, "struct_col", + iceberg::struct_({SchemaField::MakeRequired(2, "a", iceberg::int32()), + SchemaField::MakeOptional(3, "b", iceberg::string())})), + SchemaField::MakeOptional( + 4, "list_col", + iceberg::list(SchemaField::MakeOptional(5, "element", iceberg::int64()))), + SchemaField::MakeRequired( + 6, "string_map_col", + iceberg::map(SchemaField::MakeRequired(7, "key", iceberg::string()), + SchemaField::MakeOptional(8, "value", iceberg::float64()))), + SchemaField::MakeOptional( + 9, "int_map_col", + iceberg::map( + SchemaField::MakeRequired(10, "key", iceberg::int32()), + SchemaField::MakeRequired(11, "value", + iceberg::struct_({SchemaField::MakeRequired( + 12, "c", iceberg::boolean())})))), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto result, RoundTripThroughAvro(schema)); + EXPECT_EQ(*result, schema); +} + +TEST(FromAvroSchemaTest, RestoresSanitizedFieldNames) { + Schema schema({ + SchemaField::MakeRequired(1, "user-name", iceberg::string()), + SchemaField::MakeOptional(2, "1st", iceberg::int32()), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto result, RoundTripThroughAvro(schema)); + EXPECT_EQ(*result, schema); +} + +TEST(FromAvroSchemaTest, MissingFieldIdError) { + auto avro_schema = ::avro::compileJsonSchemaFromString(R"({ + "type": "record", + "name": "test_record", + "fields": [ + {"name": "id", "type": "int", "field-id": 1}, + {"name": "name", "type": "string"} + ] + })"); + + EXPECT_THAT(FromAvroSchema(avro_schema.root()), IsError(ErrorKind::kInvalidSchema)); +} + +TEST(FromAvroSchemaTest, UnsupportedTypeError) { + auto avro_schema = ::avro::compileJsonSchemaFromString(R"({ + "type": "record", + "name": "test_record", + "fields": [ + {"name": "e", "type": {"type": "enum", "name": "e", "symbols": ["A"]}, + "field-id": 1} + ] + })"); + + EXPECT_THAT(FromAvroSchema(avro_schema.root()), IsError(ErrorKind::kNotSupported)); +} + } // namespace iceberg::avro diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index a161164102..c51b2fe801 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -41,6 +41,7 @@ #include #include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/avro/avro_constants.h" #include "iceberg/avro/avro_register.h" #include "iceberg/avro/avro_stream_internal.h" @@ -323,6 +324,26 @@ TEST_F(AvroReaderTest, ReadTwoFields) { ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); } +TEST_F(AvroReaderTest, ReadWithoutProjection) { + CreateSimpleAvroFile(); + + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open( + FileFormatType::kAvro, {.path = temp_avro_file_, .io = file_io_})); + + ICEBERG_UNWRAP_OR_FAIL(auto arrow_schema, reader->Schema()); + internal::ArrowSchemaGuard guard(&arrow_schema); + ICEBERG_UNWRAP_OR_FAIL(auto read_schema, FromArrowSchema(arrow_schema, std::nullopt)); + EXPECT_EQ(*read_schema, + Schema(std::vector{ + SchemaField::MakeRequired(1, "id", std::make_shared()), + SchemaField::MakeOptional(2, "name", std::make_shared())})); + + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([[1, "Alice"], [2, "Bob"], [3, "Charlie"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + TEST_F(AvroReaderTest, 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 diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index c9f503691c..c888a1454e 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -356,6 +356,105 @@ TEST(ParquetNameMappingTest, ApplyNameMappingWithUnmappedFields) { ASSERT_EQ(SelectedColumnIndices(projection), std::vector({0})); } +TEST(ParquetSchemaConversionTest, FromParquetSchemaRoundTrip) { + Schema schema({ + SchemaField::MakeRequired(/*field_id=*/1, "bool_col", boolean()), + SchemaField::MakeOptional(/*field_id=*/2, "int_col", int32()), + SchemaField::MakeRequired(/*field_id=*/3, "long_col", int64()), + SchemaField::MakeOptional(/*field_id=*/4, "float_col", float32()), + SchemaField::MakeRequired(/*field_id=*/5, "double_col", float64()), + SchemaField::MakeRequired(/*field_id=*/6, "decimal_col", decimal(10, 2)), + SchemaField::MakeRequired(/*field_id=*/7, "date_col", date()), + SchemaField::MakeRequired(/*field_id=*/8, "time_col", time()), + SchemaField::MakeRequired(/*field_id=*/9, "ts_col", timestamp()), + SchemaField::MakeRequired(/*field_id=*/10, "ts_tz_col", timestamp_tz()), + SchemaField::MakeRequired(/*field_id=*/11, "ts_ns_col", timestamp_ns()), + SchemaField::MakeOptional(/*field_id=*/12, "string_col", string()), + SchemaField::MakeRequired(/*field_id=*/13, "uuid_col", uuid()), + SchemaField::MakeRequired(/*field_id=*/14, "fixed_col", fixed(20)), + SchemaField::MakeOptional(/*field_id=*/15, "binary_col", binary()), + SchemaField::MakeOptional( + /*field_id=*/16, "struct_col", + std::make_shared(std::vector{ + SchemaField::MakeRequired(/*field_id=*/17, "a", int32()), + SchemaField::MakeOptional(/*field_id=*/18, "b", string()), + })), + SchemaField::MakeOptional(/*field_id=*/19, "list_col", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/20, "element", int64()))), + SchemaField::MakeRequired( + /*field_id=*/21, "map_col", + std::make_shared( + SchemaField::MakeRequired(/*field_id=*/22, "key", string()), + SchemaField::MakeOptional(/*field_id=*/23, "value", float64()))), + }); + + ICEBERG_UNWRAP_OR_FAIL(auto parquet_schema, ToParquetSchema(schema)); + for (auto list_type : {::arrow::Type::LIST, ::arrow::Type::LARGE_LIST}) { + SCOPED_TRACE(list_type); + auto schema_manifest = MakeSchemaManifest(parquet_schema->schema_root(), list_type); + ICEBERG_UNWRAP_OR_FAIL(auto result, FromParquetSchema(schema_manifest)); + EXPECT_EQ(*result, schema); + } +} + +TEST(ParquetSchemaConversionTest, FromParquetSchemaWithNameMapping) { + auto parquet_schema = MakeGroupNode( + "iceberg_schema", + { + MakeInt64Node("id"), + MakeStringNode("unmapped"), + MakeGroupNode("profile", {MakeStringNode("name"), MakeStringNode("extra")}), + MakeListNode("numbers", MakeInt32Node("item")), + MakeListNode("unmapped_element", MakeInt32Node("item")), + MakeMapNode("counts", + MakeStringNode("key", /*field_id=*/-1, /*optional=*/false), + MakeInt32Node("value")), + }); + auto mapping = NameMapping::Make({ + MappedField{.names = {"id"}, .field_id = 1}, + MappedField{.names = {"profile"}, + .field_id = 2, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"name"}, .field_id = 201}, + })}, + MappedField{.names = {"numbers"}, + .field_id = 3, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"element"}, .field_id = 301}, + })}, + MappedField{.names = {"unmapped_element"}, .field_id = 4}, + MappedField{.names = {"counts"}, + .field_id = 5, + .nested_mapping = MappedFields::Make({ + MappedField{.names = {"key"}, .field_id = 501}, + MappedField{.names = {"value"}, .field_id = 502}, + })}, + }); + + auto schema_manifest = MakeSchemaManifest(parquet_schema); + ApplyNameMapping(*mapping, &schema_manifest); + ICEBERG_UNWRAP_OR_FAIL(auto result, FromParquetSchema(schema_manifest)); + + Schema expected_schema({ + SchemaField::MakeOptional(/*field_id=*/1, "id", int64()), + SchemaField::MakeOptional( + /*field_id=*/2, "profile", + std::make_shared(std::vector{ + SchemaField::MakeOptional(/*field_id=*/201, "name", string()), + })), + SchemaField::MakeOptional(/*field_id=*/3, "numbers", + std::make_shared(SchemaField::MakeOptional( + /*field_id=*/301, "element", int32()))), + SchemaField::MakeOptional( + /*field_id=*/5, "counts", + std::make_shared( + SchemaField::MakeRequired(/*field_id=*/501, "key", string()), + SchemaField::MakeOptional(/*field_id=*/502, "value", int32()))), + }); + EXPECT_EQ(*result, expected_schema); +} + 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 c35ba44717..b5ab0618c4 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -45,6 +45,7 @@ #include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/arrow/arrow_status_internal.h" +#include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/expression/literal.h" #include "iceberg/file_reader.h" #include "iceberg/file_writer.h" @@ -571,6 +572,52 @@ TEST_F(ParquetReaderTest, ReadTwoFields) { ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); } +TEST_F(ParquetReaderTest, ReadWithoutProjection) { + CreateSimpleParquetFile(); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + {.path = temp_parquet_file_, .io = file_io_})); + + ICEBERG_UNWRAP_OR_FAIL(auto arrow_schema, reader->Schema()); + internal::ArrowSchemaGuard guard(&arrow_schema); + ICEBERG_UNWRAP_OR_FAIL(auto read_schema, FromArrowSchema(arrow_schema, std::nullopt)); + EXPECT_EQ(*read_schema, Schema(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "name", string())})); + + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(*reader, R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadWithoutProjectionWithNameMapping) { + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( + ::arrow::struct_({::arrow::field("id", ::arrow::int32()), + ::arrow::field("name", ::arrow::utf8())}), + R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])")); + // Fields without a mapped ID cannot be read by ID and are skipped. + auto mapping = NameMapping::Make({MappedField{.names = {"name"}, .field_id = 2}}); + + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, + {.path = temp_parquet_file_, + .io = file_io_, + .name_mapping = std::move(mapping)})); + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([["Foo"], ["Bar"], ["Baz"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); +} + +TEST_F(ParquetReaderTest, ReadWithoutProjectionOrFieldIds) { + ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( + ::arrow::struct_({::arrow::field("id", ::arrow::int32())}), R"([[1]])")); + + auto reader_result = ReaderFactoryRegistry::Open( + FileFormatType::kParquet, {.path = temp_parquet_file_, .io = file_io_}); + ASSERT_THAT(reader_result, IsError(ErrorKind::kInvalidSchema)); + EXPECT_THAT(reader_result, HasErrorMessage("no field IDs and no name mapping")); +} + TEST_F(ParquetReaderTest, ReadWithoutFieldIdsOrNameMapping) { ASSERT_NO_FATAL_FAILURE(CreateParquetFileWithoutFieldIds( ::arrow::struct_({::arrow::field("id", ::arrow::int32())}), R"([[1]])")); From fd253e18e2e85e2cd4e910bcfdef213fedd964bd Mon Sep 17 00:00:00 2001 From: Vincent <1729065730@qq.com> Date: Sat, 24 Oct 2026 21:28:26 +0800 Subject: [PATCH 2/2] fix: export FromArrowSchema for the bundle library The Parquet reader in libiceberg_bundle now calls FromArrowSchema, which lives in libiceberg, so shared builds failed to link it. Mark it ICEBERG_EXPORT like ToArrowSchema. Signed-off-by: Vincent <1729065730@qq.com> --- src/iceberg/schema_internal.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/iceberg/schema_internal.h b/src/iceberg/schema_internal.h index fd9ec62661..1c0896379a 100644 --- a/src/iceberg/schema_internal.h +++ b/src/iceberg/schema_internal.h @@ -45,8 +45,8 @@ ICEBERG_EXPORT Status ToArrowSchema(const Schema& schema, ArrowSchema* out); /// \param[in] schema The Arrow schema to convert. /// \param[in] schema_id The schema ID of the Iceberg schema. /// \return The Iceberg schema or an error if the conversion fails. -Result> FromArrowSchema(const ArrowSchema& schema, - std::optional schema_id); +ICEBERG_EXPORT Result> FromArrowSchema( + const ArrowSchema& schema, std::optional schema_id); /// \brief Convert a struct type to an Iceberg schema. ///