NoahKusaba opened a new issue, #2622:
URL: https://github.com/apache/datafusion-ballista/issues/2622
**Is your feature request related to a problem or challenge? Please describe
what you are trying to do.**
Table formats such as Iceberg and Delta finish a write with one commit.
Every writer task's data files are gathered in one place and committed through
the catalog, exactly once, after all writes have succeeded.
DataFusion represents this commit as an ordinary plan node at the root of
the INSERT plan. For Iceberg (#2217) that plan is:
```
IcebergCommitExec 1 partition: one catalog commit, outputs a
`count` row
CoalescePartitionsExec
IcebergWriteExec N partitions: writes Parquet files, outputs their
metadata
```
Ballista sends every plan node to an executor, so the commit runs as a
one-task final stage on some executor. That has three costs, all of which come
from where the commit runs rather than from the table format:
1. **Executors need the catalog.** The plan has to carry the catalog's
configuration, credentials included, so that an executor can rebuild the
catalog client and commit.
2. **Commits must be idempotent.** Ballista retries failed tasks, so a
commit task can run twice. #2217 has to give each commit an id and refuse to
run a commit without one.
3. **Every executor has to cache catalog clients,** with logic to evict and
retry stale ones.
Spark avoids all three because its write API ends on the driver. Writer
tasks send small commit messages back, and the driver calls
`BatchWrite.commit(messages)` once. Ballista has nowhere to run code on the
scheduler at the end of a job: `succeed_job` only saves the job's final state.
**Describe the solution you'd like**
A way for a job's final step to run on the scheduler, after all of its
stages have succeeded. Its small result becomes the job's result.
A minimal version, which I have a working prototype of:
- **A config hook to mark the final step:**
`SessionConfigExt::with_ballista_scheduler_root(predicate)`. When a job's root
operator matches, the scheduler removes it before splitting the plan into
stages, and the rest runs on executors as usual.
- **Running it when the job ends:** once the job succeeds, the scheduler
reads the final stage's output (here, the data files' metadata) with a
`ShuffleReaderExec` and runs the removed root on it.
- **Returning the result:** the root's output comes back to the client in a
new field, `SuccessfulJob.inline_result`, as Arrow IPC bytes. That's a few rows
for a commit.
For Iceberg this is a one-line opt-in that matches `IcebergCommitExec`. It
needs no change to datafusion-iceberg or to the plan.
Points for discussion:
- **Scope:** the hook could run any root operator, but it is only meant for
small final steps such as commits. Should the API be narrowed to that, for
example a `JobFinalizer` trait with a capped result size?
- **Job state:** the prototype reports a finished job as running until the
commit is done. An explicit `Finalizing` state in the job graph would be
cleaner.
- **Cancellation:** if a job is cancelled while its commit runs, the commit
may still go through. Spark has the same behaviour; here it should at least be
documented.
**Describe alternatives you've considered**
- **Keep committing on an executor** (what #2217 does). It works, but has
the three costs above.
- **A Spark-style commit protocol:** writer tasks return commit messages in
their task status, and a scheduler-side hook commits them. This avoids writing
the metadata to shuffle files and reading it back, but it needs a new
writer/committer API in Ballista core, and the write node would have to report
its output through task status.
- **Run the final stage on the scheduler** as an ordinary stage that the
scheduler executes itself. This is closer to how the job graph already works,
but it changes both graph implementations, and the scheduler would have to
serve that stage's output.
**Additional context**
- **Use case:** Iceberg writes in #2217, part of #2174. Any sink that ends
with a single commit would benefit, Delta Lake and Hudi included.
- **Prototype:** I'm happy to open it as a draft PR for discussion. It
covers both the adaptive and the static planner and both pull and push job
status. It includes a standalone test and a Docker test against a real Iceberg
REST catalog.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
