S3 input source (#8903)

* add s3 input source for native batch ingestion

* add docs

* fixes

* checkstyle

* lazy splits

* fixes and hella tests

* fix it

* re-use better iterator

* use key

* javadoc and checkstyle

* exception

* oops

* refactor to use S3Coords instead of URI

* remove unused code, add retrying stream to handle s3 stream

* remove unused parameter

* update to latest master

* use list of objects instead of object

* serde test

* refactor and such

* now with the ability to compile

* fix signature and javadocs

* fix conflicts yet again, fix S3 uri stuffs

* more tests, enforce uri for bucket

* javadoc

* oops

* abstract class instead of interface

* null or empty

* better error

Author: clintropolis

core/src/main/java/org/apache/druid/data/input/InputSource.java

@@ -75,9 +75,5 @@ public interface InputSource
    * @param inputFormat        to parse data. It can be null if {@link #needsFormat()} = true
    * @param temporaryDirectory to store temp data. It will be cleaned up automatically once the task is finished.
    */
-  InputSourceReader reader(
-      InputRowSchema inputRowSchema,
-      @Nullable InputFormat inputFormat,
-      File temporaryDirectory
-  );
+  InputSourceReader reader(InputRowSchema inputRowSchema, @Nullable InputFormat inputFormat, File temporaryDirectory);
 }

core/src/main/java/org/apache/druid/data/input/RetryingInputEntity.java

@@ -19,47 +19,58 @@
 
 package org.apache.druid.data.input;
 
-import com.google.common.base.Predicate;
 import org.apache.druid.data.input.impl.RetryingInputStream;
 import org.apache.druid.data.input.impl.prefetch.ObjectOpenFunction;
 import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.utils.CompressionUtils;
 
 import java.io.IOException;
 import java.io.InputStream;
 
-public interface RetryingInputEntity extends InputEntity
+public abstract class RetryingInputEntity implements InputEntity
 {
+  /**
+   * Open a {@link RetryingInputStream} wrapper for an underlying input stream, optionally decompressing the retrying
+   * stream if the file extension matches a known compression, otherwise passing through the retrying stream directly.
+   */
   @Override
-  default InputStream open() throws IOException
+  public InputStream open() throws IOException
   {
-    return new RetryingInputStream<>(
+    RetryingInputStream<?> retryingInputStream = new RetryingInputStream<>(
         this,
         new RetryingInputEntityOpenFunction(),
         getRetryCondition(),
         RetryUtils.DEFAULT_MAX_TRIES
     );
+    return CompressionUtils.decompress(retryingInputStream, getPath());
   }
 
   /**
-   * Directly opens an {@link InputStream} on the input entity.
+   * Directly opens an {@link InputStream} on the input entity. Decompression should be handled externally, and is
+   * handled by the default implementation of {@link #open}, so this should return the raw stream for the object.
    */
-  default InputStream readFromStart() throws IOException
+  protected InputStream readFromStart() throws IOException
   {
     return readFrom(0);
   }
 
   /**
-   * Directly opens an {@link InputStream} starting at the given offset on the input entity.
+   * Directly opens an {@link InputStream} starting at the given offset on the input entity. Decompression should be
+   * handled externally, and is handled by the default implementation of {@link #open},this should return the raw stream
+   * for the object.
    *
    * @param offset an offset to start reading from. A non-negative integer counting
    *               the number of bytes from the beginning of the entity
    */
-  InputStream readFrom(long offset) throws IOException;
+  protected abstract InputStream readFrom(long offset) throws IOException;
 
-  @Override
-  Predicate<Throwable> getRetryCondition();
+  /**
+   * Get path name for this entity, used by the default implementation of {@link #open} to determine if the underlying
+   * stream needs decompressed, based on file extension of the path
+   */
+  protected abstract String getPath();
 
-  class RetryingInputEntityOpenFunction implements ObjectOpenFunction<RetryingInputEntity>
+  private static class RetryingInputEntityOpenFunction implements ObjectOpenFunction<RetryingInputEntity>
   {
     @Override
     public InputStream open(RetryingInputEntity object) throws IOException

core/src/main/java/org/apache/druid/data/input/impl/CloudObjectLocation.java

@@ -0,0 +1,129 @@
+/*
+ * 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.druid.data.input.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.net.URI;
+import java.util.Objects;
+
+/**
+ * Common type for 'bucket' and 'path' concept of cloud objects to allow code sharing between cloud specific
+ * implementations. {@link #bucket} and {@link #path} should NOT be URL encoded.
+ *
+ * The intention is that this is used as a common representation for storage objects as an alternative to dealing in
+ * {@link URI} directly, but still provide a mechanism to round-trip with a URI.
+ *
+ * In common clouds, bucket names must be dns compliant:
+ * https://docs.aws.amazon.com/AmazonS3/latest/dev/BucketRestrictions.html
+ * https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata
+ * https://cloud.google.com/storage/docs/naming
+ *
+ * The constructor ensures that bucket names are DNS compliant by checking that the URL encoded form of the bucket
+ * matches the supplied value. Technically it should probably confirm that the bucket is also all lower-case, but
+ * S3 has a legacy mode where buckets did not have to be compliant so we can't enforce that here unfortunately.
+ */
+public class CloudObjectLocation
+{
+  private final String bucket;
+  private final String path;
+
+  @JsonCreator
+  public CloudObjectLocation(@JsonProperty("bucket") String bucket, @JsonProperty("path") String path)
+  {
+    this.bucket = Preconditions.checkNotNull(StringUtils.maybeRemoveTrailingSlash(bucket));
+    this.path = Preconditions.checkNotNull(StringUtils.maybeRemoveLeadingSlash(path));
+    Preconditions.checkArgument(
+        this.bucket.equals(StringUtils.urlEncode(this.bucket)),
+        "bucket must follow DNS-compliant naming conventions"
+    );
+  }
+
+  public CloudObjectLocation(URI uri)
+  {
+    this(uri.getHost(), uri.getPath());
+  }
+
+  /**
+   * Given a scheme, encode {@link #bucket} and {@link #path} into a {@link URI}.
+   *
+   * In all clouds bucket names must be dns compliant, so it does not require encoding
+   * There is no such restriction on object names, so they will be URL encoded when constructing the URI
+   */
+  public URI toUri(String scheme)
+  {
+    // Encode path, except leave '/' characters unencoded
+    return URI.create(
+        StringUtils.format(
+            "%s://%s/%s",
+            scheme,
+            bucket,
+            StringUtils.replace(StringUtils.urlEncode(path), "%2F", "/")
+        )
+    );
+  }
+
+  @JsonProperty
+  public String getBucket()
+  {
+    return bucket;
+  }
+
+  @JsonProperty
+  public String getPath()
+  {
+    return path;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "CloudObjectLocation{" +
+           "bucket='" + bucket + '\'' +
+           ", path='" + path + '\'' +
+           '}';
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    final CloudObjectLocation that = (CloudObjectLocation) o;
+    return Objects.equals(bucket, that.bucket) &&
+           Objects.equals(path, that.path);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(bucket, path);
+  }
+
+}

core/src/main/java/org/apache/druid/data/input/impl/HttpEntity.java

@@ -27,7 +27,6 @@
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.metadata.PasswordProvider;
-import org.apache.druid.utils.CompressionUtils;
 
 import javax.annotation.Nullable;
 import java.io.IOException;
@@ -36,7 +35,7 @@
 import java.net.URLConnection;
 import java.util.Base64;
 
-public class HttpEntity implements RetryingInputEntity
+public class HttpEntity extends RetryingInputEntity
 {
   private static final Logger LOG = new Logger(HttpEntity.class);
 
@@ -64,12 +63,15 @@ public URI getUri()
   }
 
   @Override
-  public InputStream readFrom(long offset) throws IOException
+  protected InputStream readFrom(long offset) throws IOException
   {
-    return CompressionUtils.decompress(
-        openInputStream(uri, httpAuthenticationUsername, httpAuthenticationPasswordProvider, offset),
-        uri.toString()
-    );
+    return openInputStream(uri, httpAuthenticationUsername, httpAuthenticationPasswordProvider, offset);
+  }
+
+  @Override
+  protected String getPath()
+  {
+    return uri.getPath();
   }
 
   @Override

core/src/main/java/org/apache/druid/java/util/common/StringUtils.java

@@ -237,6 +237,16 @@ public static String urlDecode(String s)
     }
   }
 
+  public static String maybeRemoveLeadingSlash(String s)
+  {
+    return s != null && s.startsWith("/") ? s.substring(1) : s;
+  }
+
+  public static String maybeRemoveTrailingSlash(String s)
+  {
+    return s != null && s.endsWith("/") ? s.substring(0, s.length() - 1) : s;
+  }
+
   /**
    * Removes all occurrences of the given char from the given string. This method is an optimal version of
    * {@link String#replace(CharSequence, CharSequence) s.replace("c", "")}.

core/src/main/java/org/apache/druid/utils/CollectionUtils.java

@@ -23,11 +23,13 @@
 import com.google.common.collect.Maps;
 import org.apache.druid.java.util.common.ISE;
 
+import javax.annotation.Nullable;
 import java.util.AbstractCollection;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Spliterator;
 import java.util.TreeSet;
@@ -116,6 +118,11 @@ public static <K, V, K2> Map<K2, V> mapKeys(Map<K, V> map, Function<K, K2> keyMa
     return result;
   }
 
+  public static boolean isNullOrEmpty(@Nullable List<?> list)
+  {
+    return list == null || list.isEmpty();
+  }
+
   private CollectionUtils()
   {
   }