apacheGH-514: [Parquet] Infer schema when projection is null in ParquetReader

This change enables reading Parquet files without an explicit projection schema.
If `options.projection` is not provided, the reader now infers the Iceberg schema
from the Parquet file's Arrow schema using the Arrow C++ API.

*   Modified `src/iceberg/parquet/parquet_reader.cc`:
    *   Removed null check for `projection` in `Open`.
    *   Implemented `InferIcebergSchema` and `ConvertArrowType` to convert `arrow::Schema` to `iceberg::Schema` directly, avoiding complex C-ABI/nanoarrow dependencies.
    *   Used inferred schema when `projection` is null.
*   Added `src/iceberg/test/parquet_reader_no_projection_test.cc` to verify the fix.
*   Updated `src/iceberg/test/CMakeLists.txt` to register the new test file.

Co-authored-by: wgtmac <4684607+wgtmac@users.noreply.github.com>

Author: google-labs-jules[bot]

src/iceberg/CMakeLists.txt

@@ -173,7 +173,6 @@ if(ICEBERG_BUILD_BUNDLE)
   set(ICEBERG_BUNDLE_SOURCES
       arrow/arrow_fs_file_io.cc
       arrow/metadata_column_util.cc
-      arrow_c_data_guard_internal.cc
       avro/avro_data_util.cc
       avro/avro_direct_decoder.cc
       avro/avro_direct_encoder.cc
@@ -186,8 +185,7 @@ if(ICEBERG_BUILD_BUNDLE)
       parquet/parquet_reader.cc
       parquet/parquet_register.cc
       parquet/parquet_schema_util.cc
-      parquet/parquet_writer.cc
-      schema_internal.cc)
+      parquet/parquet_writer.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.
   set(ICEBERG_BUNDLE_STATIC_BUILD_INTERFACE_LIBS)

src/iceberg/parquet/parquet_reader.cc

@@ -19,7 +19,11 @@
 
 #include "iceberg/parquet/parquet_reader.h"
 
+#include <charconv>
+#include <cstring>
 #include <numeric>
+#include <optional>
+#include <string>
 
 #include <arrow/c/bridge.h>
 #include <arrow/memory_pool.h>
@@ -35,20 +39,148 @@
 #include "iceberg/arrow/arrow_fs_file_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/metadata_column_util_internal.h"
-#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/constants.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
 #include "iceberg/result.h"
+#include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/schema_util.h"
+#include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::parquet {
 
 namespace {
 
+constexpr int32_t kUnknownFieldId = -1;
+
+int32_t GetFieldId(const std::shared_ptr<arrow::Field>& field) {
+  if (!field->metadata()) {
+    return kUnknownFieldId;
+  }
+
+  int idx = field->metadata()->FindKey(kParquetFieldIdKey);
+  if (idx == -1) {
+    return kUnknownFieldId;
+  }
+
+  std::string value = field->metadata()->value(idx);
+  int32_t field_id = kUnknownFieldId;
+  std::from_chars(value.data(), value.data() + value.size(), field_id);
+
+  return field_id;
+}
+
+// Forward declaration
+Result<std::shared_ptr<Type>> ConvertArrowType(
+    const std::shared_ptr<arrow::DataType>& type);
+
+Result<std::unique_ptr<SchemaField>> ToSchemaField(
+    const std::shared_ptr<arrow::Field>& field) {
+  ICEBERG_ASSIGN_OR_RAISE(auto field_type, ConvertArrowType(field->type()));
+
+  auto field_id = GetFieldId(field);
+  return std::make_unique<SchemaField>(field_id, field->name(), std::move(field_type),
+                                       field->nullable());
+}
+
+Result<std::shared_ptr<Type>> ConvertArrowType(
+    const std::shared_ptr<arrow::DataType>& type) {
+  switch (type->id()) {
+    case arrow::Type::BOOL:
+      return iceberg::boolean();
+    case arrow::Type::INT32:
+      return iceberg::int32();
+    case arrow::Type::INT64:
+      return iceberg::int64();
+    case arrow::Type::FLOAT:
+      return iceberg::float32();
+    case arrow::Type::DOUBLE:
+      return iceberg::float64();
+    case arrow::Type::DECIMAL128: {
+      const auto& decimal_type = static_cast<const arrow::Decimal128Type&>(*type);
+      return iceberg::decimal(decimal_type.precision(), decimal_type.scale());
+    }
+    case arrow::Type::DATE32:
+      return iceberg::date();
+    case arrow::Type::TIME64: {
+      const auto& time_type = static_cast<const arrow::Time64Type&>(*type);
+      if (time_type.unit() != arrow::TimeUnit::MICRO) {
+        return InvalidSchema("Unsupported time unit for Arrow time type: {}",
+                             time_type.unit());
+      }
+      return iceberg::time();
+    }
+    case arrow::Type::TIMESTAMP: {
+      const auto& timestamp_type = static_cast<const arrow::TimestampType&>(*type);
+      if (timestamp_type.unit() != arrow::TimeUnit::MICRO) {
+        return InvalidSchema("Unsupported time unit for Arrow timestamp type: {}",
+                             timestamp_type.unit());
+      }
+      if (timestamp_type.timezone().empty()) {
+        return iceberg::timestamp();
+      } else {
+        return iceberg::timestamp_tz();
+      }
+    }
+    case arrow::Type::STRING:
+    case arrow::Type::LARGE_STRING:
+      return iceberg::string();
+    case arrow::Type::BINARY:
+    case arrow::Type::LARGE_BINARY:
+      return iceberg::binary();
+    case arrow::Type::FIXED_SIZE_BINARY: {
+      const auto& fixed_type = static_cast<const arrow::FixedSizeBinaryType&>(*type);
+      return iceberg::fixed(fixed_type.byte_width());
+    }
+    case arrow::Type::EXTENSION: {
+      const auto& ext_type = static_cast<const arrow::ExtensionType&>(*type);
+      if (ext_type.extension_name() == "arrow.uuid") {
+        return iceberg::uuid();
+      }
+      return ConvertArrowType(ext_type.storage_type());
+    }
+    case arrow::Type::STRUCT: {
+      const auto& struct_type = static_cast<const arrow::StructType&>(*type);
+      std::vector<SchemaField> fields;
+      fields.reserve(struct_type.num_fields());
+      for (const auto& field : struct_type.fields()) {
+        ICEBERG_ASSIGN_OR_RAISE(auto schema_field, ToSchemaField(field));
+        fields.emplace_back(std::move(*schema_field));
+      }
+      return std::make_shared<StructType>(std::move(fields));
+    }
+    case arrow::Type::LIST: {
+      const auto& list_type = static_cast<const arrow::ListType&>(*type);
+      ICEBERG_ASSIGN_OR_RAISE(auto element_field, ToSchemaField(list_type.value_field()));
+      return std::make_shared<ListType>(std::move(*element_field));
+    }
+    case arrow::Type::MAP: {
+      const auto& map_type = static_cast<const arrow::MapType&>(*type);
+      ICEBERG_ASSIGN_OR_RAISE(auto key_field, ToSchemaField(map_type.key_field()));
+      ICEBERG_ASSIGN_OR_RAISE(auto value_field, ToSchemaField(map_type.item_field()));
+      return std::make_shared<MapType>(std::move(*key_field), std::move(*value_field));
+    }
+    default:
+      return InvalidSchema("Unsupported Arrow type: {}", type->ToString());
+  }
+}
+
+Result<std::unique_ptr<Schema>> InferIcebergSchema(
+    const std::shared_ptr<arrow::Schema>& schema, std::optional<int32_t> schema_id) {
+  std::vector<SchemaField> fields;
+  fields.reserve(schema->num_fields());
+  for (const auto& field : schema->fields()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_field, ToSchemaField(field));
+    fields.emplace_back(std::move(*schema_field));
+  }
+  auto id = schema_id.value_or(Schema::kInitialSchemaId);
+  return std::make_unique<Schema>(std::move(fields), id);
+}
+
 Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenInputStream(
     const ReaderOptions& options) {
   ::arrow::fs::FileInfo file_info(options.path, ::arrow::fs::FileType::File);
@@ -135,10 +267,8 @@ class ParquetReader::Impl {
     } else {
       std::shared_ptr<::arrow::Schema> arrow_schema;
       ICEBERG_ARROW_RETURN_NOT_OK(reader_->GetSchema(&arrow_schema));
-      ArrowSchema c_arrow_schema;
-      ICEBERG_ARROW_RETURN_NOT_OK(::arrow::ExportSchema(*arrow_schema, &c_arrow_schema));
-      internal::ArrowSchemaGuard guard(&c_arrow_schema);
-      ICEBERG_ASSIGN_OR_RAISE(auto schema, FromArrowSchema(c_arrow_schema, std::nullopt));
+      ICEBERG_ASSIGN_OR_RAISE(auto schema,
+                              InferIcebergSchema(arrow_schema, std::nullopt));
       read_schema_ = std::move(schema);
     }
 

src/iceberg/test/CMakeLists.txt

@@ -165,6 +165,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    USE_BUNDLE
                    SOURCES
                    parquet_data_test.cc
+                   parquet_reader_no_projection_test.cc
                    parquet_schema_test.cc
                    parquet_test.cc)