refactor(scheduler): extract REST API wire types into ballista-api-types by andygrove · Pull Request #2256 · apache/datafusion-ballista

andygrove · 2026-08-08T15:52:49Z

Which issue does this PR close?

Part of #1923. This is the first of four PRs splitting up #1925, which is too large to review as one change.

Rationale for this change

Ballista's live TUI reads the scheduler's REST API, but that state is ephemeral: completed jobs are cleaned up after finished_job_state_clean_up_interval_seconds, and everything is gone when the scheduler restarts. #1923 asks for a Spark History Server equivalent, where a standalone server replays durable event logs and serves the same /api/* responses so the existing TUI can browse completed jobs with no scheduler running.

The design is write once, replay verbatim. When a job finishes, the scheduler builds the REST responses once against the live execution graph and writes them into the event log's terminal record; the history server deserializes those values and re-serializes them unchanged. Byte-identical output is therefore a structural property, not something two implementations have to keep agreeing on.

Two things have to be true before any of that can be built, and both are refactors rather than features:

The response types have to live somewhere every party to the /api/* contract can reach. Today they are private types inside ballista-scheduler.
Response construction has to be callable outside an axum handler, since the event-log writer runs at job completion rather than on an HTTP request. Today it is inlined in the handler bodies.

This PR does both, and nothing else. No new feature, no new config, no behavior change. Landing it separately keeps the reviewable question down to one thing: is this behavior preserving?

Why `ballista-api-types` and not `ballista-history`

The crate started out named for the history server, but the /api/* contract has three parties, not one. The scheduler serves it, the history server will serve replayed copies of it, and the web TUI already consumes it — with its own hand-maintained duplicate of these structs in ballista-cli/src/tui/domain/jobs*.rs.

That duplicate has already drifted: 2ed3464df changed TaskSummary::partition_id from u32 to Vec<u32> and the TUI copy was not updated, so it still declares u32. See #2257. Naming the crate after the contract rather than after one of its consumers is what makes it natural for the TUI to drop its copy, which is the follow-up in #2258.

The crate stays serde-only, which is what lets the TUI depend on it from a wasm32 build.

What changes are included in this PR?

New ballista-api-types crate. A leaf crate whose only dependency is serde. It is the single definition of what /api/* puts on the wire: JobResponse, TaskSummary, TaskStatus, Percentiles, QueryStageSummary, QueryStagesResponse, and PlanFormat. The crate doc names the three consumers and the rule for what is admitted.

Types describing live scheduler internals (SchedulerStateResponse, SchedulerVersionResponse, CancelJobResponse) stay in handlers.rs; no other party needs them. ExecutorResponse stays too, for a different reason worth being explicit about: it embeds ballista-core types, and keeping this crate serde-only is what makes it cheap to depend on from a wasm32 build.

New api::dto_build module. The graph-to-DTO translation moves out of the handler bodies into pure functions: job_overview_to_response, graph_to_job_response, and graph_to_query_stages, plus the formatting and percentile helpers they use. State in, DTO out, no I/O and no wall-clock reads. handlers.rs drops from 1427 lines to 768 and is now mostly HTTP concerns.

These builders stay in ballista-scheduler on purpose. They need the live execution graph, and nothing on the replay path calls them, so there is no reason to push them into the shared crate.

Some consolidation the extraction made obvious. The three ExecutionStage arms in get_query_stages were near-identical; they collapse into one destructuring match, which also removes the mutable placeholder-zero summary. render_stage_plan, task_summaries, percentiles_of, percent_complete, min_start_time, and format_millis each replace two or three copies of the same code.

Two things pulled out of the handler layer. PlanFormat is part of the wire contract, so it moved to ballista-api-types, and the builders take it by value rather than taking the axum JobQueryParams extractor. And graph_to_query_stages takes now from its caller instead of reading the clock, so replaying a stored log renders stable elapsed times rather than ones that grow on every request.

One type change. JobResponse::job_id is a String rather than ballista_core::JobId, so the new crate can stay serde-only. JobId is #[serde(transparent)] over String, so the serialized JSON is identical.

impl From<&task_status::Status> for TaskStatus becomes the free function task_status_to_dto. Both types are now foreign to ballista-scheduler, so the impl would violate the orphan rule.

Release tooling. ballista-api-types is registered in dev/update_ballista_versions.py (both the crate list and the inter-crate dependency list), in the publish order in dev/release/README.md, and in dev/release/crate-deps.dot. Without this the next version bump would leave the crate behind while ballista-scheduler pinned the old version.

The 12 helper unit tests move to dto_build.rs alongside the functions they cover. The 8 get_webtui tests stay in handlers.rs.

Are there any user-facing changes?

No. REST responses are byte-identical, no public API of ballista-scheduler changes, and no configuration is added. ballista-api-types is new but nothing outside the scheduler depends on it yet.

Verified locally: cargo test -p ballista-scheduler --lib passes (324 tests), clippy is clean for ballista-scheduler and ballista-api-types with --all-features -D warnings, cargo doc is clean, cargo fmt --all --check and taplo are clean, and the --no-default-features check CI runs still passes.

Follow-ups

The remaining three slices of #1925, in order:

Event schema, writer, and reader in a new ballista-history crate depending on this one (no scheduler changes).
Scheduler event-log wiring behind a new event_log_dir config, off by default.
The history server itself, its binary, docs, and the byte-identical-JSON end-to-end test.

One design question that belongs to step 3 rather than here, but is worth flagging early: because responses are rendered once at write time, the stored record carries a single plan format. A history server cannot honour ?plan_format=tree unless step 3 stores every rendering, stores the plan and renders at read time, or the server simply ignores the parameter. The crate doc notes the constraint.

Open question for reviewers

ballista-api-types does not set publish = false, so it would be published on the next release. Given the TUI will depend on it that is probably right, but it is a permanent commitment and worth deciding deliberately rather than by omission. The release tooling above assumes it is published; if we would rather it were not, that is a one-line change here plus dropping it back out of the publish order.

…or out dto_build Move the scheduler's REST response types into a new leaf crate, `ballista-history`, and pull the graph-to-DTO construction out of the axum handlers into a pure `api::dto_build` module. Behavior preserving: the same DTOs are produced from the same state, so live REST responses are byte-identical. The existing handler tests cover this, and the helper unit tests move alongside the functions they test. This is the first step toward a history server that replays completed jobs and serves the same `/api/*` responses without a live scheduler. Splitting the DTOs into a serde-only crate lets that server build the identical wire types without depending on the scheduler's live execution graph, and moving construction out of the handlers means it can run against state that did not come from a handler request. `JobResponse::job_id` becomes a `String` rather than `ballista_core::JobId` so the new crate stays serde-only. `JobId` is `#[serde(transparent)]` over `String`, so the JSON is unchanged.

Follow-up cleanups on the extraction: - Collapse the three near-identical ExecutionStage arms in graph_to_query_stages into one destructuring match, dropping the mutable placeholder-zero summary. - Take PlanFormat by value instead of &JobQueryParams, and move PlanFormat into ballista-history. It is part of the wire contract, and the pure builder no longer imports an axum query-param type back out of the handler module. - Inject `now` into graph_to_query_stages rather than reading the clock, so replaying a stored log renders stable elapsed times. - Share percent_complete and min_start_time; use displayable() instead of the longhand DisplayableExecutionPlan::new(). - Drop the dead JobConfig alias and the unused serde_json dev-dependency, make task_status_to_dto private, and remove a duplicated test. - Enable #![warn(missing_docs)] on ballista-history and document the types, matching the other Ballista crates. - Register ballista-history with the release tooling: version bump script, publish order, and crate dependency graph.

The previous wording left it ambiguous whether the history server re-derives responses from stored execution state or replays stored DTOs. It replays them: the scheduler builds each response once against the live graph and writes it to the event log, so byte-identical output is a structural property rather than two implementations agreeing. Also records the consequence, that anything not captured at write time cannot be recovered at replay time.

The crate holds the /api/* wire types, and it has three parties, not one: the scheduler serves them, the web TUI deserializes them, and a future history server will serve replayed copies. Naming it after the history server made it awkward for the TUI, which parses live scheduler responses and today keeps its own duplicate declarations. Renaming it after the contract it defines removes that friction. The event-log schema, writer, and reader can then land as a separate ballista-history crate that depends on this one.

andygrove · 2026-08-08T16:42:20Z

cc @phillipleblanc

phillipleblanc

Makes sense to me. I verified that 'ballista-api-types' isn't taken on crates.io

milenkovicm

Thanks @andygrove make sense to me

And thank you for reminding me of good old times of JEE, 3 layered architecture, Gang of Four book ... with Data Travel Objects naming 😀😂

* refactor(scheduler): extract REST DTOs into ballista-history and factor out dto_build Move the scheduler's REST response types into a new leaf crate, `ballista-history`, and pull the graph-to-DTO construction out of the axum handlers into a pure `api::dto_build` module. Behavior preserving: the same DTOs are produced from the same state, so live REST responses are byte-identical. The existing handler tests cover this, and the helper unit tests move alongside the functions they test. This is the first step toward a history server that replays completed jobs and serves the same `/api/*` responses without a live scheduler. Splitting the DTOs into a serde-only crate lets that server build the identical wire types without depending on the scheduler's live execution graph, and moving construction out of the handlers means it can run against state that did not come from a handler request. `JobResponse::job_id` becomes a `String` rather than `ballista_core::JobId` so the new crate stays serde-only. `JobId` is `#[serde(transparent)]` over `String`, so the JSON is unchanged. * refactor(scheduler): tighten the DTO extraction Follow-up cleanups on the extraction: - Collapse the three near-identical ExecutionStage arms in graph_to_query_stages into one destructuring match, dropping the mutable placeholder-zero summary. - Take PlanFormat by value instead of &JobQueryParams, and move PlanFormat into ballista-history. It is part of the wire contract, and the pure builder no longer imports an axum query-param type back out of the handler module. - Inject `now` into graph_to_query_stages rather than reading the clock, so replaying a stored log renders stable elapsed times. - Share percent_complete and min_start_time; use displayable() instead of the longhand DisplayableExecutionPlan::new(). - Drop the dead JobConfig alias and the unused serde_json dev-dependency, make task_status_to_dto private, and remove a duplicated test. - Enable #![warn(missing_docs)] on ballista-history and document the types, matching the other Ballista crates. - Register ballista-history with the release tooling: version bump script, publish order, and crate dependency graph. * docs(history): state the write-once, replay-verbatim data flow The previous wording left it ambiguous whether the history server re-derives responses from stored execution state or replays stored DTOs. It replays them: the scheduler builds each response once against the live graph and writes it to the event log, so byte-identical output is a structural property rather than two implementations agreeing. Also records the consequence, that anything not captured at write time cannot be recovered at replay time. * refactor: rename ballista-history to ballista-api-types The crate holds the /api/* wire types, and it has three parties, not one: the scheduler serves them, the web TUI deserializes them, and a future history server will serve replayed copies. Naming it after the history server made it awkward for the TUI, which parses live scheduler responses and today keeps its own duplicate declarations. Renaming it after the contract it defines removes that friction. The event-log schema, writer, and reader can then land as a separate ballista-history crate that depends on this one. * feat(history): add the event-log schema, writer, and reader Second step toward the history server (#1923), after the wire-type extraction in #2256. Adds the durable format and the machinery to read and write it. Nothing in the scheduler calls this yet. - A versioned JSONL schema. JobStart / StageStart / StageEnd / TaskEnd form an incremental timeline; the terminal JobEnd embeds the finished REST responses, so replay re-serves what the scheduler already built rather than re-deriving it. - An async buffered EventLogWriter. All file I/O runs on a background task so the scheduler's event loop never waits on disk. Timeline events are dropped rather than allowed to block when the queue backs up, since losing a progress record beats stalling scheduling. JobEnd is the exception and waits for capacity, because a job missing it is invisible to the history server. - A reader that folds a completed log back into the served payload, skipping malformed lines rather than failing. Restores the JobConfig alias to ballista-api-types, which now has a real consumer in the JobEnd record. TaskEnd names a task rather than a partition: under the multi-partition task model a task owns a slice of partitions, and TaskStatus carries task_id, not partition_id. * feat(history): make the event log survive future Ballista versions A log is written once and may be read years later by a much newer binary, so the guarantee has to run one way: a reader accepts any log whose version is not newer than its own. That is the opposite of BALLISTA_PROTOCOL_VERSION, the strict-equality handshake between scheduler and executor, where both ends are live and upgraded together. The format did not support that yet. Four changes: Self-describing envelope. Every line is now a LogRecord carrying `ev`, `version` and an opaque `data` payload, so a reader can route on the kind and check the version before committing to a shape it may not understand. Previously only JobStart and JobEnd carried a version at all. Stored responses are opaque. JobEnd holds the finished /api/* responses as raw JSON rather than typed structs, plus a small frozen JobIndex for listing. Those responses are ballista-api-types shapes, which change with the live REST contract: partition_id went from u32 to Vec<u32> and TaskStatus::Failed gained a field within one release cycle. Stored typed, either change would have made every older log unreadable, and the job would have silently disappeared. Stored raw, nothing ever parses the inner shape and replay relays the exact bytes. Unreadable is no longer indistinguishable from absent. read_completed_job returned Ok(None) both when a log had no JobEnd and when it had one that could not be parsed, and the loader treated that as "still running" and said nothing. It now returns a ReadError distinguishing an unsupported version from a malformed record. The version is actually checked. A record newer than SCHEMA_VERSION is reported as such rather than skipped. Adds testdata/schema-v1.eventlog, a frozen log replayed by CI on every build. It includes a record kind this build does not know and a stages payload carrying the multi-partition shape, so it exercises both forward-compatibility paths. Verified it fails as intended: a field rename in a stored type breaks all four compatibility tests. * feat(history): carry the job-list fields in JobIndex The index exists so the history server can render GET /api/jobs without parsing the stored payloads, but it was missing num_stages, completed_stages and percent_complete, which that response includes. It could not actually serve the list it was there for. Adds the three fields and updates the v1 fixture to match. Doing this before release, while the schema is still unpublished, so the frozen fixture stays a faithful record of what v1 actually looks like.

…2264) * refactor(scheduler): extract REST DTOs into ballista-history and factor out dto_build Move the scheduler's REST response types into a new leaf crate, `ballista-history`, and pull the graph-to-DTO construction out of the axum handlers into a pure `api::dto_build` module. Behavior preserving: the same DTOs are produced from the same state, so live REST responses are byte-identical. The existing handler tests cover this, and the helper unit tests move alongside the functions they test. This is the first step toward a history server that replays completed jobs and serves the same `/api/*` responses without a live scheduler. Splitting the DTOs into a serde-only crate lets that server build the identical wire types without depending on the scheduler's live execution graph, and moving construction out of the handlers means it can run against state that did not come from a handler request. `JobResponse::job_id` becomes a `String` rather than `ballista_core::JobId` so the new crate stays serde-only. `JobId` is `#[serde(transparent)]` over `String`, so the JSON is unchanged. * refactor(scheduler): tighten the DTO extraction Follow-up cleanups on the extraction: - Collapse the three near-identical ExecutionStage arms in graph_to_query_stages into one destructuring match, dropping the mutable placeholder-zero summary. - Take PlanFormat by value instead of &JobQueryParams, and move PlanFormat into ballista-history. It is part of the wire contract, and the pure builder no longer imports an axum query-param type back out of the handler module. - Inject `now` into graph_to_query_stages rather than reading the clock, so replaying a stored log renders stable elapsed times. - Share percent_complete and min_start_time; use displayable() instead of the longhand DisplayableExecutionPlan::new(). - Drop the dead JobConfig alias and the unused serde_json dev-dependency, make task_status_to_dto private, and remove a duplicated test. - Enable #![warn(missing_docs)] on ballista-history and document the types, matching the other Ballista crates. - Register ballista-history with the release tooling: version bump script, publish order, and crate dependency graph. * docs(history): state the write-once, replay-verbatim data flow The previous wording left it ambiguous whether the history server re-derives responses from stored execution state or replays stored DTOs. It replays them: the scheduler builds each response once against the live graph and writes it to the event log, so byte-identical output is a structural property rather than two implementations agreeing. Also records the consequence, that anything not captured at write time cannot be recovered at replay time. * refactor: rename ballista-history to ballista-api-types The crate holds the /api/* wire types, and it has three parties, not one: the scheduler serves them, the web TUI deserializes them, and a future history server will serve replayed copies. Naming it after the history server made it awkward for the TUI, which parses live scheduler responses and today keeps its own duplicate declarations. Renaming it after the contract it defines removes that friction. The event-log schema, writer, and reader can then land as a separate ballista-history crate that depends on this one. * feat(history): add the event-log schema, writer, and reader Second step toward the history server (#1923), after the wire-type extraction in #2256. Adds the durable format and the machinery to read and write it. Nothing in the scheduler calls this yet. - A versioned JSONL schema. JobStart / StageStart / StageEnd / TaskEnd form an incremental timeline; the terminal JobEnd embeds the finished REST responses, so replay re-serves what the scheduler already built rather than re-deriving it. - An async buffered EventLogWriter. All file I/O runs on a background task so the scheduler's event loop never waits on disk. Timeline events are dropped rather than allowed to block when the queue backs up, since losing a progress record beats stalling scheduling. JobEnd is the exception and waits for capacity, because a job missing it is invisible to the history server. - A reader that folds a completed log back into the served payload, skipping malformed lines rather than failing. Restores the JobConfig alias to ballista-api-types, which now has a real consumer in the JobEnd record. TaskEnd names a task rather than a partition: under the multi-partition task model a task owns a slice of partitions, and TaskStatus carries task_id, not partition_id. * feat(history): make the event log survive future Ballista versions A log is written once and may be read years later by a much newer binary, so the guarantee has to run one way: a reader accepts any log whose version is not newer than its own. That is the opposite of BALLISTA_PROTOCOL_VERSION, the strict-equality handshake between scheduler and executor, where both ends are live and upgraded together. The format did not support that yet. Four changes: Self-describing envelope. Every line is now a LogRecord carrying `ev`, `version` and an opaque `data` payload, so a reader can route on the kind and check the version before committing to a shape it may not understand. Previously only JobStart and JobEnd carried a version at all. Stored responses are opaque. JobEnd holds the finished /api/* responses as raw JSON rather than typed structs, plus a small frozen JobIndex for listing. Those responses are ballista-api-types shapes, which change with the live REST contract: partition_id went from u32 to Vec<u32> and TaskStatus::Failed gained a field within one release cycle. Stored typed, either change would have made every older log unreadable, and the job would have silently disappeared. Stored raw, nothing ever parses the inner shape and replay relays the exact bytes. Unreadable is no longer indistinguishable from absent. read_completed_job returned Ok(None) both when a log had no JobEnd and when it had one that could not be parsed, and the loader treated that as "still running" and said nothing. It now returns a ReadError distinguishing an unsupported version from a malformed record. The version is actually checked. A record newer than SCHEMA_VERSION is reported as such rather than skipped. Adds testdata/schema-v1.eventlog, a frozen log replayed by CI on every build. It includes a record kind this build does not know and a stages payload carrying the multi-partition shape, so it exercises both forward-compatibility paths. Verified it fails as intended: a field rename in a stored type breaks all four compatibility tests. * feat(scheduler): record a per-job event log behind --event-log-dir Third step toward the history server (#1923). Wires the event-log crate from #2260 into the scheduler. Nothing reads these logs yet; the history server that serves them is the next slice. - New --event-log-dir flag, off by default. When unset there is no channel, no background task, no file, and no per-event work beyond one Option check in the event loop. - event_log.rs builds HistoryEvents from execution-graph state, reusing the same api::dto_build builders that back the live REST API, so a job's stored record and its GET /api/job/{id} response are the same bytes for the same graph. - A tee at the top of QueryStageScheduler::on_receive maps JobSubmitted, TaskUpdating, JobFinished, JobRunningFailed and JobCancel onto history events. JobCancel is handled in the tee specifically because the handler below it drops the graph, making that the last point at which a cancelled job can be recorded. Its status is overridden on the typed DTO before anything is serialized, rather than by rewriting the stored JSON, so the payload stays a faithful serialization of one value and can be relayed verbatim. JobPlanningFailed is deliberately absent: it is posted instead of JobSubmitted, so the job has neither an execution graph nor an open log. Failure to build an event costs the job its record, never its execution. A missing graph or a serialization error is logged and skipped. The stage snapshot embedded in JobEnd is rendered as of completed_at rather than the wall clock, so replaying a log is deterministic. * feat(history): carry the job-list fields in JobIndex The index exists so the history server can render GET /api/jobs without parsing the stored payloads, but it was missing num_stages, completed_stages and percent_complete, which that response includes. It could not actually serve the list it was there for. Adds the three fields and updates the v1 fixture to match. Doing this before release, while the schema is still unpublished, so the frozen fixture stays a faithful record of what v1 actually looks like. * fix: populate the new JobIndex fields

andygrove marked this pull request as ready for review August 8, 2026 15:53

andygrove requested review from martin-g and milenkovicm August 8, 2026 16:02

github-actions Bot added the development-process label Aug 8, 2026

andygrove marked this pull request as draft August 8, 2026 16:34

andygrove marked this pull request as ready for review August 8, 2026 16:35

andygrove mentioned this pull request Aug 8, 2026

TUI's duplicated API response types have drifted from the scheduler, breaking the stages popup #2257

Closed

andygrove changed the title ~~refactor(scheduler): extract REST DTOs into ballista-history and factor out dto_build~~ refactor(scheduler): extract REST API wire types into ballista-api-types Aug 8, 2026

andygrove requested a review from avantgardnerio August 8, 2026 16:42

phillipleblanc approved these changes Aug 8, 2026

View reviewed changes

milenkovicm approved these changes Aug 9, 2026

View reviewed changes

andygrove merged commit 6610280 into apache:main Aug 9, 2026
27 checks passed

andygrove deleted the history-dto-extract branch August 9, 2026 12:56

This was referenced Aug 9, 2026

feat(scheduler): record a per-job event log behind --event-log-dir #2264

Merged

feat(scheduler): add the history server #2265

Open

Navigation Menu

Uh oh!

refactor(scheduler): extract REST API wire types into ballista-api-types - #2256

refactor(scheduler): extract REST API wire types into ballista-api-types#2256
andygrove merged 4 commits into
apache:mainfrom
andygrove:history-dto-extract

andygrove commented Aug 8, 2026 •

edited

Loading

Uh oh!

andygrove commented Aug 8, 2026

Uh oh!

phillipleblanc left a comment

Uh oh!

milenkovicm left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

andygrove commented Aug 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

Why ballista-api-types and not ballista-history

What changes are included in this PR?

Are there any user-facing changes?

Follow-ups

Open question for reviewers

Uh oh!

andygrove commented Aug 8, 2026

Uh oh!

phillipleblanc left a comment

Choose a reason for hiding this comment

Uh oh!

milenkovicm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

andygrove commented Aug 8, 2026 •

edited

Loading

Why `ballista-api-types` and not `ballista-history`