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 4a05a83396 [vortex] Preserve physical row order in scans (#9003)
4a05a83396 is described below
commit 4a05a833966346d947f19a6277b3b134924d9f7b
Author: Weihua Hu <[email protected]>
AuthorDate: Tue Aug 4 11:06:19 2026 +0800
[vortex] Preserve physical row order in scans (#9003)
---
.../paimon/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 2d3cfcbbc8..c41ac27e42 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 class VortexRecordsReader implements
FileRecordReader<InternalRow> {
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 7ccca17c09..e97e3cf80f 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.DataTypes;
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 class VortexReaderWriterTest {
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);
+ }
}
