HADOOP-18637:S3A to support upload of files greater than 2 GB using DiskBlocks

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

@@ -1255,4 +1255,8 @@ private Constants() {
    */
   public static final String PREFETCH_BLOCK_COUNT_KEY = "fs.s3a.prefetch.block.count";
   public static final int PREFETCH_BLOCK_DEFAULT_COUNT = 8;
+
+  public static final String MULTIPART_UPLOADS_ENABLED = "fs.s3a.multipart.uploads.enabled";
+
+  public static final boolean MULTIPART_UPLOAD_ENABLED_DEFAULT = true;
 }

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

@@ -101,7 +101,7 @@ class S3ABlockOutputStream extends OutputStream implements
   private final String key;
 
   /** Size of all blocks. */
-  private final int blockSize;
+  private final long blockSize;
 
   /** IO Statistics. */
   private final IOStatistics iostatistics;
@@ -169,6 +169,9 @@ class S3ABlockOutputStream extends OutputStream implements
   /** Thread level IOStatistics Aggregator. */
   private final IOStatisticsAggregator threadIOStatisticsAggregator;
 
+  /**Is multipart upload allowed? */
+  private final boolean isMultipartEnabled;
+
   /**
    * An S3A output stream which uploads partitions in a separate pool of
    * threads; different {@link S3ADataBlocks.BlockFactory}
@@ -181,7 +184,7 @@ class S3ABlockOutputStream extends OutputStream implements
     this.builder = builder;
     this.key = builder.key;
     this.blockFactory = builder.blockFactory;
-    this.blockSize = (int) builder.blockSize;
+    this.blockSize = builder.blockSize;
     this.statistics = builder.statistics;
     // test instantiations may not provide statistics;
     this.iostatistics = statistics.getIOStatistics();
@@ -200,6 +203,7 @@ class S3ABlockOutputStream extends OutputStream implements
     createBlockIfNeeded();
     LOG.debug("Initialized S3ABlockOutputStream for {}" +
         " output to {}", key, activeBlock);
+    this.isMultipartEnabled = builder.isMultipartAllowed;
     if (putTracker.initialize()) {
       LOG.debug("Put tracker requests multipart upload");
       initMultipartUpload();
@@ -318,7 +322,7 @@ public synchronized void write(byte[] source, int offset, int len)
     statistics.writeBytes(len);
     S3ADataBlocks.DataBlock block = createBlockIfNeeded();
     int written = block.write(source, offset, len);
-    int remainingCapacity = block.remainingCapacity();
+    int remainingCapacity = (int) block.remainingCapacity();
     if (written < len) {
       // not everything was written —the block has run out
       // of capacity
@@ -369,6 +373,9 @@ private synchronized void uploadCurrentBlock(boolean isLast)
    */
   @Retries.RetryTranslated
   private void initMultipartUpload() throws IOException {
+    if (!isMultipartEnabled){
+      return;
+    }
     if (multiPartUpload == null) {
       LOG.debug("Initiating Multipart upload");
       multiPartUpload = new MultiPartUpload(key);
@@ -558,19 +565,21 @@ public String toString() {
   }
 
   /**
-   * Upload the current block as a single PUT request; if the buffer
-   * is empty a 0-byte PUT will be invoked, as it is needed to create an
-   * entry at the far end.
-   * @throws IOException any problem.
-   * @return number of bytes uploaded. If thread was interrupted while
-   * waiting for upload to complete, returns zero with interrupted flag set
-   * on this thread.
+   * Upload the current block as a single PUT request; if the buffer is empty a
+   * 0-byte PUT will be invoked, as it is needed to create an entry at the far
+   * end.
+   *
+   * @return number of bytes uploaded. If thread was interrupted while waiting
+   *     for upload to complete, returns zero with interrupted flag set on this
+   *     thread.
+   * @throws IOException
+   *     any problem.
    */
-  private int putObject() throws IOException {
+  private long putObject() throws IOException {
     LOG.debug("Executing regular upload for {}", writeOperationHelper);
 
     final S3ADataBlocks.DataBlock block = getActiveBlock();
-    int size = block.dataSize();
+    long size = block.dataSize();
     final S3ADataBlocks.BlockUploadData uploadData = block.startUpload();
     final PutObjectRequest putObjectRequest = uploadData.hasFile() ?
         writeOperationHelper.createPutObjectRequest(
@@ -835,7 +844,7 @@ private void uploadBlockAsync(final S3ADataBlocks.DataBlock block,
       Preconditions.checkNotNull(uploadId, "Null uploadId");
       maybeRethrowUploadFailure();
       partsSubmitted++;
-      final int size = block.dataSize();
+      final long size = block.dataSize();
       bytesSubmitted += size;
       final int currentPartNumber = partETagsFutures.size() + 1;
       final UploadPartRequest request;
@@ -1011,7 +1020,7 @@ public void progressChanged(ProgressEvent progressEvent) {
       ProgressEventType eventType = progressEvent.getEventType();
       long bytesTransferred = progressEvent.getBytesTransferred();
 
-      int size = block.dataSize();
+      long size = block.dataSize();
       switch (eventType) {
 
       case REQUEST_BYTE_TRANSFER_EVENT:
@@ -1126,6 +1135,11 @@ public static final class BlockOutputStreamBuilder {
      */
     private IOStatisticsAggregator ioStatisticsAggregator;
 
+    /**
+     * Is Multipart Uploads enabled for the given upload
+     */
+    private boolean isMultipartAllowed;
+
     private BlockOutputStreamBuilder() {
     }
 
@@ -1276,5 +1290,11 @@ public BlockOutputStreamBuilder withIOStatisticsAggregator(
       ioStatisticsAggregator = value;
       return this;
     }
+
+    public BlockOutputStreamBuilder withMultipartAllowed(
+        final boolean value) {
+      isMultipartAllowed = value;
+      return this;
+    }
   }
 }

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

@@ -180,7 +180,7 @@ protected BlockFactory(S3AFileSystem owner) {
      * @param statistics stats to work with
      * @return a new block.
      */
-    abstract DataBlock create(long index, int limit,
+    abstract DataBlock create(long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException;
 
@@ -258,7 +258,7 @@ final DestState getState() {
      * Return the current data size.
      * @return the size of the data
      */
-    abstract int dataSize();
+    abstract long dataSize();
 
     /**
      * Predicate to verify that the block has the capacity to write
@@ -280,7 +280,7 @@ boolean hasData() {
      * The remaining capacity in the block before it is full.
      * @return the number of bytes remaining.
      */
-    abstract int remainingCapacity();
+    abstract long remainingCapacity();
 
     /**
      * Write a series of bytes from the buffer, from the offset.
@@ -391,7 +391,7 @@ static class ArrayBlockFactory extends BlockFactory {
     }
 
     @Override
-    DataBlock create(long index, int limit,
+    DataBlock create(long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       return new ByteArrayBlock(0, limit, statistics);
@@ -436,11 +436,11 @@ static class ByteArrayBlock extends DataBlock {
     private Integer dataSize;
 
     ByteArrayBlock(long index,
-        int limit,
+        long limit,
         BlockOutputStreamStatistics statistics) {
       super(index, statistics);
-      this.limit = limit;
-      buffer = new S3AByteArrayOutputStream(limit);
+      this.limit = (limit > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) limit;
+      buffer = new S3AByteArrayOutputStream((int) limit);
       blockAllocated();
     }
 
@@ -449,7 +449,7 @@ static class ByteArrayBlock extends DataBlock {
      * @return the amount of data available to upload.
      */
     @Override
-    int dataSize() {
+    long dataSize() {
       return dataSize != null ? dataSize : buffer.size();
     }
 
@@ -468,14 +468,14 @@ boolean hasCapacity(long bytes) {
     }
 
     @Override
-    int remainingCapacity() {
+    long remainingCapacity() {
       return limit - dataSize();
     }
 
     @Override
     int write(byte[] b, int offset, int len) throws IOException {
       super.write(b, offset, len);
-      int written = Math.min(remainingCapacity(), len);
+      int written = (int) Math.min(remainingCapacity(), len);
       buffer.write(b, offset, written);
       return written;
     }
@@ -514,7 +514,7 @@ static class ByteBufferBlockFactory extends BlockFactory {
     }
 
     @Override
-    ByteBufferBlock create(long index, int limit,
+    ByteBufferBlock create(long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       return new ByteBufferBlock(index, limit, statistics);
@@ -564,11 +564,12 @@ class ByteBufferBlock extends DataBlock {
        * @param statistics statistics to update
        */
       ByteBufferBlock(long index,
-          int bufferSize,
+          long bufferSize,
           BlockOutputStreamStatistics statistics) {
         super(index, statistics);
-        this.bufferSize = bufferSize;
-        blockBuffer = requestBuffer(bufferSize);
+        this.bufferSize = bufferSize > Integer.MAX_VALUE ?
+            Integer.MAX_VALUE : (int) bufferSize;
+        blockBuffer = requestBuffer(this.bufferSize);
         blockAllocated();
       }
 
@@ -577,7 +578,7 @@ class ByteBufferBlock extends DataBlock {
        * @return the amount of data available to upload.
        */
       @Override
-      int dataSize() {
+      long dataSize() {
         return dataSize != null ? dataSize : bufferCapacityUsed();
       }
 
@@ -598,7 +599,7 @@ public boolean hasCapacity(long bytes) {
       }
 
       @Override
-      public int remainingCapacity() {
+      public long remainingCapacity() {
         return blockBuffer != null ? blockBuffer.remaining() : 0;
       }
 
@@ -609,7 +610,7 @@ private int bufferCapacityUsed() {
       @Override
       int write(byte[] b, int offset, int len) throws IOException {
         super.write(b, offset, len);
-        int written = Math.min(remainingCapacity(), len);
+        int written = (int) Math.min(remainingCapacity(), len);
         blockBuffer.put(b, offset, written);
         return written;
       }
@@ -809,7 +810,7 @@ static class DiskBlockFactory extends BlockFactory {
      */
     @Override
     DataBlock create(long index,
-        int limit,
+        long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       File destFile = getOwner()
@@ -825,14 +826,14 @@ DataBlock create(long index,
    */
   static class DiskBlock extends DataBlock {
 
-    private int bytesWritten;
+    private long bytesWritten;
     private final File bufferFile;
-    private final int limit;
+    private final long limit;
     private BufferedOutputStream out;
     private final AtomicBoolean closed = new AtomicBoolean(false);
 
     DiskBlock(File bufferFile,
-        int limit,
+        long limit,
         long index,
         BlockOutputStreamStatistics statistics)
         throws FileNotFoundException {
@@ -844,7 +845,7 @@ static class DiskBlock extends DataBlock {
     }
 
     @Override
-    int dataSize() {
+    long dataSize() {
       return bytesWritten;
     }
 
@@ -854,14 +855,14 @@ boolean hasCapacity(long bytes) {
     }
 
     @Override
-    int remainingCapacity() {
+    long remainingCapacity() {
       return limit - bytesWritten;
     }
 
     @Override
     int write(byte[] b, int offset, int len) throws IOException {
       super.write(b, offset, len);
-      int written = Math.min(remainingCapacity(), len);
+      int written = (int) Math.min(remainingCapacity(), len);
       out.write(b, offset, written);
       bytesWritten += written;
       return written;

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

@@ -595,7 +595,6 @@ public void initialize(URI name, Configuration originalConf)
       }
       blockOutputBuffer = conf.getTrimmed(FAST_UPLOAD_BUFFER,
           DEFAULT_FAST_UPLOAD_BUFFER);
-      partSize = ensureOutputParameterInRange(MULTIPART_SIZE, partSize);
       blockFactory = S3ADataBlocks.createFactory(this, blockOutputBuffer);
       blockOutputActiveBlocks = intOption(conf,
           FAST_UPLOAD_ACTIVE_BLOCKS, DEFAULT_FAST_UPLOAD_ACTIVE_BLOCKS, 1);
@@ -1831,6 +1830,11 @@ private FSDataOutputStream innerCreateFile(
     final PutObjectOptions putOptions =
         new PutObjectOptions(keep, null, options.getHeaders());
 
+    if(!checkDiskBuffer(getConf())){
+      throw new IOException("The filesystem conf is not " +
+          "proper for the output stream");
+    }
+
     final S3ABlockOutputStream.BlockOutputStreamBuilder builder =
         S3ABlockOutputStream.builder()
         .withKey(destKey)
@@ -1854,7 +1858,9 @@ private FSDataOutputStream innerCreateFile(
         .withCSEEnabled(isCSEEnabled)
         .withPutOptions(putOptions)
         .withIOStatisticsAggregator(
-            IOStatisticsContext.getCurrentIOStatisticsContext().getAggregator());
+            IOStatisticsContext.getCurrentIOStatisticsContext().getAggregator())
+            .withMultipartAllowed(getConf().getBoolean(
+                MULTIPART_UPLOADS_ENABLED, MULTIPART_UPLOAD_ENABLED_DEFAULT));
     return new FSDataOutputStream(
         new S3ABlockOutputStream(builder),
         null);

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

@@ -1505,7 +1505,7 @@ public void blockReleased() {
      * of block uploads pending (1) and the bytes pending (blockSize).
      */
     @Override
-    public void blockUploadQueued(int blockSize) {
+    public void blockUploadQueued(long blockSize) {
       incCounter(StreamStatisticNames.STREAM_WRITE_BLOCK_UPLOADS);
       incAllGauges(STREAM_WRITE_BLOCK_UPLOADS_PENDING, 1);
       incAllGauges(STREAM_WRITE_BLOCK_UPLOADS_BYTES_PENDING, blockSize);
@@ -1518,7 +1518,7 @@ public void blockUploadQueued(int blockSize) {
      * {@code STREAM_WRITE_BLOCK_UPLOADS_ACTIVE}.
      */
     @Override
-    public void blockUploadStarted(Duration timeInQueue, int blockSize) {
+    public void blockUploadStarted(Duration timeInQueue, long blockSize) {
       // the local counter is used in toString reporting.
       queueDuration.addAndGet(timeInQueue.toMillis());
       // update the duration fields in the IOStatistics.
@@ -1546,7 +1546,7 @@ private IOStatisticsStore localIOStatistics() {
     @Override
     public void blockUploadCompleted(
         Duration timeSinceUploadStarted,
-        int blockSize) {
+        long blockSize) {
       transferDuration.addAndGet(timeSinceUploadStarted.toMillis());
       incAllGauges(STREAM_WRITE_BLOCK_UPLOADS_ACTIVE, -1);
       blockUploadsCompleted.incrementAndGet();
@@ -1560,7 +1560,7 @@ public void blockUploadCompleted(
     @Override
     public void blockUploadFailed(
         Duration timeSinceUploadStarted,
-        int blockSize) {
+        long blockSize) {
       incCounter(StreamStatisticNames.STREAM_WRITE_EXCEPTIONS);
     }