sunchao opened a new pull request, #5505: URL: https://github.com/apache/datafusion-comet/pull/5505
## Which issue does this PR close? Closes #5504. ## Rationale for this change Early plan close can collect final metrics while the async batch producer still owns its stream. Updates performed when stream guards are dropped then occur after the snapshot. The producer handle is currently discarded, so closing the plan cannot explicitly cancel that work. This extracts the producer-lifecycle change from #5453. It adds no scan counters, does not change object-store registration, and does not depend on the metrics PR. ## What changes are included in this PR? Keep the producer `JoinHandle` in the execution context. On the Spark executor thread, drop the batch receiver, abort the producer, and wait for cooperative completion before collecting final metrics. The helper returns an explicit finished/timeout result and logs when the final snapshot may be incomplete. Waiting uses a 100 ms budget; a missing or already finished producer does not pay that delay, and a cooperative producer can stop earlier. This is best-effort final accounting. The producer owns its stream and sender rather than a raw execution-context pointer, but synchronous decoding cannot be preempted by `abort`, and a timed-out producer may still consult external task state or update counters later. The waiting budget is not a memory-safety guarantee or a hard wall-clock limit under arbitrary scheduler delays. ## How are these changes tested? Dedicated one-worker Tokio runtimes keep blocking stand-ins out of the global runtime. Four tests verify cooperative drop-time metric visibility, cancellation while a batch is queued, an explicitly incomplete timeout snapshot followed by a later update, and an already finished producer under exhausted Tokio cooperative budget. The blocked test releases its worker after shutdown returns, so it does not depend on a tight upper timing assertion. A manual ignored test measures the helper with 100 already finished producers, 100 pending cooperative producers, and five blocked synchronous stand-ins. In the full debug-build test binary, observed median / p95 durations were **2.124 / 2.714 microseconds** for finished producers, **1.123 / 1.142 milliseconds** for cooperative pending producers, and **100.129 / 100.989 milliseconds** for the five blocked stand-ins. Cooperative cancellation exposed the drop-time counter before the snapshot; every blocked sample returned with that counter still incomplete. These are local helper/scheduler measurements, not end-to-end Spark close timings or a latency guarantee. The full native crate compiled with the two disjoint extracted changes combined at their shared base. All **four lifecycle regressions passed**, and the ignored manual measurement passed when invoked explicitly. All **90 Parquet tests** also passed with a confirmed zero process exit status. `cargo fmt --all --check` and `git diff --check` pass. Default native features and JDK 21 were used. The measurement isolates the shutdown helper and does not measure a full Spark task or production decoding workload. -- 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]
