Repository navigation
fix(spark): release native read resources deterministically - #11367
robert3005 merged 4 commits into
Conversation
robert3005
left a comment
robert3005
left a comment
There was a problem hiding this comment.
I might be missing something but we are doing too much to secure things here
| splitConcurrency); | ||
| return Scan.fromPointer(session, scanPtr); | ||
| } finally { | ||
| Reference.reachabilityFence(this); |
There was a problem hiding this comment.
| Reference.reachabilityFence(this); | |
| Reference.reachabilityFence(options); | |
| Reference.reachabilityFence(this); |
|
|
||
| /** Arrow schema of the data source (and of scans produced from it). */ | ||
| public Schema arrowSchema(BufferAllocator allocator) { | ||
| public synchronized Schema arrowSchema(BufferAllocator allocator) { |
There was a problem hiding this comment.
Do we need all the synchronised blocks? Reference.reachabilityFence(this) is a noop for synchronised. I think we should add Reference.reachabilityFence and remove synchronised unless I am missing something
| } | ||
| if (!scan.hasNext()) { | ||
| exhausted = true; | ||
| close(); |
There was a problem hiding this comment.
I don' think we should close here - let spark close it for us
| if (primed) { | ||
| return nextPartitionPointer != 0; | ||
| public synchronized boolean hasNext() { | ||
| if (closed) { |
There was a problem hiding this comment.
How about making this branch throw?
| public boolean hasNext() { | ||
| if (primed) { | ||
| return nextPartitionPointer != 0; | ||
| public synchronized boolean hasNext() { |
There was a problem hiding this comment.
this likely shouldn't be synchronised like datasource?
| scan.close(); | ||
| input.source().close(); | ||
| assertFalse(input.readable().refersTo(null)); | ||
| // Arrow 19 allocates two 40-byte stream structs, rounded to 64 bytes each. |
There was a problem hiding this comment.
you can do something like this instead
AllocationListener listener = new AllocationListener() {
@Override
public void onPreAllocation(long size) {
if (allocations.incrementAndGet() == failingAllocation) {
throw new OutOfMemoryException("injected allocation failure");
}
}
};
try (BufferAllocator limited = new RootAllocator(listener, Long.MAX_VALUE)) {
...
}
Signed-off-by: Christopher Wirt <[email protected]>
robert3005
left a comment
robert3005
left a comment
There was a problem hiding this comment.
Ok, I think we can go with a bit more of synchronised here
Merging this PR will improve performance by 10.08%2 warnings
Performance Changes
Tip Curious why performance improved? Comment Comparing 
Footnotes
|
Summary
Fixes #11364.
Closing a Spark reader currently leaves its Java-owned native read handles and file caches alive until Cleaner processing. Make that release explicit and idempotent, while preserving the resources still needed by derived scans, returned partitions and Arrow batches. Ports the lifecycle fix from QC #103.
Changes
DataSourceandScanimplementAutoCloseable, retaining Cleaner fallback. Keep native calls synchronized with explicit close; fence scan options so their expressions remain alive across JNI.hasNext()but never returned.IllegalStateException. An open, exhausted scan still returnsfalsefromhasNext()and throwsNoSuchElementExceptionfromnext().Testing
ReadLifecycleTestcases passed, including deterministic failures on the first and second Arrow allocations, explicit release, surviving child handles, and closed versus exhausted scans.-Werrorpassed../gradlew spotlessApplyandgit diff --checkpassed.Workspace-wide tests and benchmarks were not run. The unrelated list-view normalization from QC #103 is excluded.
AI tools assisted with this port, review and description. Christopher Wirt is the responsible contributor.