diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapIndexResult.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapIndexResult.java index cc20918a6933..ece6cf115270 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapIndexResult.java +++ b/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). + * + *

{@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) { diff --git a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java index 1fc80d1aea98..243b3dff65e1 100644 --- a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java @@ -43,6 +43,18 @@ default FileRecordReader createReader(Context context, long offset, getClass().getName())); } + /** + * Whether this format supports row-range / row-group level skipping by consuming the {@link + * Context#selection()} bitmap. + * + *

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 { diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/RowRange.java b/paimon-common/src/main/java/org/apache/paimon/predicate/RowRange.java new file mode 100644 index 000000000000..9b7bcb0c9186 --- /dev/null +++ b/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. + * + *

"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. + * + *

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. + * + *

The semantics is the 0-based effective-row position space, not the logical {@code + * firstRowId} space and not the row-id space used by {@code IndexedSplit.rowRanges()}. + * {@code ROW_ID = firstRowId + returnedPosition()} is unaffected. + * + *

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. + * + *

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 + "]"; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/reader/RangeSkipReader.java b/paimon-common/src/main/java/org/apache/paimon/reader/RangeSkipReader.java new file mode 100644 index 000000000000..221aef746722 --- /dev/null +++ b/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. + * + *

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 effective output stream (rows + * already filtered by deletion vectors / predicates), at the cost of decoding skipped rows. + * + *

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 implements RecordReader { + + private final RecordReader delegate; + + private final long skip; + + private final long limit; + + private long skipped; + + private long returned; + + private RecordReader.RecordIterator currentBatch; + + public RangeSkipReader(RecordReader delegate, long skip, long limit) { + this.delegate = delegate; + this.skip = skip; + this.limit = limit; + } + + @Nullable + @Override + public RecordReader.RecordIterator 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 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 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 implements RecordReader.RecordIterator { + + private final RecordReader.RecordIterator delegate; + + private long remaining; + + private final RangeSkipReader owner; + + private LimitedRecordIterator( + RecordReader.RecordIterator delegate, long remaining, RangeSkipReader 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(); + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/RowRangeTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/RowRangeTest.java new file mode 100644 index 000000000000..21e91b6ab91b --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/RowRangeTest.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.predicate; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link RowRange}, focused on {@link RowRange#localOf}. */ +public class RowRangeTest { + + /** No global range set -> read the whole file (null local range). */ + @Test + public void testNullGlobalRangeReturnsNull() { + assertThat(RowRange.localOf(null, 0, 100)).isNull(); + assertThat(RowRange.localOf(null, 50, 10)).isNull(); + } + + /** The file fully contains the requested range. */ + @Test + public void testRangeFullyInsideFile() { + // global [10, 19] over a 100-row file starting at global 0 -> local [10, 19]. + RowRange local = RowRange.localOf(RowRange.of(10, 19), 0, 100); + assertThat(local).isEqualTo(RowRange.of(10, 19)); + } + + /** The range starts before this file: local start is clamped to 0. */ + @Test + public void testRangeStartsBeforeFile() { + // global [5, 25] over a 20-row file starting at global 10 -> local [0, 15]. + RowRange local = RowRange.localOf(RowRange.of(5, 25), 10, 20); + assertThat(local).isEqualTo(RowRange.of(0, 15)); + } + + /** The range ends after this file: local end is clamped to the file's last row. */ + @Test + public void testRangeEndsAfterFile() { + // global [10, 200] over a 20-row file starting at global 10 -> local [0, 19]. + RowRange local = RowRange.localOf(RowRange.of(10, 200), 10, 20); + assertThat(local).isEqualTo(RowRange.of(0, 19)); + } + + /** The file lies entirely before the range -> EMPTY (skip the file). */ + @Test + public void testFileEntirelyBeforeRange() { + // global [50, 60] over a 20-row file starting at global 0 -> EMPTY. + RowRange local = RowRange.localOf(RowRange.of(50, 60), 0, 20); + assertThat(local).isSameAs(RowRange.EMPTY); + } + + /** The file lies entirely after the range -> EMPTY (skip the file). */ + @Test + public void testFileEntirelyAfterRange() { + // global [0, 5] over a 20-row file starting at global 10 -> start clamps to 0 but + // end = 5 - 10 = -5 < 0 -> EMPTY. + RowRange local = RowRange.localOf(RowRange.of(0, 5), 10, 20); + assertThat(local).isSameAs(RowRange.EMPTY); + } + + /** The range touches only the first row of the file. */ + @Test + public void testRangeCoversOnlyFirstRow() { + // global [10, 10] over a 20-row file starting at global 10 -> local [0, 0]. + RowRange local = RowRange.localOf(RowRange.of(10, 10), 10, 20); + assertThat(local).isEqualTo(RowRange.of(0, 0)); + assertThat(local.count()).isEqualTo(1L); + } + + /** The range touches only the last row of the file. */ + @Test + public void testRangeCoversOnlyLastRow() { + // global [29, 29] over a 20-row file starting at global 10 -> local [19, 19]. + RowRange local = RowRange.localOf(RowRange.of(29, 29), 10, 20); + assertThat(local).isEqualTo(RowRange.of(19, 19)); + } + + /** An unbounded end reads to the file's last row. */ + @Test + public void testUnboundedEndClampsToLastRow() { + // global [15, MAX] over a 20-row file starting at global 10 -> local [5, 19]. + RowRange local = RowRange.localOf(RowRange.of(15, Long.MAX_VALUE), 10, 20); + assertThat(local).isEqualTo(RowRange.of(5, 19)); + assertThat(local.count()).isEqualTo(15L); + } + + /** FULL range (from 0 to MAX) over a file starting at a non-zero offset. */ + @Test + public void testFullRangeOverOffsetFile() { + RowRange local = RowRange.localOf(RowRange.FULL, 30, 20); + assertThat(local).isEqualTo(RowRange.of(0, 19)); + } + + /** A range that starts exactly at the file boundary. */ + @Test + public void testRangeStartsAtFileBoundary() { + // global [10, 15] over a 20-row file starting at global 10 -> local [0, 5]. + RowRange local = RowRange.localOf(RowRange.of(10, 15), 10, 20); + assertThat(local).isEqualTo(RowRange.of(0, 5)); + } + + /** A range that ends exactly at the file boundary. */ + @Test + public void testRangeEndsAtFileBoundary() { + // global [25, 29] over a 20-row file starting at global 10 -> local [15, 19]. + RowRange local = RowRange.localOf(RowRange.of(25, 29), 10, 20); + assertThat(local).isEqualTo(RowRange.of(15, 19)); + } + + /** start exactly at fileRowCount (one past the last row) -> EMPTY. */ + @Test + public void testStartAtFileRowCount() { + RowRange local = RowRange.localOf(RowRange.of(30, 40), 10, 20); + assertThat(local).isSameAs(RowRange.EMPTY); + } + + /** Multiple files in a split: each local range is computed against its own offset. */ + @Test + public void testMultipleFilesConcat() { + // Split: file0 [0,99] (100 rows), file1 [100,199] (100 rows). Global range [50, 149]. + RowRange global = RowRange.of(50, 149); + RowRange local0 = RowRange.localOf(global, 0, 100); + RowRange local1 = RowRange.localOf(global, 100, 100); + assertThat(local0).isEqualTo(RowRange.of(50, 99)); // last 50 rows of file0 + assertThat(local1).isEqualTo(RowRange.of(0, 49)); // first 50 rows of file1 + assertThat(local0.count() + local1.count()).isEqualTo(100L); + } + + /** A file in the middle of the split that is fully outside the range. */ + @Test + public void testMiddleFileFullyOutside() { + // Split: file0 [0,49], file1 [50,99], file2 [100,149]. Global range [60, 140]. + RowRange global = RowRange.of(60, 140); + assertThat(RowRange.localOf(global, 0, 50)).isSameAs(RowRange.EMPTY); // file0 before range + assertThat(RowRange.localOf(global, 50, 50)) + .isEqualTo(RowRange.of(10, 49)); // file1 partial + assertThat(RowRange.localOf(global, 100, 50)) + .isEqualTo(RowRange.of(0, 40)); // file2 partial + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/reader/RangeSkipReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/reader/RangeSkipReaderTest.java new file mode 100644 index 000000000000..0e5990bcb80d --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/reader/RangeSkipReaderTest.java @@ -0,0 +1,138 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link RangeSkipReader}. */ +public class RangeSkipReaderTest { + + /** A simple in-memory RecordReader over a list of integers, one element per batch. */ + private static final class ListRecordReader implements RecordReader { + + private final List data; + + private int idx; + + ListRecordReader(List data) { + this.data = data; + } + + @Nullable + @Override + public RecordIterator readBatch() { + if (idx >= data.size()) { + return null; + } + final int current = idx++; + return new RecordIterator() { + private boolean consumed = false; + + @Nullable + @Override + public Integer next() { + if (consumed) { + return null; + } + consumed = true; + return data.get(current); + } + + @Override + public void releaseBatch() {} + }; + } + + @Override + public void close() {} + } + + private List readAll(RecordReader reader) throws IOException { + List result = new ArrayList<>(); + RecordReader.RecordIterator batch; + while ((batch = reader.readBatch()) != null) { + Integer v; + while ((v = batch.next()) != null) { + result.add(v); + } + batch.releaseBatch(); + } + return result; + } + + @Test + public void testSkipAndLimit() throws IOException { + List data = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); + RangeSkipReader reader = new RangeSkipReader<>(new ListRecordReader(data), 3, 4); + // skip first 3 (0,1,2), take 4 (3,4,5,6) + assertThat(readAll(reader)).containsExactly(3, 4, 5, 6); + } + + @Test + public void testSkipZero() throws IOException { + List data = Arrays.asList(0, 1, 2, 3, 4); + RangeSkipReader reader = new RangeSkipReader<>(new ListRecordReader(data), 0, 3); + assertThat(readAll(reader)).containsExactly(0, 1, 2); + } + + @Test + public void testLimitToEnd() throws IOException { + List data = Arrays.asList(0, 1, 2, 3, 4); + RangeSkipReader reader = + new RangeSkipReader<>(new ListRecordReader(data), 2, Long.MAX_VALUE); + // skip 2, read to end + assertThat(readAll(reader)).containsExactly(2, 3, 4); + } + + @Test + public void testSkipBeyondData() throws IOException { + List data = Arrays.asList(0, 1, 2); + RangeSkipReader reader = new RangeSkipReader<>(new ListRecordReader(data), 5, 3); + // skip beyond data -> empty + assertThat(readAll(reader)).isEmpty(); + } + + @Test + public void testLimitBeyondData() throws IOException { + List data = Arrays.asList(0, 1, 2); + RangeSkipReader reader = new RangeSkipReader<>(new ListRecordReader(data), 1, 10); + // skip 1, limit exceeds data -> read remaining + assertThat(readAll(reader)).containsExactly(1, 2); + } + + @Test + public void testEarlyTermination() throws IOException { + // with limit 2 and skip 1 from 10 elements, the reader must stop after 2 returned rows. + List data = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + data.add(i); + } + RangeSkipReader reader = new RangeSkipReader<>(new ListRecordReader(data), 1, 2); + assertThat(readAll(reader)).containsExactly(1, 2); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java index a82194cc7616..5c48bd4e4f3f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java @@ -27,6 +27,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.utils.RoaringBitmap32; @@ -53,12 +54,36 @@ public static FileIndexResult evaluate( DataFileMeta file, @Nullable DeletionVector dv) throws IOException { + return evaluate( + fileIO, dataSchema, dataFilter, topN, limit, null, dataFilePathFactory, file, dv); + } + + public static FileIndexResult evaluate( + FileIO fileIO, + TableSchema dataSchema, + List dataFilter, + @Nullable TopN topN, + @Nullable Integer limit, + @Nullable RowRange rowRange, + DataFilePathFactory dataFilePathFactory, + DataFileMeta file, + @Nullable DeletionVector dv) + throws IOException { // File index selections use 32-bit positions. Fall back when they cannot safely represent // the file or its deletion vector. if (file.rowCount() > RoaringBitmap32.MAX_VALUE || dv instanceof Bitmap64DeletionVector) { return FileIndexResult.REMAIN; } + // rowRange is only valid for a pure range read: no filter, no topN, no limit. The caller + // (RawFileSplitRead full-scan range path) already guarantees this, but defend against a + // future caller that violates it: fall back to the plain path and let RangeSkipReader + // handle + // the range over the effective output stream. + if (rowRange != null && isNullOrEmpty(dataFilter) && topN == null && limit == null) { + return evaluateRowRange(file, rowRange); + } + if (isNullOrEmpty(dataFilter) && topN == null) { if (limit == null) { return FileIndexResult.REMAIN; @@ -104,6 +129,24 @@ public static FileIndexResult evaluate( } } + /** + * Evaluate a pure row-range read (no filter / topN / limit / deletion vector): push the full + * local effective-row range {@code [start, end]} down as a selection bitmap so the parquet + * reader prunes row groups by both endpoints. Starting the bitmap from {@code + * rowRange.startInclusive()} (instead of 0) lets row-group skipping avoid reading leading row + * groups that fall before the requested range. + * + *

Deletion vectors are not considered here: a range read carries no DV (physical == + * effective), and any DV-aware effective-row correction is handled by a higher-layer + * RangeSkipReader. + */ + private static FileIndexResult evaluateRowRange(DataFileMeta file, RowRange rowRange) { + // No DV in the range-read path; pass null so createBaseSelection skips the andNot step. + BitmapIndexResult selection = createBaseSelection(file, null); + long end = Math.min(rowRange.endInclusive(), file.rowCount() - 1); + return selection.range(rowRange.startInclusive(), end); + } + private static BitmapIndexResult createBaseSelection( DataFileMeta file, @Nullable DeletionVector dv) { BitmapIndexResult selection = diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index 919100ec1f8f..83acfd9e45a2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -43,9 +43,11 @@ import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.partition.PartitionUtils; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.DataEvolutionFileReader; import org.apache.paimon.reader.EmptyFileRecordReader; import org.apache.paimon.reader.FileRecordReader; +import org.apache.paimon.reader.RangeSkipReader; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.ReaderSupplier; import org.apache.paimon.reader.RecordReader; @@ -128,6 +130,8 @@ public class DataEvolutionSplitRead implements SplitRead { protected RowType readRowType; @Nullable private List filters; @Nullable private ReadBatchSizer readBatchSizer; + @Nullable private RowRange rowRange; + private boolean fullScanRangeApplied; public DataEvolutionSplitRead( FileIO fileIO, @@ -183,6 +187,12 @@ public SplitRead withReadBatchSizer(ReadBatchSizer sizer) { return this; } + @Override + public SplitRead withRowRange(@Nullable RowRange rowRange) { + this.rowRange = rowRange; + return this; + } + /** * Row tracking fields are assigned from the manifest entry instead of being read from the file, * and data evolution may reassign row ids, so a physical copy in the file can be stale. Never @@ -199,11 +209,20 @@ private static List pushDownFilters(List filters) { @Override public RecordReader createReader(Split split) throws IOException { + fullScanRangeApplied = false; + RecordReader reader; if (split instanceof DataSplit) { - return createReader((DataSplit) split); + reader = createReader((DataSplit) split); } else { - return createReader((IndexedSplit) split); + reader = createReader((IndexedSplit) split); + } + // When a full-scan range was applied as per-file / per-group local ranges inside + // createReader(DataSplit, ...), fullScanRangeApplied is true and no outer wrap is needed. + // Otherwise apply a naive skip+limit over the effective output stream. + if (rowRange != null && !fullScanRangeApplied) { + reader = new RangeSkipReader<>(reader, rowRange.startInclusive(), rowRange.count()); } + return reader; } private RecordReader createReader(DataSplit dataSplit) throws IOException { @@ -229,8 +248,35 @@ private RecordReader createReader( // already taken by the caller List filters = readTypeFilters(this.filters, readRowType); + // The per-file local range pushdown (selection bitmap -> parquet row-group skipping) is + // only valid for a full-table scan: no filter, no row-id ranges (IndexedSplit) and no + // deletion vector. In that case physical == effective, so the local range can be computed + // from the physical row count without reading any DV. See RawFileSplitRead for the same + // pattern. + boolean fullScanRange = + rowRange != null && rowRanges == null && deletionVectorFactory == null; + if (fullScanRange && !isNullOrEmpty(filters)) { + fullScanRange = false; + } else if (fullScanRange) { + fullScanRangeApplied = true; + } + List> splitByRowId = mergeRangesAndSort(files); + long groupStartGlobal = 0L; for (List needMergeFiles : splitByRowId) { + RowRange groupRowRange = null; + if (fullScanRange) { + long groupRowCount = needMergeFiles.get(0).rowCount(); + groupRowRange = RowRange.localOf(rowRange, groupStartGlobal, groupRowCount); + groupStartGlobal += groupRowCount; + if (groupRowRange != null && groupRowRange.isEmpty()) { + // entire group lies outside the range -> skip without reading + final EmptyFileRecordReader empty = new EmptyFileRecordReader<>(); + suppliers.add(() -> empty); + continue; + } + } + final RowRange range = groupRowRange; if (needMergeFiles.size() == 1 || readRowType.getFields().isEmpty()) { // No need to merge fields, just create a single file reader suppliers.add( @@ -244,7 +290,8 @@ private RecordReader createReader( filters, rowRanges, readRowType, - deletionVector); + deletionVector, + range); }); } else { @@ -261,7 +308,8 @@ private RecordReader createReader( dataFilePathFactory, rowRanges, readRowType, - deletionVector); + deletionVector, + range); }); } } @@ -284,7 +332,8 @@ private DataEvolutionFileReader createUnionReader( DataFilePathFactory dataFilePathFactory, List rowRanges, RowType readRowType, - @Nullable DeletionVectorWithRange deletionVector) + @Nullable DeletionVectorWithRange deletionVector, + @Nullable RowRange fileRowRange) throws IOException { List fieldsFiles = splitFieldBunches( @@ -369,7 +418,8 @@ private DataEvolutionFileReader createUnionReader( formatReaderMapping, rowRanges, partialReadRowType, - deletionVector)); + deletionVector, + fileRowRange)); } return nestedFieldEnabled @@ -441,7 +491,8 @@ private RecordReader createFieldBunchReader( FormatReaderMapping formatReaderMapping, List rowRanges, RowType readRowType, - @Nullable DeletionVectorWithRange deletionVector) + @Nullable DeletionVectorWithRange deletionVector, + @Nullable RowRange fileRowRange) throws IOException { if (bunch instanceof DataBunch) { // for data bunch, directly read the single file @@ -452,38 +503,56 @@ private RecordReader createFieldBunchReader( formatReaderMapping, rowRanges, readRowType, - deletionVector); + readTarget(bunch.files().get(0), dataFilePathFactory, rowRanges), + deletionVector, + null, + fileRowRange); } else if (bunch instanceof VectorFileBunch) { // for vector bunch, sequential read all data files and concat them - return sequentialReadFiles( - bunch.files(), - partition, - dataFilePathFactory, - formatReaderMapping, - rowRanges, - deletionVector); + RecordReader vectorReader = + sequentialReadFiles( + bunch.files(), + partition, + dataFilePathFactory, + formatReaderMapping, + rowRanges, + deletionVector); + if (fileRowRange != null) { + vectorReader = + new RangeSkipReader<>( + vectorReader, fileRowRange.startInclusive(), fileRowRange.count()); + } + return vectorReader; } else if (bunch instanceof BlobFileBunch) { // for blob bunch, fallback on placeholders BlobFileBunch blobBunch = (BlobFileBunch) bunch; int blobIndex = findBlobFieldIndex(readRowType); checkArgument(blobIndex >= 0, "Blob bunch read type should contain a blob field."); - return new BlobFallbackRecordReader( - bunch.files(), - file -> - createFileReader( - partition, - file, - dataFilePathFactory, - formatReaderMapping, - rowRanges, - readRowType, - deletionVector), - (reader, range) -> - applyDeletionVector(reader, range, rowRanges, deletionVector), - blobBunch.logicalRange(), - rowRanges, - readRowType, - blobIndex); + RecordReader blobReader = + new BlobFallbackRecordReader( + bunch.files(), + file -> + createFileReader( + partition, + file, + dataFilePathFactory, + formatReaderMapping, + rowRanges, + readRowType, + deletionVector), + (reader, range) -> + applyDeletionVector(reader, range, rowRanges, deletionVector), + blobBunch.logicalRange(), + rowRanges, + readRowType, + blobIndex); + // Apply the per-group local range as an outer skip+limit on the blob fallback output. + if (fileRowRange != null) { + blobReader = + new RangeSkipReader<>( + blobReader, fileRowRange.startInclusive(), fileRowRange.count()); + } + return blobReader; } else { throw new UnsupportedOperationException("Unsupported bunch type: " + bunch); } @@ -504,6 +573,7 @@ private RecordReader sequentialReadFiles( createFileReader( partition, file, + dataFilePathFactory, formatReaderMapping, rowRanges, readRowType, @@ -512,6 +582,7 @@ private RecordReader sequentialReadFiles( dataFilePathFactory.toPath(file), file.fileSize()), deletionVector, + null, null)); } return ConcatRecordReader.create(readerSuppliers); @@ -526,14 +597,15 @@ private static int findBlobFieldIndex(RowType rowType) { return -1; } - private FileRecordReader createFileReader( + private RecordReader createFileReader( BinaryRow partition, DataFilePathFactory dataFilePathFactory, DataFileMeta file, @Nullable List filters, List rowRanges, RowType readRowType, - @Nullable DeletionVectorWithRange deletionVector) + @Nullable DeletionVectorWithRange deletionVector, + @Nullable RowRange fileRowRange) throws IOException { FileReadTarget readTarget = readTarget(file, dataFilePathFactory, rowRanges); String formatIdentifier = readTarget.formatIdentifier; @@ -555,9 +627,8 @@ private FileRecordReader createFileReader( key -> formatBuilder(readRowType, fileFilters, nestedFieldEnabled) .build(formatIdentifier, schema, dataSchema)); - FileIndexResult fileIndexResult = null; - if (fileIndexReadEnabled) { + if (fileRowRange == null && fileIndexReadEnabled) { fileIndexResult = FileIndexEvaluator.evaluate( fileIO, @@ -576,14 +647,17 @@ private FileRecordReader createFileReader( return createFileReader( partition, file, + dataFilePathFactory, formatReaderMapping, rowRanges, readRowType, readTarget, deletionVector, - fileIndexResult); + fileIndexResult, + fileRowRange); } + /** Blob/vector path: no per-file range pushdown, returns {@link FileRecordReader}. */ private FileRecordReader createFileReader( BinaryRow partition, DataFileMeta file, @@ -593,27 +667,51 @@ private FileRecordReader createFileReader( RowType readRowType, @Nullable DeletionVectorWithRange deletionVector) throws IOException { - return createFileReader( - partition, - file, - formatReaderMapping, - rowRanges, - readRowType, - readTarget(file, dataFilePathFactory, rowRanges), - deletionVector, - null); + return (FileRecordReader) + createFileReader( + partition, + file, + dataFilePathFactory, + formatReaderMapping, + rowRanges, + readRowType, + readTarget(file, dataFilePathFactory, rowRanges), + deletionVector, + null, + null); } - private FileRecordReader createFileReader( + private RecordReader createFileReader( BinaryRow partition, DataFileMeta file, + DataFilePathFactory dataFilePathFactory, FormatReaderMapping formatReaderMapping, List rowRanges, RowType readRowType, FileReadTarget readTarget, @Nullable DeletionVectorWithRange deletionVector, - @Nullable FileIndexResult fileIndexResult) + @Nullable FileIndexResult fileIndexResult, + @Nullable RowRange fileRowRange) throws IOException { + boolean rangePushdown = + fileRowRange != null + && formatReaderMapping.getReaderFactory().supportsRowRangeSkip(); + if (rangePushdown && fileIndexResult == null) { + fileIndexResult = + FileIndexEvaluator.evaluate( + fileIO, + formatReaderMapping.getDataSchema(), + null, + null, + null, + fileRowRange, + dataFilePathFactory, + file, + null); + if (!fileIndexResult.remain()) { + return new EmptyFileRecordReader<>(); + } + } RoaringBitmap32 selection = file.toFileSelection(rowRanges); BitmapIndexResult bitmapIndexResult = fileIndexResult instanceof BitmapIndexResult @@ -652,7 +750,13 @@ private FileRecordReader createFileReader( fileRecordReader = new ApplyBitmapIndexRecordReader(fileRecordReader, bitmapIndexResult); } - + if (fileRowRange != null) { + if (rangePushdown) { + return fileRecordReader; + } + return new RangeSkipReader<>( + fileRecordReader, fileRowRange.startInclusive(), fileRowRange.count()); + } return applyDeletionVector( fileRecordReader, file.nonNullRowIdRange(), rowRanges, deletionVector); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java index 9e66aad52fca..f2c4b06f72f5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java @@ -42,7 +42,9 @@ import org.apache.paimon.mergetree.compact.MergeFunctionWrapper; import org.apache.paimon.mergetree.compact.ReducerMergeFunctionWrapper; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.EmptyRecordReader; +import org.apache.paimon.reader.RangeSkipReader; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.ReaderSupplier; import org.apache.paimon.reader.RecordReader; @@ -97,6 +99,8 @@ public class MergeFileSplitRead implements SplitRead { @Nullable private List filtersForKeys; @Nullable private List filtersForAll; + @Nullable private RowRange rowRange; + private boolean forceKeepDelete = false; public MergeFileSplitRead( @@ -192,6 +196,12 @@ public MergeFileSplitRead withReadBatchSizer(ReadBatchSizer sizer) { return this; } + @Override + public MergeFileSplitRead withRowRange(@Nullable RowRange rowRange) { + this.rowRange = rowRange; + return this; + } + @Override public MergeFileSplitRead forceKeepDelete() { this.forceKeepDelete = true; @@ -250,21 +260,32 @@ public RecordReader createReader(Split split) throws IOException { } public RecordReader createReader(DataSplit split) throws IOException { + RecordReader reader; if (split.isStreaming() || split.bucket() == BucketMode.POSTPONE_BUCKET) { - return createNoMergeReader( - split.partition(), - split.bucket(), - split.dataFiles(), - split.deletionFiles().orElse(null), - split.isStreaming()); + reader = + createNoMergeReader( + split.partition(), + split.bucket(), + split.dataFiles(), + split.deletionFiles().orElse(null), + split.isStreaming()); } else { - return createMergeReader( - split.partition(), - split.bucket(), - split.dataFiles(), - split.deletionFiles().orElse(null), - forceKeepDelete); + reader = + createMergeReader( + split.partition(), + split.bucket(), + split.dataFiles(), + split.deletionFiles().orElse(null), + forceKeepDelete); } + // Primary-key table cannot skip row groups physically (merge-tree reorders rows), but the + // sort-merge output stream has a deterministic row order, so apply a naive skip+limit over + // the effective output stream. This also naturally handles deletion vectors / filters + // (already applied upstream in the merge reader). + if (rowRange != null) { + reader = new RangeSkipReader<>(reader, rowRange.startInclusive(), rowRange.count()); + } + return reader; } /** Reads a writer-grouped postpone split with key predicates only. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java index 0c79a11db288..6a308e79f0fb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java @@ -38,10 +38,12 @@ import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.partition.PartitionUtils; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.EmptyFileRecordReader; import org.apache.paimon.reader.FileRecordReader; import org.apache.paimon.reader.LimitRecordReader; +import org.apache.paimon.reader.RangeSkipReader; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.ReaderSupplier; import org.apache.paimon.reader.RecordReader; @@ -94,6 +96,7 @@ public class RawFileSplitRead implements SplitRead { @Nullable private TopN topN; @Nullable private Integer limit; @Nullable private ReadBatchSizer readBatchSizer; + @Nullable private RowRange rowRange; public RawFileSplitRead( FileIO fileIO, @@ -162,6 +165,12 @@ public SplitRead withReadBatchSizer(ReadBatchSizer sizer) { return this; } + @Override + public RawFileSplitRead withRowRange(@Nullable RowRange rowRange) { + this.rowRange = rowRange; + return this; + } + @Override public RecordReader createReader(Split s) throws IOException { if (s instanceof DataSplit) { @@ -212,7 +221,36 @@ public RecordReader createReader( Builder formatReaderMappingBuilder = createFormatReaderMappingBuilder(outputRowType, topN, limit); + boolean hasFilter = filters != null && !filters.isEmpty(); + boolean hasDv = hasDeletionVector(files, dvFactories); + boolean fullScanRange = + rowRange != null && !hasFilter && topN == null && limit == null && !hasDv; + boolean canPushdown = fullScanRange && !files.isEmpty(); + if (canPushdown) { + for (DataFileMeta file : files) { + if (!formatReaderMapping(file, formatReaderMappingBuilder) + .getReaderFactory() + .supportsRowRangeSkip()) { + canPushdown = false; + break; + } + } + } + + long fileStartGlobal = 0L; for (DataFileMeta file : files) { + RowRange fileRowRange = null; + Map> fileDvFactories = dvFactories; + if (canPushdown) { + fileRowRange = RowRange.localOf(rowRange, fileStartGlobal, file.rowCount()); + fileStartGlobal += file.rowCount(); + if (fileRowRange != null && fileRowRange.isEmpty()) { + final EmptyFileRecordReader empty = new EmptyFileRecordReader<>(); + suppliers.add(() -> empty); + continue; + } + fileDvFactories = null; + } suppliers.add( createFileReader( partition, @@ -220,11 +258,18 @@ public RecordReader createReader( file, formatReaderMappingBuilder, outputRowType, - dvFactories, - null)); + fileDvFactories, + null, + fileRowRange)); } RecordReader reader = ConcatRecordReader.create(suppliers); + // When the range could not be pushed into the formats, enforce it once over the + // effective-row output stream with a naive skip + limit. The pushdown path (canPushdown) + // already returns exactly the range per file, so no wrap is added there. + if (rowRange != null && !canPushdown) { + return new RangeSkipReader<>(reader, rowRange.startInclusive(), rowRange.count()); + } // Apply the final limit after deletion vectors when no later predicate can drop rows. if (topN == null && (filters == null || filters.isEmpty())) { return LimitRecordReader.limit(reader, limit); @@ -232,6 +277,29 @@ public RecordReader createReader( return reader; } + /** Whether any file in the split carries a non-empty deletion vector. */ + private boolean hasDeletionVector( + List files, + @Nullable Map> dvFactories) { + if (dvFactories == null) { + return false; + } + for (DataFileMeta file : files) { + IOExceptionSupplier supplier = dvFactories.get(file.fileName()); + if (supplier != null) { + try { + DeletionVector dv = supplier.get(); + if (dv != null && !dv.isEmpty()) { + return true; + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + return false; + } + FileRecordReader createFileReader( DataSplit dataSplit, RoaringBitmap32 selectedPositions) throws IOException { DataFileMeta dataFile = dataSplit.dataFiles().get(0); @@ -254,7 +322,8 @@ FileRecordReader createFileReader( createFormatReaderMappingBuilder(outputRowType, null, null), outputRowType, dvFactories, - selectedPositions) + selectedPositions, + null) .get(); } @@ -284,20 +353,9 @@ private ReaderSupplier createFileReader( Builder formatBuilder, RowType outputRowType, @Nullable Map> dvFactories, - @Nullable RoaringBitmap32 selectedPositions) { - String formatIdentifier = DataFilePathFactory.formatIdentifier(file.fileName()); - long schemaId = file.schemaId(); - - FormatReaderMapping formatReaderMapping = - formatReaderMappings.computeIfAbsent( - new FormatKey(file.schemaId(), formatIdentifier), - key -> - formatBuilder.build( - formatIdentifier, - schema, - schemaId == schema.id() - ? schema - : schemaManager.schema(schemaId))); + @Nullable RoaringBitmap32 selectedPositions, + @Nullable RowRange fileRowRange) { + FormatReaderMapping formatReaderMapping = formatReaderMapping(file, formatBuilder); IOExceptionSupplier dvFactory = dvFactories == null ? null : dvFactories.get(file.fileName()); @@ -309,21 +367,51 @@ private ReaderSupplier createFileReader( formatReaderMapping, outputRowType, dvFactory, - selectedPositions); + selectedPositions, + fileRowRange); } - private FileRecordReader createFileReader( + /** Resolve (and cache) the {@link FormatReaderMapping} for a file's format and schema. */ + private FormatReaderMapping formatReaderMapping(DataFileMeta file, Builder formatBuilder) { + String formatIdentifier = DataFilePathFactory.formatIdentifier(file.fileName()); + long schemaId = file.schemaId(); + return formatReaderMappings.computeIfAbsent( + new FormatKey(file.schemaId(), formatIdentifier), + key -> + formatBuilder.build( + formatIdentifier, + schema, + schemaId == schema.id() ? schema : schemaManager.schema(schemaId))); + } + + private RecordReader createFileReader( BinaryRow partition, DataFileMeta file, DataFilePathFactory dataFilePathFactory, FormatReaderMapping formatReaderMapping, RowType outputRowType, IOExceptionSupplier dvFactory, - @Nullable RoaringBitmap32 selectedPositions) + @Nullable RoaringBitmap32 selectedPositions, + @Nullable RowRange fileRowRange) throws IOException { FileIndexResult fileIndexResult = null; DeletionVector deletionVector = dvFactory == null ? null : dvFactory.get(); - if (fileIndexReadEnabled) { + if (fileRowRange != null) { + fileIndexResult = + FileIndexEvaluator.evaluate( + fileIO, + formatReaderMapping.getDataSchema(), + null, + null, + null, + fileRowRange, + dataFilePathFactory, + file, + deletionVector); + if (!fileIndexResult.remain()) { + return new EmptyFileRecordReader<>(); + } + } else if (fileIndexReadEnabled) { fileIndexResult = FileIndexEvaluator.evaluate( fileIO, @@ -382,7 +470,7 @@ private FileRecordReader createFileReader( } if (deletionVector != null && !deletionVector.isEmpty()) { - return new ApplyDeletionVectorReader(fileRecordReader, deletionVector); + fileRecordReader = new ApplyDeletionVectorReader(fileRecordReader, deletionVector); } return fileRecordReader; } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java index 6a1ed5fc00f0..93c750a26652 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java @@ -20,6 +20,7 @@ import org.apache.paimon.disk.IOManager; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.RecordReader; @@ -58,6 +59,14 @@ default SplitRead withReadBatchSizer(ReadBatchSizer sizer) { return this; } + /** + * Set a {@link RowRange} to restrict the read to a 0-based effective-row interval of the split. + * Default no-op; implementations supporting range reads override this. + */ + default SplitRead withRowRange(@Nullable RowRange rowRange) { + return this; + } + /** Create a {@link RecordReader} from split. */ RecordReader createReader(Split split) throws IOException; @@ -94,6 +103,12 @@ public SplitRead withReadBatchSizer(ReadBatchSizer sizer) { return this; } + @Override + public SplitRead withRowRange(@Nullable RowRange rowRange) { + read.withRowRange(rowRange); + return this; + } + @Override public RecordReader createReader(Split split) throws IOException { return splitConvert.apply(split); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java index 7b14dd1cac69..77cf78c669c4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java @@ -25,7 +25,9 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateProjectionConverter; import org.apache.paimon.predicate.PredicateVisitor; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.Transform; +import org.apache.paimon.reader.RangeSkipReader; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.types.DataField; @@ -49,6 +51,7 @@ public abstract class AbstractDataTableRead implements InnerTableRead { private RowType readType; protected boolean executeFilter = false; private Predicate predicate; + @Nullable protected RowRange rowRange; private final TableSchema schema; // reader-level filtering sees raw values, so it stays off for auth-enabled tables, @@ -78,7 +81,19 @@ public AbstractDataTableRead(@Nullable TableSchema schema) { public abstract void applyReadType(RowType readType); - public abstract RecordReader reader(Split split) throws IOException; + /** + * Create the underlying reader for a split with an optional {@link RowRange}. + * + *

Subclasses forward {@code rowRange} to the chosen {@link + * org.apache.paimon.operation.SplitRead#withRowRange} before {@code createReader(split)}. + */ + public abstract RecordReader reader(Split split, @Nullable RowRange rowRange) + throws IOException; + + /** Backward-compatible entry: read the whole split (no row range). */ + public RecordReader reader(Split split) throws IOException { + return reader(split, null); + } @Override public TableRead withIOManager(IOManager ioManager) { @@ -133,8 +148,15 @@ protected Predicate predicate() { @Override public RecordReader createReader(Split split) throws IOException { + return createReader(split, null); + } + + @Override + public RecordReader createReader(Split split, @Nullable RowRange rowRange) + throws IOException { + this.rowRange = rowRange; QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); - return createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); + return createDataReader(queryAuthContext.split(), queryAuthContext.authResult(), rowRange); } protected final QueryAuthContext unwrapQueryAuthSplit(Split split) { @@ -146,7 +168,8 @@ protected final QueryAuthContext unwrapQueryAuthSplit(Split split) { } protected final RecordReader createDataReader( - Split split, @Nullable TableQueryAuthResult authResult) throws IOException { + Split split, @Nullable TableQueryAuthResult authResult, @Nullable RowRange rowRange) + throws IOException { // A TableRead can be reused for multiple splits. Authentication may have expanded an // explicitly configured physical projection for the previous split, so restore it before // applying the current split's authorization dependencies. Without an explicit projection, @@ -155,20 +178,30 @@ protected final RecordReader createDataReader( applyReadType(readType); appliedReadType = null; } + // rowRange slices the *filtered* effective-row stream, so RangeSkipReader must run AFTER + // every filter / auth layer. When there is no outer filter / auth layer, forward rowRange + // to the inner reader so it can push it down (parquet selection / per-file + // RangeSkipReader). + // Otherwise forward null here and wrap RangeSkipReader outside the filter / auth output. + boolean outerWrap = authResult != null || executeFilter; + RowRange innerRange = outerWrap ? null : rowRange; RecordReader reader; if (authResult == null) { - reader = backProject(reader(split)); + reader = backProject(reader(split, innerRange)); } else { - reader = authedReader(split, authResult); + reader = authedReader(split, authResult, innerRange); } if (executeFilter) { reader = executeFilter(reader); } - + if (rowRange != null && outerWrap) { + reader = new RangeSkipReader<>(reader, rowRange.startInclusive(), rowRange.count()); + } return reader; } - private RecordReader authedReader(Split split, TableQueryAuthResult authResult) + private RecordReader authedReader( + Split split, TableQueryAuthResult authResult, @Nullable RowRange rowRange) throws IOException { List readFields = currentReadType().getFieldNames(); // masked filter columns are read and masked like rule fields, then evaluated post-mask @@ -198,7 +231,7 @@ private RecordReader authedReader(Split split, TableQueryAuthResult } RecordReader reader = authResult.doAuth( - reader(split), + reader(split, rowRange), outputType, authResult.extractPredicate(), selectedColumnMasking); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java index 99561b75510b..8e03d6e0e96b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AppendTableRead.java @@ -22,6 +22,7 @@ import org.apache.paimon.operation.MergeFileSplitRead; import org.apache.paimon.operation.SplitRead; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.RecordReader; @@ -123,10 +124,13 @@ protected ReadBatchSizer readBatchSizer() { } @Override - public RecordReader reader(Split split) throws IOException { + public RecordReader reader(Split split, @Nullable RowRange rowRange) + throws IOException { for (SplitReadProvider readProvider : readProviders) { if (readProvider.match(split, new SplitReadProvider.Context(false))) { - return readProvider.get().get().createReader(split); + SplitRead read = readProvider.get().get(); + read.withRowRange(rowRange); + return read.createReader(split); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java index 26672c2a8ace..f2a7259b94ba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.TableSchema; @@ -54,7 +55,9 @@ public DataEvolutionTableRead( } @Override - public RecordReader createReader(Split split) throws IOException { + public RecordReader createReader(Split split, @Nullable RowRange rowRange) + throws IOException { + this.rowRange = rowRange; QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); int[] blobViewFields = BlobViewTableReadSupport.blobViewFieldIndexes(currentReadType(), options); @@ -74,7 +77,11 @@ public RecordReader createReader(Split split) throws IOException { topN, limit, executeFilter, - () -> createDataReader(queryAuthContext.split(), queryAuthContext.authResult()), + () -> + createDataReader( + queryAuthContext.split(), + queryAuthContext.authResult(), + rowRange), () -> { InnerTableRead prescanRead = readFactory.get(); if (sizer != null) { @@ -87,6 +94,6 @@ public RecordReader createReader(Split split) throws IOException { return prescanRead; }); } - return createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); + return createDataReader(queryAuthContext.split(), queryAuthContext.authResult(), rowRange); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java index 117798d56279..918ac5ceacc9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java @@ -28,6 +28,7 @@ import org.apache.paimon.operation.RawFileSplitRead; import org.apache.paimon.operation.SplitRead; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.LimitRecordReader; import org.apache.paimon.reader.ReadBatchSizer; @@ -158,19 +159,29 @@ public RecordReader createReader(List splits) throws IOExcep @Override public RecordReader createReader(Split split) throws IOException { + return createReader(split, null); + } + + @Override + public RecordReader createReader(Split split, @Nullable RowRange rowRange) + throws IOException { + this.rowRange = rowRange; QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); RecordReader reader; int[] blobViewFields = blobViewFieldIndexes(currentReadType(), options); if (catalogContext != null && blobViewFields.length > 0) { - reader = createReaderWithBlobView(queryAuthContext, blobViewFields); + reader = createReaderWithBlobView(queryAuthContext, blobViewFields, rowRange); } else { - reader = createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); + reader = + createDataReader( + queryAuthContext.split(), queryAuthContext.authResult(), rowRange); } return LimitRecordReader.limit(reader, limit); } private RecordReader createReaderWithBlobView( - QueryAuthContext queryAuthContext, int[] blobViewFields) throws IOException { + QueryAuthContext queryAuthContext, int[] blobViewFields, @Nullable RowRange rowRange) + throws IOException { RecordReader reader; reader = BlobViewTableReadSupport.createBlobViewReader( @@ -185,7 +196,9 @@ private RecordReader createReaderWithBlobView( executeFilter, () -> createDataReader( - queryAuthContext.split(), queryAuthContext.authResult()), + queryAuthContext.split(), + queryAuthContext.authResult(), + rowRange), this::createBlobViewPrescanRead); return reader; } @@ -224,10 +237,13 @@ public InnerTableRead withReadBatchSizer(ReadBatchSizer sizer) { } @Override - public RecordReader reader(Split split) throws IOException { + public RecordReader reader(Split split, @Nullable RowRange rowRange) + throws IOException { for (SplitReadProvider readProvider : readProviders) { if (readProvider.match(split, new SplitReadProvider.Context(forceKeepDelete))) { - return readProvider.get().get().createReader(split); + SplitRead read = readProvider.get().get(); + read.withRowRange(rowRange); + return read.createReader(split); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java index 930a924ae6da..56e2f7037614 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java @@ -24,10 +24,13 @@ import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.metrics.MetricRegistry; import org.apache.paimon.operation.SplitRead; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.ReaderSupplier; import org.apache.paimon.reader.RecordReader; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -64,6 +67,21 @@ default TableRead withReadBatchSizer(ReadBatchSizer sizer) { RecordReader createReader(Split split) throws IOException; + /** + * Read a single {@link Split} restricted to the given {@link RowRange}. + * + *

{@code rowRange} is expressed in the 0-based effective-row position space of this split + * (concatenated effective rows of the split's files, deletion-vector-aware). It is a + * split-local parameter applied to this read invocation. + * + * @param split the split to read + * @param rowRange the effective-row range to read, {@code null} to read the whole split + */ + default RecordReader createReader(Split split, @Nullable RowRange rowRange) + throws IOException { + return createReader(split); + } + default RecordReader createReader(List splits) throws IOException { List> readers = new ArrayList<>(); for (Split split : splits) { diff --git a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java index 891ae8380a8c..67634cd14308 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java @@ -28,6 +28,7 @@ import org.apache.paimon.fileindex.bitmap.BitmapIndexResult; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.types.DataField; @@ -145,6 +146,92 @@ public void testDataFilterIntersectsDeletionVector() throws Exception { assertThat(result).isSameAs(FileIndexResult.SKIP); } + @Test + public void testRowRangeStartsFromRangeStartNotZero() throws Exception { + // Request only the last five rows of a 100-row file. The pushed bitmap must start at 95, + // not 0, so parquet row-group skipping can avoid reading the leading row groups. + RowRange rowRange = RowRange.of(95, 99); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + null, + Collections.emptyList(), + null, + null, + rowRange, + null, + fileWithRowCount(100), + null); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + RoaringBitmap32 bitmap = ((BitmapIndexResult) result).get(); + assertThat(bitmap).isEqualTo(RoaringBitmap32.bitmapOfRange(95, 100)); + assertThat(bitmap.first()).isEqualTo(95); + } + + @Test + public void testRowRangeEndClampedToFileRowCount() throws Exception { + // endInclusive beyond the file is clamped to the last physical row index. + RowRange rowRange = RowRange.of(95, Long.MAX_VALUE); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + null, + Collections.emptyList(), + null, + null, + rowRange, + null, + fileWithRowCount(100), + null); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()) + .isEqualTo(RoaringBitmap32.bitmapOfRange(95, 100)); + } + + @Test + public void testRowRangeBeyondFileSkips() throws Exception { + // start past the end of the file -> empty selection (no rows match). + RowRange rowRange = RowRange.of(100, 110); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + null, + Collections.emptyList(), + null, + null, + rowRange, + null, + fileWithRowCount(100), + null); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(result.remain()).isFalse(); + assertThat(((BitmapIndexResult) result).get().isEmpty()).isTrue(); + } + + @Test + public void testFullRowRangeRemains() throws Exception { + // No filter/topN/limit and a FULL range -> REMAIN (no pruning needed). + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + null, + Collections.emptyList(), + null, + null, + null, + null, + fileWithRowCount(100), + null); + + assertThat(result).isSameAs(FileIndexResult.REMAIN); + } + private static TableSchema tableSchema() { return new TableSchema( 1, diff --git a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java index 0ee799c03239..65ca46ed4470 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java @@ -52,6 +52,7 @@ import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; @@ -2850,4 +2851,129 @@ public void testMergePlainBranchSucceedsWithCompleteHistory() throws Exception { "1|10|100|binary|varbinary|mapKey:mapVal|multiset", "2|20|200|binary|varbinary|mapKey:mapVal|multiset"); } + + // -------------------------------------------------------------------------------------------- + // RowRange (effective-row slice) read via TableRead::createReader(Split, RowRange). + // An append table pushes the local range into parquet as a selection bitmap; orc falls back to + // a single outer RangeSkipReader. Rows are written in order (pt=0, a=0..N-1, b=a*10). + // -------------------------------------------------------------------------------------------- + + private void writeAppendRows(FileStoreTable table, int from, int to) throws Exception { + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + for (int i = from; i < to; i++) { + write.write(rowData(0, i, (long) i * 10)); + } + commit.commit(0, write.prepareCommit(true, 0)); + } + } + + /** Parquet pushdown: RowRange [2, 4] over one parquet file returns a = 2, 3, 4. */ + @Test + public void testAppendParquetRowRangeReturnsExactSlice() throws Exception { + FileStoreTable table = + createFileStoreTable(conf -> conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET)); + writeAppendRows(table, 0, 10); + // a = 0..9 ; RowRange [2, 4] -> a = 2,3,4 + List actual = readAppendRowRange(table, RowRange.of(2L, 4L), null, false); + assertThat(actual).containsExactly("0|2|20", "0|3|30", "0|4|40"); + } + + /** + * ORC fallback: orc does not support row-range skipping, so the range is enforced by an outer + * RangeSkipReader with the real skip (not 0). RowRange [2, 4] returns a = 2, 3, 4 — guards the + * OrcReaderFactory supportsRowRangeSkip()=false fix. + */ + @Test + public void testAppendOrcRowRangeFallsBackToSkipAndLimit() throws Exception { + FileStoreTable table = + createFileStoreTable(conf -> conf.set(FILE_FORMAT, CoreOptions.FILE_FORMAT_ORC)); + writeAppendRows(table, 0, 10); + List actual = readAppendRowRange(table, RowRange.of(2L, 4L), null, false); + assertThat(actual).containsExactly("0|2|20", "0|3|30", "0|4|40"); + } + + /** + * A range read with executeFilter: rowRange slices the *filtered* output, so the + * RangeSkipReader wraps outside the filter (AbstractDataTableRead.outerWrap). Filter a >= 5 + * keeps a = 5..9 (5 effective rows), RowRange [1, 2] -> a = 6, 7. + */ + @Test + public void testAppendParquetRowRangeWithFilterWrapsOutsideFilter() throws Exception { + FileStoreTable table = + createFileStoreTable(conf -> conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET)); + writeAppendRows(table, 0, 10); + // field "a" is index 1 ; filtered stream: a = 5..9 ; RowRange [1, 2] -> a = 6, 7 + Predicate filter = new PredicateBuilder(table.rowType()).greaterOrEqual(1, 5); + List actual = readAppendRowRange(table, RowRange.of(1L, 2L), filter, true); + assertThat(actual).containsExactly("0|6|60", "0|7|70"); + } + + /** + * A range spanning two files (two commits, each writing 5 rows). file0: a = 0..4 ; file1: a = + * 5..9. RowRange [3, 6] -> file0 a = 3,4 + file1 a = 5,6. + */ + @Test + public void testAppendParquetRowRangeAcrossMultipleFiles() throws Exception { + FileStoreTable table = + createFileStoreTable(conf -> conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET)); + writeAppendRows(table, 0, 5); + writeAppendRows(table, 5, 10); + // global effective rows: file0 -> [0,4], file1 -> [5,9] ; RowRange [3, 6] -> a = 3,4,5,6 + List actual = readAppendRowRange(table, RowRange.of(3L, 6L), null, false); + assertThat(actual).containsExactly("0|3|30", "0|4|40", "0|5|50", "0|6|60"); + } + + /** + * Regression: a split mixing formats after an append table changes from parquet to orc. The + * range pushdown check must cover every file: if only the first (parquet) file is checked, + * canPushdown is true, the outer RangeSkipReader is disabled, and the orc file (which cannot + * push down) returns its whole file — leaking rows beyond the range. With the all-files check, + * canPushdown is false and the single outer RangeSkipReader slices the concatenated stream. + * + *

file0 (parquet): a = 0..4 ; file1 (orc): a = 5..9 ; RowRange [3, 6] -> a = 3,4,5,6. + */ + @Test + public void testAppendMixedFormatRowRangeSlicesConcatenatedStream() throws Exception { + FileStoreTable table = + createFileStoreTable(conf -> conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET)); + writeAppendRows(table, 0, 5); + // switch format to orc and write the next 5 rows into the same table (mixed-format split) + FileStoreTable orcTable = + table.copy(Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "orc")); + writeAppendRows(orcTable, 5, 10); + // global effective rows: parquet file0 -> [0,4], orc file1 -> [5,9] ; RowRange [3, 6] + List actual = readAppendRowRange(table, RowRange.of(3L, 6L), null, false); + assertThat(actual).containsExactly("0|3|30", "0|4|40", "0|5|50", "0|6|60"); + } + + /** + * Reads the "pt|a|b" projection of a slice via {@code TableRead::createReader(Split, + * RowRange)}. + */ + private List readAppendRowRange( + FileStoreTable table, RowRange rowRange, Predicate filter, boolean filterExecute) + throws Exception { + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0, 1, 2}); + if (filter != null) { + readBuilder = readBuilder.withFilter(filter); + } + TableRead read = readBuilder.newRead(); + if (filterExecute) { + // executeFilter() makes the reader evaluate the filter itself, so rowRange wraps + // outside the filter (AbstractDataTableRead.outerWrap), not forwarded to the inner + // reader. + read = read.executeFilter(); + } + List splits = toSplits(table.newSnapshotReader().read().dataSplits()); + Function toString = + r -> r.getInt(0) + "|" + r.getInt(1) + "|" + r.getLong(2); + List result = new ArrayList<>(); + for (Split split : splits) { + try (RecordReader reader = read.createReader(split, rowRange)) { + reader.forEachRemaining(row -> result.add(toString.apply(row))); + } + } + return result; + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java index 2cd688c9e9e1..f6dc0691c051 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java @@ -38,6 +38,7 @@ import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.DataEvolutionFileReader; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.Schema; @@ -53,6 +54,7 @@ import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.StreamTableScan; +import org.apache.paimon.table.source.TableRead; import org.apache.paimon.table.source.TableScan; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; @@ -2524,6 +2526,170 @@ public void testDropStatsWithoutFilterOrReadType() throws Exception { } } + // -------------------------------------------------------------------------------------------- + // RowRange (effective-row slice) read via the TableRead::createReader(Split, RowRange) entry + // -------------------------------------------------------------------------------------------- + + /** + * A full-scan range read pushes the local range down to the parquet readers of the column-merge + * group: each bunch returns its local slice and the merged output is exactly [start, end]. This + * exercises the full chain {@code TableRead -> AbstractDataTableRead -> DataEvolutionTableRead + * -> DataEvolutionSplitRead -> createReader(Split)} with parquet row-group skipping. + */ + @Test + public void testRowRangeReadReturnsExactSlice() throws Exception { + int count = 100; + write(count); + // write() produces f0 = 0..count-1, so RowRange [10, 19] -> f0 = 10..19 + List actual = readRowRange(getTableDefault(), RowRange.of(10L, 19L)); + assertThat(actual).isEqualTo(intRange(10, 19)); + } + + /** + * A range read combined with a filter: rowRange slices the *filtered* effective-row stream, so + * the RangeSkipReader wraps outside the executeFilter layer (AbstractDataTableRead.outerWrap). + * Filter f0 >= 50 keeps f0 = 50..99 (50 effective rows), RowRange [10, 19] over those selects + * f0 = 60..69. + */ + @Test + public void testRowRangeWithFilterWrapsOutsideFilter() throws Exception { + int count = 100; + write(count); + PredicateBuilder builder = new PredicateBuilder(schemaDefault().rowType()); + Predicate filter = builder.greaterOrEqual(0, 50); + // filtered stream: f0 = 50..99 ; RowRange [10, 19] -> f0 = 60..69 + List actual = readRowRange(getTableDefault(), RowRange.of(10L, 19L), filter, true); + assertThat(actual).isEqualTo(intRange(60, 69)); + } + + /** + * A range read spanning multiple files (several merge groups) is sliced in the split-global + * effective-row space: each file gets its own local range and files entirely outside are + * skipped. Writing twice creates two files of `count` rows each, so RowRange [count-5, count+5] + * straddles the file boundary. + */ + @Test + public void testRowRangeAcrossMultipleFiles() throws Exception { + int count = 100; + write(count); + write(count); + // file0: f0 = 0..99 ; file1: f0 = 0..99 (each file-internal index 0..99). + // global effective rows: file0 -> [0, 99], file1 -> [100, 199]. + // RowRange [95, 104] -> file0 last 5 (f0 95..99) + file1 first 5 (f0 0..4). + List actual = readRowRange(getTableDefault(), RowRange.of(95L, 104L)); + List expected = new ArrayList<>(); + for (int i = 95; i < 100; i++) { + expected.add(i); // file0 + } + for (int i = 0; i <= 4; i++) { + expected.add(i); // file1 + } + assertThat(actual).isEqualTo(expected); + } + + /** + * Regression: a partial-column ORC read over a DataEvolution column merge. ORC cannot push a + * row range down (supportsRowRangeSkip() == false), so the range must be enforced exactly once + * by an outer RangeSkipReader, not both by a selection bitmap (ApplyBitmapIndexRecordReader) + * and a RangeSkipReader — the double application used to yield [] for RowRange.of(10, 19). + * + *

The table is stored as two ORC column files (f0+f1 in one, f2 in another) so the read goes + * through the column-merge / DataBunch path; the existing parquet case does not cover this + * branch. + */ + @Test + public void testRowRangeOrcColumnMergeDoesNotDoubleApply() throws Exception { + int count = 100; + // build an ORC data-evolution table with a column-merge layout (f0+f1 | f2) + Schema schema = schemaDefault(); + Map orcOptions = new HashMap<>(schema.options()); + orcOptions.put(CoreOptions.FILE_FORMAT.key(), "orc"); + Schema orcSchema = + new Schema( + schema.rowType().getFields(), + schema.partitionKeys(), + schema.primaryKeys(), + orcOptions, + schema.comment()); + catalog.createTable(identifier(), orcSchema, true); + FileStoreTable table = getTableDefault(); + writeColumnMerge(table, count); + + // f0 = 0..99 ; RowRange [10, 19] -> f0 = 10..19 (must NOT be empty) + List actual = readRowRange(table, RowRange.of(10L, 19L)); + assertThat(actual).isEqualTo(intRange(10, 19)); + } + + /** + * Writes {@code count} rows split across two column files (f0+f1, then f2) — a column merge. + */ + private void writeColumnMerge(FileStoreTable table, int count) throws Exception { + Schema schema = schemaDefault(); + RowType writeType0 = schema.rowType().project(Arrays.asList("f0", "f1")); + RowType writeType1 = schema.rowType().project(Collections.singletonList("f2")); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write0 = builder.newWrite().withWriteType(writeType0)) { + for (int i = 0; i < count; i++) { + write0.write(GenericRow.of(i, BinaryString.fromString("a" + i))); + } + BatchTableCommit commit = builder.newCommit(); + commit.commit(write0.prepareCommit()); + } + long rowId = table.snapshotManager().latestSnapshot().nextRowId() - count; + builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write1 = builder.newWrite().withWriteType(writeType1)) { + for (int i = 0; i < count; i++) { + write1.write(GenericRow.of(BinaryString.fromString("b" + i))); + } + BatchTableCommit commit = builder.newCommit(); + List commitables = write1.prepareCommit(); + setFirstRowId(commitables, rowId); + commit.commit(commitables); + } + } + + /** Reads the f0 column of a slice via {@code TableRead::createReader(Split, RowRange)}. */ + private List readRowRange(FileStoreTable table, RowRange rowRange) throws Exception { + return readRowRange(table, rowRange, null, false); + } + + private List readRowRange(FileStoreTable table, RowRange rowRange, Predicate filter) + throws Exception { + return readRowRange(table, rowRange, filter, false); + } + + private List readRowRange( + FileStoreTable table, RowRange rowRange, Predicate filter, boolean filterExecute) + throws Exception { + ReadBuilder readBuilder = table.newReadBuilder(); + if (filter != null) { + readBuilder = readBuilder.withFilter(filter); + } + TableRead read = readBuilder.newRead(); + if (filterExecute) { + // executeFilter() makes the reader evaluate the filter itself, so the rowRange must be + // wrapped outside the filter (AbstractDataTableRead.outerWrap), not forwarded into the + // inner reader — verifying that wrapping, not pushdown, is used in this case. + read = read.executeFilter(); + } + List splits = readBuilder.newScan().plan().splits(); + List result = new ArrayList<>(); + for (Split split : splits) { + try (RecordReader reader = read.createReader(split, rowRange)) { + reader.forEachRemaining(row -> result.add(row.getInt(0))); + } + } + return result; + } + + private static List intRange(int from, int toInclusive) { + List list = new ArrayList<>(); + for (int i = from; i <= toInclusive; i++) { + list.add(i); + } + return list; + } + private List writeOneFullRowAndCollectNewFiles(FileStoreTable table) throws Exception { Schema schema = schemaDefault(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index 6a7a722e855b..8c26b053b473 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java @@ -47,6 +47,7 @@ import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.TopN; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; @@ -3806,4 +3807,109 @@ private List wrapWithAuth(List splits, TableQueryAuthResult authRe .map(split -> new QueryAuthSplit(split, authResult)) .collect(Collectors.toList()); } + + // -------------------------------------------------------------------------------------------- + // RowRange (effective-row slice) read via TableRead::createReader(Split, RowRange). + // A primary-key table cannot push a row range into the formats (the merge-tree reorders rows), + // so the range is enforced by a single outer RangeSkipReader over the merged output stream + // (MergeFileSplitRead.createReader). + // -------------------------------------------------------------------------------------------- + + /** + * Basic range slice: write 10 rows (pk a = 0..9, pt = 0), the merged output is sorted by pk + * (pt, a), so RowRange [2, 4] selects pk a = 2, 3, 4. + */ + @Test + public void testPrimaryKeyRowRangeReturnsExactSlice() throws Exception { + FileStoreTable table = createFileStoreTable(); + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + for (int i = 0; i < 10; i++) { + write.write(rowData(0, i, (long) i * 10)); + } + commit.commit(0, write.prepareCommit(true, 0)); + } + + // merged output sorted by pk (pt=0, a): a = 0..9 ; RowRange [2, 4] -> a = 2,3,4 + List actual = readRowRange(table, RowRange.of(2L, 4L), null, false); + assertThat(actual).containsExactly("0|2|20", "0|3|30", "0|4|40"); + } + + /** + * Range slice over a merge read: the same primary key is updated across two commits, so reading + * must merge the versions (keep the latest) before slicing. RowRange covering the updated key + * returns the merged (latest) value, proving the RangeSkipReader runs over the *merged* output, + * not the raw files. + */ + @Test + public void testPrimaryKeyRowRangeOverMergeRead() throws Exception { + FileStoreTable table = createFileStoreTable(); + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + // initial: pk a = 0..9, b = i*10 + for (int i = 0; i < 10; i++) { + write.write(rowData(0, i, (long) i * 10)); + } + commit.commit(0, write.prepareCommit(true, 0)); + // update pk a = 5 to b = 999 (a second version -> merge keeps the latest) + write.write(rowData(0, 5, 999L)); + commit.commit(1, write.prepareCommit(true, 1)); + } + + // merged output: a = 0..9 with a=5 -> b=999 ; RowRange [3, 6] -> a = 3,4,5,6 + List actual = readRowRange(table, RowRange.of(3L, 6L), null, false); + assertThat(actual).containsExactly("0|3|30", "0|4|40", "0|5|999", "0|6|60"); + } + + /** + * A range read with executeFilter: rowRange slices the *filtered* merged output, so the + * RangeSkipReader wraps outside the filter (AbstractDataTableRead.outerWrap). Filter a >= 5 + * keeps a = 5..9 (5 effective rows), RowRange [1, 2] over those -> a = 6, 7. + */ + @Test + public void testPrimaryKeyRowRangeWithFilterWrapsOutsideFilter() throws Exception { + FileStoreTable table = createFileStoreTable(); + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + for (int i = 0; i < 10; i++) { + write.write(rowData(0, i, (long) i * 10)); + } + commit.commit(0, write.prepareCommit(true, 0)); + } + + // field "a" is index 1 ; filtered stream: a = 5..9 ; RowRange [1, 2] -> a = 6, 7 + Predicate filter = new PredicateBuilder(table.rowType()).greaterOrEqual(1, 5); + List actual = readRowRange(table, RowRange.of(1L, 2L), filter, true); + assertThat(actual).containsExactly("0|6|60", "0|7|70"); + } + + /** + * Reads the "pt|a|b" projection of a slice via {@code TableRead::createReader(Split, + * RowRange)}. + */ + private List readRowRange( + FileStoreTable table, RowRange rowRange, Predicate filter, boolean filterExecute) + throws Exception { + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0, 1, 2}); + if (filter != null) { + readBuilder = readBuilder.withFilter(filter); + } + TableRead read = readBuilder.newRead(); + if (filterExecute) { + // executeFilter() makes the reader evaluate the filter itself, so rowRange wraps + // outside the filter (AbstractDataTableRead.outerWrap), not forwarded to the inner + // reader. + read = read.executeFilter(); + } + List splits = toSplits(table.newSnapshotReader().read().dataSplits()); + Function toString = + r -> r.getInt(0) + "|" + r.getInt(1) + "|" + r.getLong(2); + List result = new ArrayList<>(); + for (Split split : splits) { + try (RecordReader reader = read.createReader(split, rowRange)) { + reader.forEachRemaining(row -> result.add(toString.apply(row))); + } + } + return result; + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java index fa1934c4fe43..1588439e3c50 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java @@ -22,6 +22,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.predicate.UpperTransform; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.TableSchema; @@ -130,7 +131,7 @@ public void applyReadType(RowType readType) { } @Override - public RecordReader reader(Split split) { + public RecordReader reader(Split split, RowRange rowRange) { return new RecordReader() { @Override public RecordIterator readBatch() { @@ -148,7 +149,7 @@ protected InnerTableRead innerWithFilter(Predicate predicate) { } private void createAuthedReader(TableQueryAuthResult authResult) throws IOException { - createDataReader(mock(Split.class), authResult); + createDataReader(mock(Split.class), authResult, null); } private RowType appliedReadType() { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupCompactDiffRead.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupCompactDiffRead.java index 0b390b8bf17b..8d798ecafaf7 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupCompactDiffRead.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupCompactDiffRead.java @@ -23,6 +23,7 @@ import org.apache.paimon.operation.MergeFileSplitRead; import org.apache.paimon.operation.SplitRead; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.RowRange; import org.apache.paimon.reader.ReadBatchSizer; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.TableSchema; @@ -34,6 +35,8 @@ import org.apache.paimon.table.source.TableRead; import org.apache.paimon.types.RowType; +import javax.annotation.Nullable; + import java.io.IOException; /** An {@link InnerTableRead} that reads the data changed before and after compaction. */ @@ -60,7 +63,9 @@ public void applyReadType(RowType readType) { } @Override - public RecordReader reader(Split split) throws IOException { + public RecordReader reader(Split split, @Nullable RowRange rowRange) + throws IOException { + // rowRange is not supported by lookup reads; ignored. if (split instanceof DataSplit) { return fullPhaseMergeRead.createReader(split); // full reading phase } else { diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java index ab7d277a6221..26b8f20af29d 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java @@ -121,6 +121,13 @@ Map requestedSchemaCache() { return requestedSchemaCache; } + @Override + public boolean supportsRowRangeSkip() { + // Parquet consumes the selection bitmap to skip whole row groups (filterRowGroups) and + // pages (offset index), enabling efficient row-range reads. + return true; + } + @Override public FileRecordReader createReader(FormatReaderFactory.Context context) throws IOException {