From 82b798581d12a5cbc9ae17fa290aa81e8ebf6a45 Mon Sep 17 00:00:00 2001
From: Sean Mackrory <mackrorysd@apache.org>
Date: Wed, 12 Dec 2018 09:29:53 -0700
Subject: [PATCH] HADOOP-15988. DynamoDBMetadataStore#innerGet should support
 empty directory flag when using authoritative listings. Contributed by Gabor
 Bota.

---
 .../fs/s3a/s3guard/DynamoDBMetadataStore.java | 16 ++++---
 .../s3guard/ITestDynamoDBMetadataStore.java   | 47 +++++++++++++++++++
 2 files changed, 57 insertions(+), 6 deletions(-)

diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java
index 90f6180c0aa..e8b3c0c548c 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java
@@ -576,12 +576,16 @@ private DDBPathMetadata innerGet(Path path, boolean wantEmptyDirectoryFlag)
             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);
+        }
       }
     }
 
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java
index e958919d545..72863f0c21a 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java
@@ -687,6 +687,53 @@ public void testTableTagging() throws IOException {
     }
   }
 
+  @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.
    *