andygrove commented on PR #5493: URL: https://github.com/apache/datafusion-comet/pull/5493#issuecomment-5441371302
> **Note on this review:** this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled. The core of this is clearly right. A static registry that lets task A spill task B's buffers, then reads B's `allocatedPages` unsynchronized while B mutates it, is a genuine correctness and safety bug, and scoping the list to the task's `CometBypassMergeSortShuffleWriter` is the obvious fix. Deleting the dead `spillingWriters` field along the way is a bonus. The reproducer that shows the cross-task spill plus the `AssertionError` is convincing. My concerns are all about the second half of the PR. **The description does not mention `allocateBlocking`** The "What changes are included in this PR?" section only describes the registry change. But this PR also adds about 80 lines of new blocking-allocation logic to `CometBoundedShuffleMemoryAllocator`, including a hand-rolled deadlock-avoidance check, `wait`/`notifyAll` coordination, and two new pieces of per-thread accounting. That is the riskiest code in the diff and a reader going by the description would not know it was here. Could you write up what it does and why it is needed? I do understand the connection: once a task can no longer take memory from other tasks, it needs somewhere to go when the pool is full. But that is an argument that deserves to be made explicitly, and it may well be an argument for a second PR rather than a second half. **`wait()` has no timeout** `allocateBlocking` calls bare `wait()` and only ever wakes on `free()` or on another waiter joining. The two fail-fast checks cover the cases where the waiting set can prove progress is impossible. They do not cover a thread that holds pool memory, is not in `waitingThreads`, and is not going to free anything soon, for example one blocked in a long shuffle read or stuck on network I/O. In that situation `allocatedMemory > retainedByWaitingThreads()` holds, the check passes, and the task waits forever after a single `logger.warn`. From an operator's point of view that is a hung task with one line of explanation. Would you add a bounded wait, either `wait(timeoutMillis)` with periodic re-logging so the stall is visible, or a configurable maximum after which it throws the original `SparkOutOfMemoryError`? An OOM that fails the task is much easier to diagnose than a job that never finishes. **`retainedMemory` holds strong references to `Thread` objects** Entries are removed from `retainedMemory` only when the retained total drops to zero through `free`. Any page that is never freed, which is exactly what happens on the task-failure paths this PR is partly about, leaves a permanent `Thread` key in an executor-lifetime map. `retainedByWaitingThreads` also walks the whole map under the allocator lock on every failed allocation. Would a `WeakHashMap`, or clearing the entry from a `TaskCompletionListener`, be safer here? **Interrupt is rethrown as `RuntimeException`** The comment says throwing a plain `RuntimeException` keeps an intentional task kill classified as `TaskKilled` rather than `ExceptionFailure`. I do not think the exception type is what drives that classification. Spark decides based on `TaskContext.isInterrupted` and the kill reason in `Executor.run`, and a `RuntimeException` surfacing out of the writer is not obviously different from any other. Could you point at where that behavior comes from, or verify it with a test that kills a task mid-wait and checks the reported failure reason? **Test flakiness** `CometDiskBlockWriterSuite` is 439 lines of two-thread coordination with `CountDownLatch` and `TimeLimits`, and it is now wired into both the Linux and macOS PR builds. That is the right place for it, but concurrency tests of this shape have a habit of becoming the flaky suite everyone learns to rerun. How long does it take, and are the timeouts generous enough for a loaded CI runner? If any assertion depends on one thread reaching a point before another without a latch enforcing it, that is the one to harden now rather than after the first flake. -- 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]
