[HUDI-7367] Add makeQualified APIs (#10607)

* [HUDI-7367] Add makeQualified APIs

* Fix checkstyle

hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java

@@ -40,6 +40,8 @@
 import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
 import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
 import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.HoodieLocation;
+import org.apache.hudi.storage.HoodieStorage;
 import org.apache.hudi.storage.StorageSchemes;
 
 import org.apache.hadoop.conf.Configuration;
@@ -120,6 +122,17 @@ public static Path makeQualified(FileSystem fs, Path path) {
     return path.makeQualified(fs.getUri(), fs.getWorkingDirectory());
   }
 
+  /**
+   * Makes location qualified with {@link HoodieStorage}'s URI.
+   *
+   * @param storage  instance of {@link HoodieStorage}.
+   * @param location to be qualified.
+   * @return qualified location, prefixed with the URI of the target HoodieStorage object provided.
+   */
+  public static HoodieLocation makeQualified(HoodieStorage storage, HoodieLocation location) {
+    return location.makeQualified(storage.getUri());
+  }
+
   /**
    * A write token uniquely identifies an attempt at one of the IOHandle operations (Merge/Create/Append).
    */

hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java

@@ -31,6 +31,11 @@
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
+import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
+import org.apache.hudi.storage.HoodieLocation;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -548,6 +553,22 @@ public void testGetFileStatusAtLevel() throws IOException {
             .collect(Collectors.toSet()));
   }
 
+  @Test
+  public void testMakeQualified() {
+    FileSystem fs = HadoopFSUtils.getFs("file:///a/b/c", new Configuration());
+    FileSystem wrapperFs = new HoodieWrapperFileSystem(fs, new NoOpConsistencyGuard());
+    HoodieStorage storage = new HoodieHadoopStorage(fs);
+    HoodieStorage wrapperStorage = new HoodieHadoopStorage(wrapperFs);
+    assertEquals(new HoodieLocation("file:///x/y"),
+        FSUtils.makeQualified(storage, new HoodieLocation("/x/y")));
+    assertEquals(new HoodieLocation("file:///x/y"),
+        FSUtils.makeQualified(wrapperStorage, new HoodieLocation("/x/y")));
+    assertEquals(new HoodieLocation("s3://x/y"),
+        FSUtils.makeQualified(storage, new HoodieLocation("s3://x/y")));
+    assertEquals(new HoodieLocation("s3://x/y"),
+        FSUtils.makeQualified(wrapperStorage, new HoodieLocation("s3://x/y")));
+  }
+
   private Path getHoodieTempDir() {
     return new Path(baseUri.toString(), ".hoodie/.temp");
   }

hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -53,6 +54,11 @@ public String getScheme() {
     return fs.getScheme();
   }
 
+  @Override
+  public URI getUri() {
+    return fs.getUri();
+  }
+
   @Override
   public OutputStream create(HoodieLocation location, boolean overwrite) throws IOException {
     return fs.create(convertHoodieLocationToPath(location), overwrite);

hudi-io/src/main/java/org/apache/hudi/storage/HoodieLocation.java

@@ -187,6 +187,51 @@ public URI toUri() {
     return uri;
   }
 
+  /**
+   * Returns a qualified location object.
+   *
+   * @param defaultUri if this location is missing the scheme or authority
+   *                   components, borrow them from this URI.
+   * @return this location if it contains a scheme and authority, or
+   * a new path that includes a path and authority and is fully qualified.
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  public HoodieLocation makeQualified(URI defaultUri) {
+    if (!isAbsolute()) {
+      throw new IllegalStateException("Only an absolute path can be made qualified");
+    }
+    HoodieLocation location = this;
+    URI locationUri = location.toUri();
+
+    String scheme = locationUri.getScheme();
+    String authority = locationUri.getAuthority();
+    String fragment = locationUri.getFragment();
+
+    if (scheme != null && (authority != null || defaultUri.getAuthority() == null)) {
+      return location;
+    }
+
+    if (scheme == null) {
+      scheme = defaultUri.getScheme();
+    }
+
+    if (authority == null) {
+      authority = defaultUri.getAuthority();
+      if (authority == null) {
+        authority = "";
+      }
+    }
+
+    URI newUri;
+    try {
+      newUri = new URI(scheme, authority,
+          normalize(locationUri.getPath(), true), null, fragment);
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(e);
+    }
+    return new HoodieLocation(newUri);
+  }
+
   @Override
   public String toString() {
     // This value could be overwritten concurrently and that's okay, since

hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -51,6 +52,14 @@ public abstract class HoodieStorage implements Closeable {
   @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
   public abstract String getScheme();
 
+  /**
+   * Returns a URI which identifies this HoodieStorage.
+   *
+   * @return the URI of this storage instance.
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  public abstract URI getUri();
+
   /**
    * Creates an OutputStream at the indicated location.
    *

hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieLocation.java

@@ -182,6 +182,21 @@ public void testDepth() throws URISyntaxException {
     assertEquals(4, new HoodieLocation(new HoodieLocation("s3://a/b/c"), "d/e").depth());
   }
 
+  @Test
+  public void testMakeQualified() throws URISyntaxException {
+    URI defaultUri = new URI("hdfs://host1/dir1");
+    assertEquals(new HoodieLocation("hdfs://host1/a/b/c"),
+        new HoodieLocation("/a/b/c").makeQualified(defaultUri));
+    assertEquals(new HoodieLocation("hdfs://host2/a/b/c"),
+        new HoodieLocation("hdfs://host2/a/b/c").makeQualified(defaultUri));
+    assertEquals(new HoodieLocation("hdfs://host1/a/b/c"),
+        new HoodieLocation("hdfs:/a/b/c").makeQualified(defaultUri));
+    assertEquals(new HoodieLocation("s3://a/b/c"),
+        new HoodieLocation("s3://a/b/c/").makeQualified(defaultUri));
+    assertThrows(IllegalStateException.class,
+        () -> new HoodieLocation("a").makeQualified(defaultUri));
+  }
+
   @Test
   public void testEquals() {
     assertEquals(new HoodieLocation("/foo"), new HoodieLocation("/foo"));

hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java

@@ -33,6 +33,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -99,6 +101,11 @@ public void testGetScheme() {
     assertEquals("file", getHoodieStorage().getScheme());
   }
 
+  @Test
+  public void testGetUri() throws URISyntaxException {
+    assertEquals(new URI("file:///"), getHoodieStorage().getUri());
+  }
+
   @Test
   public void testCreateWriteAndRead() throws IOException {
     HoodieStorage storage = getHoodieStorage();