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

paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapIndexResult.java

@@ -63,6 +63,23 @@ public FileIndexResult limit(int limit) {
         return new BitmapIndexResult(() -> get().limit(limit));
     }
 
+    /**
+     * Intersects this selection with the physical row range {@code [startInclusive, endInclusive]}
+     * (both endpoints inclusive), used to push down a contiguous row range read (range query).
+     *
+     * <p>{@code endInclusive} may be {@link Long#MAX_VALUE} to denote "until the end of the file".
+     * Note this operates on the physical row index space; effective-row (deletion-vector-aware)
+     * mapping is handled at a higher layer.
+     */
+    public FileIndexResult range(long startInclusive, long endInclusive) {
+        // RoaringBitmap32.bitmapOfRange is half-open [min, max), so +1 for inclusive end.
+        final long max = (endInclusive == Long.MAX_VALUE) ? Long.MAX_VALUE : endInclusive + 1;
+        return new BitmapIndexResult(
+                () ->
+                        RoaringBitmap32.and(
+                                get(), RoaringBitmap32.bitmapOfRange(startInclusive, max)));
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java

@@ -43,6 +43,18 @@ default FileRecordReader<InternalRow> createReader(Context context, long offset,
                         getClass().getName()));
     }
 
+    /**
+     * Whether this format supports row-range / row-group level skipping by consuming the {@link
+     * Context#selection()} bitmap.
+     *
+     * <p>When {@code false}, a higher layer falls back to a naive skip+limit over the read output
+     * stream to guarantee range-read correctness. Default is {@code false} for backward
+     * compatibility; parquet/orc override to {@code true}.
+     */
+    default boolean supportsRowRangeSkip() {
+        return false;
+    }
+
     /** Context for creating reader. */
     interface Context {
 

paimon-common/src/main/java/org/apache/paimon/predicate/RowRange.java

@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * A range of rows to read from a single data split, expressed as a 0-based global effective-row
+ * position range {@code [startInclusive, endInclusive]} (both endpoints inclusive) across the
+ * concatenated effective rows of all files in the split.
+ *
+ * <p>"Effective row" means the row is actually returned (deletion-vector-deleted rows are not
+ * counted). When there is no deletion vector, the effective-row position equals the physical row
+ * index. When a deletion vector exists, the reader maps effective-row positions to physical row
+ * indices internally (see {@code RangeSkipReader} / selection bitmap), so the user always reasons
+ * in the continuous effective-row space.
+ *
+ * <p>This is used to support range queries that read a contiguous slice of rows (e.g. AI training
+ * data slicing by effective sample count). The reader skips row groups / pages outside the range
+ * without decoding them when the underlying format supports it; otherwise it falls back to a naive
+ * skip+limit over the output stream.
+ *
+ * <p>The semantics is the 0-based effective-row position space, <b>not</b> the logical {@code
+ * firstRowId} space and <b>not</b> the row-id space used by {@code IndexedSplit.rowRanges()}.
+ * {@code ROW_ID = firstRowId + returnedPosition()} is unaffected.
+ *
+ * <p>Note: this is distinct from {@link org.apache.paimon.utils.Range}, which is a row-id interval
+ * used by the global-index / {@code IndexedSplit} path. {@code RowRange} is an effective-row
+ * position interval used by range-query reads.
+ *
+ * @since 1.1.0
+ */
+public final class RowRange implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /** From the 0th effective row to the end of the split (no upper bound). */
+    public static final RowRange FULL = new RowRange(0, Long.MAX_VALUE);
+
+    /** Empty range: used as a per-file intersection result meaning "skip this file". */
+    public static final RowRange EMPTY = new RowRange(0, -1);
+
+    private final long startInclusive;
+    private final long endInclusive;
+
+    private RowRange(long startInclusive, long endInclusive) {
+        this.startInclusive = startInclusive;
+        this.endInclusive = endInclusive;
+    }
+
+    /**
+     * Creates a row range {@code [startInclusive, endInclusive]} (both endpoints inclusive).
+     *
+     * @param startInclusive the first effective-row position to read (inclusive), must be {@code >=
+     *     0}
+     * @param endInclusive the last effective-row position to read (inclusive), must be {@code >=
+     *     startInclusive}; {@link Long#MAX_VALUE} means "to the end of the split"
+     */
+    public static RowRange of(long startInclusive, long endInclusive) {
+        if (startInclusive < 0) {
+            throw new IllegalArgumentException(
+                    "startInclusive must be >= 0, but is " + startInclusive);
+        }
+        if (endInclusive != Long.MAX_VALUE && endInclusive < startInclusive) {
+            throw new IllegalArgumentException(
+                    "endInclusive must be >= startInclusive (or Long.MAX_VALUE), but startInclusive="
+                            + startInclusive
+                            + ", endInclusive="
+                            + endInclusive);
+        }
+        return new RowRange(startInclusive, endInclusive);
+    }
+
+    public long startInclusive() {
+        return startInclusive;
+    }
+
+    public long endInclusive() {
+        return endInclusive;
+    }
+
+    /** Whether this range selects no rows at all. */
+    public boolean isEmpty() {
+        return endInclusive < startInclusive;
+    }
+
+    /**
+     * Number of rows covered by this range ({@code end - start + 1}); {@link Long#MAX_VALUE} if
+     * unbounded.
+     */
+    public long count() {
+        if (endInclusive == Long.MAX_VALUE) {
+            return Long.MAX_VALUE;
+        }
+        return endInclusive - startInclusive + 1;
+    }
+
+    /**
+     * Translate a split-global {@link RowRange} into a per-file local {@link RowRange} over the
+     * file's own effective-row space.
+     *
+     * <p>For a file occupying the global effective-row interval {@code [fileStartGlobal,
+     * fileStartGlobal + fileRowCount - 1]}, the local range is the intersection: {@code localStart
+     * = max(start - fileStartGlobal, 0)}, {@code localEnd = min(end - fileStartGlobal, fileRowCount
+     * - 1)} (with {@link Long#MAX_VALUE} end left unbounded before the clamp).
+     *
+     * @param globalRange the split-global range, or {@code null} when no range is set
+     * @param fileStartGlobal the global effective-row position of the first row in this file
+     *     (accumulated across preceding files in the split)
+     * @param fileRowCount the effective row count of this file (physical row count when there is no
+     *     deletion vector)
+     * @return the per-file local range; {@code null} if {@code globalRange} is null (read the whole
+     *     file); {@link #EMPTY} if the file lies entirely outside the range (skip the file)
+     */
+    @Nullable
+    public static RowRange localOf(
+            @Nullable RowRange globalRange, long fileStartGlobal, long fileRowCount) {
+        if (globalRange == null) {
+            return null;
+        }
+        long start = Math.max(globalRange.startInclusive() - fileStartGlobal, 0);
+        long end =
+                Math.min(
+                        globalRange.endInclusive() == Long.MAX_VALUE
+                                ? Long.MAX_VALUE
+                                : globalRange.endInclusive() - fileStartGlobal,
+                        fileRowCount - 1);
+        if (start >= fileRowCount || end < 0 || end < start) {
+            return RowRange.EMPTY;
+        }
+        return RowRange.of(start, end);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        RowRange rowRange = (RowRange) o;
+        return startInclusive == rowRange.startInclusive && endInclusive == rowRange.endInclusive;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(startInclusive, endInclusive);
+    }
+
+    @Override
+    public String toString() {
+        return "RowRange[" + startInclusive + ", " + endInclusive + "]";
+    }
+}

paimon-common/src/main/java/org/apache/paimon/reader/RangeSkipReader.java

@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.reader;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * A {@link RecordReader} wrapper that implements a naive range read: skip the first {@code skip}
+ * rows of the delegate's output stream, then return at most {@code limit} rows, and stop early once
+ * the limit is reached.
+ *
+ * <p>This is the fallback implementation for {@code withRowRange} when the underlying format does
+ * not support row-range skipping (e.g. avro), for primary-key tables (merge-tree output stream),
+ * and for the precise effective-row correction of append tables that carry deletion vectors or
+ * filters. It guarantees range-read correctness over the <b>effective output stream</b> (rows
+ * already filtered by deletion vectors / predicates), at the cost of decoding skipped rows.
+ *
+ * <p>Both {@code skip} and {@code limit} are expressed in the effective-row space of the delegate's
+ * output. {@code limit} may be {@link Long#MAX_VALUE} to mean "read to the end after skipping".
+ */
+public class RangeSkipReader<T> implements RecordReader<T> {
+
+    private final RecordReader<T> delegate;
+
+    private final long skip;
+
+    private final long limit;
+
+    private long skipped;
+
+    private long returned;
+
+    private RecordReader.RecordIterator<T> currentBatch;
+
+    public RangeSkipReader(RecordReader<T> delegate, long skip, long limit) {
+        this.delegate = delegate;
+        this.skip = skip;
+        this.limit = limit;
+    }
+
+    @Nullable
+    @Override
+    public RecordReader.RecordIterator<T> readBatch() throws IOException {
+        // Already reached the limit: stop early without reading further from the delegate.
+        if (returned >= limit) {
+            return null;
+        }
+
+        // Phase 1: skip the first `skip` effective rows.
+        while (skipped < skip) {
+            RecordReader.RecordIterator<T> batch =
+                    currentBatch == null ? delegate.readBatch() : currentBatch;
+            if (batch == null) {
+                // delegate exhausted before reaching `skip` rows: nothing to return.
+                currentBatch = null;
+                return null;
+            }
+            currentBatch = batch;
+            long canSkip = Math.min(skip - skipped, Long.MAX_VALUE);
+            long skippedNow = 0;
+            T record;
+            // Consume records from the batch until we have skipped enough or the batch is drained.
+            while (skippedNow < canSkip) {
+                record = currentBatch.next();
+                if (record == null) {
+                    break;
+                }
+                skippedNow++;
+            }
+            skipped += skippedNow;
+            if (skippedNow == 0) {
+                // batch drained, move to next batch
+                currentBatch.releaseBatch();
+                currentBatch = null;
+            }
+        }
+
+        // Phase 2: return a limited view of the remaining stream.
+        RecordReader.RecordIterator<T> batch =
+                currentBatch == null ? delegate.readBatch() : currentBatch;
+        if (batch == null) {
+            currentBatch = null;
+            return null;
+        }
+        currentBatch = null;
+        long remaining = limit - returned;
+        return new LimitedRecordIterator<>(batch, remaining, this);
+    }
+
+    @Override
+    public void close() throws IOException {
+        try {
+            if (currentBatch != null) {
+                currentBatch.releaseBatch();
+            }
+        } finally {
+            delegate.close();
+        }
+    }
+
+    /** A {@link RecordReader.RecordIterator} that yields at most {@code remaining} records. */
+    private static final class LimitedRecordIterator<T> implements RecordReader.RecordIterator<T> {
+
+        private final RecordReader.RecordIterator<T> delegate;
+
+        private long remaining;
+
+        private final RangeSkipReader<T> owner;
+
+        private LimitedRecordIterator(
+                RecordReader.RecordIterator<T> delegate, long remaining, RangeSkipReader<T> owner) {
+            this.delegate = delegate;
+            this.remaining = remaining;
+            this.owner = owner;
+        }
+
+        @Nullable
+        @Override
+        public T next() throws IOException {
+            if (remaining <= 0) {
+                return null;
+            }
+            T record = delegate.next();
+            if (record == null) {
+                return null;
+            }
+            remaining--;
+            owner.returned++;
+            return record;
+        }
+
+        @Override
+        public void releaseBatch() {
+            delegate.releaseBatch();
+        }
+    }
+}