viirya commented on PR #59551: URL: https://github.com/apache/spark/pull/59551#issuecomment-6227254340
@HyukjinKwon Thanks for laying out the options. I'd go with 4 for the backports, and for master I'd rather settle path 1 as below than copy every pass-through column. `DataSourceRDD` closes the reader, not the batches it has returned; whether those are released at that point is up to the reader's own `close()`. Nothing in the `PartitionReader` contract lets a reader release, in `close()`, batches it has already returned: `get()` only says it returns the same value until `next()` is called. A producer that releases or rewrites batches it has returned cannot know whether its consumer still holds them. That is the case here: `DataSourceRDD` closes the reader as soon as `next()` returns false, inside the consumer's `hasNext`, and path 1 calls `hasNext` right after writing a batch to Python, while it still holds that batch's pass-through columns. What a consumer can rely on is the batch itself. `closeIfFreeable` (SPARK-50235) already works that way: a writable or constant vector is reused by its producer, and a consumer must not release it, while any other vector is the consumer's to release once it is done with it. `NativePartitionReader` in #59424 follows this: it imports every batch into the same vectors, and wraps them in `ReusedArrowColumnVector`, whose `closeIfFreeable` does nothing, for that reason. Path 1 ignores that signal, since it only checks for `ArrowColumnVector`. So instead of copying, path 1 could respect it. For example, an internal marker interface in `sql/core`, implemented by `WritableColumnVector` and `ReusedArrowColumnVector`, which keeps extending `ArrowColumnVector`, would say that a vector is reused by its producer. Path 1 would send batches with such a column to path 2, and hold the other columns without copying them, leaving them to the consumer of its output batch to release with `closeIfFreeable`. Batches of a native data source would then take path 2, and how fast path 2 is is a separate question from this bug. We'd be happy to open a PR that adds this marker interface and closes the gaps in path 1 and in the producers and consumers described below. That leaves the producers that reuse or release Arrow vectors they have returned without saying so: - `ArrowCachedBatchToColumnarBatchIterator` closes the previous root in `next()`, on the premise that a consumer which asks for the next batch is done with the previous one. It should only close a root once none of its vectors is still in use, or mark its vectors as reused. - `ArrowOutputProcessorImpl` loads every batch of the Python runner into the same `ArrowColumnVector`s, which an outer `ArrowEvalPythonExec` would hold on path 1. Its data buffers are already new for every batch, as the reader deserializes each record batch into new buffers, so loading into a new root for every batch would cost little; marking its vectors as reused also works. On the consumer side, `closeIfFreeable` is not called by `ArrowEvalPythonExec.doExecute` and the non-codegen `ColumnarToRowEvaluatorFactory`, which flatten the batches without it, so a source that relies on its consumer keeps its vectors until the end of the task. SPARK-60306 could cover these consumers, rather than releasing the input batches of path 1 once they are written. What do you think? -- 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]
