HADOOP-19543. [ABFS][FnsOverBlob] Remove Duplicates from Blob Endpoint Listing Across Iterations

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

@@ -45,6 +45,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.WeakHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -76,6 +77,8 @@
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidUriException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.TrileanConversionException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
+import org.apache.hadoop.fs.azurebfs.contracts.services.BlobListResultEntrySchema;
+import org.apache.hadoop.fs.azurebfs.services.AbfsBlobClient;
 import org.apache.hadoop.fs.azurebfs.services.ListResponseData;
 import org.apache.hadoop.fs.azurebfs.enums.Trilean;
 import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
@@ -1273,6 +1276,7 @@ public String listStatus(final Path path, final String startFrom,
       }
     }
 
+    TreeMap<String, VersionedFileStatus> nameToEntryMap = new TreeMap<>();
     do {
       try (AbfsPerfInfo perfInfo = startTracking("listStatus", "listPath")) {
         ListResponseData listResponseData = listingClient.listPath(relativePath,
@@ -1281,9 +1285,17 @@ public String listStatus(final Path path, final String startFrom,
         AbfsRestOperation op = listResponseData.getOp();
         perfInfo.registerResult(op.getResult());
         continuation = listResponseData.getContinuationToken();
-        List<FileStatus> fileStatusListInCurrItr = listResponseData.getFileStatusList();
+        List<VersionedFileStatus> fileStatusListInCurrItr = listResponseData.getFileStatusList();
         if (fileStatusListInCurrItr != null && !fileStatusListInCurrItr.isEmpty()) {
-          fileStatuses.addAll(fileStatusListInCurrItr);
+          if (listingClient instanceof AbfsBlobClient) {
+            /* Blob Endpoint can return duplicate entries for non-empty explicit
+             * directories. Such entries can come across multiple iterations of
+             * list call and hence need to be filtered here.
+             */
+            filterDuplicateEntriesForBlobClient(nameToEntryMap, fileStatusListInCurrItr, fileStatuses);
+          } else {
+            fileStatuses.addAll(fileStatusListInCurrItr);
+          }
         }
         perfInfo.registerSuccess(true);
         countAggregate++;
@@ -1299,6 +1311,29 @@ public String listStatus(final Path path, final String startFrom,
     return continuation;
   }
 
+  private void filterDuplicateEntriesForBlobClient(
+      TreeMap<String, VersionedFileStatus> nameToEntryMap,
+      List<VersionedFileStatus> fileStatusListInCurrItr,
+      List<FileStatus> fileStatuses) {
+    for (VersionedFileStatus fileStatus : fileStatusListInCurrItr) {
+      String entryName = fileStatus.getPath().getName();
+      if (StringUtils.isNotEmpty(fileStatus.getEtag())) {
+        // This is a blob entry. It is either a file or a marker blob.
+        // In both cases we will add this.
+        nameToEntryMap.put(entryName, fileStatus);
+        fileStatuses.add(fileStatus);
+      } else {
+        // This is a BlobPrefix entry.
+        // It is a directory with file inside
+        // This might have already been added as a marker blob.
+        if (!nameToEntryMap.containsKey(entryName)) {
+          nameToEntryMap.put(entryName, fileStatus);
+          fileStatuses.add(fileStatus);
+        }
+      }
+    }
+  }
+
   // generate continuation token for xns account
   private String generateContinuationTokenForXns(final String firstEntryName) {
     Preconditions.checkArgument(!Strings.isNullOrEmpty(firstEntryName)

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobClient.java

@@ -402,7 +402,7 @@ public ListResponseData listPath(final String relativePath, final boolean recurs
       AbfsRestOperation pathStatus = this.getPathStatus(relativePath, tracingContext, null, false);
       BlobListResultSchema listResultSchema = getListResultSchemaFromPathStatus(relativePath, pathStatus);
       LOG.debug("ListBlob attempted on a file path. Returning file status.");
-      List<FileStatus> fileStatusList = new ArrayList<>();
+      List<VersionedFileStatus> fileStatusList = new ArrayList<>();
       for (BlobListResultEntrySchema entry : listResultSchema.paths()) {
         fileStatusList.add(getVersionedFileStatusFromEntry(entry, uri));
       }
@@ -1617,7 +1617,7 @@ public ListResponseData parseListPathResults(AbfsHttpOperation result, URI uri)
         LOG.debug("ListBlobs listed {} blobs with {} as continuation token",
             listResultSchema.paths().size(),
             listResultSchema.getNextMarker());
-        return filterDuplicateEntriesAndRenamePendingFiles(listResultSchema, uri);
+        return filterRenamePendingFiles(listResultSchema, uri);
       } catch (SAXException | IOException ex) {
         throw new AbfsDriverException(ERR_BLOB_LIST_PARSING, ex);
       }
@@ -1927,29 +1927,16 @@ private List<AbfsHttpHeader> getMetadataHeadersList(final Hashtable<String, Stri
    * @throws IOException if path conversion fails.
    */
   @VisibleForTesting
-  public ListResponseData filterDuplicateEntriesAndRenamePendingFiles(
+  public ListResponseData filterRenamePendingFiles(
       BlobListResultSchema listResultSchema, URI uri) throws IOException {
-    List<FileStatus> fileStatuses = new ArrayList<>();
+    List<VersionedFileStatus> fileStatuses = new ArrayList<>();
     Map<Path, Integer> renamePendingJsonPaths = new HashMap<>();
-    TreeMap<String, BlobListResultEntrySchema> nameToEntryMap = new TreeMap<>();
 
     for (BlobListResultEntrySchema entry : listResultSchema.paths()) {
-      if (StringUtils.isNotEmpty(entry.eTag())) {
-        // This is a blob entry. It is either a file or a marker blob.
-        // In both cases we will add this.
-        if (isRenamePendingJsonPathEntry(entry)) {
-          renamePendingJsonPaths.put(entry.path(), entry.contentLength().intValue());
-        } else {
-          nameToEntryMap.put(entry.name(), entry);
-          fileStatuses.add(getVersionedFileStatusFromEntry(entry, uri));
-        }
+      if (isRenamePendingJsonPathEntry(entry)) {
+        renamePendingJsonPaths.put(entry.path(), entry.contentLength().intValue());
       } else {
-        // This is a BlobPrefix entry. It is a directory with file inside
-        // This might have already been added as a marker blob.
-        if (!nameToEntryMap.containsKey(entry.name())) {
-          nameToEntryMap.put(entry.name(), entry);
-          fileStatuses.add(getVersionedFileStatusFromEntry(entry, uri));
-        }
+        fileStatuses.add(getVersionedFileStatusFromEntry(entry, uri));
       }
     }
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsDfsClient.java

@@ -1353,7 +1353,7 @@ public ListResponseData parseListPathResults(AbfsHttpOperation result, URI uri)
         LOG.debug("ListPath listed {} paths with {} as continuation token",
             listResultSchema.paths().size(),
             getContinuationFromResponse(result));
-        List<FileStatus> fileStatuses = new ArrayList<>();
+        List<VersionedFileStatus> fileStatuses = new ArrayList<>();
         for (DfsListResultEntrySchema entry : listResultSchema.paths()) {
           fileStatuses.add(getVersionedFileStatusFromEntry(entry, uri));
         }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ListResponseData.java

@@ -31,7 +31,7 @@
  */
 public class ListResponseData {
 
-  private List<FileStatus> fileStatusList;
+  private List<VersionedFileStatus> fileStatusList;
   private Map<Path, Integer> renamePendingJsonPaths;
   private AbfsRestOperation executedRestOperation;
   private String continuationToken;
@@ -40,15 +40,15 @@ public class ListResponseData {
    * Returns the list of FileStatus objects.
    * @return the list of FileStatus objects
    */
-  public List<FileStatus> getFileStatusList() {
+  public List<VersionedFileStatus> getFileStatusList() {
     return fileStatusList;
   }
 
   /**
    * Sets the list of FileStatus objects.
    * @param fileStatusList the list of FileStatus objects
    */
-  public void setFileStatusList(final List<FileStatus> fileStatusList) {
+  public void setFileStatusList(final List<VersionedFileStatus> fileStatusList) {
     this.fileStatusList = fileStatusList;
   }
 

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemListStatus.java

@@ -197,7 +197,7 @@ public void testListPathParsingFailure() throws Exception {
     Mockito.doReturn(spiedStore).when(spiedFs).getAbfsStore();
     Mockito.doReturn(spiedClient).when(spiedStore).getClient();
 
-    Mockito.doThrow(new SocketException(CONNECTION_RESET_MESSAGE)).when(spiedClient).filterDuplicateEntriesAndRenamePendingFiles(any(), any());
+    Mockito.doThrow(new SocketException(CONNECTION_RESET_MESSAGE)).when(spiedClient).filterRenamePendingFiles(any(), any());
     List<FileStatus> fileStatuses = new ArrayList<>();
     AbfsDriverException ex = intercept(AbfsDriverException.class,
       () -> {
@@ -532,6 +532,28 @@ public void testEmptyContinuationToken() throws Exception {
         .describedAs("Listing Size Not as expected").hasSize(1);
   }
 
+  @Test
+  public void testDuplicateEntriesAcrossListBlobIterations() throws Exception {
+    AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
+    AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
+    store.getAbfsConfiguration().setListMaxResults(1);
+    AbfsClient client = Mockito.spy(store.getClient());
+
+    Mockito.doReturn(store).when(fs).getAbfsStore();
+    Mockito.doReturn(client).when(store).getClient();
+
+    // Create a non-empty explicit directory under root
+    Path dir = new Path("/a");
+    Path fileInsideDir = new Path("/a/file");
+    createAzCopyFolder(dir);
+    fs.create(fileInsideDir);
+
+    FileStatus[] fileStatuses = fs.listStatus(new Path(ROOT_PATH));
+    Assertions.assertThat(fileStatuses.length)
+        .describedAs("List size is not expected").isEqualTo(1);
+    assertExplicitDirectoryFileStatus(fileStatuses[0], fs.makeQualified(dir));
+  }
+
   private void assertFilePathFileStatus(final FileStatus fileStatus,
       final Path qualifiedPath) {
     Assertions.assertThat(fileStatus.getPath())