[andygrove]

Which issue does this PR close?

Stacked on #2670. The first six commits here are #2670 (the pipelined exchange operators, with its review fixes, the executor listen backlog, the retry of a stream open whose connection failed, and one shared connection per executor pair). The last commit, "feat: run pipelined jobs end to end", is the rest of the feature. This PR is kept as a draft for end-to-end testing while the pieces are split out for review.

Closes #2652.

Rationale for this change

#2652 proposes an opt-in pipelined execution mode. In this mode the scheduler places every task of a job up front, and stages stream record batches to each other over Arrow Flight instead of running stage by stage through shuffle files. The goal is lower latency for interactive queries. The blocking shuffle stays the default, and nothing changes unless a session sets ballista.execution.mode = pipelined.

What changes are included in this PR?

The feature works end to end and is ready for review. The answers to the open questions in #2652 are in this comment.

• Setting. ballista.execution.mode = blocking | pipelined, plus ballista.pipelined.queue_capacity and ballista.pipelined.registration_timeout_seconds.
• Exchange (ballista/core/src/execution_plans/pipelined/).
  • PipelinedShuffleWriterExec routes each output partition into a bounded queue in a per-executor PipelinedRegistry.
  • PipelinedShuffleReaderExec drains all of its producers concurrently, either locally or over a new FetchPipelinedPartition Flight ticket.
  • Broadcast sides, and inputs that every task reads in full, are replayed from a buffer.
  • The protobuf messages for these are new.
• Executor. The executor owns the registry and serves it from do_get. All of a job's tasks on one executor share one memory pool, sized by the job's reservation there. Job cleanup drops the job's registry entries.
• Scheduler (state/pipelined_graph.rs).
  • PipelinedExecutionGraph plans the job statically and places all of its tasks at once, or waits until it can.
  • Each executor reserves the vcores of the widest stage it runs.
  • A stage wider than the cluster waits for every vcore and then shares them.
  • Placement is push-staged only; a pull-staged scheduler rejects pipelined jobs.
  • A failed or cancelled job releases its reservation and revives offers.
• Join selection (state/pipelined_join_selection.rs). Without AQE, pipelined jobs are planned with hash joins. A partitioned hash join is demoted to a sort-merge join when its estimated build size per partition exceeds hash_join_max_build_partition_bytes or is unknown, since every hash table in a pipelined job is resident at once and none can spill.
• Client and Flight SQL.
  • The client reads every final-stage partition while the job runs.
  • Dropping the result stream cancels the job.
  • A failed fetch reports the job's own failure.
  • The Flight SQL backend returns a pipelined result through one merged endpoint.
• Standalone. A standalone cluster runs push-staged when its session asks for pipelined execution.
• Docs. A user guide page (user-guide/pipelined-execution.md) covers when to use the mode, how a job runs, memory and settings.

Design notes

These are the decisions made while implementing, each with its reason. The answers to #2652's open questions are linked above.

Scheduling and placement

• Push-staged only. Under pull-staged scheduling each executor claims work on its own poll, so a job can't be placed atomically. The executor's vcore permits would also block tasks that wait on each other. For this reason standalone mode runs push-staged when the session asks for pipelined execution.
• The reservation is held for the whole job. The vcores are reserved once, at placement, and released once, when the job ends. Releasing them per task would hand vcores back while other tasks of the job are still running on that executor. Task infos carry vcores_consumed = 0 so the normal per-task refunds stay correct.
• A stage wider than the cluster shares vcores. The reservation is already a soft budget, and a waiting task is a parked future, not a thread. A UNION ALL is as wide as all of its inputs together, so rejecting wide stages would fail ordinary queries. The cost is that a wide job waits until the whole cluster is free.
• Placement and reservation never hold two locks at once. The placement is planned under the graph read lock, vcores are reserved atomically, and the placement is committed under the graph write lock. This avoids the vcore-then-graph lock order that bind_schedulable_tasks uses.
• One memory pool per job per executor. All of a job's tasks run at the same time, so a pool per task would let the job exceed the executor's memory. The shared pool is sized by the job's reservation, the same way a blocking task's pool is sized by its own vcores. The executor holds the runtime weakly, so it is dropped when the job's last task there finishes.

Exchange

• Bounded queues. Each non-broadcast output partition is a bounded mpsc queue, so a slow consumer stalls its producer. In DataFusion Distributed, by contrast, data for a slow consumer piles up in the producer. Bounded queues can't deadlock for tree-shaped plans as long as every reader drains all of its producers concurrently, which PipelinedShuffleReaderExec does. The hash writer finishes its partitions concurrently, so one full queue doesn't hold back the end markers of the others.
• Explicit end-of-stream marker. A queue that closes without Message::End came from a producer that failed or was cancelled. The consumer gets the producer's error, never a silently truncated partition.
• A producer finishes when its input does. It doesn't wait for consumers to drain, so its vcores are released promptly. Undrained queues stay in the registry until a consumer claims them or the job is cleaned up. If every consumer goes away, for example after a satisfied LIMIT, the writer stops reading its input.
• Broadcast outputs are replayed. They are kept until job cleanup, so that every consumer task can read them in full. Their size is bounded by broadcast_join_threshold_bytes. Any output that a stage of several tasks reads in full is replayed the same way, such as a cross join's left side. Nothing bounds those, but blocking mode reads that data once per task too.
• One timeout bounds both sides. ballista.pipelined.registration_timeout_seconds limits how long a consumer waits for its producer to register, and how long a producer waits for someone to open a full queue. A slow consumer that has already opened its queue is ordinary backpressure and isn't timed out. The producer waits on a queue reservation rather than a send, so a timeout loses no batch.
• The fetch responds before the producer exists. do_get returns at once and waits for the producer inside the stream body, so a consumer that arrives early isn't cut off by its client's request timeout.

Executor

• The registry reaches tasks through a SessionConfig extension. Plans aren't rewritten, so custom execution engines that leave the plan alone still work.
• One connection per executor pair. The executor's client pool keeps one connection to each other executor, and the streams of all its tasks share it. The stream window (ballista.pipelined.stream_window_bytes, 512 KiB) is small and the connection window is the largest HTTP/2 allows, so stalled streams cannot starve the ones something is waiting on. A stream open whose connection fails before the other executor answers is retried; a stream that has started is not. The executors' listeners take a 4096 backlog for the burst the first job after a restart opens.
• Co-located reads skip Flight. When the producer runs on the same executor (matched by executor id), the reader opens the registry directly. ballista.shuffle.force_remote_read forces the Flight path for tests.
• The registry remembers finished jobs. It keeps the last 1024 cleaned-up job ids, so a producer that starts after its job was cleaned up fails at once instead of waiting for consumers that are gone.

Client and Flight SQL

• The client reads every final partition at once. The final stage can't finish until its output is read, so the client can't wait for SuccessfulJob or read the partitions one after another, as it does in blocking mode.
• Flight SQL serves a pipelined result through one endpoint. A client may read endpoints one at a time, and the job can't finish writing one partition until the others are being read. The partitions are therefore merged on the scheduler, which every Flight SQL result already passes through.

How are these changes tested?

• Unit tests for the exchange, the registry, placement, the pipelined graph and join selection.
• ballista/client/tests/pipelined.rs compares queries against blocking mode in three setups: a standalone cluster, the same with forced remote reads, and a scheduler with two executors. It also covers LIMIT, a divide-by-zero reaching the client, a pull-staged cluster rejecting the job, and a dropped result stream freeing the cluster.
• ballista/scheduler/tests/flight_sql.rs runs a pipelined job through Flight SQL.
• The TPC-H workflow gains a pipelined leg that runs SF10 with --verify against DataFusion.
• TPC-H SF1000 on a two-executor Kubernetes cluster, with 8 vcores, a 48 GB pool and 16 partitions. All 22 queries pass in 2450 s, against 3374 s in blocking mode with AQE (0.73x). Per-query numbers are in this comment.
• TPC-H SF1000 on a 32-executor Kubernetes cluster, with 8 concurrent tasks each and 256 partitions, from an image built at cb49f30f (2 iterations per query). All 22 queries pass in 278.8 s, against 517.0 s for blocking mode on the same image and cluster (0.54x). Every query is faster, from 7% (q1) to 80% (q14). The 22 queries return the same row counts in both modes; I compared counts only, not contents. q11 returns 0 rows and q13 returns 30 in both modes. Per-query numbers are in this comment.
• The same cluster's kernel TCP counters, over the run: ListenOverflows 0 and PassiveOpens 3,590. A single query (q7, two iterations) on d0572fc7, before the shared connection per executor pair, gave 37,779 and 56,230. Earlier builds failed 7 to 8 of the 22 queries at this size with ConnectionReset.

Per-query times on the 32-executor cluster, mean of 2 iterations, in seconds, from the image at cb49f30f:

Query	blocking	pipelined	change		Query	blocking	pipelined	change
q1	14.8	13.8	-7%	q12	10.1	4.4	-57%
q2	36.3	13.5	-63%	q13	10.8	5.8	-46%
q3	17.9	7.3	-60%	q14	7.0	1.4	-80%
q4	10.9	6.1	-45%	q15	10.6	6.9	-35%
q5	41.4	13.9	-66%	q16	13.3	5.1	-61%
q6	1.7	1.4	-19%	q17	23.6	15.4	-35%
q7	35.1	18.7	-47%	q18	55.4	35.4	-36%
q8	21.6	14.1	-35%	q19	12.1	7.4	-39%
q9	35.8	24.2	-32%	q20	15.6	6.3	-60%
q10	45.1	22.0	-51%	q21	68.6	45.9	-33%
q11	18.7	4.4	-77%	q22	10.5	5.4	-48%

Total: 278.8 s pipelined against 517.0 s blocking (0.54x).

Known limitations:

• On the two-executor cluster, q8 at SF1000 is 3.4x slower than blocking. AQE broadcasts the filtered part side once it knows its size, but the static estimate is too large to broadcast. On the 32-executor cluster q8 is faster than blocking (14.1 s against 21.6 s).
• A failed task fails the whole job. No task is retried; the only retry is of a stream open whose connection failed before the other executor answered.
• A pipelined job can need more memory than the same query in blocking mode.

Are there any user-facing changes?

There are four new session settings: ballista.execution.mode, ballista.pipelined.queue_capacity, ballista.pipelined.registration_timeout_seconds and ballista.pipelined.stream_window_bytes. The mode is experimental and off by default, and blocking behavior is unchanged.

[andygrove]

TPC-H SF1000: pipelined vs blocking

I ran TPC-H SF1000 with ballista.execution.mode=pipelined at 83edc87, after rebasing onto main so that #2659 is included. All 22 queries complete. The total is 3795 s, 1.12x the blocking total. Most queries are faster, but a few regress badly: Q8 is 8x slower and Q9 3x.

Setup: 2 executors on 2 nodes, each with 8 vCPU, a 56 GiB pod limit, --memory-pool-size=48GB and --max-temp-directory-size=1TB. target_partitions=16, Parquet read from node-local disk, 1 iteration.

Caveat: the blocking column is an earlier run on the same cluster, built from the #2659 branch rather than this PR's image, so treat the comparison as approximate.

An earlier pipelined run, on this branch before the rebase, failed Q5, Q8, Q9, Q17, Q18 and Q21. Tasks exceeded DataFusion's default 100 GB spill limit, which executors could not raise before #2659. Pipelined mode holds every stage's state at the same time, so it spills more than blocking does. That makes --max-temp-directory-size effectively required for pipelined runs at this scale.

Query	Blocking (s)	Pipelined (s)	Pipelined / blocking
1	27.2	26.4	0.97x
2	51.2	98.4	1.92x
3	137.4	135.0	0.98x
4	51.4	46.8	0.91x
5	430.0	344.3	0.80x
6	12.6	8.6	0.69x
7	374.4	360.8	0.96x
8	45.0	361.2	8.03x
9	199.6	598.5	3.00x
10	136.6	109.5	0.80x
11	55.5	85.5	1.54x
12	73.4	53.6	0.73x
13	69.9	41.5	0.59x
14	19.0	15.0	0.79x
15	43.0	32.2	0.75x
16	23.8	16.0	0.67x
17	225.9	374.8	1.66x
18	459.7	409.2	0.89x
19	26.2	22.6	0.86x
20	63.8	57.6	0.90x
21	817.1	575.4	0.70x
22	31.6	22.1	0.70x
Total	3374	3795	1.12x

17 of 22 queries run at 0.59–0.98x of blocking. Q8 and Q9 alone add about 715 s, so they are the ones to look at next. I have scheduler event logs for every query in this run.

[andygrove]

Why the slower SF1000 queries regress

This follows up on the SF1000 numbers above, using the scheduler event logs from the blocking and pipelined runs.

Root cause: with no AQE, every join is a sort-merge join. Pipelined jobs are always planned with the static planner (task_manager.rs, "A pipelined job plans with the static planner whatever the AQE setting"). With datafusion.optimizer.prefer_hash_join=false (the default) and no runtime statistics, every join in all 22 pipelined plans is a SortMergeJoinExec, and none is a hash join or a broadcast. In blocking mode, AQE (on by default) replans joins from runtime sizes and produces broadcast (CollectLeft) and partitioned hash joins wherever an input is small. ballista.optimizer.broadcast_join_threshold_rows/bytes only takes effect through AQE.

Q8 is the clearest case:

• Blocking: the 1.33M filtered part rows are broadcast into the lineitem scan stage, a HashJoinExec with mode=CollectLeft. That stage takes 23 s, and lineitem is never shuffled.
• Pipelined: all 6B lineitem rows are hash-repartitioned on l_partkey and then sorted for a SortMergeJoinExec against the 1.33M parts. That one stage takes 346 s and 1624 CPU-s. These large sorts are also what pushed the spills past the 100 GB default limit before fix: stream remote shuffle fetches, make the executor spill limit configurable, and spill to --work-dir #2659.

Q9, Q17, Q2 and Q11 have the same shape. Q17 sort-merge-joins about 12B rows, where blocking mode has no sort-merge join at all.

Experiment: pipelined with datafusion.optimizer.prefer_hash_join=true

Query	Blocking (AQE)	Pipelined	Pipelined + prefer_hash_join
8	45.0 s	361.2 s	137.9 s
17	225.9 s	374.8 s	155.5 s
9	199.6 s	598.5 s	fails, out of memory

• Q9 fails. The 1.5B-row orders table becomes the build side of a partitioned hash join against lineitem. A pipelined job keeps every build resident at once, hash-join builds can't spill, and pipelined tasks aren't retried. The error was ResourcesExhausted with several HashJoinInput reservations of 3.2 GB each in a 44.7 GB pool. AQE keeps a sort-merge join for this join in blocking mode.
• Q8 is still 3x blocking with hash joins. The part⋈lineitem join stage uses only 75 CPU-s over 138 s. It is waiting on the 6B-row lineitem repartition over the 2.5 GbE network. Blocking avoids that by broadcasting the filtered parts into the scan.

Implication: a global hash-join or sort-merge-join switch won't fix this. Pipelined mode needs size-aware join selection at static planning time: estimate build-side sizes from Parquet and row-count statistics, and apply the existing broadcast thresholds, roughly what AQE decides at runtime. Broadcasting small filtered build sides into the large scan looks like the biggest win (Q8, Q9, Q17), with partitioned hash joins for medium inputs and sort-merge joins kept for very large ones.

Setup: same cluster as above, at 83edc87.

[andygrove]

SF1000 results with size-aware join selection (34a1151)

Following up on the slow-query analysis above: pipelined jobs now plan with hash joins and PipelinedJoinSelection demotes a partitioned hash join to a sort-merge join when its estimated build side per partition exceeds ballista.hash_join_max_build_partition_bytes (64 MiB default) or the size is unknown. CollectLeft joins are left alone, and a limit of 0 disables the check.

Same setup as before: 2 executors x 8 vCPU, 48 GB pool, 1 TB spill cap, 16 partitions, local data, one iteration.

All 22 queries pass. Total 2450s, down from 3795s for pipelined before this change, and 0.73x the 3374s blocking baseline.

Query	Blocking (s)	Pipelined before (s)	Pipelined + join selection (s)	vs blocking
1	27.2	26.4	26.6	0.98x
2	51.2	98.4	26.3	0.51x
3	137.4	135.0	120.3	0.88x
4	51.4	46.8	16.4	0.32x
5	430.0	344.3	284.4	0.66x
6	12.6	8.6	9.2	0.73x
7	374.4	360.8	248.4	0.66x
8	45.0	361.2	151.4	3.37x
9	199.6	598.5	246.5	1.24x
10	136.6	109.5	94.5	0.69x
11	55.5	85.5	23.5	0.42x
12	73.4	53.6	49.1	0.67x
13	69.9	41.5	41.5	0.59x
14	19.0	15.0	14.9	0.78x
15	43.0	32.2	26.1	0.61x
16	23.8	16.0	10.0	0.42x
17	225.9	374.8	157.4	0.70x
18	459.7	409.2	360.7	0.78x
19	26.2	22.6	22.1	0.84x
20	63.8	57.6	49.3	0.77x
21	817.1	575.4	423.2	0.52x
22	31.6	22.1	48.1	1.52x
Total	3374	3795	2450	0.73x

Notes:

• Q9 no longer OOMs the way it did with a global prefer_hash_join=true. The large orders build is demoted to SMJ, and the small dimension joins stay as hash joins.
• Q8 is still 3.4x blocking. In blocking mode AQE sees the actual size of the filtered part side and broadcasts it. The static estimate is too large to broadcast, so lineitem is shuffled over the network. Better static estimates, or runtime join adaptation within pipelined mode, would be needed to close this gap.
• The Q22 plan is unchanged from the previous pipelined run, so the 22s to 48s change looks like run-to-run noise and not an effect of this change.

[andygrove]

cc @phillipleblanc

[andygrove]

I'm splitting this into smaller PRs for review, and keeping this one open as a draft for reference (full design notes, end-to-end tests, SF1000 results). Each split PR is cut from this branch's final state.

1. feat: let a standalone cluster run push-staged #2669: let a standalone cluster run push-staged (no pipelined concepts)
2. feat: add pipelined exchange operators #2670: pipelined exchange operators, registry, protobuf and settings (nothing plans them yet; the scheduler rejects pipelined jobs)
3. Executor wiring (registry ownership, serving FetchPipelinedPartition, shared memory pool per job)
4. Scheduler: pipelined execution graph and all-at-once placement
5. Client streaming results, which turns the mode on, plus the end-to-end tests, TPC-H CI leg and user guide
6. Flight SQL support
7. Size-aware join selection for pipelined jobs

I'll link the rest here as they're opened.

[andygrove]

Benchmark of the pipelined mode at TPC-H SF1000: 15 of 22 queries run and are much faster, 7 fail with ConnectionReset between executors

I ran TPC-H at SF1000 (Parquet, zstd, Hive-partitioned, S3) on Kubernetes at 34a1151a: 32 executors with 8 concurrent tasks each (target partitions 256), --task-scheduling-policy push-staged on the scheduler and executors, 2 iterations per query. The comparison is the same build and cluster shape with ballista.execution.mode left at blocking (pull-staged).

Speed

Pipelined, for the queries that completed (mean of 2 iterations, seconds):

Query	blocking	pipelined	change
q1	15.7	13.5	-14%
q3	19.9	8.7	-56%
q4	12.6	5.7	-55%
q6	1.7	2.0	+23%
q7	36.7	17.2	-53%
q9	34.0	23.4	-31%
q10	37.4	25.6	-31%
q12	10.8	3.6	-66%
q13	10.9	5.7	-48%
q14	6.1	1.6	-74%
q15	11.2	6.4	-43%
q17	23.6	14.6	-38%
q18	57.1	34.8	-39%
q19	10.6	7.2	-33%
q22	10.8	5.8	-46%

q6 is a 2 s query, so I'd read that as noise. The other 14 are 14 to 74% faster.

Failures

q2, q5, q8, q11, q16, q20 and q21 failed. Each failed job reports a task failing because a FetchPipelinedPartition stream was reset, for example:

Pipelined job 038BX5GTR0000 failed because task 14 of stage 7 failed: Task failed due to runtime
execution error: GrpcActionError("failed to stream pipelined partition PartitionId { job_id:
JobId("038BX5GTR0000"), stage_id: 6, partition_id: 113 } of task 25: code: 'Unknown error',
message: "transport error", source: tonic::transport::Error(Transport, hyper::Error(Io,
Kind(ConnectionReset)))"). Pipelined jobs do not retry tasks

The failing stage, task and partition differ from job to job, so it is not one bad partition. The failing queries tend to be the ones with the most joins and stages (q5, q8, q21).

What the executor logs show for q2 (job 038BX5GTR0000):

• All 32 executors have RESTARTS 0, so no executor was killed.
• Memory is not close to the limit. Executor logs show about 315 MiB allocated against a 45,875 MiB pool during the failure.
• The first error is on the reading executor (stage 7 task 14, 13.8 s after the job launched). The other 33 failures within 0.1 s are the same error propagating through downstream tasks and the job being cancelled.
• The producing executor logged nothing about the stream. It has no WARN or ERROR lines, and its stage 6 task had finished about 10 s earlier. So the producer did not report a failure, and the reset is visible only on the reading side.

What I don't know

• I don't know why the connection resets. My guess, untested, is the fan-in: each reader drains all 256 upstream partitions concurrently, so a stage 7 task opens a stream to every stage 6 producer at once, and with 8 tasks per executor over 32 executors that is a large all-to-all mesh. I'll test it by rerunning the failed queries on fewer, smaller executors, and with hyper/h2 debug logging to find which side sends the RST.
• I don't know whether the queries that passed were just lucky.

Suggestions

• If a reset is plausible under load, retrying only a stream open (before any batch was consumed) would keep one transient failure from failing the whole job.
• Log on the producing side when a FetchPipelinedPartition stream ends abnormally, so a reset can be attributed.
• Say in pipelined-execution.md what cluster width this is expected to work at.

Logs and the result JSON are available if useful.

[andygrove]

SF1000 results after stacking on #2670 (5b57a9e)

This run is on the current head, which includes #2670's review fixes:

• A remote pipelined stream now keeps its pooled connection leased while it is read.
• Restricting a broadcast reader to some partitions keeps every location.
• A consumer that times out leaves no waiting slot in the registry.
• The pipelined timeout and queue-capacity settings are clamped to at least 1.
• The writer and registry were simplified.

Setup is the same as before: 2 executors x 8 vCPU, 48 GB pool, 1 TB spill cap, 16 partitions, local data, one iteration.

All 22 queries pass in 2475s. That is 1.01x the previous pipelined run (2450s) and 0.73x blocking mode (3374s).

Query	Blocking (s)	Pipelined, 34a1151 (s)	Pipelined, 5b57a9e (s)	vs previous	vs blocking
1	27.2	26.6	29.1	1.10x	1.07x
2	51.2	26.3	24.9	0.95x	0.49x
3	137.4	120.3	119.0	0.99x	0.87x
4	51.4	16.4	16.5	1.01x	0.32x
5	430.0	284.4	316.9	1.11x	0.74x
6	12.6	9.2	9.1	0.99x	0.72x
7	374.4	248.4	288.3	1.16x	0.77x
8	45.0	151.4	138.3	0.91x	3.08x
9	199.6	246.5	272.8	1.11x	1.37x
10	136.6	94.5	103.9	1.10x	0.76x
11	55.5	23.5	24.3	1.03x	0.44x
12	73.4	49.1	42.7	0.87x	0.58x
13	69.9	41.5	40.1	0.97x	0.57x
14	19.0	14.9	14.3	0.96x	0.75x
15	43.0	26.1	25.8	0.99x	0.60x
16	23.8	10.0	10.9	1.09x	0.46x
17	225.9	157.4	160.2	1.02x	0.71x
18	459.7	360.7	340.4	0.94x	0.74x
19	26.2	22.1	22.1	1.00x	0.85x
20	63.8	49.3	52.2	1.06x	0.82x
21	817.1	423.2	401.2	0.95x	0.49x
22	31.6	48.1	22.1	0.46x	0.70x
Total	3374	2450	2475	1.01x	0.73x

Notes:

• The totals are within noise. Q5, Q7, Q9 and Q10 were 10–16% slower than the previous run, and Q8, Q12, Q18 and Q21 were faster. With one iteration each, I can't separate noise from the effect of leasing a pooled connection to each remote stream. Repeated runs of the join-heavy queries would settle it.
• Q22 is back to 22s, which confirms that the 48s in the previous run was noise.
• Q8 is still the one large regression against blocking mode, for the reason described above (the static size estimate of the filtered part side).

One gap showed up while running this. If a client process is killed outright, as opposed to dropping its result stream, its pipelined job keeps running on the scheduler and holds its vcores until it is cancelled through PATCH /api/job/{job_id}. Cancel-on-drop only works while the client process is alive, and the scheduler doesn't notice a client that has gone away. Blocking jobs behave the same way, but a pipelined job's reservation blocks every other pipelined job, so it matters more here. Possible fixes are a lease or heartbeat from the client, or cancelling when the result fetch connections all close.

[andygrove]

Follow-up: the ConnectionResets are accept-queue overflows on the executors

I reran q2 at 5b57a9e8 (32 executors × 8 tasks, SF1000, pipelined, h2/hyper/tonic debug logging) and read the kernel TCP counters from every executor pod (/proc/net/netstat, /proc/net/snmp) right after the job failed. The pods were fresh, so the counters cover only this run's startup and the failed q2 job.

counter (summed over 32 executors)	value
TcpExt.ListenOverflows	54,476 (nonzero on all 32)
TcpExt.ListenDrops	54,476
TcpExt.TCPSynRetrans	49,793
TcpExt.TCPAbortOnData	12,042
Tcp.EstabResets	2,630
Tcp.OutRsts	26,296
Tcp.PassiveOpens (accepted connections)	72,887, up to 2,392 on one executor

So within about 1.6 s of the job starting, each executor was asked to accept thousands of connections and its listen queue overflowed. A dropped final ACK leaves the client believing the connection is open, and the server then answers its first data with an RST. That produces the Kind(ConnectionReset) in the task error, and Connection reset by peer (os error 104) in the executors' tonic logs. No process logs anything before the first error, because the drop happens in the kernel.

The timing matches. The scheduler placed the job at 13:44:13.36, the first h2 IO error error=ConnectionReset on a client connection is at 13:44:14.921, and the scheduler failed the job right after. The same happened in the earlier runs: 1.7 s and 1.6 s after the job started. (An even earlier run took 13.8 s; I haven't captured counters for that one.)

I think the cause is the number of connections: each reader drains all producers concurrently, so at 256 partitions the job opens about 73k connections across 32 executors, compared with the 31 that one HTTP/2 connection per executor pair would need. The executor binds its gRPC listener with TcpIncoming::bind, which uses the default backlog. I haven't confirmed whether each stream gets its own connection. I'm guessing that from the PassiveOpens count.

Things that might help, in the order I'd try them:

• Share one Channel (one HTTP/2 connection, multiplexed) per executor pair for the Flight fetches instead of connecting per stream.
• Retry a stream open when the connection is reset before any batch has been consumed. That is safe, and it would turn this failure into a retry.
• Let the executor bind with a larger backlog, and document net.core.somaxconn for pipelined mode.

This explains the 7 failing queries of 22 in my earlier comment. It would also explain why they are the ones with the most stages and the widest fan-in. I can run any of these changes against the same cluster if that helps.

[andygrove]

Update: confirmed in the code. The reader checked a pooled connection out for every stream and held it until the stream ended, so every producer and consumer task pair got its own connection.

e573d27 in #2670 makes each reader partition share one connection per producer executor, with at most 64 streams on a connection. It stops short of one connection per executor pair. With the 64 MB connection window and 16 MB stream window, a few stalled streams (say a join's probe side while it builds) would fill the connection and starve the build side's streams on it. Within one reader partition all the streams feed the same output, so that can't happen.

This should cut q2 from about 73k connections to roughly 8k. #2657 is rebased on it. Next step is rerunning q2 at d0572fc. If it still resets, I'll add the retry before the first batch and the backlog setting.

[andygrove]

Follow-up: d0572fc7 still fails 8 of 22 TPC-H queries with ConnectionReset, and the accept-queue counters don't separate the passing runs from the failing ones

I rebuilt at d0572fc7 (includes e573d27, one shared connection per producer executor) and ran the whole TPC-H SF1000 suite (32 executors × 8 tasks, pipelined, 2 iterations). The runner confirmed commit=d0572fc7….

• Passed (14): q1, q3, q4, q6, q9, q10, q12, q13, q14, q15, q17, q18, q19, q22. Their sum is 167.3 s against 262.3 s for the same queries in blocking mode.
• Failed (8): q2, q5, q7, q8, q11, q16, q20, q21, all with Kind(ConnectionReset) while streaming a pipelined partition, as before. q7 passed on 34a1151a and fails now; q2 failed in this run although it passed when I ran it alone on the same image (14.6 s).

So the shared connection did not fix the failures, and q2's single passing run was luck.

I also need to correct my previous comment, where I said accept-queue overflow explains the resets. The kernel counters I read from the 32 executors after a failing q2 on 5b57a9e8 and after the passing q2 on d0572fc7 are about the same:

counter (summed over 32 executors)	failing q2 (5b57a9e8)	passing q2 (d0572fc7)
ListenOverflows / ListenDrops	54,476	61,480
TCPSynRetrans	49,793	52,613
Tcp.PassiveOpens	72,887	72,764
Tcp.EstabResets	2,630	2,529

The queue overflowed in the run that passed too, and the number of accepted connections didn't fall, so I can't say overflow is the cause, or that most of the 73k connections are the reader's streams. These counters cover each pod's whole life, so they include startup and other traffic. I'm going to snapshot them before the job and sample them during the run, so I can line counter jumps up with the failures. What I do know is the first error in each failing run is a TCP reset seen by a reader (os error 104, Kind(ConnectionReset)), with nothing logged by either process before it.

One thing that might be useful regardless: a stream open that gets reset before any batch is consumed could be retried. All 8 failures are of that kind, and pipelined jobs don't retry tasks.

[andygrove]

Two speculative fixes for the ConnectionResets, in #2670 at 3c6c5cf and rebased here at 667fb67. The executor's Flight listener had a backlog of 128, inherited from std::net::TcpListener::bind via tonic's serve, so the burst of connections a wide job opens overflowed both the accept and SYN queues. Once the kernel falls back to SYN cookies, a connection whose first segment was dropped is reset as soon as it sends more, which matches the timing, the lack of logs on either side, and the counters. The Flight server now listens with a 4096 backlog (GrpcServerConfig::listen_backlog), and the reader retries a stream open whose connection failed before the server answered. SyncookiesFailed in /proc/net/netstat on a producer executor is the counter that confirms or rules out the mechanism.

[andygrove]

cb49f30f: all 22 TPC-H queries pass at SF1000, 46% less total time than blocking on the same image, and the accept queues no longer overflow

Edited: the blocking numbers are now from a back-to-back run on the same image (cb49f30f), replacing an earlier run from the same day, and the row counts are compared.

I reran the full suite at cb49f30f (includes 5d4aa6c, one connection per executor pair). Same setup as before: 32 executors × 8 tasks, SF1000 Parquet on S3, ballista.execution.mode = pipelined (push-staged), 2 iterations per query. The runner reported commit=cb49f30f…. Blocking mode ran on the same image and cluster shape, pull-staged.

All 22 queries completed, where d0572fc7 failed 8 and the first build failed 7, all with ConnectionReset.

Time

Mean of 2 iterations, in seconds:

Query	blocking	pipelined	change		Query	blocking	pipelined	change
q1	14.8	13.8	-7%	q12	10.1	4.4	-57%
q2	36.3	13.5	-63%	q13	10.8	5.8	-46%
q3	17.9	7.3	-60%	q14	7.0	1.4	-80%
q4	10.9	6.1	-45%	q15	10.6	6.9	-35%
q5	41.4	13.9	-66%	q16	13.3	5.1	-61%
q6	1.7	1.4	-19%	q17	23.6	15.4	-35%
q7	35.1	18.7	-47%	q18	55.4	35.4	-36%
q8	21.6	14.1	-35%	q19	12.1	7.4	-39%
q9	35.8	24.2	-32%	q20	15.6	6.3	-60%
q10	45.1	22.0	-51%	q21	68.6	45.9	-33%
q11	18.7	4.4	-77%	q22	10.5	5.4	-48%

The total is 278.8 s against 517.0 s (0.54x). Every query is faster, from 7% (q1) to 80% (q14). Blocking runs on this cluster vary between runs (q2 took 29.7 s and 36.3 s, q10 37.4 s and 45.1 s in two blocking runs today), so the per-query ratios are rough, but the gap is much larger than that variation.

Connections

I read each executor's kernel TCP counters (/proc/net/netstat, /proc/net/snmp) before the first query and again after the last, so these are deltas over the run, summed over the 32 executors:

counter	d0572fc7 (q7 alone, 2 iterations)	cb49f30f (full suite)
ListenOverflows / ListenDrops	37,779	0 (0 on every executor)
TCPSynRetrans	36,833	0
Tcp.PassiveOpens	56,230	3,590
Tcp.OutRsts	21,562	278

The earlier builds overflowed the executors' accept queues by tens of thousands in every run, passing ones included. With the reworked connection sharing there are about 16 times fewer accepted connections and no overflows. I can't tell which of the three fixes (backlog, retry, shared connections) matters most in this run, and I don't need to.

Results

The 22 queries return the same number of rows in both modes (for example q1 4, q3 10, q9 175, q16 27,840, q18 100, q20 112,013, q22 7). q11 returns 0 rows and q13 returns 30 in both modes, so that is not specific to pipelined execution. I compared row counts only, not the contents.

Caveats

This is one run of each mode with 2 iterations.