fix(fs): stop depending on the optional FileSystem#getScheme()

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java

@@ -53,7 +53,7 @@ public static WriteMarkers get(MarkerType markerType, HoodieTable table, String
         }
         String basePath = table.getMetaClient().getBasePath().toString();
         if (StorageSchemes.HDFS.getScheme().equals(
-            HadoopFSUtils.getFs(basePath, table.getContext().getStorageConf(), true).getScheme())) {
+            HadoopFSUtils.getScheme(HadoopFSUtils.getFs(basePath, table.getContext().getStorageConf(), true)))) {
           log.warn("Timeline-server-based markers are not supported for HDFS: "
               + "base path {}.  Falling back to direct markers.", basePath);
           return getDirectWriteMarkers(table, instantTime);

hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java

@@ -26,6 +26,7 @@
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.util.collection.ImmutablePair;
 import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.storage.StoragePath;
@@ -277,15 +278,50 @@ private static FSDataInputStream getFSDataInputStreamForGCS(FSDataInputStream fs
    * @return true if the inputstream or the wrapped one is of type GoogleHadoopFSInputStream
    */
   public static boolean isGCSFileSystem(FileSystem fs) {
-    return fs.getScheme().equals(StorageSchemes.GCS.getScheme());
+    return StorageSchemes.GCS.getScheme().equals(getScheme(fs));
   }
 
   /**
    * Chdfs will throw {@code IOException} instead of {@code EOFException}. It will cause error in isBlockCorrupted().
    * Wrapped by {@code BoundedFsDataInputStream}, to check whether the desired offset is out of the file size in advance.
    */
   public static boolean isCHDFileSystem(FileSystem fs) {
-    return StorageSchemes.CHDFS.getScheme().equals(fs.getScheme());
+    return StorageSchemes.CHDFS.getScheme().equals(getScheme(fs));
+  }
+
+  /**
+   * Resolves the scheme of {@code fs} without depending on {@link FileSystem#getScheme()}.
+   *
+   * <p>{@code getScheme()} is optional in Hadoop: {@link FileSystem}'s own implementation throws
+   * {@link UnsupportedOperationException}, and proxy implementations such as Presto's
+   * {@code PrestoS3FileSystem} do not override it, so calling it unguarded turns an unrelated read into
+   * "Not implemented by the PrestoS3FileSystem FileSystem implementation" (HUDI-4602).
+   * {@link FileSystem#getUri()} is abstract, so every implementation supplies one to fall back on.
+   *
+   * <p>The two are not interchangeable, which is why {@code getScheme()} is tried first:
+   * {@code InLineFileSystem} returns {@code "inlinefs"} from {@code getScheme()} while its
+   * {@code getUri()} is {@code URI.create("inlinefs")}, which has no colon and so carries no scheme at all.
+   * A URI with no scheme is therefore a resolution failure rather than a value to pass on - returning null
+   * would surface much later as {@code does not support scheme null} or {@code Unsupported scheme :null},
+   * with the original {@code UnsupportedOperationException} discarded.
+   *
+   * @param fs instance of {@link FileSystem} in use.
+   * @return the scheme of {@code fs}, never null.
+   * @throws HoodieException if {@code getScheme()} is unimplemented and the URI carries no scheme.
+   */
+  public static String getScheme(FileSystem fs) {
+    try {
+      return fs.getScheme();
+    } catch (UnsupportedOperationException e) {
+      String scheme = fs.getUri().getScheme();
+      if (scheme == null) {
+        // HoodieException rather than HoodieIOException: the latter only accepts an IOException cause, and
+        // discarding the UnsupportedOperationException is the thing being fixed here.
+        throw new HoodieException("Cannot resolve the scheme of " + fs.getClass().getName()
+            + ": getScheme() is unimplemented and its URI " + fs.getUri() + " carries no scheme", e);
+      }
+      return scheme;
+    }
   }
 
   private static StorageConfiguration<Configuration> getStorageConf(Configuration conf, boolean copy) {
@@ -294,7 +330,7 @@ private static StorageConfiguration<Configuration> getStorageConf(Configuration
 
   public static Configuration registerFileSystem(StoragePath file, Configuration conf) {
     Configuration returnConf = new Configuration(conf);
-    String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
+    String scheme = getScheme(HadoopFSUtils.getFs(file.toString(), conf));
     returnConf.set("fs." + HoodieWrapperFileSystem.getHoodieScheme(scheme) + ".impl",
         HoodieWrapperFileSystem.class.getName());
     return returnConf;

hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java

@@ -277,7 +277,7 @@ public Configuration getConf() {
 
   @Override
   public String getScheme() {
-    return fileSystem.getScheme();
+    return HadoopFSUtils.getScheme(fileSystem);
   }
 
   @Override

hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java

@@ -159,12 +159,8 @@ public HoodieWrapperFileSystem(FileSystem fileSystem, ConsistencyGuard consisten
   }
 
   public static Path convertToHoodiePath(StoragePath file, Configuration conf) {
-    try {
-      String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
-      return convertPathWithScheme(convertToHadoopPath(file), getHoodieScheme(scheme));
-    } catch (HoodieIOException e) {
-      throw e;
-    }
+    String scheme = HadoopFSUtils.getScheme(HadoopFSUtils.getFs(file.toString(), conf));
+    return convertPathWithScheme(convertToHadoopPath(file), getHoodieScheme(scheme));
   }
 
   public static Path convertPathWithScheme(Path oldPath, String newScheme) {

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

@@ -20,6 +20,7 @@
 package org.apache.hudi.storage.hadoop;
 
 import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.util.Lazy;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 import org.apache.hudi.hadoop.fs.HoodieRetryWrapperFileSystem;
@@ -58,6 +59,13 @@
  */
 public class HoodieHadoopStorage extends HoodieStorage {
   private final FileSystem fs;
+  /**
+   * Resolved once. On a filesystem that does not implement {@code getScheme()} the fallback in
+   * {@link HadoopFSUtils#getScheme} costs a thrown-and-caught exception, and this is called once per log
+   * block via {@code StorageSchemes.isWriteTransactional} and three times per immutable-file write via
+   * {@code needCreateTempFile}. {@code fs} is final, so the answer cannot change.
+   */
+  private final Lazy<String> scheme = Lazy.lazily(this::resolveScheme);
 
   public HoodieHadoopStorage(StoragePath path, StorageConfiguration<?> conf) {
     super(conf);
@@ -111,7 +119,11 @@ public HoodieStorage newInstance(StoragePath path, StorageConfiguration<?> stora
 
   @Override
   public String getScheme() {
-    return fs.getScheme();
+    return scheme.get();
+  }
+
+  private String resolveScheme() {
+    return HadoopFSUtils.getScheme(fs);
   }
 
   @Override

hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java

@@ -45,7 +45,6 @@
 import java.util.Arrays;
 import java.util.List;
 
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -107,9 +106,13 @@ public void testGetSchema() {
     FileSystem fileSystem =
         new HoodieRetryWrapperFileSystem(fakeFs, maxRetryIntervalMs, maxRetryNumbers,
             initialRetryIntervalMs, "");
-    HoodieWrapperFileSystem fs =
-        new HoodieWrapperFileSystem(fileSystem, new NoOpConsistencyGuard());
-    assertDoesNotThrow(fs::getScheme, "Method #getSchema does not implement correctly");
+    // FakeRemoteFileSystem deliberately does not override getScheme(), so FileSystem's own implementation
+    // throws - the PrestoS3FileSystem shape (HUDI-4602). Assert on the retry wrapper itself: asserting on
+    // HoodieWrapperFileSystem instead would only exercise its own uri.getScheme() and never reach here,
+    // which is why this guard was inert from the day HUDI-5286 added it.
+    assertThrows(UnsupportedOperationException.class, fakeFs::getScheme);
+    assertEquals("file", ((HoodieRetryWrapperFileSystem) fileSystem).getScheme(),
+        "the retry wrapper should resolve the scheme of a filesystem that does not implement getScheme()");
   }
 
   @Test
@@ -254,11 +257,6 @@ public Configuration getConf() {
       return fs.getConf();
     }
 
-    @Override
-    public String getScheme() {
-      return fs.getScheme();
-    }
-
     @Override
     public short getDefaultReplication(Path path) {
       return defaultReplication;

hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java

@@ -19,25 +19,188 @@
 
 package org.apache.hudi.hadoop.fs;
 
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+
 import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopFileStatus;
 import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
 import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
 import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePathInfo;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Tests {@link HadoopFSUtils}
  */
 public class TestHadoopFSUtils {
+  /**
+   * HUDI-4602: {@link FileSystem#getScheme()} is optional in Hadoop -- the base implementation throws
+   * {@link UnsupportedOperationException} -- and proxy implementations such as Presto's
+   * {@code PrestoS3FileSystem} do not override it. Opening a log file went straight through
+   * {@code isGCSFileSystem}, so a MOR {@code _rt} query on Presto failed with
+   * "Not implemented by the PrestoS3FileSystem FileSystem implementation" rather than reading anything.
+   *
+   * <p>{@link FilterFileSystem} has the same shape: it leaves {@code getScheme()} to the throwing base
+   * implementation while overriding {@code getUri()}.
+   */
+  @Test
+  public void testGetFSDataInputStreamWhenGetSchemeIsUnimplemented(@TempDir java.nio.file.Path tempDir) throws IOException {
+    java.nio.file.Path file = tempDir.resolve("log.file");
+    byte[] contents = new byte[] {1, 2, 3, 4};
+    Files.write(file, contents);
+    // newInstanceLocal rather than getLocal, so closing this does not evict a cached FileSystem that
+    // other tests in the same JVM share.
+    try (FileSystem fs = newFsWithoutGetScheme(FileSystem.newInstanceLocal(new Configuration()))) {
+      try (FSDataInputStream stream =
+               HadoopFSUtils.getFSDataInputStream(fs, new StoragePath(file.toUri()), 1024, true)) {
+        byte[] read = new byte[contents.length];
+        stream.readFully(read);
+        assertArrayEquals(contents, read, "The read path should not depend on the optional getScheme()");
+      }
+    }
+  }
+
+  @Test
+  public void testGetSchemeFallsBackToTheUriWhenUnimplemented() throws IOException {
+    try (FileSystem localFs = FileSystem.newInstanceLocal(new Configuration())) {
+      assertEquals("file", HadoopFSUtils.getScheme(localFs),
+          "LocalFileSystem overrides getScheme(), so the helper should return what it reports "
+              + "rather than falling back to getUri()");
+
+      // FilterFileSystem#close closes the delegate, so the wrapper is not given its own block: it owns
+      // nothing, and closing it here would close localFs a second time.
+      FileSystem noScheme = newFsWithoutGetScheme(localFs);
+      assertEquals("file", HadoopFSUtils.getScheme(noScheme),
+          "FilterFileSystem does not override getScheme(), so the helper should fall back to "
+              + "getUri().getScheme()");
+    }
+  }
+
+  /**
+   * A URI with no scheme cannot stand in for an unimplemented {@code getScheme()}. {@code InLineFileSystem}
+   * is the case in this module: {@code getScheme()} returns "inlinefs" while {@code getUri()} is
+   * {@code URI.create("inlinefs")}, which has no colon and so no scheme. Returning null there would surface
+   * far away as "does not support scheme null" with the original failure discarded, so it must fail here.
+   */
+  @Test
+  public void testGetSchemeFailsLoudlyWhenNeitherSourceHasOne() throws IOException {
+    try (FileSystem localFs = FileSystem.newInstanceLocal(new Configuration())) {
+      FileSystem schemeless = new NoSchemeFileSystem(localFs, URI.create("inlinefs"));
+
+      HoodieException thrown =
+          assertThrows(HoodieException.class, () -> HadoopFSUtils.getScheme(schemeless));
+      assertTrue(thrown.getMessage().contains("carries no scheme"),
+          () -> "the failure should say the URI carries no scheme, but was: " + thrown.getMessage());
+      assertInstanceOf(UnsupportedOperationException.class, thrown.getCause(),
+          "the original getScheme() failure must be chained rather than discarded");
+    }
+  }
+
+  /**
+   * The three call sites this rerouted that no test in the repo reached: {@code registerFileSystem},
+   * {@code HoodieWrapperFileSystem#convertToHoodiePath} - which is on the write path, via
+   * {@code HoodieBaseParquetWriter} and friends - and {@code HoodieHadoopStorage#getScheme}. All three threw
+   * {@link UnsupportedOperationException} on a filesystem without {@code getScheme()} before this change.
+   */
+  @Test
+  public void testCallSitesWorkOnAFileSystemWithoutGetScheme(@TempDir java.nio.file.Path tempDir) {
+    Configuration conf = new Configuration();
+    conf.setClass("fs.file.impl", NoSchemeLocalFileSystem.class, FileSystem.class);
+    StoragePath path = new StoragePath(tempDir.toUri());
+
+    assertDoesNotThrow(() -> HadoopFSUtils.registerFileSystem(path, conf),
+        "registerFileSystem resolves the scheme to build the fs.<scheme>.impl key");
+    assertDoesNotThrow(() -> HoodieWrapperFileSystem.convertToHoodiePath(path, conf),
+        "convertToHoodiePath is on the write path and resolves the scheme to rewrite it");
+    assertEquals("file", new HoodieHadoopStorage(path, HadoopFSUtils.getStorageConf(conf)).getScheme(),
+        "HoodieHadoopStorage#getScheme is what HoodieStorage callers reach");
+  }
+
+  /**
+   * {@code isGCSFileSystem} and {@code isCHDFileSystem} become reachable for a filesystem without
+   * {@code getScheme()} for the first time with this change, and they select different stream wrappers.
+   * Neither predicate had a test before.
+   */
+  @ParameterizedTest
+  @CsvSource({
+      "gs://bucket, org.apache.hudi.hadoop.fs.SchemeAwareFSDataInputStream",
+      "ofs://cluster, org.apache.hudi.hadoop.fs.BoundedFsDataInputStream"
+  })
+  public void testSchemeSpecificStreamIsSelectedWithoutGetScheme(String uri, String expectedStream,
+                                                                 @TempDir java.nio.file.Path tempDir) throws IOException {
+    java.nio.file.Path file = tempDir.resolve("log.file");
+    Files.write(file, new byte[] {1, 2, 3, 4});
+    try (FileSystem localFs = FileSystem.newInstanceLocal(new Configuration())) {
+      // Reports a gs:// or ofs:// URI while leaving getScheme() to the throwing base implementation.
+      FileSystem fs = new NoSchemeFileSystem(localFs, URI.create(uri));
+      assertThrows(UnsupportedOperationException.class, fs::getScheme);
+
+      try (FSDataInputStream stream =
+               HadoopFSUtils.getFSDataInputStream(fs, new StoragePath(file.toUri()), 1024, true)) {
+        assertEquals(expectedStream, stream.getClass().getName(),
+            "the scheme-specific wrapper should be selected from the fallback-resolved scheme");
+      }
+    }
+  }
+
+  /** A FileSystem with the reported shape: {@code getUri()} works, {@code getScheme()} throws. */
+  private static FileSystem newFsWithoutGetScheme(FileSystem delegate) {
+    FileSystem fs = new FilterFileSystem(delegate);
+    // The premise of every assertion below: this is the call the read path used to make unguarded.
+    assertThrows(UnsupportedOperationException.class, fs::getScheme);
+    return fs;
+  }
+
+  /** Same shape, but reporting a URI of our choosing so scheme-specific branches can be reached. */
+  private static class NoSchemeFileSystem extends FilterFileSystem {
+    private final URI uri;
+
+    NoSchemeFileSystem(FileSystem delegate, URI uri) {
+      super(delegate);
+      this.uri = uri;
+    }
+
+    @Override
+    public URI getUri() {
+      return uri;
+    }
+  }
+
+  /**
+   * A {@link LocalFileSystem} that does not implement {@code getScheme()}, so it can be registered as
+   * {@code fs.file.impl} and reached through the normal {@code FileSystem.get} path.
+   */
+  public static class NoSchemeLocalFileSystem extends LocalFileSystem {
+    @Override
+    public String getScheme() {
+      throw new UnsupportedOperationException(
+          "Not implemented by the NoSchemeLocalFileSystem FileSystem implementation");
+    }
+  }
+
   @ParameterizedTest
   @ValueSource(strings = {
       "/a/b/c",