Detect and mitigate classic pipeline buffer deadlocks (#7752)

[mattcasters]

Summary

Addresses #7752 for the classic local multi-threaded pipeline engine only (Beam/Spark are out of scope: they materialize between stages).

The issue proposed extra loader threads + CountDownLatch in Stream Lookup. That is transform-specific, risks loading whole streams into memory, and does not help Merge Join / Append / similar multi-input cases. This PR instead:

1. Detects bounded-buffer deadlock risk (split–rejoin with a shared ancestor).
2. Mitigates by spilling only the minimal set of recommended hops to disk when the in-memory rowset would block.

Goal: prefer correctness and progress over silent hangs for data engineers, with clear logging when risk or spill engages.

Problem (what we are actually fixing)

Pipeline hops are already a DAG (hasLoop / GUI loop checks). Stream Lookup / Merge Join hangs are not hop-graph cycles. They are bounded-buffer wait-for cycles:

• Rowsets (BlockingRowSet) have fixed capacity (rowset size).
• A shared source that copies to multiple paths blocks if any output buffer is full.
• Transforms that drain one input fully before another (e.g. Stream Lookup info-first) leave the other path full → source never finishes → never setDone() → hang.

Raising rowset size, redesigning streams, or inserting Blocking are valid workarounds; they are not a generic engine fix.

Decisions and rationale

Rejected: Stream Lookup-only loader threads

• Not generic (Merge Join and friends still hang).
• Eager full-stream load increases OOM risk (why rowsets are bounded in the first place).
• Extra transform lifecycle / concurrency complexity.

Rejected: spill “info hops only”

Info-only spill does not fix classic Stream Lookup: the main hop is the one left unread while the lookup cache loads, so the shared source still blocks on main. Merge Join happens to list both inputs as INFO, so info-only would accidentally help there — wrong rule for the headline case.

Chosen: detect split–rejoin, then spill recommended inbound hops

Detection (PipelineBufferDeadlockAnalyzer):

• Multi-input transform T (≥2 predecessors via enabled hops, including info).
• At least two predecessors share a common ancestor (walk findPreviousTransforms).
• Risk record includes reconvergence, ancestor, inbound preds, and recommended spill hops.

Spill hop set (v1, safe minimal): all inbound hops predecessor → T involved in the risk — typically two hops per site, not the whole pipeline. That includes main+info for Stream Lookup without encoding every transform’s read order.

Not in v1: mid-path spill (e.g. before Sort on a long branch); auto-inserting Blocking transforms; unbounded pure-memory queues.

Spilling rowset (SpillingRowSet)

• Same contract as BlockingRowSet while under capacity.
• When full (or FIFO requires append behind existing spill): serialize row data only via IRowMeta.writeData (not metadata per row; same pattern as Sort / Blocking / Join Rows).
• putRow succeeds without waiting on the consumer — that is what breaks the wait-for cycle.
• Temp files via HopVfs under configurable directory (default system temp).
• Cleanup on clear() / done path.

Critical performance fix: size() and BaseTransform sleeps

First mitigation builds were pathologically slow (~38k rows/s, near-zero CPU/disk). Cause:

• BaseTransform does Thread.sleep(0, 1) when size() >= 99% of rowset size (and a low-water sleep when nearly empty).
• On many Linux/JVMs that “1 ns” sleep rounds to ~1 ms.
• While spilling, memory stays full, so every put hit that sleep.

Fix: SpillingRowSet.size() is a flow-control signal: with pending work it reports a mid-level occupancy so producers/consumers are not artificially throttled; idle still reports 0. After the fix, ~825k rows/s on a large test pipeline.

Defaults (local run configuration GUI)

Option	Default	Why
Detect buffer deadlocks	On	Always surface risk in prepare log + pipeline verify
Mitigate (spill to disk)	On	Prefer progress over hanging; only analyzer-flagged hops spill
Spill directory	empty → system temp	Standard Hop temp behavior

Existing saved run configs keep their stored flags. Only new local configs get these defaults.

Product rationale for default-on mitigate: detection already logs risks and spill usage; locking up for hours is worse for data engineers than occasional disk I/O on known-bad topologies. Users who want strict bounded memory only can turn mitigation off.

Scope boundaries

• Local multi-threaded engine only (LocalPipelineEngine + Pipeline rowset allocation).
• Single-threaded engine keeps QueueRowSet.
• No transform plugin code changes (Stream Lookup unchanged).
• Beam / Spark unchanged.

Surfaces

1. Pipeline verify — WARNING per risk site.
2. Prepare (local) — log risks; if mitigate on, log hop count using spilling rowsets; if mitigate off, warn that hang is possible.
3. User guide — avoiding-deadlocks.adoc section 5 + short design notes.

Test plan

• Unit: SpillingRowSetTest (FIFO past capacity, clear, size() not advertising full while spilling)
• Unit: PipelineBufferDeadlockAnalyzerTest (split–rejoin risk, independent sources no risk, linear chain no risk)
• Manual: large split–rejoin / Stream Lookup-style pipeline with mitigate on — completes; throughput after size() fix
• CI: full build / spotless
• Review: existing local run configs without new keys still load; new configs show both checkboxes on in the GUI

Related

• Closes / fixes discussion in [Bug]: [Stream Lookup] Proposal: Use threads + CountDownLatch to eliminate deadlocks #7752
• Related docs history: avoiding-deadlocks guide, [Task]: Document how Pipelines using Stream Lookup can run into deadlock #3740