feat(reader): support late materialization with probe/payload two-phase reads

[zhf999]

This PR is inspired by #196, thanks @gripleaf for the initial work.

Purpose

This PR introduces a late-materializing file batch reader (LateMaterializingFileBatchReader)
that performs probe/payload two-phase reads when a predicate is pushed down through
SetReadSchema. Without a predicate the reader degrades to a plain passthrough, so it is safe to
install unconditionally.

Motivation. In a standard read, all projected columns are materialized together, even though a
pushed-down predicate may filter out the vast majority of rows. Late materialization splits the
read into two passes:

1. Probe pass — read only the predicate columns, evaluate the predicate batch by batch, and
   build a matched_bitmap_ of surviving rows.
2. Payload pass — read the remaining (payload) columns, compacted to only the matched rows,
   then assemble the probe and payload columns into a single struct array in full_schema_ field
   order.

This avoids decoding and holding the wide payload columns for rows that will be discarded.

Architecture. The reader is installed below the prefetch layer by wrapping the
format-specific ReaderBuilder with a LateMaterializingReaderBuilder
(AbstractSplitRead::CreateFileBatchReader). Each parallel reader under the prefetch layer thus
gets its own late-materializing wrapper. The reader implements the full
PrefetchFileBatchReader interface, forwarding schema, row-count, seek, read-range, and metric
queries to the inner reader while interposing the two-phase logic in SetReadSchema /
NextBatch.

The feature is controlled by ReadContextBuilder::EnableLateMaterializing(bool) (default false).
When disabled, no wrapping occurs and the read path is unchanged.

TODOs

Key files:

• src/paimon/common/reader/late_materializing_file_batch_reader.{h,cpp} — core reader with a
  state machine (kInit → kProbing → kRunning | kNoLatMat → kEOF), probe data filtering, payload
  batch reading with bitmap compaction, and full-batch assembly.
• src/paimon/common/reader/late_materializing_reader_builder.h — ReaderBuilder wrapper that
  installs the late-materializing reader around each format reader produced by the inner builder.
• src/paimon/core/operation/abstract_split_read.{h,cpp} — integration: the reader builder is
  wrapped with LateMaterializingReaderBuilder inside CreateFileBatchReader.
• src/paimon/format/parquet/parquet_file_batch_reader.{h,cpp} — fixed read-range loss after
  reentrant SetReadSchema by caching and re-applying read ranges.
• src/paimon/common/utils/arrow/arrow_utils.{h,cpp} — added ArrowUtils::NormalizeArrayOffsets()
  helper for post-slice Arrow array offset normalization.
• src/paimon/testing/mock/mock_file_batch_reader.h — enhanced to faithfully emulate a real
  format reader: SetReadSchema now resets position and read ranges; NextBatchWithBitmap
  honours assigned read ranges; a ProjectBatch helper supports column re-selection.
• src/paimon/CMakeLists.txt — registers the new source and test files.

Safety guards:

• If probe-filter binding fails (e.g. type mismatch between predicate and schema), SetReadSchema
  returns Status::Invalid to surface the configuration error early.
• If SetReadSchema has not been called before NextBatch, the reader transitions directly to
  kNoLatMat, matching the FileBatchReader contract for schema-less reads.

Tests

Unit tests. A dedicated test suite LateMaterializingFileBatchReaderTest is added in
src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp (16 cases):

Test case	Verifies
PassThroughWhenNoPredicate	No predicate → passthrough, single batch returned
PassThroughWhenPayloadEmpty	All fields are probe fields → passthrough
ContiguousSubsetAcrossBatches	Contiguous matched rows spanning multiple batches
ScatteredAlternatingMatch	Non-contiguous alternating matches compacted correctly
MatchedIntersectsSelection	Predicate bitmap intersected with selection_bitmap
EmptyMatchReturnsEof	Zero matched rows → immediate EOF
SeekToRowRealignsProbeCursor	SeekToRow realigns the probe cursor and state
ReadRangesForwardedAcrossPhases	Read ranges forwarded to inner reader across phases
ReentrantSetReadSchema	Repeated SetReadSchema resets state cleanly
ForwardsRowCountAndFileSchema	GetNumberOfRows / GetFileSchema forwarded
MultiFieldPreservesColumnOrder	Multi-field schema preserves field order in assembled batch
NestedPayloadColumn	Nested-type payload column handled correctly
FailsOnPredicateTypeMismatch	Predicate type mismatch returns error instead of silent fallback
WorksAsInnerOfPrefetchReader	Works as inner reader of PrefetchFileBatchReaderImpl
PrefetchInnerReentrantSetReadSchema	Reentrant SetReadSchema under prefetch
PrefetchInnerParallelReadersWithSeek	Parallel prefetch readers with seek interleave correctly

Integration tests:

• test/inte/read_inte_test.cpp — new LM-enabled read tests with and without predicates.
• test/inte/scan_and_read_inte_test.cpp — new LM-enabled scan-and-read tests.
• test/inte/read_inte_with_index_test.cpp — TestBitmapIndexWithLateMaterializing.
• src/paimon/core/operation/merge_file_split_read_test.cpp — TestReadWithPredicateAndLateMaterializing.
• Existing tests incompatible with LM (blob table, global index) explicitly disable it via
  EnableLateMaterializing(false).

Known Limitations / TODOs

1. ORC SetReadRanges is a no-op — the ORC reader currently ignores read ranges and always
   reads the entire file, causing read amplification in the payload pass.
2. PrefetchFileBatchReader only pre-buffers probe columns — the prefetch layer currently
   issues PreBufferRange only for the probe schema; payload columns are read synchronously on
   demand without pre-buffering.
3. Predicate evaluation not SIMD-optimized — row filtering currently uses
   PredicateFilter::Test; a future optimization will switch to arrow::compute::Filter which
   supports SIMD-accelerated batch filtering.

API and Format

No storage format or protocol change.

Public API changes (include/paimon/):

• PrefetchFileBatchReader::GetNextRowToRead() — return type changed from uint64_t to
  Result<uint64_t> to propagate errors from seek operations. All implementors updated
  (Parquet, ORC, PrefetchImpl, LateMat).
• ReadContext — added EnableLateMaterializing() const getter and
  ReadContextBuilder::EnableLateMaterializing(bool) builder method (default: false).

Internal signature changes in the split-read path:

• AbstractSplitRead::CreateFileBatchReader — const ReaderBuilder* →
  std::unique_ptr<ReaderBuilder> (ownership transfer to allow wrapping).
• AbstractSplitRead::CreateFieldMappingReader — same change for the reader_builder parameter.

Both are private methods of AbstractSplitRead; all call sites within the class are updated in the
same change.

Documentation

No user-facing documentation change yet. User-facing documentation for the late-materialization
read mode will be added in a follow-up once the feature stabilizes.

Generative AI tooling

Generated-by: Qoder

[lxy-9602]

+1