HADOOP-17428. ABFS: Implementation for getContentSummary

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java

@@ -37,6 +37,9 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
+import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.ContentSummary.Builder;
+import org.apache.hadoop.fs.azurebfs.services.ContentSummaryProcessor;
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.slf4j.Logger;
@@ -363,6 +366,16 @@ public boolean delete(final Path f, final boolean recursive) throws IOException
 
   }
 
+  @Override
+  public ContentSummary getContentSummary(Path f) throws IOException {
+    org.apache.hadoop.fs.azurebfs.utils.ContentSummary contentSummary =
+        (new ContentSummaryProcessor(abfsStore)).getContentSummary(f);
+    return new Builder().length(contentSummary.getLength())
+        .directoryCount(contentSummary.getDirectoryCount())
+        .fileCount(contentSummary.getFileCount())
+        .spaceConsumed(contentSummary.getSpaceConsumed()).build();
+  }
+
   @Override
   public FileStatus[] listStatus(final Path f) throws IOException {
     LOG.debug(

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ContentSummaryProcessor.java

@@ -0,0 +1,102 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.azurebfs.services;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore;
+import org.apache.hadoop.fs.azurebfs.utils.ContentSummary;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class ContentSummaryProcessor {
+  private final AtomicLong fileCount = new AtomicLong(0L);
+  private final AtomicLong directoryCount = new AtomicLong(0L);
+  private final AtomicLong totalBytes = new AtomicLong(0L);
+  private final ProcessingQueue<FileStatus> queue = new ProcessingQueue<>();
+  private final AzureBlobFileSystemStore abfsStore;
+  private static final int NUM_THREADS = 16;
+
+  public ContentSummaryProcessor(AzureBlobFileSystemStore abfsStore) {
+    this.abfsStore = abfsStore;
+  }
+
+  public ContentSummary getContentSummary(Path path) throws IOException {
+    processDirectoryTree(path);
+    Thread[] threads = new Thread[16];
+
+    for (int i = 0; i < NUM_THREADS; ++i) {
+      threads[i] = new Thread(new ContentSummaryProcessor.ThreadProcessor());
+      threads[i].start();
+    }
+
+    for (Thread t : threads) {
+      try {
+        t.join();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
+    }
+    return new ContentSummary(totalBytes.get(), directoryCount.get(),
+        fileCount.get(), totalBytes.get());
+  }
+
+  private void processDirectoryTree(Path path) throws IOException {
+    FileStatus[] fileStatuses = abfsStore.listStatus(path);
+    for (FileStatus fileStatus : fileStatuses) {
+      if (fileStatus.isDirectory()) {
+        this.processDirectory();
+        this.queue.add(fileStatus);
+      } else {
+        this.processFile(fileStatus);
+      }
+    }
+  }
+
+  private void processDirectory() {
+    this.directoryCount.incrementAndGet();
+  }
+
+  private void processFile(FileStatus fileStatus) {
+    this.fileCount.incrementAndGet();
+    this.totalBytes.addAndGet(fileStatus.getLen());
+  }
+
+  private final class ThreadProcessor implements Runnable {
+    private ThreadProcessor() {
+    }
+
+    public void run() {
+      try {
+        FileStatus fileStatus;
+        while ((fileStatus = ContentSummaryProcessor.this.queue.poll())
+            != null) {
+          if (fileStatus.isDirectory()) {
+            ContentSummaryProcessor.this
+                .processDirectoryTree(fileStatus.getPath());
+          }
+          ContentSummaryProcessor.this.queue.unregister();
+        }
+      } catch (IOException e) {
+        throw new RuntimeException("IOException processing Directory tree", e);
+      }
+    }
+  }
+}

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ProcessingQueue.java

@@ -0,0 +1,80 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.azurebfs.services;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+public class ProcessingQueue<T> {
+
+  private final Queue<T> internalQueue = new LinkedList<>();
+  private int processorCount = 0;
+
+  ProcessingQueue() {
+  }
+
+  public synchronized void add(T item) {
+    if (item == null) {
+      throw new IllegalArgumentException("Cannot put null into queue");
+    } else {
+      this.internalQueue.add(item);
+      this.notifyAll();
+    }
+  }
+
+  public synchronized T poll() {
+    while (true) {
+      try {
+        if (this.isQueueEmpty() && !this.done()) {
+          this.wait();
+          continue;
+        }
+        if (!this.isQueueEmpty()) {
+          ++this.processorCount;
+          return this.internalQueue.poll();
+        }
+        return null;
+      } catch (InterruptedException var2) {
+        Thread.currentThread().interrupt();
+      }
+      return null;
+    }
+  }
+
+  public synchronized void unregister() {
+    --this.processorCount;
+    if (this.processorCount < 0) {
+      throw new IllegalStateException(
+          "too many unregister()'s. processorCount is now "
+              + this.processorCount);
+    } else {
+      if (this.done()) {
+        this.notifyAll();
+      }
+    }
+  }
+
+  private boolean done() {
+    return this.processorCount == 0 && this.isQueueEmpty();
+  }
+
+  private boolean isQueueEmpty() {
+    return this.internalQueue.peek() == null;
+  }
+}

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/ContentSummary.java

@@ -0,0 +1,49 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.azurebfs.utils;
+
+public class ContentSummary {
+  private final long length;
+  private final long directoryCount;
+  private final long fileCount;
+  private final long spaceConsumed;
+
+  public ContentSummary(long length, long directoryCount, long fileCount,
+      long spaceConsumed) {
+    this.length = length;
+    this.directoryCount = directoryCount;
+    this.fileCount = fileCount;
+    this.spaceConsumed = spaceConsumed;
+  }
+
+  public long getLength() {
+    return length;
+  }
+
+  public long getDirectoryCount() {
+    return directoryCount;
+  }
+
+  public long getFileCount() {
+    return fileCount;
+  }
+  public long getSpaceConsumed() {
+    return spaceConsumed;
+  }
+}