HDDS-4040. [OFS] BasicRootedOzoneFileSystem to support batchDelete

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneClientAdapterImpl.java

@@ -176,7 +176,7 @@ public void close() throws IOException {
 
   @Override
   public InputStream readFile(String key) throws IOException {
-    incrementCounter(Statistic.OBJECTS_READ);
+    incrementCounter(Statistic.OBJECTS_READ, 1);
     try {
       return bucket.readFile(key).getInputStream();
     } catch (OMException ex) {
@@ -190,14 +190,19 @@ public InputStream readFile(String key) throws IOException {
     }
   }
 
+  @Deprecated
   protected void incrementCounter(Statistic objectsRead) {
     //noop: Use OzoneClientAdapterImpl which supports statistics.
   }
 
+  protected void incrementCounter(Statistic objectsRead, long count) {
+    //noop: Use OzoneClientAdapterImpl which supports statistics.
+  }
+
   @Override
   public OzoneFSOutputStream createFile(String key, short replication,
       boolean overWrite, boolean recursive) throws IOException {
-    incrementCounter(Statistic.OBJECTS_CREATED);
+    incrementCounter(Statistic.OBJECTS_CREATED, 1);
     try {
       OzoneOutputStream ozoneOutputStream = null;
       if (replication == ReplicationFactor.ONE.getValue()
@@ -224,7 +229,7 @@ public OzoneFSOutputStream createFile(String key, short replication,
 
   @Override
   public void renameKey(String key, String newKeyName) throws IOException {
-    incrementCounter(Statistic.OBJECTS_RENAMED);
+    incrementCounter(Statistic.OBJECTS_RENAMED, 1);
     bucket.renameKey(key, newKeyName);
   }
 
@@ -242,7 +247,7 @@ public void rename(String pathStr, String newPath) throws IOException {
   @Override
   public boolean createDirectory(String keyName) throws IOException {
     LOG.trace("creating dir for key:{}", keyName);
-    incrementCounter(Statistic.OBJECTS_CREATED);
+    incrementCounter(Statistic.OBJECTS_CREATED, 1);
     try {
       bucket.createDirectory(keyName);
     } catch (OMException e) {
@@ -264,7 +269,7 @@ public boolean createDirectory(String keyName) throws IOException {
   public boolean deleteObject(String keyName) {
     LOG.trace("issuing delete for key {}", keyName);
     try {
-      incrementCounter(Statistic.OBJECTS_DELETED);
+      incrementCounter(Statistic.OBJECTS_DELETED, 1);
       bucket.deleteKey(keyName);
       return true;
     } catch (IOException ioe) {
@@ -281,9 +286,8 @@ public boolean deleteObject(String keyName) {
    */
   @Override
   public boolean deleteObjects(List<String> keyNameList) {
-    LOG.trace("issuing delete for key {}", keyNameList);
     try {
-      incrementCounter(Statistic.OBJECTS_DELETED);
+      incrementCounter(Statistic.OBJECTS_DELETED, keyNameList.size());
       bucket.deleteKeys(keyNameList);
       return true;
     } catch (IOException ioe) {
@@ -296,7 +300,7 @@ public FileStatusAdapter getFileStatus(String key, URI uri,
       Path qualifiedPath, String userName)
       throws IOException {
     try {
-      incrementCounter(Statistic.OBJECTS_QUERY);
+      incrementCounter(Statistic.OBJECTS_QUERY, 1);
       OzoneFileStatus status = bucket.getFileStatus(key);
       return toFileStatusAdapter(status, userName, uri, qualifiedPath);
 
@@ -312,15 +316,15 @@ public FileStatusAdapter getFileStatus(String key, URI uri,
 
   @Override
   public Iterator<BasicKeyInfo> listKeys(String pathKey) {
-    incrementCounter(Statistic.OBJECTS_LIST);
+    incrementCounter(Statistic.OBJECTS_LIST, 1);
     return new IteratorAdapter(bucket.listKeys(pathKey));
   }
 
   public List<FileStatusAdapter> listStatus(String keyName, boolean recursive,
       String startKey, long numEntries, URI uri,
       Path workingDir, String username) throws IOException {
     try {
-      incrementCounter(Statistic.OBJECTS_LIST);
+      incrementCounter(Statistic.OBJECTS_LIST, 1);
       List<OzoneFileStatus> statuses = bucket
           .listStatus(keyName, recursive, startKey, numEntries);
 

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java

@@ -208,7 +208,7 @@ public String getScheme() {
 
   @Override
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
-    incrementCounter(Statistic.INVOCATION_OPEN);
+    incrementCounter(Statistic.INVOCATION_OPEN, 1);
     statistics.incrementReadOps(1);
     LOG.trace("open() path:{}", f);
     final String key = pathToKey(f);
@@ -220,17 +220,22 @@ protected InputStream createFSInputStream(InputStream inputStream) {
     return new OzoneFSInputStream(inputStream, statistics);
   }
 
+  @Deprecated
   protected void incrementCounter(Statistic statistic) {
     //don't do anyting in this default implementation.
   }
 
+  protected void incrementCounter(Statistic statistic, long count) {
+    //don't do anyting in this default implementation.
+  }
+
   @Override
   public FSDataOutputStream create(Path f, FsPermission permission,
       boolean overwrite, int bufferSize,
       short replication, long blockSize,
       Progressable progress) throws IOException {
     LOG.trace("create() path:{}", f);
-    incrementCounter(Statistic.INVOCATION_CREATE);
+    incrementCounter(Statistic.INVOCATION_CREATE, 1);
     statistics.incrementWriteOps(1);
     final String key = pathToKey(f);
     return createOutputStream(key, replication, overwrite, true);
@@ -244,7 +249,7 @@ public FSDataOutputStream createNonRecursive(Path path,
       short replication,
       long blockSize,
       Progressable progress) throws IOException {
-    incrementCounter(Statistic.INVOCATION_CREATE_NON_RECURSIVE);
+    incrementCounter(Statistic.INVOCATION_CREATE_NON_RECURSIVE, 1);
     statistics.incrementWriteOps(1);
     final String key = pathToKey(path);
     return createOutputStream(key,
@@ -302,7 +307,7 @@ boolean processKey(List<String> keyList) throws IOException {
    */
   @Override
   public boolean rename(Path src, Path dst) throws IOException {
-    incrementCounter(Statistic.INVOCATION_RENAME);
+    incrementCounter(Statistic.INVOCATION_RENAME, 1);
     statistics.incrementWriteOps(1);
     super.checkPath(src);
     super.checkPath(dst);
@@ -479,16 +484,9 @@ private boolean innerDelete(Path f, boolean recursive) throws IOException {
     }
   }
 
-  /**
-   * {@inheritDoc}
-   *
-   * OFS supports volume and bucket deletion, recursive or non-recursive.
-   * e.g. delete(new Path("/volume1"), true)
-   * But root deletion is explicitly disallowed for safety concerns.
-   */
   @Override
   public boolean delete(Path f, boolean recursive) throws IOException {
-    incrementCounter(Statistic.INVOCATION_DELETE);
+    incrementCounter(Statistic.INVOCATION_DELETE, 1);
     statistics.incrementWriteOps(1);
     LOG.debug("Delete path {} - recursive {}", f, recursive);
     FileStatus status;
@@ -568,7 +566,7 @@ private boolean o3Exists(final Path f) throws IOException {
 
   @Override
   public FileStatus[] listStatus(Path f) throws IOException {
-    incrementCounter(Statistic.INVOCATION_LIST_STATUS);
+    incrementCounter(Statistic.INVOCATION_LIST_STATUS, 1);
     statistics.incrementReadOps(1);
     LOG.trace("listStatus() path:{}", f);
     int numEntries = LISTING_PAGE_SIZE;
@@ -705,7 +703,7 @@ public boolean mkdirs(Path f, FsPermission permission) throws IOException {
 
   @Override
   public FileStatus getFileStatus(Path f) throws IOException {
-    incrementCounter(Statistic.INVOCATION_GET_FILE_STATUS);
+    incrementCounter(Statistic.INVOCATION_GET_FILE_STATUS, 1);
     statistics.incrementReadOps(1);
     LOG.trace("getFileStatus() path:{}", f);
     Path qualifiedPath = f.makeQualified(uri, workingDir);

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java

@@ -27,6 +27,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import com.google.common.base.Preconditions;
 import org.apache.commons.collections.CollectionUtils;
@@ -284,7 +285,7 @@ public void close() throws IOException {
 
   @Override
   public InputStream readFile(String pathStr) throws IOException {
-    incrementCounter(Statistic.OBJECTS_READ);
+    incrementCounter(Statistic.OBJECTS_READ, 1);
     OFSPath ofsPath = new OFSPath(pathStr);
     String key = ofsPath.getKeyName();
     try {
@@ -301,14 +302,19 @@ public InputStream readFile(String pathStr) throws IOException {
     }
   }
 
+  @Deprecated
   protected void incrementCounter(Statistic objectsRead) {
     //noop: Use OzoneClientAdapterImpl which supports statistics.
   }
 
+  protected void incrementCounter(Statistic objectsRead, long count) {
+    //noop: Use OzoneClientAdapterImpl which supports statistics.
+  }
+
   @Override
   public OzoneFSOutputStream createFile(String pathStr, short replication,
       boolean overWrite, boolean recursive) throws IOException {
-    incrementCounter(Statistic.OBJECTS_CREATED);
+    incrementCounter(Statistic.OBJECTS_CREATED, 1);
     OFSPath ofsPath = new OFSPath(pathStr);
     if (ofsPath.isRoot() || ofsPath.isVolume() || ofsPath.isBucket()) {
       throw new IOException("Cannot create file under root or volume.");
@@ -358,7 +364,7 @@ public void renameKey(String key, String newKeyName) throws IOException {
    */
   @Override
   public void rename(String path, String newPath) throws IOException {
-    incrementCounter(Statistic.OBJECTS_RENAMED);
+    incrementCounter(Statistic.OBJECTS_RENAMED, 1);
     OFSPath ofsPath = new OFSPath(path);
     OFSPath ofsNewPath = new OFSPath(newPath);
 
@@ -384,7 +390,7 @@ public void rename(String path, String newPath) throws IOException {
    */
   void rename(OzoneBucket bucket, String path, String newPath)
       throws IOException {
-    incrementCounter(Statistic.OBJECTS_RENAMED);
+    incrementCounter(Statistic.OBJECTS_RENAMED, 1);
     OFSPath ofsPath = new OFSPath(path);
     OFSPath ofsNewPath = new OFSPath(newPath);
     // No same-bucket policy check here since this call path is controlled
@@ -402,7 +408,7 @@ void rename(OzoneBucket bucket, String path, String newPath)
   @Override
   public boolean createDirectory(String pathStr) throws IOException {
     LOG.trace("creating dir for path: {}", pathStr);
-    incrementCounter(Statistic.OBJECTS_CREATED);
+    incrementCounter(Statistic.OBJECTS_CREATED, 1);
     OFSPath ofsPath = new OFSPath(pathStr);
     if (ofsPath.getVolumeName().isEmpty()) {
       // Volume name unspecified, invalid param, return failure
@@ -441,7 +447,7 @@ public boolean createDirectory(String pathStr) throws IOException {
   @Override
   public boolean deleteObject(String path) {
     LOG.trace("issuing delete for path to key: {}", path);
-    incrementCounter(Statistic.OBJECTS_DELETED);
+    incrementCounter(Statistic.OBJECTS_DELETED, 1);
     OFSPath ofsPath = new OFSPath(path);
     String keyName = ofsPath.getKeyName();
     if (keyName.length() == 0) {
@@ -457,45 +463,83 @@ public boolean deleteObject(String path) {
     }
   }
 
+  /**
+   * Helper function to check if the list of key paths are in the same volume
+   * and same bucket.
+   */
+  private boolean areInSameBucket(List<String> keyNameList) {
+    if (keyNameList.size() == 0) {
+      return true;
+    }
+    String firstKeyPath = keyNameList.get(0);
+    final String volAndBucket = new OFSPath(firstKeyPath).getNonKeyPath();
+    // If any key path's volume and bucket from the second element and on
+    // in the list doesn't match the first element's, hasDifferentVolAndBucket
+    // would be true
+    boolean hasDifferentVolAndBucket = keyNameList.stream().skip(1)
+        .anyMatch(p -> !(new OFSPath(p).getNonKeyPath().equals(volAndBucket)));
+    return !hasDifferentVolAndBucket;
+  }
+
   /**
    * Helper method to delete an object specified by key name in bucket.
    *
-   * @param pathList key name list to be deleted
-   * @return true if the key is deleted, false otherwise
+   * Only supports deleting keys in the same bucket in one call.
+   *
+   * Each item in the given list should be the String of an OFS path:
+   * e.g. ofs://om/vol1/buck1/k1
+   *
+   * @param keyNameList key name list to be deleted
+   * @return true if the key deletion is successful, false otherwise
    */
   @Override
-  public boolean deleteObjects(List<String> pathList) {
-    // TODO: we will support deleteObjects in ofs.
-    LOG.error("ofs currently does not support deleteObjects");
-    return false;
+  public boolean deleteObjects(List<String> keyNameList) {
+    if (keyNameList.size() == 0) {
+      return true;
+    }
+    // Sanity check. Support only deleting a list of keys in the same bucket
+    if (!areInSameBucket(keyNameList)) {
+      LOG.error("Deleting keys from different buckets in a single batch "
+          + "is not supported.");
+      return false;
+    }
+    try {
+      OFSPath firstKeyPath = new OFSPath(keyNameList.get(0));
+      OzoneBucket bucket = getBucket(firstKeyPath, false);
+      return deleteObjects(bucket, keyNameList);
+    } catch (IOException ioe) {
+      LOG.error("delete key failed: {}", ioe.getMessage());
+      return false;
+    }
   }
 
   /**
    * Package-private helper function to reduce calls to getBucket().
+   *
+   * This will be faster than the public variant of the method since this
+   * doesn't verify the same-bucket condition.
+   *
    * @param bucket Bucket to operate in.
-   * @param path Path to delete.
-   * @return true if operation succeeded, false upon IOException.
+   * @param keyNameList key name list to be deleted.
+   * @return true if operation succeeded, false on IOException.
    */
-  boolean deleteObject(OzoneBucket bucket, String path) {
-    LOG.trace("issuing delete for path to key: {}", path);
-    incrementCounter(Statistic.OBJECTS_DELETED);
-    OFSPath ofsPath = new OFSPath(path);
-    String keyName = ofsPath.getKeyName();
-    if (keyName.length() == 0) {
-      return false;
-    }
+  boolean deleteObjects(OzoneBucket bucket, List<String> keyNameList) {
+    List<String> keyList = keyNameList.stream()
+        .map(p -> new OFSPath(p).getKeyName())
+        .collect(Collectors.toList());
     try {
-      bucket.deleteKey(keyName);
+      incrementCounter(Statistic.OBJECTS_DELETED, keyNameList.size());
+      bucket.deleteKeys(keyList);
       return true;
     } catch (IOException ioe) {
-      LOG.error("delete key failed " + ioe.getMessage());
+      LOG.error("delete key failed: {}", ioe.getMessage());
       return false;
     }
   }
 
   public FileStatusAdapter getFileStatus(String path, URI uri,
       Path qualifiedPath, String userName) throws IOException {
-    incrementCounter(Statistic.OBJECTS_QUERY);
+    incrementCounter(Statistic.OBJECTS_QUERY, 1);
     OFSPath ofsPath = new OFSPath(path);
     String key = ofsPath.getKeyName();
     if (ofsPath.isRoot()) {
@@ -579,7 +623,7 @@ public Collection<FileStatus> getTrashRoots(boolean allUsers,
 
   @Override
   public Iterator<BasicKeyInfo> listKeys(String pathStr) {
-    incrementCounter(Statistic.OBJECTS_LIST);
+    incrementCounter(Statistic.OBJECTS_LIST, 1);
     OFSPath ofsPath = new OFSPath(pathStr);
     String key = ofsPath.getKeyName();
     OzoneBucket bucket;
@@ -668,7 +712,7 @@ public List<FileStatusAdapter> listStatus(String pathStr, boolean recursive,
       String startPath, long numEntries, URI uri,
       Path workingDir, String username) throws IOException {
 
-    incrementCounter(Statistic.OBJECTS_LIST);
+    incrementCounter(Statistic.OBJECTS_LIST, 1);
     // Remove authority from startPath if it exists
     if (startPath.startsWith(uri.toString())) {
       try {