adds in integration tests for parquet files

Author: ahmarsuhail

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StreamStatisticNames.java

@@ -97,6 +97,13 @@ public final class StreamStatisticNames {
    */
   public static final String STREAM_READ_OPENED = "stream_read_opened";
 
+  /**
+   * Total count of times an analytics input stream was opened.
+   *
+   * Value: {@value}.
+   */
+  public static final String STREAM_READ_ANALYTICS_OPENED = "stream_read_analytics_opened";
+
   /**
    * Count of exceptions raised during input stream reads.
    * Value: {@value}.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java

@@ -28,6 +28,7 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.impl.WeakRefMetricsSource;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.statistics.BlockOutputStreamStatistics;
 import org.apache.hadoop.fs.s3a.statistics.ChangeTrackerStatistics;
 import org.apache.hadoop.fs.s3a.statistics.CommitterStatistics;
@@ -840,6 +841,7 @@ private final class InputStreamStatistics
     private final AtomicLong closed;
     private final AtomicLong forwardSeekOperations;
     private final AtomicLong openOperations;
+    private final AtomicLong analyticsStreamOpenOperations;
     private final AtomicLong readExceptions;
     private final AtomicLong readsIncomplete;
     private final AtomicLong readOperations;
@@ -888,7 +890,8 @@ private InputStreamStatistics(
               StreamStatisticNames.STREAM_READ_VECTORED_OPERATIONS,
               StreamStatisticNames.STREAM_READ_VECTORED_READ_BYTES_DISCARDED,
               StreamStatisticNames.STREAM_READ_VERSION_MISMATCHES,
-              StreamStatisticNames.STREAM_EVICT_BLOCKS_FROM_FILE_CACHE)
+              StreamStatisticNames.STREAM_EVICT_BLOCKS_FROM_FILE_CACHE,
+              StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED)
           .withGauges(STREAM_READ_GAUGE_INPUT_POLICY,
               STREAM_READ_BLOCKS_IN_FILE_CACHE.getSymbol(),
               STREAM_READ_ACTIVE_PREFETCH_OPERATIONS.getSymbol(),
@@ -927,6 +930,9 @@ private InputStreamStatistics(
           StreamStatisticNames.STREAM_READ_SEEK_FORWARD_OPERATIONS);
       openOperations = st.getCounterReference(
           StreamStatisticNames.STREAM_READ_OPENED);
+      analyticsStreamOpenOperations = st.getCounterReference(
+          StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED
+      );
       readExceptions = st.getCounterReference(
           StreamStatisticNames.STREAM_READ_EXCEPTIONS);
       readsIncomplete = st.getCounterReference(
@@ -1030,6 +1036,16 @@ public long streamOpened() {
       return openOperations.getAndIncrement();
     }
 
+    @Override
+    public long streamOpened(InputStreamType type) {
+      switch (type) {
+       case Analytics:
+         return analyticsStreamOpenOperations.getAndIncrement();
+      default:
+        return openOperations.getAndIncrement();
+      }
+    }
+
     /**
      * {@inheritDoc}.
      * If the connection was aborted, increment {@link #aborted}

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

@@ -50,6 +50,7 @@ public AnalyticsStream(final ObjectReadParameters parameters, final S3SeekableIn
     super(InputStreamType.Analytics, parameters);
     S3ObjectAttributes s3Attributes = parameters.getObjectAttributes();
     this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(s3Attributes.getBucket(), s3Attributes.getKey()));
+    getS3AStreamStatistics().streamOpened(InputStreamType.Analytics);
   }
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/statistics/S3AInputStreamStatistics.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.fs.s3a.statistics;
 
 import org.apache.hadoop.fs.impl.prefetch.PrefetchingStatistics;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 
 /**
@@ -53,6 +54,13 @@ public interface S3AInputStreamStatistics extends AutoCloseable,
    */
   long streamOpened();
 
+  /**
+   * A stream of the given type was opened.
+   * @param type type of input stream
+   *  @return the previous count or zero if this is the first opening.
+   */
+  long streamOpened(InputStreamType type);
+
   /**
    * The inner stream was closed.
    * @param abortedConnection flag to indicate the stream was aborted,

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/statistics/impl/EmptyS3AStatisticsContext.java

@@ -22,6 +22,7 @@
 import java.time.Duration;
 
 import org.apache.hadoop.fs.s3a.Statistic;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.statistics.BlockOutputStreamStatistics;
 import org.apache.hadoop.fs.s3a.statistics.ChangeTrackerStatistics;
 import org.apache.hadoop.fs.s3a.statistics.CommitterStatistics;
@@ -164,6 +165,11 @@ public long streamOpened() {
       return 0;
     }
 
+    @Override
+    public long streamOpened(InputStreamType type) {
+      return 0;
+    }
+
     @Override
     public void streamClose(final boolean abortedConnection,
         final long remainingInCurrentRequest) {

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

@@ -0,0 +1,201 @@
+/*
+ * 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.s3a;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.UUID;
+
+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.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 {
+
+  private static final String PHYSICAL_IO_PREFIX = "physicalio";
+  private static final String LOGICAL_IO_PREFIX = "logicalio";
+
+
+  private Configuration conf;
+  private Path testFile;
+
+  @Before
+  public void setUp() throws Exception {
+    super.setup();
+    conf = createConfiguration();
+    testFile = getExternalData(conf);
+  }
+
+  @Override
+  public Configuration createConfiguration() {
+    Configuration configuration = super.createConfiguration();
+    if (isUsingDefaultExternalDataFile(configuration)) {
+      S3ATestUtils.removeBaseAndBucketOverrides(configuration,
+          ENDPOINT);
+    }
+    enableAnalyticsAccelerator(configuration);
+    return configuration;
+  }
+
+  @Test
+  public void testConnectorFrameWorkIntegration() throws IOException {
+    describe("Verify S3 connector framework integration");
+
+    S3AFileSystem fs =
+        (S3AFileSystem) FileSystem.get(testFile.toUri(), conf);
+    byte[] buffer = new byte[500];
+    IOStatistics ioStats;
+
+    try (FSDataInputStream inputStream = fs.open(testFile)) {
+      ioStats = inputStream.getIOStatistics();
+      inputStream.seek(5);
+      inputStream.read(buffer, 0, 500);
+    }
+    verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
+  }
+
+  @Test
+  public void testMalformedParquetFooter() throws IOException {
+    describe("Reading a malformed parquet file should not throw an exception");
+
+    // File with malformed footer take from https://github.com/apache/parquet-testing/blob/master/bad_data/PARQUET-1481.parquet.
+    // This test ensures AAL does not throw exceptions if footer parsing fails. It will only emit a WARN log,
+    // "Unable to parse parquet footer for test/malformedFooter.parquet, parquet prefetch optimisations will be disabled for this key."
+    Path dest = path("malformed_footer.parquet");
+
+    File file = new File("src/test/resources/malformed_footer.parquet");
+    Path sourcePath = new Path(file.toURI().getPath());
+    getFileSystem().copyFromLocalFile(false, true, sourcePath, dest);
+
+    byte[] buffer = new byte[500];
+    IOStatistics ioStats;
+
+    try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
+      ioStats = inputStream.getIOStatistics();
+      inputStream.seek(5);
+      inputStream.read(buffer, 0, 500);
+    }
+
+    verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
+  }
+
+ @Test
+ public void testMultiRowGroupParquet() throws IOException {
+    describe("A parquet file is read successfully");
+
+    Path dest = path("multi_row_group.parquet");
+
+   File file = new File("src/test/resources/multi_row_group.parquet");
+   Path sourcePath = new Path(file.toURI().getPath());
+   getFileSystem().copyFromLocalFile(false, true, sourcePath, dest);
+
+   FileStatus fileStatus = getFileSystem().getFileStatus(dest);
+
+   byte[] buffer = new byte[3000];
+   IOStatistics ioStats;
+
+   try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
+     ioStats = inputStream.getIOStatistics();
+     inputStream.readFully(buffer, 0, (int) fileStatus.getLen());
+   }
+
+   verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
+ }
+
+  @Test
+  public void testConnectorFrameworkConfigurable() {
+    describe("Verify S3 connector framework reads configuration");
+
+    Configuration conf = getConfiguration();
+    removeBaseAndBucketOverrides(conf);
+
+    //Disable Predictive Prefetching
+    conf.set(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
+        "." + LOGICAL_IO_PREFIX + ".prefetching.mode", "all");
+
+    //Set Blobstore Capacity
+    conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
+        "." + PHYSICAL_IO_PREFIX + ".blobstore.capacity", 1);
+
+    ConnectorConfiguration connectorConfiguration =
+        new ConnectorConfiguration(conf, ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX);
+
+    S3SeekableInputStreamConfiguration configuration =
+        S3SeekableInputStreamConfiguration.fromConfiguration(connectorConfiguration);
+
+    Assertions.assertThat(configuration.getLogicalIOConfiguration().getPrefetchingMode())
+            .as("AnalyticsStream configuration is not set to expected value")
+            .isSameAs(PrefetchMode.ALL);
+
+    Assertions.assertThat(configuration.getPhysicalIOConfiguration().getBlobStoreCapacity())
+            .as("AnalyticsStream configuration is not set to expected value")
+            .isEqualTo(1);
+  }
+
+  @Test
+  public void testInvalidConfigurationThrows() throws Exception {
+    describe("Verify S3 connector framework throws with invalid configuration");
+
+    Configuration conf = getConfiguration();
+    removeBaseAndBucketOverrides(conf);
+    //Disable Sequential Prefetching
+    conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
+        "." + PHYSICAL_IO_PREFIX + ".blobstore.capacity", -1);
+
+    ConnectorConfiguration connectorConfiguration =
+        new ConnectorConfiguration(conf, ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX);
+    Assertions.assertThatExceptionOfType(IllegalArgumentException.class)
+            .isThrownBy(() ->
+                    S3SeekableInputStreamConfiguration.fromConfiguration(connectorConfiguration));
+  }
+
+}