feat(scheduler): add the history server by andygrove · Pull Request #2265 · apache/datafusion-ballista

andygrove · 2026-08-09T13:59:02Z

Which issue does this PR close?

Closes #1923. Last of four PRs splitting up #1925, after #2256 (merged), #2260 and #2264.

Stacked on #2264. GitHub shows the whole chain until the earlier ones merge, so review the last three commits here (feat(scheduler): add the history server, refactor(history): index event logs instead of loading them into memory and fix(history): list completed jobs newest first).

Rationale for this change

#2264 makes the scheduler write durable per-job logs, but nothing reads them. This serves them.

The scheduler forgets a job shortly after it finishes: completed jobs are cleaned up after finished_job_state_clean_up_interval_seconds, and everything is gone on restart. With this, the TUI can point at a history server and browse completed jobs with no scheduler running at all.

user-personas.md lists "a history server / UI" among the things Persona 2 depends on, so this fills in a guarantee already written down.

What changes are included in this PR?

ballista-history-server --event-log-dir <dir> indexes every completed log at startup and serves the same /api/* paths as the live scheduler: jobs, job/{id}, job/{id}/stages, job/{id}/config, job/{id}/dot.

Corrupt or unreadable logs are logged and skipped rather than failing startup, so one file truncated by a crash cannot hide every other job. /api/executors returns empty and /api/state a static payload, since there is no cluster behind a history server and the TUI calls both at startup.

A user-guide page covering both flags, the endpoints, and the operational caveats.

What is kept in memory

Only each job's JobIndex, which is exactly what GET /api/jobs reports. Everything else is read back out of the job's log per request, inside spawn_blocking.

The alternative, holding the loaded payload for every job, means keeping both plan-bearing REST responses, the session config and the DOT graph resident. A job with many tasks stores megabytes of per-task detail, so a directory retained for any length of time would put the server's memory use at the mercy of the retention policy, for data almost none of which anyone will open. Request rate on the detail endpoints is whatever a person clicks through a UI, so a read per request is the cheaper side of that trade.

Startup still scans the directory once, since a job's metadata only exists inside its log, but it decodes only the index field out of the JobEnd record rather than materialising payloads it would immediately drop.

Two consequences, both covered by tests:

Corruption confined to the payloads is not seen at index time. It surfaces as a 500 naming the failing log when that job is opened.
A log that loses its terminal record after indexing (rotated, truncated) returns 404.

Job list ordering

Newest first, by start time, with the job id as a tiebreaker so the order is total. A job id is a random 7-character string (TaskManager::generate_job_id), so ordering by it would have been arbitrary. This also matches what the TUI does with the list once it has it, and is the order a future ?limit= would want to truncate.

There is no paging yet. GET /api/jobs returns every job in one response, which is fine for the job counts a single directory realistically holds but will not stay fine forever. Adding it here alone would leave the history server and the live scheduler with different /api/jobs contracts, and the TUI talks to both, so it belongs in a change that covers both endpoints. Tracked in #2270, and noted as a limitation on the user-guide page in the meantime.

How the payloads are served

The stored /api/job/{id} and /api/job/{id}/stages payloads are relayed as raw JSON rather than deserialized and re-serialized. Clients get the exact bytes the live scheduler produced, and a later change to the REST types cannot make an existing log unservable, which is the compatibility property #2260 is built around.

The job list is the one endpoint that needs structure, since it omits the plan fields. It is rebuilt from the frozen JobIndex rather than by editing a stored payload, so that path never parses a response it would mostly discard, and it is also the reason the list endpoint never touches disk.

The end-to-end guarantee

history_store_serves_byte_identical_json_to_live_scheduler was held out of #2264 because it needs HistoryStore. It lands here. It emits a real JobEnd through the real async writer, loads it back through HistoryStore::load, and compares the served bytes against what the live builders produce for the same graph.

Storing raw JSON made it a stronger test than it was in #1925: it is now a literal byte comparison rather than a structural one.

Are there any user-facing changes?

Yes, all additive:

New ballista-history-server binary.
The TUI can point at it with BALLISTA__SCHEDULER__URL to browse completed jobs with no live scheduler. Its --host / --port flags set the gRPC address queries run against and do not move the TUI, so the user-guide page documents the environment variable instead.

No API changes, and nothing about a cluster that has not enabled event logging.

Verified locally: cargo test -p ballista-scheduler --lib passes (342 tests) and cargo test -p ballista-history passes (18 tests), clippy is clean with --all-features -D warnings, fmt, taplo and prettier are clean, and the --no-default-features check still passes.

I also ran the binary against the checked-in schema-v1.eventlog fixture and curled every endpoint. It indexes the job, serves the list, the job, the stages (relaying the multi-partition partition_id array verbatim) and the config, and 404s on an unknown job.

Known limitations

Documented on the user-guide page rather than left to be discovered:

Local filesystem only, and the directory is scanned once at startup. Restart to pick up newly finished jobs.
Disk is not reclaimed automatically. Logs accumulate until pruned.
Plans are rendered once, at write time, so ?plan_format= has no effect against a history server.
Only the terminal record is served. The per-task timeline is captured so a future UI can show a job progressing, but nothing reads it yet.
GET /api/jobs is unpaged. Tracked in Add paging to GET /api/jobs on the scheduler and history server #2270.

Now that nothing is cached, dropping the first of those is mostly a matter of rescanning the directory, but that is out of scope here.

Verified against a live cluster

The two halves have now been exercised together, which earlier revisions of this description listed as outstanding.

A scheduler started with --event-log-dir and one executor ran a TPC-H join and aggregate over Parquet, producing a real event log. A history server pointed at that directory indexed the job and served all five endpoints, and /api/job/{unknown} returned 404.

Diffing the history server's responses against the still-running scheduler's, /api/jobs, /api/job/{id}, /api/job/{id}/stages and /api/job/{id}/dot are byte-identical, which is the property the raw-JSON relay is meant to give.

/api/job/{id}/config is the one endpoint whose bytes differ, and the cause is on the live side: it serializes a HashMap, so two consecutive requests to the scheduler itself disagree on key order. The parsed maps compare equal, and the history server's copy is the stable one because it is a frozen snapshot.

Driving the TUI against the history server is how the --host / --port documentation error above surfaced. With BALLISTA__SCHEDULER__URL set, the header shows the history server's address and the job list renders from it.

Two things this did not cover:

The per-job TUI screens. Selecting a row could not be driven from a scripted terminal, so those screens are unverified, though every endpoint behind them is checked above.
The static /api/state payload sets started to 0, which the TUI header renders as 1969-12-31. Cosmetic, and left as is here.

…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.

Second step toward the history server (apache#1923), after the wire-type extraction in apache#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.

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.

# Conflicts: # Cargo.toml # ballista/api-types/src/dto.rs # dev/release/README.md # dev/release/crate-deps.dot # dev/update_ballista_versions.py

Third step toward the history server (apache#1923). Wires the event-log crate from apache#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.

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.

Final step of apache#1923, after apache#2256, apache#2260 and apache#2264. Serves completed jobs from stored event logs, so the existing TUI can browse them with no scheduler running. - ballista-history-server --event-log-dir <dir> loads every completed log at startup and serves the same /api/* paths as the live scheduler. - Corrupt or unreadable logs are logged and skipped rather than failing startup, so one bad file cannot hide every other job. - /api/executors returns empty and /api/state a static payload, since there is no cluster behind a history server, so TUI screens expecting them still load. - A user-guide page covering both flags, the endpoints, and the operational caveats. The stored /api/job/{id} and /api/job/{id}/stages payloads are relayed as raw JSON rather than deserialized and re-serialized, so clients receive the exact bytes the live scheduler produced and a later change to the REST types cannot make an existing log unservable. The job list is rebuilt from the frozen JobIndex instead, which carries exactly the fields that endpoint includes. Brings back history_store_serves_byte_identical_json_to_live_scheduler, held out of apache#2264 because it needs HistoryStore. It emits a real JobEnd through the real async writer, loads it back through HistoryStore::load, and compares the served bytes against what the live builders produce for the same graph. Now a literal byte comparison rather than a structural one.

# Conflicts: # ballista/scheduler/Cargo.toml # ballista/scheduler/src/scheduler_server/event_log.rs

milenkovicm

do we need to load all jobs to memory instead of reading them from FS when they are needed ?

milenkovicm · 2026-08-09T21:04:01Z

+#[derive(Default)]
+pub struct HistoryStore {
+    /// Completed jobs keyed by job id.
+    pub jobs: HashMap<String, ReplayedJob>,


why do we want to keep them all in the memory? cant we just read file on request

Thanks, this wasn't intended ... fixed now

milenkovicm · 2026-08-09T21:05:27Z

+    /// crash mid-write) is logged and skipped rather than failing the whole
+    /// load — one bad log must not hide every other completed job. Only a
+    /// failure to read the directory itself is propagated.
+    pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {


same like previous, why do we need to load it all instead of reading it on request ?

HistoryStore kept a full ReplayedJob per log: both plan-bearing REST payloads, the session config and the DOT graph. For a job with many tasks that runs to megabytes, held resident for every job in the directory whether or not anyone ever opens it. Keep only the JobIndex and the file path, which is all GET /api/jobs needs, and read a job's payload back from its log per request inside spawn_blocking. Startup still scans the directory once, but decodes only the summary out of each JobEnd rather than materialising payloads it would immediately drop. Corruption confined to the payloads is no longer caught at startup, so it surfaces as a 500 naming the failing log. A log that loses its terminal record after indexing returns 404.

milenkovicm

A few more comments

milenkovicm · 2026-08-09T22:05:38Z

+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
+    let mut jobs: Vec<JobResponse> =
+        store.jobs.values().map(|j| list_entry(&j.index)).collect();
+    jobs.sort_by(|a, b| a.job_id.cmp(&b.job_id));


Should jobs be sorted on time rather than name (if job ids are not monotonically increasing)

Good catch. Fixed.

Job ids are random 7-character strings, so ordering /api/jobs by id put the list in an order that means nothing. Sort by start time descending, with the id as a tiebreaker so two jobs that started in the same millisecond cannot swap places between requests. This is also the order the TUI puts the list into once it has it, and the order a future ?limit= would want to truncate.

milenkovicm

one question regarding visibility of new events after history server starts, not sure if i'm missing something but to me it looks job entries are read at (history server) startup

can we handle situation where we have one history server running and we have one or more scheduler servers producing events (to a directory which history server reads), will new events be visible to history server (after history server startup)

milenkovicm · 2026-08-10T12:48:49Z

+    /// costs a pass over the directory rather than a copy of it in memory.
+    /// Corruption confined to the payloads therefore surfaces when the job is
+    /// requested rather than at startup.
+    pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {


will this be called ONLY when history server starts?

can we handle situation where we have one history server running and we have one or more scheduler servers producing events, will new events be visible to history server (after history server startup)

milenkovicm · 2026-08-10T19:53:01Z

simple test scenario

i have a running server which writes events to predefined directory,
i have a history server which reads same predefined directory
run a job
new event file gets written to predefined directory

number of jobs returned by history server does not change, new job not detected

milenkovicm · 2026-08-10T20:26:24Z

#2269 landed, it generates monotonically increasing ids, perhaps it could be used to avoid loading and caching data from a fs

andygrove · 2026-08-11T01:38:03Z

Sorry for the quality of this PR @milenkovicm and thanks for reviews so far. Moving to draft until this is more complete.

milenkovicm · 2026-08-11T06:21:56Z

Not a bother at all, this is good functionality to have, happy to help

milenkovicm · 2026-08-11T06:38:05Z

I'll use this comments for follow up ideas, we can filter them later

Https support
authentication
web TUI integration

The TUI reads its scheduler URL from configuration, not from --host and --port, so the documented command silently browsed the live scheduler on the default port instead of the history server.

milenkovicm · 2026-08-16T08:05:01Z

+the TUI on its default of `http://localhost:50050`, where it either finds your
+live scheduler or reports that the scheduler is down.
+
+The directory is scanned once at startup, so restart the history server to pick


Do you want to history sever to serve only static list of jobs discovered at startup @andygrove ?

This was my main comment last time, I think it would make sense to discover new events automatically during execution rather than on startup.

Anyway your call, please let me know what you think

andygrove added 12 commits August 8, 2026 09:52

Merge remote-tracking branch 'apache/main' into history-event-crate

19f2d81

# Conflicts: # Cargo.toml # ballista/api-types/src/dto.rs # dev/release/README.md # dev/release/crate-deps.dot # dev/update_ballista_versions.py

Merge branch 'history-event-crate' into history-scheduler-wiring

a9f5277

fix: populate the new JobIndex fields

7c15a33

github-actions Bot added documentation Improvements or additions to documentation development-process labels Aug 9, 2026

andygrove marked this pull request as draft August 9, 2026 14:29

andygrove mentioned this pull request Aug 9, 2026

feat: event log and history server (Spark History Server equivalent) #1925

Closed

Merge remote-tracking branch 'apache/main' into history-server-binary

634a199

# Conflicts: # ballista/scheduler/Cargo.toml # ballista/scheduler/src/scheduler_server/event_log.rs

andygrove marked this pull request as ready for review August 9, 2026 20:29

github-actions Bot removed the development-process label Aug 9, 2026

andygrove requested a review from milenkovicm August 9, 2026 20:37

milenkovicm reviewed Aug 9, 2026

View reviewed changes

andygrove mentioned this pull request Aug 9, 2026

Add paging to GET /api/jobs on the scheduler and history server #2270

Open

docs(history): link the paging limitation to apache#2270

02cd203

andygrove requested a review from milenkovicm August 10, 2026 12:40

milenkovicm reviewed Aug 10, 2026

View reviewed changes

andygrove marked this pull request as draft August 11, 2026 01:38

docs(history): correct the TUI invocation against a history server

b30a6cf

The TUI reads its scheduler URL from configuration, not from --host and --port, so the documented command silently browsed the live scheduler on the default port instead of the history server.

andygrove marked this pull request as ready for review August 15, 2026 16:00

andygrove mentioned this pull request Aug 15, 2026

DOT graph renders SortMergeJoinExec and SortPreservingMergeExec as "Unknown Operator" #2303

Closed

milenkovicm reviewed Aug 16, 2026

View reviewed changes

Uh oh!

Conversation

andygrove commented Aug 9, 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

What changes are included in this PR?

What is kept in memory

Job list ordering

How the payloads are served

The end-to-end guarantee

Are there any user-facing changes?

Known limitations

Verified against a live cluster

Uh oh!

milenkovicm left a comment

Choose a reason for hiding this comment

Uh oh!

milenkovicm Aug 9, 2026

Choose a reason for hiding this comment

Uh oh!

andygrove Aug 9, 2026

Choose a reason for hiding this comment

Uh oh!

milenkovicm Aug 9, 2026

Choose a reason for hiding this comment

Uh oh!

milenkovicm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

milenkovicm Aug 9, 2026

Choose a reason for hiding this comment

Uh oh!

andygrove Aug 9, 2026

Choose a reason for hiding this comment

Uh oh!

milenkovicm left a comment

Choose a reason for hiding this comment

Uh oh!

milenkovicm Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

milenkovicm commented Aug 10, 2026

Uh oh!

milenkovicm commented Aug 10, 2026

Uh oh!

andygrove commented Aug 11, 2026

Uh oh!

milenkovicm commented Aug 11, 2026

Uh oh!

milenkovicm commented Aug 11, 2026

Uh oh!

milenkovicm Aug 16, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

andygrove commented Aug 9, 2026 •

edited

Loading