HADOOP-16746 mkdirs and s3guard auth mode

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -3524,7 +3524,7 @@ void finishedWrite(String key, long length, String eTag, String versionId,
           // information gleaned from addAncestors is preserved into the
           // subsequent put.
           stateToClose = S3Guard.initiateBulkWrite(metadataStore,
-              BulkOperationState.OperationType.Put,
+              BulkOperationState.OperationType.Mkdir,
               keyToPath(key));
           activeState = stateToClose;
         }
@@ -3533,13 +3533,20 @@ void finishedWrite(String key, long length, String eTag, String versionId,
         S3AFileStatus status = createUploadFileStatus(p,
             isDir, length,
             getDefaultBlockSize(p), username, eTag, versionId);
-        if (!isDir) {
+        boolean authoritative = false;
+        if (isDir) {
+          // this is a directory marker so put it as such.
+          status.setIsEmptyDirectory(Tristate.TRUE);
+          // and maybe mark as auth
+          authoritative = allowAuthoritative(p);
+        }
+        if (!authoritative) {
+          // for files and non-auth directories
           S3Guard.putAndReturn(metadataStore, status,
               ttlTimeProvider,
               activeState);
         } else {
-          // this is a directory marker so put it as such.
-          status.setIsEmptyDirectory(Tristate.TRUE);
+          // authoritative directory
           S3Guard.putAuthDirectoryMarker(metadataStore, status,
               ttlTimeProvider,
               activeState);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/BulkOperationState.java

@@ -98,5 +98,9 @@ public enum OperationType {
      * Listing update.
      */
     Listing,
+    /**
+     * Mkdir operation.
+     */
+    Mkdir,
   }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java

@@ -202,6 +202,7 @@
  * same region. The region may also be set explicitly by setting the config
  * parameter {@code fs.s3a.s3guard.ddb.region} to the corresponding region.
  */
+@SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class DynamoDBMetadataStore implements MetadataStore,
@@ -899,19 +900,18 @@ private Collection<DDBPathMetadata> completeAncestry(
     List<DDBPathMetadata> sortedPaths = new ArrayList<>(pathsToCreate);
     sortedPaths.sort(PathOrderComparators.TOPMOST_PM_FIRST);
     // iterate through the paths.
-    for (DDBPathMetadata meta : sortedPaths) {
-      Preconditions.checkArgument(meta != null);
-      Path path = meta.getFileStatus().getPath();
+    for (DDBPathMetadata entry : sortedPaths) {
+      Preconditions.checkArgument(entry != null);
+      Path path = entry.getFileStatus().getPath();
       LOG.debug("Adding entry {}", path);
       if (path.isRoot()) {
         // this is a root entry: do not add it.
         break;
       }
-      // create the new entry
-      DDBPathMetadata entry = new DDBPathMetadata(meta);
       // add it to the ancestor state, failing if it is already there and
       // of a different type.
       DDBPathMetadata oldEntry = ancestorState.put(path, entry);
+      boolean addAncestors = true;
       if (oldEntry != null) {
         if (!oldEntry.getFileStatus().isDirectory()
             || !entry.getFileStatus().isDirectory()) {
@@ -928,12 +928,18 @@ private Collection<DDBPathMetadata> completeAncestry(
           // a directory is already present. Log and continue.
           LOG.debug("Directory at {} being updated with value {}",
               path, entry);
+          // and we skip the the subsequent parent scan as we've already been
+          // here
+          addAncestors = false;
         }
       }
       // add the entry to the ancestry map as an explicitly requested entry.
       ancestry.put(path, Pair.of(EntryOrigin.Requested, entry));
+      // now scan up the ancestor tree to see if there are any
+      // immediately missing entries.
       Path parent = path.getParent();
-      while (!parent.isRoot() && !ancestry.containsKey(parent)) {
+      while (addAncestors
+          && !parent.isRoot() && !ancestry.containsKey(parent)) {
         if (!ancestorState.findEntry(parent, true)) {
           // there is no entry in the ancestor state.
           // look in the store
@@ -947,6 +953,9 @@ private Collection<DDBPathMetadata> completeAncestry(
             md = itemToPathMetadata(item, username);
             LOG.debug("Found existing entry for parent: {}", md);
             newEntry = Pair.of(EntryOrigin.Retrieved, md);
+            // and we break, assuming that if there is an entry, its parents
+            // are valid too.
+            addAncestors = false;
           } else {
             // A directory entry was not found in the DB. Create one.
             LOG.debug("auto-create ancestor path {} for child path {}",
@@ -1439,13 +1448,15 @@ static S3AFileStatus makeDirStatus(Path f, String owner) {
    * {@link #processBatchWriteRequest(DynamoDBMetadataStore.AncestorState, PrimaryKey[], Item[])}
    * is only tried once.
    * @param meta Directory listing metadata.
+   * @param unchangedEntries unchanged child entry paths
    * @param operationState operational state for a bulk update
    * @throws IOException IO problem
    */
   @Override
   @Retries.RetryTranslated
   public void put(
       final DirListingMetadata meta,
+      final List<Path> unchangedEntries,
       @Nullable final BulkOperationState operationState) throws IOException {
     LOG.debug("Saving {} dir meta for {} to table {} in region {}: {}",
         meta.isAuthoritative() ? "auth" : "nonauth",
@@ -1463,8 +1474,14 @@ public void put(
     final List<DDBPathMetadata> metasToPut = fullPathsToPut(ddbPathMeta,
         ancestorState);
 
-    // next add all children of the directory
-    metasToPut.addAll(pathMetaToDDBPathMeta(meta.getListing()));
+    // next add all changed children of the directory
+    // ones that came from the previous listing are left as-is
+    final Collection<PathMetadata> children = meta.getListing()
+        .stream()
+        .filter(e -> !unchangedEntries.contains(e.getFileStatus().getPath()))
+        .collect(Collectors.toList());
+
+    metasToPut.addAll(pathMetaToDDBPathMeta(children));
 
     // sort so highest-level entries are written to the store first.
     // if a sequence fails, no orphan entries will have been written.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java

@@ -45,6 +45,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -341,13 +342,14 @@ public void put(PathMetadata meta,
 
   @Override
   public synchronized void put(DirListingMetadata meta,
+      final List<Path> unchangedEntries,
       final BulkOperationState operationState) throws IOException {
     if (LOG.isDebugEnabled()) {
       LOG.debug("put dirMeta {}", meta.prettyPrint());
     }
     LocalMetadataEntry entry =
         localCache.getIfPresent(standardize(meta.getPath()));
-    if(entry == null){
+    if (entry == null) {
       localCache.put(standardize(meta.getPath()), new LocalMetadataEntry(meta));
     } else {
       entry.setDirListingMetadata(meta);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java

@@ -22,6 +22,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -265,11 +266,19 @@ void put(Collection<? extends PathMetadata> metas,
    * missing metadata updates (create, delete) made to the same path by
    * another process.
    *
+   * To optimize updates and avoid overwriting existing entries which
+   * may contain extra data, entries in the list of unchangedEntries may
+   * be excluded. That is: the listing metadata has the full list of
+   * what it believes are children, but implementations can opt to ignore
+   * some.
    * @param meta Directory listing metadata.
+   * @param unchangedEntries list of entries in the dir listing which have
+   * not changed since the directory was list scanned on s3guard.
    * @param operationState operational state for a bulk update
    * @throws IOException if there is an error
    */
   void put(DirListingMetadata meta,
+      final List<Path> unchangedEntries,
       @Nullable BulkOperationState operationState) throws IOException;
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/NullMetadataStore.java

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -113,6 +114,7 @@ public void put(Collection<? extends PathMetadata> meta,
 
   @Override
   public void put(DirListingMetadata meta,
+      final List<Path> unchangedEntries,
       final BulkOperationState operationState) throws IOException {
   }