HADOOP-15988. DynamoDBMetadataStore#innerGet should support empty directory flag when using authoritative listings. Contributed by Gabor Bota.

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

@@ -576,12 +576,16 @@ public class DynamoDBMetadataStore implements MetadataStore {
             path.toString(),
             true,
             () -> table.query(spec).iterator().hasNext());
-        // When this class has support for authoritative
-        // (fully-cached) directory listings, we may also be able to answer
-        // TRUE here.  Until then, we don't know if we have full listing or
-        // not, thus the UNKNOWN here:
-        meta.setIsEmptyDirectory(
-            hasChildren ? Tristate.FALSE : Tristate.UNKNOWN);
+
+        // If directory is authoritative, we can set the empty directory flag
+        // to TRUE or FALSE. Otherwise FALSE, or UNKNOWN.
+        if(meta.isAuthoritativeDir()) {
+          meta.setIsEmptyDirectory(
+              hasChildren ? Tristate.FALSE : Tristate.TRUE);
+        } else {
+          meta.setIsEmptyDirectory(
+              hasChildren ? Tristate.FALSE : Tristate.UNKNOWN);
+        }
       }
     }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java

@@ -687,6 +687,53 @@ public class ITestDynamoDBMetadataStore extends MetadataStoreTestBase {
     }
   }
 
+  @Test
+  public void testGetEmptyDirFlagCanSetTrue() throws IOException {
+    boolean authoritativeDirectoryListing = true;
+    testGetEmptyDirFlagCanSetTrueOrUnknown(authoritativeDirectoryListing);
+  }
+
+  @Test
+  public void testGetEmptyDirFlagCanSetUnknown() throws IOException {
+    boolean authoritativeDirectoryListing = false;
+    testGetEmptyDirFlagCanSetTrueOrUnknown(authoritativeDirectoryListing);
+  }
+
+  private void testGetEmptyDirFlagCanSetTrueOrUnknown(boolean auth)
+      throws IOException {
+    // setup
+    final DynamoDBMetadataStore ms = getDynamoMetadataStore();
+    String rootPath = "/testAuthoritativeEmptyDirFlag"+ UUID.randomUUID();
+    String filePath = rootPath + "/file1";
+    final Path dirToPut = fileSystem.makeQualified(new Path(rootPath));
+    final Path fileToPut = fileSystem.makeQualified(new Path(filePath));
+
+    // Create non-auth DirListingMetadata
+    DirListingMetadata dlm =
+        new DirListingMetadata(dirToPut, new ArrayList<>(), auth);
+    if(auth){
+      assertEquals(Tristate.TRUE, dlm.isEmpty());
+    } else {
+      assertEquals(Tristate.UNKNOWN, dlm.isEmpty());
+    }
+    assertEquals(auth, dlm.isAuthoritative());
+
+    // Test with non-authoritative listing, empty dir
+    ms.put(dlm);
+    final PathMetadata pmdResultEmpty = ms.get(dirToPut, true);
+    if(auth){
+      assertEquals(Tristate.TRUE, pmdResultEmpty.isEmptyDirectory());
+    } else {
+      assertEquals(Tristate.UNKNOWN, pmdResultEmpty.isEmptyDirectory());
+    }
+
+    // Test with non-authoritative listing, non-empty dir
+    dlm.put(basicFileStatus(fileToPut, 1, false));
+    ms.put(dlm);
+    final PathMetadata pmdResultNotEmpty = ms.get(dirToPut, true);
+    assertEquals(Tristate.FALSE, pmdResultNotEmpty.isEmptyDirectory());
+  }
+
   /**
    * This validates the table is created and ACTIVE in DynamoDB.
    *