gengliangwang commented on code in PR #59961:
URL: https://github.com/apache/spark/pull/59961#discussion_r4483289144
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -858,9 +897,21 @@ class NativePartitionReader(
override def close(): Unit = synchronized {
if (!closed) {
closed = true
- if (!reading) {
+ // Another thread may be in next, or still read the batch that next
returned to it.
+ if (!reading && consumerDone) {
Review Comment:
You're right, the rework isn't in yet. 3b96894 only adds the path 2
fallback. The release logic of `NativePartitionReader` is still the one from
050cc61: `close()` frees at once when `consumerDone`, and otherwise hands the
release to the `releaser` pool. So the P1 still holds at the current head. To
your three questions:
1. I'll do it in this PR, not in a follow-up. This PR is what turns late
reads from an Arrow Java exception into reads of freed memory, so I don't want
it merged with that still open.
2. I'm not waiting on reviewers for the GC trade-off. sarutak's case (a
reader read to the end and then closed) gets an immediate free. Now that path 2
copies the pass-through rows, nothing reads the buffers after `next()` returns
false. What's left is a reader closed before the end of its stream, for example
with a LIMIT or when a script exits early. There, the last batch's buffers and
the child allocator wait for a GC. That's at most one batch per such reader.
I'd take that over a late read of freed memory, and I'll note it in the class
doc. If anyone thinks it's too much, please say so. What I still have to check
before writing it is the two points from my earlier reply. First, that an array
can outlive its stream: the C stream interface says their lifetimes are
independent. Second, that an array's release callback doesn't use the task,
because the task's context is closed when the stream is released. I'll check
the second one against the Rust and C++ SDKs and write it into the
`SupportsArrowStreamRead` contract.
3. It isn't going to a follow-up, so this doesn't apply. Until the rework is
pushed, please treat this thread as blocking.
The next revision will have: `close()` frees everything at once after the
end of the stream. Before the end, it frees the stream, the dictionaries and
the task at once, unless a thread is inside `next()`, and a Cleaner frees the
last batch's buffers and the allocator. A `next()` after `close()` fails. The
consumer tracking and the `releaser` pool go away. It will also include the
end-of-stream test I promised sarutak.
<sub>Automated reply (Opus)</sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
