viirya commented on PR #22038: URL: https://github.com/apache/datafusion/pull/22038#issuecomment-5012905466
@andygrove Thanks for looking at this closely — your analysis is correct, and it's a real deadlock, not a stall in theory. To confirm the mechanism: the coordinator initializes `probe_threads_counter` to the global `right_partition_count`, and a chunk is only released (and the next one loaded) once the partition that drives the counter to zero notifies the waiters. When a process builds its own `FallbackCoordinator` but only executes one right partition, the counter stops at N-1, nobody is elected to release the chunk, and the other `next_chunk` waiters block forever on the `Notify`. One nuance I want to be honest about: the single-process assumption itself isn't new — the single-pass path (`collect_left_input(..., right_partition_count)`) already sets the counter to the global partition count, and a distributed run there produces *incorrect* results (missing unmatched-left rows) because non-emitter partitions just return early. What this PR changes is that the coordinator makes that assumption load-bearing for *liveness*, turning a wrong-results degradation into a hang — a strictly worse failure mode. So the concern is fair and specific to the coordinated fallback. I've pushed an opt-out along the lines you suggested: a new config flag `datafusion.execution.enable_nlj_coordinated_fallback` (default `true`). When a distributed consumer sets it to `false`, the coordinated fallback is disabled for exactly the cases that would deadlock — left-emitting joins with a multi-partition right side — and those fall back to `SpillState::Disabled` (resource exhaustion under memory pressure) instead of coordinating, mirroring how `enable_dynamic_filter_pushdown` lets consumers opt out of another single-process assumption. Single-partition and non-left-emitting joins keep the coordinated fallback regardless. I kept the default as `true` so single-process users (the common case) get the coordinated spill behavior without configuration, and distributed engines flip it off — but I don't have a strong opinion on the default and am happy to invert it if you think distributed-safety should be the default, given a hang is a harder failure than an OOM. -- 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]
