review comments

Author: ahmarsuhail

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsRequestCallback.java

@@ -47,7 +47,7 @@ public void onHeadRequest() {
 
   @Override
   public void onBlockPrefetch(long start, long end) {
-    statistics.bytesPrefetched(end - start);
+    statistics.bytesPrefetched(end - start + 1);
   }
 
   @Override

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractAnalyticsStreamVectoredRead.java

@@ -38,6 +38,11 @@
 import static org.apache.hadoop.fs.contract.ContractTestUtils.skip;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult;
 import static org.apache.hadoop.fs.s3a.Constants.ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX;
+
+import static org.apache.hadoop.fs.s3a.S3ATestConstants.AAL_CACHE_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.S3ATestConstants.AAL_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.s3a.S3ATestConstants.AAL_REQUEST_COALESCE_TOLERANCE;
+import static org.apache.hadoop.fs.s3a.S3ATestConstants.AAL_SMALL_OBJECT_PREFETCH_ENABLED;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.enableAnalyticsAccelerator;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipForAnyEncryptionExceptSSES3;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
@@ -73,20 +78,20 @@ protected Configuration createConfiguration() {
     Configuration conf = super.createConfiguration();
     // Set the coalesce tolerance to 1KB, default is 1MB.
     conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
-            "."  + "physicalio.request.coalesce.tolerance", S_16K);
+            "."  + AAL_REQUEST_COALESCE_TOLERANCE, S_16K);
 
     // Set the minimum block size to 32KB. AAL uses a default block size of 128KB, which means the minimum size a S3
     // request will be is 128KB. Since the file being read is 128KB, we need to  use this here to demonstrate that
     // separate GET requests are made for ranges that are not coalesced.
     conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
-            "."  + "physicalio.readbuffersize", S_32K);
+            "."  + AAL_READ_BUFFER_SIZE, S_32K);
 
     // Disable small object prefetched, otherwise anything less than 8MB is fetched in a single GET.
     conf.set(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
-            "."  + "physicalio.small.objects.prefetching.enabled", "false");
+            "."  + AAL_SMALL_OBJECT_PREFETCH_ENABLED, "false");
 
     conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
-            "."  + "physicalio.cache.timeout", 5000);
+            "."  + AAL_CACHE_TIMEOUT, 5000);
 
     enableAnalyticsAccelerator(conf);
     // If encryption is set, some AAL tests will fail.

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAnalyticsAcceleratorStreamReading.java

@@ -52,7 +52,13 @@
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_HTTP_GET_REQUEST;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_HTTP_HEAD_REQUEST;
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.*;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.ANALYTICS_STREAM_FACTORY_CLOSED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_CACHE_HIT;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_OPERATIONS;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PARQUET_FOOTER_PARSING_FAILED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCHED_BYTES;
 import static org.apache.hadoop.io.Sizes.S_1K;
 import static org.apache.hadoop.io.Sizes.S_1M;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
@@ -95,6 +101,12 @@ public void testConnectorFrameWorkIntegration() throws Throwable {
 
     S3AFileSystem fs =
         (S3AFileSystem) FileSystem.get(externalTestFile.toUri(), getConfiguration());
+
+   long fileLength = fs.getFileStatus(externalTestFile).getLen();
+
+   // Head request for the file length.
+    verifyStatisticCounterValue(fs.getIOStatistics(), AUDIT_REQUEST_EXECUTION, 1);
+
     byte[] buffer = new byte[500];
     IOStatistics ioStats;
 
@@ -123,9 +135,9 @@ public void testConnectorFrameWorkIntegration() throws Throwable {
 
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
 
-    // Total file size is: 21511173, and read starts from pos 5. Since policy is WHOLE_FILE, the whole file starts
-    // getting prefetched as soon as the stream to it is opened. So prefetched bytes is 21511173 - 5 = 21511168
-    verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 21511168);
+    // Since policy is WHOLE_FILE, the whole file starts getting prefetched as soon as the stream to it is opened.
+    // So prefetched bytes is fileLen - 5
+    verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, fileLength - 5);
 
     fs.close();
     verifyStatisticCounterValue(fs.getIOStatistics(), ANALYTICS_STREAM_FACTORY_CLOSED, 1);
@@ -134,7 +146,7 @@ public void testConnectorFrameWorkIntegration() throws Throwable {
     // in which case, AAL will start prefetching till EoF on file open in 8MB chunks. The file read here
     // s3://noaa-cors-pds/raw/2023/017/ohfh/OHFH017d.23_.gz, has a size of ~21MB, resulting in 3 GETS:
     // [0-8388607, 8388608-16777215, 16777216-21511173].
-    verifyStatisticCounterValue(fs.getIOStatistics(), AUDIT_REQUEST_EXECUTION, 4);
+    verifyStatisticCounterValue(fs.getIOStatistics(), AUDIT_REQUEST_EXECUTION, 5);
   }
 
   @Test
@@ -146,13 +158,15 @@ public void testSequentialPrefetching() throws IOException {
     // While this works well when running on EC2, for local testing, it can take more than 1s to download large chunks
     // of data. Set this value to higher for testing to prevent early cache evictions.
     conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
-            "."  + "physicalio.cache.timeout", 10000);
+            "."  + AAL_CACHE_TIMEOUT, 10000);
 
     S3AFileSystem fs =
             (S3AFileSystem) FileSystem.get(externalTestFile.toUri(), getConfiguration());
     byte[] buffer = new byte[10 * ONE_MB];
     IOStatistics ioStats;
 
+    long fileLength = fs.getFileStatus(externalTestFile).getLen();
+
     // Here we read through the 21MB external test file, but do not pass in the WHOLE_FILE policy. Instead, we rely
     // on AAL detecting a sequential pattern being read, and then prefetching bytes in a geometrical progression.
     // AAL's sequential prefetching starts prefetching in increments 4MB, 8MB, 16MB etc. depending on how many
@@ -181,20 +195,19 @@ public void testSequentialPrefetching() throws IOException {
       verifyStatisticCounterValue(ioStats, STREAM_READ_CACHE_HIT, 2);
       // A total of 10MB is prefetched - 3MB and then 7MB.
       verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB);
-
+      long bytesRemainingForPrefetch = fileLength - (inputStream.getPos() + 10 * ONE_MB);
       inputStream.readFully(buffer, 0, 10 * ONE_MB);
 
-      // Though the next GP should prefetch 16MB, since the file is ~23MB, only the bytes till EoF are prefetched:
-      // 6291456 remaining bytes.
-      verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB + 6291456);
+
+      // Though the next GP should prefetch 16MB, since the file is ~23MB, only the bytes till EoF are prefetched.
+      verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB + bytesRemainingForPrefetch);
       inputStream.readFully(buffer, 0, 3 * ONE_MB);
       verifyStatisticCounterValue(ioStats, STREAM_READ_CACHE_HIT, 3);
     }
 
     // verify all AAL stats are passed to the FS.
     verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_CACHE_HIT, 3);
     verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_PARQUET_FOOTER_PARSING_FAILED, 0);
-    verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB + 6291456);
   }
 
   @Test
@@ -213,6 +226,8 @@ public void testMalformedParquetFooter() throws IOException {
     Path sourcePath = new Path(file.toURI().getPath());
     getFileSystem().copyFromLocalFile(false, true, sourcePath, dest);
 
+    long fileLength = getFileSystem().getFileStatus(dest).getLen();
+
     byte[] buffer = new byte[500];
     IOStatistics ioStats;
     int bytesRead;
@@ -232,9 +247,9 @@ public void testMalformedParquetFooter() throws IOException {
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
     verifyStatisticCounterValue(ioStats, ACTION_HTTP_GET_REQUEST, 1);
     verifyStatisticCounterValue(ioStats, ACTION_HTTP_HEAD_REQUEST, 0);
-    // This file has a content length of 450. Since it's a parquet file, AAL will prefetch the footer bytes (last 32KB),
+    // This file has a content length of 451. Since it's a parquet file, AAL will prefetch the footer bytes (last 32KB),
     // as soon as the file is opened, but because the file is < 32KB, the whole file is prefetched.
-    verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 450);
+    verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, fileLength);
 
     // The footer is only prefetched once, but parsing is attempted on each stream open.
     verifyStatisticCounterValue(ioStats, STREAM_READ_PARQUET_FOOTER_PARSING_FAILED, 1);
@@ -367,8 +382,10 @@ public void testSequentialStreamsNoDuplicateGets() throws Throwable {
     describe("Sequential streams reading same object should not duplicate GETs");
 
     Path dest = path("sequential-test.txt");
-    byte[] data = dataset(S_1M, 256, 255);
-    writeDataset(getFileSystem(), dest, data, S_1M, 1024, true);
+    int fileLen = S_1M;
+
+    byte[] data = dataset(fileLen, 256, 255);
+    writeDataset(getFileSystem(), dest, data, fileLen, 1024, true);
 
     byte[] buffer = new byte[ONE_MB];
     try (FSDataInputStream stream1 = getFileSystem().open(dest);
@@ -386,14 +403,14 @@ public void testSequentialStreamsNoDuplicateGets() throws Throwable {
 
       // Since it's a small file (ALL will prefetch the whole file for size < 8MB), the whole file is prefetched
       // on the first read.
-      verifyStatisticCounterValue(stats1, STREAM_READ_PREFETCHED_BYTES, 1048575);
+      verifyStatisticCounterValue(stats1, STREAM_READ_PREFETCHED_BYTES, fileLen);
 
       // The second stream will not prefetch any bytes, as they have already been prefetched by stream 1.
       verifyStatisticCounterValue(stats2, STREAM_READ_PREFETCHED_BYTES, 0);
     }
 
     // verify value is passed up to the FS
-    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_PREFETCHED_BYTES, 1048575);
+    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_PREFETCHED_BYTES, fileLen);
 
     // We did 3 reads, all of them were served from the small object cache. In this case, the whole object was
     // downloaded as soon as the stream to it was opened.

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestConstants.java

@@ -284,4 +284,24 @@ public interface S3ATestConstants {
    * Default policy on root tests: {@value}.
    */
   boolean DEFAULT_ROOT_TESTS_ENABLED = true;
+
+  /**
+   * Ranges within this distance of each other will be coalesced.
+   */
+  String AAL_REQUEST_COALESCE_TOLERANCE = "physicalio.request.coalesce.tolerance";
+
+  /**
+   * The minimum size of a block in AAL.
+   */
+  String AAL_READ_BUFFER_SIZE = "physicalio.readbuffersize";
+
+  /**
+   * Objects smaller than this will be downloaded completely.
+   */
+  String AAL_SMALL_OBJECT_PREFETCH_ENABLED = "physicalio.small.objects.prefetching.enabled";
+
+  /**
+   * Objects in AAL's cache will expire after this duration.
+   */
+  String AAL_CACHE_TIMEOUT = "physicalio.cache.timeout";
 }