feat(avro,parquet): read the full file schema when no projection is set

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));

src/iceberg/avro/avro_schema_util.cc

@@ -903,6 +903,166 @@ Result<SchemaProjection> Project(const Schema& expected_schema,
 
 namespace {
 
+Result<std::shared_ptr<Type>> FromAvroNode(const ::avro::NodePtr& node);
+
+std::optional<std::string> 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<SchemaField> 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<std::vector<SchemaField>> FromAvroRecordFields(const ::avro::NodePtr& node) {
+  if (node->type() != ::avro::AVRO_RECORD) {
+    return InvalidSchema("Expected AVRO_RECORD type, but got {}", ToString(node));
+  }
+  std::vector<SchemaField> 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<std::shared_ptr<Type>> 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<Type>(iceberg::timestamptz_ns())
+                         : std::shared_ptr<Type>(iceberg::timestamp_ns());
+  }
+  return adjust_to_utc ? std::shared_ptr<Type>(iceberg::timestamp_tz())
+                       : std::shared_ptr<Type>(iceberg::timestamp());
+}
+
+Result<std::shared_ptr<Type>> 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<int32_t>(logical_type.precision()),
+                                static_cast<int32_t>(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<int32_t>(node->fixedSize()));
+      }
+      break;
+    default:
+      break;
+  }
+  return NotSupported("Cannot convert Avro type {} to an Iceberg type", ToString(node));
+}
+
+}  // namespace
+
+Result<std::unique_ptr<Schema>> FromAvroSchema(const ::avro::NodePtr& avro_node) {
+  ICEBERG_ASSIGN_OR_RAISE(auto fields, FromAvroRecordFields(avro_node));
+  return std::make_unique<Schema>(std::move(fields));
+}
+
+namespace {
+
 class NamesGuard {
  public:
   NamesGuard(std::vector<std::string>& names, const std::string& name) : names_(names) {

src/iceberg/avro/avro_schema_util_internal.h

@@ -146,6 +146,17 @@ class HasIdVisitor {
 Result<SchemaProjection> 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<std::unique_ptr<Schema>> 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);

src/iceberg/file_reader.h

@@ -102,7 +102,8 @@ struct ICEBERG_EXPORT ReaderOptions {
   std::optional<Split> split;
   /// \brief FileIO instance to open the file.
   std::shared_ptr<class FileIO> 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<class Schema> projection;
   /// \brief The filter to apply to the data. Reader implementations may ignore this if
   /// the file format does not support filtering.

src/iceberg/parquet/parquet_reader.cc

@@ -60,27 +60,25 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenInputStream(
   return arrow::OpenArrowInputStream(options.io, options.path, options.length);
 }
 
-Result<SchemaProjection> 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};

src/iceberg/parquet/parquet_schema_util.cc

@@ -17,13 +17,15 @@
  * under the License.
  */
 
+#include <arrow/c/bridge.h>
 #include <arrow/extension_type.h>
 #include <arrow/type.h>
 #include <arrow/type_fwd.h>
 #include <arrow/util/key_value_metadata.h>
 #include <parquet/arrow/schema.h>
 #include <parquet/schema.h>
 
+#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<std::string
   }
 }
 
+// Rebuilds the Arrow field of a Parquet schema field so that nested fields carry the
+// field IDs stored in the SchemaField tree (e.g. those assigned by a name mapping).
+// List elements and map keys/values use the Iceberg names. Returns nullptr if the field,
+// or any list element or map key/value inside it, has no field ID, since such a field
+// cannot be read by ID.
+std::shared_ptr<::arrow::Field> 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<SchemaProjection> Project(const Schema& expected_schema,
@@ -830,6 +886,27 @@ void ApplyNameMapping(const NameMapping& mapping,
   }
 }
 
+Result<std::unique_ptr<Schema>> 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<std::shared_ptr<::parquet::SchemaDescriptor>> ToParquetSchema(
     const Schema& iceberg_schema) {
   ::parquet::schema::NodeVector fields;