feat(data): declare a stable physical Arrow schema on task streams

src/iceberg/arrow_c_data_util.cc

@@ -22,6 +22,7 @@
 #include <memory>
 #include <mutex>
 #include <span>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -221,9 +222,7 @@ Status AppendValue(const ArrowSchema& input_schema, const ArrowArray& input_arra
 }  // namespace
 
 ProjectionContext::ProjectionContext(ProjectionContext&& other) noexcept
-    : input_schema_(std::exchange(other.input_schema_, nullptr)),
-      output_schema_(std::exchange(other.output_schema_, nullptr)),
-      selected_field_indices_(std::move(other.selected_field_indices_)),
+    : selected_field_indices_(std::move(other.selected_field_indices_)),
       input_arrow_schema_(other.input_arrow_schema_),
       output_arrow_schema_(other.output_arrow_schema_),
       project_batch_function_(std::exchange(other.project_batch_function_, nullptr)),
@@ -244,8 +243,6 @@ ProjectionContext& ProjectionContext::operator=(ProjectionContext&& other) noexc
     output_arrow_schema_.release(&output_arrow_schema_);
   }
 
-  input_schema_ = std::exchange(other.input_schema_, nullptr);
-  output_schema_ = std::exchange(other.output_schema_, nullptr);
   selected_field_indices_ = std::move(other.selected_field_indices_);
   input_arrow_schema_ = other.input_arrow_schema_;
   other.input_arrow_schema_.release = nullptr;
@@ -267,26 +264,53 @@ ProjectionContext::~ProjectionContext() {
 
 Result<ProjectionContext> ProjectionContext::Make(
     const Schema& input_schema, const Schema& output_schema,
+    const ArrowSchema& physical_input_schema,
     ProjectionContext::ProjectBatchFunction project_batch_function) {
+  ArrowSchemaView schema_view;
+  ArrowError error;
+  ArrowErrorInit(&error);
+  ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+      ArrowSchemaViewInit(&schema_view, &physical_input_schema, &error), error);
+  ICEBERG_PRECHECK(schema_view.type == NANOARROW_TYPE_STRUCT,
+                   "Physical input schema must be a struct, but is {}",
+                   ArrowTypeString(schema_view.type));
+  ICEBERG_PRECHECK(
+      std::cmp_equal(physical_input_schema.n_children, input_schema.fields().size()),
+      "Physical input schema must have {} fields to match the logical schema, but has {}",
+      input_schema.fields().size(), physical_input_schema.n_children);
+
   ICEBERG_ASSIGN_OR_RAISE(
       auto selected_field_indices,
       BuildSelectedFieldIndices(input_schema.fields(), output_schema.fields()));
-
   ProjectionContext context;
-  context.input_schema_ = &input_schema;
-  context.output_schema_ = &output_schema;
-  context.selected_field_indices_ = std::move(selected_field_indices);
-  ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(input_schema, &context.input_arrow_schema_));
-  ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(output_schema, &context.output_arrow_schema_));
   context.project_batch_function_ = project_batch_function;
+  ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+      ArrowSchemaDeepCopy(&physical_input_schema, &context.input_arrow_schema_));
+  ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(output_schema, &context.output_arrow_schema_));
 
+  // Replace each logical output field with the physical layout of its selected input
+  // field. The deep copy also carries the physical metadata (e.g. field ids and
+  // extension annotations); only the top-level name and nullability come from
+  // output_schema.
+  const auto& output_fields = output_schema.fields();
+  for (size_t i = 0; i < output_fields.size(); ++i) {
+    auto* output_field = context.output_arrow_schema_.children[i];
+    output_field->release(output_field);
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowSchemaDeepCopy(
+        physical_input_schema.children[selected_field_indices[i]], output_field));
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+        ArrowSchemaSetName(output_field, std::string(output_fields[i].name()).c_str()));
+    // Preserve type-specific flags such as sorted map keys when setting nullability.
+    if (output_fields[i].optional()) {
+      output_field->flags |= ARROW_FLAG_NULLABLE;
+    } else {
+      output_field->flags &= ~ARROW_FLAG_NULLABLE;
+    }
+  }
+  context.selected_field_indices_ = std::move(selected_field_indices);
   return context;
 }
 
-const Schema& ProjectionContext::input_schema() const { return *input_schema_; }
-
-const Schema& ProjectionContext::output_schema() const { return *output_schema_; }
-
 const ArrowSchema& ProjectionContext::input_arrow_schema() const {
   return input_arrow_schema_;
 }

src/iceberg/arrow_c_data_util_internal.h

@@ -61,19 +61,26 @@ class ICEBERG_EXPORT ProjectionContext {
   /// \brief Resolve the registered ProjectBatch implementation.
   static auto ResolveProjectBatchFunction() -> ProjectBatchFunction;
 
-  /// \brief Build reusable projection state for a validated schema pair.
+  /// \brief Build reusable projection state using the reader's physical Arrow layout.
   ///
   /// \param input_schema Schema that describes every input batch.
   /// \param output_schema Final schema and column order requested by the caller.
+  /// \param physical_input_schema Physical fields corresponding to input_schema in
+  ///   the same order. Each selected output field becomes a deep copy of its physical
+  ///   counterpart, preserving the physical layout, nested list offset widths,
+  ///   metadata, and type-specific flags; only the top-level name and nullability come
+  ///   from output_schema.
   /// \param project_batch_function Optional implementation returned by
   ///   ProjectionContext::ResolveProjectBatchFunction, or nullptr to use the nanoarrow
   ///   path.
   /// \note It validates that output_schema selects or reorders complete top-level fields
-  /// by field id. Nested pruning and type changes are rejected. The input_schema and
-  /// output_schema passed to Make must outlive the context. ProjectBatch may lazily
-  /// initialize backend cache; do not share one context across concurrent calls.
+  /// by field id. Nested pruning and type changes are rejected. Schema inputs are only
+  /// used during construction; the context owns its physical Arrow schemas.
+  /// ProjectBatch may lazily initialize backend cache; do not share one context across
+  /// concurrent calls.
   static Result<ProjectionContext> Make(const Schema& input_schema,
                                         const Schema& output_schema,
+                                        const ArrowSchema& physical_input_schema,
                                         ProjectBatchFunction project_batch_function);
 
   ProjectionContext(ProjectionContext&&) noexcept;
@@ -83,10 +90,6 @@ class ICEBERG_EXPORT ProjectionContext {
   ProjectionContext(const ProjectionContext&) = delete;
   ProjectionContext& operator=(const ProjectionContext&) = delete;
 
-  const Schema& input_schema() const;
-
-  const Schema& output_schema() const;
-
   const ArrowSchema& input_arrow_schema() const;
 
   const ArrowSchema& output_arrow_schema() const;
@@ -100,10 +103,6 @@ class ICEBERG_EXPORT ProjectionContext {
  private:
   ProjectionContext() = default;
 
-  // Raw schema pointers are borrowed from caller-owned schemas. FileScanTaskReader
-  // keeps those schema objects alive in the same stream source that owns this context.
-  const Schema* input_schema_ = nullptr;
-  const Schema* output_schema_ = nullptr;
   std::vector<int32_t> selected_field_indices_;
   ArrowSchema input_arrow_schema_{};
   ArrowSchema output_arrow_schema_{};
@@ -237,7 +236,7 @@ ICEBERG_EXPORT Result<ArrowArrayStream> MakeArrowArrayStream(
 /// \param row_indices Zero-based row positions to copy from `input_batch`.
 /// \param projection Reusable schema/projection state created by
 ///   ProjectionContext::Make.
-/// \return A newly owned ArrowArray matching `projection.output_schema()`.
+/// \return A newly owned ArrowArray matching `projection.output_arrow_schema()`.
 ICEBERG_EXPORT Result<ArrowArray> ProjectBatch(ArrowArray* input_batch,
                                                std::span<const int32_t> row_indices,
                                                ProjectionContext& projection);

src/iceberg/data/file_scan_task_reader.cc

@@ -25,13 +25,15 @@
 #include <utility>
 #include <vector>
 
+#include <nanoarrow/nanoarrow.h>
+
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/arrow_c_data_util_internal.h"
 #include "iceberg/data/delete_filter.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/nanoarrow_status_internal.h"
 #include "iceberg/schema.h"
-#include "iceberg/schema_internal.h"
 #include "iceberg/table_scan.h"
 #include "iceberg/util/macros.h"
 
@@ -68,22 +70,12 @@ class MergeOnReadStreamSource {
  public:
   MergeOnReadStreamSource(std::unique_ptr<Reader> reader,
                           std::unique_ptr<DeleteFilter> delete_filter,
-                          std::shared_ptr<::iceberg::Schema> required_schema,
-                          std::shared_ptr<::iceberg::Schema> projected_schema,
-                          ProjectionContext projection_context)
+                          bool project_all_columns, ProjectionContext projection_context)
       : reader_(std::move(reader)),
         delete_filter_(std::move(delete_filter)),
-        required_schema_(std::move(required_schema)),
-        projected_schema_(std::move(projected_schema)),
-        project_all_rows_(required_schema_->SameSchema(*projected_schema_)),
+        project_all_columns_(project_all_columns),
         projection_context_(std::move(projection_context)) {}
 
-  ~MergeOnReadStreamSource() {
-    if (cached_schema_.has_value() && cached_schema_->release != nullptr) {
-      cached_schema_->release(&cached_schema_.value());
-    }
-  }
-
   Status Close() {
     if (reader_ == nullptr) {
       return {};
@@ -92,12 +84,6 @@ class MergeOnReadStreamSource {
   }
 
   Result<std::optional<ArrowArray>> Next() {
-    if (!cached_schema_.has_value()) {
-      // File readers expose one stable Arrow schema for every batch in the stream.
-      ICEBERG_ASSIGN_OR_RAISE(cached_schema_, reader_->Schema());
-    }
-    ArrowSchema& input_arrow_schema = cached_schema_.value();
-
     while (true) {
       ICEBERG_ASSIGN_OR_RAISE(auto next_batch, reader_->Next());
       if (!next_batch.has_value()) {
@@ -107,13 +93,14 @@ class MergeOnReadStreamSource {
       ArrowArray input_batch = std::move(next_batch.value());
       internal::ArrowArrayGuard input_batch_guard(&input_batch);
 
-      ICEBERG_ASSIGN_OR_RAISE(
-          auto alive, delete_filter_->ComputeAliveRows(input_arrow_schema, input_batch));
+      ICEBERG_ASSIGN_OR_RAISE(auto alive,
+                              delete_filter_->ComputeAliveRows(
+                                  projection_context_.input_arrow_schema(), input_batch));
       if (alive.empty()) {
         continue;
       }
 
-      if (alive.alive_count() == input_batch.length && project_all_rows_) {
+      if (alive.alive_count() == input_batch.length && project_all_columns_) {
         // Transfer ownership to the stream result; the local guard must not release it.
         ArrowArray output_batch = input_batch;
         input_batch.release = nullptr;
@@ -125,19 +112,17 @@ class MergeOnReadStreamSource {
   }
 
   Result<ArrowSchema> Schema() {
-    ArrowSchema schema;
-    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*projected_schema_, &schema));
+    ArrowSchema schema{};
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+        ArrowSchemaDeepCopy(&projection_context_.output_arrow_schema(), &schema));
     return schema;
   }
 
  private:
   std::unique_ptr<Reader> reader_;
   std::unique_ptr<DeleteFilter> delete_filter_;
-  std::shared_ptr<::iceberg::Schema> required_schema_;
-  std::shared_ptr<::iceberg::Schema> projected_schema_;
-  bool project_all_rows_ = false;
+  bool project_all_columns_ = false;
   ProjectionContext projection_context_;
-  std::optional<ArrowSchema> cached_schema_;
 };
 
 }  // namespace
@@ -209,21 +194,25 @@ class FileScanTaskReader::Impl {
                            io_, field_lookup_, has_position_deletes, delete_counter_));
 
     auto required_schema = delete_filter->RequiredSchema();
-    auto project_batch_function = ProjectionContext::ResolveProjectBatchFunction();
-    // ProjectionContext borrows schemas that are kept in MergeOnReadStreamSource.
-    ICEBERG_ASSIGN_OR_RAISE(auto projection_context,
-                            ProjectionContext::Make(*required_schema, *projected_schema_,
-                                                    project_batch_function));
-
     auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(),
                                      name_mapping_, properties_, data_file->first_row_id,
                                      data_file->data_sequence_number);
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
                             ReaderFactoryRegistry::Open(data_file->file_format, options));
 
+    // Resolve the physical layout before emitting any batch. Both pass-through and
+    // projected batches must use the same types, including nested list offset widths.
+    ICEBERG_ASSIGN_OR_RAISE(auto input_arrow_schema, reader->Schema());
+    internal::ArrowSchemaGuard input_schema_guard(&input_arrow_schema);
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto projection_context,
+        ProjectionContext::Make(*required_schema, *projected_schema_, input_arrow_schema,
+                                ProjectionContext::ResolveProjectBatchFunction()));
+
+    const bool project_all_columns = required_schema->SameSchema(*projected_schema_);
     auto mor_reader = std::make_unique<MergeOnReadStreamSource>(
-        std::move(reader), std::move(delete_filter), std::move(required_schema),
-        projected_schema_, std::move(projection_context));
+        std::move(reader), std::move(delete_filter), project_all_columns,
+        std::move(projection_context));
     return MakeArrowArrayStream(std::move(mor_reader));
   }
 

src/iceberg/data/file_scan_task_reader.h

@@ -38,7 +38,9 @@ namespace iceberg {
 ///
 /// FileScanTaskReader chooses the copy-on-write path for tasks without deletes and
 /// the merge-on-read path for tasks with v2 position or equality deletes. The returned
-/// stream always exposes `projected_schema`.
+/// stream exposes the fields of `projected_schema` in one stable physical Arrow layout,
+/// which may depend on the file format and reader properties (for example, `large_list`
+/// instead of `list`). Every batch matches the schema reported by the stream.
 ///
 /// TODO(gangwu): Add a mode that emits a `_deleted` column instead of filtering rows.
 /// TODO(gangwu): Use evaluator to apply residual expression filters.