From bd9c5ceb8b6fbcd5cdd89d7c12dce0b1b87debd0 Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Wed, 16 Sep 2026 09:19:44 +0900 Subject: [PATCH] [common] Delegate listFilesIterative through the remaining FileIO wrappers The three wrappers did not forward listFilesIterative, so the interface default ran on the wrapper and listed each directory with listStatus instead of using the inner object-store FileIO's flat paginated listing. Forward it like #9071 did for RESTTokenFileIO and #9034 did for tryToWriteAtomic, with a mock test per wrapper. Generated-by: Claude Code --- .../org/apache/paimon/fs/PluginFileIO.java | 8 ++++++ .../org/apache/paimon/fs/ResolvingFileIO.java | 8 ++++++ .../apache/paimon/fs/cache/CachingFileIO.java | 9 +++++++ .../apache/paimon/fs/PluginFileIOTest.java | 27 +++++++++++++++++++ .../apache/paimon/fs/ResolvingFileIOTest.java | 20 ++++++++++++++ .../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 587c1f2d4423..b1ffb43a9c91 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 boolean tryToWriteAtomic(Path path, String content) throws IOException { return wrap(() -> fileIO(path).tryToWriteAtomic(path, content)); } + @Override + public RemoteIterator 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 cc3aba497e65..5091acaaff59 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 TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite return wrap(() -> fileIO(path).newTwoPhaseOutputStream(path, overwrite)); } + @Override + public RemoteIterator 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 65eeaa3ebfd7..d943abcfbcf0 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.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 boolean tryToWriteAtomic(Path path, String content) throws IOException { return delegate.tryToWriteAtomic(path, content); } + @Override + public RemoteIterator 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 2411cce3d030..2be00e899e78 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.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 @@ void testCreateBlobPresignedUrlUsesPluginClassLoader() throws IOException { 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 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 e5203833aa27..6330d68daafd 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 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 void testNewTwoPhaseOutputStreamReachesResolvedOverride() throws IOExcept // 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 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 d3ae9d06633e..b4f3c46373dd 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.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 @@ void testTryToWriteAtomicReachesDelegateOverride() throws IOException { 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 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 whitelist, int blockSize) { return new CachingFileIO(delegate, cache, whitelist);