You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by cn...@apache.org on 2016/08/17 21:58:54 UTC
[2/2] hadoop git commit: HADOOP-13208. S3A listFiles(recursive=true)
to do a bulk listObjects instead of walking the pseudo-tree of directories.
Contributed by Steve Loughran.
HADOOP-13208. S3A listFiles(recursive=true) to do a bulk listObjects instead of walking the pseudo-tree of directories. Contributed by Steve Loughran.
(cherry picked from commit 822d661b8fcc42bec6eea958d9fd02ef1aaa4b6c)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/986162f9
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/986162f9
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/986162f9
Branch: refs/heads/branch-2
Commit: 986162f9776837b1343073d721f89be8f3362085
Parents: 2486c4c
Author: Chris Nauroth <cn...@apache.org>
Authored: Wed Aug 17 14:54:54 2016 -0700
Committer: Chris Nauroth <cn...@apache.org>
Committed: Wed Aug 17 14:55:07 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/hadoop/fs/s3a/Listing.java | 594 +++++++++++++++++++
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 372 +++++++-----
.../hadoop/fs/s3a/S3AInstrumentation.java | 1 +
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 12 +
.../org/apache/hadoop/fs/s3a/Statistic.java | 2 +
.../s3a/TestS3AContractGetFileStatus.java | 16 +
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 8 +-
.../s3a/scale/TestS3ADirectoryPerformance.java | 58 +-
8 files changed, 897 insertions(+), 166 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
new file mode 100644
index 0000000..4120b20
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
@@ -0,0 +1,594 @@
+/*
+ * 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.s3a;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.NoSuchElementException;
+
+import static org.apache.hadoop.fs.s3a.Constants.S3N_FOLDER_SUFFIX;
+import static org.apache.hadoop.fs.s3a.S3AUtils.createFileStatus;
+import static org.apache.hadoop.fs.s3a.S3AUtils.objectRepresentsDirectory;
+import static org.apache.hadoop.fs.s3a.S3AUtils.stringify;
+import static org.apache.hadoop.fs.s3a.S3AUtils.translateException;
+
+/**
+ * Place for the S3A listing classes; keeps all the small classes under control.
+ */
+public class Listing {
+
+ private final S3AFileSystem owner;
+ private static final Logger LOG = S3AFileSystem.LOG;
+
+ public Listing(S3AFileSystem owner) {
+ this.owner = owner;
+ }
+
+ /**
+ * Create a FileStatus iterator against a path, with a given
+ * list object request.
+ * @param listPath path of the listing
+ * @param request initial request to make
+ * @param filter the filter on which paths to accept
+ * @param acceptor the class/predicate to decide which entries to accept
+ * in the listing based on the full file status.
+ * @return the iterator
+ * @throws IOException IO Problems
+ */
+ FileStatusListingIterator createFileStatusListingIterator(
+ Path listPath,
+ ListObjectsRequest request,
+ PathFilter filter,
+ Listing.FileStatusAcceptor acceptor) throws IOException {
+ return new FileStatusListingIterator(
+ new ObjectListingIterator(listPath, request),
+ filter,
+ acceptor);
+ }
+
+ /**
+ * Create a located status iterator over a file status iterator.
+ * @param statusIterator an iterator over the remote status entries
+ * @return a new remote iterator
+ */
+ LocatedFileStatusIterator createLocatedFileStatusIterator(
+ RemoteIterator<FileStatus> statusIterator) {
+ return new LocatedFileStatusIterator(statusIterator);
+ }
+
+ /**
+ * Interface to implement by the logic deciding whether to accept a summary
+ * entry or path as a valid file or directory.
+ */
+ interface FileStatusAcceptor {
+
+ /**
+ * Predicate to decide whether or not to accept a summary entry.
+ * @param keyPath qualified path to the entry
+ * @param summary summary entry
+ * @return true if the entry is accepted (i.e. that a status entry
+ * should be generated.
+ */
+ boolean accept(Path keyPath, S3ObjectSummary summary);
+
+ /**
+ * Predicate to decide whether or not to accept a prefix.
+ * @param keyPath qualified path to the entry
+ * @param commonPrefix the prefix
+ * @return true if the entry is accepted (i.e. that a status entry
+ * should be generated.)
+ */
+ boolean accept(Path keyPath, String commonPrefix);
+ }
+
+ /**
+ * A remote iterator which only iterates over a single `LocatedFileStatus`
+ * value.
+ *
+ * If the status value is null, the iterator declares that it has no data.
+ * This iterator is used to handle {@link listStatus()} calls where the path
+ * handed in refers to a file, not a directory: this is the iterator
+ * returned.
+ */
+ static final class SingleStatusRemoteIterator
+ implements RemoteIterator<LocatedFileStatus> {
+
+ /**
+ * The status to return; set to null after the first iteration.
+ */
+ private LocatedFileStatus status;
+
+ /**
+ * Constructor.
+ * @param status status value: may be null, in which case
+ * the iterator is empty.
+ */
+ public SingleStatusRemoteIterator(LocatedFileStatus status) {
+ this.status = status;
+ }
+
+ /**
+ * {@inheritDoc}
+ * @return true if there is a file status to return: this is always false
+ * for the second iteration, and may be false for the first.
+ * @throws IOException never
+ */
+ @Override
+ public boolean hasNext() throws IOException {
+ return status != null;
+ }
+
+ /**
+ * {@inheritDoc}
+ * @return the non-null status element passed in when the instance was
+ * constructed, if it ha not already been retrieved.
+ * @throws IOException never
+ * @throws NoSuchElementException if this is the second call, or it is
+ * the first call and a null {@link LocatedFileStatus} entry was passed
+ * to the constructor.
+ */
+ @Override
+ public LocatedFileStatus next() throws IOException {
+ if (hasNext()) {
+ LocatedFileStatus s = this.status;
+ status = null;
+ return s;
+ } else {
+ throw new NoSuchElementException();
+ }
+ }
+ }
+
+ /**
+ * Wraps up object listing into a remote iterator which will ask for more
+ * listing data if needed.
+ *
+ * This is a complex operation, especially the process to determine
+ * if there are more entries remaining. If there are no more results
+ * remaining in the (filtered) results of the current listing request, then
+ * another request is made <i>and those results filtered</i> before the
+ * iterator can declare that there is more data available.
+ *
+ * The need to filter the results precludes the iterator from simply
+ * declaring that if the {@link S3AFileSystem.ObjectListingIterator#hasNext()}
+ * is true then there are more results. Instead the next batch of results must
+ * be retrieved and filtered.
+ *
+ * What does this mean? It means that remote requests to retrieve new
+ * batches of object listings are made in the {@link #hasNext()} call;
+ * the {@link #next()} call simply returns the filtered results of the last
+ * listing processed. However, do note that {@link #next()} calls
+ * {@link #hasNext()} during its operation. This is critical to ensure
+ * that a listing obtained through a sequence of {@link #next()} will
+ * complete with the same set of results as a classic
+ * {@code while(it.hasNext()} loop.
+ *
+ * Thread safety: None.
+ */
+ class FileStatusListingIterator
+ implements RemoteIterator<FileStatus> {
+
+ /** Source of objects. */
+ private final ObjectListingIterator source;
+ /** Filter of paths from API call. */
+ private final PathFilter filter;
+ /** Filter of entries from file status. */
+ private final FileStatusAcceptor acceptor;
+ /** request batch size. */
+ private int batchSize;
+ /** Iterator over the current set of results. */
+ private ListIterator<FileStatus> statusBatchIterator;
+
+ /**
+ * Create an iterator over file status entries.
+ * @param source the listing iterator from a listObjects call.
+ * @param filter the filter on which paths to accept
+ * @param acceptor the class/predicate to decide which entries to accept
+ * in the listing based on the full file status.
+ * @throws IOException IO Problems
+ */
+ FileStatusListingIterator(ObjectListingIterator source,
+ PathFilter filter,
+ FileStatusAcceptor acceptor) throws IOException {
+ this.source = source;
+ this.filter = filter;
+ this.acceptor = acceptor;
+ // build the first set of results. This will not trigger any
+ // remote IO, assuming the source iterator is in its initial
+ // iteration
+ requestNextBatch();
+ }
+
+ /**
+ * Report whether or not there is new data available.
+ * If there is data in the local filtered list, return true.
+ * Else: request more data util that condition is met, or there
+ * is no more remote listing data.
+ * @return true if a call to {@link #next()} will succeed.
+ * @throws IOException
+ */
+ @Override
+ public boolean hasNext() throws IOException {
+ return statusBatchIterator.hasNext() || requestNextBatch();
+ }
+
+ @Override
+ public FileStatus next() throws IOException {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return statusBatchIterator.next();
+ }
+
+ /**
+ * Try to retrieve another batch.
+ * Note that for the initial batch,
+ * {@link S3AFileSystem.ObjectListingIterator} does not generate a request;
+ * it simply returns the initial set.
+ *
+ * @return true if a new batch was created.
+ * @throws IOException IO problems
+ */
+ private boolean requestNextBatch() throws IOException {
+ // look for more object listing batches being available
+ while (source.hasNext()) {
+ // if available, retrieve it and build the next status
+ if (buildNextStatusBatch(source.next())) {
+ // this batch successfully generated entries matching the filters/
+ // acceptors; declare that the request was successful
+ return true;
+ } else {
+ LOG.debug("All entries in batch were filtered...continuing");
+ }
+ }
+ // if this code is reached, it means that all remaining
+ // object lists have been retrieved, and there are no new entries
+ // to return.
+ return false;
+ }
+
+ /**
+ * Build the next status batch from a listing.
+ * @param objects the next object listing
+ * @return true if this added any entries after filtering
+ */
+ private boolean buildNextStatusBatch(ObjectListing objects) {
+ // counters for debug logs
+ int added = 0, ignored = 0;
+ // list to fill in with results. Initial size will be list maximum.
+ List<FileStatus> stats = new ArrayList<>(
+ objects.getObjectSummaries().size() +
+ objects.getCommonPrefixes().size());
+ // objects
+ for (S3ObjectSummary summary : objects.getObjectSummaries()) {
+ String key = summary.getKey();
+ Path keyPath = owner.keyToQualifiedPath(key);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{}: {}", keyPath, stringify(summary));
+ }
+ // Skip over keys that are ourselves and old S3N _$folder$ files
+ if (acceptor.accept(keyPath, summary) && filter.accept(keyPath)) {
+ FileStatus status = createFileStatus(keyPath, summary,
+ owner.getDefaultBlockSize(keyPath));
+ LOG.debug("Adding: {}", status);
+ stats.add(status);
+ added++;
+ } else {
+ LOG.debug("Ignoring: {}", keyPath);
+ ignored++;
+ }
+ }
+
+ // prefixes: always directories
+ for (String prefix : objects.getCommonPrefixes()) {
+ Path keyPath = owner.keyToQualifiedPath(prefix);
+ if (acceptor.accept(keyPath, prefix) && filter.accept(keyPath)) {
+ FileStatus status = new S3AFileStatus(true, false, keyPath);
+ LOG.debug("Adding directory: {}", status);
+ added++;
+ stats.add(status);
+ } else {
+ LOG.debug("Ignoring directory: {}", keyPath);
+ ignored++;
+ }
+ }
+
+ // finish up
+ batchSize = stats.size();
+ statusBatchIterator = stats.listIterator();
+ boolean hasNext = statusBatchIterator.hasNext();
+ LOG.debug("Added {} entries; ignored {}; hasNext={}; hasMoreObjects={}",
+ added, ignored, hasNext, objects.isTruncated());
+ return hasNext;
+ }
+
+ /**
+ * Get the number of entries in the current batch.
+ * @return a number, possibly zero.
+ */
+ public int getBatchSize() {
+ return batchSize;
+ }
+ }
+
+ /**
+ * Wraps up AWS `ListObjects` requests in a remote iterator
+ * which will ask for more listing data if needed.
+ *
+ * That is:
+ *
+ * 1. The first invocation of the {@link #next()} call will return the results
+ * of the first request, the one created during the construction of the
+ * instance.
+ *
+ * 2. Second and later invocations will continue the ongoing listing,
+ * calling {@link #continueListObjects(ObjectListing)} to request the next
+ * batch of results.
+ *
+ * 3. The {@link #hasNext()} predicate returns true for the initial call,
+ * where {@link #next()} will return the initial results. It declares
+ * that it has future results iff the last executed request was truncated.
+ *
+ * Thread safety: none.
+ */
+ class ObjectListingIterator implements RemoteIterator<ObjectListing> {
+
+ /** The path listed. */
+ private final Path listPath;
+
+ /** The most recent listing results. */
+ private ObjectListing objects;
+
+ /** Indicator that this is the first listing. */
+ private boolean firstListing = true;
+
+ /**
+ * Count of how many listings have been requested
+ * (including initial result).
+ */
+ private int listingCount = 1;
+
+ /**
+ * Maximum keys in a request.
+ */
+ private int maxKeys;
+
+ /**
+ * Constructor -calls `listObjects()` on the request to populate the
+ * initial set of results/fail if there was a problem talking to the bucket.
+ * @param listPath path of the listing
+ * @param request initial request to make
+ * */
+ ObjectListingIterator(
+ Path listPath,
+ ListObjectsRequest request) {
+ this.listPath = listPath;
+ this.maxKeys = owner.getMaxKeys();
+ this.objects = owner.listObjects(request);
+ }
+
+ /**
+ * Declare that the iterator has data if it is either is the initial
+ * iteration or it is a later one and the last listing obtained was
+ * incomplete.
+ * @throws IOException never: there is no IO in this operation.
+ */
+ @Override
+ public boolean hasNext() throws IOException {
+ return firstListing || objects.isTruncated();
+ }
+
+ /**
+ * Ask for the next listing.
+ * For the first invocation, this returns the initial set, with no
+ * remote IO. For later requests, S3 will be queried, hence the calls
+ * may block or fail.
+ * @return the next object listing.
+ * @throws IOException if a query made of S3 fails.
+ * @throws NoSuchElementException if there is no more data to list.
+ */
+ @Override
+ public ObjectListing next() throws IOException {
+ if (firstListing) {
+ // on the first listing, don't request more data.
+ // Instead just clear the firstListing flag so that it future calls
+ // will request new data.
+ firstListing = false;
+ } else {
+ try {
+ if (!objects.isTruncated()) {
+ // nothing more to request: fail.
+ throw new NoSuchElementException("No more results in listing of "
+ + listPath);
+ }
+ // need to request a new set of objects.
+ LOG.debug("[{}], Requesting next {} objects under {}",
+ listingCount, maxKeys, listPath);
+ objects = owner.continueListObjects(objects);
+ listingCount++;
+ LOG.debug("New listing status: {}", this);
+ } catch (AmazonClientException e) {
+ throw translateException("listObjects()", listPath, e);
+ }
+ }
+ return objects;
+ }
+
+ @Override
+ public String toString() {
+ return "Object listing iterator against " + listPath
+ + "; listing count "+ listingCount
+ + "; isTruncated=" + objects.isTruncated();
+ }
+
+ /**
+ * Get the path listed.
+ * @return the path used in this listing.
+ */
+ public Path getListPath() {
+ return listPath;
+ }
+
+ /**
+ * Get the count of listing requests.
+ * @return the counter of requests made (including the initial lookup).
+ */
+ public int getListingCount() {
+ return listingCount;
+ }
+ }
+
+ /**
+ * Accept all entries except the base path and those which map to S3N
+ * pseudo directory markers.
+ */
+ static class AcceptFilesOnly implements FileStatusAcceptor {
+ private final Path qualifiedPath;
+
+ public AcceptFilesOnly(Path qualifiedPath) {
+ this.qualifiedPath = qualifiedPath;
+ }
+
+ /**
+ * Reject a summary entry if the key path is the qualified Path, or
+ * it ends with {@code "_$folder$"}.
+ * @param keyPath key path of the entry
+ * @param summary summary entry
+ * @return true if the entry is accepted (i.e. that a status entry
+ * should be generated.
+ */
+ @Override
+ public boolean accept(Path keyPath, S3ObjectSummary summary) {
+ return !keyPath.equals(qualifiedPath)
+ && !summary.getKey().endsWith(S3N_FOLDER_SUFFIX)
+ && !objectRepresentsDirectory(summary.getKey(), summary.getSize());
+ }
+
+ /**
+ * Accept no directory paths.
+ * @param keyPath qualified path to the entry
+ * @param prefix common prefix in listing.
+ * @return false, always.
+ */
+ @Override
+ public boolean accept(Path keyPath, String prefix) {
+ return false;
+ }
+ }
+
+ /**
+ * Take a remote iterator over a set of {@link FileStatus} instances and
+ * return a remote iterator of {@link LocatedFileStatus} instances.
+ */
+ class LocatedFileStatusIterator
+ implements RemoteIterator<LocatedFileStatus> {
+ private final RemoteIterator<FileStatus> statusIterator;
+
+ /**
+ * Constructor.
+ * @param statusIterator an iterator over the remote status entries
+ */
+ LocatedFileStatusIterator(RemoteIterator<FileStatus> statusIterator) {
+ this.statusIterator = statusIterator;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ return statusIterator.hasNext();
+ }
+
+ @Override
+ public LocatedFileStatus next() throws IOException {
+ return owner.toLocatedFileStatus(statusIterator.next());
+ }
+ }
+
+ /**
+ * Accept all entries except the base path and those which map to S3N
+ * pseudo directory markers.
+ */
+ static class AcceptAllButSelfAndS3nDirs implements FileStatusAcceptor {
+
+ /** Base path. */
+ private final Path qualifiedPath;
+
+ /**
+ * Constructor.
+ * @param qualifiedPath an already-qualified path.
+ */
+ public AcceptAllButSelfAndS3nDirs(Path qualifiedPath) {
+ this.qualifiedPath = qualifiedPath;
+ }
+
+ /**
+ * Reject a summary entry if the key path is the qualified Path, or
+ * it ends with {@code "_$folder$"}.
+ * @param keyPath key path of the entry
+ * @param summary summary entry
+ * @return true if the entry is accepted (i.e. that a status entry
+ * should be generated.)
+ */
+ @Override
+ public boolean accept(Path keyPath, S3ObjectSummary summary) {
+ return !keyPath.equals(qualifiedPath) &&
+ !summary.getKey().endsWith(S3N_FOLDER_SUFFIX);
+ }
+
+ /**
+ * Accept all prefixes except the one for the base path, "self".
+ * @param keyPath qualified path to the entry
+ * @param prefix common prefix in listing.
+ * @return true if the entry is accepted (i.e. that a status entry
+ * should be generated.
+ */
+ @Override
+ public boolean accept(Path keyPath, String prefix) {
+ return !keyPath.equals(qualifiedPath);
+ }
+ }
+
+ /**
+ * A Path filter which accepts all filenames.
+ */
+ static final PathFilter ACCEPT_ALL = new PathFilter() {
+ @Override
+ public boolean accept(Path file) {
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "ACCEPT_ALL";
+ }
+ };
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index c45f4a5..61727c9 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -85,6 +85,7 @@ import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.VersionInfo;
import static org.apache.hadoop.fs.s3a.Constants.*;
+import static org.apache.hadoop.fs.s3a.Listing.ACCEPT_ALL;
import static org.apache.hadoop.fs.s3a.S3AUtils.*;
import static org.apache.hadoop.fs.s3a.Statistic.*;
@@ -116,6 +117,7 @@ public class S3AFileSystem extends FileSystem {
private AmazonS3Client s3;
private String bucket;
private int maxKeys;
+ private Listing listing;
private long partSize;
private boolean enableMultiObjectsDelete;
private TransferManager transfers;
@@ -184,6 +186,7 @@ public class S3AFileSystem extends FileSystem {
initAmazonS3Client(conf, credentials, awsConf);
maxKeys = intOption(conf, MAX_PAGING_KEYS, DEFAULT_MAX_PAGING_KEYS, 1);
+ listing = new Listing(this);
partSize = conf.getLong(MULTIPART_SIZE, DEFAULT_MULTIPART_SIZE);
if (partSize < 5 * 1024 * 1024) {
LOG.error(MULTIPART_SIZE + " must be at least 5 MB");
@@ -524,7 +527,11 @@ public class S3AFileSystem extends FileSystem {
super();
}
- /* Turns a path (relative or otherwise) into an S3 key
+ /**
+ * Turns a path (relative or otherwise) into an S3 key.
+ *
+ * @param path input path, may be relative to the working dir
+ * @return a key excluding the leading "/", or, if it is the root path, ""
*/
private String pathToKey(Path path) {
if (!path.isAbsolute()) {
@@ -538,11 +545,50 @@ public class S3AFileSystem extends FileSystem {
return path.toUri().getPath().substring(1);
}
+ /**
+ * Turns a path (relative or otherwise) into an S3 key, adding a trailing
+ * "/" if the path is not the root <i>and</i> does not already have a "/"
+ * at the end.
+ *
+ * @param key s3 key or ""
+ * @return the with a trailing "/", or, if it is the root key, "",
+ */
+ private String maybeAddTrailingSlash(String key) {
+ if (!key.isEmpty() && !key.endsWith("/")) {
+ return key + '/';
+ } else {
+ return key;
+ }
+ }
+
+ /**
+ * Convert a path back to a key.
+ * @param key input key
+ * @return the path from this key
+ */
private Path keyToPath(String key) {
return new Path("/" + key);
}
/**
+ * Convert a key to a fully qualified path.
+ * @param key input key
+ * @return the fully qualified path including URI scheme and bucket name.
+ */
+ Path keyToQualifiedPath(String key) {
+ return qualify(keyToPath(key));
+ }
+
+ /**
+ * Qualify a path.
+ * @param path path to qualify
+ * @return a qualified path.
+ */
+ Path qualify(Path path) {
+ return path.makeQualified(uri, workingDir);
+ }
+
+ /**
* Check that a Path belongs to this FileSystem.
* Unlike the superclass, this version does not look at authority,
* only hostnames.
@@ -560,10 +606,10 @@ public class S3AFileSystem extends FileSystem {
}
/**
- * Opens an FSDataInputStream at the indicated Path.
- * @param f the file name to open
- * @param bufferSize the size of the buffer to be used.
- */
+ * Opens an FSDataInputStream at the indicated Path.
+ * @param f the file name to open
+ * @param bufferSize the size of the buffer to be used.
+ */
public FSDataInputStream open(Path f, int bufferSize)
throws IOException {
@@ -885,7 +931,7 @@ public class S3AFileSystem extends FileSystem {
* @return the next result object
*/
protected ObjectListing continueListObjects(ObjectListing objects) {
- incrementStatistic(OBJECT_LIST_REQUESTS);
+ incrementStatistic(OBJECT_CONTINUE_LIST_REQUESTS);
incrementReadOperations();
return s3.listNextBatchOfObjects(objects);
}
@@ -1142,12 +1188,7 @@ public class S3AFileSystem extends FileSystem {
} else {
LOG.debug("Getting objects for directory prefix {} to delete", key);
- ListObjectsRequest request = new ListObjectsRequest();
- request.setBucketName(bucket);
- request.setPrefix(key);
- // Hopefully not setting a delimiter will cause this to find everything
- //request.setDelimiter("/");
- request.setMaxKeys(maxKeys);
+ ListObjectsRequest request = createListObjectsRequest(key, null);
ObjectListing objects = listObjects(request);
List<DeleteObjectsRequest.KeyVersion> keys =
@@ -1221,66 +1262,57 @@ public class S3AFileSystem extends FileSystem {
*/
public FileStatus[] innerListStatus(Path f) throws FileNotFoundException,
IOException, AmazonClientException {
- String key = pathToKey(f);
- LOG.debug("List status for path: {}", f);
+ Path path = qualify(f);
+ String key = pathToKey(path);
+ LOG.debug("List status for path: {}", path);
incrementStatistic(INVOCATION_LIST_STATUS);
- final List<FileStatus> result = new ArrayList<FileStatus>();
- final FileStatus fileStatus = getFileStatus(f);
+ List<FileStatus> result;
+ final FileStatus fileStatus = getFileStatus(path);
if (fileStatus.isDirectory()) {
if (!key.isEmpty()) {
- key = key + "/";
+ key = key + '/';
}
- ListObjectsRequest request = new ListObjectsRequest();
- request.setBucketName(bucket);
- request.setPrefix(key);
- request.setDelimiter("/");
- request.setMaxKeys(maxKeys);
-
+ ListObjectsRequest request = createListObjectsRequest(key, "/");
LOG.debug("listStatus: doing listObjects for directory {}", key);
- ObjectListing objects = listObjects(request);
-
- Path fQualified = f.makeQualified(uri, workingDir);
-
- while (true) {
- for (S3ObjectSummary summary : objects.getObjectSummaries()) {
- Path keyPath = keyToPath(summary.getKey()).makeQualified(uri, workingDir);
- // Skip over keys that are ourselves and old S3N _$folder$ files
- if (keyPath.equals(fQualified) ||
- summary.getKey().endsWith(S3N_FOLDER_SUFFIX)) {
- LOG.debug("Ignoring: {}", keyPath);
- } else {
- S3AFileStatus status = createFileStatus(keyPath, summary,
- getDefaultBlockSize(keyPath));
- result.add(status);
- LOG.debug("Adding: {}", status);
- }
- }
-
- for (String prefix : objects.getCommonPrefixes()) {
- Path keyPath = keyToPath(prefix).makeQualified(uri, workingDir);
- if (!keyPath.equals(f)) {
- result.add(new S3AFileStatus(true, false, keyPath));
- LOG.debug("Adding: rd: {}", keyPath);
- }
- }
-
- if (objects.isTruncated()) {
- LOG.debug("listStatus: list truncated - getting next batch");
- objects = continueListObjects(objects);
- } else {
- break;
- }
+ Listing.FileStatusListingIterator files =
+ listing.createFileStatusListingIterator(path,
+ request,
+ ACCEPT_ALL,
+ new Listing.AcceptAllButSelfAndS3nDirs(path));
+ result = new ArrayList<>(files.getBatchSize());
+ while (files.hasNext()) {
+ result.add(files.next());
}
+ return result.toArray(new FileStatus[result.size()]);
} else {
- LOG.debug("Adding: rd (not a dir): {}", f);
- result.add(fileStatus);
+ LOG.debug("Adding: rd (not a dir): {}", path);
+ FileStatus[] stats = new FileStatus[1];
+ stats[0]= fileStatus;
+ return stats;
}
+ }
- return result.toArray(new FileStatus[result.size()]);
+ /**
+ * Create a {@code ListObjectsRequest} request against this bucket,
+ * with the maximum keys returned in a query set by {@link #maxKeys}.
+ * @param key key for request
+ * @param delimiter any delimiter
+ * @return the request
+ */
+ private ListObjectsRequest createListObjectsRequest(String key,
+ String delimiter) {
+ ListObjectsRequest request = new ListObjectsRequest();
+ request.setBucketName(bucket);
+ request.setMaxKeys(maxKeys);
+ request.setPrefix(key);
+ if (delimiter != null) {
+ request.setDelimiter(delimiter);
+ }
+ return request;
}
/**
@@ -1381,11 +1413,11 @@ public class S3AFileSystem extends FileSystem {
* @throws java.io.FileNotFoundException when the path does not exist;
* @throws IOException on other problems.
*/
- public S3AFileStatus getFileStatus(Path f) throws IOException {
- String key = pathToKey(f);
+ public S3AFileStatus getFileStatus(final Path f) throws IOException {
incrementStatistic(INVOCATION_GET_FILE_STATUS);
- LOG.debug("Getting path status for {} ({})", f , key);
-
+ final Path path = qualify(f);
+ String key = pathToKey(path);
+ LOG.debug("Getting path status for {} ({})", path , key);
if (!key.isEmpty()) {
try {
ObjectMetadata meta = getObjectMetadata(key);
@@ -1393,20 +1425,20 @@ public class S3AFileSystem extends FileSystem {
if (objectRepresentsDirectory(key, meta.getContentLength())) {
LOG.debug("Found exact file: fake directory");
return new S3AFileStatus(true, true,
- f.makeQualified(uri, workingDir));
+ path);
} else {
LOG.debug("Found exact file: normal file");
return new S3AFileStatus(meta.getContentLength(),
dateToLong(meta.getLastModified()),
- f.makeQualified(uri, workingDir),
- getDefaultBlockSize(f.makeQualified(uri, workingDir)));
+ path,
+ getDefaultBlockSize(path));
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
- throw translateException("getFileStatus", f, e);
+ throw translateException("getFileStatus", path, e);
}
} catch (AmazonClientException e) {
- throw translateException("getFileStatus", f, e);
+ throw translateException("getFileStatus", path, e);
}
// Necessary?
@@ -1417,14 +1449,14 @@ public class S3AFileSystem extends FileSystem {
if (objectRepresentsDirectory(newKey, meta.getContentLength())) {
LOG.debug("Found file (with /): fake directory");
- return new S3AFileStatus(true, true, f.makeQualified(uri, workingDir));
+ return new S3AFileStatus(true, true, path);
} else {
LOG.warn("Found file (with /): real file? should not happen: {}", key);
return new S3AFileStatus(meta.getContentLength(),
dateToLong(meta.getLastModified()),
- f.makeQualified(uri, workingDir),
- getDefaultBlockSize(f.makeQualified(uri, workingDir)));
+ path,
+ getDefaultBlockSize(path));
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
@@ -1437,9 +1469,7 @@ public class S3AFileSystem extends FileSystem {
}
try {
- if (!key.isEmpty() && !key.endsWith("/")) {
- key = key + "/";
- }
+ key = maybeAddTrailingSlash(key);
ListObjectsRequest request = new ListObjectsRequest();
request.setBucketName(bucket);
request.setPrefix(key);
@@ -1463,11 +1493,10 @@ public class S3AFileSystem extends FileSystem {
}
}
- return new S3AFileStatus(true, false,
- f.makeQualified(uri, workingDir));
+ return new S3AFileStatus(true, false, path);
} else if (key.isEmpty()) {
LOG.debug("Found root directory");
- return new S3AFileStatus(true, true, f.makeQualified(uri, workingDir));
+ return new S3AFileStatus(true, true, path);
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
@@ -1477,8 +1506,8 @@ public class S3AFileSystem extends FileSystem {
throw translateException("getFileStatus", key, e);
}
- LOG.debug("Not Found: {}", f);
- throw new FileNotFoundException("No such file or directory: " + f);
+ LOG.debug("Not Found: {}", path);
+ throw new FileNotFoundException("No such file or directory: " + path);
}
/**
@@ -1839,7 +1868,15 @@ public class S3AFileSystem extends FileSystem {
}
/**
- * Override superclass so as to add statistic collection.
+ * Get the maximum key count.
+ * @return a value, valid after initialization
+ */
+ int getMaxKeys() {
+ return maxKeys;
+ }
+
+ /**
+ * Increments the statistic {@link Statistic#INVOCATION_GLOB_STATUS}.
* {@inheritDoc}
*/
@Override
@@ -1864,24 +1901,6 @@ public class S3AFileSystem extends FileSystem {
* {@inheritDoc}
*/
@Override
- public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path f)
- throws FileNotFoundException, IOException {
- incrementStatistic(INVOCATION_LIST_LOCATED_STATUS);
- return super.listLocatedStatus(f);
- }
-
- @Override
- public RemoteIterator<LocatedFileStatus> listFiles(Path f,
- boolean recursive) throws FileNotFoundException, IOException {
- incrementStatistic(INVOCATION_LIST_FILES);
- return super.listFiles(f, recursive);
- }
-
- /**
- * Override superclass so as to add statistic collection.
- * {@inheritDoc}
- */
- @Override
public boolean exists(Path f) throws IOException {
incrementStatistic(INVOCATION_EXISTS);
return super.exists(f);
@@ -1908,6 +1927,129 @@ public class S3AFileSystem extends FileSystem {
}
/**
+ * {@inheritDoc}.
+ *
+ * This implementation is optimized for S3, which can do a bulk listing
+ * off all entries under a path in one single operation. Thus there is
+ * no need to recursively walk the directory tree.
+ *
+ * Instead a {@link ListObjectsRequest} is created requesting a (windowed)
+ * listing of all entries under the given path. This is used to construct
+ * an {@code ObjectListingIterator} instance, iteratively returning the
+ * sequence of lists of elements under the path. This is then iterated
+ * over in a {@code FileStatusListingIterator}, which generates
+ * {@link S3AFileStatus} instances, one per listing entry.
+ * These are then translated into {@link LocatedFileStatus} instances.
+ *
+ * This is essentially a nested and wrapped set of iterators, with some
+ * generator classes; an architecture which may become less convoluted
+ * using lambda-expressions.
+ * @param f a path
+ * @param recursive if the subdirectories need to be traversed recursively
+ *
+ * @return an iterator that traverses statuses of the files/directories
+ * in the given path
+ * @throws FileNotFoundException if {@code path} does not exist
+ * @throws IOException if any I/O error occurred
+ */
+ @Override
+ public RemoteIterator<LocatedFileStatus> listFiles(Path f,
+ boolean recursive) throws FileNotFoundException, IOException {
+ incrementStatistic(INVOCATION_LIST_FILES);
+ Path path = qualify(f);
+ LOG.debug("listFiles({}, {})", path, recursive);
+ try {
+ // lookup dir triggers existence check
+ final FileStatus fileStatus = getFileStatus(path);
+ if (fileStatus.isFile()) {
+ // simple case: File
+ LOG.debug("Path is a file");
+ return new Listing.SingleStatusRemoteIterator(
+ toLocatedFileStatus(fileStatus));
+ } else {
+ // directory: do a bulk operation
+ String key = maybeAddTrailingSlash(pathToKey(path));
+ String delimiter = recursive ? null : "/";
+ LOG.debug("Requesting all entries under {} with delimiter '{}'",
+ key, delimiter);
+ return listing.createLocatedFileStatusIterator(
+ listing.createFileStatusListingIterator(path,
+ createListObjectsRequest(key, delimiter),
+ ACCEPT_ALL,
+ new Listing.AcceptFilesOnly(path)));
+ }
+ } catch (AmazonClientException e) {
+ throw translateException("listFiles", path, e);
+ }
+ }
+
+ /**
+ * Override superclass so as to add statistic collection.
+ * {@inheritDoc}
+ */
+ @Override
+ public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path f)
+ throws FileNotFoundException, IOException {
+ return listLocatedStatus(f, ACCEPT_ALL);
+ }
+
+ /**
+ * {@inheritDoc}.
+ *
+ * S3 Optimized directory listing. The initial operation performs the
+ * first bulk listing; extra listings will take place
+ * when all the current set of results are used up.
+ * @param f a path
+ * @param filter a path filter
+ * @return an iterator that traverses statuses of the files/directories
+ * in the given path
+ * @throws FileNotFoundException if {@code path} does not exist
+ * @throws IOException if any I/O error occurred
+ */
+ @Override
+ public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f,
+ final PathFilter filter)
+ throws FileNotFoundException, IOException {
+ incrementStatistic(INVOCATION_LIST_LOCATED_STATUS);
+ Path path = qualify(f);
+ LOG.debug("listLocatedStatus({}, {}", path, filter);
+ try {
+ // lookup dir triggers existence check
+ final FileStatus fileStatus = getFileStatus(path);
+ if (fileStatus.isFile()) {
+ // simple case: File
+ LOG.debug("Path is a file");
+ return new Listing.SingleStatusRemoteIterator(
+ filter.accept(path) ? toLocatedFileStatus(fileStatus) : null);
+ } else {
+ // directory: trigger a lookup
+ String key = maybeAddTrailingSlash(pathToKey(path));
+ return listing.createLocatedFileStatusIterator(
+ listing.createFileStatusListingIterator(path,
+ createListObjectsRequest(key, "/"),
+ filter,
+ new Listing.AcceptAllButSelfAndS3nDirs(path)));
+ }
+ } catch (AmazonClientException e) {
+ throw translateException("listLocatedStatus", path, e);
+ }
+ }
+
+ /**
+ * Build a {@link LocatedFileStatus} from a {@link FileStatus} instance.
+ * @param status file status
+ * @return a located status with block locations set up from this FS.
+ * @throws IOException IO Problems.
+ */
+ LocatedFileStatus toLocatedFileStatus(FileStatus status)
+ throws IOException {
+ return new LocatedFileStatus(status,
+ status.isFile() ?
+ getFileBlockLocations(status, 0, status.getLen())
+ : null);
+ }
+
+ /**
* Get a integer option >= the minimum allowed value.
* @param conf configuration
* @param key key to look up
@@ -1944,38 +2086,4 @@ public class S3AFileSystem extends FileSystem {
return v;
}
- /**
- * This is a simple encapsulation of the
- * S3 access key and secret.
- */
- static class AWSAccessKeys {
- private String accessKey = null;
- private String accessSecret = null;
-
- /**
- * Constructor.
- * @param key - AWS access key
- * @param secret - AWS secret key
- */
- public AWSAccessKeys(String key, String secret) {
- accessKey = key;
- accessSecret = secret;
- }
-
- /**
- * Return the AWS access key.
- * @return key
- */
- public String getAccessKey() {
- return accessKey;
- }
-
- /**
- * Return the AWS secret key.
- * @return secret
- */
- public String getAccessSecret() {
- return accessSecret;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java
index 514b974..b4c4063 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java
@@ -95,6 +95,7 @@ public class S3AInstrumentation {
OBJECT_COPY_REQUESTS,
OBJECT_DELETE_REQUESTS,
OBJECT_LIST_REQUESTS,
+ OBJECT_CONTINUE_LIST_REQUESTS,
OBJECT_METADATA_REQUESTS,
OBJECT_MULTIPART_UPLOAD_ABORTED,
OBJECT_PUT_BYTES,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
index 8033ac3..3d48fcd 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
@@ -280,4 +280,16 @@ public final class S3AUtils {
return val;
}
}
+
+ /**
+ * String information about a summary entry for debug messages.
+ * @param summary summary object
+ * @return string value
+ */
+ public static String stringify(S3ObjectSummary summary) {
+ StringBuilder builder = new StringBuilder(summary.getKey().length() + 100);
+ builder.append(summary.getKey()).append(' ');
+ builder.append("size=").append(summary.getSize());
+ return builder.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java
index 36d163c..cbc34d6 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java
@@ -69,6 +69,8 @@ public enum Statistic {
OBJECT_DELETE_REQUESTS("object_delete_requests", "Object delete requests"),
OBJECT_LIST_REQUESTS("object_list_requests",
"Number of object listings made"),
+ OBJECT_CONTINUE_LIST_REQUESTS("object_continue_list_requests",
+ "Number of continued object listings made"),
OBJECT_METADATA_REQUESTS("object_metadata_requests",
"Number of requests for object metadata"),
OBJECT_MULTIPART_UPLOAD_ABORTED("object_multipart_aborted",
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java
index d7b8fe3..5937d49 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java
@@ -20,6 +20,8 @@ package org.apache.hadoop.fs.contract.s3a;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.contract.AbstractFSContract;
import org.apache.hadoop.fs.contract.AbstractContractGetFileStatusTest;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.S3ATestUtils;
public class TestS3AContractGetFileStatus extends AbstractContractGetFileStatusTest {
@@ -28,4 +30,18 @@ public class TestS3AContractGetFileStatus extends AbstractContractGetFileStatusT
return new S3AContract(conf);
}
+ @Override
+ public void teardown() throws Exception {
+ getLog().info("FS details {}", getFileSystem());
+ super.teardown();
+ }
+
+ @Override
+ protected Configuration createConfiguration() {
+ Configuration conf = super.createConfiguration();
+ S3ATestUtils.disableFilesystemCaching(conf);
+ // aggressively low page size forces tests to go multipage
+ conf.setInt(Constants.MAX_PAGING_KEYS, 2);
+ return conf;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
index 39c6028..e45db48 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
@@ -124,7 +124,7 @@ public class S3ATestUtils {
try {
callback.call();
return;
- } catch (FailFastException e) {
+ } catch (InterruptedException | FailFastException e) {
throw e;
} catch (Exception e) {
lastException = e;
@@ -321,7 +321,7 @@ public class S3ATestUtils {
* @return true if the value is {@code ==} the other's
*/
public boolean diffEquals(MetricDiff that) {
- return this.currentValue() == that.currentValue();
+ return this.diff() == that.diff();
}
/**
@@ -330,7 +330,7 @@ public class S3ATestUtils {
* @return true if the value is {@code <} the other's
*/
public boolean diffLessThan(MetricDiff that) {
- return this.currentValue() < that.currentValue();
+ return this.diff() < that.diff();
}
/**
@@ -339,7 +339,7 @@ public class S3ATestUtils {
* @return true if the value is {@code <=} the other's
*/
public boolean diffLessThanOrEquals(MetricDiff that) {
- return this.currentValue() <= that.currentValue();
+ return this.diff() <= that.diff();
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java
index 7ece394..35ea3ad 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.fs.s3a.scale;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.Statistic;
import org.junit.Test;
@@ -51,6 +50,8 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase {
int files = scale;
MetricDiff metadataRequests = new MetricDiff(fs, OBJECT_METADATA_REQUESTS);
MetricDiff listRequests = new MetricDiff(fs, OBJECT_LIST_REQUESTS);
+ MetricDiff listContinueRequests =
+ new MetricDiff(fs, OBJECT_CONTINUE_LIST_REQUESTS);
MetricDiff listStatusCalls = new MetricDiff(fs, INVOCATION_LIST_FILES);
MetricDiff getFileStatusCalls =
new MetricDiff(fs, INVOCATION_GET_FILE_STATUS);
@@ -69,34 +70,29 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase {
printThenReset(LOG,
metadataRequests,
listRequests,
+ listContinueRequests,
listStatusCalls,
getFileStatusCalls);
+ describe("Listing files via treewalk");
try {
// Scan the directory via an explicit tree walk.
// This is the baseline for any listing speedups.
- MetricDiff treewalkMetadataRequests =
- new MetricDiff(fs, OBJECT_METADATA_REQUESTS);
- MetricDiff treewalkListRequests = new MetricDiff(fs,
- OBJECT_LIST_REQUESTS);
- MetricDiff treewalkListStatusCalls = new MetricDiff(fs,
- INVOCATION_LIST_FILES);
- MetricDiff treewalkGetFileStatusCalls =
- new MetricDiff(fs, INVOCATION_GET_FILE_STATUS);
NanoTimer treeWalkTimer = new NanoTimer();
TreeScanResults treewalkResults = treeWalk(fs, listDir);
- treeWalkTimer.end("List status via treewalk");
+ treeWalkTimer.end("List status via treewalk of %s", created);
- print(LOG,
- treewalkMetadataRequests,
- treewalkListRequests,
- treewalkListStatusCalls,
- treewalkGetFileStatusCalls);
+ printThenReset(LOG,
+ metadataRequests,
+ listRequests,
+ listContinueRequests,
+ listStatusCalls,
+ getFileStatusCalls);
assertEquals("Files found in listFiles(recursive=true) " +
" created=" + created + " listed=" + treewalkResults,
created.getFileCount(), treewalkResults.getFileCount());
-
+ describe("Listing files via listFiles(recursive=true)");
// listFiles() does the recursion internally
NanoTimer listFilesRecursiveTimer = new NanoTimer();
@@ -108,31 +104,33 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase {
" created=" + created + " listed=" + listFilesResults,
created.getFileCount(), listFilesResults.getFileCount());
- treewalkListRequests.assertDiffEquals(listRequests);
- printThenReset(LOG,
- metadataRequests, listRequests,
- listStatusCalls, getFileStatusCalls);
-
- NanoTimer globStatusTimer = new NanoTimer();
- FileStatus[] globStatusFiles = fs.globStatus(listDir);
- globStatusTimer.end("Time to globStatus() %s", globStatusTimer);
- LOG.info("Time for glob status {} entries: {}",
- globStatusFiles.length,
- toHuman(createTimer.duration()));
- printThenReset(LOG,
+ // only two list operations should have taken place
+ print(LOG,
metadataRequests,
listRequests,
+ listContinueRequests,
listStatusCalls,
getFileStatusCalls);
+ assertEquals(listRequests.toString(), 2, listRequests.diff());
+ reset(metadataRequests,
+ listRequests,
+ listContinueRequests,
+ listStatusCalls,
+ getFileStatusCalls);
+
} finally {
+ describe("deletion");
// deletion at the end of the run
NanoTimer deleteTimer = new NanoTimer();
fs.delete(listDir, true);
deleteTimer.end("Deleting directory tree");
printThenReset(LOG,
- metadataRequests, listRequests,
- listStatusCalls, getFileStatusCalls);
+ metadataRequests,
+ listRequests,
+ listContinueRequests,
+ listStatusCalls,
+ getFileStatusCalls);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org