jacques-n commented on a change in pull request #1573: URL: https://github.com/apache/iceberg/pull/1573#discussion_r502725465
########## File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.aws.s3; + +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.AmazonS3URI; +import com.amazonaws.services.s3.model.GetObjectRequest; +import com.amazonaws.services.s3.model.S3ObjectInputStream; +import java.io.IOException; +import java.util.Arrays; +import org.apache.commons.compress.utils.IOUtils; +import org.apache.iceberg.io.SeekableInputStream; +import org.apache.iceberg.relocated.com.google.common.base.Joiner; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class S3InputStream extends SeekableInputStream { + private static final Logger LOG = LoggerFactory.getLogger(S3InputStream.class); + + private final StackTraceElement[] createStack; + private final AmazonS3 s3; + private final AmazonS3URI location; + + private S3ObjectInputStream stream; + private long pos = 0; + private long next = 0; + private boolean closed = false; + + private int skipSize = 1024 * 1024; + + public S3InputStream(AmazonS3 s3, AmazonS3URI location) { + this.s3 = s3; + this.location = location; + + createStack = Thread.currentThread().getStackTrace(); + } + + @Override + public long getPos() { + return pos; + } + + @Override + public void seek(long newPos) { + Preconditions.checkState(!closed, "already closed"); + Preconditions.checkArgument(newPos >= 0, "position is negative: %s", newPos); + + // this allows a seek beyond the end of the stream but the next read will fail + next = newPos; + } + + @Override + public int read() { + // This stream is wrapped with BufferedInputStream, so this method should never be called + throw new UnsupportedOperationException(); Review comment: Good questions. I haven't looked at the underlying reader to confirm so it may be a moot point. Generally, in our other performance work we've tried to move away from InputStream entirely. For example, we expose [this](https://github.com/dremio/dremio-oss/blob/master/common/src/main/java/com/dremio/io/AsyncByteReader.java#L41 ) as the only input mechanism for streams. Then higher levels consumers should be preallocating space based on say, page size, do a bulk pull from the underlying system (say s3) and then do all byte reads using unsafe byte reads or similar (as opposed to an inputstream stack). For reference, the impl for S3 of the above are in these two classes: [Using AWS Sync Client](https://github.com/dremio/dremio-oss/blob/master/plugins/s3/src/main/java/com/dremio/plugins/s3/store/S3AsyncByteReaderUsingSyncClient.java), [Useing AWS Async Client](https://github.com/dremio/dremio-oss/blob/master/plugins/s3/src/main/java/com/dremio/plugins/s3/store/S3AsyncByteReader.java) Feel free to grab anything that seems useful. Unfortunately, we don't actually use the latter true async implementation because we've found the AWS Java SDk2 async api for S3 to be very buggy (or at least it was when were last testing it). In cases for Azure and similar, we actually chose to use the relatively strong [AsyncHttpClient](https://github.com/AsyncHttpClient/async-http-client) as you can find [here](https://github.com/dremio/dremio-oss/blob/master/plugins/azure/src/main/java/com/dremio/plugins/azure/AzureAsyncReader.java). ---------------------------------------------------------------- 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: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
