diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java index 2765ffdd6286a..8191e6d04cba4 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java +++ b/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); diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java index e167657803143..8c0293dbe7957 100644 --- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java +++ b/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,7 +278,7 @@ 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)); } /** @@ -285,7 +286,42 @@ public static boolean isGCSFileSystem(FileSystem fs) { * 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()}. + * + *

{@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. + * + *

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 getStorageConf(Configuration conf, boolean copy) { @@ -294,7 +330,7 @@ private static StorageConfiguration 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; diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java index c9d8fff3fbbd7..d7c1ca5f72fc8 100644 --- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java +++ b/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 diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java index 24674ee725a90..f8d731fd0ce21 100644 --- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java +++ b/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) { diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java index 87a9ad1019f63..067f77d491bd6 100644 --- a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java +++ b/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 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 diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java index bb0b3608c7fc8..aaaf749e85d46 100644 --- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java +++ b/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; diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java index 7768ff4feae7f..da4e9a7500f42 100644 --- a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java +++ b/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. + * + *

{@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..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",