[java][python] Read Python-planned splits with the native reader

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java

@@ -177,9 +177,9 @@ public CommitMessage doCompact(FileStoreTable table, String commitUser) throws E
         long minSequenceNumber = minSequenceId(compactBefore);
         long maxSequenceNumber = maxSequenceId(compactBefore);
         long nextRowId = firstRowId;
-        long[] columnMaxSequenceNumbers =
+        long[] writeColsSequences =
                 options.ignoreIndexColumnUpdate() && !writeResult.isEmpty()
-                        ? compactedColumnMaxSequenceNumbers(
+                        ? compactedWriteColsSequences(
                                 table,
                                 writeResult
                                         .get(0)
@@ -189,8 +189,8 @@ public CommitMessage doCompact(FileStoreTable table, String commitUser) throws E
             DataFileMeta dataFileMeta =
                     file.assignFirstRowId(nextRowId)
                             .assignSequenceNumber(minSequenceNumber, maxSequenceNumber);
-            if (columnMaxSequenceNumbers != null) {
-                dataFileMeta = dataFileMeta.withColumnMaxSequenceNumbers(columnMaxSequenceNumbers);
+            if (writeColsSequences != null) {
+                dataFileMeta = dataFileMeta.withWriteColsSequences(writeColsSequences);
             }
             compactAfter.add(dataFileMeta);
             nextRowId += dataFileMeta.rowCount();
@@ -230,8 +230,7 @@ List<Range> planOutputRanges(long targetFileSize) {
     }
 
     @Nullable
-    private long[] compactedColumnMaxSequenceNumbers(
-            FileStoreTable table, DataFileMeta outputFile) {
+    private long[] compactedWriteColsSequences(FileStoreTable table, DataFileMeta outputFile) {
         SchemaManager schemaManager = table.schemaManager();
         Map<Long, TableSchema> schemaCache = new HashMap<>();
         Function<Long, TableSchema> schemaLoader =
@@ -244,12 +243,12 @@ private long[] compactedColumnMaxSequenceNumbers(
                     fileFieldsCache.computeIfAbsent(
                             Pair.of(input.schemaId(), input.writeCols()),
                             key -> fileFields(schemaLoader, input));
-            long[] inputColumnSequences = input.columnMaxSequenceNumbers();
+            long[] inputWriteColsSequences = input.writeColsSequences();
             for (int inputPosition = 0; inputPosition < inputFields.size(); inputPosition++) {
                 fieldMaxSequences.merge(
                         inputFields.get(inputPosition).id(),
                         fieldMaxSequenceNumber(
-                                input, inputColumnSequences, inputPosition, inputFields.size()),
+                                input, inputWriteColsSequences, inputPosition, inputFields.size()),
                         Math::max);
             }
         }

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

@@ -255,15 +255,15 @@ private static void addIfUpdatesIndexedFields(
                             TableSchema fileSchema = schemaLoader.apply(file.schemaId());
                             return fileFields(fileSchema, file);
                         });
-        long[] columnSequences = file.columnMaxSequenceNumbers();
+        long[] writeColsSequences = file.writeColsSequences();
         long indexedMaxSequence = Long.MIN_VALUE;
         for (int position = 0; position < physicalFields.size(); position++) {
             if (indexedFieldIds.contains(physicalFields.get(position).id())) {
                 indexedMaxSequence =
                         Math.max(
                                 indexedMaxSequence,
                                 fieldMaxSequenceNumber(
-                                        file, columnSequences, position, physicalFields.size()));
+                                        file, writeColsSequences, position, physicalFields.size()));
             }
         }
         if (indexedMaxSequence != Long.MIN_VALUE) {

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -266,7 +266,7 @@ static DataFileMeta create(
             @Nullable String externalPath,
             @Nullable Long firstRowId,
             @Nullable List<String> writeCols,
-            @Nullable long[] columnMaxSequenceNumbers) {
+            @Nullable long[] writeColsSequences) {
         return new PojoDataFileMeta(
                 fileName,
                 fileSize,
@@ -288,7 +288,7 @@ static DataFileMeta create(
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     String fileName();
@@ -377,7 +377,7 @@ default Range nonNullRowIdRange() {
      * value otherwise. A null value means that only the file-level sequence range is available.
      */
     @Nullable
-    long[] columnMaxSequenceNumbers();
+    long[] writeColsSequences();
 
     DataFileMeta upgrade(int newLevel);
 
@@ -387,7 +387,7 @@ default Range nonNullRowIdRange() {
 
     DataFileMeta assignSequenceNumber(long minSequenceNumber, long maxSequenceNumber);
 
-    DataFileMeta withColumnMaxSequenceNumbers(long[] columnMaxSequenceNumbers);
+    DataFileMeta withWriteColsSequences(long[] writeColsSequences);
 
     DataFileMeta assignFirstRowId(long firstRowId);
 

paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java

@@ -43,7 +43,7 @@ public DataFileMetaSerializer() {
 
     @Override
     public InternalRow toRow(DataFileMeta meta) {
-        long[] columnMaxSequenceNumbers = meta.columnMaxSequenceNumbers();
+        long[] writeColsSequences = meta.writeColsSequences();
         return GenericRow.of(
                 BinaryString.fromString(meta.fileName()),
                 meta.fileSize(),
@@ -65,9 +65,7 @@ public InternalRow toRow(DataFileMeta meta) {
                 meta.externalPath().map(BinaryString::fromString).orElse(null),
                 meta.firstRowId(),
                 meta.writeCols() == null ? null : toStringArrayData(meta.writeCols()),
-                columnMaxSequenceNumbers == null
-                        ? null
-                        : new GenericArray(columnMaxSequenceNumbers));
+                writeColsSequences == null ? null : new GenericArray(writeColsSequences));
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java

@@ -78,7 +78,7 @@ public class PojoDataFileMeta implements DataFileMeta {
 
     private final @Nullable List<String> writeCols;
 
-    private final @Nullable long[] columnMaxSequenceNumbers;
+    private final @Nullable long[] writeColsSequences;
 
     public PojoDataFileMeta(
             String fileName,
@@ -101,7 +101,7 @@ public PojoDataFileMeta(
             @Nullable String externalPath,
             @Nullable Long firstRowId,
             @Nullable List<String> writeCols,
-            @Nullable long[] columnMaxSequenceNumbers) {
+            @Nullable long[] writeColsSequences) {
         this.fileName = fileName;
         this.fileSize = fileSize;
 
@@ -126,8 +126,7 @@ public PojoDataFileMeta(
         this.externalPath = externalPath;
         this.firstRowId = firstRowId;
         this.writeCols = writeCols;
-        this.columnMaxSequenceNumbers =
-                columnMaxSequenceNumbers == null ? null : columnMaxSequenceNumbers.clone();
+        this.writeColsSequences = writeColsSequences == null ? null : writeColsSequences.clone();
     }
 
     @Override
@@ -246,8 +245,8 @@ public List<String> writeCols() {
 
     @Nullable
     @Override
-    public long[] columnMaxSequenceNumbers() {
-        return columnMaxSequenceNumbers == null ? null : columnMaxSequenceNumbers.clone();
+    public long[] writeColsSequences() {
+        return writeColsSequences == null ? null : writeColsSequences.clone();
     }
 
     @Override
@@ -274,7 +273,7 @@ public PojoDataFileMeta upgrade(int newLevel) {
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -301,7 +300,7 @@ public PojoDataFileMeta rename(String newFileName) {
                 newExternalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -327,7 +326,7 @@ public PojoDataFileMeta copyWithoutStats() {
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -353,11 +352,11 @@ public PojoDataFileMeta assignSequenceNumber(long minSequenceNumber, long maxSeq
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
-    public PojoDataFileMeta withColumnMaxSequenceNumbers(long[] columnMaxSequenceNumbers) {
+    public PojoDataFileMeta withWriteColsSequences(long[] writeColsSequences) {
         return new PojoDataFileMeta(
                 fileName,
                 fileSize,
@@ -379,7 +378,7 @@ public PojoDataFileMeta withColumnMaxSequenceNumbers(long[] columnMaxSequenceNum
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -405,7 +404,7 @@ public PojoDataFileMeta assignFirstRowId(long firstRowId) {
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -431,7 +430,7 @@ public PojoDataFileMeta newFirstRowId(@Nullable Long newFirstRowId) {
                 externalPath,
                 newFirstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -457,7 +456,7 @@ public PojoDataFileMeta copy(List<String> newExtraFiles) {
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -483,7 +482,7 @@ public PojoDataFileMeta newExternalPath(String newExternalPath) {
                 newExternalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -509,7 +508,7 @@ public PojoDataFileMeta copy(byte[] newEmbeddedIndex) {
                 externalPath,
                 firstRowId,
                 writeCols,
-                columnMaxSequenceNumbers);
+                writeColsSequences);
     }
 
     @Override
@@ -541,7 +540,7 @@ public boolean equals(Object o) {
                 && Objects.equals(externalPath, that.externalPath().orElse(null))
                 && Objects.equals(firstRowId, that.firstRowId())
                 && Objects.equals(writeCols, that.writeCols())
-                && Arrays.equals(columnMaxSequenceNumbers, that.columnMaxSequenceNumbers());
+                && Arrays.equals(writeColsSequences, that.writeColsSequences());
     }
 
     @Override
@@ -567,7 +566,7 @@ public int hashCode() {
                 externalPath,
                 firstRowId,
                 writeCols,
-                Arrays.hashCode(columnMaxSequenceNumbers));
+                Arrays.hashCode(writeColsSequences));
     }
 
     @Override
@@ -578,7 +577,7 @@ public String toString() {
                         + "minSequenceNumber: %d, maxSequenceNumber: %d, "
                         + "schemaId: %d, level: %d, extraFiles: %s, creationTime: %s, "
                         + "deleteRowCount: %d, fileSource: %s, valueStatsCols: %s, externalPath: %s, "
-                        + "firstRowId: %s, writeCols: %s, columnMaxSequenceNumbers: %s}",
+                        + "firstRowId: %s, writeCols: %s, writeColsSequences: %s}",
                 fileName,
                 fileSize,
                 rowCount,
@@ -599,6 +598,6 @@ public String toString() {
                 externalPath,
                 firstRowId,
                 writeCols,
-                Arrays.toString(columnMaxSequenceNumbers));
+                Arrays.toString(writeColsSequences));
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java

@@ -256,7 +256,7 @@ public List<String> writeCols() {
 
     @Nullable
     @Override
-    public long[] columnMaxSequenceNumbers() {
+    public long[] writeColsSequences() {
         int position = requiredPosition(Fields.WRITE_COLS_SEQUENCES);
         InternalRow row = currentRow();
         if (row.isNullAt(position)) {
@@ -308,8 +308,8 @@ public DataFileMeta assignSequenceNumber(long minSequenceNumber, long maxSequenc
     }
 
     @Override
-    public DataFileMeta withColumnMaxSequenceNumbers(long[] columnMaxSequenceNumbers) {
-        throw unsupportedOperation("withColumnMaxSequenceNumbers(long[])");
+    public DataFileMeta withWriteColsSequences(long[] writeColsSequences) {
+        throw unsupportedOperation("withWriteColsSequences(long[])");
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java

@@ -291,13 +291,13 @@ private static List<DataField> fileFields(
     /** Returns the latest sequence known for a physical field position in the file. */
     public static long fieldMaxSequenceNumber(
             DataFileMeta file,
-            @Nullable long[] columnSequences,
+            @Nullable long[] writeColsSequences,
             int fieldPosition,
             int physicalFieldCount) {
-        if (columnSequences == null || columnSequences.length != physicalFieldCount) {
+        if (writeColsSequences == null || writeColsSequences.length != physicalFieldCount) {
             return file.maxSequenceNumber();
         }
-        return columnSequences[fieldPosition];
+        return writeColsSequences[fieldPosition];
     }
 
     /**