[andygrove]

Which issue does this PR close?

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.

Most of the feature is already on main. Those parts were split out of this PR and merged on their own, and they do nothing until the scheduler accepts pipelined jobs:

• the pipelined exchange operators and settings (feat: add pipelined exchange operators #2670)
• size-aware join selection for pipelined jobs (feat: add size-aware join selection for pipelined jobs #2685)
• the executor serving pipelined partitions (feat: serve pipelined partitions from the executor #2695)
• the client and Flight SQL streaming pipelined results (feat: stream pipelined job results to the client and Flight SQL #2696)
• push-staged scheduling for a standalone cluster (feat: let a standalone cluster run push-staged #2669)

This PR adds the scheduler side and turns the mode on:

• Scheduler (state/pipelined_graph.rs).
  • PipelinedExecutionGraph plans the job statically, with the join selection from feat: add size-aware join selection for pipelined jobs #2685, 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 finished, failed or cancelled job releases its reservation and revives offers. Losing an executor fails the job.
  • submit_job no longer refuses ballista.execution.mode = pipelined.
• Standalone. Pipelined execution overrides ballista.standalone.task_scheduling_policy to push-staged, so a standalone session only has to set ballista.execution.mode = pipelined.
• Docs. The user guide page is in a separate draft PR, docs: add a user guide page for pipelined execution #2713, to be merged after this one.
• CI. The TPC-H workflow gains a pipelined leg.

Reading guide

About 1,000 of the added lines are tests. The core is ballista/scheduler/src/state/pipelined_graph.rs. A suggested order:

1. plan_placement (top of pipelined_graph.rs). A pure function: given each stage's width and the free vcores per executor, can the whole job fit right now, and where does each task go? Its unit tests show the cases: spreading, collapse stages, capped task size, waiting, and a stage wider than the cluster.
2. PipelinedExecutionGraph::new and place. new plans the job statically into stages. place commits a placement: it resolves every reader to its producers' executors (resolve_pipelined_readers) and hands out every task at once.
3. SchedulerState::bind_pipelined_jobs (state/mod.rs). Runs before normal binding on each offer round. It plans under the graph read lock, reserves the vcores atomically (ClusterState::reserve_vcores), then commits under the write lock.
4. Job end (task_manager.rs, query_stage_scheduler.rs). succeed_job and abort_job take the job's reservation (take_reservation) and release it. executor_lost fails a pipelined job instead of rolling it back.
5. submit_job (task_manager.rs). Builds a PipelinedExecutionGraph when the session asks for pipelined execution, and no longer rejects it.

Changes on the path blocking jobs use are small:

• Pipelined tasks are left out of the per-task vcore reservation ledger (holds_vcores). No blocking task is affected.
• executor_lost returns an ExecutorLostOutcome struct. Jobs that lost DataFrame::cache() data are failed through its failed_jobs, with the same message as before.
• Under push-staged scheduling, a failed or cancelled job now triggers a ReviveOffers, and after an executor is lost the ReviveOffers is posted after the job failures instead of before them.

Design notes

These are the scheduler decisions in this PR, each with its reason. The design notes for the exchange, executor and client are in #2670, #2695 and #2696. 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 and pipelined tasks are left out of the per-task reservation ledger, so the normal per-task refunds never hand those vcores back early.
• 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.

How are these changes tested?

• Unit tests for placement, the pipelined graph, reservations and the standalone policy override.
• 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?

Yes. ballista.execution.mode = pipelined now runs jobs instead of being rejected, and in a standalone cluster it implies push-staged scheduling. The setting and its companions (ballista.pipelined.queue_capacity, ballista.pipelined.registration_timeout_seconds and ballista.pipelined.stream_window_bytes) are already on main from #2670. 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.

[akshaychitneni]

Reviewed against the DataFrame::cache() design. One blocker (1), plus fixes so a retry rebuilds lost cache data (3, 4). Should these go into this PR now, or come in a follow-up right after it merges?

Tests: nothing runs both modes on one cluster or session, though #2713 documents SET. Suggest a same-context blocking-then-pipelined client test, plus a cached-frame variant.
cc @andygrove @milenkovicm

[akshaychitneni]

(1) Blocker. Cache builds inherit the session's mode, so a pipelined session builds them as pipelined jobs, which leave nothing to cache (see 2). Suggest: if self.cache_linker.key_of(job_id) is set, submit with a copy of the session that has ballista.execution.mode = blocking, made with session_manager.create_or_update_session so the session id is kept. Read jobs keep the user's mode; reading a blocking-built cache from a pipelined job already works.

[andygrove]

Fixed in 81ff2e3. Cache builds now run blocking whatever the session mode. Rather than going through create_or_update_session, I override ballista.execution.mode on a copy of the job's SessionState, which keeps the session id and leaves the session manager out of it. Read jobs keep the user's mode.

[akshaychitneni]

(2) This empty list is what a pipelined cache build records: the cache ends up empty and reads return no rows without an error (reproduced in the scheduler test harness). A cache larger than its queue fails after registration_timeout_seconds instead (from reading the code). With 1 this can't happen, but record_materialized_cache could also refuse a pipelined graph as a guard.

[andygrove]

Added the guard too, so record_materialized_cache now refuses a pipelined graph.

[akshaychitneni]

(3) This branch skips reads_external_data_on for pipelined jobs. Check both, and use CACHE_DATA_LOST when cache data was lost, so the lost build is dropped.

[andygrove]

Fixed. executor_lost now checks reads_external_data_on before the pipelined branch, so a pipelined job that loses cache data fails with CACHE_DATA_LOST and the build is dropped.

[akshaychitneni]

(4) A FetchPartitionError on cache data (files gone, executor alive) should carry CACHE_DATA_LOST, detected with is_external_fetch_failure as in the static graph. Otherwise every later pipelined query on that cache fails.

[andygrove]

Fixed. A FetchPartitionError on cache data, detected with is_external_fetch_failure, now fails the pipelined job with CACHE_DATA_LOST, so a retry rebuilds the cache.

[andygrove]

Thanks for the careful review against the cache design, @akshaychitneni! I put all four fixes into this PR in 81ff2e3 rather than a follow-up (replies inline).

On tests, there are new scheduler tests for (1), (3) and (4), plus the client tests you suggested. One switches a single session between blocking and pipelined with SET and checks the results match. The other caches a frame from a pipelined session and reads it back twice.

[mbutrovich]

Thanks @andygrove! Splitting the exchange, executor, client and join selection out into their own PRs made the scheduler side easy to follow, and the reading guide helped.

[milenkovicm]

Thanks @andygrove
Let's merge this as experimental and iterate on it

[andygrove]

Thanks @andygrove Let's merge this as experimental and iterate on it

Thank you @milenkovicm