DOT graph renders SortMergeJoinExec and SortPreservingMergeExec as "Unknown Operator"

[andygrove]

Describe the bug

ExecutionGraphDot labels a physical operator by walking a chain of downcast_ref arms in ballista/scheduler/src/state/execution_graph_dot.rs, falling through to a literal "Unknown Operator" at line 353. Two operators that Ballista plans routinely are missing from that chain:

• SortMergeJoinExec — the chain handles HashJoinExec (line 292) and CrossJoinExec (line 311), but no sort-merge join.
• SortPreservingMergeExec — the chain handles SortExec (line 244) and CoalescePartitionsExec (line 282), but not the order-preserving merge.

This is not an edge case. new_with_ballista() sets datafusion.optimizer.prefer_hash_join = false (ballista/core/src/extension.rs:840), deliberately, because DataFusion's hash join cannot spill (#1648). Sort-merge is therefore the default join for Ballista, so an ordinary joined query renders an unlabeled box where the join should be. SortPreservingMergeExec typically sits at the top of a sorted plan, so sorted queries lose their final operator too.

The graph still has the right shape and edges. Only the label is lost, which makes the node useless for the thing the graph exists for: seeing what a stage actually does.

To Reproduce

Against a scheduler built with rest-api, with nation and region registered as Parquet tables:

select r.r_name, count(*) as num_nations
from nation n join region r on n.n_regionkey = r.r_regionkey
group by r.r_name
order by r.r_name;

Then fetch the graph:

curl -s localhost:50050/api/job/<job_id>/dot

Two of the nodes come back unlabeled — one per unhandled operator:

	subgraph cluster0 {
		label = "Stage 1 [Successful]";
		stage_1_0 [shape=box, label="SortShuffleWriter [1 partitions]"]
		stage_1_0_0 [shape=box, label="Aggregate
groupBy=[r_name@0]
aggr=[count(Int64(1))]"]
		stage_1_0_0_0 [shape=box, label="Projection: ..."]
		stage_1_0_0_0_0 [shape=box, label="Unknown Operator"]
		stage_1_0_0_0_0_0 [shape=box, label="Sort: n_regionkey@0 NULLS FIRST"]
		stage_1_0_0_0_0_0_0 [shape=box, label="DataSourceExec: (...) [1 partitions]"]
...
	subgraph cluster2 {
		label = "Stage 3 [Successful]";
		stage_3_0 [shape=box, label="ShuffleWriter [1 partitions]"]
		stage_3_0_0 [shape=box, label="Unknown Operator"]
		stage_3_0_0_0 [shape=box, label="ShuffleReader [16 partitions]"]

The fallback arm logs what it could not name, so running the scheduler with

RUST_LOG=info,ballista_scheduler::state::execution_graph_dot=debug

identifies them directly:

DEBUG ballista_scheduler::state::execution_graph_dot: Unknown physical operator when producing DOT graph: SortMergeJoinExec
DEBUG ballista_scheduler::state::execution_graph_dot: Unknown physical operator when producing DOT graph: SortPreservingMergeExec

The same two nodes appear unlabeled in the TUI's job graph popup (g on a completed job), which is where this is likely to be noticed.

Expected behavior

Both operators are labeled, in the style of the surrounding arms: the join naming its join keys the way the HashJoinExec arm does with join_expr, and the merge naming its sort expressions the way the SortExec arm does.

Additional context

Worth considering alongside the specific fix: the fallback is silent at info level, so an operator dropping out of the chain shows up only as a blank box in a UI, and only if someone happens to look. DataFusion adds and renames physical operators every release, so this chain drifts by default rather than by accident. A test asserting that no node in a representative plan's graph is labeled "Unknown Operator" would catch the next one at CI time instead of in a screenshot.

Found while verifying the history server in #2265. Not caused by it — the history server relays the scheduler's stored bytes, and the two graphs are byte-identical. It reproduces against a live scheduler on main.