This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 81e68d152239c70865159b8e1eea8186071f2c8b Author: Eunbin Son <[email protected]> AuthorDate: Thu Sep 17 14:07:48 2026 +0900 [common] Delegate listFilesIterative through the remaining FileIO wrappers (#9865) --- .../java/org/apache/paimon/fs/PluginFileIO.java | 8 +++++++ .../java/org/apache/paimon/fs/ResolvingFileIO.java | 8 +++++++ .../org/apache/paimon/fs/cache/CachingFileIO.java | 9 ++++++++ .../org/apache/paimon/fs/PluginFileIOTest.java | 27 ++++++++++++++++++++++ .../org/apache/paimon/fs/ResolvingFileIOTest.java | 20 ++++++++++++++++ .../apache/paimon/fs/cache/CachingFileIOTest.java | 21 +++++++++++++++++ 6 files changed, 93 insertions(+) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java index 587c1f2d44..b1ffb43a9c 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java @@ -98,6 +98,14 @@ public abstract class PluginFileIO implements FileIO, HadoopOptionsProvider { return wrap(() -> fileIO(path).tryToWriteAtomic(path, content)); } + @Override + public RemoteIterator<FileStatus> listFilesIterative(Path path, boolean recursive) + throws IOException { + // the interface default would hide the plugin FileIO's iterative listing override and + // list each directory with listStatus instead + return wrap(() -> fileIO(path).listFilesIterative(path, recursive)); + } + @Override public String createBlobPresignedUrl( Path tableRoot, BlobDescriptor descriptor, Duration validity) throws IOException { diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java index cc3aba497e..5091acaaff 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java @@ -124,6 +124,14 @@ public class ResolvingFileIO implements FileIO { return wrap(() -> fileIO(path).newTwoPhaseOutputStream(path, overwrite)); } + @Override + public RemoteIterator<FileStatus> listFilesIterative(Path path, boolean recursive) + throws IOException { + // the interface default would hide the resolved FileIO's iterative listing override and + // list each directory with listStatus instead + return wrap(() -> fileIO(path).listFilesIterative(path, recursive)); + } + @Override public String createBlobPresignedUrl( Path tableRoot, BlobDescriptor descriptor, Duration validity) throws IOException { diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java index 65eeaa3ebf..d943abcfbc 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java @@ -24,6 +24,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.RemoteIterator; import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.MemorySize; @@ -186,6 +187,14 @@ public class CachingFileIO implements FileIO { return delegate.tryToWriteAtomic(path, content); } + @Override + public RemoteIterator<FileStatus> listFilesIterative(Path path, boolean recursive) + throws IOException { + // the interface default would hide the delegate's iterative listing override and list + // each directory with listStatus instead + return delegate.listFilesIterative(path, recursive); + } + @Override public String createBlobPresignedUrl( Path tableRoot, BlobDescriptor descriptor, Duration validity) throws IOException { diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java index 2411cce3d0..2be00e899e 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java @@ -26,7 +26,10 @@ import java.io.IOException; import java.time.Duration; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Tests for {@link PluginFileIO}. */ @@ -56,6 +59,30 @@ class PluginFileIOTest { assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(original); } + @Test + void testListFilesIterativeReachesPluginOverride() throws IOException { + FileIO delegate = mock(FileIO.class); + ClassLoader pluginClassLoader = new ClassLoader() {}; + TestPluginFileIO fileIO = new TestPluginFileIO(delegate, pluginClassLoader); + Path tableRoot = new Path("oss://bucket/table"); + @SuppressWarnings("unchecked") + RemoteIterator<FileStatus> marker = mock(RemoteIterator.class); + ClassLoader original = Thread.currentThread().getContextClassLoader(); + when(delegate.listFilesIterative(tableRoot, true)) + .thenAnswer( + ignored -> { + assertThat(Thread.currentThread().getContextClassLoader()) + .isSameAs(pluginClassLoader); + return marker; + }); + + assertThat(fileIO.listFilesIterative(tableRoot, true)).isSameAs(marker); + verify(delegate).listFilesIterative(tableRoot, true); + // the interface default would construct its own iterator backed by listStatus + verify(delegate, never()).listStatus(any()); + assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(original); + } + private static class TestPluginFileIO extends PluginFileIO { private final FileIO delegate; diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java index e5203833aa..6330d68daa 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java @@ -36,6 +36,7 @@ import java.util.concurrent.Future; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -202,4 +203,23 @@ public class ResolvingFileIOTest { // the interface default would have renamed a temp file on the resolver instead verify(delegate, never()).rename(any(), any()); } + + @Test + public void testListFilesIterativeReachesResolvedOverride() throws IOException { + FileIO delegate = mock(FileIO.class); + FileIOLoader loader = mock(FileIOLoader.class); + when(loader.load(any())).thenReturn(delegate); + when(loader.getScheme()).thenReturn("oss"); + resolvingFileIO.configure(CatalogContext.create(new Options(), loader, null)); + + Path tableRoot = new Path("oss://bucket/table"); + @SuppressWarnings("unchecked") + RemoteIterator<FileStatus> marker = mock(RemoteIterator.class); + when(delegate.listFilesIterative(tableRoot, true)).thenReturn(marker); + + assertSame(marker, resolvingFileIO.listFilesIterative(tableRoot, true)); + verify(delegate).listFilesIterative(tableRoot, true); + // the interface default would construct its own iterator backed by listStatus + verify(delegate, never()).listStatus(any()); + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java index d3ae9d0663..b4f3c46373 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java @@ -25,6 +25,7 @@ import org.apache.paimon.fs.FileRange; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.RemoteIterator; import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.VectoredReadUtils; import org.apache.paimon.options.MemorySize; @@ -190,6 +191,26 @@ class CachingFileIOTest { verify(delegate, never()).rename(any(), any()); } + @Test + void testListFilesIterativeReachesDelegateOverride() throws IOException { + FileIO delegate = mock(FileIO.class); + CachingFileIO cachingIO = + newCachingFileIO( + delegate, + new LocalMemoryCacheManager(1024, 64), + EnumSet.of(FileType.DATA), + 64); + Path tableRoot = new Path("oss://bucket/table"); + @SuppressWarnings("unchecked") + RemoteIterator<FileStatus> marker = mock(RemoteIterator.class); + when(delegate.listFilesIterative(tableRoot, true)).thenReturn(marker); + + assertThat(cachingIO.listFilesIterative(tableRoot, true)).isSameAs(marker); + verify(delegate).listFilesIterative(tableRoot, true); + // the interface default would construct its own iterator backed by listStatus + verify(delegate, never()).listStatus(any()); + } + private CachingFileIO newCachingFileIO( FileIO delegate, LocalCacheManager cache, EnumSet<FileType> whitelist, int blockSize) { return new CachingFileIO(delegate, cache, whitelist);
