[accumulo] branch main updated: Update du command to use hdfs iterator (#1259)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmiller pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git


The following commit(s) were added to refs/heads/main by this push:
     new 23fa7d4  Update du command to use hdfs iterator (#1259)
23fa7d4 is described below

commit 23fa7d48e7908b9af0a761fda431ffa0fc472a12
Author: EdColeman <de...@etcoleman.com>
AuthorDate: Wed Jan 19 15:18:39 2022 -0500

    Update du command to use hdfs iterator (#1259)
    
    * Update TableDiskUsage to use hdfs remote iterator to hopefully
    reduce memory requirements and improve performance.
    * Create new listFiles in VolumeManager to return remote iterator
    * Add comment to new listFiles() about globbing
    * Replace globbing in TableDiskUsage with recursive call to listFiles
    
    Co-authored-by: Mike Miller <mm...@apache.org>
---
 .../java/org/apache/accumulo/server/fs/VolumeManager.java |  6 ++++++
 .../org/apache/accumulo/server/fs/VolumeManagerImpl.java  |  8 ++++++++
 .../org/apache/accumulo/server/util/TableDiskUsage.java   | 15 ++++++++-------
 3 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
index 66b5d52..7fb212b 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
@@ -33,7 +33,9 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -135,6 +137,10 @@ public interface VolumeManager extends AutoCloseable {
   // return the item in options that is in the same file system as source
   Path matchingFileSystem(Path source, Set<String> options);
 
+  // forward to appropriate FileSystem object. Does not support globbing.
+  RemoteIterator<LocatedFileStatus> listFiles(final Path path, final boolean recursive)
+      throws IOException;
+
   // forward to the appropriate FileSystem object
   FileStatus[] listStatus(Path path) throws IOException;
 
diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
index 504e6c4..4e5533e 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
@@ -53,7 +53,9 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.Trash;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
@@ -262,6 +264,12 @@ public class VolumeManagerImpl implements VolumeManager {
   }
 
   @Override
+  public RemoteIterator<LocatedFileStatus> listFiles(final Path path, final boolean recursive)
+      throws IOException {
+    return getFileSystemByPath(path).listFiles(path, recursive);
+  }
+
+  @Override
   public FileStatus[] listStatus(Path path) throws IOException {
     return getFileSystemByPath(path).listStatus(path);
   }
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java b/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java
index d77699b..8b281f6 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java
@@ -50,7 +50,9 @@ import org.apache.accumulo.core.util.NumUtil;
 import org.apache.accumulo.server.cli.ServerUtilOpts;
 import org.apache.accumulo.server.fs.VolumeManager;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -212,13 +214,12 @@ public class TableDiskUsage {
     for (TableId tableId : tablesReferenced) {
       for (String tableDir : nameSpacesReferenced) {
         // Find each file and add its size
-        FileStatus[] files = fs.globStatus(new Path(tableDir + "/" + tableId + "/*/*"));
-        if (files != null) {
-          for (FileStatus fileStatus : files) {
-            // Assumes that all filenames are unique
-            String name = fileStatus.getPath().getName();
-            tdu.addFileSize(name, fileStatus.getLen());
-          }
+        Path path = new Path(tableDir + "/" + tableId);
+        RemoteIterator<LocatedFileStatus> ri = fs.listFiles(path, true);
+        while (ri.hasNext()) {
+          FileStatus status = ri.next();
+          String name = status.getPath().getName();
+          tdu.addFileSize(name, status.getLen());
         }
       }
     }