sarutak commented on code in PR #59961:
URL: https://github.com/apache/spark/pull/59961#discussion_r4445437083
##########
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:
The direction makes sense to me. One concern about the Cleaner: as I read
the plan, every close leaves the buffers of the last imported batch and the
child allocator to it, including the close that `DataSourceRDD` makes right
after `next()` reaches the end of the stream, so even a reader that was read to
the end gives those buffers back only after a GC. The JVM does not count the
memory that the library allocated, so holding it never starts a GC, and in a
long task the `ColumnarBatch` and the column vectors, which the reader creates
once and reuses, have often been promoted, so the release may wait for an
old-generation collection. Could `close()` release the buffers right away once
`next()` has reached the end of the stream, and keep the Cleaner for a close
while a batch that `next()` returned may still be in use? This would depend on
your planned fallback for the columnar Python input: its Arrow path reads the
pass-through columns of the last batch when Python returns that batch
's result, which is after the reader reached the end of the stream and was
closed.
--
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]
