pass down file open options such as read policy, file status to AAL

Author: ahmarsuhail

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

@@ -25,12 +25,16 @@
 import org.apache.hadoop.fs.FSExceptionMessages;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.s3a.Retries;
+import org.apache.hadoop.fs.s3a.S3AInputPolicy;
 import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStream;
+import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
+import software.amazon.s3.analyticsaccelerator.util.InputPolicy;
+import software.amazon.s3.analyticsaccelerator.util.OpenFileInformation;
 import software.amazon.s3.analyticsaccelerator.util.S3URI;
 
 /**
@@ -49,7 +53,7 @@ public class AnalyticsStream extends ObjectInputStream implements StreamCapabili
   public AnalyticsStream(final ObjectReadParameters parameters, final S3SeekableInputStreamFactory s3SeekableInputStreamFactory) throws IOException {
     super(InputStreamType.Analytics, parameters);
     S3ObjectAttributes s3Attributes = parameters.getObjectAttributes();
-    this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(s3Attributes.getBucket(), s3Attributes.getKey()));
+    this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(s3Attributes.getBucket(), s3Attributes.getKey()), buildOpenFileInformation(parameters));
     getS3AStreamStatistics().streamOpened(InputStreamType.Analytics);
   }
 
@@ -197,6 +201,38 @@ private void onReadFailure(IOException ioe) throws IOException {
     this.close();
   }
 
+  private OpenFileInformation buildOpenFileInformation(ObjectReadParameters parameters) {
+    OpenFileInformation openFileInformation = OpenFileInformation.builder().inputPolicy(mapS3AInputPolicyToAAL(parameters.getContext()
+        .getInputPolicy()))
+        .objectMetadata(ObjectMetadata.builder()
+            .contentLength(parameters.getObjectAttributes().getLen())
+            .etag(parameters.getObjectAttributes().getETag()).build())
+        .build();
+
+    return openFileInformation;
+  }
+
+  /**
+   * If S3A's input policy is Sequential, that is, if the file format to be read is sequential
+   * (CSV, JSON), or the file policy passed down is WHOLE_FILE, then AAL's parquet specific
+   * optimisations will be turned off, regardless of the file extension. This is to allow for
+   * applications like DISTCP that read parquet files, but will read them whole, and so do not
+   * follow the typical parquet read patterns of reading footer first etc. and will not benefit
+   * from parquet optimisations.
+   * Else, AAL will make a decision on which optimisations based on the file extension,
+   * if the file ends in .par or .parquet, then parquet specific optimisations are used.
+   *
+   * @param inputPolicy S3A's input file policy passed down when opening the file
+   * @return the AAL read policy
+   */
+  private InputPolicy mapS3AInputPolicyToAAL(S3AInputPolicy inputPolicy) {
+    switch (inputPolicy) {
+     case Sequential:
+       return InputPolicy.Sequential;
+     default:
+       return InputPolicy.None;
+    }
+  }
 
   protected void throwIfClosed() throws IOException {
     if (closed) {

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

@@ -21,42 +21,34 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.UUID;
+import java.io.InputStream;
 
 import org.junit.Before;
 import org.junit.Test;
 
-import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathHandle;
-import org.apache.hadoop.fs.s3a.impl.AwsSdkWorkarounds;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStream;
 import org.apache.hadoop.fs.statistics.IOStatistics;
-import org.apache.hadoop.test.GenericTestUtils;
 
-import static org.apache.commons.io.FileUtils.ONE_KB;
 import static org.apache.hadoop.fs.s3a.Constants.*;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.enableAnalyticsAccelerator;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.enablePrefetching;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.getExternalData;
 import static org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.isUsingDefaultExternalDataFile;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED;
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_OPENED;
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS;
-import static org.apache.hadoop.test.GenericTestUtils.LogCapturer.captureLogs;
+
 
 import org.assertj.core.api.Assertions;
 
-import org.slf4j.LoggerFactory;
+
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.common.ConnectorConfiguration;
-import software.amazon.s3.analyticsaccelerator.io.logical.parquet.ParquetMetadataParsingTask;
 import software.amazon.s3.analyticsaccelerator.util.PrefetchMode;
 
 public class ITestS3AAnalyticsAcceleratorStream extends AbstractS3ATestBase {
@@ -90,6 +82,9 @@ public Configuration createConfiguration() {
   public void testConnectorFrameWorkIntegration() throws IOException {
     describe("Verify S3 connector framework integration");
 
+    removeBaseAndBucketOverrides(conf, INPUT_FADVISE);
+    conf.set(INPUT_FADVISE, "whole-file");
+
     S3AFileSystem fs =
         (S3AFileSystem) FileSystem.get(testFile.toUri(), conf);
     byte[] buffer = new byte[500];
@@ -99,7 +94,13 @@ public void testConnectorFrameWorkIntegration() throws IOException {
       ioStats = inputStream.getIOStatistics();
       inputStream.seek(5);
       inputStream.read(buffer, 0, 500);
+
+      final InputStream wrappedStream = inputStream.getWrappedStream();
+      ObjectInputStream objectInputStream = (ObjectInputStream) wrappedStream;
+      assertEquals(objectInputStream.streamType(), InputStreamType.Analytics);
+      assertEquals(objectInputStream.getInputPolicy(), S3AInputPolicy.Sequential);
     }
+
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
   }