readVectored stata

Author: ahmarsuhail

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

@@ -55,6 +55,11 @@ public void footerParsingFailed() {
     statistics.footerParsingFailed();
   }
 
+  @Override
+  public void onReadVectored(int numIncomingRanges, int numCombinedRanges) {
+    statistics.readVectoredOperationStarted(numIncomingRanges, numCombinedRanges);
+  }
+
 
 }
 

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

@@ -202,8 +202,6 @@ public void readVectored(final List<? extends FileRange> ranges,
       range.setData(result);
     }
 
-    // AAL does not do any range coalescing, so input and combined ranges are the same.
-    this.getS3AStreamStatistics().readVectoredOperationStarted(ranges.size(), ranges.size());
     inputStream.readVectored(objectRanges, allocate, release);
   }
 

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

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.contract.s3a;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.hadoop.conf.Configuration;
@@ -34,11 +35,11 @@
 import org.junit.jupiter.params.ParameterizedClass;
 import org.junit.jupiter.params.provider.MethodSource;
 
-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.S3ATestUtils.enableAnalyticsAccelerator;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipForAnyEncryptionExceptSSES3;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
+import static org.apache.hadoop.fs.s3a.Constants.ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.*;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
+import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_HTTP_GET_REQUEST;
 
 /**
  * S3A contract tests for vectored reads with the Analytics stream.
@@ -53,6 +54,8 @@
 @MethodSource("params")
 public class ITestS3AContractAnalyticsStreamVectoredRead extends AbstractContractVectoredReadTest {
 
+  private static final int ONE_KB = 1024;
+
   public ITestS3AContractAnalyticsStreamVectoredRead(String bufferType) {
     super(bufferType);
   }
@@ -64,6 +67,20 @@ public ITestS3AContractAnalyticsStreamVectoredRead(String bufferType) {
   @Override
   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", 10 * ONE_KB);
+
+    // 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", 32 * ONE_KB);
+
+    // 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");
+
     enableAnalyticsAccelerator(conf);
     // If encryption is set, some AAL tests will fail.
     // This is because AAL caches the head request response, and uses
@@ -102,21 +119,30 @@ public void testNullReleaseOperation()  {
 
   @Test
   public void testReadVectoredWithAALStatsCollection() throws Exception {
+    List<FileRange> fileRanges = new ArrayList<>();
+    fileRanges.add(FileRange.createFileRange(0, 100));
+    fileRanges.add(FileRange.createFileRange(800, 200));
+    fileRanges.add(FileRange.createFileRange(4 * ONE_KB , 4 * ONE_KB));
+    fileRanges.add(FileRange.createFileRange(80 * ONE_KB , 4 * ONE_KB));
 
-    List<FileRange> fileRanges = createSampleNonOverlappingRanges();
     try (FSDataInputStream in = openVectorFile()) {
       in.readVectored(fileRanges, getAllocate());
 
       validateVectoredReadResult(fileRanges, DATASET, 0);
       IOStatistics st = in.getIOStatistics();
 
-      // Statistics such as GET requests will be added after IoStats support.
       verifyStatisticCounterValue(st,
               StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED, 1);
 
       verifyStatisticCounterValue(st,
               StreamStatisticNames.STREAM_READ_VECTORED_OPERATIONS,
               1);
+
+      // Verify ranges are coalesced, we are using a coalescing tolerance of 10KB, so [0-100, 800-200, 4KB-8KB] will
+      // get coalesced into a single range.
+      verifyStatisticCounterValue(st, StreamStatisticNames.STREAM_READ_VECTORED_INCOMING_RANGES, 4);
+      verifyStatisticCounterValue(st, StreamStatisticNames.STREAM_READ_VECTORED_COMBINED_RANGES, 2);
+      verifyStatisticCounterValue(st, ACTION_HTTP_GET_REQUEST, 2);
     }
   }
 }