From b9ef0fdb58b3e79e1fe7f20d1519ce25e274f055 Mon Sep 17 00:00:00 2001 From: Christopher Wirt Date: Fri, 13 Nov 2026 11:53:01 +0000 Subject: [PATCH 1/2] feat(spark): add opt-in ordered reads and JMH comparison Signed-off-by: Christopher Wirt --- java/vortex-spark/README.md | 28 ++++++ .../spark/bench/OrderedScanBenchmark.java | 95 +++++++++++++++++++ .../spark/read/VortexPartitionReader.java | 2 +- 3 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 java/vortex-spark/common/src/jmh/java/dev/vortex/spark/bench/OrderedScanBenchmark.java diff --git a/java/vortex-spark/README.md b/java/vortex-spark/README.md index a662b62d677..c7ffa9119e5 100644 --- a/java/vortex-spark/README.md +++ b/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//`. 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. diff --git a/java/vortex-spark/common/src/jmh/java/dev/vortex/spark/bench/OrderedScanBenchmark.java b/java/vortex-spark/common/src/jmh/java/dev/vortex/spark/bench/OrderedScanBenchmark.java new file mode 100644 index 00000000000..ee1f727382b --- /dev/null +++ b/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; + } +} diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java index 2fd70f71202..a4341271030 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java +++ b/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()); } From bd142d0a6790cc0c3d2c250801bcd6683c90616b Mon Sep 17 00:00:00 2001 From: Christopher Wirt Date: Fri, 13 Nov 2026 16:24:24 +0000 Subject: [PATCH 2/2] test(spark): verify opt-in ordering across reader paths Signed-off-by: Christopher Wirt --- .../dev/vortex/spark/VortexFileSplitTest.java | 46 ++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFileSplitTest.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFileSplitTest.java index 7cdcf389f38..0fa402f8437 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFileSplitTest.java +++ b/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 read = read(output, mode, splitSize, Map.of("VORTEX.READ.ORDERED", "TrUe")); + for (boolean filtered : new boolean[] {false, true}) { + Dataset 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 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,6 +341,10 @@ private Path write(Dataset data, String name, String partitionColumn) { * {@code splitSize} bytes. */ private Dataset read(Path output, String mode, long splitSize) { + return read(output, mode, splitSize, Map.of()); + } + + private Dataset read(Path output, String mode, long splitSize, Map 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. @@ -308,7 +352,7 @@ private Dataset read(Path output, String mode, long splitSize) { 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 read = spark.read().format("vortex").load(output.toString()); + Dataset 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;