From d789aef5ccfd6f9a75d932420300ae7d89582cb9 Mon Sep 17 00:00:00 2001 From: ashishk Date: Mon, 1 Dec 2025 10:33:17 +0530 Subject: [PATCH 1/5] HDDS-13660. Ozone client support readVectored. --- .../hdds/scm/storage/ExtendedInputStream.java | 1 + .../scm/storage/MultipartInputStream.java | 74 ++++++++++++ .../ozone/client/io/OzoneInputStream.java | 24 ++++ .../fs/ozone/TestOzoneFSInputStream.java | 109 ++++++++++++++++++ .../hadoop/fs/ozone/OzoneFSInputStream.java | 60 +++++++++- 5 files changed, 266 insertions(+), 2 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java index 0ff7f1fdd787..0c8f0393165a 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java @@ -94,6 +94,7 @@ public boolean hasCapability(String capability) { switch (StringUtils.toLowerCase(capability)) { case StreamCapabilities.READBYTEBUFFER: case StreamCapabilities.UNBUFFER: + case StreamCapabilities.VECTOREDIO: return true; default: return false; diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index e48b704aade2..aecf2fbe02fe 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -22,9 +22,13 @@ import java.io.EOFException; import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; import org.apache.hadoop.fs.FSExceptionMessages; +import org.apache.hadoop.fs.FileRange; /** * A stream for accessing multipart streams. @@ -232,6 +236,76 @@ public synchronized long skip(long n) throws IOException { return toSkip; } + /** + * Implements vectored read for multipart input stream. + * This method reads multiple byte ranges asynchronously, potentially + * from different underlying part streams. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + */ + public void readVectored(List ranges, + IntFunction allocate) throws IOException { + checkOpen(); + if (!initialized) { + initialize(); + } + + // Perform vectored read using positioned read operations + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + if (result == null) { + result = new CompletableFuture<>(); + range.setData(result); + } + + final CompletableFuture finalResult = result; + final long offset = range.getOffset(); + final int length = range.getLength(); + + // Submit async read task for this range + CompletableFuture.runAsync(() -> { + try { + ByteBuffer buffer = allocate.apply(length); + readRangeData(offset, buffer); + buffer.flip(); + finalResult.complete(buffer); + } catch (Exception e) { + finalResult.completeExceptionally(e); + } + }); + } + } + + /** + * Helper method to read data for a specific range. + * Uses synchronized seeks to read data from the correct position. + * + * @param offset the starting offset in the stream + * @param buffer the buffer to read data into + * @throws IOException if there is an error reading data + */ + private void readRangeData(long offset, ByteBuffer buffer) throws IOException { + synchronized (this) { + long savedPos = getPos(); + try { + seek(offset); + int remaining = buffer.remaining(); + byte[] temp = new byte[remaining]; + int bytesRead = read(temp, 0, remaining); + if (bytesRead < remaining) { + throw new EOFException("Could not read all requested bytes. " + + "Requested: " + remaining + ", Read: " + bytesRead); + } + buffer.put(temp, 0, bytesRead); + } finally { + seek(savedPos); + } + } + } + + @Override public synchronized void close() throws IOException { closed = true; diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java index 2366893ef562..9439a5a12008 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java @@ -20,9 +20,13 @@ import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.util.List; +import java.util.function.IntFunction; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.Seekable; +import org.apache.hadoop.hdds.scm.storage.MultipartInputStream; /** * OzoneInputStream is used to read data from Ozone. @@ -121,4 +125,24 @@ public boolean seekToNewSource(long targetPos) throws IOException { "underlying inputStream"); } } + + /** + * Read data from multiple byte ranges asynchronously. + * This allows reading multiple discontiguous ranges from the same file + * efficiently with a single API call. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + */ + public void readVectored(List ranges, + IntFunction allocate) throws IOException { + if (inputStream instanceof MultipartInputStream) { + ((MultipartInputStream) inputStream).readVectored(ranges, allocate); + } else { + throw new UnsupportedOperationException("Vectored read is not supported " + + "on the underlying inputStream of type " + + inputStream.getClass().getName()); + } + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java index e039e14e47d0..1143bc5e6d11 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java @@ -32,10 +32,15 @@ import java.net.URI; import java.nio.ByteBuffer; import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.RandomStringUtils; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hdds.client.DefaultReplicationConfig; @@ -324,4 +329,108 @@ public void testSequenceFileReaderSyncEC() throws IOException { assertEquals(srcfile.length(), blockStart); in.close(); } + + @Test + public void testVectoredRead() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Create multiple file ranges to read + List ranges = new ArrayList<>(); + + // Read from different parts of the file + ranges.add(FileRange.createFileRange(0L, 100)); // First 100 bytes + ranges.add(FileRange.createFileRange(500L, 200)); // 200 bytes from offset 500 + ranges.add(FileRange.createFileRange(1024L, 512)); // 512 bytes from offset 1024 + ranges.add(FileRange.createFileRange(10L * 1024, 1024)); // 1KB from offset 10KB + + // Perform vectored read + inputStream.readVectored(ranges, ByteBuffer::allocate); + + // Wait for all reads to complete and validate + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + + assertEquals(range.getLength(), buffer.remaining(), + "Buffer size mismatch for range at offset " + range.getOffset()); + + // Validate data by comparing with sequential read + byte[] vectoredData = new byte[buffer.remaining()]; + buffer.get(vectoredData); + + byte[] expectedData = new byte[range.getLength()]; + System.arraycopy(data, (int)range.getOffset(), expectedData, 0, range.getLength()); + + assertArrayEquals(expectedData, vectoredData, + "Data mismatch for range at offset " + range.getOffset()); + } + } + } + + @Test + public void testVectoredReadWithOverlappingRanges() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Create overlapping ranges + List ranges = new ArrayList<>(); + ranges.add(FileRange.createFileRange(100L, 500)); + ranges.add(FileRange.createFileRange(300L, 400)); // Overlaps with first range + ranges.add(FileRange.createFileRange(600L, 200)); + + inputStream.readVectored(ranges, ByteBuffer::allocate); + + // Validate all reads complete successfully + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + assertEquals(range.getLength(), buffer.remaining()); + } + } + } + + @Test + public void testVectoredReadAcrossMultipleBlocks() throws Exception { + // This test reads ranges that span across different blocks in the file + try (FSDataInputStream inputStream = fs.open(filePath)) { + List ranges = new ArrayList<>(); + + // Assuming blocks are 1MB each, read ranges that cross block boundaries + ranges.add(FileRange.createFileRange(1024L * 1024 - 100, 200)); // Crosses first block boundary + ranges.add(FileRange.createFileRange(2L * 1024 * 1024 - 50, 100)); // Crosses second block boundary + + inputStream.readVectored(ranges, ByteBuffer::allocate); + + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + + byte[] vectoredData = new byte[buffer.remaining()]; + buffer.get(vectoredData); + + byte[] expectedData = new byte[range.getLength()]; + System.arraycopy(data, (int)range.getOffset(), expectedData, 0, range.getLength()); + + assertArrayEquals(expectedData, vectoredData, + "Data mismatch for range crossing block boundary at offset " + range.getOffset()); + } + } + } + + @Test + public void testVectoredReadWithSmallRanges() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Test with many small ranges + List ranges = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + ranges.add(FileRange.createFileRange((long) i * 100, 50)); + } + + inputStream.readVectored(ranges, ByteBuffer::allocate); + + // Validate all small reads + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + assertEquals(range.getLength(), buffer.remaining()); + } + } + } } diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java index e4133ae57a59..0004e2f644d3 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java @@ -22,10 +22,14 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.ReadOnlyBufferException; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.FSInputStream; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.FileSystem.Statistics; import org.apache.hadoop.fs.Seekable; import org.apache.hadoop.hdds.annotation.InterfaceAudience; @@ -164,7 +168,7 @@ public void unbuffer() { * @throws IOException if there is some error performing the read */ @Override - public int read(long position, ByteBuffer buf) throws IOException { + public synchronized int read(long position, ByteBuffer buf) throws IOException { if (!buf.hasRemaining()) { return 0; } @@ -189,7 +193,7 @@ public int read(long position, ByteBuffer buf) throws IOException { * @throws EOFException if end of file reached before reading fully */ @Override - public void readFully(long position, ByteBuffer buf) throws IOException { + public synchronized void readFully(long position, ByteBuffer buf) throws IOException { int bytesRead; for (int readCount = 0; buf.hasRemaining(); readCount += bytesRead) { bytesRead = this.read(position + (long)readCount, buf); @@ -199,4 +203,56 @@ public void readFully(long position, ByteBuffer buf) throws IOException { } } } + + /** + * Implements vectored read by reading each range asynchronously. + * This allows clients to read multiple byte ranges from the same file + * in a single call, potentially improving performance by enabling + * parallel reads and reducing round-trip overhead. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + */ + @Override + public void readVectored(List ranges, + IntFunction allocate) throws IOException { + TracingUtil.executeInNewSpan("OzoneFSInputStream.readVectored", () -> { + // Perform vectored read using positioned read operations + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + if (result == null) { + result = new CompletableFuture<>(); + range.setData(result); + } + + final CompletableFuture finalResult = result; + final long offset = range.getOffset(); + final int length = range.getLength(); + + // Submit async read task for this range + CompletableFuture.runAsync(() -> { + try { + ByteBuffer buffer = allocate.apply(length); + int bytesRead = read(offset, buffer); + + if (bytesRead < length) { + finalResult.completeExceptionally( + new EOFException("Requested " + length + + " bytes but only read " + bytesRead)); + } else { + buffer.flip(); + if (statistics != null) { + statistics.incrementBytesRead(bytesRead); + } + finalResult.complete(buffer); + } + } catch (Exception e) { + finalResult.completeExceptionally(e); + } + }); + } + return null; + }); + } } From 272edca99ba35f686d32d035c35cd6048ceabeaf Mon Sep 17 00:00:00 2001 From: ashishk Date: Mon, 1 Dec 2025 12:19:02 +0530 Subject: [PATCH 2/5] Fix checkstyle error --- .../apache/hadoop/hdds/scm/storage/MultipartInputStream.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index aecf2fbe02fe..19cf0f5d5931 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -262,12 +262,12 @@ public void readVectored(List ranges, final CompletableFuture finalResult = result; final long offset = range.getOffset(); - final int length = range.getLength(); + final int rangeLength = range.getLength(); // Submit async read task for this range CompletableFuture.runAsync(() -> { try { - ByteBuffer buffer = allocate.apply(length); + ByteBuffer buffer = allocate.apply(rangeLength); readRangeData(offset, buffer); buffer.flip(); finalResult.complete(buffer); @@ -305,7 +305,6 @@ private void readRangeData(long offset, ByteBuffer buffer) throws IOException { } } - @Override public synchronized void close() throws IOException { closed = true; From ba9f85e09b2641cc3690353764ce9d61f83cfad5 Mon Sep 17 00:00:00 2001 From: "Doroszlai, Attila" Date: Wed, 3 Dec 2025 19:49:11 +0100 Subject: [PATCH 3/5] add contract test (cherry picked from commit ee11cf36ae4ded41c0627906af79023fef21ae75) --- .../AbstractContractVectoredReadTest.java | 684 ++++++++++++++++++ .../hadoop/fs/contract/ContractTestUtils.java | 51 +- .../contract/AbstractOzoneContractTest.java | 22 + 3 files changed, 752 insertions(+), 5 deletions(-) create mode 100644 hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java new file mode 100644 index 000000000000..2bf3d59bcce9 --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java @@ -0,0 +1,684 @@ +/* + * 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.hadoop.fs.contract; + +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_LENGTH; +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY; +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_VECTOR; +import static org.apache.hadoop.fs.StreamCapabilities.VECTOREDIO_BUFFERS_SLICED; +import static org.apache.hadoop.fs.contract.ContractTestUtils.VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS; +import static org.apache.hadoop.fs.contract.ContractTestUtils.assertDatasetEquals; +import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile; +import static org.apache.hadoop.fs.contract.ContractTestUtils.range; +import static org.apache.hadoop.fs.contract.ContractTestUtils.returnBuffersToPoolPostRead; +import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult; +import static org.apache.hadoop.io.Sizes.S_128K; +import static org.apache.hadoop.io.Sizes.S_4K; +import static org.apache.hadoop.test.LambdaTestUtils.intercept; +import static org.apache.hadoop.test.LambdaTestUtils.interceptFuture; +import static org.apache.hadoop.util.functional.FutureIO.awaitFuture; +import static org.junit.jupiter.api.Assertions.fail; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntFunction; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.impl.TrackingByteBufferPool; +import org.apache.hadoop.io.ElasticByteBufferPool; +import org.apache.hadoop.io.WeakReferencedElasticByteBufferPool; +import org.apache.hadoop.util.concurrent.HadoopExecutors; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedClass; +import org.junit.jupiter.params.provider.MethodSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Test Vectored Reads. + *

+ * Both the original readVectored(allocator) and the readVectored(allocator, release) + * operations are tested. + */ +@ParameterizedClass(name = "buffer-{0}") +@MethodSource("params") +public abstract class AbstractContractVectoredReadTest extends AbstractFSContractTestBase { + + private static final Logger LOG = + LoggerFactory.getLogger(AbstractContractVectoredReadTest.class); + + private static final int DATASET_LEN = S_128K; + private static final byte[] DATASET = ContractTestUtils.dataset(DATASET_LEN, 'a', 32); + private static final String VECTORED_READ_FILE_NAME = "vectored_file.txt"; + + /** + * Buffer allocator for vector IO. + */ + private final IntFunction allocate; + + /** + * Buffer pool for vector IO. + */ + private final ElasticByteBufferPool pool = + new WeakReferencedElasticByteBufferPool(); + + private final String bufferType; + + private final boolean isDirect; + + /** + * Path to the vector file. + */ + private Path vectorPath; + + /** + * Counter of buffer releases. + * Because not all implementations release buffers on failures, + * this is not yet used in assertions. + */ + private final AtomicInteger bufferReleases = new AtomicInteger(); + + public static List params() { + return Arrays.asList("direct", "array"); + } + + protected AbstractContractVectoredReadTest(String bufferType) { + this.bufferType = bufferType; + this.isDirect = !"array".equals(bufferType); + this.allocate = size -> pool.getBuffer(isDirect, size); + } + + /** + * Get the buffer allocator. + * @return allocator function for vector IO. + */ + protected IntFunction getAllocate() { + return allocate; + } + + /** + * The buffer release operation. + */ + protected void release(ByteBuffer buffer) { + LOG.info("Released buffer {}", buffer); + bufferReleases.incrementAndGet(); + pool.putBuffer(buffer); + } + + /** + * Get the vector IO buffer pool. + * @return a pool. + */ + + protected ElasticByteBufferPool getPool() { + return pool; + } + + @BeforeEach + @Override + public void setup() throws Exception { + super.setup(); + vectorPath = path(VECTORED_READ_FILE_NAME); + FileSystem fs = getFileSystem(); + createFile(fs, vectorPath, true, DATASET); + } + + @AfterEach + @Override + public void teardown() throws Exception { + pool.release(); + super.teardown(); + } + + /** + * Open the vector file. + * @return the input stream. + * @throws IOException failure. + */ + protected FSDataInputStream openVectorFile() throws IOException { + return openVectorFile(getFileSystem()); + } + + /** + * Open the vector file. + * @param fs filesystem to use + * @return the input stream. + * @throws IOException failure. + */ + protected FSDataInputStream openVectorFile(final FileSystem fs) throws IOException { + return awaitFuture( + fs.openFile(vectorPath) + .opt(FS_OPTION_OPENFILE_LENGTH, DATASET_LEN) + .opt(FS_OPTION_OPENFILE_READ_POLICY, + FS_OPTION_OPENFILE_READ_POLICY_VECTOR) + .build()); + } + + @Test + public void testVectoredReadMultipleRanges() throws Exception { + List fileRanges = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + FileRange fileRange = FileRange.createFileRange(i * 100, 100); + fileRanges.add(fileRange); + } + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + CompletableFuture[] completableFutures = new CompletableFuture[fileRanges.size()]; + int i = 0; + for (FileRange res : fileRanges) { + completableFutures[i++] = res.getData(); + } + CompletableFuture combinedFuture = CompletableFuture.allOf(completableFutures); + combinedFuture.get(); + + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadAndReadFully() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 100); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + byte[] readFullRes = new byte[100]; + in.readFully(100, readFullRes); + ByteBuffer vecRes = awaitFuture(fileRanges.get(0).getData()); + Assertions.assertThat(vecRes) + .describedAs("Result from vectored read and readFully must match") + .isEqualByComparingTo(ByteBuffer.wrap(readFullRes)); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadWholeFile() throws Exception { + describe("Read the whole file in one single vectored read"); + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, DATASET_LEN); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + ByteBuffer vecRes = awaitFuture(fileRanges.get(0).getData()); + Assertions.assertThat(vecRes) + .describedAs("Result from vectored read and readFully must match") + .isEqualByComparingTo(ByteBuffer.wrap(DATASET)); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024,none of the below ranges + * will get merged. + */ + @Test + public void testDisjointRanges() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, 100); + range(fileRanges, 4_000 + 101, 100); + range(fileRanges, 16_000 + 101, 100); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024, all the below ranges + * will get merged into one. + */ + @Test + public void testAllRangesMergedIntoOne() throws Exception { + List fileRanges = new ArrayList<>(); + final int length = 100; + range(fileRanges, 0, length); + range(fileRanges, 4_000 - length - 1, length); + range(fileRanges, 8_000 - length - 1, length); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024, the first three ranges will be + * merged into and other two will remain as it is. + */ + @Test + public void testSomeRangesMergedSomeUnmerged() throws Exception { + FileSystem fs = getFileSystem(); + List fileRanges = new ArrayList<>(); + range(fileRanges, 8 * 1024, 100); + range(fileRanges, 14 * 1024, 100); + range(fileRanges, 10 * 1024, 100); + range(fileRanges, 2 * 1024 - 101, 100); + range(fileRanges, 40 * 1024, 1024); + FileStatus fileStatus = fs.getFileStatus(path(VECTORED_READ_FILE_NAME)); + CompletableFuture builder = + fs.openFile(path(VECTORED_READ_FILE_NAME)) + .withFileStatus(fileStatus) + .build(); + try (FSDataInputStream in = builder.get()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * Most file systems won't support overlapping ranges. + * Currently, only Raw Local supports it. + */ + @Test + public void testOverlappingRanges() throws Exception { + if (!isSupported(VECTOR_IO_OVERLAPPING_RANGES)) { + verifyExceptionalVectoredRead( + getSampleOverlappingRanges(), + IllegalArgumentException.class); + } else { + try (FSDataInputStream in = openVectorFile()) { + List fileRanges = getSampleOverlappingRanges(); + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + } + + /** + * Same ranges are special case of overlapping. + */ + @Test + public void testSameRanges() throws Exception { + if (!isSupported(VECTOR_IO_OVERLAPPING_RANGES)) { + verifyExceptionalVectoredRead( + getSampleSameRanges(), + IllegalArgumentException.class); + } else { + try (FSDataInputStream in = openVectorFile()) { + List fileRanges = getSampleSameRanges(); + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + } + + /** + * A null range is not permitted. + */ + @Test + public void testNullRange() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 500, 100); + fileRanges.add(null); + verifyExceptionalVectoredRead( + fileRanges, + NullPointerException.class); + } + + /** + * A null range is not permitted. + */ + @Test + public void testNullRangeList() throws Exception { + verifyExceptionalVectoredRead( + null, + NullPointerException.class); + } + + @Test + public void testSomeRandomNonOverlappingRanges() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 500, 100); + range(fileRanges, 1000, 200); + range(fileRanges, 50, 10); + range(fileRanges, 10, 5); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testConsecutiveRanges() throws Exception { + List fileRanges = new ArrayList<>(); + final int offset = 500; + final int length = 2011; + range(fileRanges, offset, length); + range(fileRanges, offset + length, length); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testEmptyRanges() throws Exception { + List fileRanges = new ArrayList<>(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + Assertions.assertThat(fileRanges) + .describedAs("Empty ranges must stay empty") + .isEmpty(); + } + } + + /** + * Test to validate EOF ranges. + *

+ * Default implementation fails with EOFException + * while reading the ranges. Some implementation like s3, checksum fs fail fast + * as they already have the file length calculated. + * The contract option {@link ContractOptions#VECTOR_IO_EARLY_EOF_CHECK} is used + * to determine which check to perform. + */ + @Test + public void testEOFRanges() throws Exception { + describe("Testing reading with an offset past the end of the file"); + List fileRanges = range(DATASET_LEN + 1, 100); + + if (isSupported(VECTOR_IO_EARLY_EOF_CHECK)) { + LOG.info("Expecting early EOF failure"); + verifyExceptionalVectoredRead(fileRanges, EOFException.class); + } else { + expectEOFinRead(fileRanges); + } + } + + @Test + public void testVectoredReadWholeFilePlusOne() throws Exception { + describe("Try to read whole file plus 1 byte"); + List fileRanges = range(0, DATASET_LEN + 1); + + if (isSupported(VECTOR_IO_EARLY_EOF_CHECK)) { + LOG.info("Expecting early EOF failure"); + verifyExceptionalVectoredRead(fileRanges, EOFException.class); + } else { + expectEOFinRead(fileRanges); + } + } + + private void expectEOFinRead(final List fileRanges) throws Exception { + LOG.info("Expecting late EOF failure"); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + for (FileRange res : fileRanges) { + CompletableFuture data = res.getData(); + interceptFuture(EOFException.class, + "", + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, + TimeUnit.SECONDS, + data); + } + } + } + + @Test + public void testNegativeLengthRange() throws Exception { + verifyExceptionalVectoredRead(range(0, -50), IllegalArgumentException.class); + } + + @Test + public void testNegativeOffsetRange() throws Exception { + verifyExceptionalVectoredRead(range(-1, 50), EOFException.class); + } + + @Test + public void testNullReleaseOperation() throws Exception { + + final List range = range(0, 10); + try (FSDataInputStream in = openVectorFile()) { + intercept(NullPointerException.class, () -> + in.readVectored(range, allocate, null)); + } + } + + @Test + public void testNormalReadAfterVectoredRead() throws Exception { + List fileRanges = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + // read starting 200 bytes + final int len = 200; + byte[] res = new byte[len]; + in.readFully(res, 0, len); + ByteBuffer buffer = ByteBuffer.wrap(res); + assertDatasetEquals(0, "normal_read", buffer, len, DATASET); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadAfterNormalRead() throws Exception { + List fileRanges = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + // read starting 200 bytes + final int len = 200; + byte[] res = new byte[len]; + in.readFully(res, 0, len); + ByteBuffer buffer = ByteBuffer.wrap(res); + assertDatasetEquals(0, "normal_read", buffer, len, DATASET); + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testMultipleVectoredReads() throws Exception { + List fileRanges1 = createSampleNonOverlappingRanges(); + List fileRanges2 = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges1, allocate); + in.readVectored(fileRanges2, allocate, this::release); + validateVectoredReadResult(fileRanges2, DATASET, 0); + validateVectoredReadResult(fileRanges1, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges1, pool); + returnBuffersToPoolPostRead(fileRanges2, pool); + } + } + + /** + * This test creates list of ranges and then submit a readVectored + * operation and then uses a separate thread pool to process the + * results asynchronously. + */ + @Test + public void testVectoredIOEndToEnd() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 8 * 1024, 100); + range(fileRanges, 14 * 1024, 100); + range(fileRanges, 10 * 1024, 100); + range(fileRanges, 2 * 1024 - 101, 100); + range(fileRanges, 40 * 1024, 1024); + + ExecutorService dataProcessor = Executors.newFixedThreadPool(5); + CountDownLatch countDown = new CountDownLatch(fileRanges.size()); + + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, this.allocate); + for (FileRange res : fileRanges) { + dataProcessor.submit(() -> { + try { + readBufferValidateDataAndReturnToPool(res, countDown); + } catch (Exception e) { + String error = String.format("Error while processing result for %s", res); + LOG.error(error, e); + fail(error, e); + } + }); + } + // user can perform other computations while waiting for IO. + if (!countDown.await(VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + fail("Timeout/Error while processing vectored io results"); + } + } finally { + HadoopExecutors.shutdown(dataProcessor, LOG, + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + } + + private void readBufferValidateDataAndReturnToPool(FileRange res, + CountDownLatch countDownLatch) + throws IOException, TimeoutException { + try { + CompletableFuture data = res.getData(); + // Read the data and perform custom operation. Here we are just + // validating it with original data. + awaitFuture(data.thenAccept(buffer -> { + assertDatasetEquals((int) res.getOffset(), + "vecRead", buffer, res.getLength(), DATASET); + // return buffer to the pool once read. + // If the read failed, this doesn't get invoked. + pool.putBuffer(buffer); + }), + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } finally { + // countdown to notify main thread that processing has been done. + countDownLatch.countDown(); + } + } + + protected List createSampleNonOverlappingRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, 100); + range(fileRanges, 110, 50); + return fileRanges; + } + + protected List getSampleSameRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 8_000, 1000); + range(fileRanges, 8_000, 1000); + range(fileRanges, 8_000, 1000); + return fileRanges; + } + + protected List getSampleOverlappingRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 500); + range(fileRanges, 400, 500); + return fileRanges; + } + + protected List getConsecutiveRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 500); + range(fileRanges, 600, 500); + return fileRanges; + } + + /** + * Validate that exceptions must be thrown during a vectored + * read operation with specific input ranges. + * @param fileRanges input file ranges. + * @param clazz type of exception expected. + * @throws Exception any other exception. + */ + protected void verifyExceptionalVectoredRead( + List fileRanges, + Class clazz) throws Exception { + + try (FSDataInputStream in = openVectorFile()) { + intercept(clazz, () -> { + in.readVectored(fileRanges, allocate); + return "triggered read of " + fileRanges.size() + " ranges" + " against " + in; + }); + } + } + + @Test + public void testBufferSlicing() throws Throwable { + describe("Test buffer slicing behavior in vectored IO"); + + final int numBuffers = 8; + final int bufferSize = S_4K; + long offset = 0; + final List fileRanges = new ArrayList<>(); + for (int i = 0; i < numBuffers; i++) { + fileRanges.add(FileRange.createFileRange(offset, bufferSize)); + // increment and add a non-binary-aligned gap, so as to force + // offsets to be misaligned with possible page sizes. + offset += bufferSize + 4000; + } + TrackingByteBufferPool trackerPool = TrackingByteBufferPool.wrap(getPool()); + int unknownBuffers = 0; + boolean slicing; + try (FSDataInputStream in = openVectorFile()) { + slicing = in.hasCapability(VECTOREDIO_BUFFERS_SLICED); + LOG.info("Slicing is {} for vectored IO with stream {}", slicing, in); + in.readVectored(fileRanges, s -> trackerPool.getBuffer(isDirect, s), trackerPool::putBuffer); + + // check that all buffers are from the the pool, unless they are sliced. + for (FileRange res : fileRanges) { + CompletableFuture data = res.getData(); + ByteBuffer buffer = awaitFuture(data); + Assertions.assertThat(buffer) + .describedAs("Buffer must not be null") + .isNotNull(); + Assertions.assertThat(slicing || trackerPool.containsBuffer(buffer)) + .describedAs("Buffer must be from the pool") + .isTrue(); + try { + trackerPool.putBuffer(buffer); + } catch (TrackingByteBufferPool.ReleasingUnallocatedByteBufferException e) { + // this can happen if the buffer was sliced, as it is not in the pool. + if (!slicing) { + throw e; + } + LOG.info("Sliced buffer detected: {}", buffer); + unknownBuffers++; + } + } + } + try { + trackerPool.close(); + } catch (TrackingByteBufferPool.LeakedByteBufferException e) { + if (!slicing) { + throw e; + } + LOG.info("Slicing is enabled; we saw leaked buffers: {} after {}" + + " releases of unknown buffers", + e.getCount(), unknownBuffers); + } + + } +} diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java index 1ddaa079bb95..d3e31369a172 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java @@ -1130,11 +1130,14 @@ public static void validateFileContent(byte[] concat, byte[][] bytes) { * Utility to validate vectored read results. * @param fileRanges input ranges. * @param originalData original data. + * @param baseOffset base offset of the original data * @throws IOException any ioe. */ - public static void validateVectoredReadResult(List fileRanges, - byte[] originalData) - throws IOException, TimeoutException { + public static void validateVectoredReadResult( + final List fileRanges, + final byte[] originalData, + final long baseOffset) + throws IOException, TimeoutException { CompletableFuture[] completableFutures = new CompletableFuture[fileRanges.size()]; int i = 0; for (FileRange res : fileRanges) { @@ -1150,8 +1153,8 @@ public static void validateVectoredReadResult(List fileRanges, ByteBuffer buffer = FutureIO.awaitFuture(data, VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); - assertDatasetEquals((int) res.getOffset(), "vecRead", - buffer, res.getLength(), originalData); + assertDatasetEquals((int) (res.getOffset() - baseOffset), "vecRead", + buffer, res.getLength(), originalData); } } @@ -1758,6 +1761,44 @@ public static long readStream(InputStream in) { } } + /** + * Create a range list with a single range within it. + * @param offset offset + * @param length length + * @return the list. + */ + public static List range( + final long offset, + final int length) { + return range(new ArrayList<>(), offset, length); + } + + /** + * Create a range and add it to the supplied list. + * @param fileRanges list of ranges + * @param offset offset + * @param length length + * @return the list. + */ + public static List range( + final List fileRanges, + final long offset, + final int length) { + fileRanges.add(FileRange.createFileRange(offset, length)); + return fileRanges; + } + + /** + * Given a list of ranges, calculate the total size. + * @param fileRanges range list. + * @return total size of all reads. + */ + public static long totalReadSize(final List fileRanges) { + return fileRanges.stream() + .mapToLong(FileRange::getLength) + .sum(); + } + /** * Results of recursive directory creation/scan operations. */ diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java index 7315abd1553a..95f50d2cbec4 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java @@ -34,6 +34,7 @@ import org.apache.hadoop.fs.contract.AbstractContractRootDirectoryTest; import org.apache.hadoop.fs.contract.AbstractContractSeekTest; import org.apache.hadoop.fs.contract.AbstractContractUnbufferTest; +import org.apache.hadoop.fs.contract.AbstractContractVectoredReadTest; import org.apache.hadoop.fs.contract.AbstractFSContract; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.ozone.MiniOzoneCluster; @@ -43,6 +44,8 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.params.ParameterizedClass; +import org.junit.jupiter.params.provider.MethodSource; /** * Base class for Ozone contract tests. Manages lifecycle of {@link MiniOzoneCluster}. @@ -271,6 +274,25 @@ protected AbstractFSContract createContract(Configuration conf) { } } + @Nested + @ParameterizedClass(name = "buffer-{0}") + @MethodSource("params") + class TestContractVectoredRead extends AbstractContractVectoredReadTest { + TestContractVectoredRead(String bufferType) { + super(bufferType); + } + + @Override + protected Configuration createConfiguration() { + return createOzoneConfig(); + } + + @Override + protected AbstractFSContract createContract(Configuration conf) { + return createOzoneContract(conf); + } + } + @Nested class TestContractLeaseRecovery extends AbstractContractLeaseRecoveryTest { From a2eff5524f21cc1ba4a78a11cab342f4f7718240 Mon Sep 17 00:00:00 2001 From: ashishk Date: Thu, 4 Dec 2025 20:57:43 +0530 Subject: [PATCH 4/5] Fix review comments --- .../scm/storage/MultipartInputStream.java | 67 ++++----- .../hadoop/hdds/utils/VectoredReadUtils.java | 138 ++++++++++++++++++ .../ozone/client/io/OzoneInputStream.java | 6 +- .../fs/ozone/TestOzoneFSInputStream.java | 12 +- .../hadoop/fs/ozone/OzoneFSInputStream.java | 51 +++---- 5 files changed, 197 insertions(+), 77 deletions(-) create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index 19cf0f5d5931..9bd48f6b48e8 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -25,10 +25,10 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; -import java.util.concurrent.CompletableFuture; import java.util.function.IntFunction; import org.apache.hadoop.fs.FSExceptionMessages; import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.hdds.utils.VectoredReadUtils; /** * A stream for accessing multipart streams. @@ -245,36 +245,26 @@ public synchronized long skip(long n) throws IOException { * @param allocate function to allocate ByteBuffer for each range * @throws IOException if there is an error performing the reads */ - public void readVectored(List ranges, - IntFunction allocate) throws IOException { + public void readVectored( + List ranges, + IntFunction allocate + ) throws IOException { checkOpen(); if (!initialized) { initialize(); } - // Perform vectored read using positioned read operations - for (FileRange range : ranges) { - CompletableFuture result = range.getData(); - if (result == null) { - result = new CompletableFuture<>(); - range.setData(result); - } + // Save the initial position + final long initialPosition = getPos(); - final CompletableFuture finalResult = result; - final long offset = range.getOffset(); - final int rangeLength = range.getLength(); - - // Submit async read task for this range - CompletableFuture.runAsync(() -> { - try { - ByteBuffer buffer = allocate.apply(rangeLength); - readRangeData(offset, buffer); - buffer.flip(); - finalResult.complete(buffer); - } catch (Exception e) { - finalResult.completeExceptionally(e); - } - }); + try { + VectoredReadUtils.performVectoredRead(ranges, allocate, + (offset, buffer) -> readRangeData(offset, buffer, initialPosition)); + } finally { + // Restore position + synchronized (this) { + seek(initialPosition); + } } } @@ -286,21 +276,28 @@ public void readVectored(List ranges, * @param buffer the buffer to read data into * @throws IOException if there is an error reading data */ - private void readRangeData(long offset, ByteBuffer buffer) throws IOException { + private void readRangeData(long offset, ByteBuffer buffer, long initialPosition) throws IOException { synchronized (this) { - long savedPos = getPos(); try { seek(offset); - int remaining = buffer.remaining(); - byte[] temp = new byte[remaining]; - int bytesRead = read(temp, 0, remaining); - if (bytesRead < remaining) { - throw new EOFException("Could not read all requested bytes. " + - "Requested: " + remaining + ", Read: " + bytesRead); + int totalBytesToRead = buffer.remaining(); + byte[] temp = new byte[totalBytesToRead]; + int totalBytesRead = 0; + + // Read in a loop to handle partial reads + while (totalBytesRead < totalBytesToRead) { + int bytesRead = read(temp, totalBytesRead, totalBytesToRead - totalBytesRead); + if (bytesRead < 0) { + throw new EOFException("End of file reached before reading fully. " + + "Requested: " + totalBytesToRead + ", Read: " + totalBytesRead); + } + totalBytesRead += bytesRead; } - buffer.put(temp, 0, bytesRead); + + buffer.put(temp, 0, totalBytesRead); } finally { - seek(savedPos); + // Restore position + seek(initialPosition); } } } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java new file mode 100644 index 000000000000..eb4f21c238cf --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.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.hadoop.hdds.utils; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; +import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.hdds.annotation.InterfaceAudience; + +/** + * Utility class for vectored read operations. + */ +@InterfaceAudience.Private +public final class VectoredReadUtils { + private VectoredReadUtils() { + // Utility class, no instances + } + + /** + * Functional interface for reading data into a buffer from a specific offset. + */ + @FunctionalInterface + public interface RangeReader { + /** + * Read data from the specified offset into the buffer. + * + * @param offset the offset to read from + * @param buffer the buffer to read into + * @throws IOException if an error occurs during reading + */ + void readRange(long offset, ByteBuffer buffer) throws IOException; + } + + /** + * Validates the ranges for vectored read operations. + * Checks for null ranges, negative offsets, negative lengths, and overlapping ranges. + * + * @param ranges list of file ranges to validate + * @throws NullPointerException if ranges list is null or contains null elements + * @throws EOFException if any range has a negative offset + * @throws IllegalArgumentException if any range has negative length or ranges overlap + */ + public static void validateRanges(List ranges) throws IOException { + if (ranges == null) { + throw new NullPointerException("Ranges list cannot be null"); + } + for (int i = 0; i < ranges.size(); i++) { + FileRange range = ranges.get(i); + if (range == null) { + throw new NullPointerException("Range at index " + i + " is null"); + } + // Check for negative offset + if (range.getOffset() < 0) { + throw new EOFException("Range " + i + " has negative offset: " + range.getOffset()); + } + // Check for negative length + if (range.getLength() < 0) { + throw new IllegalArgumentException("Range " + i + " has negative length: " + range.getLength()); + } + } + // Check for overlapping ranges + for (int i = 0; i < ranges.size(); i++) { + FileRange current = ranges.get(i); + long currentEnd = current.getOffset() + current.getLength(); + for (int j = i + 1; j < ranges.size(); j++) { + FileRange other = ranges.get(j); + long otherEnd = other.getOffset() + other.getLength(); + // Check if ranges overlap + boolean overlaps = (current.getOffset() < otherEnd && currentEnd > other.getOffset()); + if (overlaps) { + throw new IllegalArgumentException( + "Range[" + i + "] (" + current.getOffset() + ", " + current.getLength() + + ") overlaps with Range[" + j + "] (" + other.getOffset() + ", " + other.getLength() + ")"); + } + } + } + } + + /** + * Performs vectored read by reading each range asynchronously. + * This method handles the common logic of setting up CompletableFutures + * and submitting async read tasks for each range. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @param reader the function that performs the actual read operation + * @throws IOException if there is an error during validation + */ + public static void performVectoredRead( + List ranges, + IntFunction allocate, + RangeReader reader) throws IOException { + + // Validate ranges before processing + validateRanges(ranges); + // Perform vectored read using positioned read operations + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + if (result == null) { + result = new CompletableFuture<>(); + range.setData(result); + } + final CompletableFuture finalResult = result; + final long offset = range.getOffset(); + final int length = range.getLength(); + // Submit async read task for this range + CompletableFuture.runAsync(() -> { + try { + ByteBuffer buffer = allocate.apply(length); + reader.readRange(offset, buffer); + buffer.flip(); + finalResult.complete(buffer); + } catch (Exception e) { + finalResult.completeExceptionally(e); + } + }); + } + } +} diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java index 9439a5a12008..1443079342b7 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java @@ -135,8 +135,10 @@ public boolean seekToNewSource(long targetPos) throws IOException { * @param allocate function to allocate ByteBuffer for each range * @throws IOException if there is an error performing the reads */ - public void readVectored(List ranges, - IntFunction allocate) throws IOException { + public void readVectored( + List ranges, + IntFunction allocate + ) throws IOException { if (inputStream instanceof MultipartInputStream) { ((MultipartInputStream) inputStream).readVectored(ranges, allocate); } else { diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java index 1143bc5e6d11..83cda4b22198 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java @@ -375,14 +375,10 @@ public void testVectoredReadWithOverlappingRanges() throws Exception { ranges.add(FileRange.createFileRange(300L, 400)); // Overlaps with first range ranges.add(FileRange.createFileRange(600L, 200)); - inputStream.readVectored(ranges, ByteBuffer::allocate); - - // Validate all reads complete successfully - for (FileRange range : ranges) { - CompletableFuture result = range.getData(); - ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); - assertEquals(range.getLength(), buffer.remaining()); - } + // Overlapping ranges should throw IllegalArgumentException + assertThrows(IllegalArgumentException.class, () -> { + inputStream.readVectored(ranges, ByteBuffer::allocate); + }); } } diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java index 0004e2f644d3..da5a7c161e14 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java @@ -23,7 +23,6 @@ import java.nio.ByteBuffer; import java.nio.ReadOnlyBufferException; import java.util.List; -import java.util.concurrent.CompletableFuture; import java.util.function.IntFunction; import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.fs.ByteBufferReadable; @@ -35,6 +34,7 @@ import org.apache.hadoop.hdds.annotation.InterfaceAudience; import org.apache.hadoop.hdds.annotation.InterfaceStability; import org.apache.hadoop.hdds.tracing.TracingUtil; +import org.apache.hadoop.hdds.utils.VectoredReadUtils; /** * The input stream for Ozone file system. @@ -215,42 +215,29 @@ public synchronized void readFully(long position, ByteBuffer buf) throws IOExcep * @throws IOException if there is an error performing the reads */ @Override - public void readVectored(List ranges, - IntFunction allocate) throws IOException { + public void readVectored( + List ranges, + IntFunction allocate + ) throws IOException { TracingUtil.executeInNewSpan("OzoneFSInputStream.readVectored", () -> { - // Perform vectored read using positioned read operations - for (FileRange range : ranges) { - CompletableFuture result = range.getData(); - if (result == null) { - result = new CompletableFuture<>(); - range.setData(result); - } - - final CompletableFuture finalResult = result; - final long offset = range.getOffset(); - final int length = range.getLength(); + // Save the initial position + final long initialPosition = getPos(); - // Submit async read task for this range - CompletableFuture.runAsync(() -> { - try { - ByteBuffer buffer = allocate.apply(length); - int bytesRead = read(offset, buffer); + try { + VectoredReadUtils.performVectoredRead(ranges, allocate, (offset, buffer) -> { + int length = buffer.remaining(); + readFully(offset, buffer); - if (bytesRead < length) { - finalResult.completeExceptionally( - new EOFException("Requested " + length + - " bytes but only read " + bytesRead)); - } else { - buffer.flip(); - if (statistics != null) { - statistics.incrementBytesRead(bytesRead); - } - finalResult.complete(buffer); - } - } catch (Exception e) { - finalResult.completeExceptionally(e); + // Update statistics + if (statistics != null) { + statistics.incrementBytesRead(length); } }); + } finally { + // Restore position + synchronized (this) { + seek(initialPosition); + } } return null; }); From bc0bafd7c2455891ef360b3479f0ee1ef899e40d Mon Sep 17 00:00:00 2001 From: ashishk Date: Tue, 9 Dec 2025 14:00:44 +0530 Subject: [PATCH 5/5] Fix review comments --- .../scm/storage/MultipartInputStream.java | 53 ++++++--- .../hadoop/hdds/utils/VectoredReadUtils.java | 102 +++++++++++------- .../hadoop/fs/ozone/OzoneFSInputStream.java | 39 +++---- 3 files changed, 124 insertions(+), 70 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index 9bd48f6b48e8..7a95aebac3e3 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -244,8 +244,10 @@ public synchronized long skip(long n) throws IOException { * @param ranges list of file ranges to read * @param allocate function to allocate ByteBuffer for each range * @throws IOException if there is an error performing the reads + * @apiNote This method is synchronized to prevent race conditions from + * concurrent readVectored() calls on the same stream instance. */ - public void readVectored( + public synchronized void readVectored( List ranges, IntFunction allocate ) throws IOException { @@ -257,36 +259,51 @@ public void readVectored( // Save the initial position final long initialPosition = getPos(); - try { - VectoredReadUtils.performVectoredRead(ranges, allocate, - (offset, buffer) -> readRangeData(offset, buffer, initialPosition)); - } finally { - // Restore position - synchronized (this) { - seek(initialPosition); - } - } + // Use common vectored read implementation + VectoredReadUtils.performVectoredRead( + ranges, + allocate, + (offset, buffer) -> readRangeData(offset, buffer, initialPosition) + ); + + // Restore position + seek(initialPosition); } /** * Helper method to read data for a specific range. * Uses synchronized seeks to read data from the correct position. + * Reads data fully, handling partial reads in a loop. * * @param offset the starting offset in the stream * @param buffer the buffer to read data into + * @param initialPosition the initial position to restore after reading * @throws IOException if there is an error reading data */ - private void readRangeData(long offset, ByteBuffer buffer, long initialPosition) throws IOException { + private void readRangeData(long offset, ByteBuffer buffer, long initialPosition) + throws IOException { synchronized (this) { try { seek(offset); int totalBytesToRead = buffer.remaining(); - byte[] temp = new byte[totalBytesToRead]; - int totalBytesRead = 0; + // Read directly into buffer's backing array if available + byte[] readBuffer; + int arrayOffset; + if (buffer.hasArray()) { + readBuffer = buffer.array(); + arrayOffset = buffer.arrayOffset() + buffer.position(); + } else { + // Use temp array for direct ByteBuffers + readBuffer = new byte[totalBytesToRead]; + arrayOffset = 0; + } + + int totalBytesRead = 0; // Read in a loop to handle partial reads while (totalBytesRead < totalBytesToRead) { - int bytesRead = read(temp, totalBytesRead, totalBytesToRead - totalBytesRead); + int bytesRead = read(readBuffer, arrayOffset + totalBytesRead, + totalBytesToRead - totalBytesRead); if (bytesRead < 0) { throw new EOFException("End of file reached before reading fully. " + "Requested: " + totalBytesToRead + ", Read: " + totalBytesRead); @@ -294,7 +311,13 @@ private void readRangeData(long offset, ByteBuffer buffer, long initialPosition) totalBytesRead += bytesRead; } - buffer.put(temp, 0, totalBytesRead); + // If we used a temp array, copy to buffer + if (!buffer.hasArray()) { + buffer.put(readBuffer, 0, totalBytesRead); + } else { + // Update buffer position + buffer.position(buffer.position() + totalBytesRead); + } } finally { // Restore position seek(initialPosition); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java index eb4f21c238cf..4b0d160b9fdf 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java @@ -17,9 +17,13 @@ package org.apache.hadoop.hdds.utils; +import static java.util.Objects.requireNonNull; + import java.io.EOFException; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.function.IntFunction; @@ -28,6 +32,7 @@ /** * Utility class for vectored read operations. + * Based on Hadoop's org.apache.hadoop.fs.VectoredReadUtils. */ @InterfaceAudience.Private public final class VectoredReadUtils { @@ -36,7 +41,7 @@ private VectoredReadUtils() { } /** - * Functional interface for reading data into a buffer from a specific offset. + * Functional interface for reading a range of data. */ @FunctionalInterface public interface RangeReader { @@ -51,65 +56,90 @@ public interface RangeReader { } /** - * Validates the ranges for vectored read operations. - * Checks for null ranges, negative offsets, negative lengths, and overlapping ranges. + * Validate a single range. + * @param range range to validate. + * @return the range. + * @param range type + * @throws IllegalArgumentException the range length is negative + * @throws EOFException the range offset is negative + * @throws NullPointerException if the range is null. + */ + public static T validateRangeRequest(T range) + throws EOFException { + requireNonNull(range, "range is null"); + + if (range.getLength() < 0) { + throw new IllegalArgumentException("length is negative in " + range); + } + if (range.getOffset() < 0) { + throw new EOFException("position is negative in range " + range); + } + return range; + } + + /** + * Sort the input ranges by offset; no validation is done. + * @param input input ranges. + * @return a new list of the ranges, sorted by offset. + */ + public static List sortRangeList(List input) { + final List l = new ArrayList<>(input); + l.sort(Comparator.comparingLong(FileRange::getOffset)); + return l; + } + + /** + * Validate a list of ranges (including overlapping checks). + * Based on Hadoop's validateAndSortRanges. + * Two ranges overlap when the start offset of second is less than + * the end offset of first. End offset is calculated as start offset + length. * * @param ranges list of file ranges to validate * @throws NullPointerException if ranges list is null or contains null elements * @throws EOFException if any range has a negative offset - * @throws IllegalArgumentException if any range has negative length or ranges overlap + * @throws IllegalArgumentException if there are overlapping ranges or a range element is invalid */ public static void validateRanges(List ranges) throws IOException { - if (ranges == null) { - throw new NullPointerException("Ranges list cannot be null"); + requireNonNull(ranges, "Null input list"); + + if (ranges.isEmpty()) { + return; } - for (int i = 0; i < ranges.size(); i++) { - FileRange range = ranges.get(i); - if (range == null) { - throw new NullPointerException("Range at index " + i + " is null"); - } - // Check for negative offset - if (range.getOffset() < 0) { - throw new EOFException("Range " + i + " has negative offset: " + range.getOffset()); - } - // Check for negative length - if (range.getLength() < 0) { - throw new IllegalArgumentException("Range " + i + " has negative length: " + range.getLength()); - } + + if (ranges.size() == 1) { + validateRangeRequest(ranges.get(0)); + return; } - // Check for overlapping ranges - for (int i = 0; i < ranges.size(); i++) { - FileRange current = ranges.get(i); - long currentEnd = current.getOffset() + current.getLength(); - for (int j = i + 1; j < ranges.size(); j++) { - FileRange other = ranges.get(j); - long otherEnd = other.getOffset() + other.getLength(); - // Check if ranges overlap - boolean overlaps = (current.getOffset() < otherEnd && currentEnd > other.getOffset()); - if (overlaps) { + + // Sort ranges to check for overlaps efficiently + List sortedRanges = sortRangeList(ranges); + FileRange prev = null; + for (final FileRange current : sortedRanges) { + validateRangeRequest(current); + if (prev != null) { + // Check for overlap: current start < prev end + if (current.getOffset() < prev.getOffset() + prev.getLength()) { throw new IllegalArgumentException( - "Range[" + i + "] (" + current.getOffset() + ", " + current.getLength() + - ") overlaps with Range[" + j + "] (" + other.getOffset() + ", " + other.getLength() + ")"); + "Overlapping ranges " + prev + " and " + current); } } + prev = current; } } /** - * Performs vectored read by reading each range asynchronously. - * This method handles the common logic of setting up CompletableFutures - * and submitting async read tasks for each range. + * Common implementation for vectored read operations. + * Validates ranges and submits async read tasks for each range. * * @param ranges list of file ranges to read * @param allocate function to allocate ByteBuffer for each range - * @param reader the function that performs the actual read operation + * @param reader function that performs the actual read operation * @throws IOException if there is an error during validation */ public static void performVectoredRead( List ranges, IntFunction allocate, RangeReader reader) throws IOException { - // Validate ranges before processing validateRanges(ranges); // Perform vectored read using positioned read operations diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java index da5a7c161e14..b5c79525dcc4 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java @@ -213,32 +213,33 @@ public synchronized void readFully(long position, ByteBuffer buf) throws IOExcep * @param ranges list of file ranges to read * @param allocate function to allocate ByteBuffer for each range * @throws IOException if there is an error performing the reads + * @apiNote This method is synchronized to prevent race conditions from + * concurrent readVectored() calls on the same stream instance. */ @Override - public void readVectored( - List ranges, - IntFunction allocate - ) throws IOException { + public synchronized void readVectored(List ranges, + IntFunction allocate) throws IOException { TracingUtil.executeInNewSpan("OzoneFSInputStream.readVectored", () -> { // Save the initial position final long initialPosition = getPos(); - try { - VectoredReadUtils.performVectoredRead(ranges, allocate, (offset, buffer) -> { - int length = buffer.remaining(); - readFully(offset, buffer); - - // Update statistics - if (statistics != null) { - statistics.incrementBytesRead(length); + // Use common vectored read implementation + VectoredReadUtils.performVectoredRead( + ranges, + allocate, + (offset, buffer) -> { + // readFully is synchronized and uses positioned reads + // which automatically preserve stream position + readFully(offset, buffer); + if (statistics != null) { + statistics.incrementBytesRead(buffer.remaining()); + } } - }); - } finally { - // Restore position - synchronized (this) { - seek(initialPosition); - } - } + ); + + // Restore position before returning from method + seek(initialPosition); + return null; }); }