[HUDI-9681] Remove mkdir in partition listing and add try catch to listStatus of partition (#13739)

Author: yihua

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -398,9 +398,7 @@ protected Map<Pair<String, StoragePath>, List<StoragePathInfo>> listPartitions(
         pathInfoMap.put(partitionPair,
             metaClient.getStorage().listDirectEntries(absolutePartitionPath));
       } catch (IOException e) {
-        // Create the path if it does not exist already
         if (!metaClient.getStorage().exists(absolutePartitionPath)) {
-          metaClient.getStorage().createDirectory(absolutePartitionPath);
           pathInfoMap.put(partitionPair, Collections.emptyList());
         } else {
           // in case the partition path was created by another caller
@@ -471,9 +469,7 @@ protected List<StoragePathInfo> listPartition(StoragePath partitionPath) throws
     try {
       return metaClient.getStorage().listDirectEntries(partitionPath);
     } catch (IOException e) {
-      // Create the path if it does not exist already
       if (!metaClient.getStorage().exists(partitionPath)) {
-        metaClient.getStorage().createDirectory(partitionPath);
         return Collections.emptyList();
       } else {
         // in case the partition path was created by another caller

hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java

@@ -44,6 +44,7 @@
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.storage.StoragePathInfo;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -52,6 +53,7 @@
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Implementation of {@link HoodieTableMetadata} based file-system-backed table metadata.
@@ -177,7 +179,12 @@ private List<String> getPartitionPathWithPathPrefixUsingFilterExpression(String
           "Listing all partitions with prefix " + relativePathPrefix);
       // Need to use serializable file status here, see HUDI-5936
       List<StoragePathInfo> dirToFileListing = engineContext.flatMap(pathsToList, path -> {
-        return getStorage().listDirectEntries(path).stream();
+        try {
+          return getStorage().listDirectEntries(path).stream();
+        } catch (FileNotFoundException e) {
+          // The partition may have been cleaned.
+          return Stream.empty();
+        }
       }, listingParallelism);
       pathsToList.clear();