| /* |
| * 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 javax.annotation.Nullable; |
| |
| import com.amazonaws.AmazonClientException; |
| import com.amazonaws.services.s3.model.S3ObjectSummary; |
| import com.google.common.annotations.VisibleForTesting; |
| |
| import org.apache.commons.lang3.tuple.Triple; |
| import org.apache.hadoop.classification.InterfaceAudience; |
| 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.apache.hadoop.fs.s3a.impl.AbstractStoreOperation; |
| import org.apache.hadoop.fs.s3a.impl.ListingOperationCallbacks; |
| import org.apache.hadoop.fs.s3a.impl.StoreContext; |
| import org.apache.hadoop.fs.s3a.s3guard.DirListingMetadata; |
| import org.apache.hadoop.fs.s3a.s3guard.MetadataStoreListFilesIterator; |
| import org.apache.hadoop.fs.s3a.s3guard.PathMetadata; |
| import org.apache.hadoop.fs.s3a.s3guard.S3Guard; |
| |
| import com.google.common.base.Preconditions; |
| import org.slf4j.Logger; |
| |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.time.Instant; |
| import java.time.OffsetDateTime; |
| import java.time.ZoneOffset; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.ListIterator; |
| import java.util.Map; |
| import java.util.NoSuchElementException; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| |
| import static org.apache.hadoop.fs.impl.FutureIOSupport.awaitFuture; |
| import static org.apache.hadoop.fs.s3a.Constants.S3N_FOLDER_SUFFIX; |
| import static org.apache.hadoop.fs.s3a.S3AUtils.ACCEPT_ALL; |
| import static org.apache.hadoop.fs.s3a.S3AUtils.createFileStatus; |
| import static org.apache.hadoop.fs.s3a.S3AUtils.maybeAddTrailingSlash; |
| 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; |
| import static org.apache.hadoop.fs.s3a.auth.RoleModel.pathToKey; |
| |
| /** |
| * Place for the S3A listing classes; keeps all the small classes under control. |
| */ |
| @InterfaceAudience.Private |
| public class Listing extends AbstractStoreOperation { |
| |
| private static final Logger LOG = S3AFileSystem.LOG; |
| |
| static final FileStatusAcceptor ACCEPT_ALL_BUT_S3N = |
| new AcceptAllButS3nDirs(); |
| |
| private final ListingOperationCallbacks listingOperationCallbacks; |
| |
| public Listing(ListingOperationCallbacks listingOperationCallbacks, |
| StoreContext storeContext) { |
| super(storeContext); |
| this.listingOperationCallbacks = listingOperationCallbacks; |
| } |
| |
| /** |
| * Create a FileStatus iterator against a provided list of file status, with |
| * a given status filter. |
| * |
| * @param fileStatuses the provided list of file status. NO remote calls. |
| * @param filter file path filter on which paths to accept |
| * @param acceptor the file status acceptor |
| * @return the file status iterator |
| */ |
| ProvidedFileStatusIterator createProvidedFileStatusIterator( |
| S3AFileStatus[] fileStatuses, |
| PathFilter filter, |
| FileStatusAcceptor acceptor) { |
| return new ProvidedFileStatusIterator(fileStatuses, filter, acceptor); |
| } |
| |
| /** |
| * Create a FileStatus iterator against a provided list of file status. |
| * @param fileStatuses array of file status. |
| * @return the file status iterator. |
| */ |
| @VisibleForTesting |
| public static ProvidedFileStatusIterator toProvidedFileStatusIterator( |
| S3AFileStatus[] fileStatuses) { |
| return new ProvidedFileStatusIterator(fileStatuses, |
| ACCEPT_ALL, |
| Listing.ACCEPT_ALL_BUT_S3N); |
| } |
| |
| /** |
| * 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 |
| */ |
| public FileStatusListingIterator createFileStatusListingIterator( |
| Path listPath, |
| S3ListRequest request, |
| PathFilter filter, |
| Listing.FileStatusAcceptor acceptor) throws IOException { |
| return createFileStatusListingIterator(listPath, request, filter, acceptor, |
| null); |
| } |
| |
| /** |
| * 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. |
| * @param providedStatus the provided list of file status, which may contain |
| * items that are not listed from source. |
| * @return the iterator |
| * @throws IOException IO Problems |
| */ |
| @Retries.RetryRaw |
| public FileStatusListingIterator createFileStatusListingIterator( |
| Path listPath, |
| S3ListRequest request, |
| PathFilter filter, |
| Listing.FileStatusAcceptor acceptor, |
| RemoteIterator<S3AFileStatus> providedStatus) throws IOException { |
| return new FileStatusListingIterator( |
| createObjectListingIterator(listPath, request), |
| filter, |
| acceptor, |
| providedStatus); |
| } |
| |
| /** |
| * Create an object listing iterator against a path, with a given |
| * list object request. |
| * @param listPath path of the listing |
| * @param request initial request to make |
| * @return the iterator |
| * @throws IOException IO Problems |
| */ |
| @Retries.RetryRaw |
| public ObjectListingIterator createObjectListingIterator( |
| final Path listPath, |
| final S3ListRequest request) throws IOException { |
| return new ObjectListingIterator(listPath, request); |
| } |
| |
| /** |
| * Create a located status iterator over a file status iterator. |
| * @param statusIterator an iterator over the remote status entries |
| * @return a new remote iterator |
| */ |
| @VisibleForTesting |
| public LocatedFileStatusIterator createLocatedFileStatusIterator( |
| RemoteIterator<S3AFileStatus> statusIterator) { |
| return new LocatedFileStatusIterator(statusIterator); |
| } |
| |
| /** |
| * Create an located status iterator that wraps another to filter out a set |
| * of recently deleted items. |
| * @param iterator an iterator over the remote located status entries. |
| * @param tombstones set of paths that are recently deleted and should be |
| * filtered. |
| * @return a new remote iterator. |
| */ |
| @VisibleForTesting |
| TombstoneReconcilingIterator createTombstoneReconcilingIterator( |
| RemoteIterator<S3ALocatedFileStatus> iterator, Set<Path> tombstones) { |
| return new TombstoneReconcilingIterator(iterator, tombstones); |
| } |
| |
| |
| /** |
| * List files under a path assuming the path to be a directory. |
| * @param path input path. |
| * @param recursive recursive listing? |
| * @param acceptor file status filter |
| * @param collectTombstones should tombstones be collected from S3Guard? |
| * @param forceNonAuthoritativeMS forces metadata store to act like non |
| * authoritative. This is useful when |
| * listFiles output is used by import tool. |
| * @return an iterator over listing. |
| * @throws IOException any exception. |
| */ |
| public RemoteIterator<S3ALocatedFileStatus> getListFilesAssumingDir( |
| Path path, |
| boolean recursive, Listing.FileStatusAcceptor acceptor, |
| boolean collectTombstones, |
| boolean forceNonAuthoritativeMS) throws IOException { |
| |
| String key = maybeAddTrailingSlash(pathToKey(path)); |
| String delimiter = recursive ? null : "/"; |
| if (recursive) { |
| LOG.debug("Recursive list of all entries under {}", key); |
| } else { |
| LOG.debug("Requesting all entries under {} with delimiter '{}'", |
| key, delimiter); |
| } |
| final RemoteIterator<S3AFileStatus> cachedFilesIterator; |
| final Set<Path> tombstones; |
| boolean allowAuthoritative = listingOperationCallbacks |
| .allowAuthoritative(path); |
| if (recursive) { |
| final PathMetadata pm = getStoreContext() |
| .getMetadataStore() |
| .get(path, true); |
| if (pm != null) { |
| if (pm.isDeleted()) { |
| OffsetDateTime deletedAt = OffsetDateTime |
| .ofInstant(Instant.ofEpochMilli( |
| pm.getFileStatus().getModificationTime()), |
| ZoneOffset.UTC); |
| throw new FileNotFoundException("Path " + path + " is recorded as " + |
| "deleted by S3Guard at " + deletedAt); |
| } |
| } |
| MetadataStoreListFilesIterator metadataStoreListFilesIterator = |
| new MetadataStoreListFilesIterator( |
| getStoreContext().getMetadataStore(), |
| pm, |
| allowAuthoritative); |
| tombstones = metadataStoreListFilesIterator.listTombstones(); |
| // if all of the below is true |
| // - authoritative access is allowed for this metadatastore |
| // for this directory, |
| // - all the directory listings are authoritative on the client |
| // - the caller does not force non-authoritative access |
| // return the listing without any further s3 access |
| if (!forceNonAuthoritativeMS && |
| allowAuthoritative && |
| metadataStoreListFilesIterator.isRecursivelyAuthoritative()) { |
| S3AFileStatus[] statuses = S3AUtils.iteratorToStatuses( |
| metadataStoreListFilesIterator, tombstones); |
| cachedFilesIterator = createProvidedFileStatusIterator( |
| statuses, ACCEPT_ALL, acceptor); |
| return createLocatedFileStatusIterator(cachedFilesIterator); |
| } |
| cachedFilesIterator = metadataStoreListFilesIterator; |
| } else { |
| DirListingMetadata meta = |
| S3Guard.listChildrenWithTtl( |
| getStoreContext().getMetadataStore(), |
| path, |
| listingOperationCallbacks.getUpdatedTtlTimeProvider(), |
| allowAuthoritative); |
| if (meta != null) { |
| tombstones = meta.listTombstones(); |
| } else { |
| tombstones = null; |
| } |
| cachedFilesIterator = createProvidedFileStatusIterator( |
| S3Guard.dirMetaToStatuses(meta), ACCEPT_ALL, acceptor); |
| if (allowAuthoritative && meta != null && meta.isAuthoritative()) { |
| // metadata listing is authoritative, so return it directly |
| return createLocatedFileStatusIterator(cachedFilesIterator); |
| } |
| } |
| return createTombstoneReconcilingIterator( |
| createLocatedFileStatusIterator( |
| createFileStatusListingIterator(path, |
| listingOperationCallbacks |
| .createListObjectsRequest(key, delimiter), |
| ACCEPT_ALL, |
| acceptor, |
| cachedFilesIterator)), |
| collectTombstones ? tombstones : null); |
| } |
| |
| /** |
| * Generate list located status for a directory. |
| * Also performing tombstone reconciliation for guarded directories. |
| * @param dir directory to check. |
| * @param filter a path filter. |
| * @return an iterator that traverses statuses of the given dir. |
| * @throws IOException in case of failure. |
| */ |
| public RemoteIterator<S3ALocatedFileStatus> getLocatedFileStatusIteratorForDir( |
| Path dir, PathFilter filter) throws IOException { |
| final String key = maybeAddTrailingSlash(pathToKey(dir)); |
| final Listing.FileStatusAcceptor acceptor = |
| new Listing.AcceptAllButSelfAndS3nDirs(dir); |
| boolean allowAuthoritative = listingOperationCallbacks |
| .allowAuthoritative(dir); |
| DirListingMetadata meta = |
| S3Guard.listChildrenWithTtl(getStoreContext().getMetadataStore(), |
| dir, |
| listingOperationCallbacks |
| .getUpdatedTtlTimeProvider(), |
| allowAuthoritative); |
| Set<Path> tombstones = meta != null |
| ? meta.listTombstones() |
| : null; |
| final RemoteIterator<S3AFileStatus> cachedFileStatusIterator = |
| createProvidedFileStatusIterator( |
| S3Guard.dirMetaToStatuses(meta), filter, acceptor); |
| return (allowAuthoritative && meta != null |
| && meta.isAuthoritative()) |
| ? createLocatedFileStatusIterator( |
| cachedFileStatusIterator) |
| : createTombstoneReconcilingIterator( |
| createLocatedFileStatusIterator( |
| createFileStatusListingIterator(dir, |
| listingOperationCallbacks |
| .createListObjectsRequest(key, "/"), |
| filter, |
| acceptor, |
| cachedFileStatusIterator)), |
| tombstones); |
| } |
| |
| /** |
| * Calculate list of file statuses assuming path |
| * to be a non-empty directory. |
| * @param path input path. |
| * @return Triple of file statuses, metaData, auth flag. |
| * @throws IOException Any IO problems. |
| */ |
| public Triple<RemoteIterator<S3AFileStatus>, DirListingMetadata, Boolean> |
| getFileStatusesAssumingNonEmptyDir(Path path) |
| throws IOException { |
| String key = pathToKey(path); |
| List<S3AFileStatus> result; |
| if (!key.isEmpty()) { |
| key = key + '/'; |
| } |
| |
| boolean allowAuthoritative = listingOperationCallbacks |
| .allowAuthoritative(path); |
| DirListingMetadata dirMeta = |
| S3Guard.listChildrenWithTtl( |
| getStoreContext().getMetadataStore(), |
| path, |
| listingOperationCallbacks.getUpdatedTtlTimeProvider(), |
| allowAuthoritative); |
| // In auth mode return directly with auth flag. |
| if (allowAuthoritative && dirMeta != null && dirMeta.isAuthoritative()) { |
| ProvidedFileStatusIterator mfsItr = createProvidedFileStatusIterator( |
| S3Guard.dirMetaToStatuses(dirMeta), |
| ACCEPT_ALL, |
| Listing.ACCEPT_ALL_BUT_S3N); |
| return Triple.of(mfsItr, |
| dirMeta, Boolean.TRUE); |
| } |
| |
| S3ListRequest request = createListObjectsRequest(key, "/"); |
| LOG.debug("listStatus: doing listObjects for directory {}", key); |
| |
| FileStatusListingIterator filesItr = createFileStatusListingIterator( |
| path, |
| request, |
| ACCEPT_ALL, |
| new Listing.AcceptAllButSelfAndS3nDirs(path)); |
| |
| // return the results obtained from s3. |
| return Triple.of( |
| filesItr, |
| dirMeta, |
| Boolean.FALSE); |
| } |
| |
| public S3ListRequest createListObjectsRequest(String key, String delimiter) { |
| return listingOperationCallbacks.createListObjectsRequest(key, delimiter); |
| } |
| |
| /** |
| * 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); |
| |
| /** |
| * Predicate to decide whether or not to accept a file status. |
| * @param status file status containing file path information |
| * @return true if the status is accepted else false |
| */ |
| boolean accept(FileStatus status); |
| } |
| |
| /** |
| * 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 S3AFileSystem#listStatus(Path)} |
| * calls where the path handed in refers to a file, not a directory: |
| * this is the iterator returned. |
| */ |
| static final class SingleStatusRemoteIterator |
| implements RemoteIterator<S3ALocatedFileStatus> { |
| |
| /** |
| * The status to return; set to null after the first iteration. |
| */ |
| private S3ALocatedFileStatus status; |
| |
| /** |
| * Constructor. |
| * @param status status value: may be null, in which case |
| * the iterator is empty. |
| */ |
| SingleStatusRemoteIterator(S3ALocatedFileStatus 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 S3ALocatedFileStatus next() throws IOException { |
| if (hasNext()) { |
| S3ALocatedFileStatus s = this.status; |
| status = null; |
| return s; |
| } else { |
| throw new NoSuchElementException(); |
| } |
| } |
| } |
| |
| /** |
| * This wraps up a provided non-null list of file status as a remote iterator. |
| * |
| * It firstly filters the provided list and later {@link #next} call will get |
| * from the filtered list. This suffers from scalability issues if the |
| * provided list is too large. |
| * |
| * There is no remote data to fetch. |
| */ |
| static class ProvidedFileStatusIterator |
| implements RemoteIterator<S3AFileStatus> { |
| private final ArrayList<S3AFileStatus> filteredStatusList; |
| private int index = 0; |
| |
| ProvidedFileStatusIterator(S3AFileStatus[] fileStatuses, PathFilter filter, |
| FileStatusAcceptor acceptor) { |
| Preconditions.checkArgument(fileStatuses != null, "Null status list!"); |
| |
| filteredStatusList = new ArrayList<>(fileStatuses.length); |
| for (S3AFileStatus status : fileStatuses) { |
| if (filter.accept(status.getPath()) && acceptor.accept(status)) { |
| filteredStatusList.add(status); |
| } |
| } |
| filteredStatusList.trimToSize(); |
| } |
| |
| @Override |
| public boolean hasNext() throws IOException { |
| return index < filteredStatusList.size(); |
| } |
| |
| @Override |
| public S3AFileStatus next() throws IOException { |
| if (!hasNext()) { |
| throw new NoSuchElementException(); |
| } |
| return filteredStatusList.get(index++); |
| } |
| } |
| |
| /** |
| * 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 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<S3AFileStatus> { |
| |
| /** 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<S3AFileStatus> statusBatchIterator; |
| |
| private final Map<Path, S3AFileStatus> providedStatus; |
| private Iterator<S3AFileStatus> providedStatusIterator; |
| |
| /** |
| * 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. |
| * @param providedStatus the provided list of file status, which may contain |
| * items that are not listed from source. |
| * @throws IOException IO Problems |
| */ |
| @Retries.RetryTranslated |
| FileStatusListingIterator(ObjectListingIterator source, |
| PathFilter filter, |
| FileStatusAcceptor acceptor, |
| @Nullable RemoteIterator<S3AFileStatus> providedStatus) |
| throws IOException { |
| this.source = source; |
| this.filter = filter; |
| this.acceptor = acceptor; |
| this.providedStatus = new HashMap<>(); |
| for (; providedStatus != null && providedStatus.hasNext();) { |
| final S3AFileStatus status = providedStatus.next(); |
| Path path = status.getPath(); |
| if (filter.accept(path) && acceptor.accept(status)) { |
| this.providedStatus.put(path, status); |
| } |
| } |
| // 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. |
| * Lastly, return true if the {@code providedStatusIterator} |
| * has left items. |
| * @return true if a call to {@link #next()} will succeed. |
| * @throws IOException |
| */ |
| @Override |
| @Retries.RetryTranslated |
| public boolean hasNext() throws IOException { |
| return sourceHasNext() || providedStatusIterator.hasNext(); |
| } |
| |
| @Retries.RetryTranslated |
| private boolean sourceHasNext() throws IOException { |
| if (statusBatchIterator.hasNext() || requestNextBatch()) { |
| return true; |
| } else { |
| // turn to file status that are only in provided list |
| if (providedStatusIterator == null) { |
| LOG.debug("Start iterating the provided status."); |
| providedStatusIterator = providedStatus.values().iterator(); |
| } |
| return false; |
| } |
| } |
| |
| @Override |
| @Retries.RetryTranslated |
| public S3AFileStatus next() throws IOException { |
| final S3AFileStatus status; |
| if (sourceHasNext()) { |
| status = statusBatchIterator.next(); |
| // We remove from provided map the file status listed by S3 so that |
| // this does not return duplicate items. |
| |
| // The provided status is returned as it is assumed to have the better |
| // metadata (i.e. the eTag and versionId from S3Guard) |
| S3AFileStatus provided = providedStatus.remove(status.getPath()); |
| if (provided != null) { |
| LOG.debug( |
| "Removed and returned the status from provided file status {}", |
| status); |
| return provided; |
| } |
| } else { |
| if (providedStatusIterator.hasNext()) { |
| status = providedStatusIterator.next(); |
| LOG.debug("Returning provided file status {}", status); |
| } else { |
| throw new NoSuchElementException(); |
| } |
| } |
| return status; |
| } |
| |
| /** |
| * Try to retrieve another batch. |
| * Note that for the initial batch, |
| * {@link ObjectListingIterator} does not generate a request; |
| * it simply returns the initial set. |
| * |
| * @return true if a new batch was created. |
| * @throws IOException IO problems |
| */ |
| @Retries.RetryTranslated |
| 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(S3ListResult objects) { |
| // counters for debug logs |
| int added = 0, ignored = 0; |
| // list to fill in with results. Initial size will be list maximum. |
| List<S3AFileStatus> stats = new ArrayList<>( |
| objects.getObjectSummaries().size() + |
| objects.getCommonPrefixes().size()); |
| // objects |
| for (S3ObjectSummary summary : objects.getObjectSummaries()) { |
| String key = summary.getKey(); |
| Path keyPath = getStoreContext().getContextAccessors().keyToPath(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)) { |
| S3AFileStatus status = createFileStatus(keyPath, summary, |
| listingOperationCallbacks.getDefaultBlockSize(keyPath), |
| getStoreContext().getUsername(), |
| summary.getETag(), null); |
| LOG.debug("Adding: {}", status); |
| stats.add(status); |
| added++; |
| } else { |
| LOG.debug("Ignoring: {}", keyPath); |
| ignored++; |
| } |
| } |
| |
| // prefixes: always directories |
| for (String prefix : objects.getCommonPrefixes()) { |
| Path keyPath = getStoreContext() |
| .getContextAccessors() |
| .keyToPath(prefix); |
| if (acceptor.accept(keyPath, prefix) && filter.accept(keyPath)) { |
| S3AFileStatus status = new S3AFileStatus(Tristate.FALSE, keyPath, |
| getStoreContext().getUsername()); |
| 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 S3AFileSystem#continueListObjects} 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<S3ListResult> { |
| |
| /** The path listed. */ |
| private final Path listPath; |
| |
| /** The most recent listing results. */ |
| private S3ListResult objects; |
| |
| /** The most recent listing request. */ |
| private S3ListRequest request; |
| |
| /** 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; |
| |
| /** |
| * Future to store current batch listing result. |
| */ |
| private CompletableFuture<S3ListResult> s3ListResultFuture; |
| |
| /** |
| * Result of previous batch. |
| */ |
| private S3ListResult objectsPrev; |
| |
| /** |
| * 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 |
| * @throws IOException if listObjects raises one. |
| */ |
| @Retries.RetryRaw |
| ObjectListingIterator( |
| Path listPath, |
| S3ListRequest request) throws IOException { |
| this.listPath = listPath; |
| this.maxKeys = listingOperationCallbacks.getMaxKeys(); |
| this.s3ListResultFuture = listingOperationCallbacks |
| .listObjectsAsync(request); |
| this.request = request; |
| this.objectsPrev = null; |
| } |
| |
| /** |
| * 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 || |
| (objectsPrev != null && objectsPrev.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 |
| @Retries.RetryTranslated |
| public S3ListResult next() throws IOException { |
| if (firstListing) { |
| // clear the firstListing flag for future calls. |
| firstListing = false; |
| // Calculating the result of last async list call. |
| objects = awaitFuture(s3ListResultFuture); |
| fetchNextBatchAsyncIfPresent(); |
| } else { |
| try { |
| if (objectsPrev!= null && !objectsPrev.isTruncated()) { |
| // nothing more to request: fail. |
| throw new NoSuchElementException("No more results in listing of " |
| + listPath); |
| } |
| // Calculating the result of last async list call. |
| objects = awaitFuture(s3ListResultFuture); |
| // Requesting next batch of results. |
| fetchNextBatchAsyncIfPresent(); |
| listingCount++; |
| LOG.debug("New listing status: {}", this); |
| } catch (AmazonClientException e) { |
| throw translateException("listObjects()", listPath, e); |
| } |
| } |
| // Storing the current result to be used by hasNext() call. |
| objectsPrev = objects; |
| return objectsPrev; |
| } |
| |
| /** |
| * If there are more listings present, call for next batch async. |
| * @throws IOException |
| */ |
| private void fetchNextBatchAsyncIfPresent() throws IOException { |
| if (objects.isTruncated()) { |
| LOG.debug("[{}], Requesting next {} objects under {}", |
| listingCount, maxKeys, listPath); |
| s3ListResultFuture = listingOperationCallbacks |
| .continueListObjectsAsync(request, 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; |
| } |
| |
| @Override |
| public boolean accept(FileStatus status) { |
| return (status != null) && status.isFile(); |
| } |
| } |
| |
| /** |
| * Take a remote iterator over a set of {@link FileStatus} instances and |
| * return a remote iterator of {@link LocatedFileStatus} instances. |
| */ |
| class LocatedFileStatusIterator |
| implements RemoteIterator<S3ALocatedFileStatus> { |
| private final RemoteIterator<S3AFileStatus> statusIterator; |
| |
| /** |
| * Constructor. |
| * @param statusIterator an iterator over the remote status entries |
| */ |
| LocatedFileStatusIterator(RemoteIterator<S3AFileStatus> statusIterator) { |
| this.statusIterator = statusIterator; |
| } |
| |
| @Override |
| public boolean hasNext() throws IOException { |
| return statusIterator.hasNext(); |
| } |
| |
| @Override |
| public S3ALocatedFileStatus next() throws IOException { |
| return listingOperationCallbacks |
| .toLocatedFileStatus(statusIterator.next()); |
| } |
| } |
| |
| /** |
| * Wraps another iterator and filters out files that appear in the provided |
| * set of tombstones. Will read ahead in the iterator when necessary to |
| * ensure that emptiness is detected early enough if only deleted objects |
| * remain in the source iterator. |
| */ |
| static class TombstoneReconcilingIterator implements |
| RemoteIterator<S3ALocatedFileStatus> { |
| private S3ALocatedFileStatus next = null; |
| private final RemoteIterator<S3ALocatedFileStatus> iterator; |
| private final Set<Path> tombstones; |
| |
| /** |
| * @param iterator Source iterator to filter |
| * @param tombstones set of tombstone markers to filter out of results |
| */ |
| TombstoneReconcilingIterator(RemoteIterator<S3ALocatedFileStatus> |
| iterator, Set<Path> tombstones) { |
| this.iterator = iterator; |
| if (tombstones != null) { |
| this.tombstones = tombstones; |
| } else { |
| this.tombstones = Collections.emptySet(); |
| } |
| } |
| |
| private boolean fetch() throws IOException { |
| while (next == null && iterator.hasNext()) { |
| S3ALocatedFileStatus candidate = iterator.next(); |
| if (!tombstones.contains(candidate.getPath())) { |
| next = candidate; |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| public boolean hasNext() throws IOException { |
| if (next != null) { |
| return true; |
| } |
| return fetch(); |
| } |
| |
| public S3ALocatedFileStatus next() throws IOException { |
| if (hasNext()) { |
| S3ALocatedFileStatus result = next; |
| next = null; |
| fetch(); |
| return result; |
| } |
| throw new NoSuchElementException(); |
| } |
| } |
| |
| /** |
| * Accept all entries except those which map to S3N pseudo directory markers. |
| */ |
| static class AcceptAllButS3nDirs implements FileStatusAcceptor { |
| |
| public boolean accept(Path keyPath, S3ObjectSummary summary) { |
| return !summary.getKey().endsWith(S3N_FOLDER_SUFFIX); |
| } |
| |
| public boolean accept(Path keyPath, String prefix) { |
| return !keyPath.toString().endsWith(S3N_FOLDER_SUFFIX); |
| } |
| |
| public boolean accept(FileStatus status) { |
| return !status.getPath().toString().endsWith(S3N_FOLDER_SUFFIX); |
| } |
| |
| } |
| |
| /** |
| * Accept all entries except the base path and those which map to S3N |
| * pseudo directory markers. |
| */ |
| public 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); |
| } |
| |
| @Override |
| public boolean accept(FileStatus status) { |
| return (status != null) && !status.getPath().equals(qualifiedPath); |
| } |
| } |
| |
| } |
