diff --git a/java/vortex-jni/src/main/java/dev/vortex/api/DataSource.java b/java/vortex-jni/src/main/java/dev/vortex/api/DataSource.java index 51657d5a014..b94d5cb3fde 100644 --- a/java/vortex-jni/src/main/java/dev/vortex/api/DataSource.java +++ b/java/vortex-jni/src/main/java/dev/vortex/api/DataSource.java @@ -73,9 +73,12 @@ public static DataSource open(Session session, List uris, Map readables, i lengths[i] = readable.length(); Preconditions.checkArgument(lengths[i] >= 0, "readable for %s reported negative length", names[i]); } - long pointer = - NativeDataSource.openFiles(session.nativePointer(), readableArray, names, lengths, readConcurrency); - return new DataSource(session, pointer); + try { + long pointer = + NativeDataSource.openFiles(session.nativePointer(), readableArray, names, lengths, readConcurrency); + return new DataSource(session, pointer); + } finally { + Reference.reachabilityFence(session); + } } /** Arrow schema of the data source (and of scans produced from it). */ @@ -247,12 +254,12 @@ public OptionalLong asOptional() { /** Submit a scan. */ public synchronized Scan scan(ScanOptions options) { Preconditions.checkState(!closed, "data source is closed"); + Objects.requireNonNull(options, "options"); + Expression projection = options.projection().orElse(null); + Expression filter = options.filter().orElse(null); try { - Objects.requireNonNull(options, "options"); - - long projectionPtr = - options.projection().map(Expression::nativePointer).orElse(0L); - long filterPtr = options.filter().map(Expression::nativePointer).orElse(0L); + long projectionPtr = projection == null ? 0L : projection.nativePointer(); + long filterPtr = filter == null ? 0L : filter.nativePointer(); long begin = options.rowRangeBegin().orElse(0L); long end = options.rowRangeEnd().orElse(0L); ScanOptions.SelectionMode selectionMode = options.selectionMode(); @@ -276,6 +283,8 @@ public synchronized Scan scan(ScanOptions options) { splitConcurrency); return Scan.fromPointer(session, scanPtr); } finally { + Reference.reachabilityFence(projection); + Reference.reachabilityFence(filter); Reference.reachabilityFence(this); } } diff --git a/java/vortex-jni/src/main/java/dev/vortex/api/Expression.java b/java/vortex-jni/src/main/java/dev/vortex/api/Expression.java index 89ec0c5a0f5..74bf14602fc 100644 --- a/java/vortex-jni/src/main/java/dev/vortex/api/Expression.java +++ b/java/vortex-jni/src/main/java/dev/vortex/api/Expression.java @@ -6,6 +6,7 @@ import com.google.common.base.Preconditions; import dev.vortex.VortexCleaner; import dev.vortex.jni.NativeExpression; +import java.lang.ref.Reference; import java.math.BigInteger; import java.nio.ByteBuffer; import java.nio.ByteOrder; @@ -18,7 +19,8 @@ *

Expressions are composed via the static factories ({@link #root()}, {@link #getItem(String, Expression)}, etc.). * Each returned {@code Expression} owns its native pointer; the pointer is released automatically when the * {@code Expression} is no longer reachable. Passing an expression as an input to a builder does not transfer - * ownership — the resulting expression is an independent copy on the native side. + * ownership — the resulting expression is an independent copy on the native side. Factories keep their input wrappers + * reachable until native code has finished borrowing their pointers. */ public final class Expression { /** Number of bytes in a UUID's big-endian representation. */ @@ -53,7 +55,11 @@ public static Expression rowIdx() { /** Access a named field from a struct expression. */ public static Expression getItem(String fieldName, Expression child) { - return new Expression(NativeExpression.getItem(fieldName, child.nativePointer())); + try { + return new Expression(NativeExpression.getItem(fieldName, child.nativePointer())); + } finally { + Reference.reachabilityFence(child); + } } /** Shortcut for {@code getItem(fieldName, root())}. */ @@ -76,12 +82,20 @@ public static Expression column(String[] fieldNames) { /** Project a subset of fields out of a struct expression. */ public static Expression select(String[] fieldNames, Expression child) { - return new Expression(NativeExpression.select(fieldNames, child.nativePointer())); + try { + return new Expression(NativeExpression.select(fieldNames, child.nativePointer())); + } finally { + Reference.reachabilityFence(child); + } } /** Creates an expression that packs values into a struct with named fields. */ public static Expression pack(String[] fieldNames, Expression[] expressions, boolean nullable) { - return new Expression(NativeExpression.pack(fieldNames, nativePointers(expressions), nullable)); + try { + return new Expression(NativeExpression.pack(fieldNames, nativePointers(expressions), nullable)); + } finally { + Reference.reachabilityFence(expressions); + } } /** @@ -92,7 +106,11 @@ public static Expression pack(String[] fieldNames, Expression[] expressions, boo * an earlier one with the same name. Merging zero expressions yields an empty struct. */ public static Expression merge(DuplicateHandling duplicateHandling, Expression... expressions) { - return new Expression(NativeExpression.merge(nativePointers(expressions), duplicateHandling.tag())); + try { + return new Expression(NativeExpression.merge(nativePointers(expressions), duplicateHandling.tag())); + } finally { + Reference.reachabilityFence(expressions); + } } /** @@ -106,17 +124,30 @@ public static Expression merge(Expression... expressions) { /** Logical AND. Requires at least one operand. */ public static Expression and(Expression... operands) { Preconditions.checkArgument(operands.length > 0, "and requires at least one operand"); - return new Expression(NativeExpression.and(nativePointers(operands))); + try { + return new Expression(NativeExpression.and(nativePointers(operands))); + } finally { + Reference.reachabilityFence(operands); + } } /** Logical OR. Requires at least one operand. */ public static Expression or(Expression... operands) { Preconditions.checkArgument(operands.length > 0, "or requires at least one operand"); - return new Expression(NativeExpression.or(nativePointers(operands))); + try { + return new Expression(NativeExpression.or(nativePointers(operands))); + } finally { + Reference.reachabilityFence(operands); + } } public static Expression binary(BinaryOp op, Expression lhs, Expression rhs) { - return new Expression(NativeExpression.binary(op.code(), lhs.nativePointer(), rhs.nativePointer())); + try { + return new Expression(NativeExpression.binary(op.code(), lhs.nativePointer(), rhs.nativePointer())); + } finally { + Reference.reachabilityFence(lhs); + Reference.reachabilityFence(rhs); + } } /** @@ -130,19 +161,35 @@ public static Expression spatial(SpatialFunction function, Expression... operand function, function.arity(), operands.length); - return new Expression(NativeExpression.spatial(function.code(), nativePointers(operands))); + try { + return new Expression(NativeExpression.spatial(function.code(), nativePointers(operands))); + } finally { + Reference.reachabilityFence(operands); + } } public static Expression not(Expression child) { - return new Expression(NativeExpression.not(child.nativePointer())); + try { + return new Expression(NativeExpression.not(child.nativePointer())); + } finally { + Reference.reachabilityFence(child); + } } public static Expression isNull(Expression child) { - return new Expression(NativeExpression.isNull(child.nativePointer())); + try { + return new Expression(NativeExpression.isNull(child.nativePointer())); + } finally { + Reference.reachabilityFence(child); + } } public static Expression isNotNull(Expression child) { - return new Expression(NativeExpression.isNotNull(child.nativePointer())); + try { + return new Expression(NativeExpression.isNotNull(child.nativePointer())); + } finally { + Reference.reachabilityFence(child); + } } /** @@ -150,7 +197,12 @@ public static Expression isNotNull(Expression child) { * must be a non-nullable boolean. */ public static Expression mask(Expression child, Expression mask) { - return new Expression(NativeExpression.mask(child.nativePointer(), mask.nativePointer())); + try { + return new Expression(NativeExpression.mask(child.nativePointer(), mask.nativePointer())); + } finally { + Reference.reachabilityFence(child); + Reference.reachabilityFence(mask); + } } /** @@ -160,8 +212,13 @@ public static Expression mask(Expression child, Expression mask) { * @param caseInsensitive whether to perform case-insensitive matching ({@code ILIKE}) */ public static Expression like(Expression child, Expression pattern, boolean negated, boolean caseInsensitive) { - return new Expression( - NativeExpression.like(child.nativePointer(), pattern.nativePointer(), negated, caseInsensitive)); + try { + return new Expression( + NativeExpression.like(child.nativePointer(), pattern.nativePointer(), negated, caseInsensitive)); + } finally { + Reference.reachabilityFence(child); + Reference.reachabilityFence(pattern); + } } /** @@ -172,8 +229,14 @@ public static Expression like(Expression child, Expression pattern, boolean nega */ public static Expression between( Expression value, Expression lower, Expression upper, boolean lowerStrict, boolean upperStrict) { - return new Expression(NativeExpression.between( - value.nativePointer(), lower.nativePointer(), upper.nativePointer(), lowerStrict, upperStrict)); + try { + return new Expression(NativeExpression.between( + value.nativePointer(), lower.nativePointer(), upper.nativePointer(), lowerStrict, upperStrict)); + } finally { + Reference.reachabilityFence(value); + Reference.reachabilityFence(lower); + Reference.reachabilityFence(upper); + } } /** @@ -202,8 +265,13 @@ public static Expression listContains(Expression list, Expression needle) { * against an empty list with Vortex's default rules, where the result is {@code false}. */ public static Expression listContains(Expression list, Expression needle, boolean sqlNullSemantics) { - return new Expression( - NativeExpression.listContains(list.nativePointer(), needle.nativePointer(), sqlNullSemantics)); + try { + return new Expression( + NativeExpression.listContains(list.nativePointer(), needle.nativePointer(), sqlNullSemantics)); + } finally { + Reference.reachabilityFence(list); + Reference.reachabilityFence(needle); + } } /** @@ -236,7 +304,12 @@ public static Expression variable(String name) { *

{@code body} may read only {@code param}, so a transform nested in it takes a parameter of its own. */ public static Expression listTransform(Expression list, String param, Expression body) { - return new Expression(NativeExpression.listTransform(list.nativePointer(), param, body.nativePointer())); + try { + return new Expression(NativeExpression.listTransform(list.nativePointer(), param, body.nativePointer())); + } finally { + Reference.reachabilityFence(list); + Reference.reachabilityFence(body); + } } public static Expression literal(boolean value) { @@ -441,14 +514,23 @@ public static Expression nullLiteral(DType dtype) { * {@code elementsNullable}, and every element, which must itself be a literal, is cast to it. */ public static Expression literalList(Expression elementType, boolean elementsNullable, Expression... elements) { - return new Expression(NativeExpression.literalList( - nativePointers(elements), elementType.nativePointer(), elementsNullable, false)); + try { + return new Expression(NativeExpression.literalList( + nativePointers(elements), elementType.nativePointer(), elementsNullable, false)); + } finally { + Reference.reachabilityFence(elements); + Reference.reachabilityFence(elementType); + } } /** Null list literal. See {@link #literalList(Expression, boolean, Expression...)} for the element dtype. */ public static Expression nullLiteralList(Expression elementType, boolean elementsNullable) { - return new Expression( - NativeExpression.literalList(new long[0], elementType.nativePointer(), elementsNullable, true)); + try { + return new Expression( + NativeExpression.literalList(new long[0], elementType.nativePointer(), elementsNullable, true)); + } finally { + Reference.reachabilityFence(elementType); + } } /** @@ -457,15 +539,24 @@ public static Expression nullLiteralList(Expression elementType, boolean element */ public static Expression literalFixedSizeList( Expression elementType, boolean elementsNullable, Expression... elements) { - return new Expression(NativeExpression.literalFixedSizeList( - nativePointers(elements), elementType.nativePointer(), elementsNullable, elements.length, false)); + try { + return new Expression(NativeExpression.literalFixedSizeList( + nativePointers(elements), elementType.nativePointer(), elementsNullable, elements.length, false)); + } finally { + Reference.reachabilityFence(elements); + Reference.reachabilityFence(elementType); + } } /** Null fixed-size list literal of {@code size} elements. */ public static Expression nullLiteralFixedSizeList(Expression elementType, boolean elementsNullable, int size) { Preconditions.checkArgument(size >= 0, "fixed-size list size must not be negative: %s", size); - return new Expression(NativeExpression.literalFixedSizeList( - new long[0], elementType.nativePointer(), elementsNullable, size, true)); + try { + return new Expression(NativeExpression.literalFixedSizeList( + new long[0], elementType.nativePointer(), elementsNullable, size, true)); + } finally { + Reference.reachabilityFence(elementType); + } } /** Create a struct literal. Each field must be a literal expression; its dtype becomes the field's dtype. */ @@ -475,7 +566,11 @@ public static Expression literalStruct(String[] fieldNames, Expression[] fields) "struct literal has %s field names but %s fields", fieldNames.length, fields.length); - return new Expression(NativeExpression.literalStruct(fieldNames, nativePointers(fields), false)); + try { + return new Expression(NativeExpression.literalStruct(fieldNames, nativePointers(fields), false)); + } finally { + Reference.reachabilityFence(fields); + } } /** @@ -488,7 +583,11 @@ public static Expression nullLiteralStruct(String[] fieldNames, Expression[] fie "struct literal has %s field names but %s field types", fieldNames.length, fieldTypes.length); - return new Expression(NativeExpression.literalStruct(fieldNames, nativePointers(fieldTypes), true)); + try { + return new Expression(NativeExpression.literalStruct(fieldNames, nativePointers(fieldTypes), true)); + } finally { + Reference.reachabilityFence(fieldTypes); + } } /** @@ -501,19 +600,36 @@ public static Expression literalMap( Expression keyType, Expression valueType, boolean valuesNullable, Expression[] keys, Expression[] values) { Preconditions.checkArgument( keys.length == values.length, "map literal has %s keys but %s values", keys.length, values.length); - return new Expression(NativeExpression.literalMap( - nativePointers(keys), - nativePointers(values), - keyType.nativePointer(), - valueType.nativePointer(), - valuesNullable, - false)); + try { + return new Expression(NativeExpression.literalMap( + nativePointers(keys), + nativePointers(values), + keyType.nativePointer(), + valueType.nativePointer(), + valuesNullable, + false)); + } finally { + Reference.reachabilityFence(keys); + Reference.reachabilityFence(values); + Reference.reachabilityFence(keyType); + Reference.reachabilityFence(valueType); + } } /** Null map literal. See {@link #literalMap} for how the key and value dtypes are derived. */ public static Expression nullLiteralMap(Expression keyType, Expression valueType, boolean valuesNullable) { - return new Expression(NativeExpression.literalMap( - new long[0], new long[0], keyType.nativePointer(), valueType.nativePointer(), valuesNullable, true)); + try { + return new Expression(NativeExpression.literalMap( + new long[0], + new long[0], + keyType.nativePointer(), + valueType.nativePointer(), + valuesNullable, + true)); + } finally { + Reference.reachabilityFence(keyType); + Reference.reachabilityFence(valueType); + } } private static long[] nativePointers(Expression[] exprs) { diff --git a/java/vortex-jni/src/main/java/dev/vortex/api/Session.java b/java/vortex-jni/src/main/java/dev/vortex/api/Session.java index 073ab2c92e1..a69f447749e 100644 --- a/java/vortex-jni/src/main/java/dev/vortex/api/Session.java +++ b/java/vortex-jni/src/main/java/dev/vortex/api/Session.java @@ -6,6 +6,7 @@ import com.google.common.base.Preconditions; import dev.vortex.VortexCleaner; import dev.vortex.jni.NativeSession; +import java.lang.ref.Reference; import java.util.Set; /** @@ -38,7 +39,11 @@ public static Session create() { *

Select an older edition to write files that older Vortex readers can read. */ public void enableEdition(String edition) { - NativeSession.enableEdition(pointer, edition); + try { + NativeSession.enableEdition(pointer, edition); + } finally { + Reference.reachabilityFence(this); + } } /** @@ -46,10 +51,17 @@ public void enableEdition(String edition) { * {@link #enableEdition} accepts. */ public Set knownEditions() { - return Set.of(NativeSession.knownEditions(pointer)); + try { + return Set.of(NativeSession.knownEditions(pointer)); + } finally { + Reference.reachabilityFence(this); + } } - /** Internal: returns the native pointer. Do not free directly. */ + /** + * Internal: returns the native pointer. Do not free directly. The caller must keep this session reachable while + * native code borrows the pointer. + */ public long nativePointer() { return pointer; } diff --git a/java/vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java b/java/vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java index 9c8318066eb..73517338436 100644 --- a/java/vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java +++ b/java/vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java @@ -9,6 +9,7 @@ import dev.vortex.jni.NativeBuffers; import dev.vortex.jni.NativeWriter; import java.io.IOException; +import java.lang.ref.Reference; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -179,6 +180,8 @@ public VortexWriter build() throws IOException { // Native failures arrive as RuntimeException. Wrap them so callers can handle every // failure to open a writer as an IOException, as writeBatch and finish already do. throw new IOException("failed to create writer for " + destination(), e); + } finally { + Reference.reachabilityFence(session); } if (pointer <= 0) { throw new IOException("failed to create writer for " + destination() + " (ptr=" + pointer + ")"); @@ -226,6 +229,8 @@ public synchronized void writeBatch(VectorSchemaRoot root, BufferAllocator alloc } } catch (RuntimeException e) { throw new IOException("failed to write batch", e); + } finally { + Reference.reachabilityFence(this); } } @@ -241,6 +246,8 @@ public synchronized void writeBatch(long arrowArrayAddr, long arrowSchemaAddr) t ok = NativeWriter.writeBatch(pointer, arrowArrayAddr, arrowSchemaAddr); } catch (RuntimeException e) { throw new IOException("failed to write batch", e); + } finally { + Reference.reachabilityFence(this); } if (!ok) { throw new IOException("failed to write batch"); @@ -259,7 +266,12 @@ public synchronized long bytesWritten() { return summary.fileSize(); } Preconditions.checkState(!closed.get(), "writer closed without a write summary"); - long bytesWritten = NativeWriter.bytesWritten(pointer); + long bytesWritten; + try { + bytesWritten = NativeWriter.bytesWritten(pointer); + } finally { + Reference.reachabilityFence(this); + } Preconditions.checkState(bytesWritten >= 0, "native writer returned an invalid byte count"); return bytesWritten; } @@ -276,7 +288,12 @@ public synchronized long bufferedBytes() { return 0; } Preconditions.checkState(!closed.get(), "writer closed without a write summary"); - long bufferedBytes = NativeWriter.bufferedBytes(pointer); + long bufferedBytes; + try { + bufferedBytes = NativeWriter.bufferedBytes(pointer); + } finally { + Reference.reachabilityFence(this); + } Preconditions.checkState(bufferedBytes >= 0, "native writer returned an invalid buffered byte count"); return bufferedBytes; } diff --git a/java/vortex-jni/src/main/java/dev/vortex/jni/NativeFiles.java b/java/vortex-jni/src/main/java/dev/vortex/jni/NativeFiles.java index 445b0ac7691..d9a7403efbe 100644 --- a/java/vortex-jni/src/main/java/dev/vortex/jni/NativeFiles.java +++ b/java/vortex-jni/src/main/java/dev/vortex/jni/NativeFiles.java @@ -7,6 +7,7 @@ import dev.vortex.api.Session; import dev.vortex.api.VortexColumnStatistics; import dev.vortex.io.NativeReadable; +import java.lang.ref.Reference; import java.util.List; import java.util.Map; import java.util.Objects; @@ -24,12 +25,20 @@ private NativeFiles() {} /** List all Vortex files reachable under {@code uri}. */ public static List listFiles(Session session, String uri, Map options) { - return listFiles(session.nativePointer(), uri, options); + try { + return listFiles(session.nativePointer(), uri, options); + } finally { + Reference.reachabilityFence(session); + } } /** Delete files at the given URIs. Silently tolerates an empty list. */ public static void delete(Session session, String[] uris, Map options) { - delete(session.nativePointer(), uris, options); + try { + delete(session.nativePointer(), uris, options); + } finally { + Reference.reachabilityFence(session); + } } /** @@ -40,7 +49,11 @@ public static void delete(Session session, String[] uris, Map op * default, so a segment that falls outside the file-tail read costs an extra round trip. */ public static Map readMetadata(Session session, String uri, Map options) { - return readMetadata(session.nativePointer(), uri, options); + try { + return readMetadata(session.nativePointer(), uri, options); + } finally { + Reference.reachabilityFence(session); + } } /** @@ -58,7 +71,11 @@ public static Map readMetadata(Session session, NativeReadable r Preconditions.checkArgument(name != null && !name.isEmpty(), "readable reported an empty name"); long length = readable.length(); Preconditions.checkArgument(length >= 0, "readable for %s reported negative length", name); - return readMetadataFromReadable(session.nativePointer(), readable, name, length); + try { + return readMetadataFromReadable(session.nativePointer(), readable, name, length); + } finally { + Reference.reachabilityFence(session); + } } /** @@ -69,7 +86,11 @@ public static Map readMetadata(Session session, NativeReadable r */ public static List readStatistics( Session session, String uri, Map options) { - return List.of(readStatistics(session.nativePointer(), uri, options)); + try { + return List.of(readStatistics(session.nativePointer(), uri, options)); + } finally { + Reference.reachabilityFence(session); + } } /** @@ -87,7 +108,11 @@ public static List readStatistics(Session session, Nativ Preconditions.checkArgument(name != null && !name.isEmpty(), "readable reported an empty name"); long length = readable.length(); Preconditions.checkArgument(length >= 0, "readable for %s reported negative length", name); - return List.of(readStatisticsFromReadable(session.nativePointer(), readable, name, length)); + try { + return List.of(readStatisticsFromReadable(session.nativePointer(), readable, name, length)); + } finally { + Reference.reachabilityFence(session); + } } private static native List listFiles(long sessionPointer, String uri, Map options); diff --git a/java/vortex-jni/src/test/java/dev/vortex/api/ExpressionReachabilityTest.java b/java/vortex-jni/src/test/java/dev/vortex/api/ExpressionReachabilityTest.java new file mode 100644 index 00000000000..05b8f937d63 --- /dev/null +++ b/java/vortex-jni/src/test/java/dev/vortex/api/ExpressionReachabilityTest.java @@ -0,0 +1,163 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.lang.management.ManagementFactory; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.ArrowReader; +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.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** Exercise compiled expression factories while the Cleaner processes their temporary inputs. */ +public final class ExpressionReachabilityTest { + private static final int LIST_SIZE = 2000; + private static final int ITERATIONS = 2000; + private static volatile Expression result; + + @TempDir + Path directory; + + @Test + void temporaryLiteralInputsSurviveNativeCalls() throws Exception { + // A lifetime regression can crash native code, so keep the test runner in a separate JVM. + Path log = directory.resolve("child.log"); + List command = new ArrayList<>(); + command.add(Path.of(System.getProperty("java.home"), "bin", "java").toString()); + command.addAll(List.of( + "-Xms32m", + "-Xmx32m", + "-Xbatch", + "-XX:-TieredCompilation", + "-XX:CompileThreshold=100", + "-XX:CompileCommand=dontinline,dev.vortex.api.Expression::literalList", + "-XX:ErrorFile=" + directory.resolve("hs_err_pid%p.log"))); + ManagementFactory.getRuntimeMXBean().getInputArguments().stream() + .filter(arg -> arg.startsWith("--add-opens=")) + .forEach(command::add); + command.addAll( + List.of("-cp", System.getProperty("java.class.path"), getClass().getName(), directory.toString())); + Process child = new ProcessBuilder(command) + .redirectErrorStream(true) + .redirectOutput(log.toFile()) + .start(); + boolean exited; + try { + exited = child.waitFor(120, TimeUnit.SECONDS); + } finally { + if (child.isAlive()) { + child.destroyForcibly().waitFor(); + } + } + String childOutput = Files.readString(log); + assertTrue(exited, "GC stress child timed out:\n" + childOutput); + assertEquals(0, child.exitValue(), childOutput); + } + + private static Expression buildList() { + Expression[] elements = new Expression[LIST_SIZE]; + for (int i = 0; i < elements.length; i++) { + elements[i] = Expression.literal((long) i); + } + return Expression.literalList(Expression.nullLiteral(Expression.DType.I64), false, elements); + } + + /** Child entry point: validate native list values with a real filtered scan during GC pressure. */ + public static void main(String[] args) throws Exception { + AtomicBoolean running = new AtomicBoolean(true); + Thread gc = new Thread( + () -> { + while (running.get()) { + System.gc(); + try { + Thread.sleep(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + }, + "expression-gc-pressure"); + try (BufferAllocator allocator = new RootAllocator()) { + Session session = Session.create(); + String uri = Path.of(args[0], "values.vortex").toString(); + Schema schema = new Schema(List.of(Field.notNullable("id", new ArrowType.Int(64, true)))); + try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator); + VortexWriter writer = VortexWriter.builder(session, uri, schema, allocator) + .build()) { + BigIntVector ids = (BigIntVector) root.getVector(0); + ids.allocateNew(LIST_SIZE + 1); + for (int i = 0; i <= LIST_SIZE; i++) { + ids.set(i, i); + } + root.setRowCount(LIST_SIZE + 1); + writer.writeBatch(root, allocator); + } + try (DataSource source = DataSource.open(session, uri)) { + long collectionsBefore = collectionCount(); + gc.start(); + try { + for (int i = 0; i < ITERATIONS; i++) { + result = buildList(); + if (i % 100 == 0 || i == ITERATIONS - 1) { + checkValues(source, allocator, result); + } + } + } finally { + running.set(false); + gc.join(); + } + if (collectionCount() <= collectionsBefore) { + throw new AssertionError("no garbage collection occurred during the stress loop"); + } + } + } + System.out.println("completed=" + ITERATIONS); + } + + private static long collectionCount() { + return ManagementFactory.getGarbageCollectorMXBeans().stream() + .mapToLong(collector -> Math.max(0, collector.getCollectionCount())) + .sum(); + } + + private static void checkValues(DataSource source, BufferAllocator allocator, Expression list) throws Exception { + try (Scan scan = source.scan(ScanOptions.builder() + .filter(Expression.in(Expression.column("id"), list)) + .ordered(true) + .build())) { + long expected = 0; + while (scan.hasNext()) { + try (ArrowReader reader = scan.next().scanArrow(allocator)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + BigIntVector ids = (BigIntVector) root.getVector(0); + for (int row = 0; row < root.getRowCount(); row++) { + if (ids.isNull(row) || ids.get(row) != expected++) { + throw new AssertionError("incorrect list membership at row " + row); + } + } + } + } + } + if (expected != LIST_SIZE) { + throw new AssertionError("expected " + LIST_SIZE + " matches, got " + expected); + } + } + } +} diff --git a/java/vortex-jni/src/test/java/dev/vortex/jni/IndependentBuffersTest.java b/java/vortex-jni/src/test/java/dev/vortex/jni/IndependentBuffersTest.java index 02a3315bb8e..48814986df0 100644 --- a/java/vortex-jni/src/test/java/dev/vortex/jni/IndependentBuffersTest.java +++ b/java/vortex-jni/src/test/java/dev/vortex/jni/IndependentBuffersTest.java @@ -14,6 +14,7 @@ import dev.vortex.api.VortexWriter; import dev.vortex.io.NativeWritable; import java.io.IOException; +import java.lang.ref.Reference; import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.util.ArrayList; @@ -114,13 +115,19 @@ public void failedImportReleasesRetains() throws Exception { ArrowArray array = ArrowArray.allocateNew(allocator)) { fill(root, 32); Data.exportSchema(allocator, schema, null, ffiSchema); - long writer = NativeWriter.create( - Session.create().nativePointer(), - directory.resolve("failure.vortex").toString(), - ffiSchema.memoryAddress(), - Map.of(), - Map.of(), - false); + Session session = Session.create(); + long writer; + try { + writer = NativeWriter.create( + session.nativePointer(), + directory.resolve("failure.vortex").toString(), + ffiSchema.memoryAddress(), + Map.of(), + Map.of(), + false); + } finally { + Reference.reachabilityFence(session); + } try { Data.exportVectorSchemaRoot(allocator, root, null, array); // Omit the struct validity buffer while retaining its child's allocations.