HDFS-15484. Add new method batchRename for DistributedFileSystem and W…

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/BatchOperations.java

@@ -0,0 +1,43 @@
+/*
+ * 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;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import java.io.IOException;
+
+/**
+ * Interface filesystems MAY implement to offer a batched operations.
+ */
+
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public interface BatchOperations {
+
+  /**
+   * Batched rename API that rename a batch of files.
+   *
+   * @param srcs source file list.
+   * @param dsts target file list.
+   * @throws IOException failure exception.
+   */
+   void batchRename(String[] srcs, String[] dsts, Options.Rename... options)
+       throws IOException;
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java

@@ -139,4 +139,11 @@ private CommonPathCapabilities() {
   public static final String FS_MULTIPART_UPLOADER =
       "fs.capability.multipart.uploader";
 
+  /**
+   * Does the store support multipart uploading?
+   * Value: {@value}.
+   */
+  public static final String FS_BATCH_RENAME =
+      "fs.capability.batch.rename";
+
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -105,6 +105,7 @@
 import org.apache.hadoop.hdfs.net.Peer;
 import org.apache.hadoop.hdfs.protocol.AclException;
 import org.apache.hadoop.hdfs.protocol.AddErasureCodingPolicyResponse;
+import org.apache.hadoop.hdfs.protocol.BatchOpsException;
 import org.apache.hadoop.hdfs.protocol.BatchedDirectoryListing;
 import org.apache.hadoop.hdfs.protocol.BlockStoragePolicy;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
@@ -1609,6 +1610,26 @@ public void rename(String src, String dst, Options.Rename... options)
     }
   }
 
+  /**
+   * Rename a batch files or directories.
+   * @see ClientProtocol#batchRename(String[] , String[], Options.Rename...)
+   */
+  public void batchRename(String[] srcs, String[] dsts,
+      Options.Rename... options) throws IOException {
+    checkOpen();
+    try {
+      namenode.batchRename(srcs, dsts, options);
+    } catch(RemoteException re) {
+      throw re.unwrapRemoteException(AccessControlException.class,
+          NSQuotaExceededException.class,
+          DSQuotaExceededException.class,
+          UnresolvedPathException.class,
+          SnapshotAccessControlException.class,
+          BatchOpsException.class);
+    }
+  }
+
+
   /**
    * Truncate a file to an indicated size
    * See {@link ClientProtocol#truncate}.

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOpsCountStatistics.java

@@ -114,7 +114,8 @@ public enum OpType {
     GET_SNAPSHOT_LIST("op_get_snapshot_list"),
     TRUNCATE(CommonStatisticNames.OP_TRUNCATE),
     UNSET_EC_POLICY("op_unset_ec_policy"),
-    UNSET_STORAGE_POLICY("op_unset_storage_policy");
+    UNSET_STORAGE_POLICY("op_unset_storage_policy"),
+    BATCH_RENAME("op_batch_rename");
 
     private static final Map<String, OpType> SYMBOL_MAP =
         new HashMap<>(OpType.values().length);

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -30,6 +30,7 @@
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderTokenIssuer;
 import org.apache.hadoop.fs.BatchListingOperations;
+import org.apache.hadoop.fs.BatchOperations;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.BlockStoragePolicySpi;
 import org.apache.hadoop.fs.CacheFlag;
@@ -50,6 +51,7 @@
 import org.apache.hadoop.fs.FsStatus;
 import org.apache.hadoop.fs.GlobalStorageStatistics;
 import org.apache.hadoop.fs.GlobalStorageStatistics.StorageStatisticsProvider;
+import org.apache.hadoop.fs.InvalidPathException;
 import org.apache.hadoop.fs.InvalidPathHandleException;
 import org.apache.hadoop.fs.PartialListing;
 import org.apache.hadoop.fs.MultipartUploaderBuilder;
@@ -147,7 +149,7 @@
 @InterfaceAudience.LimitedPrivate({ "MapReduce", "HBase" })
 @InterfaceStability.Unstable
 public class DistributedFileSystem extends FileSystem
-    implements KeyProviderTokenIssuer, BatchListingOperations {
+    implements KeyProviderTokenIssuer, BatchListingOperations, BatchOperations{
   private Path workingDir;
   private URI uri;
 
@@ -965,6 +967,25 @@ public Void next(final FileSystem fs, final Path p)
     }
   }
 
+  protected String[] getBatchPathName(String[] files) {
+    List<String> ret = new ArrayList<>();
+    for(String f :  files) {
+      ret.add(getPathName(new Path(f)));
+    }
+    return ret.toArray(new String[ret.size()]);
+  }
+
+  @Override
+  public void batchRename(final String[] srcs, final String[] dsts,
+      final Options.Rename... options) throws IOException {
+    if (srcs.length != dsts.length) {
+      throw new InvalidPathException("mismatch batch path src: " +
+          Arrays.toString(srcs) + " dst: " + Arrays.toString(dsts));
+    }
+    statistics.incrementWriteOps(1);
+    dfs.batchRename(getBatchPathName(srcs), getBatchPathName(dsts));
+  }
+
   @Override
   public boolean truncate(Path f, final long newLength) throws IOException {
     statistics.incrementWriteOps(1);

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/DfsPathCapabilities.java

@@ -53,6 +53,7 @@ public static Optional<Boolean> hasPathCapability(final Path path,
     case CommonPathCapabilities.FS_SNAPSHOTS:
     case CommonPathCapabilities.FS_STORAGEPOLICY:
     case CommonPathCapabilities.FS_XATTRS:
+    case CommonPathCapabilities.FS_BATCH_RENAME:
       return Optional.of(true);
     case CommonPathCapabilities.FS_SYMLINKS:
       return Optional.of(FileSystem.areSymlinksEnabled());

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/BatchOpsException.java

@@ -0,0 +1,87 @@
+/**
+ * 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.hdfs.protocol;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import java.io.IOException;
+
+/**
+ * Thrown when break during a batch operation .
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class BatchOpsException extends IOException {
+  private static final long serialVersionUID = 1L;
+  private static final String TAG_INDEX = "index";
+  private static final String TAG_TOTAL = "total";
+  private static final String TAG_REASON = "reason";
+
+  /**
+   * Used by RemoteException to instantiate an BatchOpsException.
+   */
+  public BatchOpsException(String msg) {
+    super(msg);
+  }
+
+  public BatchOpsException(long index, long total, Throwable cause) {
+    this(index, total,
+        cause.getClass().getName() + ": " + cause.getMessage());
+  }
+
+  public BatchOpsException(long index, long total,
+                           String cause) {
+    super("Batch operation break! " +
+        getTagHeader(TAG_INDEX) + index + getTagTailer(TAG_INDEX) +
+        getTagHeader(TAG_TOTAL) + total + getTagTailer(TAG_TOTAL) +
+        getTagHeader(TAG_REASON) + cause + getTagTailer(TAG_REASON));
+  }
+
+  public long getIndex() {
+    return Long.parseLong(getValue(TAG_INDEX));
+  }
+
+  public long getTotal() {
+    return Long.parseLong(getValue(TAG_TOTAL));
+  }
+
+  public String getReason() {
+    return getValue(TAG_REASON);
+  }
+
+  private static String getTagHeader(String tag) {
+    return "<"+tag + ">";
+  }
+
+  private static String getTagTailer(String tag) {
+    return "</"+tag + ">";
+  }
+
+  private String getValue(String target) {
+    String msg = getMessage();
+    String header = getTagHeader(target);
+    String tailer = getTagTailer(target);
+    int pos1 = msg.indexOf(header) + header.length();
+    int pos2 = msg.indexOf(tailer, pos1);
+
+    assert pos2 > pos1;
+    return msg.substring(pos1, pos2);
+  }
+}

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -583,6 +583,19 @@ void concat(String trg, String[] srcs)
   void rename2(String src, String dst, Options.Rename... options)
       throws IOException;
 
+
+  /**
+   * Rename an batch items in the file system namespace.
+   * @param srcs existing files or directories name.216
+   * @param dsts new names.
+   * @param options Rename options
+   *
+   * @throws IOException an I/O error occurred
+   */
+  @AtMostOnce
+  void batchRename(String[] srcs, String[] dsts,
+      Options.Rename... options) throws IOException;
+
   /**
    * Truncate file src to new size.
    * <ul>

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java

@@ -104,6 +104,7 @@
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.AllowSnapshotRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.AppendRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.AppendResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.BatchRenameRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CachePoolEntryProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CheckAccessRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CompleteRequestProto;
@@ -637,6 +638,33 @@ public void rename2(String src, String dst, Rename... options)
 
   }
 
+  @Override
+  public void batchRename(String[] srcs, String[] dsts, Rename... options)
+      throws IOException {
+    boolean overwrite = false;
+    boolean toTrash = false;
+    if (options != null) {
+      for (Rename option : options) {
+        if (option == Rename.OVERWRITE) {
+          overwrite = true;
+        }
+        if (option == Rename.TO_TRASH) {
+          toTrash = true;
+        }
+      }
+    }
+    BatchRenameRequestProto req = BatchRenameRequestProto.newBuilder()
+        .addAllSrcs(Arrays.asList(srcs))
+        .addAllDsts(Arrays.asList(dsts))
+        .setOverwriteDest(overwrite)
+        .setMoveToTrash(toTrash).build();
+    try {
+      rpcProxy.batchRename(null, req);
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
   @Override
   public void concat(String trg, String[] srcs) throws IOException {
     ConcatRequestProto req = ConcatRequestProto.newBuilder().

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java

@@ -40,6 +40,7 @@
 import java.nio.charset.StandardCharsets;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.Base64.Decoder;
 import java.util.Collection;
@@ -61,9 +62,9 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderTokenIssuer;
+import org.apache.hadoop.fs.BatchOperations;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.hadoop.fs.CommonPathCapabilities;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.DelegationTokenRenewer;
@@ -76,9 +77,9 @@
 import org.apache.hadoop.fs.FsServerDefaults;
 import org.apache.hadoop.fs.GlobalStorageStatistics;
 import org.apache.hadoop.fs.GlobalStorageStatistics.StorageStatisticsProvider;
+import org.apache.hadoop.fs.InvalidPathException;
 import org.apache.hadoop.fs.MultipartUploaderBuilder;
 import org.apache.hadoop.fs.QuotaUsage;
-import org.apache.hadoop.fs.PathCapabilities;
 import org.apache.hadoop.fs.StorageStatistics;
 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.fs.impl.FileSystemMultipartUploaderBuilder;
@@ -144,7 +145,8 @@
 /** A FileSystem for HDFS over the web. */
 public class WebHdfsFileSystem extends FileSystem
     implements DelegationTokenRenewer.Renewable,
-    TokenAspect.TokenManagementDelegator, KeyProviderTokenIssuer {
+    TokenAspect.TokenManagementDelegator, KeyProviderTokenIssuer,
+    BatchOperations {
   public static final Logger LOG = LoggerFactory
       .getLogger(WebHdfsFileSystem.class);
   /** WebHdfs version. */
@@ -1184,6 +1186,34 @@ public void rename(final Path src, final Path dst,
     ).run();
   }
 
+  protected String[] getBatchPathName(String[] files) throws IOException{
+    List<String> ret = new ArrayList<>();
+    for(String f :  files) {
+      if(!f.startsWith(Path.SEPARATOR)) {
+        throw new InvalidPathException("Path is not absolute! " + f);
+      }
+      ret.add(makeQualified(new Path(f)).toUri().getPath());
+    }
+    return ret.toArray(new String[ret.size()]);
+  }
+
+  @Override
+  public void batchRename(final String[] srcs, final String[] dsts,
+      final Options.Rename... options) throws IOException {
+    statistics.incrementWriteOps(1);
+    storageStatistics.incrementOpCounter(OpType.BATCH_RENAME);
+    final HttpOpParam.Op op = PutOpParam.Op.BATCH_RENAME;
+    if (srcs.length != dsts.length) {
+      throw new InvalidPathException("mismatch batch path src: " +
+          Arrays.toString(srcs) + " dst: " + Arrays.toString(dsts));
+    }
+    new FsPathRunner(op,
+        new Path(StringUtils.join(":", getBatchPathName(srcs))),
+        new DestinationParam(StringUtils.join(":", getBatchPathName(dsts))),
+        new RenameOptionSetParam(options)
+    ).run();
+  }
+
   @Override
   public void setXAttr(Path p, String name, byte[] value,
       EnumSet<XAttrSetFlag> flag) throws IOException {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/resources/PutOpParam.java

@@ -60,6 +60,7 @@ public enum Op implements HttpOpParam.Op {
     SETQUOTA(false, HttpURLConnection.HTTP_OK),
     SETQUOTABYSTORAGETYPE(false, HttpURLConnection.HTTP_OK),
 
+    BATCH_RENAME(false, HttpURLConnection.HTTP_OK),
     NULL(false, HttpURLConnection.HTTP_NOT_IMPLEMENTED);
 
     final boolean doOutputAndRedirect;