gengliangwang commented on code in PR #59961:
URL: https://github.com/apache/spark/pull/59961#discussion_r4430136415
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -858,7 +896,9 @@ 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:
Done in 076f7112929c: When close() hands the release to a live consumer
thread that is not inside next(), it now also queues a task on a shared daemon
pool. That task joins the consumer thread and then calls releaseClosed(). So
the reader is released even if the thread exits without calling next() again,
as in the PipedRDD writer case. Added a test for this, and updated the class
doc.
<sub>Automated reply (Opus)</sub>
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -858,7 +896,9 @@ class NativePartitionReader(
override def close(): Unit = synchronized {
if (!closed) {
Review Comment:
Done in 076f7112929c: close() on a reader that is already closed now calls
releaseClosed() when no thread is in next() and consumerDone holds. So the
thread the release was handed to can release the reader through close(), for
example after next() returns false. Added a test where that thread closes the
reader.
<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]
