From fbebd5a051bf244d7ddfcb01bfbf03dec31e8f32 Mon Sep 17 00:00:00 2001
From: huweihua <huweihua.ckl@gmail.com>
Date: Mon, 3 Aug 2026 19:35:40 +0800
Subject: [PATCH] [vortex] Preserve physical row order in scans

Vortex scans are unordered by default, while Paimon derives returned positions monotonically. Request ordered scans so dedicated Vortex files stay aligned with their corresponding main-file rows.

Generated-by: OpenAI Codex
---
 .../format/vortex/VortexRecordsReader.java    |  3 +-
 .../format/vortex/VortexReaderWriterTest.java | 73 +++++++++++++++++++
 2 files changed, 75 insertions(+), 1 deletion(-)

diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java
index 2d3cfcbbc816..c41ac27e42dd 100644
--- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java
+++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexRecordsReader.java
@@ -87,7 +87,8 @@ public VortexRecordsReader(
             try {
                 this.dataSource = DataSource.open(session, path.toUri().toString(), storageOptions);
                 try {
-                    ImmutableScanOptions.Builder scanBuilder = ImmutableScanOptions.builder();
+                    ImmutableScanOptions.Builder scanBuilder =
+                            ImmutableScanOptions.builder().ordered(true);
 
                     java.util.List<String> columns = physicalReadRowType.getFieldNames();
                     scanBuilder.projection(
diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java
index 7ccca17c09e6..e97e3cf80fcc 100644
--- a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java
+++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java
@@ -44,6 +44,7 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.RoaringBitmap32;
 
+import dev.vortex.jni.NativeRuntime;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -597,4 +598,76 @@ public void testReturnedPositionSequential(@TempDir java.nio.file.Path tempDir)
             assertEquals(5, expectedPos, "Should have read exactly 5 rows");
         }
     }
+
+    @Test
+    public void testReturnedPositionWithMultipleScanPartitions(@TempDir java.nio.file.Path tempDir)
+            throws Exception {
+        RowType rowType =
+                RowType.builder()
+                        .field("id", DataTypes.INT())
+                        .field("payload", DataTypes.STRING())
+                        .build();
+        VortexFileFormat format =
+                new VortexFileFormatFactory()
+                        .create(new FileFormatFactory.FormatContext(new Options(), 1024, 1024));
+
+        FileIO fileIO = new LocalFileIO();
+        Path testFile =
+                new Path(new Path(tempDir.toUri()), "test_ordered_scan_" + UUID.randomUUID());
+
+        // Create multiple scan tasks and make the first one slower, so an unordered scan returns
+        // a later task first.
+        int firstRangeRowCount = 1_024;
+        int lastSelectedRow = 4_999;
+        try (FormatWriter writer =
+                ((SupportsDirectWrite) format.createWriterFactory(rowType))
+                        .create(fileIO, testFile, "")) {
+            for (int i = 0; i <= lastSelectedRow; i++) {
+                String payload = i < firstRangeRowCount ? payload(i) : "x";
+                writer.addElement(GenericRow.of(i, BinaryString.fromString(payload)));
+            }
+        }
+
+        long[] selectedRows = new long[firstRangeRowCount + 1];
+        for (int i = 0; i < firstRangeRowCount; i++) {
+            selectedRows[i] = i;
+        }
+        selectedRows[firstRangeRowCount] = lastSelectedRow;
+
+        int previousWorkerCount = NativeRuntime.workerCount();
+        NativeRuntime.setWorkerThreads(2);
+        try {
+            try (VortexRecordsReader reader =
+                    new VortexRecordsReader(
+                            testFile,
+                            rowType,
+                            rowType,
+                            selectedRows,
+                            null,
+                            Collections.emptyMap())) {
+                int readCount = 0;
+                FileRecordIterator<InternalRow> batch;
+                while ((batch = reader.readBatch()) != null) {
+                    InternalRow row;
+                    while ((row = batch.next()) != null) {
+                        assertEquals(batch.returnedPosition(), row.getInt(0));
+                        readCount++;
+                    }
+                }
+                assertEquals(selectedRows.length, readCount);
+            }
+        } finally {
+            NativeRuntime.setWorkerThreads(previousWorkerCount);
+        }
+    }
+
+    private static String payload(int rowId) {
+        char[] chars = new char[4_096];
+        int state = rowId + 1;
+        for (int i = 0; i < chars.length; i++) {
+            state = state * 1_103_515_245 + 12_345;
+            chars[i] = (char) ('a' + ((state >>> 16) & 15));
+        }
+        return new String(chars);
+    }
 }