From bb8cd9fb2e8ac7d65510830b771afcd51d0fa2ae Mon Sep 17 00:00:00 2001 From: Vincent <1729065730@qq.com> Date: Fri, 30 Oct 2026 21:38:40 +0800 Subject: [PATCH] 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. Also export FromArrowSchema for the bundle library so shared builds can link it. Closes #514 Signed-off-by: Vincent <1729065730@qq.com> --- 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 | 38 ++++- src/iceberg/parquet/parquet_schema_util.cc | 77 +++++++++ .../parquet/parquet_schema_util_internal.h | 12 ++ src/iceberg/schema_internal.h | 4 +- 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 +++++ 12 files changed, 573 insertions(+), 15 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 eded9191c6..213578a323 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -108,7 +108,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 b88dfd000e..6f6b9bddbf 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -62,6 +62,27 @@ Result> OpenInputStream( return arrow::OpenArrowInputStream(options.io, options.path, options.length); } +// 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(); + + 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); + } + + return {}; +} + Result BuildProjection(::parquet::arrow::FileReader* reader, const Schema& read_schema, const NameMapping* name_mapping) { @@ -278,11 +299,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; stats_filter_.reset(); // Prepare reader properties @@ -304,6 +320,16 @@ class ParquetReader::Impl { reader_, ::parquet::arrow::FileReader::Make(pool_, std::move(file_reader), arrow_reader_properties)); + ::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())); @@ -313,7 +339,7 @@ class ParquetReader::Impl { ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter)); if (!is_bound) { ICEBERG_ASSIGN_OR_RAISE( - filter, Binder::Bind(*options.projection, filter, options.case_sensitive)); + filter, Binder::Bind(*read_schema_, filter, options.case_sensitive)); } ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( filter, *reader_->manifest().descr)); diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index d16b8a1ae6..5736588ff7 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" @@ -781,6 +783,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, @@ -826,6 +882,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 49c62ff090..500cd3d3d8 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -81,6 +81,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/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. /// 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]])"));