HDDS-6810. Add a optional flag to trigger listStatus as part of listKeys for FSO buckets.

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -937,6 +937,28 @@ public List<OzoneFileStatus> listStatus(String keyName, boolean recursive,
         .listStatus(volumeName, name, keyName, recursive, startKey, numEntries);
   }
 
+  /**
+   * List the status for a file or a directory and its contents.
+   *
+   * @param keyName    Absolute path of the entry to be listed
+   * @param recursive  For a directory if true all the descendants of a
+   *                   particular directory are listed
+   * @param startKey   Key from which listing needs to start. If startKey exists
+   *                   its status is included in the final list.
+   * @param numEntries Number of entries to list from the start key
+   * @param allowPartialPrefix allow partial prefixes during listStatus,
+   *                           this is used in context of listKeys calling
+   *                           listStatus
+   * @return list of file status
+   */
+  public List<OzoneFileStatus> listStatus(String keyName, boolean recursive,
+      String startKey, long numEntries, boolean allowPartialPrefix)
+      throws IOException {
+    return proxy
+        .listStatus(volumeName, name, keyName, recursive, startKey,
+            numEntries, allowPartialPrefix);
+  }
+
   /**
    * Return with the list of the in-flight multipart uploads.
    *
@@ -1223,7 +1245,7 @@ private boolean getChildrenKeys(String keyPrefix, String startKey,
 
       // 2. Get immediate children of keyPrefix, starting with startKey
       List<OzoneFileStatus> statuses = proxy.listStatus(volumeName, name,
-              keyPrefix, false, startKey, listCacheSize);
+              keyPrefix, false, startKey, listCacheSize, true);
 
       // 3. Special case: ListKey expects keyPrefix element should present in
       // the resultList, only if startKey is blank. If startKey is not blank

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -815,6 +815,25 @@ List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
       throws IOException;
 
 
+  /**
+   * List the status for a file or a directory and its contents.
+   *
+   * @param volumeName Volume name
+   * @param bucketName Bucket name
+   * @param keyName    Absolute path of the entry to be listed
+   * @param recursive  For a directory if true all the descendants of a
+   *                   particular directory are listed
+   * @param startKey   Key from which listing needs to start. If startKey exists
+   *                   its status is included in the final list.
+   * @param numEntries Number of entries to list from the start key
+   * @param allowPartialPrefixes if partial prefixes should be allowed,
+   *                             this is needed in context of ListKeys
+   * @return list of file status
+   */
+  List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
+      String keyName, boolean recursive, String startKey,
+      long numEntries, boolean allowPartialPrefixes) throws IOException;
+
   /**
    * Add acl for Ozone object. Return true if acl is added successfully else
    * false.

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -1719,22 +1719,38 @@ public OzoneOutputStream createFile(String volumeName, String bucketName,
     return createOutputStream(keySession, UUID.randomUUID().toString());
   }
 
-  @Override
-  public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
-      String keyName, boolean recursive, String startKey, long numEntries)
-      throws IOException {
-    OmKeyArgs keyArgs = new OmKeyArgs.Builder()
+  private OmKeyArgs prepareOmKeyArgs(String volumeName, String bucketName,
+      String keyName) {
+    return new OmKeyArgs.Builder()
         .setVolumeName(volumeName)
         .setBucketName(bucketName)
         .setKeyName(keyName)
         .setRefreshPipeline(true)
         .setSortDatanodesInPipeline(topologyAwareReadEnabled)
         .setLatestVersionLocation(getLatestVersionLocation)
         .build();
+  }
+
+
+  @Override
+  public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
+      String keyName, boolean recursive, String startKey, long numEntries)
+      throws IOException {
+    OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName);
     return ozoneManagerClient
         .listStatus(keyArgs, recursive, startKey, numEntries);
   }
 
+  @Override
+  public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
+      String keyName, boolean recursive, String startKey,
+      long numEntries, boolean allowPartialPrefixes) throws IOException {
+    OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName);
+    return ozoneManagerClient
+        .listStatus(keyArgs, recursive, startKey, numEntries,
+            allowPartialPrefixes);
+  }
+
   /**
    * Add acl for Ozone object. Return true if acl is added successfully else
    * false.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OzoneFSUtils.java

@@ -150,6 +150,28 @@ public static String getFileName(@Nonnull String keyName) {
     return keyName;
   }
 
+  /**
+   * Verifies whether the childKey is a sibling of a given
+   * parentKey.
+   *
+   * @param parentKey parent key name
+   * @param childKey  child key name
+   * @return true if childKey is a sibling of parentKey
+   */
+  public static boolean isSibling(String parentKey, String childKey) {
+    // Empty childKey has no parent, so just returning false.
+    if (org.apache.commons.lang3.StringUtils.isBlank(childKey)) {
+      return false;
+    }
+    java.nio.file.Path parentPath = Paths.get(parentKey);
+    java.nio.file.Path childPath = Paths.get(childKey);
+
+    java.nio.file.Path childParent = childPath.getParent();
+    java.nio.file.Path parentParent = parentPath.getParent();
+
+    return childParent == parentParent;
+  }
+
   /**
    * Verifies whether the childKey is an immediate path under the given
    * parentKey.
@@ -168,6 +190,7 @@ public static boolean isImmediateChild(String parentKey, String childKey) {
     java.nio.file.Path childPath = Paths.get(childKey);
 
     java.nio.file.Path childParent = childPath.getParent();
+
     // Following are the valid parentKey formats:
     // parentKey="" or parentKey="/" or parentKey="/a" or parentKey="a"
     // Following are the valid childKey formats:

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -748,6 +748,24 @@ default OpenKeySession createFile(OmKeyArgs keyArgs, boolean overWrite,
   List<OzoneFileStatus> listStatus(OmKeyArgs keyArgs, boolean recursive,
       String startKey, long numEntries) throws IOException;
 
+  /**
+   * List the status for a file or a directory and its contents.
+   *
+   * @param keyArgs    Key args
+   * @param recursive  For a directory if true all the descendants of a
+   *                   particular directory are listed
+   * @param startKey   Key from which listing needs to start. If startKey exists
+   *                   its status is included in the final list.
+   * @param numEntries Number of entries to list from the start key
+   * @param allowPartialPrefixes if partial prefixes should be allowed,
+   *                             this is needed in context of ListKeys
+   * @return list of file status
+   */
+  List<OzoneFileStatus> listStatus(OmKeyArgs keyArgs, boolean recursive,
+                                   String startKey, long numEntries,
+                                   boolean allowPartialPrefixes)
+      throws IOException;
+
   /**
    * Add acl for Ozone object. Return true if acl is added successfully else
    * false.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -1835,23 +1835,28 @@ public OpenKeySession createFile(OmKeyArgs args,
 
   @Override
   public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
-      String startKey, long numEntries) throws IOException {
+      String startKey, long numEntries, boolean allowPartialPrefixes)
+      throws IOException {
     KeyArgs keyArgs = KeyArgs.newBuilder()
         .setVolumeName(args.getVolumeName())
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName())
         .setSortDatanodes(args.getSortDatanodes())
         .setLatestVersionLocation(args.getLatestVersionLocation())
         .build();
-    ListStatusRequest listStatusRequest =
+    ListStatusRequest.Builder listStatusRequestBuilder =
         ListStatusRequest.newBuilder()
             .setKeyArgs(keyArgs)
             .setRecursive(recursive)
             .setStartKey(startKey)
-            .setNumEntries(numEntries)
-            .build();
+            .setNumEntries(numEntries);
+
+    if (allowPartialPrefixes) {
+      listStatusRequestBuilder.setAllowPartialPrefix(allowPartialPrefixes);
+    }
+
     OMRequest omRequest = createOMRequest(Type.ListStatus)
-        .setListStatusRequest(listStatusRequest)
+        .setListStatusRequest(listStatusRequestBuilder.build())
         .build();
     ListStatusResponse listStatusResponse =
         handleError(submitRequest(omRequest)).getListStatusResponse();
@@ -1864,6 +1869,12 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
     return statusList;
   }
 
+  @Override
+  public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
+      String startKey, long numEntries) throws IOException {
+    return listStatus(args, recursive, startKey, numEntries, false);
+  }
+
   @Override
   public List<RepeatedOmKeyInfo> listTrash(String volumeName,
       String bucketName, String startKeyName, String keyPrefix, int maxKeys)

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestListStatus.java

@@ -92,29 +92,41 @@ public static void teardownClass() {
   @Test
   public void testSortedListStatus() throws Exception {
     // a) test if output is sorted
-    checkKeyList("", "", 1000, 10);
+    checkKeyList("", "", 1000, 10, false);
 
     // b) number of keys returns is expected
-    checkKeyList("", "", 2, 2);
+    checkKeyList("", "", 2, 2, false);
 
     // c) check if full prefix works
-    checkKeyList("a1", "", 100, 3);
+    checkKeyList("a1", "", 100, 3, false);
 
     //  d) check if full prefix with numEntries work
-    checkKeyList("a1", "", 2, 2);
+    checkKeyList("a1", "", 2, 2, false);
 
     // e) check if existing start key >>>
-    checkKeyList("a1", "a1/a12", 100, 2);
+    checkKeyList("a1", "a1/a12", 100, 2, false);
 
-    // f) check with non existing start key>>>
-    checkKeyList("", "a7", 100, 6);
+    // f) check with non-existing start key
+    checkKeyList("", "a7", 100, 6, false);
 
-    // TODO: Enable the following test after listKeys changes
-//    // g) check if half prefix works <<<<
-//     checkKeyList("b", "", 100, 4);
-//
-//    // h) check half prefix with non-existing start key
-//     checkKeyList("b", "b5", 100, 2);
+    // g) check if half prefix works
+    checkKeyList("b", "", 100, 4, true);
+
+    // h) check half prefix with non-existing start key
+    checkKeyList("b", "b5", 100, 2, true);
+
+    // i) check half prefix with non-existing parent in start key
+    checkKeyList("b", "c", 100, 0, true);
+
+    // i) check half prefix with non-existing parent in start key
+    checkKeyList("b", "b/g5", 100, 4, true);
+
+    // i) check half prefix with non-existing parent in start key
+    checkKeyList("b", "c/g5", 100, 0, true);
+
+    // j) check prefix with non-existing prefix key
+    //    and non-existing parent in start key
+    checkKeyList("a1/a111", "a1/a111/a100", 100, 0, true);
   }
 
   private static void createFile(OzoneBucket bucket, String keyName)
@@ -143,8 +155,8 @@ private static void buildNameSpaceTree(OzoneBucket ozoneBucket)
 
     "b1"      File
     "b2"      File
-    "b3"      File
-    "b4"      File
+    "b7"      File
+    "b8"      File
      */
     ozoneBucket.createDirectory("/a1");
     createFile(ozoneBucket, "/a2");
@@ -167,11 +179,13 @@ private static void buildNameSpaceTree(OzoneBucket ozoneBucket)
   }
 
   private void checkKeyList(String keyPrefix, String startKey,
-                            long numEntries, int expectedNumKeys)
+                            long numEntries, int expectedNumKeys,
+                            boolean isPartialPrefix)
       throws Exception {
 
     List<OzoneFileStatus> statuses =
-        fsoOzoneBucket.listStatus(keyPrefix, false, startKey, numEntries);
+        fsoOzoneBucket.listStatus(keyPrefix, false, startKey,
+            numEntries, isPartialPrefix);
     Assert.assertEquals(expectedNumKeys, statuses.size());
 
     System.out.println("BEGIN:::keyPrefix---> " + keyPrefix + ":::---> " +

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -973,6 +973,7 @@ message ListStatusRequest {
     required bool recursive = 2;
     required string startKey = 3;
     required uint64 numEntries = 4;
+    optional bool allowPartialPrefix = 5;
 }
 
 message ListStatusResponse {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -1476,6 +1476,14 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
     return listStatus(args, recursive, startKey, numEntries, null);
   }
 
+  public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
+                                          String startKey, long numEntries,
+                                          String clientAddress)
+      throws IOException {
+    return listStatus(args, recursive, startKey, numEntries,
+        clientAddress, false);
+  }
+
   /**
    * List the status for a file or a directory and its contents.
    *
@@ -1492,8 +1500,8 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
   @Override
   @SuppressWarnings("methodlength")
   public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
-      String startKey, long numEntries, String clientAddress)
-          throws IOException {
+      String startKey, long numEntries, String clientAddress,
+      boolean allowPartialPrefixes) throws IOException {
     Preconditions.checkNotNull(args, "Key args can not be null");
     String volName = args.getVolumeName();
     String buckName = args.getBucketName();
@@ -1511,7 +1519,7 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
                 this::getOzoneFileStatusFSO);
         Collection<OzoneFileStatus> statuses =
             statusHelper.listStatusFSO(args, startKey, numEntries,
-            clientAddress);
+            clientAddress, allowPartialPrefixes);
         return buildFinalStatusList(statuses, args, clientAddress);
       } else {
         return listStatusFSO(args, recursive, startKey, numEntries,