fix(parquet): keep output schema consistent with produced list arrays

Arrow honors set_list_type(LARGE_LIST) only when it derives the Arrow schema from the
Parquet schema. When a file carries serialized ARROW:schema metadata, the reader keeps
producing plain list arrays, but the output schema was rewritten to large_list
unconditionally. ProjectRecordBatch then built the projected batch against a large_list
schema while the incoming arrays were list arrays, which casts a ListArray to a
LargeListArray.

The output schema is the target of the projection, so it keeps being derived from the
projected Iceberg schema, and the large_list rewrite is now applied only when the reader
actually produces large lists. Adds a regression test that reads a file written through
parquet::arrow::WriteTable, which serializes ARROW:schema, with use-large-list enabled.

Also addresses review comments:
- comment the forward declaration of UseLargeListField
- extract the duplicated field rewriting into UseLargeListFields

Author: rahulsmahadev

src/iceberg/parquet/parquet_reader.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/parquet/parquet_reader.h"
 
+#include <algorithm>
 #include <numeric>
 
 #include <arrow/c/bridge.h>
@@ -85,6 +86,7 @@ class EmptyRecordBatchReader : public ::arrow::RecordBatchReader {
   }
 };
 
+// forward declaration to unblock cycle dependence.
 std::shared_ptr<::arrow::Field> UseLargeListField(
     const std::shared_ptr<::arrow::Field>& field);
 
@@ -120,6 +122,42 @@ std::shared_ptr<::arrow::Field> UseLargeListField(
   return field->WithType(UseLargeListType(field->type()));
 }
 
+// Rewrite all fields in a field vector to use large_list instead of list.
+::arrow::FieldVector UseLargeListFields(const ::arrow::FieldVector& fields) {
+  ::arrow::FieldVector rewritten;
+  rewritten.reserve(fields.size());
+  for (const auto& field : fields) {
+    rewritten.push_back(UseLargeListField(field));
+  }
+  return rewritten;
+}
+
+// Returns true if the type contains a large_list, at any level of nesting.
+bool ContainsLargeList(const ::arrow::DataType& type) {
+  if (type.id() == ::arrow::Type::LARGE_LIST) {
+    return true;
+  }
+  return std::ranges::any_of(
+      type.fields(), [](const auto& field) { return ContainsLargeList(*field->type()); });
+}
+
+// Returns true if the reader produces large_list arrays.
+//
+// Arrow honors the requested large_list type only when it derives the Arrow schema from
+// the Parquet schema. A file that carries serialized ARROW:schema metadata keeps its
+// original list type instead, so whether large lists are produced can only be told from
+// the schema of the reader.
+bool ProducesLargeList(const ::arrow::RecordBatchReader& reader) {
+  const auto& schema = reader.schema();
+  if (schema == nullptr) {
+    // an empty reader produces no arrays to be described
+    return false;
+  }
+  return std::ranges::any_of(schema->fields(), [](const auto& field) {
+    return ContainsLargeList(*field->type());
+  });
+}
+
 }  // namespace
 
 // A stateful context to keep track of the reading progress.
@@ -252,23 +290,6 @@ class ParquetReader::Impl {
   Status InitReadContext() {
     context_ = std::make_unique<ReadContext>();
 
-    // Build the output Arrow schema
-    ArrowSchema arrow_schema;
-    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_,
-                                   ::arrow::ImportSchema(&arrow_schema));
-    if (use_large_list_) {
-      // Align the output schema with the large_list arrays produced by the
-      // Parquet reader when kArrowUseLargeList is enabled.
-      ::arrow::FieldVector fields;
-      fields.reserve(context_->output_arrow_schema_->fields().size());
-      for (const auto& field : context_->output_arrow_schema_->fields()) {
-        fields.push_back(UseLargeListField(field));
-      }
-      context_->output_arrow_schema_ =
-          ::arrow::schema(std::move(fields), context_->output_arrow_schema_->metadata());
-    }
-
     // Row group pruning based on the split
     // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc.
     std::vector<int> row_group_indices;
@@ -301,6 +322,24 @@ class ParquetReader::Impl {
           reader_->GetRecordBatchReader(row_group_indices, column_indices));
     }
 
+    // Build the output Arrow schema from the projected Iceberg schema. This schema is the
+    // target of ProjectRecordBatch, so it must describe the projected schema rather than
+    // the schema of the file.
+    ArrowSchema arrow_schema;
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_,
+                                   ::arrow::ImportSchema(&arrow_schema));
+
+    if (use_large_list_ && ProducesLargeList(*context_->record_batch_reader_)) {
+      // Align the output schema with the large_list arrays produced by the Parquet
+      // reader. Note that Arrow ignores the requested list type when the file carries
+      // serialized ARROW:schema metadata, in which case the reader keeps producing plain
+      // list arrays and the output schema must keep describing them as such.
+      context_->output_arrow_schema_ =
+          ::arrow::schema(UseLargeListFields(context_->output_arrow_schema_->fields()),
+                          context_->output_arrow_schema_->metadata());
+    }
+
     return {};
   }
 

src/iceberg/test/parquet_test.cc

@@ -293,6 +293,38 @@ class ParquetReaderTest : public TempFileTestBase {
                                         .data_sequence_number = data_sequence_number});
   }
 
+  // Writes a list parquet file through parquet::arrow::WriteTable, which serializes the
+  // Arrow schema of the table into the ARROW:schema key value metadata of the file.
+  void CreateListParquetFileWithArrowSchema() {
+    const std::string kParquetFieldIdKey = "PARQUET:field_id";
+    auto arrow_schema = ::arrow::schema(
+        {::arrow::field("id", ::arrow::int32(), /*nullable=*/false,
+                        ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey}, {"1"})),
+         ::arrow::field(
+             "numbers",
+             ::arrow::list(::arrow::field(
+                 "element", ::arrow::int32(), /*nullable=*/true,
+                 ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey}, {"101"}))),
+             /*nullable=*/true,
+             ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey}, {"2"}))});
+    auto batch =
+        ::arrow::RecordBatch::FromStructArray(
+            ::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_schema->fields()),
+                                               R"([[1, [1, 2]], [2, [3]]])")
+                .ValueOrDie())
+            .ValueOrDie();
+    auto table = ::arrow::Table::FromRecordBatches(arrow_schema, {batch}).ValueOrDie();
+
+    auto io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
+    auto outfile = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie();
+
+    // write a single row group so that one batch holds every row
+    ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(),
+                                             outfile, table->num_rows())
+                    .ok());
+    ASSERT_TRUE(outfile->Close().ok());
+  }
+
   void VerifyNextBatch(Reader& reader, std::string_view expected_json) {
     // Boilerplate to get Arrow schema
     auto schema_result = reader.Schema();
@@ -552,6 +584,62 @@ TEST_F(ParquetReaderTest, ReadListAsLargeList) {
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
 }
 
+TEST_F(ParquetReaderTest, ReadListAsLargeListWithArrowSchema) {
+  // A file written with serialized ARROW:schema metadata keeps its original list type, as
+  // Arrow ignores the requested large_list type in that case. The output schema must keep
+  // describing the arrays that are actually produced, otherwise projecting the record
+  // batch casts a list array to a large_list array.
+  CreateListParquetFileWithArrowSchema();
+
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(2, "numbers",
+                                std::make_shared<ListType>(SchemaField::MakeOptional(
+                                    /*field_id=*/101, "element", int32())))});
+
+  ReaderProperties reader_properties;
+  reader_properties.Set(ReaderProperties::kArrowUseLargeList, true);
+
+  auto reader_result = ReaderFactoryRegistry::Open(
+      FileFormatType::kParquet, {.path = temp_parquet_file_,
+                                 .io = file_io_,
+                                 .projection = schema,
+                                 .properties = std::move(reader_properties)});
+  ASSERT_THAT(reader_result, IsOk());
+  auto reader = std::move(reader_result.value());
+
+  auto schema_result = reader->Schema();
+  ASSERT_THAT(schema_result, IsOk());
+  auto arrow_c_schema = std::move(schema_result.value());
+  auto arrow_type = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+  ASSERT_EQ(arrow_type->field(1)->type()->id(), ::arrow::Type::LIST);
+
+  auto data = reader->Next();
+  ASSERT_THAT(data, IsOk());
+  ASSERT_TRUE(data.value().has_value());
+  auto arrow_c_array = data.value().value();
+
+  // Importing the array against the reported schema fails if the two disagree.
+  auto import_result = ::arrow::ImportArray(&arrow_c_array, arrow_type);
+  ASSERT_TRUE(import_result.ok()) << import_result.status().ToString();
+  auto arrow_array = import_result.ValueOrDie();
+  ASSERT_TRUE(arrow_array->ValidateFull().ok());
+
+  const auto& struct_array =
+      internal::checked_cast<const ::arrow::StructArray&>(*arrow_array);
+  ASSERT_EQ(struct_array.length(), 2);
+
+  const auto& id_array =
+      internal::checked_cast<const ::arrow::Int32Array&>(*struct_array.field(0));
+  ASSERT_EQ(id_array.Value(0), 1);
+  ASSERT_EQ(id_array.Value(1), 2);
+
+  const auto& numbers_array =
+      internal::checked_cast<const ::arrow::ListArray&>(*struct_array.field(1));
+  ASSERT_EQ(numbers_array.value_slice(0)->length(), 2);
+  ASSERT_EQ(numbers_array.value_slice(1)->length(), 1);
+}
+
 TEST_F(ParquetReaderTest, ReadSplit) {
   CreateSplitParquetFile();