diff --git a/Cargo.lock b/Cargo.lock index 16b0a83f9ac..7767fa1a5b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9979,6 +9979,7 @@ dependencies = [ "async-fs", "async-lock", "async-trait", + "codspeed-divan-compat", "futures", "jni", "object_store", diff --git a/java/settings.gradle.kts b/java/settings.gradle.kts index a601cfa9488..a71a1b9fd77 100644 --- a/java/settings.gradle.kts +++ b/java/settings.gradle.kts @@ -17,8 +17,10 @@ toolchainManagement { rootProject.name = "vortex-root" -// API bindings +// API bindings (JMH benchmarks live in vortex-jni's `jmh` source set; see vortex-jni/BENCHMARKS.md) include("vortex-jni") + +// Spark integration include("vortex-spark_2.12") project(":vortex-spark_2.12").projectDir = file("vortex-spark") diff --git a/java/vortex-jni/build.gradle.kts b/java/vortex-jni/build.gradle.kts index 8f5106a1262..146cc9559f9 100644 --- a/java/vortex-jni/build.gradle.kts +++ b/java/vortex-jni/build.gradle.kts @@ -2,13 +2,16 @@ // SPDX-FileCopyrightText: Copyright the Vortex contributors import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar +import net.ltgt.gradle.errorprone.errorprone import org.gradle.api.tasks.Copy import org.gradle.api.tasks.Exec +import org.gradle.kotlin.dsl.support.serviceOf plugins { `java-library` `jvm-test-suite` id("com.gradleup.shadow") version "9.4.2" + id("me.champeau.jmh") version "0.7.3" } dependencies { @@ -93,8 +96,9 @@ tasks.withType().all { ) } -// shade guava and arrow dependencies -tasks.withType { +// shade guava and arrow dependencies in the published jar only. The JMH benchmark links the real +// (unrelocated) Arrow classes, so its jar must not be relocated — scope this to the `shadowJar` task. +tasks.named("shadowJar") { relocate("com.google.common", "dev.vortex.relocated.com.google.common") relocate("org.apache.arrow", "dev.vortex.relocated.org.apache.arrow") { // exclude C Data Interface since JNI cannot be relocated @@ -216,4 +220,140 @@ tasks.register("generateJniHeaders") { dependsOn("compileJava") } +// --------------------------------------------------------------------------- +// JMH benchmarks (src/jmh). See BENCHMARKS.md. +// +// The read-boundary benchmark is meaningless against a debug native lib, so the `jmh` task builds +// and stages the release_debug cdylib itself (buildJmhNativeLib) rather than reusing the dev +// `makeTestFiles` debug build. The benchmark links the real Arrow classes off the runtime classpath +// (it is not run from the relocated shadowJar), so no relocation applies to it. +// --------------------------------------------------------------------------- +// Shared canonical benchmark file, generated by the Rust side and read by BOTH the JMH benchmark and +// the Rust `read_boundary` Divan bench so the two measure reads of the exact same bytes. +val workspaceRoot = rootProject.projectDir.absoluteFile.parentFile +val benchFile = workspaceRoot.resolve("target/vortex-jni-bench/data.vortex") + +jmh { + jmhVersion.set("1.37") + // These reach the forked benchmark JVM. The Arrow C Data Interface needs the --add-opens; the + // system property points the benchmark at the shared canonical file. + jvmArgsAppend.addAll( + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + "-Dvortex.jni.bench.file=${benchFile.absolutePath}", + ) +} + +// Generate the shared canonical .vortex file via the Rust generator example. Idempotent: skipped +// while the file exists (delete it to regenerate). Both `jmh` and the Rust bench read this file. +val generateBenchFile = + tasks.register("generateBenchFile") { + description = "Generate the shared canonical .vortex file read by the JMH and Rust read benchmarks" + group = "verification" + + outputs.file(benchFile) + + doLast { + benchFile.parentFile.mkdirs() + serviceOf().exec { + workingDir = workspaceRoot + executable = "cargo" + args( + "run", + "--profile", + "release_debug", + "--quiet", + "--package", + "vortex-jni", + "--example", + "gen_bench_data", + "--", + benchFile.absolutePath, + ) + } + } + } + +// JMH benchmark classes/methods must be public and non-final, which the nopen checker forbids, and +// the generated JMH glue trips error-prone under -Werror. Relax both for the jmh source set only; +// main and test keep full strictness. +tasks.withType().configureEach { + if (name.lowercase().contains("jmh")) { + options.errorprone.enabled.set(false) + options.compilerArgs.remove("-Werror") + } +} + +// Skip the redundant debug `makeTestFiles` build when this invocation runs the benchmark; the +// benchmark consumes the release_debug lib staged by buildJmhNativeLib instead. +val benchmarkRequested = objects.property(Boolean::class.java).convention(false) +gradle.taskGraph.whenReady { + benchmarkRequested.set(allTasks.any { it.project == project && (it.name == "jmh" || it.name == "jmhJar") }) +} +tasks.named("makeTestFiles").configure { + onlyIf { !benchmarkRequested.get() } +} + +val buildJmhNativeLib = + tasks.register("buildJmhNativeLib") { + description = "Build the release_debug vortex-jni cdylib and stage it for the JMH benchmark" + group = "verification" + + // Stage on top of the processed resources so the benchmark loads it from the runtime classpath. + dependsOn("processResources") + + doLast { + val workspaceRoot = rootProject.projectDir.absoluteFile.parentFile + + serviceOf().exec { + workingDir = workspaceRoot + executable = "cargo" + args("build", "--profile", "release_debug", "--package", "vortex-jni") + } + + val osName = System.getProperty("os.name").lowercase() + val osArch = System.getProperty("os.arch").lowercase() + val osShortName = + when { + osName.contains("mac") -> "darwin" + osName.contains("nix") || osName.contains("nux") -> "linux" + osName.contains("win") -> "win" + else -> throw GradleException("Unsupported OS for buildJmhNativeLib: $osName") + } + val libExt = + when (osShortName) { + "darwin" -> ".dylib" + "linux" -> ".so" + "win" -> ".dll" + else -> throw GradleException("Unsupported OS short name: $osShortName") + } + + copy { + from("$workspaceRoot/target/release_debug/libvortex_jni$libExt") + into(layout.buildDirectory.dir("resources/main/native/$osShortName-$osArch")) + } + } + } + +tasks.named("jmh").configure { + dependsOn(buildJmhNativeLib) + dependsOn(generateBenchFile) +} +tasks.named("jmhJar").configure { dependsOn(buildJmhNativeLib) } + +// Standalone read-batch-granularity diagnostic (VortexJniBatchDiagnostic, not a JMH benchmark). Run +// it off the jmh runtime classpath, which carries the real Arrow classes and the staged +// release_debug lib (me.champeau.jmh's fat `jmhJar` does not bundle deps under com.gradleup.shadow). +tasks.register("batchDiagnostic") { + description = "Run the standalone read-batch-granularity diagnostic (VortexJniBatchDiagnostic)" + group = "verification" + dependsOn("buildJmhNativeLib") + classpath = sourceSets["jmh"].runtimeClasspath + mainClass.set("dev.vortex.bench.VortexJniBatchDiagnostic") + jvmArgs( + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + ) +} + description = "JNI bindings for the Vortex format" diff --git a/java/vortex-jni/src/jmh/java/dev/vortex/bench/BenchData.java b/java/vortex-jni/src/jmh/java/dev/vortex/bench/BenchData.java new file mode 100644 index 00000000000..7c243ee9781 --- /dev/null +++ b/java/vortex-jni/src/jmh/java/dev/vortex/bench/BenchData.java @@ -0,0 +1,100 @@ +// 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.Session; +import dev.vortex.api.VortexWriter; +import java.util.HashMap; +import java.util.List; +import java.util.Random; +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.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.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; + +/** + * The synthetic table shape shared across the JNI benchmarks: six columns (2× int64, 2× float64, 2× Utf8View) over + * {@link #ROWS} rows. {@code id} is sequential, {@code cat} is a periodic low-cardinality column kept non-null so a + * {@code cat='alpha'} filter has selectivity exactly {@code 1/|CATS|}, and {@code tag} is high-cardinality with a 10% + * null rate to exercise a validity buffer. + * + *

{@link #ROWS} and {@link #CATS} must stay in lockstep with the Rust generator in + * {@code vortex-jni/benches/jni_bench_data/mod.rs}, whose output {@link VortexJniReadBenchmark} reads. + * {@link #writeTable} is used only by {@link VortexJniBatchDiagnostic}, which needs to write at several chunk sizes. + */ +final class BenchData { + + static final int ROWS = 2_000_000; + static final String[] CATS = { + "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", + "india", "juliet", "kilo", "lima", "mike", "november", "oscar", "papa" + }; + + private BenchData() {} + + 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))); + } + + 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 stays non-null and deterministic so filter selectivity is exactly 1/|CATS|. + cat.setSafe(i, CATS[(int) (r % CATS.length)].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); + ArrowSchema sch = ArrowSchema.allocateNew(allocator)) { + Data.exportVectorSchemaRoot(allocator, root, null, arr, sch); + writer.writeBatch(arr.memoryAddress(), sch.memoryAddress()); + } + written += batch; + } + } + } +} diff --git a/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniBatchDiagnostic.java b/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniBatchDiagnostic.java new file mode 100644 index 00000000000..1b1658de678 --- /dev/null +++ b/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniBatchDiagnostic.java @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.bench; + +import dev.vortex.api.DataSource; +import dev.vortex.api.Partition; +import dev.vortex.api.Scan; +import dev.vortex.api.ScanOptions; +import dev.vortex.api.Session; +import dev.vortex.jni.NativeLoader; +import java.nio.file.Files; +import java.nio.file.Path; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.ipc.ArrowReader; + +/** + * Standalone diagnostic (not a JMH benchmark): writes the shared {@link BenchData} table at several writer chunk sizes + * and prints the resulting read-batch row-count distribution, showing that Vortex coalesces to a stable read-batch + * granularity (~64K rows) independent of how the file was written. + * + *

Run it with {@code ./gradlew :vortex-jni:batchDiagnostic}. + */ +public final class VortexJniBatchDiagnostic { + + private VortexJniBatchDiagnostic() {} + + public static void main(String[] args) throws Exception { + NativeLoader.loadJni(); + 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"); + Files.deleteIfExists(f); + String uri = f.toAbsolutePath().toUri().toString(); + BenchData.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); + } + } +} diff --git a/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java b/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java new file mode 100644 index 00000000000..29c5847b344 --- /dev/null +++ b/java/vortex-jni/src/jmh/java/dev/vortex/bench/VortexJniReadBenchmark.java @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.bench; + +import dev.vortex.api.DataSource; +import dev.vortex.api.Expression; +import dev.vortex.api.Scan; +import dev.vortex.api.ScanOptions; +import dev.vortex.api.Session; +import dev.vortex.jni.NativeLoader; +import dev.vortex.jni.NativeRuntime; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.ipc.ArrowReader; +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.OperationsPerInvocation; +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.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). + * + *

Every invocation scans the full {@link BenchData#ROWS}-row table, so {@code @OperationsPerInvocation(ROWS)} makes + * JMH report input rows scanned per second directly. Each lane materializes the result batches and sums their + * {@code getRowCount()} — no per-value work — so the numbers reflect scan + boundary cost rather than JVM-side + * arithmetic. Vortex coalesces to ~64K-row read batches regardless of the writer's chunk size (see + * {@link VortexJniBatchDiagnostic}), so boundary cost is amortized over large batches by construction. + * + *

    + *
  • {@code fullScan} — read all six columns. + *
  • {@code projection} — native projection of {@code id, y} (two of six columns). + *
  • {@code selectiveFilter} — native filter {@code cat = 'alpha'} (~1/16 selectivity). + *
+ * + *

This reads a shared canonical file generated by the Rust side ({@code :vortex-jni:generateBenchFile}), so + * these {@code ops/s} are directly comparable to the Rust {@code read_boundary} Divan bench, which runs the same lanes + * over the same bytes through the same native code — the difference is the JNI + Arrow C Data boundary cost. The Gradle + * {@code jmh} task passes the file path via {@code -Dvortex.jni.bench.file}; running the benchmark by hand requires + * setting it (or {@code VORTEX_JNI_BENCH_FILE}). + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@OperationsPerInvocation(BenchData.ROWS) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 2) +@Fork(1) +@Threads(1) +@State(Scope.Benchmark) +public class VortexJniReadBenchmark { + + /** System property (set by the Gradle {@code jmh} task) pointing at the shared canonical {@code .vortex} file. */ + static final String BENCH_FILE_PROPERTY = "vortex.jni.bench.file"; + + /** + * Background worker threads driving the JVM-wide pool: {@code 0} keeps the reading thread the only driver + * (single-threaded), {@code -1} sizes the pool to the available parallelism so the scan's split tasks decode in + * parallel in the background. + */ + @Param({"0", "-1"}) + public int workerThreads; + + BufferAllocator allocator; + Session session; + DataSource dataSource; + + @Setup(Level.Trial) + public void setup() throws Exception { + NativeLoader.loadJni(); + if (workerThreads < 0) { + NativeRuntime.setWorkerThreadsToAvailableParallelism(); + } else { + NativeRuntime.setWorkerThreads(workerThreads); + } + allocator = new RootAllocator(Long.MAX_VALUE); + session = Session.create(); + String uri = benchFile().toUri().toString(); + dataSource = DataSource.open(session, uri); + } + + /** Locate the shared canonical file (generated by {@code :vortex-jni:generateBenchFile}); fail loudly if absent. */ + private static Path benchFile() { + String configured = System.getProperty(BENCH_FILE_PROPERTY); + if (configured == null || configured.isBlank()) { + configured = System.getenv("VORTEX_JNI_BENCH_FILE"); + } + if (configured == null || configured.isBlank()) { + throw new IllegalStateException("canonical bench file not configured: set -D" + BENCH_FILE_PROPERTY + + " (the Gradle `jmh` task does this). Run `./gradlew :vortex-jni:jmh`."); + } + Path path = Path.of(configured).toAbsolutePath(); + if (!Files.isRegularFile(path)) { + throw new IllegalStateException("canonical bench file does not exist: " + path + + " — run `./gradlew :vortex-jni:generateBenchFile`."); + } + return path; + } + + @TearDown(Level.Trial) + public void teardown() { + // The canonical file is shared with the Rust bench and managed by Gradle, so it is not deleted here. + // Intentionally does not close the allocator either: 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. + dataSource = null; + } + + /** Full scan of all six columns. */ + @Benchmark + public void fullScan(Blackhole bh) throws Exception { + bh.consume(countRows(dataSource.scan(ScanOptions.of()))); + } + + /** 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()); + bh.consume(countRows( + dataSource.scan(ScanOptions.builder().projection(projection).build()))); + } + + /** 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(BenchData.CATS[0])); + bh.consume( + countRows(dataSource.scan(ScanOptions.builder().filter(filter).build()))); + } + + /** Materialize every result batch and return the total row count — no per-value consumption. */ + private long countRows(Scan scan) throws Exception { + long rows = 0; + while (scan.hasNext()) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { + while (reader.loadNextBatch()) { + rows += reader.getVectorSchemaRoot().getRowCount(); + } + } + } + return rows; + } +} diff --git a/vortex-jni/Cargo.toml b/vortex-jni/Cargo.toml index 80f46a67b82..29cdefecdd3 100644 --- a/vortex-jni/Cargo.toml +++ b/vortex-jni/Cargo.toml @@ -38,6 +38,7 @@ vortex-object-store-opendal = { path = "../vortex-object-store-opendal", optiona vortex-parquet-variant = { workspace = true } [dev-dependencies] +divan = { workspace = true } jni = { workspace = true, features = ["invocation"] } [features] @@ -46,7 +47,11 @@ jni = { workspace = true, features = ["invocation"] } opendal = ["dep:vortex-object-store-opendal"] [lib] -crate-type = ["cdylib"] +crate-type = ["cdylib", "rlib"] + +[[bench]] +name = "read_boundary" +harness = false [lints] workspace = true diff --git a/vortex-jni/benches/jni_bench_data/mod.rs b/vortex-jni/benches/jni_bench_data/mod.rs new file mode 100644 index 00000000000..9e48cb58ee8 --- /dev/null +++ b/vortex-jni/benches/jni_bench_data/mod.rs @@ -0,0 +1,136 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Deterministic generator for the canonical read-boundary benchmark file. +//! +//! Both the `read_boundary` Divan bench and the `gen_bench_data` example write/read the SAME file +//! so the Rust "native floor" and the Java JMH benchmark (`VortexJniReadBenchmark`) measure reads of +//! the exact same bytes. The Gradle `generateBenchFile` task runs the example to produce it, then +//! points the JMH fork at it via `-Dvortex.jni.bench.file`. +//! +//! The table is six columns over [`ROWS`] rows — `id`, `x` (int64), `y`, `z` (float64), and `cat`, +//! `tag` (Utf8View) — generated by a fixed formula (no RNG) so the file is reproducible: `id` is +//! sequential, `cat` is a periodic 16-value low-cardinality column kept non-null so `cat = 'alpha'` +//! has selectivity exactly `1/|CATS|`, and `tag` is high-cardinality with a 10% null rate. + +use std::path::Path; +use std::path::PathBuf; + +use vortex::VortexSessionDefault; +use vortex::array::ArrayRef; +use vortex::array::IntoArray; +use vortex::array::arrays::ChunkedArray; +use vortex::array::arrays::PrimitiveArray; +use vortex::array::arrays::StructArray; +use vortex::array::arrays::VarBinViewArray; +use vortex::array::validity::Validity; +use vortex::dtype::FieldNames; +use vortex::error::VortexResult; +use vortex::error::vortex_err; +use vortex::file::WriteOptionsSessionExt; +use vortex::io::runtime::BlockingRuntime; +use vortex::io::runtime::current::CurrentThreadRuntime; +use vortex::io::session::RuntimeSessionExt; +use vortex::session::VortexSession; + +/// Rows in the canonical table. Must match the JMH side's `BenchData.ROWS`. +pub const ROWS: usize = 2_000_000; +/// Rows per Arrow batch handed to the writer. +pub const WRITE_CHUNK: usize = 65_536; +/// 16 low-cardinality category values; `cat = 'alpha'` matches exactly `ROWS / 16` rows. +pub const CATS: [&str; 16] = [ + "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", "juliet", + "kilo", "lima", "mike", "november", "oscar", "papa", +]; + +/// Absolute path of the canonical file, shared with the Gradle/JMH side. +/// +/// Honors `VORTEX_JNI_BENCH_FILE`; otherwise defaults to `/target/vortex-jni-bench/ +/// data.vortex` (the same location the Gradle task passes), so a standalone `cargo bench` and a +/// `./gradlew :vortex-jni:jmh` run resolve to the same file. +pub fn default_path() -> PathBuf { + if let Ok(path) = std::env::var("VORTEX_JNI_BENCH_FILE") { + return PathBuf::from(path); + } + Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/vortex-jni-bench/data.vortex") +} + +/// Build one `[start, end)` slice of the canonical table as a Vortex struct array. +fn build_chunk(start: usize, end: usize) -> ArrayRef { + let id = PrimitiveArray::from_iter((start..end).map(|r| r as i64)).into_array(); + // Deterministic spread across [0, 1_000_000); a fixed multiplicative hash, no RNG. + let xs = PrimitiveArray::from_iter( + (start..end).map(|r| (r as i64).wrapping_mul(2_654_435_761).rem_euclid(1_000_000)), + ) + .into_array(); + let ys = + PrimitiveArray::from_iter((start..end).map(|r| (r % 1000) as f64 / 1000.0)).into_array(); + let zs = PrimitiveArray::from_iter((start..end).map(|r| ((r / 1000) % 1000) as f64 / 1000.0)) + .into_array(); + // cat stays non-null and periodic so filter selectivity is exactly 1/|CATS|. + let cat = + VarBinViewArray::from_iter_str((start..end).map(|r| CATS[r % CATS.len()])).into_array(); + // tag carries nulls (every 10th row) and high-cardinality values to exercise a validity buffer. + let tag = VarBinViewArray::from_iter_nullable_str((start..end).map(|r| { + if r % 10 == 0 { + None + } else { + Some(r.to_string()) + } + })) + .into_array(); + + StructArray::new( + FieldNames::from(["id", "x", "y", "z", "cat", "tag"]), + vec![id, xs, ys, zs, cat, tag], + end - start, + Validity::NonNullable, + ) + .into_array() +} + +/// Build the canonical six-column table as a chunked array of [`WRITE_CHUNK`]-row chunks. +fn build_table() -> VortexResult { + let mut chunks = Vec::new(); + let mut start = 0; + while start < ROWS { + let end = (start + WRITE_CHUNK).min(ROWS); + chunks.push(build_chunk(start, end)); + start = end; + } + let dtype = chunks[0].dtype().clone(); + Ok(ChunkedArray::try_new(chunks, dtype)?.into_array()) +} + +/// Write the canonical file to `path`, creating parent directories as needed. +pub fn write_canonical(path: &Path) -> VortexResult<()> { + let runtime = CurrentThreadRuntime::new(); + let session = VortexSession::default().with_handle(runtime.handle()); + let table = build_table()?; + + let mut bytes: Vec = Vec::new(); + runtime.block_on(async { + session + .write_options() + .write(&mut bytes, table.to_array_stream()) + .await + })?; + + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .map_err(|e| vortex_err!("failed to create {}: {e}", parent.display()))?; + } + std::fs::write(path, &bytes) + .map_err(|e| vortex_err!("failed to write {}: {e}", path.display()))?; + Ok(()) +} + +/// Write the canonical file only if it does not already exist (idempotent), so the bench and the +/// Gradle generator share one file rather than racing to overwrite it. +pub fn ensure_canonical(path: &Path) -> VortexResult<()> { + let present = path.metadata().map(|m| m.len() > 0).unwrap_or(false); + if present { + return Ok(()); + } + write_canonical(path) +} diff --git a/vortex-jni/benches/read_boundary.rs b/vortex-jni/benches/read_boundary.rs new file mode 100644 index 00000000000..400e5786184 --- /dev/null +++ b/vortex-jni/benches/read_boundary.rs @@ -0,0 +1,262 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Native "floor" for the `VortexJniReadBenchmark` JMH lanes. +//! +//! Reads the same canonical file (see [`jni_bench_data`]) on the same [`RUNTIME`]/[`POOL`] statics, +//! in two variants: +//! +//! - `partitions` calls [`partition_record_batches`], the function the `NativePartition.scanArrow` +//! entry point calls, so the gap to the JMH `ops/s` is the JNI crossing and the Arrow C Data +//! export and nothing else. +//! - `scan_builder` reads the same file through `ScanBuilder` with the Arrow conversion mapped +//! inside the split tasks, which is how native Rust callers scan. The gap between the two +//! variants is what the JNI's partition-at-a-time consumption costs. +//! +//! `ItemsCount::new(ROWS)` matches `@OperationsPerInvocation(ROWS)`, and the `*`/`*_pooled` pairs +//! match the JMH `workerThreads` `0`/`-1` params. + +#![expect(clippy::unwrap_used)] + +mod jni_bench_data; + +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::cast::AsArray; +use arrow_schema::Field; +use divan::Bencher; +use divan::counter::ItemsCount; +use futures::StreamExt; +use vortex::array::VortexSessionExecute; +use vortex::arrow::ArrowSessionExt; +use vortex::dtype::FieldName; +use vortex::error::VortexResult; +use vortex::expr::Expression; +use vortex::expr::get_item; +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::scalar_fn::ScalarFnVTableExt; +use vortex::scalar_fn::fns::binary::Binary; +use vortex::scalar_fn::fns::operators::Operator; +use vortex::scan::DataSourceRef; +use vortex::scan::ScanRequest; +use vortex::scan::selection::Selection; +use vortex::session::VortexSession; +use vortex::utils::aliases::hash_map::HashMap; +use vortex_jni::POOL; +use vortex_jni::RUNTIME; +use vortex_jni::new_session; +use vortex_jni::open_data_source; +use vortex_jni::partition_record_batches; + +use crate::jni_bench_data::ROWS; + +fn main() { + divan::main(); +} + +/// The lanes, mirroring `VortexJniReadBenchmark`. +#[derive(Clone, Copy)] +enum Lane { + /// Read all six columns. + FullScan, + /// Native projection of `id, y`. + Projection, + /// Native filter `cat = 'alpha'` (~1/16 selectivity). + SelectiveFilter, +} + +/// Opened once per Divan sample, outside the timed region. Both variants read the same file: the +/// data source is what Java holds, the [`VortexFile`] is what `ScanBuilder` needs. +struct Env { + session: VortexSession, + data_source: DataSourceRef, + file: VortexFile, +} + +impl Env { + fn open() -> VortexResult { + let path = jni_bench_data::default_path(); + jni_bench_data::ensure_canonical(&path)?; + + let session = *new_session(); + let uri = url::Url::from_file_path(&path) + .unwrap_or_else(|()| unreachable!("canonical path is absolute")) + .to_string(); + let data_source = open_data_source(&session, &[uri], &HashMap::new())?; + let file = RUNTIME.block_on(session.open_options().open_path(&path))?; + Ok(Self { + session, + data_source, + file, + }) + } + + /// The native half of the Java `countRows`: one partition at a time, each consumed to + /// completion through the JNI's own conversion path. + fn run_partitions(&self, lane: Lane) -> VortexResult { + let scan = RUNTIME.block_on(self.data_source.scan(scan_request(lane)))?; + let mut partitions = scan.partitions(); + + let mut rows = 0u64; + while let Some(partition) = RUNTIME.block_on(partitions.next()) { + let (_schema, batches) = partition_record_batches(&self.session, partition?)?; + for batch in batches { + rows += batch?.num_rows() as u64; + } + } + Ok(rows) + } + + /// The same read as a native caller writes it: the Arrow conversion is the scan's `map`, so it + /// runs inside the split tasks rather than behind a second buffered stage. + fn run_scan_builder(&self, lane: Lane) -> VortexResult { + let mut builder = self.file.scan()?.with_ordered(false); + match lane { + Lane::Projection => builder = builder.with_projection(projection_expr()), + Lane::SelectiveFilter => builder = builder.with_filter(filter_expr()), + Lane::FullScan => {} + } + + let schema = self.session.arrow().to_arrow_schema(&builder.dtype()?)?; + let target = Arc::new(Field::new_struct("", schema.fields().clone(), false)); + let session = self.session.clone(); + + let mut rows = 0u64; + for batch in builder + .map(move |array| { + let mut ctx = session.create_execution_ctx(); + let arrow = + session + .arrow() + .execute_arrow(array, Some(target.as_ref()), &mut ctx)?; + Ok(RecordBatch::from(arrow.as_struct().clone()).num_rows() as u64) + }) + .into_iter(&*RUNTIME)? + { + rows += batch?; + } + Ok(rows) + } +} + +fn projection_expr() -> Expression { + select(vec![FieldName::from("id"), FieldName::from("y")], root()) +} + +fn filter_expr() -> Expression { + Binary.new_expr( + Operator::Eq, + [get_item(FieldName::from("cat"), root()), lit("alpha")], + ) +} + +/// What a Java `ScanOptions` produces: all defaults but the projection and filter. +fn scan_request(lane: Lane) -> ScanRequest { + ScanRequest { + projection: match lane { + Lane::Projection => projection_expr(), + _ => root(), + }, + filter: matches!(lane, Lane::SelectiveFilter).then(filter_expr), + row_range: None, + selection: Selection::All, + ordered: false, + limit: None, + partition_selection: Selection::All, + partition_range: None, + } +} + +/// `pooled` maps to the JMH `workerThreads` param: `false` is `0`, `true` is `-1`. +fn run_lane + bencher: Bencher<'_, '_>, + lane: Lane, + pooled: bool, + run: fn(&Env, Lane) -> VortexResult, +) { + if pooled { + POOL.set_workers_to_available_parallelism(); + } else { + POOL.set_workers(0); + } + bencher + .with_inputs(|| Env::open().unwrap()) + .input_counter(|_| ItemsCount::new(ROWS)) + .bench_refs(move |env| run(env, lane).unwrap()); +} + +/// Through the JNI's own partition consumption — the floor for `VortexJniReadBenchmark`. +mod partitions { + use super::*; + + #[divan::bench] + fn full_scan(bencher: Bencher) { + run_lane(bencher, Lane::FullScan, false, Env::run_partitions); + } + + #[divan::bench] + fn projection(bencher: Bencher) { + run_lane(bencher, Lane::Projection, false, Env::run_partitions); + } + + #[divan::bench] + fn selective_filter(bencher: Bencher) { + run_lane(bencher, Lane::SelectiveFilter, false, Env::run_partitions); + } + + #[divan::bench] + fn full_scan_pooled(bencher: Bencher) { + run_lane(bencher, Lane::FullScan, true, Env::run_partitions); + } + + #[divan::bench] + fn projection_pooled(bencher: Bencher) { + run_lane(bencher, Lane::Projection, true, Env::run_partitions); + } + + #[divan::bench] + fn selective_filter_pooled(bencher: Bencher) { + run_lane(bencher, Lane::SelectiveFilter, true, Env::run_partitions); + } +} + +/// Through `ScanBuilder`, as a native Rust caller would scan. +mod scan_builder { + use super::*; + + #[divan::bench] + fn full_scan(bencher: Bencher) { + run_lane(bencher, Lane::FullScan, false, Env::run_scan_builder); + } + + #[divan::bench] + fn projection(bencher: Bencher) { + run_lane(bencher, Lane::Projection, false, Env::run_scan_builder); + } + + #[divan::bench] + fn selective_filter(bencher: Bencher) { + run_lane(bencher, Lane::SelectiveFilter, false, Env::run_scan_builder); + } + + #[divan::bench] + fn full_scan_pooled(bencher: Bencher) { + run_lane(bencher, Lane::FullScan, true, Env::run_scan_builder); + } + + #[divan::bench] + fn projection_pooled(bencher: Bencher) { + run_lane(bencher, Lane::Projection, true, Env::run_scan_builder); + } + + #[divan::bench] + fn selective_filter_pooled(bencher: Bencher) { + run_lane(bencher, Lane::SelectiveFilter, true, Env::run_scan_builder); + } +} diff --git a/vortex-jni/examples/gen_bench_data.rs b/vortex-jni/examples/gen_bench_data.rs new file mode 100644 index 00000000000..5651d54a458 --- /dev/null +++ b/vortex-jni/examples/gen_bench_data.rs @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Generates the canonical read-boundary benchmark file shared by the Rust `read_boundary` bench +//! and the Java `VortexJniReadBenchmark` JMH benchmark. +//! +//! Run directly (`cargo run -p vortex-jni --example gen_bench_data -- [PATH]`) or via the Gradle +//! `generateBenchFile` task. With no argument it writes to [`jni_bench_data::default_path`]. Generation +//! is idempotent: an existing non-empty file is left untouched so both benches read the same bytes. + +#[path = "../benches/jni_bench_data/mod.rs"] +mod jni_bench_data; + +use std::path::PathBuf; +use std::process::ExitCode; + +fn main() -> ExitCode { + let path: PathBuf = std::env::args() + .nth(1) + .map(PathBuf::from) + .unwrap_or_else(jni_bench_data::default_path); + + let existed = path.metadata().map(|m| m.len() > 0).unwrap_or(false); + match jni_bench_data::ensure_canonical(&path) { + Ok(()) => { + if existed { + println!("canonical bench file already present: {}", path.display()); + } else { + println!( + "wrote canonical bench file ({} rows): {}", + jni_bench_data::ROWS, + path.display() + ); + } + ExitCode::SUCCESS + } + Err(e) => { + eprintln!("failed to generate canonical bench file: {e}"); + ExitCode::FAILURE + } + } +} diff --git a/vortex-jni/src/data_source.rs b/vortex-jni/src/data_source.rs index 516a0aa7f55..f7580b2782c 100644 --- a/vortex-jni/src/data_source.rs +++ b/vortex-jni/src/data_source.rs @@ -30,6 +30,7 @@ use vortex::io::filesystem::FileSystemRef; use vortex::io::runtime::BlockingRuntime; use vortex::io::session::RuntimeSessionExt; use vortex::scan::DataSourceRef; +use vortex::session::VortexSession; use vortex::utils::aliases::hash_map::HashMap; use vortex_arrow::ArrowSessionExt; @@ -84,39 +85,50 @@ pub extern "system" fn Java_dev_vortex_jni_NativeDataSource_open( glob_strings.push(uri.to_owned()); } } - if glob_strings.is_empty() { - return Err(vortex_err!("no paths provided").into()); - } + let inner = open_data_source(session, &glob_strings, &properties)?; + Ok(Box::new(NativeDataSource { inner }).into_raw()) + }) +} - let glob_urls: Vec = glob_strings - .iter() - .map(|g| parse_uri_or_path(g.as_str())) - .collect::>()?; +/// Open a data source over `globs`, resolving one object-store filesystem per base URL. +/// +/// Shared with the `read_boundary` benchmark so it opens its data source exactly as Java does. +pub fn open_data_source( + session: &VortexSession, + globs: &[String], + properties: &HashMap, +) -> VortexResult { + if globs.is_empty() { + return Err(vortex_err!("no paths provided")); + } - let mut fs_cache: HashMap = HashMap::new(); - for glob_url in &glob_urls { - let base = base_url(glob_url); - if !fs_cache.contains_key(&base) { - let fs = object_store_fs(glob_url, &properties, session.handle())?; - fs_cache.insert(base, fs); - } - } + let glob_urls: Vec = globs + .iter() + .map(|g| parse_uri_or_path(g.as_str())) + .collect::>()?; - let mut builder = MultiFileDataSource::new(session.clone()); - for glob_url in &glob_urls { - let base = base_url(glob_url); - let fs = fs_cache - .get(&base) - .cloned() - .unwrap_or_else(|| unreachable!("fs cached for every base url")); - builder = builder.with_glob(glob_url.path(), Some(fs)); + let mut fs_cache: HashMap = HashMap::new(); + for glob_url in &glob_urls { + let base = base_url(glob_url); + if !fs_cache.contains_key(&base) { + let fs = object_store_fs(glob_url, properties, session.handle())?; + fs_cache.insert(base, fs); } + } - let inner = RUNTIME - .block_on(builder.build()) - .map(|ds| Arc::new(ds) as DataSourceRef)?; - Ok(Box::new(NativeDataSource { inner }).into_raw()) - }) + let mut builder = MultiFileDataSource::new(session.clone()); + for glob_url in &glob_urls { + let base = base_url(glob_url); + let fs = fs_cache + .get(&base) + .cloned() + .unwrap_or_else(|| unreachable!("fs cached for every base url")); + builder = builder.with_glob(glob_url.path(), Some(fs)); + } + + RUNTIME + .block_on(builder.build()) + .map(|ds| Arc::new(ds) as DataSourceRef) } /// Open a data source over caller-provided `dev.vortex.io.NativeReadable` objects. diff --git a/vortex-jni/src/lib.rs b/vortex-jni/src/lib.rs index 4e74eb304ce..03899f70a71 100644 --- a/vortex-jni/src/lib.rs +++ b/vortex-jni/src/lib.rs @@ -6,6 +6,9 @@ //! The JNI surface mirrors the C FFI in `vortex-ffi` closely. It exposes a small //! session-oriented scan API (session → data source → scan → partition → Arrow //! array stream) so that Java callers only see Arrow at the boundary. +//! +//! The crate is an `rlib` as well as a `cdylib` so the `read_boundary` benchmark can call the +//! native half of that pipeline directly and measure the floor of the work Java drives. use std::sync::LazyLock; @@ -30,14 +33,18 @@ mod scan; mod session; mod writer; +pub use crate::data_source::open_data_source; +pub use crate::scan::partition_record_batches; +pub use crate::session::new_session; + /// Shared current-thread runtime backing every JNI call. Using a current-thread /// runtime (as opposed to multi-thread Tokio) keeps the Java side responsible for /// parallelism decisions — each partition is consumed on the caller's thread, and /// writes are bounded by a small in-flight queue on the same thread. -static RUNTIME: LazyLock = LazyLock::new(CurrentThreadRuntime::new); +pub static RUNTIME: LazyLock = LazyLock::new(CurrentThreadRuntime::new); /// Shared worker pool that can drive [`RUNTIME`]'s executor in the background. Callers /// configure its size through the `NativeRuntime.setWorkerThreads` JNI entry point. By /// default the pool has zero workers — nothing is driven unless a Java thread calls /// the blocking API or workers are added here. -pub(crate) static POOL: LazyLock = LazyLock::new(|| RUNTIME.new_pool()); +pub static POOL: LazyLock = LazyLock::new(|| RUNTIME.new_pool()); diff --git a/vortex-jni/src/scan.rs b/vortex-jni/src/scan.rs index 82fa23fa650..c846959a334 100644 --- a/vortex-jni/src/scan.rs +++ b/vortex-jni/src/scan.rs @@ -21,6 +21,7 @@ use arrow_array::ffi::FFI_ArrowSchema; use arrow_array::ffi_stream::FFI_ArrowArrayStream; use arrow_schema::ArrowError; use arrow_schema::Field; +use arrow_schema::SchemaRef; use futures::StreamExt; use jni::EnvUnowned; use jni::objects::JByteArray; @@ -43,6 +44,7 @@ use vortex::scan::PartitionRef; use vortex::scan::PartitionStream; use vortex::scan::ScanRequest; use vortex::scan::selection::Selection; +use vortex::session::VortexSession; use vortex_arrow::ArrowSessionExt; use crate::POOL; @@ -318,6 +320,50 @@ pub extern "system" fn Java_dev_vortex_jni_NativePartition_rowCount( }); } +/// Consume `partition` into Arrow record batches and their schema. +/// +/// The whole native side of `NativePartition.scanArrow`; the entry point below only wraps the +/// iterator in an `FFI_ArrowArrayStream`. Also called directly by the `read_boundary` benchmark. +pub fn partition_record_batches( + session: &VortexSession, + partition: PartitionRef, +) -> VortexResult<( + SchemaRef, + impl Iterator> + use<>, +)> { + let array_stream = partition.execute()?; + let dtype = array_stream.dtype().clone(); + + let schema = Arc::new(session.arrow().to_arrow_schema(&dtype)?); + let target = Arc::new(Field::new_struct("", schema.fields().clone(), false)); + let session = session.clone(); + + let iter = RUNTIME + .block_on_stream_thread_safe(|handle| { + array_stream + .map(move |chunk| { + let session = session.clone(); + let target = Arc::clone(&target); + handle.spawn(async move { + let chunk = chunk?; + let mut ctx = session.create_execution_ctx(); + let arrow = session.arrow().execute_arrow( + chunk, + Some(target.as_ref()), + &mut ctx, + )?; + Ok(RecordBatch::from(arrow.as_struct().clone())) + }) + }) + .buffered(POOL.worker_count().max(1)) + }) + .map(|result: VortexResult| { + result.map_err(|e| ArrowError::ExternalError(Box::new(e))) + }); + + Ok((schema, iter)) +} + /// Consume a partition into the `FFI_ArrowArrayStream` pointed to by `stream_addr`. The /// partition pointer is invalidated by this call; Java must not `free` it afterwards. /// @@ -345,36 +391,8 @@ pub extern "system" fn Java_dev_vortex_jni_NativePartition_scanArrow( _ => throw_runtime!("partition already consumed"), }; - let array_stream = partition.execute()?; - let dtype = array_stream.dtype().clone(); - let session = unsafe { session_ref(session_ptr) }; - let schema = Arc::new(session.arrow().to_arrow_schema(&dtype)?); - let target = Arc::new(Field::new_struct("", schema.fields().clone(), false)); - - let iter = RUNTIME - .block_on_stream_thread_safe(|handle| { - array_stream - .map(move |chunk| { - let session = session.clone(); - let target = Arc::clone(&target); - handle.spawn(async move { - let chunk = chunk?; - let mut ctx = session.create_execution_ctx(); - let arrow = session.arrow().execute_arrow( - chunk, - Some(target.as_ref()), - &mut ctx, - )?; - Ok(RecordBatch::from(arrow.as_struct().clone())) - }) - }) - .buffered(POOL.worker_count().max(1)) - }) - .map(|result: VortexResult| { - result.map_err(|e| ArrowError::ExternalError(Box::new(e))) - }); - + let (schema, iter) = partition_record_batches(session, partition)?; let reader = RecordBatchIteratorAdapter::new(iter, schema); let arrow_stream = FFI_ArrowArrayStream::new(Box::new(reader)); unsafe { diff --git a/vortex-jni/src/session.rs b/vortex-jni/src/session.rs index 881babc5dea..9a7bc75293b 100644 --- a/vortex-jni/src/session.rs +++ b/vortex-jni/src/session.rs @@ -13,9 +13,10 @@ use vortex::session::VortexSession; use crate::RUNTIME; -/// Constructs a fresh [`VortexSession`] bound to the JNI-shared tokio runtime and returns -/// an opaque pointer that Java must pass to [`Java_dev_vortex_jni_NativeSession_free`]. -pub(crate) fn new_session() -> Box { +/// Constructs the [`VortexSession`] every Java `Session.create()` gets: bound to the JNI-shared +/// runtime, with the extension plugins registered. Java frees it via +/// [`Java_dev_vortex_jni_NativeSession_free`]. +pub fn new_session() -> Box { let session = VortexSession::default().with_handle(RUNTIME.handle()); vortex_parquet_variant::initialize(&session); vortex_geo::initialize(&session);