[kamcheungting-db]

Problem

A FileScanTaskReader stream uses the merge-on-read path whenever a task carries v2 position or equality deletes. Until now that stream declared its schema by converting the logical Iceberg projected schema with ToArrowSchema(), which always yields the canonical Arrow layout (e.g. 32-bit list).

The batches it emits, however, follow the physical layout produced by the file reader, which can legitimately differ:

• ReaderProperties::kArrowUseLargeList asks the Parquet reader for large_list (64-bit offsets),
• files written with ARROW:schema metadata can store a mix of list and large_list,
• columns missing from the file (schema evolution) are reconstructed with the configured preference.

So one and the same stream could emit two different physical layouts at once:

BEFORE (merge-on-read stream, reader configured for large_list)

  stream->get_schema()         ToArrowSchema(projected_schema)  -> "list"       (32-bit)
  batch #1, all rows alive     pass-through from the reader     -> "large_list" (64-bit)
  batch #2, some rows deleted  ProjectBatch, output built from  -> "list"       (32-bit)
                               ToArrowSchema(projected_schema)

  The declaration disagrees with pass-through batches, and batches disagree
  with each other: importing batch #1 against the declared schema reads
  64-bit offsets as 32-bit ones.

What this PR does

ProjectionContext — the reusable state behind ProjectBatch — is now built from the reader's physical Arrow schema and owns deep copies of both its input and output Arrow schemas. The stream declares exactly the schema its batches are made of, no matter which path produces them.

How it works

                 FileScanTaskReader::Open(task with deletes)
                                 │
          ┌──────────────────────┴───────────────────────┐
          ▼                                            ▼
  DeleteFilter::Make()                        ReaderFactoryRegistry::Open()
          │                                            │
          ▼                                            ▼
  required_schema                             reader->Schema()   ← PHYSICAL Arrow schema
  (projected columns +                        (list vs large_list from reader
   equality-delete columns + _pos)             properties, ARROW:schema overrides,
          │                                   reconstructed missing columns)
          │                                   │
          └────────────────┬──────────────────┘
                           ▼
  ProjectionContext::Make(required_schema, projected_schema, physical_schema)
                           │
         ┌─────────────────┼──────────────────────┐
         ▼                 ▼                      ▼
  input_arrow_schema_  output_arrow_schema_  selected_field_indices_
  (owned deep copy    (owned; per output     (output field → input
   of the physical     field: deep copy of    column, by field id)
   schema)             the selected physical
                      child, then top-level
                      name + nullability
                      from projected_schema)
                           │
                           ▼
      MergeOnReadStreamSource  (owns the ProjectionContext)
                           │
      for each batch from the reader:
        ComputeAliveRows(input_arrow_schema_, batch)
                           │
         ┌─────────────────┴────────────────────┐
         │ all rows alive AND                   │ otherwise
         │ required_schema == projected_schema   │
         ▼                                     ▼
    PASS-THROUGH                            ProjectBatch(batch, alive rows):
    (hand the reader's                      copy the selected columns/rows into
     ArrowArray to the stream               arrays built from output_arrow_schema_
     unchanged)
         └─────────────────┬────────────────────┘
                           ▼
   stream->get_schema() = deep copy of output_arrow_schema_
   ⇒ every emitted batch — pass-through or projected — carries the
     exact physical layout the stream declares

Key changes

• ProjectionContext::Make now takes the physical input ArrowSchema (the old 3-argument overload is removed) and:
  • validates the physical schema is a struct with one child per logical input field, in the same order;
  • stores input_arrow_schema_ as an owned deep copy of it;
  • builds output_arrow_schema_ by deep-copying, for each output field, the selected physical child and then applying only the top-level name and nullability from the projected schema — the physical layout, nested list offset widths, metadata (field ids, extension annotations) and type-specific flags (e.g. sorted map keys) are preserved;
  • keeps the existing by-field-id selection rules (nested pruning and type changes are still rejected).
• The borrowed Schema* members and the input_schema()/output_schema() accessors are gone — schema inputs are only read during construction.
• FileScanTaskReader::Open resolves reader->Schema() once, before the first batch, so pass-through and projected batches derive from the same layout, and schema errors surface at Open instead of mid-stream. The pass-through flag is renamed project_all_columns to say what it actually means.
• MergeOnReadStreamSource no longer caches reader_->Schema() nor stores schema shared_ptrs; it uses the context's owned input schema for ComputeAliveRows, and Schema() returns an independent deep copy on every call (results stay valid after the stream is released).

There is no public API change: ProjectionContext is an internal utility, and the FileScanTaskReader behavior change is the guarantee itself — every batch now matches the schema the stream reports.

Testing

• arrow_c_data_util_test.cc — all existing projection tests now run through the production path with a physical schema, on both backends (nanoarrow and Arrow compute). New cases:
  • PreservesPhysicalListLayoutsWithBothBackends — mixed nested list layouts survive selection, reordering and renaming;
  • PreservesPhysicalMapFlagsWhenChangingNullability — keys_sorted survives a nullability change;
  • EmptyProjectionPreservesRowCount — zero-column projection keeps the alive row count;
  • validation tests for the physical-schema contract (must be a struct, must match the logical field count).
• file_scan_task_reader_test.cc:
  • EmptyProjectionPreservesLiveRowCount — position deletes plus a zero-column projection still streams the correct number of live rows;
  • FileScanTaskPhysicalSchemaTest — 36 scenarios: 6 delete-batch selections (delete first row / a second-batch row / no-op / whole first batch / all rows / empty split) × {all columns, pruned} × {list, large_list, stored mixed schema}. Each scenario checks the declared schema before, between and after batches; imports every batch against it; destroys the task reader before consuming the stream; and re-validates the retained batches after the stream is released.
• data_test: 231 tests pass with both projection backends; pre-commit (clang-format included) is clean.

Related

• fix: preserve list types for empty Parquet splits #1040 aligned the Parquet reader's own output schema — that already made the no-deletes path (which streams the reader directly) physically stable, including empty splits. This PR extends the same guarantee to the merge-on-read path in the task reader.: 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 such as sorted map keys, while only top-level names and nullability come from the projected schema. The borrowed logical-schema pointers and their accessors are removed, so schema inputs are only used during construction.

FileScanTaskReader::Open resolves the physical schema before emitting any batch so pass-through, filtered, and projected batches all match the schema reported by the stream, including files that mix list and large_list layouts.

Tests cover both projection backends (nanoarrow and Arrow compute): mixed list layouts, sorted-map flags under nullability changes, zero-column projections, and 36 task-reader scenarios that destroy the reader before consuming the stream and validate retained batches after the stream is released.