From b5445c1bba3a7cdb97b9a2ec3e3d341663cfb28f Mon Sep 17 00:00:00 2001 From: mprammer Date: Tue, 23 Jun 2026 22:24:50 -0400 Subject: [PATCH 1/4] java: add vortex-jni JMH read-boundary benchmark lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New `vortex-jni-bench` module (JMH) that stresses the vortex-jni read boundary — JNI plus the Arrow C Data Interface — which is the path an Iceberg FormatModel takes to read Vortex from the JVM. Three query shapes (full scan, projection, selective filter) over a synthetic six-column table, consumed column-at-a-time so the numbers reflect format/boundary cost rather than per-row JVM allocation. Includes a batch-granularity diagnostic (Vortex coalesces to ~64K-row read batches regardless of write chunk) and a README with run instructions. Must run against a --release native lib (VORTEX_SKIP_MAKE_TEST_FILES=true to preserve it). v2 TODO: a native Rust criterion read of the same file as a floor, to quote boundary overhead vs native. Co-Authored-By: Claude Signed-off-by: mprammer --- java/settings.gradle.kts | 3 + java/vortex-jni-bench/README.md | 50 ++++ java/vortex-jni-bench/build.gradle.kts | 29 ++ .../vortex/bench/VortexJniReadBenchmark.java | 281 ++++++++++++++++++ 4 files changed, 363 insertions(+) create mode 100644 java/vortex-jni-bench/README.md create mode 100644 java/vortex-jni-bench/build.gradle.kts create mode 100644 java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java diff --git a/java/settings.gradle.kts b/java/settings.gradle.kts index a601cfa9488..8cbf97feb04 100644 --- a/java/settings.gradle.kts +++ b/java/settings.gradle.kts @@ -19,6 +19,9 @@ rootProject.name = "vortex-root" // API bindings include("vortex-jni") + +// Benchmarks +include("vortex-jni-bench") include("vortex-spark_2.12") project(":vortex-spark_2.12").projectDir = file("vortex-spark") diff --git a/java/vortex-jni-bench/README.md b/java/vortex-jni-bench/README.md new file mode 100644 index 00000000000..ad065bf9b78 --- /dev/null +++ b/java/vortex-jni-bench/README.md @@ -0,0 +1,50 @@ +# vortex-jni-bench + +JMH microbenchmarks that stress the **vortex-jni read boundary** — JNI plus the Arrow C Data +Interface — which is the path an Iceberg `FormatModel` takes to read Vortex from the JVM. + +`VortexJniReadBenchmark` writes a synthetic six-column table (2M rows: 2× int64, 2× float64, +2× Utf8View) and reads it back three ways, consuming columns at the buffer level (numeric sums / +null counts) so the numbers reflect format + boundary cost, not per-row Java allocation: + +- `fullScan` — read all six columns. +- `projection` — read two of six (projection pushdown). +- `selectiveFilter` — `cat = 'alpha'` (~1/16 selectivity; filter pushdown). + +`ScanOptions` has no read-batch knob, and Vortex coalesces to ~64K-row read batches regardless of +the writer's chunk size, so the boundary is amortized over large batches by construction. Run the +`main` method to see that batch-granularity diagnostic. + +## Running + +The benchmark **must** run against a `--release` native lib (the dev `makeTestFiles` task builds a +debug lib, which would make the numbers meaningless). Build it once and drop it into vortex-jni's +resources, then run with `VORTEX_SKIP_MAKE_TEST_FILES=true` so the debug rebuild doesn't clobber it: + +```bash +# from repo root: build the release cdylib +cargo build --release -p vortex-jni +# place it for the host arch (example: macOS arm64) +cp target/release/libvortex_jni.dylib \ + java/vortex-jni/src/main/resources/native/darwin-aarch64/ + +cd java +VORTEX_SKIP_MAKE_TEST_FILES=true ./gradlew :vortex-jni-bench:jmh +``` + +Results land in `vortex-jni-bench/build/results/jmh/results.txt`. The JMH fork adds Arrow's +`--add-opens` flags via `@Fork(jvmArgsAppend=...)`. + +Batch-granularity diagnostic: + +```bash +VORTEX_SKIP_MAKE_TEST_FILES=true ./gradlew :vortex-jni-bench:jmhJar +java --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED \ + -cp vortex-jni-bench/build/libs/*-jmh.jar dev.vortex.bench.VortexJniReadBenchmark +``` + +## TODO (v2) + +These benchmarks measure absolute throughput and pushdown effectiveness *through* the boundary, not +the boundary's *overhead*. To quote a "().configureEach { enabled = false } + +dependencies { + jmhImplementation(platform(libs.netty.bom)) + jmhImplementation(project(":vortex-jni")) + jmhImplementation(libs.arrow.c.data) + jmhImplementation(libs.arrow.memory.core) + jmhImplementation(libs.arrow.memory.netty) +} + +jmh { + jmhVersion.set("1.37") +} diff --git a/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java new file mode 100644 index 00000000000..ab3b03ffd7d --- /dev/null +++ b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java @@ -0,0 +1,281 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.bench; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import dev.vortex.api.DataSource; +import dev.vortex.api.Expression; +import dev.vortex.api.Partition; +import dev.vortex.api.Scan; +import dev.vortex.api.ScanOptions; +import dev.vortex.api.Session; +import dev.vortex.api.VortexWriter; +import dev.vortex.jni.NativeLoader; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ViewVarCharVector; +import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +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.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +/** + * Measures read throughput through the vortex-jni boundary (JNI + the Arrow C Data Interface). + * + *

Three query shapes — full scan, projection, and a selective filter — over a synthetic six-column table. Rows are + * consumed column-at-a-time (numeric sums, null counts) rather than into per-row Java objects, so the numbers reflect + * the format/boundary cost rather than JVM allocation. + * + *

Note on batch size: {@code ScanOptions} exposes no read-batch knob, and Vortex coalesces to ~64K-row read batches + * regardless of the writer's chunk size (see {@link #main}), so the boundary cost is already amortized over large + * batches by construction — there is no small-batch regime to sweep from the public API. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 2) +@Fork( + value = 1, + jvmArgsAppend = { + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED" + }) +@State(Scope.Benchmark) +public class VortexJniReadBenchmark { + + static final long ROWS = 2_000_000L; + static final int WRITE_CHUNK = 65536; + static final String[] CATS = { + "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", + "india", "juliet", "kilo", "lima", "mike", "november", "oscar", "papa" + }; + + BufferAllocator allocator; + Session session; + DataSource dataSource; + Path file; + + @Setup(Level.Trial) + public void setup() throws Exception { + NativeLoader.loadJni(); + allocator = new RootAllocator(Long.MAX_VALUE); + session = Session.create(); + file = Files.createTempFile("vortex-jni-bench-", ".vortex"); + Files.deleteIfExists(file); + String uri = file.toAbsolutePath().toUri().toString(); + writeTable(session, allocator, uri, WRITE_CHUNK); + dataSource = DataSource.open(session, uri); + } + + @TearDown(Level.Trial) + public void teardown() throws Exception { + // Intentionally does not close the allocator: DataSource/Scan native resources are released by VortexCleaner + // at GC time, which races an explicit allocator.close() and trips leak detection. The JMH fork exits after the + // trial and reclaims everything; we only remove the temp file. + dataSource = null; + if (file != null) { + Files.deleteIfExists(file); + } + } + + private static Schema schema() { + return new Schema(List.of( + Field.notNullable("id", new ArrowType.Int(64, true)), + Field.notNullable("x", new ArrowType.Int(64, true)), + Field.notNullable("y", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.notNullable("z", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.nullable("cat", ArrowType.Utf8View.INSTANCE), + Field.nullable("tag", ArrowType.Utf8View.INSTANCE))); + } + + private static void writeTable(Session session, BufferAllocator allocator, String uri, int chunk) throws Exception { + Schema schema = schema(); + Random rnd = new Random(42); + try (VortexWriter writer = VortexWriter.create(session, uri, schema, new HashMap<>(), allocator); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) { + BigIntVector id = (BigIntVector) root.getVector("id"); + BigIntVector x = (BigIntVector) root.getVector("x"); + Float8Vector y = (Float8Vector) root.getVector("y"); + Float8Vector z = (Float8Vector) root.getVector("z"); + ViewVarCharVector cat = (ViewVarCharVector) root.getVector("cat"); + ViewVarCharVector tag = (ViewVarCharVector) root.getVector("tag"); + + long written = 0; + while (written < ROWS) { + int batch = (int) Math.min(chunk, ROWS - written); + for (FieldVector v : root.getFieldVectors()) { + v.reset(); + } + for (int i = 0; i < batch; i++) { + long r = written + i; + id.setSafe(i, r); + x.setSafe(i, rnd.nextInt(1_000_000)); + y.setSafe(i, rnd.nextDouble()); + z.setSafe(i, rnd.nextDouble()); + cat.setSafe(i, CATS[(int) (r % CATS.length)].getBytes(UTF_8)); + tag.setSafe(i, Long.toString(r).getBytes(UTF_8)); + } + root.setRowCount(batch); + try (ArrowArray arr = ArrowArray.allocateNew(allocator); + ArrowSchema sch = ArrowSchema.allocateNew(allocator)) { + Data.exportVectorSchemaRoot(allocator, root, null, arr, sch); + writer.writeBatch(arr.memoryAddress(), sch.memoryAddress()); + } + written += batch; + } + } + } + + @Benchmark + public void fullScan(Blackhole bh) throws Exception { + long sumId = 0; + long sumX = 0; + double sumY = 0; + long catNonNull = 0; + Scan scan = dataSource.scan(ScanOptions.of()); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(allocator)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot r = reader.getVectorSchemaRoot(); + int rows = r.getRowCount(); + BigIntVector id = (BigIntVector) r.getVector("id"); + BigIntVector x = (BigIntVector) r.getVector("x"); + Float8Vector y = (Float8Vector) r.getVector("y"); + FieldVector cat = r.getVector("cat"); + for (int i = 0; i < rows; i++) { + sumId += id.get(i); + sumX += x.get(i); + sumY += y.get(i); + if (!cat.isNull(i)) { + catNonNull++; + } + } + } + } + } + bh.consume(sumId); + bh.consume(sumX); + bh.consume(sumY); + bh.consume(catNonNull); + } + + @Benchmark + public void projection(Blackhole bh) throws Exception { + Expression projection = Expression.select(new String[] {"id", "y"}, Expression.root()); + ScanOptions options = ScanOptions.builder().projection(projection).build(); + long sumId = 0; + double sumY = 0; + Scan scan = dataSource.scan(options); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(allocator)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot r = reader.getVectorSchemaRoot(); + int rows = r.getRowCount(); + BigIntVector id = (BigIntVector) r.getVector("id"); + Float8Vector y = (Float8Vector) r.getVector("y"); + for (int i = 0; i < rows; i++) { + sumId += id.get(i); + sumY += y.get(i); + } + } + } + } + bh.consume(sumId); + bh.consume(sumY); + } + + @Benchmark + public void selectiveFilter(Blackhole bh) throws Exception { + Expression filter = + Expression.binary(Expression.BinaryOp.EQ, Expression.column("cat"), Expression.literal(CATS[0])); + ScanOptions options = ScanOptions.builder().filter(filter).build(); + long matched = 0; + long sumId = 0; + Scan scan = dataSource.scan(options); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(allocator)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot r = reader.getVectorSchemaRoot(); + int rows = r.getRowCount(); + BigIntVector id = (BigIntVector) r.getVector("id"); + for (int i = 0; i < rows; i++) { + sumId += id.get(i); + matched++; + } + } + } + } + bh.consume(matched); + bh.consume(sumId); + } + + /** + * Diagnostic (not a benchmark): prints the distribution of read batch row counts for a few writer chunk sizes, to + * show that Vortex coalesces to a stable read-batch granularity independent of how the file was written. + */ + public static void main(String[] args) throws Exception { + NativeLoader.loadJni(); + for (int chunk : new int[] {8192, 131072}) { + BufferAllocator alloc = new RootAllocator(Long.MAX_VALUE); + Session sess = Session.create(); + Path f = Files.createTempFile("vortex-jni-diag-" + chunk + "-", ".vortex"); + Files.deleteIfExists(f); + String uri = f.toAbsolutePath().toUri().toString(); + writeTable(sess, alloc, uri, chunk); + DataSource ds = DataSource.open(sess, uri); + long batches = 0; + long rowsSeen = 0; + long minRows = Long.MAX_VALUE; + long maxRows = 0; + Scan scan = ds.scan(ScanOptions.of()); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(alloc)) { + while (reader.loadNextBatch()) { + int rows = reader.getVectorSchemaRoot().getRowCount(); + batches++; + rowsSeen += rows; + minRows = Math.min(minRows, rows); + maxRows = Math.max(maxRows, rows); + } + } + } + System.out.printf( + "writeChunkRows=%d -> %d read batches over %d rows (min=%d, max=%d, avg=%d)%n", + chunk, batches, rowsSeen, minRows, maxRows, batches == 0 ? 0 : rowsSeen / batches); + Files.deleteIfExists(f); + } + } +} From a6b450900f2be409662a3c59ffdc72135e880f1a Mon Sep 17 00:00:00 2001 From: mprammer Date: Wed, 24 Jun 2026 00:20:41 -0400 Subject: [PATCH 2/4] =?UTF-8?q?java(vortex-jni-bench):=20address=20gauntle?= =?UTF-8?q?t=20review=20=E2=80=94=20isolate=20pushdown,=20fix=20units,=20a?= =?UTF-8?q?dd=20guards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A codex-run gauntlet (fresh/correctness/maint) flagged the first cut as overclaiming relative to what it measured. This commit fixes that: - Isolate native pushdown from JVM-side work: add projectionControl (full scan, consume id,y in Java) and filterControl (full scan, filter cat='alpha' in Java). The pushdown speedup is now projection-vs-projectionControl (~4.1x) and selectiveFilter-vs-filterControl (~4.6x), not the confounded ~6x-vs-fullScan. (M2) - fullScan now consumes all six columns at the buffer level (z, cat, tag added), so the "all-six-column scan" number is honest (~40M rows/s). (M1) - @OperationsPerInvocation(ROWS) so JMH reports input rows/s directly, not scans/s. (M3) - @Setup validates the file before measuring: exact row count, cat='alpha' returns ROWS/|CATS|, projection schema is exactly [id,y] — fast garbage can't be cited. (M5) - Gradle guard fails the jmh task unless VORTEX_SKIP_MAKE_TEST_FILES=true, so a plain run can't silently rebuild + measure the debug lib. (M6) - @Threads(1); tag carries a 10% null rate; README documents the synthetic-data caveats. Read path returns string columns as VarCharVector (Utf8), not ViewVarCharVector — matches the existing TestMinimal read path. Native floor for a boundary-overhead % remains the v2 TODO (M4). Co-Authored-By: Claude Signed-off-by: mprammer --- java/vortex-jni-bench/README.md | 39 ++++- java/vortex-jni-bench/build.gradle.kts | 16 ++ .../vortex/bench/VortexJniReadBenchmark.java | 156 +++++++++++++++--- 3 files changed, 178 insertions(+), 33 deletions(-) diff --git a/java/vortex-jni-bench/README.md b/java/vortex-jni-bench/README.md index ad065bf9b78..0df4a97e71e 100644 --- a/java/vortex-jni-bench/README.md +++ b/java/vortex-jni-bench/README.md @@ -4,22 +4,47 @@ JMH microbenchmarks that stress the **vortex-jni read boundary** — JNI plus th Interface — which is the path an Iceberg `FormatModel` takes to read Vortex from the JVM. `VortexJniReadBenchmark` writes a synthetic six-column table (2M rows: 2× int64, 2× float64, -2× Utf8View) and reads it back three ways, consuming columns at the buffer level (numeric sums / -null counts) so the numbers reflect format + boundary cost, not per-row Java allocation: +2× Utf8View) and reads it back, consuming columns at the buffer level (numeric sums, view lengths, +null counts) so the numbers reflect format + boundary cost, not per-row Java allocation. + +Each invocation scans the full 2M-row table, so `@OperationsPerInvocation(ROWS)` makes JMH report +**input rows scanned per second** directly (not scans/s that you have to convert by hand). + +Benchmarks: - `fullScan` — read all six columns. -- `projection` — read two of six (projection pushdown). -- `selectiveFilter` — `cat = 'alpha'` (~1/16 selectivity; filter pushdown). +- `projection` — native projection of `id,y` (two of six columns). +- `projectionControl` — full scan, but consume only `id,y` in Java (NO native projection). +- `selectiveFilter` — native filter `cat = 'alpha'` (~1/16 selectivity). +- `filterControl` — full scan, evaluate `cat = 'alpha'` in Java (NO native filter). + +The **controls are the point**: `projection` vs `projectionControl` and `selectiveFilter` vs +`filterControl` do the same Java-side work with and without native pushdown, so the remaining +speedup isolates native projection / filter pushdown from "fewer vectors / fewer rows touched in +Java." Comparing a pushdown lane against `fullScan` alone conflates the two and overstates pushdown. + +`@Setup` validates the generated file before any measurement (exact row count, `cat='alpha'` returns +exactly `ROWS/|CATS|`, projection schema is exactly `[id, y]`) and fails the trial otherwise — a +corrupt write or broken filter must not silently produce impressive throughput. `ScanOptions` has no read-batch knob, and Vortex coalesces to ~64K-row read batches regardless of the writer's chunk size, so the boundary is amortized over large batches by construction. Run the -`main` method to see that batch-granularity diagnostic. +`main` method to see the batch-granularity diagnostic across chunk sizes. + +**Workload caveats** (it is synthetic, single-machine, warm-cache — directional, not a leaderboard): +`id` is sequential, `cat` is a periodic 16-value low-cardinality column (kept non-null so filter +selectivity is exactly 1/16), `tag` is high-cardinality with a 10% null rate, numerics use a fixed +seed. This shape is friendly to compression and pushdown; a less-compressible / higher-null workload +is future work. These lanes measure throughput and pushdown *through* the boundary, **not** the +boundary's overhead versus a native floor (that comparison is the v2 TODO below). ## Running The benchmark **must** run against a `--release` native lib (the dev `makeTestFiles` task builds a -debug lib, which would make the numbers meaningless). Build it once and drop it into vortex-jni's -resources, then run with `VORTEX_SKIP_MAKE_TEST_FILES=true` so the debug rebuild doesn't clobber it: +debug lib, which would make the numbers meaningless). The `jmh` task is **guarded** to fail unless +`VORTEX_SKIP_MAKE_TEST_FILES=true`, so a plain `./gradlew :vortex-jni-bench:jmh` cannot silently +rebuild and measure the debug lib. Build the release lib once, drop it into vortex-jni's resources, +then run: ```bash # from repo root: build the release cdylib diff --git a/java/vortex-jni-bench/build.gradle.kts b/java/vortex-jni-bench/build.gradle.kts index febd4c4479e..e84b893db0f 100644 --- a/java/vortex-jni-bench/build.gradle.kts +++ b/java/vortex-jni-bench/build.gradle.kts @@ -27,3 +27,19 @@ dependencies { jmh { jmhVersion.set("1.37") } + +// Guard: the benchmark is meaningless against a debug native lib. Require the deliberate release path +// (VORTEX_SKIP_MAKE_TEST_FILES=true, with a release libvortex_jni placed in vortex-jni's resources) so a +// plain `./gradlew :vortex-jni-bench:jmh` cannot silently rebuild + measure the debug lib. +tasks.named("jmh") { + doFirst { + if (System.getenv("VORTEX_SKIP_MAKE_TEST_FILES") != "true") { + throw GradleException( + "vortex-jni-bench must run against a RELEASE native lib. Build it " + + "(cargo build --release -p vortex-jni), copy it into " + + "vortex-jni/src/main/resources/native/-/, and re-run with " + + "VORTEX_SKIP_MAKE_TEST_FILES=true. See README.md.", + ) + } + } +} diff --git a/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java index ab3b03ffd7d..ad4123d0745 100644 --- a/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java +++ b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java @@ -15,6 +15,7 @@ import dev.vortex.jni.NativeLoader; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Random; @@ -27,6 +28,7 @@ import org.apache.arrow.vector.BigIntVector; import org.apache.arrow.vector.FieldVector; import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.ViewVarCharVector; import org.apache.arrow.vector.ipc.ArrowReader; @@ -40,27 +42,40 @@ import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Measurement; import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OperationsPerInvocation; import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; import org.openjdk.jmh.annotations.Warmup; import org.openjdk.jmh.infra.Blackhole; /** * Measures read throughput through the vortex-jni boundary (JNI + the Arrow C Data Interface). * - *

Three query shapes — full scan, projection, and a selective filter — over a synthetic six-column table. Rows are - * consumed column-at-a-time (numeric sums, null counts) rather than into per-row Java objects, so the numbers reflect - * the format/boundary cost rather than JVM allocation. + *

Every invocation scans the full {@link #ROWS}-row table, so {@code @OperationsPerInvocation(ROWS)} makes JMH + * report input rows scanned per second directly (rather than scans/s that the reader must convert). * - *

Note on batch size: {@code ScanOptions} exposes no read-batch knob, and Vortex coalesces to ~64K-row read batches - * regardless of the writer's chunk size (see {@link #main}), so the boundary cost is already amortized over large - * batches by construction — there is no small-batch regime to sweep from the public API. + *

To isolate native pushdown from JVM-side consumption savings, the projection and filter lanes each have a + * control that does the SAME Java-side work but WITHOUT the native pushdown: + * + *

    + *
  • {@code projection} (native projection of id,y) vs {@code projectionControl} (full scan, consume only id,y) — + * the speedup that remains is attributable to native projection, not to touching fewer Java vectors. + *
  • {@code selectiveFilter} (native filter cat='alpha') vs {@code filterControl} (full scan, filter in Java) — + * the speedup that remains is attributable to native filter pushdown, not to summing fewer rows. + *
+ * + *

Rows are consumed column-at-a-time (numeric sums, view lengths, null counts) rather than into per-row Java + * objects, so the numbers reflect format/boundary cost rather than JVM allocation. {@code ScanOptions} exposes no + * read-batch knob, and Vortex coalesces to ~64K-row read batches regardless of the writer's chunk size (see + * {@link #main}), so boundary cost is amortized over large batches by construction. */ @BenchmarkMode(Mode.Throughput) @OutputTimeUnit(TimeUnit.SECONDS) +@OperationsPerInvocation(VortexJniReadBenchmark.ROWS) @Warmup(iterations = 3, time = 2) @Measurement(iterations = 5, time = 2) @Fork( @@ -69,15 +84,18 @@ "--add-opens=java.base/java.nio=ALL-UNNAMED", "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED" }) +@Threads(1) @State(Scope.Benchmark) public class VortexJniReadBenchmark { - static final long ROWS = 2_000_000L; + static final int ROWS = 2_000_000; static final int WRITE_CHUNK = 65536; static final String[] CATS = { "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", "juliet", "kilo", "lima", "mike", "november", "oscar", "papa" }; + static final byte[] ALPHA = "alpha".getBytes(UTF_8); + static final long EXPECTED_ALPHA_MATCHES = ROWS / CATS.length; BufferAllocator allocator; Session session; @@ -94,6 +112,7 @@ public void setup() throws Exception { String uri = file.toAbsolutePath().toUri().toString(); writeTable(session, allocator, uri, WRITE_CHUNK); dataSource = DataSource.open(session, uri); + validate(); } @TearDown(Level.Trial) @@ -107,6 +126,43 @@ public void teardown() throws Exception { } } + /** Fail the trial loudly if the generated file or pushdown semantics are wrong — fast garbage must not be cited. */ + private void validate() throws Exception { + if (!(dataSource.rowCount() instanceof DataSource.RowCount.Exact exact) || exact.value() != ROWS) { + throw new IllegalStateException("expected exactly " + ROWS + " rows, got " + dataSource.rowCount()); + } + // Native filter must return exactly ROWS/|CATS| rows for cat='alpha'. + Expression filter = + Expression.binary(Expression.BinaryOp.EQ, Expression.column("cat"), Expression.literal(CATS[0])); + long matched = 0; + Scan scan = dataSource.scan(ScanOptions.builder().filter(filter).build()); + while (scan.hasNext()) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { + while (reader.loadNextBatch()) { + matched += reader.getVectorSchemaRoot().getRowCount(); + } + } + } + if (matched != EXPECTED_ALPHA_MATCHES) { + throw new IllegalStateException("filter cat='alpha' returned " + matched + ", expected " + EXPECTED_ALPHA_MATCHES); + } + // Native projection must yield exactly [id, y]. + Expression projection = Expression.select(new String[] {"id", "y"}, Expression.root()); + Scan pscan = dataSource.scan(ScanOptions.builder().projection(projection).build()); + if (pscan.hasNext()) { + try (ArrowReader reader = pscan.next().scanArrow(allocator)) { + if (reader.loadNextBatch()) { + List names = reader.getVectorSchemaRoot().getSchema().getFields().stream() + .map(Field::getName) + .toList(); + if (!names.equals(List.of("id", "y"))) { + throw new IllegalStateException("projection schema expected [id, y], got " + names); + } + } + } + } + } + private static Schema schema() { return new Schema(List.of( Field.notNullable("id", new ArrowType.Int(64, true)), @@ -141,8 +197,14 @@ private static void writeTable(Session session, BufferAllocator allocator, Strin x.setSafe(i, rnd.nextInt(1_000_000)); y.setSafe(i, rnd.nextDouble()); z.setSafe(i, rnd.nextDouble()); + // cat stays non-null and deterministic so filter selectivity is exactly 1/|CATS|. cat.setSafe(i, CATS[(int) (r % CATS.length)].getBytes(UTF_8)); - tag.setSafe(i, Long.toString(r).getBytes(UTF_8)); + // tag carries nulls (every 10th row) and high-cardinality values to exercise a validity buffer. + if (r % 10 == 0) { + tag.setNull(i); + } else { + tag.setSafe(i, Long.toString(r).getBytes(UTF_8)); + } } root.setRowCount(batch); try (ArrowArray arr = ArrowArray.allocateNew(allocator); @@ -155,29 +217,37 @@ private static void writeTable(Session session, BufferAllocator allocator, Strin } } + /** Full scan consuming ALL six columns at the buffer level. */ @Benchmark public void fullScan(Blackhole bh) throws Exception { long sumId = 0; long sumX = 0; double sumY = 0; - long catNonNull = 0; + double sumZ = 0; + long catLen = 0; + long tagLenOrNulls = 0; Scan scan = dataSource.scan(ScanOptions.of()); while (scan.hasNext()) { - Partition partition = scan.next(); - try (ArrowReader reader = partition.scanArrow(allocator)) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { while (reader.loadNextBatch()) { VectorSchemaRoot r = reader.getVectorSchemaRoot(); int rows = r.getRowCount(); BigIntVector id = (BigIntVector) r.getVector("id"); BigIntVector x = (BigIntVector) r.getVector("x"); Float8Vector y = (Float8Vector) r.getVector("y"); - FieldVector cat = r.getVector("cat"); + Float8Vector z = (Float8Vector) r.getVector("z"); + VarCharVector cat = (VarCharVector) r.getVector("cat"); + VarCharVector tag = (VarCharVector) r.getVector("tag"); for (int i = 0; i < rows; i++) { sumId += id.get(i); sumX += x.get(i); sumY += y.get(i); - if (!cat.isNull(i)) { - catNonNull++; + sumZ += z.get(i); + catLen += cat.getValueLength(i); + if (tag.isNull(i)) { + tagLenOrNulls++; + } else { + tagLenOrNulls += tag.getValueLength(i); } } } @@ -186,19 +256,29 @@ public void fullScan(Blackhole bh) throws Exception { bh.consume(sumId); bh.consume(sumX); bh.consume(sumY); - bh.consume(catNonNull); + bh.consume(sumZ); + bh.consume(catLen); + bh.consume(tagLenOrNulls); } + /** Native projection pushdown: only id,y cross the boundary. */ @Benchmark public void projection(Blackhole bh) throws Exception { Expression projection = Expression.select(new String[] {"id", "y"}, Expression.root()); - ScanOptions options = ScanOptions.builder().projection(projection).build(); + bh.consume(consumeIdY(dataSource.scan(ScanOptions.builder().projection(projection).build()))); + } + + /** Control for {@link #projection}: full scan, but consume only id,y in Java (no native projection). */ + @Benchmark + public void projectionControl(Blackhole bh) throws Exception { + bh.consume(consumeIdY(dataSource.scan(ScanOptions.of()))); + } + + private double consumeIdY(Scan scan) throws Exception { long sumId = 0; double sumY = 0; - Scan scan = dataSource.scan(options); while (scan.hasNext()) { - Partition partition = scan.next(); - try (ArrowReader reader = partition.scanArrow(allocator)) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { while (reader.loadNextBatch()) { VectorSchemaRoot r = reader.getVectorSchemaRoot(); int rows = r.getRowCount(); @@ -211,21 +291,19 @@ public void projection(Blackhole bh) throws Exception { } } } - bh.consume(sumId); - bh.consume(sumY); + return sumId + sumY; } + /** Native filter pushdown: only matching rows cross the boundary. */ @Benchmark public void selectiveFilter(Blackhole bh) throws Exception { Expression filter = Expression.binary(Expression.BinaryOp.EQ, Expression.column("cat"), Expression.literal(CATS[0])); - ScanOptions options = ScanOptions.builder().filter(filter).build(); long matched = 0; long sumId = 0; - Scan scan = dataSource.scan(options); + Scan scan = dataSource.scan(ScanOptions.builder().filter(filter).build()); while (scan.hasNext()) { - Partition partition = scan.next(); - try (ArrowReader reader = partition.scanArrow(allocator)) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { while (reader.loadNextBatch()) { VectorSchemaRoot r = reader.getVectorSchemaRoot(); int rows = r.getRowCount(); @@ -241,13 +319,39 @@ public void selectiveFilter(Blackhole bh) throws Exception { bh.consume(sumId); } + /** Control for {@link #selectiveFilter}: full scan, evaluate cat='alpha' in Java (no native filter). */ + @Benchmark + public void filterControl(Blackhole bh) throws Exception { + long matched = 0; + long sumId = 0; + Scan scan = dataSource.scan(ScanOptions.of()); + while (scan.hasNext()) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot r = reader.getVectorSchemaRoot(); + int rows = r.getRowCount(); + BigIntVector id = (BigIntVector) r.getVector("id"); + VarCharVector cat = (VarCharVector) r.getVector("cat"); + for (int i = 0; i < rows; i++) { + if (!cat.isNull(i) && Arrays.equals(cat.get(i), ALPHA)) { + sumId += id.get(i); + matched++; + } + } + } + } + } + bh.consume(matched); + bh.consume(sumId); + } + /** * Diagnostic (not a benchmark): prints the distribution of read batch row counts for a few writer chunk sizes, to * show that Vortex coalesces to a stable read-batch granularity independent of how the file was written. */ public static void main(String[] args) throws Exception { NativeLoader.loadJni(); - for (int chunk : new int[] {8192, 131072}) { + for (int chunk : new int[] {8192, 65536, 131072}) { BufferAllocator alloc = new RootAllocator(Long.MAX_VALUE); Session sess = Session.create(); Path f = Files.createTempFile("vortex-jni-diag-" + chunk + "-", ".vortex"); From 0398bc5fd8d2b912bb083d4b866225e684b1faf2 Mon Sep 17 00:00:00 2001 From: mprammer Date: Sat, 4 Jul 2026 21:27:27 -0400 Subject: [PATCH 3/4] bench(vortex-jni): add native-Rust floor lane + shared-file hook Adds a native-Rust read "floor" for the vortex-jni read-boundary benchmark so Appendix B can quote JNI+Arrow-C-boundary throughput against a Rust-native floor from one reproducible box. - vortex-jni/src/bin/floor.rs: mirrors the JNI read path (session -> file -> scan -> array stream -> execute_arrow) on the same CurrentThreadRuntime the JNI lib uses, stopping before the Arrow C export / JNI / JVM. Reports rows/s for fullScan / projection(id,y) / selectiveFilter(cat='alpha'), normalized by the file row count to match JMH's @OperationsPerInvocation. Validates row counts. - VortexJniReadBenchmark: when VORTEX_JNI_BENCH_FILE is set, write+keep the table at that path (instead of a deleted temp file) so the floor reads the byte-identical file. Default behavior unchanged. Co-Authored-By: Claude Signed-off-by: mprammer --- .../vortex/bench/VortexJniReadBenchmark.java | 14 +- vortex-jni/src/bin/floor.rs | 215 ++++++++++++++++++ 2 files changed, 226 insertions(+), 3 deletions(-) create mode 100644 vortex-jni/src/bin/floor.rs diff --git a/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java index ad4123d0745..9f5a0279a5f 100644 --- a/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java +++ b/java/vortex-jni-bench/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java @@ -15,6 +15,7 @@ import dev.vortex.jni.NativeLoader; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.Paths; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -101,13 +102,19 @@ public class VortexJniReadBenchmark { Session session; DataSource dataSource; Path file; + // When VORTEX_JNI_BENCH_FILE is set, the table is written to (and kept at) that path instead of a deleted-on-exit + // temp file, so the native-Rust "floor" lane can read the byte-identical file this benchmark reads. Default + // behavior (unset) is unchanged: a temp file removed in teardown. + boolean keepFile; @Setup(Level.Trial) public void setup() throws Exception { NativeLoader.loadJni(); allocator = new RootAllocator(Long.MAX_VALUE); session = Session.create(); - file = Files.createTempFile("vortex-jni-bench-", ".vortex"); + String keepPath = System.getenv("VORTEX_JNI_BENCH_FILE"); + keepFile = keepPath != null && !keepPath.isEmpty(); + file = keepFile ? Paths.get(keepPath) : Files.createTempFile("vortex-jni-bench-", ".vortex"); Files.deleteIfExists(file); String uri = file.toAbsolutePath().toUri().toString(); writeTable(session, allocator, uri, WRITE_CHUNK); @@ -119,9 +126,10 @@ public void setup() throws Exception { public void teardown() throws Exception { // Intentionally does not close the allocator: DataSource/Scan native resources are released by VortexCleaner // at GC time, which races an explicit allocator.close() and trips leak detection. The JMH fork exits after the - // trial and reclaims everything; we only remove the temp file. + // trial and reclaims everything; we only remove the temp file. When keepFile is set the file is retained for + // the native floor lane. dataSource = null; - if (file != null) { + if (file != null && !keepFile) { Files.deleteIfExists(file); } } diff --git a/vortex-jni/src/bin/floor.rs b/vortex-jni/src/bin/floor.rs new file mode 100644 index 00000000000..be657c2f875 --- /dev/null +++ b/vortex-jni/src/bin/floor.rs @@ -0,0 +1,215 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Native-Rust "floor" lane for the vortex-jni read benchmark. +//! +//! Reads the SAME `.vortex` file that `VortexJniReadBenchmark` reads, but entirely in Rust: it +//! mirrors the JNI read path (session -> file -> scan -> array stream -> `execute_arrow`) using the +//! very same [`CurrentThreadRuntime`] the shipping JNI lib uses, and stops *before* the Arrow C Data +//! Interface export / JNI call / JVM vector access. The delta between this floor and the JMH numbers +//! is therefore attributable to the boundary (FFI marshaling + JNI + JVM), not to the format. +//! +//! For each op it reports rows/s normalized by the file's total row count (the same normalization +//! JMH uses via `@OperationsPerInvocation(ROWS)`), so JMH ops/s and floor rows/s are directly +//! comparable: +//! * `fullScan` — project all columns, no filter. +//! * `projection` — native projection of `id,y`. +//! * `selectiveFilter` — native filter `cat = 'alpha'` (~1/16 selectivity), all columns. +//! +//! Usage: `floor [warmup_iters] [measured_iters]`. + +use std::env::consts::ARCH; +use std::error::Error; +use std::hint::black_box; +use std::sync::LazyLock; +use std::time::Duration; +use std::time::Instant; + +use futures::StreamExt; +use futures::pin_mut; +use vortex::VortexSessionDefault; +use vortex::array::VortexSessionExecute; +use vortex::array::arrow::ArrowSessionExt; +use vortex::expr::Expression; +use vortex::expr::col; +use vortex::expr::eq; +use vortex::expr::lit; +use vortex::expr::root; +use vortex::expr::select; +use vortex::file::OpenOptionsSessionExt; +use vortex::file::VortexFile; +use vortex::io::runtime::BlockingRuntime; +use vortex::io::runtime::current::CurrentThreadRuntime; +use vortex::io::session::RuntimeSessionExt; +use vortex::session::VortexSession; + +/// Current-thread runtime, identical to the one the JNI lib drives its scans on. +static RUNTIME: LazyLock = LazyLock::new(CurrentThreadRuntime::new); + +/// Outcome of running one op: rows produced by the last iteration and per-iteration wall times. +struct OpResult { + output_rows: u64, + per_iter: Vec, +} + +/// Scan `file` once with the given projection/filter, materializing every chunk to Arrow in-process +/// (`execute_arrow`) exactly as the JNI path does before it exports across the C Data Interface. +/// Returns the number of rows produced (post-filter). +fn scan_once( + file: &VortexFile, + session: &VortexSession, + projection: Expression, + filter: Option, +) -> Result> { + let stream = file + .scan()? + .with_projection(projection) + .with_some_filter(filter) + .into_array_stream()?; + let rows = RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session.create_execution_ctx(); + let mut rows: u64 = 0; + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + let arrow = session.arrow().execute_arrow(chunk, None, &mut ctx)?; + rows += arrow.len() as u64; + black_box(&arrow); + } + Ok::>(rows) + })?; + Ok(rows) +} + +/// Warm up, then time `iters` full scans of one op. +fn bench_op( + file: &VortexFile, + session: &VortexSession, + projection: Expression, + filter: Option, + warmup: usize, + iters: usize, +) -> Result> { + for _ in 0..warmup { + scan_once(file, session, projection.clone(), filter.clone())?; + } + let mut per_iter = Vec::with_capacity(iters); + let mut output_rows = 0; + for _ in 0..iters { + let start = Instant::now(); + output_rows = scan_once(file, session, projection.clone(), filter.clone())?; + per_iter.push(start.elapsed()); + } + Ok(OpResult { + output_rows, + per_iter, + }) +} + +/// Median of a set of durations, in seconds. +fn median_secs(durations: &[Duration]) -> f64 { + let mut secs: Vec = durations.iter().map(Duration::as_secs_f64).collect(); + secs.sort_by(f64::total_cmp); + let mid = secs.len() / 2; + if secs.len() % 2 == 1 { + secs[mid] + } else { + (secs[mid - 1] + secs[mid]) / 2.0 + } +} + +fn report(name: &str, input_rows: u64, res: &OpResult) { + let median = median_secs(&res.per_iter); + let best = res + .per_iter + .iter() + .map(Duration::as_secs_f64) + .fold(f64::INFINITY, f64::min); + let worst = res + .per_iter + .iter() + .map(Duration::as_secs_f64) + .fold(0.0_f64, f64::max); + let rps_median = input_rows as f64 / median; + let rps_best = input_rows as f64 / best; + let rps_worst = input_rows as f64 / worst; + println!( + "floor {name}: input_rows={input_rows} output_rows={out} iters={n} \ + median_rows_per_s={rps_median:.0} best_rows_per_s={rps_best:.0} worst_rows_per_s={rps_worst:.0} \ + median_ms={ms:.3}", + out = res.output_rows, + n = res.per_iter.len(), + ms = median * 1000.0, + ); +} + +fn main() -> Result<(), Box> { + let mut args = std::env::args().skip(1); + let path = args + .next() + .ok_or("usage: floor [warmup_iters] [measured_iters]")?; + let warmup: usize = match args.next() { + Some(v) => v.parse()?, + None => 3, + }; + let iters: usize = match args.next() { + Some(v) => v.parse()?, + None => 7, + }; + + // Build the session exactly as the JNI lib does, so encodings match the writer's session. + let session = VortexSession::default().with_handle(RUNTIME.handle()); + vortex_parquet_variant::initialize(&session); + + let file = RUNTIME.block_on( + session + .open_options() + .with_layout_reader_cache() + .open_path(&path), + )?; + let total = file.row_count(); + if total == 0 { + return Err(format!("file {path} reports zero rows").into()); + } + let expected_alpha = total / 16; + + println!("floor: file={path} total_rows={total} warmup={warmup} iters={iters}"); + println!("floor: target_arch={ARCH}"); + + // fullScan: all columns, no filter. + let full = bench_op(&file, &session, root(), None, warmup, iters)?; + if full.output_rows != total { + return Err(format!( + "fullScan produced {} rows, expected {total}", + full.output_rows + ) + .into()); + } + report("fullScan", total, &full); + + // projection: native projection of id,y. + let proj = select(vec!["id", "y"], root()); + let projection = bench_op(&file, &session, proj, None, warmup, iters)?; + if projection.output_rows != total { + return Err(format!( + "projection produced {} rows, expected {total}", + projection.output_rows + ) + .into()); + } + report("projection", total, &projection); + + // selectiveFilter: native filter cat = 'alpha' (~1/16), all columns. + let filter = eq(col("cat"), lit("alpha")); + let selective = bench_op(&file, &session, root(), Some(filter), warmup, iters)?; + if selective.output_rows != expected_alpha { + return Err(format!( + "selectiveFilter produced {} rows, expected {expected_alpha}", + selective.output_rows + ) + .into()); + } + report("selectiveFilter", total, &selective); + + Ok(()) +} From b0360cbb307ccfc15f713ed5ec5b6e8383f074de Mon Sep 17 00:00:00 2001 From: mprammer Date: Sun, 5 Jul 2026 13:01:47 -0400 Subject: [PATCH 4/4] bench(vortex-jni): add pure-Rust reproducer for multi-chunk write corruption underflow.rs reproduces whole-row data corruption in the vortex-jni Arrow-C write path: a multi-chunk write of Arrow-imported variable-width (Utf8/Utf8View) columns with a reused VectorSchemaRoot and an uneven final chunk corrupts the tail of full-size chunks (strings -> "", ints -> 0). A pure-Rust write of the same data is clean; the JMH read bench's `cat = 'alpha'` filter consequently under-counts (111,168 vs the expected 125,000 = ROWS/16). Run: cargo run -p vortex-jni --bin underflow (knobs: UF_ROWS / UF_CHUNK / UF_VIEW) Co-Authored-By: Claude Signed-off-by: mprammer --- vortex-jni/src/bin/underflow.rs | 472 ++++++++++++++++++++++++++++++++ 1 file changed, 472 insertions(+) create mode 100644 vortex-jni/src/bin/underflow.rs diff --git a/vortex-jni/src/bin/underflow.rs b/vortex-jni/src/bin/underflow.rs new file mode 100644 index 00000000000..b71a39be0ea --- /dev/null +++ b/vortex-jni/src/bin/underflow.rs @@ -0,0 +1,472 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Pure-Rust reproducer for the low-cardinality Utf8View equality-filter under-count. +//! +//! Builds a table shaped exactly like `VortexJniReadBenchmark.writeTable` (2M rows, `cat` cycling +//! 16 low-cardinality values with "alpha" one of them), writes it with the bench's DEFAULT write +//! config (`session.write_options()`), then reads it back three ways and prints: +//! (a) fullScan -> materialize `cat` -> count == "alpha" +//! (b) filter eq(col("cat"), lit("alpha")) -> count +//! (c) total rows +//! Expected alpha = ROWS / 16. +//! +//! Knobs (env vars): +//! UF_ROWS total rows (default 2_000_000) +//! UF_CHUNK write chunk rows (default 65536) +//! UF_VIEW cat/tag as Utf8View (default 1; 0 => Utf8) +//! UF_NCATS distinct cat values (default 16) +//! UF_KEEP keep file at this path instead of a temp file (default: temp, deleted) + +use std::env::consts::ARCH; +use std::error::Error; +use std::sync::LazyLock; + +use arrow_array::Array as ArrowArray; +use arrow_array::Float64Array; +use arrow_array::Int64Array; +use arrow_array::RecordBatch; +use arrow_array::StringArray; +use arrow_array::StringViewArray; +use arrow_array::StructArray as ArrowStructArray; +use arrow_schema::DataType; +use arrow_schema::Field as ArrowField; +use arrow_schema::Schema as ArrowSchema; +use futures::StreamExt; +use futures::pin_mut; +use std::sync::Arc; +use vortex::VortexSessionDefault; +use vortex::array::VortexSessionExecute; +use vortex::array::arrow::ArrowSessionExt; +use vortex::error::VortexError; +use vortex::expr::col; +use vortex::expr::eq; +use vortex::expr::lit; +use vortex::expr::root; +use vortex::file::OpenOptionsSessionExt; +use vortex::file::VortexFile; +use vortex::file::WriteOptionsSessionExt; +use vortex::io::runtime::BlockingRuntime; +use vortex::io::runtime::current::CurrentThreadRuntime; +use vortex::io::session::RuntimeSessionExt; +use vortex::array::iter::ArrayIteratorAdapter; +use vortex::layout::scan::split_by::SplitBy; +use vortex::session::VortexSession; + +/// Read the optional `UF_SPLIT` override: `SplitBy::RowCount(n)` when set, else layout default. +fn split_override() -> Option { + std::env::var("UF_SPLIT") + .ok() + .and_then(|v| v.parse::().ok()) + .map(SplitBy::RowCount) +} + +/// Apply optional scan knobs from env: UF_SPLIT (RowCount), UF_CONC (concurrency), UF_ORDERED (0/1). +fn apply_scan_knobs( + mut scan: vortex::layout::scan::scan_builder::ScanBuilder, +) -> vortex::layout::scan::scan_builder::ScanBuilder { + if let Some(sb) = split_override() { + scan = scan.with_split_by(sb); + } + if let Some(c) = std::env::var("UF_CONC").ok().and_then(|v| v.parse::().ok()) { + scan = scan.with_concurrency(c); + } + if let Some(o) = std::env::var("UF_ORDERED").ok().and_then(|v| v.parse::().ok()) { + scan = scan.with_ordered(o != 0); + } + scan +} + +static RUNTIME: LazyLock = LazyLock::new(CurrentThreadRuntime::new); + +const CATS: [&str; 16] = [ + "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", "juliet", + "kilo", "lima", "mike", "november", "oscar", "papa", +]; + +fn env_usize(key: &str, default: usize) -> usize { + std::env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn string_field(name: &str, view: bool) -> ArrowField { + let dt = if view { + DataType::Utf8View + } else { + DataType::Utf8 + }; + ArrowField::new(name, dt, true) +} + +fn string_col(values: Vec>, view: bool, block: usize) -> Arc { + if view { + if block > 0 { + // Build with a small block size so non-inline strings spill across MANY variadic data + // buffers (views with buffer_index 0,1,2,...), mimicking Arrow-Java ViewVarCharVector. + let mut b = arrow_array::builder::StringViewBuilder::new() + .with_fixed_block_size(block as u32); + for v in &values { + b.append_option(v.as_deref()); + } + Arc::new(b.finish()) as Arc + } else { + Arc::new(StringViewArray::from_iter(values)) as Arc + } + } else { + Arc::new(StringArray::from_iter(values)) as Arc + } +} + +/// Count the number of `cat` == "alpha" entries in an executed Arrow struct array. +fn count_alpha_in_struct(arrow: &Arc) -> Result> { + let st = arrow + .as_any() + .downcast_ref::() + .ok_or("execute_arrow output was not a struct array")?; + let cat = st + .column_by_name("cat") + .ok_or("no 'cat' column in output")?; + count_alpha(cat.as_ref()) +} + +/// Accumulate a distribution of `cat` values (and nulls) from an executed struct array. +fn tally_cat( + arrow: &Arc, + hist: &mut std::collections::BTreeMap, + nulls: &mut u64, +) -> Result<(), Box> { + let st = arrow + .as_any() + .downcast_ref::() + .ok_or("not a struct array")?; + let cat = st.column_by_name("cat").ok_or("no 'cat' column")?; + let mut push = |v: Option<&str>| match v { + Some(s) => *hist.entry(s.to_string()).or_insert(0) += 1, + None => *nulls += 1, + }; + if let Some(v) = cat.as_any().downcast_ref::() { + for s in v.iter() { + push(s); + } + } else if let Some(v) = cat.as_any().downcast_ref::() { + for s in v.iter() { + push(s); + } + } else { + return Err(format!("unexpected cat type: {:?}", cat.data_type()).into()); + } + Ok(()) +} + +fn count_alpha(col: &dyn ArrowArray) -> Result> { + if let Some(v) = col.as_any().downcast_ref::() { + Ok(v.iter().filter(|s| *s == Some("alpha")).count()) + } else if let Some(v) = col.as_any().downcast_ref::() { + Ok(v.iter().filter(|s| *s == Some("alpha")).count()) + } else { + Err(format!("unexpected cat arrow type: {:?}", col.data_type()).into()) + } +} + +fn build_and_write( + session: &VortexSession, + path: &str, + rows: usize, + chunk: usize, + view: bool, + ncats: usize, +) -> Result<(), Box> { + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int64, false), + ArrowField::new("x", DataType::Int64, false), + ArrowField::new("y", DataType::Float64, false), + ArrowField::new("z", DataType::Float64, false), + string_field("cat", view), + string_field("tag", view), + ])); + let write_schema = session.arrow().from_arrow_schema(schema.as_ref())?; + + // UF_PAD: append this many 'x' chars to each string value to push it past the 12-byte + // StringView inline threshold, forcing an actual variadic data buffer (like non-inline Arrow). + let pad = env_usize("UF_PAD", 0); + let suffix: String = "x".repeat(pad); + // UF_BLOCK: StringView builder fixed block size; small values force many variadic data buffers. + let block = env_usize("UF_BLOCK", 0); + + let mut chunks: Vec = Vec::new(); + let mut written = 0usize; + while written < rows { + let batch = std::cmp::min(chunk, rows - written); + let mut id = Vec::with_capacity(batch); + let mut x = Vec::with_capacity(batch); + let mut y = Vec::with_capacity(batch); + let mut z = Vec::with_capacity(batch); + let mut cat: Vec> = Vec::with_capacity(batch); + let mut tag: Vec> = Vec::with_capacity(batch); + for i in 0..batch { + let r = (written + i) as i64; + id.push(r); + x.push((r * 2654435761) % 1_000_000); + y.push((r as f64) * 0.5); + z.push((r as f64) * 0.25); + cat.push(Some(format!("{}{}", CATS[(r as usize) % ncats], suffix))); + if r % 10 == 0 { + tag.push(None); + } else { + tag.push(Some(format!("{}{}", r, suffix))); + } + } + let columns: Vec> = vec![ + Arc::new(Int64Array::from(id)), + Arc::new(Int64Array::from(x)), + Arc::new(Float64Array::from(y)), + Arc::new(Float64Array::from(z)), + string_col(cat, view, block), + string_col(tag, view, block), + ]; + let record_batch = RecordBatch::try_new(schema.clone(), columns)?; + let vortex_batch = session + .arrow() + .from_arrow_record_batch(record_batch, schema.as_ref())?; + chunks.push(vortex_batch); + written += batch; + } + + let iter = ArrayIteratorAdapter::new( + write_schema, + chunks.into_iter().map(Ok::<_, VortexError>), + ); + let file = std::fs::File::create(path)?; + let summary = session + .write_options() + .blocking(&*RUNTIME) + .write(file, iter)?; + println!( + "write: rows={} chunk={} view={} ncats={} -> file_rows={} bytes={}", + rows, + chunk, + view, + ncats, + summary.row_count(), + summary.size() + ); + Ok(()) +} + +/// fullScan: project everything, materialize each chunk, count cat=="alpha". +fn fullscan_alpha_count( + file: &VortexFile, + session: &VortexSession, +) -> Result<(u64, u64), Box> { + let scan = apply_scan_knobs(file.scan()?.with_projection(root())); + let stream = scan.into_array_stream()?; + RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session.create_execution_ctx(); + let mut rows: u64 = 0; + let mut alpha: u64 = 0; + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + let arrow = session.arrow().execute_arrow(chunk, None, &mut ctx)?; + rows += arrow.len() as u64; + alpha += count_alpha_in_struct(&arrow)? as u64; + } + Ok::<(u64, u64), Box>((rows, alpha)) + }) +} + +/// filter eq(col("cat"), lit("alpha")): count returned rows (and re-verify all are alpha). +fn filter_alpha_count( + file: &VortexFile, + session: &VortexSession, +) -> Result<(u64, u64), Box> { + let filter = eq(col("cat"), lit("alpha")); + let scan = apply_scan_knobs( + file.scan()?.with_projection(root()).with_some_filter(Some(filter)), + ); + let stream = scan.into_array_stream()?; + RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session.create_execution_ctx(); + let mut rows: u64 = 0; + let mut alpha: u64 = 0; + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + let arrow = session.arrow().execute_arrow(chunk, None, &mut ctx)?; + rows += arrow.len() as u64; + alpha += count_alpha_in_struct(&arrow)? as u64; + } + Ok::<(u64, u64), Box>((rows, alpha)) + }) +} + +fn main() -> Result<(), Box> { + let rows = env_usize("UF_ROWS", 2_000_000); + let chunk = env_usize("UF_CHUNK", 65536); + let view = env_usize("UF_VIEW", 1) != 0; + let ncats = env_usize("UF_NCATS", 16); + + let keep = std::env::var("UF_KEEP").ok().filter(|s| !s.is_empty()); + // Read-only mode: skip writing and just read+diagnose an existing file (e.g. a Java-written one) + // passed as argv[1] or via UF_READ. + let read_only = std::env::args().nth(1).or_else(|| std::env::var("UF_READ").ok()); + let tmp = std::env::temp_dir().join(format!("underflow-{}.vortex", std::process::id())); + let path = read_only + .clone() + .or_else(|| keep.clone()) + .unwrap_or_else(|| tmp.to_string_lossy().into_owned()); + + println!("underflow: target_arch={ARCH} path={path} read_only={}", read_only.is_some()); + + let session = VortexSession::default().with_handle(RUNTIME.handle()); + vortex_parquet_variant::initialize(&session); + + if read_only.is_none() { + build_and_write(&session, &path, rows, chunk, view, ncats)?; + } + + let file = RUNTIME.block_on( + session + .open_options() + .with_layout_reader_cache() + .open_path(&path), + )?; + let total = file.row_count(); + let expected = total / ncats as u64; + + if std::env::var("UF_SPLITS").ok().filter(|s| !s.is_empty()).is_some() { + let splits = file.splits()?; + println!("--- file.splits(): {} natural splits ---", splits.len()); + for (i, r) in splits.iter().enumerate().take(40) { + println!(" split[{i}] = [{}, {}) len={}", r.start, r.end, r.end - r.start); + } + } + + let (fs_rows, fs_alpha) = fullscan_alpha_count(&file, &session)?; + let (flt_rows, flt_alpha) = filter_alpha_count(&file, &session)?; + + if std::env::var("UF_POS").ok().filter(|s| !s.is_empty()).is_some() { + // Assumes canonical Java file: row r has cat == CATS[r % ncats]. Reports positions where the + // decoded value differs (the corrupted rows), and their distribution modulo the read batch. + let stream = file.scan()?.with_projection(root()).into_array_stream()?; + let session = session.clone(); + RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session.create_execution_ctx(); + let mut bad: Vec = Vec::new(); + let mut batch_idx = 0usize; + while let Some(chunk) = stream.next().await { + let arrow = session.arrow().execute_arrow(chunk?, None, &mut ctx)?; + let st = arrow.as_any().downcast_ref::().unwrap(); + let cat = st.column_by_name("cat").unwrap(); + let v = cat.as_any().downcast_ref::().unwrap(); + let id = st + .column_by_name("id") + .and_then(|c| c.as_any().downcast_ref::().cloned()) + .expect("id must be Int64"); + let batch_len = v.len(); + let id_first = id.value(0); + let mut bad_in_batch = 0u64; + let mut first_bad_at: i64 = -1; + for (j, s) in v.iter().enumerate() { + let gr = id.value(j) as u64; // true row index + let expected = CATS[(gr as usize) % ncats]; + if s != Some(expected) { + bad.push(gr); + bad_in_batch += 1; + if first_bad_at < 0 { first_bad_at = j as i64; } + } + } + if bad_in_batch > 0 { + println!(" emit#{batch_idx}: id_first={id_first} len={batch_len} cat_bad={bad_in_batch} first_bad_at={first_bad_at}"); + } + batch_idx += 1; + } + println!(" total emitted batches={batch_idx}, total corrupted rows={}", bad.len()); + bad.sort_unstable(); + // Coalesce into contiguous runs of true row indices. + let mut runs: Vec<(u64, u64)> = Vec::new(); + for &r in &bad { + match runs.last_mut() { + Some(last) if r == last.1 + 1 => last.1 = r, + _ => runs.push((r, r)), + } + } + println!(" corrupted contiguous runs (true row index): {} runs", runs.len()); + for (s, e) in runs.iter().take(20) { + println!(" [{s}, {e}] len={}", e - s + 1); + } + Ok::<(), Box>(()) + })?; + } + + if std::env::var("UF_HIST").ok().filter(|s| !s.is_empty()).is_some() { + let stream = file.scan()?.with_projection(root()).into_array_stream()?; + let session_a = session.clone(); + let (hist, nulls) = RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session_a.create_execution_ctx(); + let mut hist = std::collections::BTreeMap::new(); + let mut nulls = 0u64; + while let Some(chunk) = stream.next().await { + let arrow = session_a.arrow().execute_arrow(chunk?, None, &mut ctx)?; + tally_cat(&arrow, &mut hist, &mut nulls)?; + } + Ok::<_, Box>((hist, nulls)) + })?; + // Independent id integrity check: id should be the exact set {0..total}. Report sum + zeros. + { + let stream = file.scan()?.with_projection(root()).into_array_stream()?; + let session2 = session.clone(); + let (id_sum, id_zeros, n) = RUNTIME.block_on(async move { + pin_mut!(stream); + let mut ctx = session2.create_execution_ctx(); + let (mut sum, mut zeros, mut n) = (0i128, 0u64, 0u64); + while let Some(chunk) = stream.next().await { + let arrow = session2.arrow().execute_arrow(chunk?, None, &mut ctx)?; + let st = arrow.as_any().downcast_ref::().unwrap(); + let id = st.column_by_name("id").unwrap().as_any().downcast_ref::().unwrap().clone(); + for j in 0..id.len() { + let v = id.value(j); + sum += v as i128; + if v == 0 { zeros += 1; } + n += 1; + } + } + Ok::<_, Box>((sum, zeros, n)) + })?; + let expected_sum: i128 = (0..total as i128).sum(); + println!("--- id integrity: n={n} sum={id_sum} expected_sum={expected_sum} match={} id==0 count={id_zeros} (expected 1) ---", id_sum == expected_sum); + } + println!("--- cat distribution (expected {} each of {} values) ---", rows / ncats, ncats); + let mut sum = 0u64; + for (k, v) in &hist { + sum += *v; + let delta = *v as i64 - (rows / ncats) as i64; + println!(" {k:>10} = {v:>8} (delta {delta:+})"); + } + println!(" {:>10} = {nulls:>8}", ""); + println!(" distinct_values={} sum(non-null)={sum} total_with_nulls={}", hist.len(), sum + nulls); + } + + println!("total_rows={total} (scanned fullScan rows={fs_rows})"); + println!("expected_alpha={expected}"); + println!("fullScan_materialized_alpha={fs_alpha}"); + println!("filter_returned_rows={flt_rows} (of which cat==alpha: {flt_alpha})"); + + let verdict = if flt_rows < fs_alpha { + "READ/filter bug: filter drops real matches (filter < fullScan-materialized)" + } else if fs_alpha < expected { + "WRITE/decode bug: fullScan-materialized alpha below expected" + } else { + "NO REPRO: filter == fullScan == expected" + }; + println!("VERDICT: {verdict}"); + + if keep.is_none() && read_only.is_none() { + drop(std::fs::remove_file(&path)); + } + Ok(()) +}