You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by st...@apache.org on 2026/04/20 20:47:49 UTC
(hadoop) branch pr/HADOOP-19863-vector-read-metrics updated: HADOOP-19863. Pull count test to LocalFS
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch pr/HADOOP-19863-vector-read-metrics
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/pr/HADOOP-19863-vector-read-metrics by this push:
new df47018ef33 HADOOP-19863. Pull count test to LocalFS
df47018ef33 is described below
commit df47018ef338049847bf2da0beb7074624624577
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Mon Apr 20 21:39:30 2026 +0100
HADOOP-19863. Pull count test to LocalFS
+ validate properly
---
.../org/apache/hadoop/fs/RawLocalFileSystem.java | 19 +++--
.../contract/AbstractContractVectoredReadTest.java | 6 +-
.../localfs/TestLocalFSContractVectoredRead.java | 85 +++++++++++++++++-----
.../rawlocal/TestRawLocalContractVectoredRead.java | 48 +-----------
4 files changed, 86 insertions(+), 72 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java
index 956a3713aa4..9d6c0788b21 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java
@@ -251,9 +251,15 @@ public int read(byte[] b, int off, int len) throws IOException {
}
}
- private void recordBytesRead(final int value) {
- statistics.incrementBytesRead(value);
- bytesRead.addAndGet(value);
+ /**
+ * Count the number of bytes read in fs and io statistics.
+ * @param count
+ */
+ private void recordBytesRead(final int count) {
+ if (count > 0) {
+ statistics.incrementBytesRead(count);
+ bytesRead.addAndGet(count);
+ }
}
@Override
@@ -375,6 +381,7 @@ private static class AsyncHandler implements CompletionHandler<Integer, Integer>
/** Buffers being read. */
private final ByteBuffer[] buffers;
+ /* Callback to update statistics. */
private final Consumer<Integer> statisticsUpdater;
/**
@@ -382,7 +389,7 @@ private static class AsyncHandler implements CompletionHandler<Integer, Integer>
* @param channel open channel.
* @param ranges ranges to read.
* @param allocateRelease pool for allocating buffers, and releasing on failure
- * @param statisticsUpdater
+ * @param statisticsUpdater callback to update statistics.
*/
AsyncHandler(
final AsynchronousFileChannel channel,
@@ -433,10 +440,10 @@ public void completed(Integer result, Integer rangeIndex) {
// issue a read for the rest of the buffer
channel.read(buffer, range.getOffset() + buffer.position(), rangeIndex, this);
} else {
+ // read finished
+ statisticsUpdater.accept(range.getLength());
// Flip the buffer and declare success.
buffer.flip();
- statisticsUpdater.accept(range.getLength());
-
range.getData().complete(buffer);
}
}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
index de31a98e12e..6d39cdbad4f 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java
@@ -206,16 +206,18 @@ public void testVectoredReadMultipleRanges() throws Exception {
combinedFuture.get();
validateVectoredReadResult(fileRanges, DATASET, 0);
+ assertionsWithinTestVectoredReadMultipleRanges(in, fileRanges);
returnBuffersToPoolPostRead(fileRanges, pool);
- assertionsWithinTestVectoredReadMultipleRanges(in);
}
}
/**
* Place to add some custom assertions within {@link #testVectoredReadMultipleRanges()}.
* @param in active input stream.
+ * @param fileRanges
*/
- protected void assertionsWithinTestVectoredReadMultipleRanges(final FSDataInputStream in) {
+ protected void assertionsWithinTestVectoredReadMultipleRanges(final FSDataInputStream in,
+ final List<FileRange> fileRanges) {
}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/localfs/TestLocalFSContractVectoredRead.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/localfs/TestLocalFSContractVectoredRead.java
index e2a9bfb444a..522f5e075cb 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/localfs/TestLocalFSContractVectoredRead.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/localfs/TestLocalFSContractVectoredRead.java
@@ -21,31 +21,38 @@
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.MethodSource;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ChecksumException;
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.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.contract.AbstractContractVectoredReadTest;
import org.apache.hadoop.fs.contract.AbstractFSContract;
import org.apache.hadoop.fs.contract.ContractTestUtils;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedClass;
-import org.junit.jupiter.params.provider.MethodSource;
+import org.apache.hadoop.fs.statistics.IOStatistics;
import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult;
+import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.assertj.core.api.Assertions.assertThat;
-@ParameterizedClass(name="buffer-{0}")
+@ParameterizedClass(name = "buffer-{0}")
@MethodSource("params")
public class TestLocalFSContractVectoredRead extends AbstractContractVectoredReadTest {
+ private long initialBytesRead;
+
public TestLocalFSContractVectoredRead(final String bufferType) {
super(bufferType);
}
@@ -87,28 +94,28 @@ public void testChecksumValidationDuringVectoredReadSmallFile() throws Exception
* @throws Exception any exception other than ChecksumException
*/
private void validateCheckReadException(Path testPath,
- int length,
- List<FileRange> ranges) throws Exception {
+ int length,
+ List<FileRange> ranges) throws Exception {
LocalFileSystem localFs = (LocalFileSystem) getFileSystem();
final byte[] datasetCorrect = ContractTestUtils.dataset(length, 'a', 32);
- try (FSDataOutputStream out = localFs.create(testPath, true)){
+ try (FSDataOutputStream out = localFs.create(testPath, true)) {
out.write(datasetCorrect);
}
Path checksumPath = localFs.getChecksumFile(testPath);
Assertions.assertThat(localFs.exists(checksumPath))
- .describedAs("Checksum file should be present")
- .isTrue();
+ .describedAs("Checksum file should be present")
+ .isTrue();
CompletableFuture<FSDataInputStream> fis = localFs.openFile(testPath).build();
- try (FSDataInputStream in = fis.get()){
+ try (FSDataInputStream in = fis.get()) {
in.readVectored(ranges, getAllocate());
validateVectoredReadResult(ranges, datasetCorrect, 0);
}
final byte[] datasetCorrupted = ContractTestUtils.dataset(length, 'a', 64);
- try (FSDataOutputStream out = localFs.getRaw().create(testPath, true)){
+ try (FSDataOutputStream out = localFs.getRaw().create(testPath, true)) {
out.write(datasetCorrupted);
}
CompletableFuture<FSDataInputStream> fisN = localFs.openFile(testPath).build();
- try (FSDataInputStream in = fisN.get()){
+ try (FSDataInputStream in = fisN.get()) {
in.readVectored(ranges, getAllocate());
// Expect checksum exception when data is updated directly through
// raw local fs instance.
@@ -123,20 +130,64 @@ public void tesChecksumVectoredReadBoundaries() throws Exception {
final int length = 1071;
LocalFileSystem localFs = (LocalFileSystem) getFileSystem();
final byte[] datasetCorrect = ContractTestUtils.dataset(length, 'a', 32);
- try (FSDataOutputStream out = localFs.create(testPath, true)){
+ try (FSDataOutputStream out = localFs.create(testPath, true)) {
out.write(datasetCorrect);
}
Path checksumPath = localFs.getChecksumFile(testPath);
Assertions.assertThat(localFs.exists(checksumPath))
- .describedAs("Checksum file should be present at {} ", checksumPath)
- .isTrue();
+ .describedAs("Checksum file should be present at {} ", checksumPath)
+ .isTrue();
CompletableFuture<FSDataInputStream> fis = localFs.openFile(testPath).build();
List<FileRange> smallRange = new ArrayList<>();
smallRange.add(FileRange.createFileRange(1000, 71));
- try (FSDataInputStream in = fis.get()){
+ try (FSDataInputStream in = fis.get()) {
in.readVectored(smallRange, getAllocate());
validateVectoredReadResult(smallRange, datasetCorrect, 0);
}
}
+ /**
+ * subclass so that the bytes read count can be cached before the test run.
+ */
+ @Test
+ @Override
+ public void testVectoredReadMultipleRanges() throws Exception {
+ initialBytesRead = getBytesRead();
+ super.testVectoredReadMultipleRanges();
+ }
+
+ /**
+ * Add some custom checks of bytes read counts.
+ * @param in active input stream.
+ * @param fileRanges
+ */
+ @Override
+ protected void assertionsWithinTestVectoredReadMultipleRanges(
+ final FSDataInputStream in,
+ final List<FileRange> fileRanges) {
+
+ // the iostats is the sum of all the ranges.
+ final long totalVectorReadLength = fileRanges.stream().mapToLong(FileRange::getLength).sum();
+ final IOStatistics stats = in.getIOStatistics();
+ assertThatStatisticCounter(stats, STREAM_READ_BYTES)
+ .describedAs(STREAM_READ_BYTES + " in bytes read in stream %s", stats)
+ .isEqualTo(totalVectorReadLength);
+ // validate filesystem stats, went up by at least that amount.
+ // expect counting of other things, crc files in particular
+ long currentBytesRead = getBytesRead();
+ assertThat(currentBytesRead)
+ .describedAs("bytes read in stream %s", in)
+ .isGreaterThanOrEqualTo(initialBytesRead + totalVectorReadLength);
+ }
+
+ /**
+ * API is deprecated, but Spark uses it, and it's how the regression was found.
+ * this is how the production code looks at our stats.
+ * @return counter of bytes read across all stores. Never reset.
+ */
+ private static long getBytesRead() {
+ AtomicLong bytes = new AtomicLong();
+ FileSystem.getAllStatistics().forEach(st -> bytes.addAndGet(st.getBytesRead()));
+ return bytes.get();
+ }
}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/rawlocal/TestRawLocalContractVectoredRead.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/rawlocal/TestRawLocalContractVectoredRead.java
index 3cc0167ccea..6fe1f4e3a97 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/rawlocal/TestRawLocalContractVectoredRead.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/rawlocal/TestRawLocalContractVectoredRead.java
@@ -18,28 +18,17 @@
package org.apache.hadoop.fs.contract.rawlocal;
-import java.util.concurrent.atomic.AtomicLong;
-
import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.provider.MethodSource;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.contract.AbstractContractVectoredReadTest;
import org.apache.hadoop.fs.contract.AbstractFSContract;
-import org.apache.hadoop.fs.statistics.IOStatistics;
-
-import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
-import static org.assertj.core.api.Assertions.assertThat;
-@ParameterizedClass(name = "buffer-{0}")
+@ParameterizedClass(name="buffer-{0}")
@MethodSource("params")
public class TestRawLocalContractVectoredRead extends AbstractContractVectoredReadTest {
- private long initialBytesRead;
-
public TestRawLocalContractVectoredRead(final String bufferType) {
super(bufferType);
}
@@ -48,39 +37,4 @@ public TestRawLocalContractVectoredRead(final String bufferType) {
protected AbstractFSContract createContract(Configuration conf) {
return new RawlocalFSContract(conf);
}
-
- @Override
- public void setup() throws Exception {
- super.setup();
- initialBytesRead = getBytesRead();
- }
-
- /**
- * API is deprecated, but Spark uses it, and it's how the regression was found.
- * this is how the production code looks at our stats.
- * @return counter of bytes read across all stores. Never reset.
- */
- private static long getBytesRead() {
- AtomicLong bytes = new AtomicLong();
- FileSystem.getAllStatistics().forEach(st -> bytes.addAndGet(st.getBytesRead()));
- return bytes.get();
- }
-
- /**
- * Add some custom checks of bytes read counts.
- * @param in active input stream.
- */
- @Override
- protected void assertionsWithinTestVectoredReadMultipleRanges(final FSDataInputStream in) {
- long currentBytesRead = getBytesRead();
- assertThat(currentBytesRead)
- .describedAs("bytes read in stream %s", in)
- .isGreaterThan(initialBytesRead);
- final long diff = currentBytesRead - initialBytesRead;
- final IOStatistics stats = in.getIOStatistics();
- assertThatStatisticCounter(stats, STREAM_READ_BYTES)
- .describedAs(STREAM_READ_BYTES + " in bytes read in stream %s", stats)
- .isEqualTo(diff);
-
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org