This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b6e0e78e52 [core] avoid reading stale blobs during fallback read.
(#9208)
b6e0e78e52 is described below
commit b6e0e78e527049f65f91f44cac255e7db862ee50
Author: Faiz <[email protected]>
AuthorDate: Thu Aug 13 22:00:33 2026 +0800
[core] avoid reading stale blobs during fallback read. (#9208)
---
.../apache/paimon/reader/FileRecordIterator.java | 20 +++
.../org/apache/paimon/reader/RecordReader.java | 18 +++
.../paimon/reader/FileRecordIteratorTest.java | 48 +++++++
.../paimon/append/ForceSingleBatchReader.java | 19 +++
.../ApplyDeletionFileRecordIterator.java | 10 ++
.../operation/AllPlaceholdersRecordReader.java | 13 +-
.../paimon/operation/BlobFallbackRecordReader.java | 20 ++-
.../paimon/deletionvectors/DeletionVectorTest.java | 27 ++++
.../operation/AllPlaceholdersRecordReaderTest.java | 23 +++
.../operation/BlobFallbackRecordReaderTest.java | 160 +++++++++++++++++++++
.../paimon/format/blob/BlobFormatReader.java | 9 ++
.../paimon/format/blob/BlobFileFormatTest.java | 63 +++++++-
12 files changed, 419 insertions(+), 11 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java
b/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java
index 6911dffe7f..dc61e586f0 100644
---
a/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java
+++
b/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java
@@ -68,6 +68,11 @@ public interface FileRecordIterator<T> extends
RecordReader.RecordIterator<T> {
return function.apply(next);
}
+ @Override
+ public boolean skip() throws IOException {
+ return thisIterator.skip();
+ }
+
@Override
public void releaseBatch() {
thisIterator.releaseBatch();
@@ -146,6 +151,21 @@ public interface FileRecordIterator<T> extends
RecordReader.RecordIterator<T> {
}
}
+ @Override
+ public boolean skip() throws IOException {
+ while (true) {
+ if (nextExpected == -1 || !thisIterator.skip()) {
+ return false;
+ }
+ while (nextExpected != -1 && nextExpected <
returnedPosition()) {
+ nextExpected = selects.hasNext() ? selects.next() : -1;
+ }
+ if (nextExpected == returnedPosition()) {
+ return true;
+ }
+ }
+ }
+
@Override
public void releaseBatch() {
thisIterator.releaseBatch();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java
b/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java
index 135c27f250..7dcbc9a554 100644
--- a/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java
@@ -71,6 +71,19 @@ public interface RecordReader<T> extends Closeable {
*/
void releaseBatch();
+ /**
+ * Consumes the next logical record without materializing it.
+ *
+ * @return true if a record was skipped, false if the iterator is
exhausted
+ * @throws UnsupportedOperationException if this iterator does not
support skipping
+ */
+ default boolean skip() throws IOException {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Iterator %s does not support skipping records.",
+ getClass().getCanonicalName()));
+ }
+
/** Returns an iterator that applies {@code function} to each element.
*/
default <R> RecordReader.RecordIterator<R> transform(Function<T, R>
function) {
RecordReader.RecordIterator<T> thisIterator = this;
@@ -85,6 +98,11 @@ public interface RecordReader<T> extends Closeable {
return function.apply(next);
}
+ @Override
+ public boolean skip() throws IOException {
+ return thisIterator.skip();
+ }
+
@Override
public void releaseBatch() {
thisIterator.releaseBatch();
diff --git
a/paimon-common/src/test/java/org/apache/paimon/reader/FileRecordIteratorTest.java
b/paimon-common/src/test/java/org/apache/paimon/reader/FileRecordIteratorTest.java
index 4fd0f44689..50001cd9f7 100644
---
a/paimon-common/src/test/java/org/apache/paimon/reader/FileRecordIteratorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/reader/FileRecordIteratorTest.java
@@ -194,6 +194,45 @@ public class FileRecordIteratorTest {
assertThat(third).isNull();
}
+ @Test
+ public void testSelectionSkip() throws IOException {
+ FileRecordIterator<Long> iterator = createIterator(Arrays.asList(0L,
1L, 2L, 3L, 4L, 5L));
+
+ RoaringBitmap32 selection = new RoaringBitmap32();
+ selection.add(1);
+ selection.add(3);
+ selection.add(5);
+
+ FileRecordIterator<Long> selected = iterator.selection(selection);
+ assertThat(selected.skip()).isTrue();
+ assertThat(selected.returnedPosition()).isEqualTo(1L);
+ assertThat(selected.next()).isEqualTo(3L);
+ assertThat(selected.returnedPosition()).isEqualTo(3L);
+ assertThat(selected.skip()).isTrue();
+ assertThat(selected.returnedPosition()).isEqualTo(5L);
+ assertThat(selected.skip()).isFalse();
+ }
+
+ @Test
+ public void testTransformSkipDoesNotApplyFunction() throws IOException {
+ int[] transformCount = {0};
+ FileRecordIterator<String> transformed =
+ createIterator(Arrays.asList(1L, 2L, 3L))
+ .transform(
+ value -> {
+ transformCount[0]++;
+ return value.toString();
+ });
+
+ assertThat(transformed.skip()).isTrue();
+ assertThat(transformCount[0]).isZero();
+ assertThat(transformed.next()).isEqualTo("2");
+ assertThat(transformCount[0]).isOne();
+ assertThat(transformed.skip()).isTrue();
+ assertThat(transformCount[0]).isOne();
+ assertThat(transformed.skip()).isFalse();
+ }
+
@Test
public void testSelectionFilePathPreserved() {
List<Long> values = Arrays.asList(0L, 1L, 2L);
@@ -231,6 +270,15 @@ public class FileRecordIteratorTest {
return values.get(position);
}
+ @Override
+ public boolean skip() {
+ if (position + 1 >= values.size()) {
+ return false;
+ }
+ position++;
+ return true;
+ }
+
@Override
public void releaseBatch() {}
};
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/ForceSingleBatchReader.java
b/paimon-core/src/main/java/org/apache/paimon/append/ForceSingleBatchReader.java
index f5a5f7f2e2..c3db10001c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/ForceSingleBatchReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/ForceSingleBatchReader.java
@@ -99,6 +99,25 @@ public class ForceSingleBatchReader implements
RecordReader<InternalRow> {
return next;
}
+ @Override
+ public boolean skip() throws IOException {
+ while (true) {
+ if (currentBatch == null) {
+ currentBatch = reader.readBatch();
+ if (currentBatch == null) {
+ return false;
+ }
+ }
+
+ if (currentBatch.skip()) {
+ return true;
+ }
+
+ currentBatch.releaseBatch();
+ currentBatch = null;
+ }
+ }
+
@Override
public void releaseBatch() {
if (currentBatch != null) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
index 39162c4783..474683685e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
@@ -75,6 +75,16 @@ public class ApplyDeletionFileRecordIterator
}
}
+ @Override
+ public boolean skip() throws IOException {
+ while (iterator.skip()) {
+ if (!deletionVector.isDeleted(returnedPosition())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public void releaseBatch() {
iterator.releaseBatch();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
index 91d706de96..f789f2833a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
@@ -143,12 +143,21 @@ class AllPlaceholdersRecordReader implements
FileRecordReader<InternalRow> {
@Nullable
@Override
public InternalRow next() {
+ return advance() ? placeholderRow(returnedRowId) : null;
+ }
+
+ @Override
+ public boolean skip() {
+ return advance();
+ }
+
+ private boolean advance() {
while (rangeIndex < selectedRanges.size()) {
Range range = selectedRanges.get(rangeIndex);
if (nextRowId <= range.to) {
returnedRowId = nextRowId;
nextRowId++;
- return placeholderRow(returnedRowId);
+ return true;
}
rangeIndex++;
@@ -156,7 +165,7 @@ class AllPlaceholdersRecordReader implements
FileRecordReader<InternalRow> {
nextRowId = selectedRanges.get(rangeIndex).from;
}
}
- return null;
+ return false;
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
index 7508a45ad8..db501c1615 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
@@ -165,14 +165,20 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
public InternalRow next() throws IOException {
InternalRow result = null;
long rowId = -1L;
- // We should always move each iterator forward
- // This may significantly increase memory usage and decrease
read efficiency
- // if `blob-as-descriptor` is disabled and many non-null blobs
are updated
- // TODO: Do not read stale records if there's a newer
non-placeholder
- // record. e.g. introduce a discard method to directly
discard the
- // next record?
+ // We should always move each iterator forward, but stale
blobs do not need to be
+ // materialized after finding a newer non-placeholder record.
for (int i = 0; i < iterators.length; i++) {
RecordIterator<InternalRow> iterator = iterators[i];
+ // If result is not null, skip all older blobs.
+ if (result != null) {
+ if (!iterator.skip()) {
+ throw new IllegalStateException(
+ "All readers of each max_seq group should
have the same number "
+ + "of records.");
+ }
+ continue;
+ }
+
InternalRow row = iterator.next();
if (row == null) {
if (i != 0) {
@@ -180,7 +186,7 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
"All readers of each max_seq group should
have the same number of records.");
}
for (int j = i + 1; j < iterators.length; j++) {
- if (iterators[j].next() != null) {
+ if (iterators[j].skip()) {
throw new IllegalStateException(
"All readers of each max_seq group
should have the same number of records.");
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
index 1800f4c332..a924edeb1f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
@@ -59,6 +59,24 @@ public class DeletionVectorTest {
assertThat(iterator.next()).isNull();
}
+ @Test
+ public void testApplyDeletionFileRecordIteratorSkip() throws Exception {
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.checkedDelete(1);
+ deletionVector.checkedDelete(3);
+
+ ApplyDeletionFileRecordIterator iterator =
+ new ApplyDeletionFileRecordIterator(
+ new TestingFileRecordIterator(5),
deletionVector::isDeleted);
+
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isZero();
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isEqualTo(2L);
+ assertThat(iterator.next().getInt(0)).isEqualTo(4);
+ assertThat(iterator.skip()).isFalse();
+ }
+
@Test
public void testApplyDeletionVectorReaderUsesOffsetAwareJudger() throws
Exception {
DeletionVector deletionVector = new BitmapDeletionVector();
@@ -226,6 +244,15 @@ public class DeletionVectorTest {
return GenericRow.of(returnedPosition);
}
+ @Override
+ public boolean skip() {
+ if (nextPosition >= rows) {
+ return false;
+ }
+ returnedPosition = nextPosition++;
+ return true;
+ }
+
@Override
public void releaseBatch() {}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
index 7fbf912a64..203d6cdf94 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
@@ -90,6 +90,29 @@ public class AllPlaceholdersRecordReaderTest {
assertThat(reader.readBatch()).isNull();
}
+ @Test
+ public void testSkipWithRowRangePushed() throws Exception {
+ AllPlaceholdersRecordReader reader =
+ new AllPlaceholdersRecordReader(
+ 10L,
+ 10L,
+ Arrays.asList(new Range(10, 11), new Range(14, 14),
new Range(18, 19)),
+ READ_ROW_TYPE,
+ BLOB_INDEX,
+ SEQUENCE_NUMBER);
+
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isZero();
+ assertNextPlaceholder(iterator, 11L, 1L);
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isEqualTo(4L);
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isEqualTo(8L);
+ assertNextPlaceholder(iterator, 19L, 9L);
+ assertThat(iterator.skip()).isFalse();
+ }
+
private static void assertNextPlaceholder(
FileRecordIterator<InternalRow> iterator, long rowId, long
returnedPosition)
throws Exception {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
index 1e9df6938a..1c41741b70 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
@@ -52,6 +52,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -59,6 +60,7 @@ import java.util.Map;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link BlobFallbackRecordReader}. */
public class BlobFallbackRecordReaderTest {
@@ -163,6 +165,142 @@ public class BlobFallbackRecordReaderTest {
assertThat(rows.sequenceNumbers).containsExactly(2L, 1L, 2L, 1L, 1L);
}
+ @Test
+ public void testBlobFallbackRecordReaderSkipsStaleRecords() throws
Exception {
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ DataFileMeta oldFile = blobFile("old-file", 0, 3, 1);
+ Map<String, ReadCounts> counts = new HashMap<>();
+
+ ReadResult rows =
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile, oldFile),
+ file ->
+ oneRowPerBatchReader(
+ file,
+ fileRows(file, null),
+ counts.computeIfAbsent(
+ file.fileName(),
+ ignored -> new
ReadCounts())),
+ (reader, range) -> reader,
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX));
+
+ assertThat(rows.sequenceNumbers).containsExactly(2L, 2L, 2L);
+ assertThat(counts.get(newFile.fileName()).nextCount).isEqualTo(3);
+ assertThat(counts.get(newFile.fileName()).skipCount).isZero();
+ assertThat(counts.get(oldFile.fileName()).nextCount).isZero();
+ assertThat(counts.get(oldFile.fileName()).skipCount).isEqualTo(3);
+ }
+
+ @Test
+ public void testBlobFallbackRecordReaderReadsOldRecordForPlaceholder()
throws Exception {
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ DataFileMeta oldFile = blobFile("old-file", 0, 3, 1);
+ Map<String, ReadCounts> counts = new HashMap<>();
+
+ ReadResult rows =
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile, oldFile),
+ file ->
+ oneRowPerBatchReader(
+ file,
+ fileRows(file, null,
placeholderRows(newFile, 1)),
+ counts.computeIfAbsent(
+ file.fileName(),
+ ignored -> new
ReadCounts())),
+ (reader, range) -> reader,
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX));
+
+ assertThat(rows.sequenceNumbers).containsExactly(2L, 1L, 2L);
+ assertThat(counts.get(oldFile.fileName()).nextCount).isOne();
+ assertThat(counts.get(oldFile.fileName()).skipCount).isEqualTo(2);
+ }
+
+ @Test
+ public void testBlobFallbackRecordReaderDoesNotFallbackOnNull() throws
Exception {
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ DataFileMeta oldFile = blobFile("old-file", 0, 3, 1);
+ List<InternalRow> newRows = fileRows(newFile, null);
+ ((GenericRow) newRows.get(1)).setField(BLOB_INDEX, null);
+ Map<String, ReadCounts> counts = new HashMap<>();
+
+ ReadResult rows =
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile, oldFile),
+ file ->
+ oneRowPerBatchReader(
+ file,
+ file == newFile ? newRows :
fileRows(file, null),
+ counts.computeIfAbsent(
+ file.fileName(),
+ ignored -> new
ReadCounts())),
+ (reader, range) -> reader,
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX));
+
+ assertThat(rows.nullBlobRowIds).containsExactly(1L);
+ assertThat(rows.nullBlobSequenceNumbers).containsExactly(2L);
+ assertThat(counts.get(oldFile.fileName()).nextCount).isZero();
+ assertThat(counts.get(oldFile.fileName()).skipCount).isEqualTo(3);
+ }
+
+ @Test
+ public void testBlobFallbackRecordReaderFailsIfOldGroupEndsEarly() {
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ DataFileMeta oldFile = blobFile("old-file", 0, 3, 1);
+ List<InternalRow> oldRows = fileRows(oldFile, null).subList(0, 2);
+
+ assertThatThrownBy(
+ () ->
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile,
oldFile),
+ file ->
+ oneRowPerBatchReader(
+ file,
+ file == oldFile
+ ?
oldRows
+ :
fileRows(file, null)),
+ (reader, range) -> reader,
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("same number of records");
+ }
+
+ @Test
+ public void testBlobFallbackRecordReaderFailsIfOldGroupHasExtraRecord() {
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ DataFileMeta oldFile = blobFile("old-file", 0, 3, 1);
+ List<InternalRow> newRows = fileRows(newFile, null).subList(0, 2);
+
+ assertThatThrownBy(
+ () ->
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile,
oldFile),
+ file ->
+ oneRowPerBatchReader(
+ file,
+ file == newFile
+ ?
newRows
+ :
fileRows(file, null)),
+ (reader, range) -> reader,
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("same number of records");
+ }
+
@Test
public void testBlobFallbackRecordReaderDerivesRowIdBoundsFromFiles()
throws Exception {
DataFileMeta newFile = blobFile("new-file", 10, 2, 2);
@@ -577,6 +715,11 @@ public class BlobFallbackRecordReaderTest {
private static FileRecordReader<InternalRow> oneRowPerBatchReader(
DataFileMeta file, List<InternalRow> rows) {
+ return oneRowPerBatchReader(file, rows, new ReadCounts());
+ }
+
+ private static FileRecordReader<InternalRow> oneRowPerBatchReader(
+ DataFileMeta file, List<InternalRow> rows, ReadCounts counts) {
return new FileRecordReader<InternalRow>() {
int index;
@@ -609,9 +752,21 @@ public class BlobFallbackRecordReaderTest {
}
returned = true;
returnedPosition = row.getLong(1) -
file.nonNullFirstRowId();
+ counts.nextCount++;
return row;
}
+ @Override
+ public boolean skip() {
+ if (returned) {
+ return false;
+ }
+ returned = true;
+ returnedPosition = row.getLong(1) -
file.nonNullFirstRowId();
+ counts.skipCount++;
+ return true;
+ }
+
@Override
public void releaseBatch() {}
};
@@ -622,6 +777,11 @@ public class BlobFallbackRecordReaderTest {
};
}
+ private static class ReadCounts {
+ private int nextCount;
+ private int skipCount;
+ }
+
private static class ReadResult {
final List<Long> rowIds = new ArrayList<>();
final List<Long> sequenceNumbers = new ArrayList<>();
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java
index 2bb96c79f7..1ef0ae113e 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java
@@ -110,6 +110,15 @@ public class BlobFormatReader implements
FileRecordReader<InternalRow> {
return row;
}
+ @Override
+ public boolean skip() {
+ if (currentPosition >= fileMeta.recordNumber()) {
+ return false;
+ }
+ currentPosition++;
+ return true;
+ }
+
@Override
public void releaseBatch() {}
};
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
index a9a2d2f99c..4134537a19 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
@@ -43,6 +43,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.reader.FileRecordIterator;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -95,6 +96,53 @@ public class BlobFileFormatTest {
innerTest(false);
}
+ @Test
+ public void testSkipDoesNotReadBlobPayload() throws IOException {
+ BlobFileFormat format =
+ new BlobFileFormat(false,
BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE);
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ FormatWriter writer =
format.createWriterFactory(rowType).create(out, null);
+ writer.addElement(GenericRow.of(new BlobData("first".getBytes())));
+ writer.addElement(GenericRow.of(new
BlobData("second".getBytes())));
+ writer.addElement(GenericRow.of(new BlobData("third".getBytes())));
+ writer.close();
+ }
+
+ RoaringBitmap32 selection = new RoaringBitmap32();
+ selection.add(0);
+ selection.add(1);
+ selection.add(2);
+ TrackingLocalFileIO trackingFileIO = new TrackingLocalFileIO();
+ FormatReaderFactory readerFactory = format.createReaderFactory(null,
rowType, null);
+ FormatReaderContext context =
+ new FormatReaderContext(
+ trackingFileIO, file,
trackingFileIO.getFileSize(file), selection);
+
+ try (FileRecordReader<InternalRow> reader =
readerFactory.createReader(context)) {
+ TrackingSeekableInputStream stream =
trackingFileIO.lastInputStream;
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+
+ int readCount = stream.readCount;
+ int seekCount = stream.seekCount;
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isZero();
+ assertThat(stream.readCount).isEqualTo(readCount);
+ assertThat(stream.seekCount).isEqualTo(seekCount);
+
+
assertThat(iterator.next().getBlob(0).toData()).isEqualTo("second".getBytes());
+ assertThat(iterator.returnedPosition()).isOne();
+
+ readCount = stream.readCount;
+ seekCount = stream.seekCount;
+ assertThat(iterator.skip()).isTrue();
+ assertThat(iterator.returnedPosition()).isEqualTo(2L);
+ assertThat(stream.readCount).isEqualTo(readCount);
+ assertThat(stream.seekCount).isEqualTo(seekCount);
+ assertThat(iterator.skip()).isFalse();
+ }
+ }
+
@Test
public void testRawDescriptorReaderDoesNotOwnInputStream() throws
IOException {
BlobFileFormat format = new BlobFileFormat(true,
BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE);
@@ -1131,6 +1179,8 @@ public class BlobFileFormatTest {
private final SeekableInputStream delegate;
private int closeCount;
+ private int readCount;
+ private int seekCount;
private TrackingSeekableInputStream(SeekableInputStream delegate) {
this.delegate = delegate;
@@ -1138,6 +1188,7 @@ public class BlobFileFormatTest {
@Override
public void seek(long desired) throws IOException {
+ seekCount++;
delegate.seek(desired);
}
@@ -1148,12 +1199,20 @@ public class BlobFileFormatTest {
@Override
public int read() throws IOException {
- return delegate.read();
+ int value = delegate.read();
+ if (value >= 0) {
+ readCount++;
+ }
+ return value;
}
@Override
public int read(byte[] bytes, int offset, int length) throws
IOException {
- return delegate.read(bytes, offset, length);
+ int read = delegate.read(bytes, offset, length);
+ if (read > 0) {
+ readCount += read;
+ }
+ return read;
}
@Override
