NoahKusaba commented on issue #2622: URL: https://github.com/apache/datafusion-ballista/issues/2622#issuecomment-6449725671
Thanks @milenkovicm, agreed. Here's a short design note. It replaces my earlier client-commit suggestion. ## Catalogs on the scheduler The scheduler can get catalogs through the existing `SchedulerConfig::with_override_session_builder`, with no core change. #2638 does this with `iceberg_session_builder(catalogs)`, which Flight SQL clients need anyway: they send SQL for the scheduler to plan, so they have no catalog of their own. Executors never need one. That leaves the question of where the commit runs. ## Options | | Trade-offs | |---|---| | **A. Commit on the client** | The client commits through its own catalog, but must stay connected until the job ends. Doesn't work for Flight SQL clients, which have no catalog. *Slowest: the file lists travel out of the cluster to the client before the commit.* | | **B. Final step on the scheduler** | Writers save their file lists as normal stage output. Once every writer is done, the scheduler fetches the lists from the executors, then commits. A small change, contained in the scheduler. *Faster than A: the file lists stop at the scheduler instead of continuing to the client.* | | **C. Spark-style commit messages** | Each writer puts its file list in its task-status message, so the scheduler already has every list when the last writer finishes and needs no fetch. But task status must carry output, which adds load to status messages and scheduler memory, needs a size cap, and changes executor, scheduler and proto. *Fastest: it skips B's final fetch from the executors, but the scheduler holds every writer's file list in memory until the job ends.* | In every option the commit runs once, so it needn't be idempotent. ## Proposal: B, marked by a plan node This is your plan-node suggestion, and it's how Trino does it: writers send their file lists through a normal exchange to a `TableFinish` node that runs on the coordinator. A marker node is explicit, shows up in `EXPLAIN`, and never leaves the scheduler, so it needs no serialization. - `ballista-iceberg` wraps `IcebergCommitExec` in the marker node, with an optimizer rule added through `iceberg_session_builder`. - The scheduler removes the marker before splitting stages. When the job succeeds, it runs the marker over the final stage's output and returns the result inline. C's inline transport could be added later without changing either. Open questions: 1. A generic marker node, or a narrower `JobFinalizer` trait with a capped result size? 2. An explicit `Finalizing` job state, instead of reporting the job as running until the commit completes? 3. Cancelling during the commit may still commit, as in Spark. Document it? If this works for you, I'll open a draft PR on top of #2638. -- 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]
