[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r448150182
##
File path:
extensions-contrib/aliyun-oss-extensions/src/main/java/org/apache/druid/storage/aliyun/OssStorageDruidModule.java
##
@@ -0,0 +1,106 @@
+/*
+ * 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.storage.aliyun;
+
+import com.aliyun.oss.OSS;
+import com.fasterxml.jackson.core.Version;
+import com.fasterxml.jackson.databind.Module;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import com.google.inject.multibindings.MapBinder;
+import org.apache.druid.data.SearchableVersionedDataFinder;
+import org.apache.druid.data.input.aliyun.OssClientConfig;
+import org.apache.druid.guice.Binders;
+import org.apache.druid.guice.JsonConfigProvider;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.initialization.DruidModule;
+
+import java.util.List;
+
+public class OssStorageDruidModule implements DruidModule
+{
+ public static final String SCHEME = "oss";
+ public static final String SCHEME_ZIP = "oss_zip";
+
+ @Override
+ public List getJacksonModules()
+ {
+return ImmutableList.of(
+new Module()
+{
+ @Override
+ public String getModuleName()
+ {
+return "DruidOss-" + System.identityHashCode(this);
Review comment:
it's been fixed.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r448148488
##
File path:
extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssDataSegmentMoverTest.java
##
@@ -0,0 +1,266 @@
+/*
+ * 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.storage.aliyun;
+
+import com.aliyun.oss.OSSClient;
+import com.aliyun.oss.OSSException;
+import com.aliyun.oss.model.CopyObjectRequest;
+import com.aliyun.oss.model.CopyObjectResult;
+import com.aliyun.oss.model.ListObjectsRequest;
+import com.aliyun.oss.model.OSSObjectSummary;
+import com.aliyun.oss.model.ObjectListing;
+import com.aliyun.oss.model.PutObjectResult;
+import com.aliyun.oss.model.StorageClass;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.MapUtils;
+import org.apache.druid.segment.loading.SegmentLoadingException;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.NoneShardSpec;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class OssDataSegmentMoverTest
+{
+ private static final DataSegment SOURCE_SEGMENT = new DataSegment(
+ "test",
+ Intervals.of("2013-01-01/2013-01-02"),
+ "1",
+ ImmutableMap.of(
+ "key",
+
"baseKey/test/2013-01-01T00:00:00.000Z_2013-01-02T00:00:00.000Z/1/0/index.zip",
+ "bucket",
+ "main"
+ ),
+ ImmutableList.of("dim1", "dim1"),
+ ImmutableList.of("metric1", "metric2"),
+ NoneShardSpec.instance(),
+ 0,
+ 1
+ );
+
+ @Test
+ public void testMove() throws Exception
+ {
+MockClient mockClient = new MockClient();
+OssDataSegmentMover mover = new OssDataSegmentMover(mockClient, new
OssStorageConfig());
+
+mockClient.putObject(
+"main",
+
"baseKey/test/2013-01-01T00:00:00.000Z_2013-01-02T00:00:00.000Z/1/0/index.zip"
+);
+
+DataSegment movedSegment = mover.move(
+SOURCE_SEGMENT,
+ImmutableMap.of("baseKey", "targetBaseKey", "bucket", "archive")
+);
+
+Map targetLoadSpec = movedSegment.getLoadSpec();
+Assert.assertEquals(
+
"targetBaseKey/test/2013-01-01T00:00:00.000Z_2013-01-02T00:00:00.000Z/1/0/index.zip",
+MapUtils.getString(targetLoadSpec, "key")
+);
+Assert.assertEquals("archive", MapUtils.getString(targetLoadSpec,
"bucket"));
+Assert.assertTrue(mockClient.didMove());
+ }
+
+ @Test
+ public void testMoveNoop() throws Exception
+ {
+MockClient mockS3Client = new MockClient();
Review comment:
all `s3` in this extension have been fixed
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898: URL: https://github.com/apache/druid/pull/9898#discussion_r448146198 ## File path: integration-tests/src/test/java/org/apache/druid/tests/TestNGGroup.java ## @@ -76,6 +76,13 @@ */ public static final String AZURE_DEEP_STORAGE = "azure-deep-storage"; + /** + * This group is not part of CI. To run this group, azure configs/credentials for your azure must be provided in a file. Review comment: it's been fixed This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org For additional commands, e-mail: commits-h...@druid.apache.org
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r446116676
##
File path:
extensions-contrib/aliyun-oss-extensions/src/main/java/org/apache/druid/storage/aliyun/ObjectSummaryIterator.java
##
@@ -0,0 +1,163 @@
+/*
+ * 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.storage.aliyun;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSException;
+import com.aliyun.oss.model.ListObjectsRequest;
+import com.aliyun.oss.model.OSSObjectSummary;
+import com.aliyun.oss.model.ObjectListing;
+import org.apache.druid.java.util.common.RE;
+
+import java.net.URI;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * Iterator class used by {@link OssUtils#objectSummaryIterator}.
+ *
+ * As required by the specification of that method, this iterator is computed
incrementally in batches of
+ * {@code maxListLength}. The first call is made at the same time the iterator
is constructed.
+ *
+ */
+public class ObjectSummaryIterator implements Iterator
+{
+ private final OSS client;
+ private final Iterator prefixesIterator;
+ private final int maxListingLength;
+
+ private ListObjectsRequest request;
+ private ObjectListing result;
+ private Iterator objectSummaryIterator;
+ private OSSObjectSummary currentObjectSummary;
+
+ ObjectSummaryIterator(
+ final OSS client,
+ final Iterable prefixes,
+ final int maxListingLength
+ )
+ {
+this.client = client;
+this.prefixesIterator = prefixes.iterator();
+this.maxListingLength = maxListingLength;
+
+prepareNextRequest();
+fetchNextBatch();
+advanceObjectSummary();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+return currentObjectSummary != null;
+ }
+
+ @Override
+ public OSSObjectSummary next()
+ {
+if (currentObjectSummary == null) {
+ throw new NoSuchElementException();
+}
+
+final OSSObjectSummary retVal = currentObjectSummary;
+advanceObjectSummary();
+return retVal;
+ }
+
+ private void prepareNextRequest()
+ {
+final URI currentUri = prefixesIterator.next();
+final String currentBucket = currentUri.getAuthority();
+final String currentPrefix = OssUtils.extractKey(currentUri);
+
+request = new ListObjectsRequest(currentBucket, currentPrefix, null, null,
maxListingLength);
+ }
+
+ private void fetchNextBatch()
+ {
+try {
+ result = OssUtils.retry(() -> client.listObjects(request));
+ request.setMarker(result.getNextMarker());
+ objectSummaryIterator = result.getObjectSummaries().iterator();
+}
+catch (OSSException e) {
+ throw new RE(
+ e,
+ "Failed to get object summaries from S3 bucket[%s], prefix[%s]; S3
error: %s",
+ request.getBucketName(),
+ request.getPrefix(),
+ e.getMessage()
+ );
+}
+catch (Exception e) {
+ throw new RE(
+ e,
+ "Failed to get object summaries from S3 bucket[%s], prefix[%s]",
+ request.getBucketName(),
+ request.getPrefix()
+ );
+}
+ }
+
+ /**
+ * Advance objectSummaryIterator to the next non-placeholder, updating
"currentObjectSummary".
+ */
+ private void advanceObjectSummary()
+ {
+while (objectSummaryIterator.hasNext() || result.isTruncated() ||
prefixesIterator.hasNext()) {
+ while (objectSummaryIterator.hasNext()) {
+currentObjectSummary = objectSummaryIterator.next();
+// skips directories and empty objects
+if (!isDirectoryPlaceholder(currentObjectSummary) &&
currentObjectSummary.getSize() > 0) {
+ return;
+}
+ }
+
+ // Exhausted "objectSummaryIterator" without finding a non-placeholder.
+ if (result.isTruncated()) {
+fetchNextBatch();
+ } else if (prefixesIterator.hasNext()) {
+prepareNextRequest();
+fetchNextBatch();
+ }
+}
+
+// Truly nothing left to read.
+currentObjectSummary = null;
+ }
+
+ /**
+ * Checks if a given object is a directory placeholder and should be ignored.
Review comment:
@jon-wei LICENSE has been updated in the lastest commit, please check
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r445616439
##
File path:
extensions-contrib/aliyun-oss-extensions/src/main/java/org/apache/druid/storage/aliyun/OssDataSegmentPusher.java
##
@@ -0,0 +1,131 @@
+/*
+ * 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.storage.aliyun;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSException;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.segment.SegmentUtils;
+import org.apache.druid.segment.loading.DataSegmentPusher;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CompressionUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+
+public class OssDataSegmentPusher implements DataSegmentPusher
+{
+ private static final EmittingLogger log = new
EmittingLogger(OssDataSegmentPusher.class);
+
+ private final OSS client;
+ private final OssStorageConfig config;
+
+ @Inject
+ public OssDataSegmentPusher(
+ OSS client,
+ OssStorageConfig config
+ )
+ {
+this.client = client;
+this.config = config;
+ }
+
+ @Override
+ public String getPathForHadoop()
+ {
+return StringUtils.format("%s/%s", config.getBucket(), config.getPrefix());
Review comment:
It has not been tested in hadoop cluster yet, because in our test
environment, all data are ingested from kafka.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r445615267
##
File path:
extensions-contrib/aliyun-oss-extensions/src/main/java/org/apache/druid/storage/aliyun/ObjectSummaryIterator.java
##
@@ -0,0 +1,163 @@
+/*
+ * 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.storage.aliyun;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSException;
+import com.aliyun.oss.model.ListObjectsRequest;
+import com.aliyun.oss.model.OSSObjectSummary;
+import com.aliyun.oss.model.ObjectListing;
+import org.apache.druid.java.util.common.RE;
+
+import java.net.URI;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * Iterator class used by {@link OssUtils#objectSummaryIterator}.
+ *
+ * As required by the specification of that method, this iterator is computed
incrementally in batches of
+ * {@code maxListLength}. The first call is made at the same time the iterator
is constructed.
+ *
+ */
+public class ObjectSummaryIterator implements Iterator
+{
+ private final OSS client;
+ private final Iterator prefixesIterator;
+ private final int maxListingLength;
+
+ private ListObjectsRequest request;
+ private ObjectListing result;
+ private Iterator objectSummaryIterator;
+ private OSSObjectSummary currentObjectSummary;
+
+ ObjectSummaryIterator(
+ final OSS client,
+ final Iterable prefixes,
+ final int maxListingLength
+ )
+ {
+this.client = client;
+this.prefixesIterator = prefixes.iterator();
+this.maxListingLength = maxListingLength;
+
+prepareNextRequest();
+fetchNextBatch();
+advanceObjectSummary();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+return currentObjectSummary != null;
+ }
+
+ @Override
+ public OSSObjectSummary next()
+ {
+if (currentObjectSummary == null) {
+ throw new NoSuchElementException();
+}
+
+final OSSObjectSummary retVal = currentObjectSummary;
+advanceObjectSummary();
+return retVal;
+ }
+
+ private void prepareNextRequest()
+ {
+final URI currentUri = prefixesIterator.next();
+final String currentBucket = currentUri.getAuthority();
+final String currentPrefix = OssUtils.extractKey(currentUri);
+
+request = new ListObjectsRequest(currentBucket, currentPrefix, null, null,
maxListingLength);
+ }
+
+ private void fetchNextBatch()
+ {
+try {
+ result = OssUtils.retry(() -> client.listObjects(request));
+ request.setMarker(result.getNextMarker());
+ objectSummaryIterator = result.getObjectSummaries().iterator();
+}
+catch (OSSException e) {
+ throw new RE(
+ e,
+ "Failed to get object summaries from S3 bucket[%s], prefix[%s]; S3
error: %s",
+ request.getBucketName(),
+ request.getPrefix(),
+ e.getMessage()
+ );
+}
+catch (Exception e) {
+ throw new RE(
+ e,
+ "Failed to get object summaries from S3 bucket[%s], prefix[%s]",
+ request.getBucketName(),
+ request.getPrefix()
+ );
+}
+ }
+
+ /**
+ * Advance objectSummaryIterator to the next non-placeholder, updating
"currentObjectSummary".
+ */
+ private void advanceObjectSummary()
+ {
+while (objectSummaryIterator.hasNext() || result.isTruncated() ||
prefixesIterator.hasNext()) {
+ while (objectSummaryIterator.hasNext()) {
+currentObjectSummary = objectSummaryIterator.next();
+// skips directories and empty objects
+if (!isDirectoryPlaceholder(currentObjectSummary) &&
currentObjectSummary.getSize() > 0) {
+ return;
+}
+ }
+
+ // Exhausted "objectSummaryIterator" without finding a non-placeholder.
+ if (result.isTruncated()) {
+fetchNextBatch();
+ } else if (prefixesIterator.hasNext()) {
+prepareNextRequest();
+fetchNextBatch();
+ }
+}
+
+// Truly nothing left to read.
+currentObjectSummary = null;
+ }
+
+ /**
+ * Checks if a given object is a directory placeholder and should be ignored.
Review comment:
There's no directory placeholder in aliyun OSS, so I made change to thi
[GitHub] [druid] FrankChen021 commented on a change in pull request #9898: support Aliyun OSS service as deep storage
FrankChen021 commented on a change in pull request #9898:
URL: https://github.com/apache/druid/pull/9898#discussion_r445612812
##
File path:
extensions-contrib/aliyun-oss-extensions/src/main/java/org/apache/druid/data/input/aliyun/OssInputSource.java
##
@@ -0,0 +1,178 @@
+/*
+ * 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.aliyun;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.model.OSSObjectSummary;
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import org.apache.druid.data.input.InputEntity;
+import org.apache.druid.data.input.InputFileAttribute;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.data.input.SplitHintSpec;
+import org.apache.druid.data.input.impl.CloudObjectInputSource;
+import org.apache.druid.data.input.impl.CloudObjectLocation;
+import org.apache.druid.data.input.impl.SplittableInputSource;
+import org.apache.druid.storage.aliyun.OssInputDataConfig;
+import org.apache.druid.storage.aliyun.OssStorageDruidModule;
+import org.apache.druid.storage.aliyun.OssUtils;
+import org.apache.druid.utils.Streams;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.net.URI;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class OssInputSource extends CloudObjectInputSource
+{
+ private final Supplier clientSupplier;
+ @JsonProperty("properties")
+ private final OssClientConfig inputSourceConfig;
+ private final OssInputDataConfig inputDataConfig;
+
+ /**
+ * Constructor for OssInputSource
+ *
+ * @param clientThe default client built with all default configs
+ * from Guice. This injected singleton client is
used when {@param inputSourceConfig}
+ * is not provided and hence
+ * @param inputDataConfig Stores the configuration for options related to
reading input data
+ * @param uris User provided uris to read input data
+ * @param prefixes User provided prefixes to read input data
+ * @param objects User provided cloud objects values to read input
data
+ * @param inputSourceConfig User provided properties for overriding the
default aliyun-oss configuration
+ */
+ @JsonCreator
+ public OssInputSource(
+ @JacksonInject OSS client,
+ @JacksonInject OssInputDataConfig inputDataConfig,
+ @JsonProperty("uris") @Nullable List uris,
+ @JsonProperty("prefixes") @Nullable List prefixes,
+ @JsonProperty("objects") @Nullable List objects,
+ @JsonProperty("properties") @Nullable OssClientConfig inputSourceConfig
+ )
+ {
+super(OssStorageDruidModule.SCHEME, uris, prefixes, objects);
+this.inputDataConfig = Preconditions.checkNotNull(inputDataConfig,
"inputDataConfig");
+Preconditions.checkNotNull(client, "client");
+this.inputSourceConfig = inputSourceConfig;
+this.clientSupplier = Suppliers.memoize(
+() -> {
+ if (inputSourceConfig != null) {
+return inputSourceConfig.buildClient();
+ } else {
+return client;
+ }
+}
+);
+ }
+
+
+ @Nullable
+ @JsonProperty("properties")
+ public OssClientConfig getOssInputSourceConfig()
+ {
+return inputSourceConfig;
+ }
+
+ @Override
+ protected InputEntity createEntity(CloudObjectLocation location)
+ {
+return new OssEntity(clientSupplier.get(), location);
+ }
+
+ @Override
+ protected Stream>>
getPrefixesSplitStream(@Nonnull SplitHintSpec splitHintSpec)
+ {
+final Iterator> splitIterator = splitHintSpec.split(
+getIterableObjectsFromPrefixes().iterator(),
+object -> new InputFileAttribute(object.getSize())
+);
+
+return Streams.sequentialStreamFrom(splitIterator)
+ .map(objects -> objects.stream()
+
