[common] Delegate listFilesIterative through the remaining FileIO wrappers

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<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 {

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<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 {

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<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 {

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<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;

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<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());
+    }
 }

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<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);