peterxcli commented on PR #5027: URL: https://github.com/apache/datafusion-comet/pull/5027#issuecomment-5441858831
@andygrove Thanks — evaluated all four; addressed in 1dc8fc71e. **(1) Threading contract**: this is a doc correction, not a behavior change. The "at most one thread" wording was already contradicted on main: `CometScalaUDFCodegen` — the only `CometUDF` implementation — has run `evaluate` under `this.synchronized` since its introduction in #4267, with a comment explaining that `HashJoinExec` pipelines build/probe via `OnceAsync` (`tokio::spawn`), so multiple Tokio workers can call one task's dispatcher. Two more sources in `jni_api.rs`: the scan-free path drives each plan from a spawned prefetching Tokio task, and one Spark task can hold several such plans concurrently. Since this PR also changes the `evaluate` signature (allocator parameter), any out-of-tree UDF must be revisited regardless. I've grounded the trait doc in these concrete causes and will call it out in the PR description. **(2) Lock ordering**: added a javadoc on `TaskState` naming the order — `TaskMemoryManager` monitor → `TaskState` monitor → `MemoryManager` monitor — verified against Spark 3.5.8 and 4.1.3 sources: `acquireExecutionMemory` holds the TMM monitor across spills (so a spill releasing Arrow buffers re-enters `onRelease` in the same TMM→TaskState order), and `releaseExecutionMemory` never takes the TMM monitor, so every release-side path is a suffix of the acquire-side order. **(3) On-heap mode**: agreed — the behavior is correct (Arrow buffers are off-heap; there is no matching Spark pool), but it is now stated in the `TaskState` javadoc and the `CometUDF` contract, with a per-task debug log when accounting is skipped. **(4) Registration order**: the failure mode is not use-after-free — the allocator only closes once in-flight evaluations finish and it holds no memory, and post-completion evaluation is rejected. But investigating this surfaced a real edge: a straggler `evaluate()` after state removal re-created state via `computeIfAbsent`, whose mapping function registered a completion listener Spark can invoke immediately — re-entering `TASKS.remove` from inside `computeIfAbsent`. Fixed by registering the listener outside `computeIfAbsent`, which makes the ordering structural (a graceful-shutdown nicety rather than a safety requirement), and added `stragglerEvaluationsAroundTaskCompletionAreSafe` covering evaluations racing and following task completion (Spark's deferred-listener-drain semantics verified identical in 3.5.8 and 4.1.3). -- 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]
