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 14d3dee2f4 [core][vector] Read vectored ranges into caller-provided
buffers (#8799)
14d3dee2f4 is described below
commit 14d3dee2f40eb67548501ac516e5bdc5b384a1c4
Author: shyjsarah <[email protected]>
AuthorDate: Wed Jul 22 19:06:39 2026 +0800
[core][vector] Read vectored ranges into caller-provided buffers (#8799)
`NativeVectorGlobalIndexReader` receives destination buffers from the
native vector index callback. The current vectored-read path allocates a
second byte array for every range and then copies the data into those
destination buffers.
---
.../main/java/org/apache/paimon/fs/FileRange.java | 28 ++++++++++++++++++++++
.../org/apache/paimon/fs/VectoredReadUtils.java | 15 ++++++++----
.../apache/paimon/fs/VectoredReadUtilsTest.java | 21 ++++++++++++++++
.../index/NativeVectorGlobalIndexReader.java | 7 +++---
.../index/SeekableStreamVectorIndexInputTest.java | 8 +++++++
5 files changed, 71 insertions(+), 8 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileRange.java
b/paimon-common/src/main/java/org/apache/paimon/fs/FileRange.java
index 2c85ae5006..be2869ef5e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/FileRange.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileRange.java
@@ -20,6 +20,8 @@ package org.apache.paimon.fs;
import java.util.concurrent.CompletableFuture;
+import static java.util.Objects.requireNonNull;
+
/* This file is based on source code from the Hadoop Project
(http://hadoop.apache.org/), licensed by the Apache
* Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
* additional information regarding copyright ownership. */
@@ -47,12 +49,24 @@ public interface FileRange {
return new FileRangeImpl(offset, length);
}
+ /**
+ * Factory method to create a FileRange object backed by a caller-provided
buffer.
+ *
+ * @param offset starting offset of the range.
+ * @param buffer buffer to store the data for this range.
+ * @return a new instance of FileRangeImpl.
+ */
+ static FileRange createFileRange(long offset, byte[] buffer) {
+ return new FileRangeImpl(offset, buffer);
+ }
+
/** An implementation for {@link FileRange}. */
class FileRangeImpl implements FileRange {
private final long offset;
private final int length;
private final CompletableFuture<byte[]> reader;
+ private byte[] buffer;
public FileRangeImpl(long offset, int length) {
this.offset = offset;
@@ -60,6 +74,13 @@ public interface FileRange {
this.reader = new CompletableFuture<>();
}
+ public FileRangeImpl(long offset, byte[] buffer) {
+ this.offset = offset;
+ this.buffer = requireNonNull(buffer, "buffer is null");
+ this.length = this.buffer.length;
+ this.reader = new CompletableFuture<>();
+ }
+
@Override
public String toString() {
return "range[" + offset + "," + (offset + length) + ")";
@@ -79,5 +100,12 @@ public interface FileRange {
public CompletableFuture<byte[]> getData() {
return reader;
}
+
+ byte[] getOrCreateBuffer() {
+ if (buffer == null) {
+ buffer = new byte[length];
+ }
+ return buffer;
+ }
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java
b/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java
index 8f1551bff9..e0933d76e1 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java
@@ -181,7 +181,7 @@ public class VectoredReadUtils {
private static void fallbackToReadSequence(
SeekableInputStream in, List<? extends FileRange> ranges) throws
IOException {
for (FileRange range : ranges) {
- byte[] bytes = new byte[range.getLength()];
+ byte[] bytes = getOrCreateBuffer(range);
in.seek(range.getOffset());
IOUtils.readFully(in, bytes);
range.getData().complete(bytes);
@@ -190,13 +190,13 @@ public class VectoredReadUtils {
private static void readSingleRange(VectoredReadable readable, FileRange
range) {
if (range.getLength() == 0) {
- range.getData().complete(new byte[0]);
+ range.getData().complete(getOrCreateBuffer(range));
return;
}
try {
long position = range.getOffset();
int length = range.getLength();
- byte[] buffer = new byte[length];
+ byte[] buffer = getOrCreateBuffer(range);
readable.preadFully(position, buffer, 0, length);
range.getData().complete(buffer);
} catch (Exception ex) {
@@ -212,7 +212,7 @@ public class VectoredReadUtils {
}
long offset = combinedRange.offset;
for (FileRange fileRange : combinedRange.underlying) {
- byte[] buffer = new byte[fileRange.getLength()];
+ byte[] buffer = getOrCreateBuffer(fileRange);
copyMultiBytesToBytes(
segments,
(int) (fileRange.getOffset() - offset),
@@ -222,6 +222,13 @@ public class VectoredReadUtils {
}
}
+ private static byte[] getOrCreateBuffer(FileRange range) {
+ if (range instanceof FileRange.FileRangeImpl) {
+ return ((FileRange.FileRangeImpl) range).getOrCreateBuffer();
+ }
+ return new byte[range.getLength()];
+ }
+
private static void completeFileRangesExceptionally(
CombinedRange combinedRange, Throwable throwable) {
for (FileRange fileRange : combinedRange.underlying) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/VectoredReadUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/VectoredReadUtilsTest.java
index 1cd5476e63..bd66f3f284 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fs/VectoredReadUtilsTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fs/VectoredReadUtilsTest.java
@@ -116,6 +116,27 @@ class VectoredReadUtilsTest {
FileRange.createFileRange(7300, 850)));
}
+ @Test
+ public void testReadIntoProvidedBuffers() throws Exception {
+ byte[] first = new byte[60];
+ byte[] second = new byte[90];
+ byte[] third = new byte[200];
+ List<FileRange> ranges =
+ Arrays.asList(
+ FileRange.createFileRange(0, first),
+ FileRange.createFileRange(100, second),
+ FileRange.createFileRange(300, third));
+
+ VectoredReadUtils.readVectored(readable, ranges);
+
+ assertThat(ranges.get(0).getData().get()).isSameAs(first);
+ assertThat(ranges.get(1).getData().get()).isSameAs(second);
+ assertThat(ranges.get(2).getData().get()).isSameAs(third);
+ assertThat(first).isEqualTo(Arrays.copyOfRange(bytes, 0, 60));
+ assertThat(second).isEqualTo(Arrays.copyOfRange(bytes, 100, 190));
+ assertThat(third).isEqualTo(Arrays.copyOfRange(bytes, 300, 500));
+ }
+
@Test
public void testRandom() throws Exception {
List<FileRange> ranges = new ArrayList<>();
diff --git
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexReader.java
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexReader.java
index 1f0b46fca9..b525fdd226 100644
---
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexReader.java
+++
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexReader.java
@@ -419,7 +419,7 @@ public class NativeVectorGlobalIndexReader implements
GlobalIndexReader {
throws IOException {
List<FileRange> ranges = new ArrayList<>(positions.length);
for (int i = 0; i < positions.length; i++) {
- ranges.add(FileRange.createFileRange(positions[i],
buffers[i].length));
+ ranges.add(FileRange.createFileRange(positions[i],
buffers[i]));
}
VectoredReadUtils.ReadOptions options =
@@ -430,9 +430,8 @@ public class NativeVectorGlobalIndexReader implements
GlobalIndexReader {
.withSequentialReadFallback(false);
VectoredReadUtils.readVectored(readable, ranges, options);
- for (int i = 0; i < ranges.size(); i++) {
- byte[] bytes = ranges.get(i).getData().join();
- System.arraycopy(bytes, 0, buffers[i], 0, bytes.length);
+ for (FileRange range : ranges) {
+ range.getData().join();
}
}
diff --git
a/paimon-vector/src/test/java/org/apache/paimon/vector/index/SeekableStreamVectorIndexInputTest.java
b/paimon-vector/src/test/java/org/apache/paimon/vector/index/SeekableStreamVectorIndexInputTest.java
index 82eeee198e..94c3a4cd2b 100644
---
a/paimon-vector/src/test/java/org/apache/paimon/vector/index/SeekableStreamVectorIndexInputTest.java
+++
b/paimon-vector/src/test/java/org/apache/paimon/vector/index/SeekableStreamVectorIndexInputTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.fs.VectoredReadable;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -48,6 +49,10 @@ public class SeekableStreamVectorIndexInputTest {
assertThat(input.positionReads).hasValue(2);
assertThat(input.sequentialReads).hasValue(0);
assertThat(input.maxActiveReads).hasValue(2);
+ assertThat(input.positionReadBuffers.stream().anyMatch(buffer ->
buffer == buffers[0]))
+ .isTrue();
+ assertThat(input.positionReadBuffers.stream().anyMatch(buffer ->
buffer == buffers[1]))
+ .isTrue();
}
@Test
@@ -90,6 +95,8 @@ public class SeekableStreamVectorIndexInputTest {
private final AtomicInteger positionReads = new AtomicInteger();
private final AtomicInteger sequentialReads = new AtomicInteger();
private final AtomicInteger maxActiveReads = new AtomicInteger();
+ private final CopyOnWriteArrayList<byte[]> positionReadBuffers =
+ new CopyOnWriteArrayList<>();
private int position;
@@ -129,6 +136,7 @@ public class SeekableStreamVectorIndexInputTest {
@Override
public int pread(long position, byte[] buffer, int offset, int length)
throws IOException {
+ positionReadBuffers.add(buffer);
int active = activeReads.incrementAndGet();
maxActiveReads.accumulateAndGet(active, Math::max);
readsStarted.countDown();
