feat(spark): add opt-in ordered reads and JMH comparison

java/vortex-spark/README.md

@@ -65,6 +65,15 @@ Catalog tables read and write through the V1 file format;
 `spark.sql.sources.useV1SourceList=vortex` puts every path on it, which is what the `_metadata`
 column and dynamic partition pruning need.
 
+## Read ordering
+
+Set `.option("vortex.read.ordered", true)` to preserve a file's row order within each
+Spark file split. The default is `false`, allowing batches to arrive as decoding finishes.
+This does not establish ordering across Spark partitions or files.
+
+The option is case-insensitive and works with both V1 and V2 reads. Session defaults use
+`vortex.read.ordered`; cluster defaults can use `spark.datasource.vortex.read.ordered`.
+
 ## Read coalescing
 
 Nearby storage reads can be merged to reduce requests. Set the maximum gap and merged span in bytes:
@@ -122,5 +131,24 @@ cd java
 ./gradlew :vortex-spark-4.0_2.13:jmh -PjmhArgs="FooterReadBenchmark -f 1 -wi 2 -i 5"
 ```
 
+Compare ordered and unordered decoding on an existing Vortex file, running once for a
+current bid-hour event file and once for a current featoid bucket file:
+
+For performance measurements, first stage an optimized JNI library from the same revision
+under `vortex-jni/src/main/resources/native/<platform>/`. The environment setting below
+keeps Gradle from replacing it with its default debug build.
+
+```bash
+VORTEX_SKIP_MAKE_TEST_FILES=true ./gradlew :vortex-spark-4.0_2.13:jmh -PjmhArgs="OrderedScanBenchmark -p path=/data/events.vortex -p ordered=false,true -rf json -rff /tmp/events-ordering.json"
+VORTEX_SKIP_MAKE_TEST_FILES=true ./gradlew :vortex-spark-4.0_2.13:jmh -PjmhArgs="OrderedScanBenchmark -p path=/data/featoids.vortex -p ordered=false,true -rf json -rff /tmp/featoids-ordering.json"
+```
+
+Each invocation opens and fully decodes every column through `VortexPartitionReader`,
+without Spark planning, aggregation, filtering, or file splitting. Schema inference runs
+outside the measured region. Use the same files and machine for both modes; repeated scans
+can use warm storage caches. The setup prints the file size and column count alongside the
+JMH parameters. Ordered overhead is `(ordered / unordered - 1) * 100%`; compare the error
+intervals before deciding whether the cost is small enough for a caller to opt in.
+
 See the [Spark user guide](https://docs.vortex.dev/user-guide/spark.html) for supported types,
 options, partitioning, and remote storage details.

java/vortex-spark/common/src/jmh/java/dev/vortex/spark/bench/OrderedScanBenchmark.java

@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.spark.bench;
+
+import dev.vortex.spark.VortexOptions;
+import dev.vortex.spark.io.VortexIo;
+import dev.vortex.spark.read.VortexFooterReader;
+import dev.vortex.spark.read.VortexPartitionReader;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.paths.SparkPath;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.execution.datasources.PartitionedFile;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.types.StructType;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/** Compares ordered and unordered decoding of every column of an existing, unsplit Vortex file. */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@Fork(2)
+public class OrderedScanBenchmark {
+    /** The same representative store file must be supplied for both modes with {@code -p path=...}. */
+    @Param({""})
+    private String path;
+
+    @Param({"false", "true"})
+    private boolean ordered;
+
+    private VortexOptions options;
+    private VortexIo io;
+    private StructType schema;
+    private PartitionedFile file;
+
+    @Setup(Level.Trial)
+    public void setUp() throws IOException {
+        if (path.isBlank()) {
+            throw new IllegalArgumentException("Supply a representative Vortex file with -p path=...");
+        }
+        Configuration conf = new Configuration();
+        Path input = new Path(path);
+        var status = input.getFileSystem(conf).getFileStatus(input);
+        if (!status.isFile()) {
+            throw new IllegalArgumentException("path must name one Vortex file: " + path);
+        }
+        options = VortexOptions.of(Map.of("vortex.read.ordered", Boolean.toString(ordered)));
+        io = VortexIo.create(options, conf);
+        schema = VortexFooterReader.inferSchema(List.of(status), io, options, true);
+        file = new PartitionedFile(
+                InternalRow.empty(),
+                SparkPath.fromPathString(status.getPath().toString()),
+                0,
+                status.getLen(),
+                new String[0],
+                status.getModificationTime(),
+                status.getLen(),
+                scala.collection.immutable.Map$.MODULE$.empty());
+        System.out.printf("Input: %s, bytes: %d, columns: %d%n", path, status.getLen(), schema.length());
+    }
+
+    /** Loading each batch materializes all columns as Arrow; no row-count or projection shortcut is used. */
+    @Benchmark
+    public long fullScan(Blackhole blackhole) throws IOException {
+        long rows = 0;
+        try (VortexPartitionReader reader =
+                new VortexPartitionReader(file, schema, schema, new StructType(), io, options, new Filter[0], true)) {
+            while (reader.next()) {
+                var batch = reader.get();
+                rows += batch.numRows();
+                blackhole.consume(batch);
+            }
+        }
+        return rows;
+    }
+}

java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java

@@ -135,7 +135,7 @@ public VortexPartitionReader(
             ReadTypes.check(
                     projectedSchema, arrowFields, caseSensitive, file.toPath().toString());
 
-            var options = ScanOptions.builder();
+            var options = ScanOptions.builder().ordered(formatOptions.getBoolean("vortex.read.ordered", false));
             if (io.splitConcurrency() > 0) {
                 options.splitConcurrency(io.splitConcurrency());
             }

java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFileSplitTest.java

@@ -13,6 +13,7 @@
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Stream;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -114,6 +115,45 @@ void everySplitSizeReadsEachRowOnce(String mode) {
                 read.filter("amount > 400000 and name is not null"));
     }
 
+    @ParameterizedTest(name = "{0}")
+    @ValueSource(strings = {"v2", "v1", "v1-rows"})
+    @DisplayName("Ordered reads preserve row order within each file split, including with a pushed filter")
+    void orderedReadsPreserveRowOrder(String mode) {
+        Path output = write(rows(ROWS).coalesce(1), "ordered", null);
+        long fileSize = fileSizes(output).get(0);
+        for (long splitSize : new long[] {fileSize + 1, fileSize / 2 + 1}) {
+            Dataset<Row> read = read(output, mode, splitSize, Map.of("VORTEX.READ.ORDERED", "TrUe"));
+            for (boolean filtered : new boolean[] {false, true}) {
+                Dataset<Row> selected = filtered ? read.filter("id >= 1000") : read;
+                String context = "mode=" + mode + " splitSize=" + splitSize + " filtered=" + filtered;
+                if (filtered) {
+                    String plan = selected.queryExecution().executedPlan().toString();
+                    assertTrue(plan.contains("GreaterThanOrEqual(id,1000)"), plan);
+                }
+                List<Long> counts = selected.javaRDD()
+                        .mapPartitions(partition -> {
+                            long previous = -1;
+                            long count = 0;
+                            while (partition.hasNext()) {
+                                long id = partition.next().getLong(0);
+                                if (id <= previous) {
+                                    throw new IllegalStateException(context + ": " + id + " follows " + previous);
+                                }
+                                previous = id;
+                                count++;
+                            }
+                            return List.of(count).iterator();
+                        })
+                        .collect();
+                assertEquals(expectedPartitions(List.of(fileSize), splitSize), counts.size(), context);
+                assertEquals(
+                        filtered ? ROWS - 1000 : ROWS,
+                        counts.stream().mapToLong(Long::longValue).sum(),
+                        context);
+            }
+        }
+    }
+
     @ParameterizedTest(name = "{0}")
     @ValueSource(strings = {"v2", "v1", "v1-rows"})
     @DisplayName("Each split reads its own contiguous rows, also when one task reads several splits of a file")
@@ -301,14 +341,18 @@ private Path write(Dataset<Row> data, String name, String partitionColumn) {
      * {@code splitSize} bytes.
      */
     private Dataset<Row> read(Path output, String mode, long splitSize) {
+        return read(output, mode, splitSize, Map.of());
+    }
+
+    private Dataset<Row> read(Path output, String mode, long splitSize, Map<String, String> options) {
         boolean v1 = !mode.equals("v2");
         spark.conf().set("spark.sql.sources.useV1SourceList", v1 ? "vortex" : "");
         // Whole-stage codegen off makes `FileSourceScanExec` ask the format for rows, not batches.
         spark.conf().set("spark.sql.codegen.wholeStage", !mode.equals("v1-rows"));
         spark.conf().set("spark.sql.files.maxPartitionBytes", splitSize);
         spark.conf().set("spark.sql.files.openCostInBytes", 0);
         spark.conf().set("spark.sql.files.minPartitionNum", 1);
-        Dataset<Row> read = spark.read().format("vortex").load(output.toString());
+        Dataset<Row> read = spark.read().format("vortex").options(options).load(output.toString());
         String plan = read.queryExecution().executedPlan().toString();
         assertTrue(plan.contains(v1 ? "FileScan vortex" : "BatchScan"), plan);
         return read;