bhalchandrap commented on a change in pull request #3736: URL: https://github.com/apache/hadoop/pull/3736#discussion_r778485852
########## File path: hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/read/S3CachingInputStream.java ########## @@ -0,0 +1,180 @@ +/* + * 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.read; + +import com.twitter.util.FuturePool; +import org.apache.hadoop.fs.common.BlockData; +import org.apache.hadoop.fs.common.BlockManager; +import org.apache.hadoop.fs.common.BufferData; +import org.apache.hadoop.fs.s3a.S3AInputStream; +import org.apache.hadoop.fs.s3a.S3AReadOpContext; +import org.apache.hadoop.fs.s3a.S3ObjectAttributes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** + * Provides an {@code InputStream} that allows reading from an S3 file. + * Prefetched blocks are cached to local disk if a seek away from the + * current block is issued. + */ +public class S3CachingInputStream extends S3InputStream { + private static final Logger LOG = LoggerFactory.getLogger(S3CachingInputStream.class); + + // Number of blocks queued for prefching. + private int numBlocksToPrefetch; + + private BlockManager blockManager; + + /** + * Initializes a new instance of the {@code S3CachingInputStream} class. + * + * @param context read-specific operation context. + * @param s3Attributes attributes of the S3 object being read. + * @param client callbacks used for interacting with the underlying S3 client. + * + * @throws IllegalArgumentException if context is null. + * @throws IllegalArgumentException if s3Attributes is null. + * @throws IllegalArgumentException if client is null. + */ + public S3CachingInputStream( + S3AReadOpContext context, + S3ObjectAttributes s3Attributes, + S3AInputStream.InputStreamCallbacks client) { + super(context, s3Attributes, client); + + this.numBlocksToPrefetch = this.getContext().getPrefetchBlockCount(); + int bufferPoolSize = this.numBlocksToPrefetch + 1; + this.blockManager = this.createBlockManager( + this.getContext().getFuturePool(), + this.getReader(), + this.getBlockData(), + bufferPoolSize); + int fileSize = (int) s3Attributes.getLen(); + LOG.debug("Created caching input stream for {} (size = {})", this.getName(), fileSize); + } + + /** + * Moves the current read position so that the next read will occur at {@code pos}. + * + * @param pos the next read will take place at this position. + * + * @throws IllegalArgumentException if pos is outside of the range [0, file size]. + */ + @Override + public void seek(long pos) throws IOException { + this.throwIfClosed(); + this.throwIfInvalidSeek(pos); + + if (!this.getFilePosition().setAbsolute(pos)) { Review comment: done. -- 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. To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-issues-h...@hadoop.apache.org
