mbutrovich commented on PR #7016: URL: https://github.com/apache/datafusion-comet/pull/7016#issuecomment-6320288002
I had another anchored comment, not sure if GitHub ate it or what. I'll just post it again: What do you think about keeping the shared state on the broadcast value instead of in a process-wide `object`? The [global singletons](https://github.com/apache/datafusion-comet/blob/04f329d3f49908dddb4e46f509c21ce6e313bd48/docs/source/contributor-guide/development.md?plain=1#L194-L201) section of the development guide says not to use one when "The state needs cleanup or refresh during process lifetime", and entries here are freed during the process lifetime by design. The cache already keys on the value instance that Spark hands every task on an executor, so the state could live on that instance. `relationFuture` could broadcast a small Comet class instead of the bare `Array[ChunkedByteBuffer]`, roughly: ```scala private[comet] final class CometBroadcastValue(val chunks: Array[ChunkedByteBuffer]) extends Serializable { // Not serialized, so each executor's deserialized instance starts with no decode. @transient private lazy val shared = new SharedDecode(chunks) def batches(context: TaskContext, source: String): Iterator[ColumnarBatch] = shared.batches(context, source) } ``` `SharedDecode` would be today's `Entry` and `SharedBatchIterator`, with the same holds and the same release on task completion, so the lifetime rules stay as they are. With the state on the value there is no map or key: `acquire`, `forget` and its `eq` check go away, along with the strong references the map keeps to broadcast values and the state that suites share through `DecodedBroadcastCache.shared`. What's left is a hold count and one decode slot per value. Spark's broadcast hash join has the same shape: one `HashedRelation` per executor, and each task takes its own view with [`asReadOnlyCopy()`](https://github.com/apache/spark/blob/77bbf77e86ad48f58b5dfbc6ac882b3e70cf1989/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala#L125-L129). The state needs to be a `@transient lazy val` rather than a field initializer, because neither Java deserialization nor Spark's Kryo setup runs the constructor of a class without a no-arg constructor. The cost is changing the payload type where it's read: `CometBatchRDD`, `CometColumnarToRowExec`, `CometNativeColumnarToRowExec` and `CometSubqueryBroadcastExec`, plus registering the new class in `Utils.arrowBytesKryoClasses` for `spark.kryo.registrationRequired`. A version with no holds at all would hand the lifecycle to Spark. `MemoryStore` calls `close()` on an `AutoCloseable` value when it removes the block ([`freeMemoryEntry`](https://github.com/apache/spark/blob/77bbf77e86ad48f58b5dfbc6ac882b3e70cf1989/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala#L412-L425)), which it does on eviction and when the broadcast is cleaned up. If the wrapper were `AutoCloseable`, the code here would shrink to a decode-once under a lock and a free in `close()`. The decoded copy would then live as long as the block. Neither `CometBroadcastExchangeExec` nor Spark's `BroadcastExchangeExec` destroys its broadcast, so on executors that is until the driver's `ContextCleaner` collects the `Broadcast`. The BlockManager sizes the block when it stores the value, before any decode, so storage memory wouldn't count the decoded buffers or evict the block to make room for them. Eviction can also call `close()` while a task is between batches, so creating a view and closing would still share a lock, with the task decoding for itself after close. Do you see a way to bound that lifetime, for example by charging the decoded size to storage memory? If so, I'd prefer Spark owning the lifecycle. If not, the holds above seem like the better trade. -- 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]
