[core][WIP] Optimize Parquet point lookup I/O on OSS

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -84,6 +84,15 @@ default void setRuntimeContext(Map<String, String> options) {}
      */
     SeekableInputStream newInputStream(Path path) throws IOException;
 
+    /**
+     * Opens a stream for a file whose length is already known, for example from a manifest.
+     * Implementations may avoid a separate file-status request and optimize positioned reads. The
+     * caller must supply the length of the file being opened.
+     */
+    default SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        return newInputStream(path);
+    }
+
     /**
      * Opens an PositionOutputStream at the indicated Path.
      *

paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java

@@ -58,6 +58,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException {
         return wrap(() -> fileIO(path).newInputStream(path));
     }
 
+    @Override
+    public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        return wrap(() -> fileIO(path).newInputStream(path, fileSize));
+    }
+
     @Override
     public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
         return wrap(() -> fileIO(path).newOutputStream(path, overwrite));

paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java

@@ -74,6 +74,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException {
         return wrap(() -> fileIO(path).newInputStream(path));
     }
 
+    @Override
+    public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        return wrap(() -> fileIO(path).newInputStream(path, fileSize));
+    }
+
     @Override
     public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
         return wrap(() -> fileIO(path).newOutputStream(path, overwrite));

paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java

@@ -222,7 +222,8 @@ private static void copyToFileRanges(
         }
     }
 
-    private static byte[] getOrCreateBuffer(FileRange range) {
+    /** Return a range's caller-provided buffer, allocating it lazily when necessary. */
+    public static byte[] getOrCreateBuffer(FileRange range) {
         if (range instanceof FileRange.FileRangeImpl) {
             return ((FileRange.FileRangeImpl) range).getOrCreateBuffer();
         }
@@ -258,7 +259,8 @@ private static void copyMultiBytesToBytes(
         }
     }
 
-    private static List<? extends FileRange> validateAndSortRanges(
+    /** Validate ranges before dispatching them to multiple readers or caches. */
+    public static List<? extends FileRange> validateAndSortRanges(
             final List<? extends FileRange> input) throws EOFException {
         requireNonNull(input, "Null input list");
         checkArgument(!input.isEmpty(), "Empty input list");

paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java

@@ -150,6 +150,15 @@ public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws
         return delegate.newOutputStream(path, overwrite);
     }
 
+    @Override
+    public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        if (!whitelist.contains(FileType.classify(path)) || FileType.isMutable(path)) {
+            return delegate.newInputStream(path, fileSize);
+        }
+        // Keep the existing cache namespace and mutable-file validation semantics.
+        return newInputStream(path);
+    }
+
     @Override
     public FileStatus getFileStatus(Path path) throws IOException {
         return delegate.getFileStatus(path);

paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -131,6 +131,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException {
         return fileIO().newInputStream(path);
     }
 
+    @Override
+    public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        return fileIO().newInputStream(path, fileSize);
+    }
+
     @Override
     public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
         return fileIO().newOutputStream(path, overwrite);

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java

@@ -125,7 +125,8 @@ private DataEvolutionGlobalIndexScanner(
                         partitionFilter,
                         coverageIndexFiles,
                         table.coreOptions().scalarIndexSearchMode());
-        GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath());
+        GlobalIndexFileReader indexFileReader =
+                meta -> fileIO.newInputStream(meta.filePath(), meta.fileSize());
         Map<Integer, IndexMetaFileGroup> indexMetas = new HashMap<>();
         Map<Integer, List<IndexMetaFileGroup>> extraIndexMetas = new HashMap<>();
         for (IndexFileMeta indexFile : indexFiles) {

paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java

@@ -138,6 +138,7 @@ public DataEvolutionSplitRead(
             FileStorePathFactory pathFactory) {
         this.fileIO = fileIO;
         final Map<Long, TableSchema> cache = new HashMap<>();
+        cache.put(schema.id(), schema);
         this.schemaFetcher =
                 schemaId -> cache.computeIfAbsent(schemaId, key -> schemaManager.schema(schemaId));
         this.schema = schema;

paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java

@@ -32,9 +32,11 @@
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.FileSystemSchemaManager;
 import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.source.DataSplit;
@@ -45,12 +47,15 @@
 
 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.ValueSource;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -63,6 +68,91 @@ class DataEvolutionSplitReadTest {
 
     @TempDir java.nio.file.Path tempDir;
 
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void testPredicateReusesCurrentSchemaAndLoadsOlderSchema(boolean evolved) throws Exception {
+        LocalFileIO fileIO = new LocalFileIO();
+        Path tableRoot = new Path(tempDir.toUri().toString());
+        CoreOptions options = new CoreOptions(new Options());
+        FileStorePathFactory paths =
+                new FileStorePathFactory(
+                        tableRoot,
+                        RowType.of(),
+                        options.partitionDefaultName(),
+                        "parquet",
+                        CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+                        CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+                        CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+                        CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+                        CoreOptions.FILE_COMPRESSION.defaultValue(),
+                        null,
+                        null,
+                        CoreOptions.ExternalPathStrategy.NONE,
+                        null,
+                        false,
+                        null);
+        SchemaManager manager = new FileSystemSchemaManager(fileIO, tableRoot);
+        TableSchema original =
+                manager.createTable(
+                        Schema.newBuilder()
+                                .column("f0", DataTypes.INT())
+                                .column("f1", DataTypes.STRING())
+                                .build());
+        TableSchema current =
+                evolved
+                        ? manager.commitChanges(SchemaChange.addColumn("extra", DataTypes.INT()))
+                        : original;
+        Path bucket = paths.bucketPath(EMPTY_ROW, 0);
+        fileIO.mkdirs(bucket);
+        Path file = new Path(bucket, "data-0.parquet");
+        writeFormatFile(fileIO, file, original.logicalRowType(), 100, "parquet");
+        AtomicInteger fetched = new AtomicInteger();
+        SchemaManager readManager =
+                new FileSystemSchemaManager(fileIO, tableRoot) {
+                    @Override
+                    public TableSchema schema(long id) {
+                        fetched.incrementAndGet();
+                        return super.schema(id);
+                    }
+                };
+        DataSplit split =
+                DataSplit.builder()
+                        .withPartition(EMPTY_ROW)
+                        .withBucket(0)
+                        .withBucketPath(bucket.toString())
+                        .withDataFiles(
+                                Collections.singletonList(
+                                        createFile(
+                                                "data-0.parquet",
+                                                fileIO.getFileSize(file),
+                                                10,
+                                                100,
+                                                1)))
+                        .rawConvertible(false)
+                        .build();
+        DataEvolutionSplitRead read =
+                new DataEvolutionSplitRead(
+                        fileIO, readManager, current, current.logicalRowType(), options, paths);
+        read.withFilter(new PredicateBuilder(current.logicalRowType()).greaterOrEqual(0, 1000));
+        List<Integer> actual = new ArrayList<>();
+        try (RecordReader<InternalRow> reader =
+                read.createReader(
+                        new IndexedSplit(
+                                split,
+                                Arrays.asList(new Range(12, 12), new Range(42, 42)),
+                                null))) {
+            reader.forEachRemaining(
+                    row -> {
+                        actual.add(row.getInt(0));
+                        if (evolved) {
+                            assertTrue(row.isNullAt(2));
+                        }
+                    });
+        }
+        assertEquals(Arrays.asList(1002, 1032), actual);
+        assertEquals(evolved ? 1 : 0, fetched.get());
+    }
+
     @Test
     public void testDifferentRowIdRange() {
         DataFileMeta file1 = createFile("file1.parquet", 1L, 100, 10);

paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.HadoopOptionsProvider;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.fs.TwoPhaseOutputStream;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.utils.IOUtils;
@@ -236,6 +237,24 @@ protected AliyunOSSFileSystem createFileSystem(org.apache.hadoop.fs.Path path) {
         }
     }
 
+    @Override
+    public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException {
+        URI uri = path.toUri();
+        if (fileSize < 0 || !"oss".equals(uri.getScheme()) || uri.getHost() == null) {
+            return super.newInputStream(path);
+        }
+        try {
+            return new OSSRangeInputStream(
+                    ossClient(path),
+                    uri.getHost(),
+                    uri.getPath().substring(1),
+                    fileSize,
+                    FileSystem.getStatistics("oss", AliyunOSSFileSystem.class));
+        } catch (Exception e) {
+            throw new IOException("Failed to open OSS file " + path, e);
+        }
+    }
+
     @Override
     public boolean tryToWriteAtomic(Path path, String content) throws IOException {
         URI uri = path.toUri();