This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-mosaic.git
The following commit(s) were added to refs/heads/main by this push:
new 1efb60b fix(jni): propagate InputFile exceptions from native reader
threads (#77)
1efb60b is described below
commit 1efb60b8cb9cec6858300b8b7bd43b63f1fd67fe
Author: jianguotian <[email protected]>
AuthorDate: Thu Aug 20 20:35:43 2026 +0800
fix(jni): propagate InputFile exceptions from native reader threads (#77)
---
docs/java-api.html | 19 ++-
.../org/apache/paimon/mosaic/MosaicReader.java | 18 ++-
.../paimon/mosaic/MosaicComprehensiveTest.java | 23 +--
.../apache/paimon/mosaic/MosaicRoundtripTest.java | 161 +++++++++++++++++----
jni/src/lib.rs | 122 ++++++++++++++--
5 files changed, 279 insertions(+), 64 deletions(-)
diff --git a/docs/java-api.html b/docs/java-api.html
index ffc95de..89ccda2 100644
--- a/docs/java-api.html
+++ b/docs/java-api.html
@@ -260,7 +260,13 @@ w.endMap();</code></pre>
<pre><code><span class="ty">BufferAllocator</span> allocator = <span
class="kw">new</span> <span class="ty">RootAllocator</span>();
<span class="ty">InputFile</span> inputFile = ...;
<span class="kw">long</span> fileLength = ...;
+<span class="cmt">// The enclosing method declares throws IOException.</span>
<span class="ty">MosaicReader</span> reader = <span
class="ty">MosaicReader</span>.open(inputFile, fileLength,
allocator);</code></pre>
+ <p>
+ <code>open()</code> and <code>readRowGroup()</code> propagate
the original
+ <code>IOException</code> thrown by
<code>InputFile.readFully()</code>. Callers must
+ catch it or declare <code>throws IOException</code>.
+ </p>
<h3>2. Inspect the Schema</h3>
<p>
@@ -283,7 +289,7 @@ w.endMap();</code></pre>
<tr><td><code>numRowGroups()</code></td><td><code>int</code></td><td>Row group
count</td></tr>
<tr><td><code>rowGroupNumRows(rg)</code></td><td><code>int</code></td><td>Number
of rows in a specific row group</td></tr>
<tr><td><code>project(String[])</code></td><td><code>void</code></td><td>Set
projection: subsequent reads return only the named columns in the specified
order</td></tr>
- <tr><td><code>readRowGroup(rg,
allocator)</code></td><td><code>VectorSchemaRoot</code></td><td>Read a row
group (all columns or projected columns if project() was called)</td></tr>
+ <tr><td><code>readRowGroup(rg, allocator) throws
IOException</code></td><td><code>VectorSchemaRoot</code></td><td>Read a row
group, propagating the original <code>InputFile</code> exception</td></tr>
<tr><td><code>getRowGroupStatistics(rg)</code></td><td><code>Map<String,
ColumnStatistics></code></td><td>Column statistics for a row group, keyed by
column name</td></tr>
</tbody>
</table>
@@ -424,12 +430,11 @@ writer.close();
<span class="kw">byte</span>[] data = baos.toByteArray();
<span class="cmt">// 2. Read</span>
-<span class="ty">MosaicReader</span> reader = <span
class="ty">MosaicReader</span>.open(
- (pos, buf, off, len) -> System.arraycopy(data, (<span
class="kw">int</span>) pos, buf, off, len),
- data.length,
- allocator);
-
-<span class="kw">try</span> (reader) {
+<span class="cmt">// The enclosing method declares throws IOException.</span>
+<span class="kw">try</span> (<span class="ty">MosaicReader</span> reader =
<span class="ty">MosaicReader</span>.open(
+ (pos, buf, off, len) -> System.arraycopy(data, (<span
class="kw">int</span>) pos, buf, off, len),
+ data.length,
+ allocator)) {
<span class="kw">for</span> (<span class="kw">int</span> rg = <span
class="num">0</span>; rg < reader.numRowGroups(); rg++) {
<span class="kw">try</span> (<span class="ty">VectorSchemaRoot</span>
batch = reader.readRowGroup(rg, allocator)) {
diff --git a/java/src/main/java/org/apache/paimon/mosaic/MosaicReader.java
b/java/src/main/java/org/apache/paimon/mosaic/MosaicReader.java
index cab99db..8c2e9f5 100644
--- a/java/src/main/java/org/apache/paimon/mosaic/MosaicReader.java
+++ b/java/src/main/java/org/apache/paimon/mosaic/MosaicReader.java
@@ -19,6 +19,7 @@
package org.apache.paimon.mosaic;
+import java.io.IOException;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -46,7 +47,14 @@ public class MosaicReader implements AutoCloseable {
}
}
- public static MosaicReader open(InputFile inputFile, long fileLength,
BufferAllocator allocator) {
+ /**
+ * Opens a Mosaic reader.
+ *
+ * @throws IOException if {@link InputFile#readFully(long, byte[], int,
int)} throws one while
+ * reading file metadata; the original exception is propagated
+ */
+ public static MosaicReader open(InputFile inputFile, long fileLength,
BufferAllocator allocator)
+ throws IOException {
long handle = NativeLib.nativeReaderOpen(inputFile, fileLength);
if (handle == 0) {
throw new RuntimeException("failed to open reader");
@@ -71,7 +79,13 @@ public class MosaicReader implements AutoCloseable {
NativeLib.nativeReaderSetProjection(handle, columns);
}
- public VectorSchemaRoot readRowGroup(int rgIndex, BufferAllocator
allocator) {
+ /**
+ * Reads a row group.
+ *
+ * @throws IOException if {@link InputFile#readFully(long, byte[], int,
int)} throws one while
+ * reading row-group data; the original exception is propagated
+ */
+ public VectorSchemaRoot readRowGroup(int rgIndex, BufferAllocator
allocator) throws IOException {
long rgHandle = NativeLib.nativeReaderOpenRowGroup(handle, rgIndex);
if (rgHandle == 0) {
throw new RuntimeException("failed to open row group " + rgIndex);
diff --git
a/java/src/test/java/org/apache/paimon/mosaic/MosaicComprehensiveTest.java
b/java/src/test/java/org/apache/paimon/mosaic/MosaicComprehensiveTest.java
index 796845f..6f2713d 100644
--- a/java/src/test/java/org/apache/paimon/mosaic/MosaicComprehensiveTest.java
+++ b/java/src/test/java/org/apache/paimon/mosaic/MosaicComprehensiveTest.java
@@ -20,6 +20,7 @@
package org.apache.paimon.mosaic;
import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
@@ -72,7 +73,7 @@ public class MosaicComprehensiveTest {
return baos.toByteArray();
}
- private MosaicReader readerFromBytes(byte[] data) {
+ private MosaicReader readerFromBytes(byte[] data) throws IOException {
InputFile inputFile = (position, buffer, offset, length) -> {
System.arraycopy(data, (int) position, buffer, offset, length);
};
@@ -81,7 +82,7 @@ public class MosaicComprehensiveTest {
// Test 1: Large data roundtrip with 1M rows
@Test
- public void testLargeDataRoundtrip() {
+ public void testLargeDataRoundtrip() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(64, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -145,7 +146,7 @@ public class MosaicComprehensiveTest {
// Test 2: All constant values - should produce small file
@Test
- public void testAllConstantValues() {
+ public void testAllConstantValues() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(64, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -208,7 +209,7 @@ public class MosaicComprehensiveTest {
// Test 3: High null rate (95%)
@Test
- public void testHighNullRate() {
+ public void testHighNullRate() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(64, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -294,7 +295,7 @@ public class MosaicComprehensiveTest {
// Test 4: Wide table with 100 columns
@Test
- public void testWideTable() {
+ public void testWideTable() throws IOException {
int numCols = 100;
int totalRows = 50_000;
@@ -367,7 +368,7 @@ public class MosaicComprehensiveTest {
// Test 5: Many small writes (10000 batches of 10 rows)
@Test
- public void testManySmallWrites() {
+ public void testManySmallWrites() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("val", ArrowType.Utf8.INSTANCE)
@@ -408,7 +409,7 @@ public class MosaicComprehensiveTest {
// Test 6: Projection with large column count
@Test
- public void testProjectionWithLargeColumnCount() {
+ public void testProjectionWithLargeColumnCount() throws IOException {
int numCols = 50;
List<Field> fields = new ArrayList<>();
for (int c = 0; c < numCols; c++) {
@@ -460,7 +461,7 @@ public class MosaicComprehensiveTest {
// Test 7: Sequential vs random data file size comparison
@Test
- public void testSequentialVsRandomData() {
+ public void testSequentialVsRandomData() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("val", new ArrowType.Int(64, true))
));
@@ -523,7 +524,7 @@ public class MosaicComprehensiveTest {
// Test 8: Multiple row groups roundtrip with small max size
@Test
- public void testMultipleRowGroupsRoundtrip() {
+ public void testMultipleRowGroupsRoundtrip() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("data", new ArrowType.Int(64, true)),
@@ -587,7 +588,7 @@ public class MosaicComprehensiveTest {
// Test 9: Empty string values mixed with non-empty
@Test
- public void testEmptyStringValues() {
+ public void testEmptyStringValues() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("text", ArrowType.Utf8.INSTANCE)
@@ -646,7 +647,7 @@ public class MosaicComprehensiveTest {
// Test 10: Decimal precisions
@Test
- public void testDecimalPrecisions() {
+ public void testDecimalPrecisions() throws IOException {
// Test Decimal128 with precision 10
Schema schema10 = new Schema(Arrays.asList(
Field.nullable("dec10", new ArrowType.Decimal(10, 2, 128))
diff --git
a/java/src/test/java/org/apache/paimon/mosaic/MosaicRoundtripTest.java
b/java/src/test/java/org/apache/paimon/mosaic/MosaicRoundtripTest.java
index 437cd5d..7357dc2 100644
--- a/java/src/test/java/org/apache/paimon/mosaic/MosaicRoundtripTest.java
+++ b/java/src/test/java/org/apache/paimon/mosaic/MosaicRoundtripTest.java
@@ -28,6 +28,9 @@ import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.arrow.c.jni.JniWrapper;
import org.apache.arrow.memory.ArrowBuf;
@@ -171,7 +174,7 @@ public class MosaicRoundtripTest {
return baos.toByteArray();
}
- private MosaicReader readerFromBytes(byte[] data) {
+ private MosaicReader readerFromBytes(byte[] data) throws IOException {
InputFile inputFile = (position, buffer, offset, length) -> {
System.arraycopy(data, (int) position, buffer, offset, length);
};
@@ -192,7 +195,14 @@ public class MosaicRoundtripTest {
System.runFinalization();
Thread.sleep(50L);
}
- assertNull("expected input file to be released after failed open",
reference.get());
+ assertNull("expected native callback object to be released",
reference.get());
+ }
+
+ private static void awaitGarbageCollection(List<WeakReference<?>>
references)
+ throws InterruptedException {
+ for (WeakReference<?> reference : references) {
+ awaitGarbageCollection(reference);
+ }
}
private WeakReference<InputFile> openReaderWithClosedAllocator(byte[]
data) {
@@ -211,8 +221,72 @@ public class MosaicRoundtripTest {
return reference;
}
+ private List<WeakReference<?>> openReaderWithFailingInput() {
+ IOException expected = new IOException("intentional native input
failure");
+ InputFile inputFile =
+ new InputFile() {
+ @Override
+ public void readFully(
+ long position, byte[] buffer, int offset, int
length)
+ throws IOException {
+ throw expected;
+ }
+ };
+ WeakReference<InputFile> inputReference = new
WeakReference<>(inputFile);
+ WeakReference<IOException> exceptionReference = new
WeakReference<>(expected);
+
+ try (MosaicReader ignored = MosaicReader.open(inputFile, 64L,
allocator)) {
+ fail("expected IOException");
+ } catch (IOException error) {
+ assertSame(expected, error);
+ }
+ return Arrays.asList(inputReference, exceptionReference);
+ }
+
+ private List<WeakReference<?>> readRowGroupWithFailingInput(byte[] data)
throws IOException {
+ IOException expected = new IOException("intentional native background
input failure");
+ long callingThreadId = Thread.currentThread().getId();
+ AtomicBoolean failReads = new AtomicBoolean();
+ AtomicInteger reads = new AtomicInteger();
+ AtomicLong failingThreadId = new AtomicLong(-1L);
+ InputFile inputFile =
+ new InputFile() {
+ @Override
+ public void readFully(
+ long position, byte[] buffer, int offset, int
length)
+ throws IOException {
+ reads.incrementAndGet();
+ if (failReads.get()) {
+ failingThreadId.compareAndSet(
+ -1L, Thread.currentThread().getId());
+ throw expected;
+ }
+ System.arraycopy(data, (int) position, buffer, offset,
length);
+ }
+ };
+ WeakReference<InputFile> inputReference = new
WeakReference<>(inputFile);
+ WeakReference<IOException> exceptionReference = new
WeakReference<>(expected);
+
+ MosaicReader reader = MosaicReader.open(inputFile, data.length,
allocator);
+ int readsAfterOpen = reads.get();
+ try {
+ failReads.set(true);
+ try (VectorSchemaRoot ignored = reader.readRowGroup(0, allocator))
{
+ fail("expected IOException");
+ } catch (IOException actual) {
+ assertSame(expected, actual);
+ }
+ assertTrue("expected a row-group read", reads.get() >
readsAfterOpen);
+ assertNotEquals(callingThreadId, failingThreadId.get());
+ assertEquals(1, reader.numRowGroups());
+ } finally {
+ reader.close();
+ }
+ return Arrays.asList(inputReference, exceptionReference);
+ }
+
@Test
- public void testBasicRoundtrip() {
+ public void testBasicRoundtrip() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -280,7 +354,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testWriteFromIndependentRootAllocator() {
+ public void testWriteFromIndependentRootAllocator() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE)
@@ -316,7 +390,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testWriteFromLimitedChildAllocator() {
+ public void testWriteFromLimitedChildAllocator() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true))
));
@@ -389,7 +463,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testSameRootExportKeepsWriterAllocatorAccounting() {
+ public void testSameRootExportKeepsWriterAllocatorAccounting() throws
IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true))
));
@@ -432,7 +506,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testCrossRootWriterAllocatorOutOfMemoryCanRetryWithoutLeak() {
+ public void testCrossRootWriterAllocatorOutOfMemoryCanRetryWithoutLeak()
throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true))
));
@@ -483,7 +557,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testCrossRootFailureAfterRootRegistrationCanRetryWithoutLeak()
{
+ public void testCrossRootFailureAfterRootRegistrationCanRetryWithoutLeak()
throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true))
));
@@ -554,7 +628,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testCrossRootPreflightValidationFailureCanRetryWithoutLeak() {
+ public void testCrossRootPreflightValidationFailureCanRetryWithoutLeak()
throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true))
));
@@ -723,7 +797,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testCrossRootPartialExportFailureCanRetryWithoutLeak() {
+ public void testCrossRootPartialExportFailureCanRetryWithoutLeak() throws
IOException {
byte[] data;
try (RootAllocator writerRoot = new RootAllocator(16L * 1024 * 1024);
RootAllocator inputRoot = new RootAllocator(16L * 1024 * 1024)) {
@@ -781,7 +855,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testWriteSequentialBundlesFromDifferentRootAllocators() {
+ public void testWriteSequentialBundlesFromDifferentRootAllocators() throws
IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.notNullable("id", new ArrowType.Int(32, true))
));
@@ -907,7 +981,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testNullValues() {
+ public void testNullValues() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -967,7 +1041,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testProjection() {
+ public void testProjection() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("a", new ArrowType.Int(32, true)),
Field.nullable("b", ArrowType.Utf8.INSTANCE),
@@ -1013,7 +1087,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testProjectionOrder() {
+ public void testProjectionOrder() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("a", new ArrowType.Int(32, true)),
Field.nullable("b", ArrowType.Utf8.INSTANCE),
@@ -1061,7 +1135,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testProjectionEmpty() {
+ public void testProjectionEmpty() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("a", new ArrowType.Int(32, true)),
Field.nullable("b", ArrowType.Utf8.INSTANCE)
@@ -1093,7 +1167,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testStats() {
+ public void testStats() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("name", ArrowType.Utf8.INSTANCE),
@@ -1141,7 +1215,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testAllTypes() {
+ public void testAllTypes() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("f_bool", ArrowType.Bool.INSTANCE),
Field.nullable("f_int8", new ArrowType.Int(8, true)),
@@ -1217,7 +1291,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testTimestampNsRoundtrip() {
+ public void testTimestampNsRoundtrip() throws IOException {
ArrowType.Timestamp tsNsType = new
ArrowType.Timestamp(TimeUnit.NANOSECOND, null);
ArrowType.Timestamp tsNsTzType = new
ArrowType.Timestamp(TimeUnit.NANOSECOND, "Asia/Shanghai");
Schema arrowSchema = new Schema(Arrays.asList(
@@ -1263,7 +1337,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testCompressionNone() {
+ public void testCompressionNone() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("x", new ArrowType.Int(32, true)),
Field.nullable("y", ArrowType.Utf8.INSTANCE)
@@ -1295,7 +1369,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testMultipleRowGroups() {
+ public void testMultipleRowGroups() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("data", new ArrowType.Int(64, true))
@@ -1342,7 +1416,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testMultipleWrites() {
+ public void testMultipleWrites() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("x", new ArrowType.Int(32, true))
));
@@ -1409,7 +1483,30 @@ public class MosaicRoundtripTest {
}
@Test
- public void testSingleRow() {
+ public void testReaderOpenReleasesInputGlobalRefWhenReadFails() throws
Exception {
+ awaitGarbageCollection(openReaderWithFailingInput());
+ }
+
+ @Test
+ public void
testReaderRestoresBackgroundInputExceptionAndReleasesGlobalRef()
+ throws Exception {
+ Schema schema = new Schema(Arrays.asList(
+ Field.nullable("value", new ArrowType.Int(32, true))
+ ));
+ byte[] data;
+ try (VectorSchemaRoot root = VectorSchemaRoot.create(schema,
allocator)) {
+ IntVector values = (IntVector) root.getVector("value");
+ values.allocateNew(1);
+ values.set(0, 7);
+ root.setRowCount(1);
+ data = writeToBytes(schema, writer -> writer.write(root));
+ }
+
+ awaitGarbageCollection(readRowGroupWithFailingInput(data));
+ }
+
+ @Test
+ public void testSingleRow() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("v", new ArrowType.Int(32, true))
));
@@ -1432,7 +1529,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testZeroRows() {
+ public void testZeroRows() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("v", new ArrowType.Int(32, true))
));
@@ -1450,7 +1547,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testStatsWithNulls() {
+ public void testStatsWithNulls() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("a", new ArrowType.Int(32, true)),
Field.nullable("b", new ArrowType.Int(64, true))
@@ -1502,7 +1599,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testStatsAllNull() {
+ public void testStatsAllNull() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("x", new ArrowType.Int(32, true))
));
@@ -1556,7 +1653,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testSchemaRoundtrip() {
+ public void testSchemaRoundtrip() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("name", ArrowType.Utf8.INSTANCE),
Field.notNullable("id", new ArrowType.Int(32, true)),
@@ -1719,7 +1816,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testWriterStatsMatchesReaderStats() {
+ public void testWriterStatsMatchesReaderStats() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("value", new
ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE))
@@ -1763,7 +1860,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testRowGroupNumRows() {
+ public void testRowGroupNumRows() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("id", new ArrowType.Int(32, true)),
Field.nullable("data", new ArrowType.Int(64, true))
@@ -1806,7 +1903,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testStatsEmptyStringMin() {
+ public void testStatsEmptyStringMin() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("s", ArrowType.Utf8.INSTANCE)
));
@@ -1849,7 +1946,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testRowGroupNumRowsSingleRowGroup() {
+ public void testRowGroupNumRowsSingleRowGroup() throws IOException {
Schema arrowSchema = new Schema(Arrays.asList(
Field.nullable("x", new ArrowType.Int(32, true))
));
@@ -1872,7 +1969,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testArrayType() {
+ public void testArrayType() throws IOException {
Field elementField = new Field("item", FieldType.nullable(new
ArrowType.Int(32, true)), null);
Field listField = new Field("tags",
FieldType.nullable(ArrowType.List.INSTANCE), Arrays.asList(elementField));
Schema arrowSchema = new Schema(Arrays.asList(
@@ -1960,7 +2057,7 @@ public class MosaicRoundtripTest {
}
@Test
- public void testMapType() {
+ public void testMapType() throws IOException {
// Use MapVector's writer to avoid schema mismatch with UnionMapWriter
Field keyField = new Field("keys", FieldType.notNullable(new
ArrowType.Int(32, true)), null);
Field valueField = new Field("values",
FieldType.nullable(ArrowType.Utf8.INSTANCE), null);
diff --git a/jni/src/lib.rs b/jni/src/lib.rs
index ac84808..7dc59f1 100644
--- a/jni/src/lib.rs
+++ b/jni/src/lib.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use std::error::Error;
+use std::fmt;
use std::io;
use std::panic::{self, AssertUnwindSafe};
use std::ptr;
@@ -198,6 +200,77 @@ impl OutputFile for JniOutputFile {
// ======================== JniInputFile ========================
+struct JavaInputException {
+ operation: &'static str,
+ throwable: GlobalRef,
+}
+
+impl fmt::Debug for JavaInputException {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("JavaInputException")
+ .field("operation", &self.operation)
+ .finish_non_exhaustive()
+ }
+}
+
+impl fmt::Display for JavaInputException {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(formatter, "{} threw a Java exception", self.operation)
+ }
+}
+
+impl Error for JavaInputException {}
+
+fn input_jni_result<T>(
+ env: &mut JNIEnv<'_>,
+ result: jni::errors::Result<T>,
+ operation: &'static str,
+) -> io::Result<T> {
+ match result {
+ Ok(value) => Ok(value),
+ Err(JniError::JavaException) => {
+ let captured = (|| -> jni::errors::Result<GlobalRef> {
+ let exception = env.exception_occurred()?;
+ env.exception_clear()?;
+ if exception.is_null() {
+ return Err(JniError::NullPtr("pending Java input
exception"));
+ }
+ let global = env.new_global_ref(exception)?;
+ let exception_pending = env.exception_check()?;
+ if exception_pending {
+ env.exception_clear()?;
+ }
+ if global.as_obj().is_null() || exception_pending {
+ return Err(JniError::NullPtr(
+ "NewGlobalRef for pending Java input exception",
+ ));
+ }
+ Ok(global)
+ })();
+
+ match captured {
+ Ok(throwable) => Err(io::Error::other(JavaInputException {
+ operation,
+ throwable,
+ })),
+ Err(capture_error) => {
+ // Native reads may run on worker threads. Do not detach a
worker while a Java
+ // exception is pending, even if preserving the original
throwable failed.
+ let _ = env.exception_clear();
+ Err(io::Error::other(format!(
+ "{} (failed to preserve Java exception from {}: {})",
+ JniError::JavaException,
+ operation,
+ capture_error
+ )))
+ }
+ }
+ }
+ Err(error) => Err(io::Error::other(error.to_string())),
+ }
+}
+
struct JniInputFile {
jvm: Arc<JavaVM>,
input_file_ref: GlobalRef,
@@ -216,11 +289,10 @@ impl InputFile for JniInputFile {
.attach_current_thread()
.map_err(|e| io::Error::other(e.to_string()))?;
- let java_buf = env
- .new_byte_array(buf.len() as i32)
- .map_err(|e| io::Error::other(e.to_string()))?;
+ let result = env.new_byte_array(buf.len() as i32);
+ let java_buf = input_jni_result(&mut env, result, "NewByteArray")?;
- env.call_method(
+ let result = env.call_method(
&self.input_file_ref,
"readFully",
"(J[BII)V",
@@ -230,13 +302,13 @@ impl InputFile for JniInputFile {
JValue::Int(0),
JValue::Int(buf.len() as jint),
],
- )
- .map_err(|e| io::Error::other(e.to_string()))?;
+ );
+ input_jni_result(&mut env, result, "InputFile.readFully")?;
let i8_buf: &mut [i8] =
unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr() as *mut
i8, buf.len()) };
- env.get_byte_array_region(&java_buf, 0, i8_buf)
- .map_err(|e| io::Error::other(e.to_string()))?;
+ let result = env.get_byte_array_region(&java_buf, 0, i8_buf);
+ input_jni_result(&mut env, result, "GetByteArrayRegion")?;
Ok(())
}
@@ -252,7 +324,9 @@ fn bytemuck_cast(data: &[u8]) -> &[i8] {
}
fn throw(env: &mut JNIEnv, msg: &str) {
- let _ = env.throw_new("java/lang/RuntimeException", msg);
+ if matches!(env.exception_check(), Ok(false)) {
+ let _ = env.throw_new("java/lang/RuntimeException", msg);
+ }
}
fn rethrow(env: &mut JNIEnv, exception: &GlobalRef) {
@@ -284,6 +358,29 @@ fn rethrow(env: &mut JNIEnv, exception: &GlobalRef) {
}
}
+fn find_java_input_exception<'a>(
+ error: &'a (dyn Error + 'static),
+) -> Option<&'a JavaInputException> {
+ if let Some(input_exception) = error.downcast_ref::<JavaInputException>() {
+ return Some(input_exception);
+ }
+ if let Some(io_error) = error.downcast_ref::<io::Error>() {
+ if let Some(inner) = io_error.get_ref() {
+ if let Some(input_exception) = find_java_input_exception(inner) {
+ return Some(input_exception);
+ }
+ }
+ }
+ error.source().and_then(find_java_input_exception)
+}
+
+fn throw_io_error(env: &mut JNIEnv<'_>, error: &io::Error, message: &str) {
+ match find_java_input_exception(error) {
+ Some(input_exception) => rethrow(env, &input_exception.throwable),
+ None => throw(env, message),
+ }
+}
+
struct WriterHandle {
inner: MosaicWriter<JniOutputFile>,
_stream_ref: GlobalRef,
@@ -723,7 +820,8 @@ pub extern "system" fn
Java_org_apache_paimon_mosaic_NativeLib_nativeReaderOpen(
Box::into_raw(Box::new(rh)) as jlong
}
Err(e) => {
- throw(&mut env, &format!("open failed: {}", e));
+ drop(global);
+ throw_io_error(&mut env, &e, &format!("open failed: {}", e));
0
}
}
@@ -819,7 +917,7 @@ pub extern "system" fn
Java_org_apache_paimon_mosaic_NativeLib_nativeReaderOpenR
Box::into_raw(rg_handle) as jlong
}
Err(e) => {
- throw(&mut env, &format!("open row group failed: {}", e));
+ throw_io_error(&mut env, &e, &format!("open row group failed:
{}", e));
0
}
}
@@ -1078,7 +1176,7 @@ pub extern "system" fn
Java_org_apache_paimon_mosaic_NativeLib_nativeRowGroupRea
let batch = match rg.inner.read_columns() {
Ok(b) => b,
Err(e) => {
- throw(&mut env, &format!("read_columns failed: {}", e));
+ throw_io_error(&mut env, &e, &format!("read_columns failed:
{}", e));
return -1;
}
};
