[core][common]support row range read of a DataSplit

[weijietong]

Purpose

This is to solve issue.

Tests

tests for pk table,append table, DE table.

[JingsongLi]

Why not just use IndexedSplit?

[weijietong]

The critical RowRange use case is slicing the stream after a filter — e.g. "give me the 1000th–2000th matching rows for training." Even composed with a filter, IndexedSplit only ANDs two physical-position bitmaps; it never renumbers rows into a
post-filter effective sequence, so it cannot express "the N-th row after filtering."

This is exactly what AbstractDataTableRead.outerWrap encodes:

• RowRange + filter (executeFilter): outerWrap=true — RowRange wraps outside the filter, slicing the filtered effective stream.
• IndexedSplit + filter: both reduce to physical-position bitmaps ANDed together — no "post-filter renumbering" step.

The AI training-shard use case (why RowRange exists)

In AI / ML data loading, a common pattern is sharding a dataset by effective sample position:

• A training job with worldSize=W, rank=R wants its local shard — the rows at effective positions [R*N, (R+1)*N - 1] of the dataset.
• With filters active (e.g. quality filters, column predicates) or deletion vectors, the "effective row order" is the filtered/compacted sequence. Each worker must receive a deterministic, contiguous slice of that sequence, not of the raw file layout.
• RowRange models exactly this: a single contiguous [start, end] in the 0-based effective-row space of a split. The reader either pushes it down as a selection bitmap (parquet, full-scan — no extra rows decoded) or wraps a single RangeSkipReader over
  the filtered output (filter/DV/ORC/merge-tree) to skip start and take count.

IndexedSplit cannot express this:

• It cuts physical row-ids, which have no relation to the post-filter effective order — worker rank=R wouldn't get a contiguous slice of matching rows.
• It requires firstRowId (so plain append tables are out) or a pre-built global index.
• For merge-tree primary-key tables, physical position ≠ output order, so the shard would be scrambled.

Conclusion

IndexedSplit and RowRange are complementary, not interchangeable:

• IndexedSplit: read a discrete set of physical row-ids (vector search / global-index hits), AND-composed with filters in the physical-position space, requires row-ids or a pre-built index, excludes append tables / post-filter slicing / merge-output
  slicing.
• RowRange: read a contiguous slice in effective-row space (AI training shards / pagination / range queries), applied as skip+limit over the filtered effective stream, works across append / primary-key / DE tables, no pre-built index needed.

RowRange was introduced precisely to cover IndexedSplit's blind spots — append tables, post-filter effective-row slicing, and merge-output slicing — all validated by tests in this PR (testAppendOrcRowRange..., testRowRangeWithFilterWrapsOutsideFilter,
testPrimaryKeyRowRangeOverMergeRead). They cover behavior IndexedSplit cannot express, so RowRange cannot be replaced by it.

[JingsongLi]

The post-filter/merge-output slicing use case is distinct from IndexedSplit's physical row IDs and has end-to-end value. Two fallback paths currently violate the requested range; both were reproduced using real table writes and reads, and both passed after focused conditional fixes. Details are inline.

The existing Parquet DE range test also passes. The new failures specifically cover ORC column-merge reads and a supported file-format change within an append table.

[JingsongLi]

The two issues from my previous review are fixed. Data-evolution reads now construct the row-range bitmap only when the underlying reader can apply it, avoiding the ORC double-skip; raw reads check every file format before choosing pushdown, so a Parquet-to-ORC mixed split retains the required fallback. I re-ran both real write/read reproducers against the updated production classes, and 32 distinct focused tests passed across the range utilities, append/raw reads and data-evolution reads. This has concrete value for range-based training reads described in #9668, and I found no remaining actionable regression in the declared full-data DataSplit scope. Validation used isolated JDK 8 compilation and local cached dependencies, not a full Maven build; the visible CI failure in S3FileIOTest is a container-fetch failure outside this diff.

[JingsongLi]

Could you clarify the concrete purpose of this PR? Is there an actual production workload that needs this Java range-read API?

The AI training motivation alone does not explain the Java-side requirement: PyPaimon's training read path uses its own Python/PyArrow implementation. Please describe the production application, how its training data loader calls the Java reader, and the specific bottleneck this change addresses.

If there is no actual production use case for this API, please close this PR.

[weijietong]

In our use case, the training framework calculates the total number of rows within a partition based on a snapshot and determines the number of row ranges to be read for each data split. Without this new API, the initial implementation would have relied on the framework obtaining row-by-row iteration via RecordReader<InternalRow> createReader(Split split) and then manually implementing functionality similar to a RangeSkipReader. Given our Parquet-based storage format—specifically when dealing with large Parquet files but reading only a small number of rows defined by the row ranges—leveraging the new API and row-range pushdown optimization allows us to achieve a significant reduction in training time compared to the previous 40-minute duration.

Note: We did not use PyPaimon.

[weijietong]

I am not necessarily advocating that this PR must be merged; I simply feel that other training workflows would likely require this same interface and performance optimization. Without this API, our training framework would be forced to implement—in an intrusive manner—custom read logic (handling primary key tables, append-only tables, and DE tables) to process raw files stored on OSS, push down row ranges, and accelerate read speeds; such an approach would lack maintainability and extensibility. However, if you do not consider this a general-purpose requirement, I am happy to close the PR.

[JingsongLi]

Reviewed b6b9ae1. Requirement fit: SUPPORTED for the clarified Java training-reader caller. Implementation: FINDINGS.

The latest explanation identifies a Java RecordReader consumer over Parquet training ranges, so efficient bounded reads have a concrete end-to-end use. The earlier ORC double-slicing and mixed-format fallback findings remain fixed. A new supported fallback case still violates the effective-row contract: ignoring a missing file leaves metadata-based offsets counting rows that the reader never emits. The real write/read reproduction is inline.

The head is unchanged from the prior 32 passing focused tests; I verified the relevant production sources match that tested head. A new exact-head missing-file probe fails with [] versus [6,7], and disabling metadata-based pushdown when files may be ignored makes it pass. No training throughput benchmark was rerun, and the current CI rollup remains failing.

[weijietong]

Although the ORC reader cannot perform page-level row filtering like Parquet to return precise row ranges, it can still ultimately yield precise row ranges through the outermost encapsulation by ApplyBitmapIndexFileRecordIterator. Therefore, the ORC format supports row range pushdown for accelerated reads.

[weijietong]

The CI test case seems always fail. The failure is caused by Testcontainers being unable to pull the MinIO image (quay.io/minio/minio:RELEASE.2022-02-07T08-17-33Z), so S3FileIOTest fails before running any assertions. Can you solve it ? @JingsongLi