(paimon) branch master updated: [core] Route plain splits to the wrapped table read in FallbackReadFileStoreTable (#10820)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 7e6cb13a41 [core] Route plain splits to the wrapped table read in 
FallbackReadFileStoreTable (#10820)
7e6cb13a41 is described below

commit 7e6cb13a41a0b8a15b9540129504f2212347c80f
Author: Dapeng Sun(孙大鹏) <dapeng....@alibaba-inc.com>
AuthorDate: Tue Oct 27 14:16:00 2026 +0800

    [core] Route plain splits to the wrapped table read in 
FallbackReadFileStoreTable (#10820)
---
 .../paimon/table/FallbackReadFileStoreTable.java   |  5 +++-
 .../table/FallbackReadFileStoreTableTest.java      | 33 ++++++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
 
b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
index 46c7cfcfc2..02624408f2 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
@@ -718,6 +718,7 @@ public class FallbackReadFileStoreTable extends 
DelegatedFileStoreTable {
 
         private final InnerTableRead mainRead;
         private final InnerTableRead fallbackRead;
+        private final InnerTableRead wrappedRead;
         private final boolean fallbackReadFailFast;
 
         private Read() {
@@ -725,6 +726,7 @@ public class FallbackReadFileStoreTable extends 
DelegatedFileStoreTable {
             FileStoreTable second = wrappedFirst ? other : wrapped;
             this.mainRead = first.newRead();
             this.fallbackRead = second.newRead();
+            this.wrappedRead = wrappedFirst ? mainRead : fallbackRead;
             this.fallbackReadFailFast =
                     wrapped.coreOptions()
                             .toConfiguration()
@@ -800,7 +802,8 @@ public class FallbackReadFileStoreTable extends 
DelegatedFileStoreTable {
                 }
                 return mainRead.createReader(wrappedSplit);
             }
-            return mainRead.createReader(split);
+            // Plain splits come from newSnapshotReader(), which plans the 
wrapped table.
+            return wrappedRead.createReader(split);
         }
     }
 }
diff --git 
a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
 
b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
index 6b8727e0c6..5c161a1a70 100644
--- 
a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
+++ 
b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
@@ -38,6 +38,7 @@ 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.SchemaUtils;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.sink.StreamTableCommit;
@@ -538,6 +539,38 @@ public class FallbackReadFileStoreTableTest {
         assertThat(mergedPartitions).containsExactlyInAnyOrder(1, 2, 3);
     }
 
+    @Test
+    public void testReadPlainSplitWithPrimaryBranch() throws Exception {
+        String branchName = "bc";
+
+        FileStoreTable mainTable = createTable();
+        writeDataIntoTable(mainTable, 0, rowData(1, 10));
+
+        mainTable.createBranch(branchName);
+        FileStoreTable branchTable = createTableFromBranch(mainTable, 
branchName);
+        writeDataIntoTable(branchTable, 0, rowData(2, 20));
+
+        // evolve only the main branch, so its new files use a schema id the 
branch lacks
+        TableSchema newSchema =
+                new FileSystemSchemaManager(fileIO, tablePath)
+                        .commitChanges(SchemaChange.setOption("my-custom-key", 
"my-value"));
+        mainTable = mainTable.copy(newSchema);
+        writeDataIntoTable(mainTable, 1, rowData(3, 30));
+
+        FallbackReadFileStoreTable table =
+                new FallbackReadFileStoreTable(mainTable, branchTable, false);
+
+        // splits planned by newSnapshotReader() belong to the wrapped table
+        List<Pair<Integer, Integer>> result = new ArrayList<>();
+        for (Split split : table.newSnapshotReader().read().splits()) {
+            
assertThat(split).isNotInstanceOf(FallbackReadFileStoreTable.FallbackSplit.class);
+            RecordReader<InternalRow> reader = 
table.newRead().createReader(split);
+            reader.forEachRemaining(r -> result.add(Pair.of(r.getInt(0), 
r.getInt(1))));
+            reader.close();
+        }
+        assertThat(result).containsExactlyInAnyOrder(Pair.of(1, 10), 
Pair.of(3, 30));
+    }
+
     @Test
     public void testFallbackReadFailFastDefaultSwallowsException() throws 
Exception {
         FallbackReadFileStoreTable table = 
setUpTableWithThrowingFallback(false);