feat: add cluster settings to the Comet Native Executor binaries by andygrove · Pull Request #7420 · apache/datafusion-comet

andygrove · 2026-11-02T19:50:14Z

Which issue does this PR close?

Part of #7288.

Rationale for this change

comet-scheduler and comet-executor only expose the settings that a local, single-machine cluster needs. Running them on a real cluster, as the TPC-H runs in #7289 do, needs a few more of Ballista's settings: an external host for the scheduler, larger gRPC message limits for tasks with many scan partitions, a fixed executor memory pool, and the scheduler's per-query event logs for looking at stage timings. This splits those settings out of #7289. Each one sets a field that the Ballista 55.0.0 release already has, so this does not depend on any unmerged Ballista change.

What changes are included in this PR?

New environment variables, each documented in the binary's module docs:

COMET_SCHEDULER_EXTERNAL_HOST: the host executors reach the scheduler at.
COMET_GRPC_MAX_MESSAGE_SIZE: the largest gRPC message the scheduler and executor send or accept (Ballista's default is 16 MiB).
COMET_EXECUTOR_MEMORY_POOL_SIZE: the executor's memory budget, which Ballista splits into a pool per task.
COMET_SCHEDULER_EVENT_LOG_DIR: a directory for one Ballista event log per query. The scheduler doesn't start if it can't write there.

The event-log writer is behind ballista-scheduler's rest-api feature, so that feature is enabled. In 55.0.0 that feature also needs the tracing feature to compile without the default features, so that is enabled too (fixed upstream in apache/datafusion-ballista#2747).

Compiling rest-api also compiles Ballista's REST API, which would be served on the scheduler's gRPC port with no authentication, including a route that cancels jobs. The scheduler sets disable_rest_api: true, so the API is not served while the event-log writer still runs. apache/datafusion-ballista#2745 proposes compiling the writer without the API, and apache/datafusion-ballista#2746 proposes not serving the API on that port by default.

Ballista sees each stage's Comet block as one operator without metrics, so for Comet stages the logs' input rows count only the rows read from other stages, and a task's compute time covers only that reading. The scheduler's docs say so, and #7465 tracks reporting the block's metrics.

How are these changes tested?

Each variable is passed straight through to the matching Ballista config field. A new cluster test, writes_event_logs_without_serving_the_rest_api, runs a query with COMET_SCHEDULER_EVENT_LOG_DIR set, waits for a finished .eventlog, and checks that GET /api/jobs on the scheduler port returns 404 (it returns 200 without disable_rest_api). Another, refuses_an_event_log_dir_it_cannot_write, sets the variable to a path inside a regular file and checks that the scheduler exits with an error naming it. The existing native-executors cluster tests still pass, and clippy is clean with --all-targets -D warnings. The settings were used for the TPC-H runs in #7289.

Executors report task status to the host the scheduler advertises, which is localhost by default. On Kubernetes they need the scheduler's Service name, which COMET_SCHEDULER_EXTERNAL_HOST sets. (cherry picked from commit fdb4db0)

… limits A task of a stage that reads files carries each of its partitions' copy of the block, which at TPC-H SF1000 can exceed Ballista's 16 MiB gRPC limit, and the scheduler then loses the executor it can't send the task to. COMET_GRPC_MAX_MESSAGE_SIZE sets the limit for both binaries. (cherry picked from commit 2327720)

COMET_EXECUTOR_MEMORY_POOL_SIZE sets the bytes the executor's tasks share. Ballista's default, 70% of the container's limit, leaves too little for the memory outside the pool at TPC-H SF1000, where an executor was OOM-killed. (cherry picked from commit 8925114)

(cherry picked from commit e9d6e3d)

COMET_SCHEDULER_EVENT_LOG_DIR set the scheduler config's event_log_dir, but Ballista's EventLogWriter is built only with the scheduler's rest-api feature, which default-features = false had left out, so no event log was ever written. Enable rest-api (adds ballista-api-types, ballista-history and utoipa to the lockfile). (cherry picked from commit fc71795)

sunchao

I reviewed the cluster configuration against Ballista 55.0.0. The host and control-RPC settings reach the intended Ballista fields. Enabling the event writer also exposes Ballista’s REST API by default. I left inline comments on keeping those routes disabled, testing event-log output through the existing cluster harness, and documenting the limits of the new settings.

sunchao · 2026-11-02T20:01:26Z

-ballista-scheduler = { version = "55.0.0", default-features = false }
+# `rest-api` compiles the scheduler's event-log writer (COMET_SCHEDULER_EVENT_LOG_DIR) and its REST
+# API. In Ballista 55.0.0 it logs through `tracing` without enabling it, so that is enabled here.
+ballista-scheduler = { version = "55.0.0", default-features = false, features = ["rest-api", "tracing"] }


Could we keep the Ballista REST routes disabled in Comet’s SchedulerConfig? Enabling rest-api makes start_grpc_service mount /api/jobs, job plans, and PATCH /api/job/{id} on the executor-facing scheduler port. A network peer can enumerate and cancel jobs over HTTP without authentication, even with COMET_SCHEDULER_EVENT_LOG_DIR unset. Setting disable_rest_api: true keeps those routes off while event_log_dir still initializes the writer. Could we use Cluster::start_with to run a small query with logging enabled, wait for a finalized .eventlog, and assert that GET /api/jobs returns 404 on the scheduler port? Route setup · REST endpoints · Writer setup

Good catch, thanks. The scheduler now sets disable_rest_api: true, so rest-api only brings in the event-log writer. I added writes_event_logs_without_serving_the_rest_api to tests/cluster.rs. It starts a cluster through Cluster::start_with with COMET_SCHEDULER_EVENT_LOG_DIR set, runs a query, waits for a finished .eventlog, and asserts that GET /api/jobs on the scheduler port returns 404. With disable_rest_api: false the same test fails with a 200. Fixed in 94010d2.

sunchao · 2026-11-02T20:01:26Z

 //! - `COMET_SCHEDULER_RESULT_RETENTION_SECONDS`: how long a query's results stay readable after
 //!   the query ends, whether or not a driver has read them (default `300`). `0` keeps them.
+//! - `COMET_GRPC_MAX_MESSAGE_SIZE`: the largest gRPC message, in bytes, the scheduler sends or
+//!   accepts, such as a task with many partitions of a scan (default: Ballista's 16 MiB)


Could the docs scope this to the Ballista RPC endpoints and tell users to set it on both scheduler and executors for large task messages? Comet’s separate Spark-facing Flight listener still sets both limits to usize::MAX below. The executor Flight server picks up this setting, while remote shuffle Flight clients keep the separate 16 MiB session default. Raising the variable on every process therefore raises the task RPC limit but can still leave a large shuffle response above the receiving client’s limit. Shuffle client configuration

Done in 94010d2. The scheduler's doc now covers only Ballista's scheduler-executor gRPC service and says to set the variable on the executors too. It notes that the drivers' Flight service has no limit, and that shuffle reads between executors keep ballista.client.grpc_max_message_size (16 MiB by default), which this variable doesn't change. The executor's doc lists where it does apply: the connections with the scheduler and its Flight service.

sunchao · 2026-11-02T20:01:26Z

+//! - `COMET_GRPC_MAX_MESSAGE_SIZE`: the largest gRPC message, in bytes, the executor accepts or
+//!   sends, such as a task with many partitions of a scan (default: Ballista's 16 MiB)
+//! - `COMET_EXECUTOR_MEMORY_POOL_SIZE`: the bytes of memory the executor's tasks share (default:
+//!   Ballista's, 70% of the container's memory limit). Memory outside the pool, such as batches


Could this describe a budget split into per-task pools and document that 0 selects an unbounded pool? Ballista divides the configured bytes by the executor’s vcores, then gives each task a separate pool proportional to the vcores it claims. A one-vcore task on an eight-vcore executor gets one eighth of the budget even if it’s the only task running. When unset, Ballista uses 70% of the smaller detected cgroup limit or host memory, with an unbounded fallback if neither is detected. Budget resolution · Per-task pool policy

Done in 94010d2. The doc now calls it a budget, says 0 means no limit, and explains that each task gets its own pool of the budget divided by COMET_EXECUTOR_TASKS, times the slots it takes, even when it runs alone. It also gives the default: 70% of the smaller of the cgroup limit and host memory, or no limit if neither is detected.

sunchao · 2026-11-02T20:01:26Z

+//!   accepts, such as a task with many partitions of a scan (default: Ballista's 16 MiB)
+//! - `COMET_SCHEDULER_EVENT_LOG_DIR`: a directory to write one event log per query to, as
+//!   Ballista's scheduler does: the query's plan, a record for each task, and each stage's rows
+//!   and compute time (default: no event logs)


Could the docs describe the per-task timeline as best effort? Ballista drops timeline records when the writer queue fills, so a completed job can have fewer records than tasks. It would also help to say that event-log files remain in this directory until removed. The result-retention setting above governs executor result data. Event-log writer

Done in 94010d2. The doc now says the task records are best effort and may be dropped when the writer falls behind, so a log can have fewer records than the query had tasks. It also says logs stay in the directory until something removes them, and that the result-retention setting applies to query results, not to logs.

…e settings' limits Compiling the event-log writer through ballista-scheduler's rest-api feature also mounts Ballista's REST API on the scheduler's gRPC port, which lists and cancels jobs without authentication. The scheduler now sets disable_rest_api, which keeps the routes off while the event-log writer still runs. A cluster test writes an event log and checks that /api/jobs is not served. The docs now say that COMET_GRPC_MAX_MESSAGE_SIZE covers Ballista's scheduler-executor RPCs and the executor's Flight service but not shuffle reads, that the executor's memory budget is split into per-task pools and 0 means no limit, and that event-log task records are best effort and logs are never removed.

comphead

Thanks for splitting these settings out of #7289. Reading Ballista 55.0.0, the host, the gRPC limits on both sides, the memory pool and disable_rest_api all reach the intended fields, and disable_rest_api covers every method under /api/*. Two comments inline.

comphead · 2026-11-03T20:32:18Z

+//! - `COMET_SCHEDULER_EVENT_LOG_DIR`: a directory to write one event log per query to, as
+//!   Ballista's scheduler does: the query's plan, a record for each task, and each stage's rows


Do the stage rows and compute time hold for Comet stages? In Ballista 55.0.0, task_row_counts (scheduler/src/api/dto_build.rs) sums elapsed_compute over the operators that report metrics, and takes rows from the output_rows of the root and the leaves. CometFragmentExec has no metrics(), ShuffleWriterExec reports no elapsed_compute of its own, and the leaves of a Comet stage are the one-row-per-partition descriptor and the shuffle readers. So I expect compute time near zero and input rows counting descriptor rows. I have not run it. Could you check elapsed_compute_nanos and input_rows in the TaskEnd records the new test writes? If they are off, the doc could name what the log does give for Comet stages, or CometFragmentExec could report baseline metrics in a follow-up.

I ran the cluster test with a two-stage query (a range stage hash-shuffled into a stage that reads it) and read the TaskEnd records:

Stage input_rows output_rows elapsed_compute_nanos

Range stage, reads no other stage 0 1000 0

Reads the range stage's shuffle 1000 1000 3328833

Compute time is as you expected: 0 for a stage with no shuffle input, and otherwise only the ShuffleReaderExec's poll time. Input rows don't count the descriptor, whose DataSourceExec reports no output_rows, but they count only rows read from other stages, so a stage that reads files or makes its own rows logs 0. Output rows and the task times are right, as is the stage summary's elapsed_compute, which runs from the first task's start to the last task's end.

In dd77ce7 the doc says this: Ballista sees a Comet block as one operator with no metrics, so the plan shows a block only by its size, input rows count only rows read from other stages, and a task's compute time covers only that reading. I opened #7465 for CometFragmentExec to report its block's metrics, which #7373 needs for the Spark UI as well.

comphead · 2026-11-03T20:32:18Z

+    if let Some(dir) = env_var::<String>("COMET_SCHEDULER_EVENT_LOG_DIR")? {
+        config.event_log_dir = Some(dir);
+    }


Ballista creates event_log_dir on a background task, and when that fails, run in ballista/history/src/writer.rs logs a warning and returns. The scheduler then starts normally and every later event logs dropping event ... (channel full or closed). So an unwritable directory gives no logs and no startup error, while the other malformed settings fail fast through env_var. Could the binary call std::fs::create_dir_all on it here and return the error?

Done in dd77ce7. The scheduler now calls create_dir_all and returns the error, naming COMET_SCHEDULER_EVENT_LOG_DIR. create_dir_all succeeds on a directory that exists but isn't writable, such as a volume mounted with another owner, and Ballista then only warns for each job, so the scheduler also creates and removes a file there. refuses_an_event_log_dir_it_cannot_write sets the variable to a path inside a regular file and checks that the scheduler exits with an error naming the variable. That fails for root too, which CI's container runs as, and the test fails without the check. By hand, an existing read-only directory gives Permission denied (os error 13).

…show for Comet stages comet-scheduler now creates COMET_SCHEDULER_EVENT_LOG_DIR and checks that it can create files there before it starts. Ballista creates the directory on a background task and, when it can't write there, only logs warnings, so a bad directory gave no logs and no startup error. Creating the directory alone doesn't catch one that exists but isn't writable, such as a volume mounted with another owner, so the check also creates and removes a file. The module docs now say what the logs record for Comet stages: Ballista sees a Comet block as one operator without metrics, so input rows count only the rows read from other stages, and a task's compute time covers only that reading.

coderfender · 2026-11-05T08:27:41Z

+/// directory that already exists, such as a volume mounted with another owner, may not allow it.
+fn check_writable(dir: &Path) -> std::io::Result<()> {
+    std::fs::create_dir_all(dir)?;
+    let probe = dir.join(format!(".comet-scheduler-{}", std::process::id()));


In a container the scheduler is usually PID 1, so two schedulers sharing a log volume (an HA pair, or an overlapping restart) use the same probe name. If A creates it, B creates it again and A removes it, B's remove_file fails with NotFound and B refuses to start. Could the name be unique (e.g. a random suffix), or NotFound on removal be ignored?

Good point. The probe name now has the time in nanoseconds as well as the PID, so two schedulers sharing a log volume no longer collide. Fixed in 10defc3.

coderfender · 2026-11-05T08:27:41Z


+/// The HTTP status of a GET of `path` on a local port.
+fn http_status(port: u16, path: &str) -> u16 {
+    let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();


Nit: without a read timeout, read_to_end blocks if the server ever keeps the connection open, so the test would hang instead of failing. stream.set_read_timeout(Some(Duration::from_secs(10))) would avoid that.

Done in 10defc3: http_status sets a 10 second read timeout.

comphead · 2026-11-06T16:00:23Z

+//! - `COMET_SCHEDULER_EVENT_LOG_DIR`: a directory to write one event log per query to, as
+//!   Ballista's scheduler does (default: no event logs). The scheduler creates it if needed and
+//!   doesn't start if it can't write there. A log has the query's plan and a record for each
+//!   stage and task, with the task's times and the rows it wrote. Ballista sees a stage's Comet


do we need to mention ballista ?

No. The doc now says "the log" sees a stage's Comet block as one operator, without naming Ballista. Done in 10defc3.

comphead · 2026-11-06T16:01:31Z

+    if let Some(host) = env_var("COMET_SCHEDULER_EXTERNAL_HOST")? {
+        config.external_host = host;
+    }
+    if let Some(size) = env_var("COMET_GRPC_MAX_MESSAGE_SIZE")? {


the similar code above sets only 2 params, and 3 here, is it expected?

Yes. The scheduler is a gRPC server for the executors and also a client of theirs, and Ballista has a separate grpc_client_max_message_size for that client. An executor has one pair of settings, which covers both its server and its connection to the scheduler. I added a comment saying so in 10defc3.

comphead

looks good, thanks @andygrove some nits

…, and explain the gRPC settings

andygrove added 5 commits November 2, 2026 12:42

feat: write the scheduler's event logs to COMET_SCHEDULER_EVENT_LOG_DIR

55f4733

(cherry picked from commit e9d6e3d)

github-actions Bot added the enhancement New feature or request label Nov 2, 2026

andygrove requested a review from sunchao November 2, 2026 19:55

sunchao reviewed Nov 2, 2026

View reviewed changes

comphead reviewed Nov 3, 2026

View reviewed changes

This was referenced Nov 3, 2026

Comet Native Executors: report each stage's Comet block metrics to Ballista #7465

Open

Scheduler runs without event logs, and only warns, when it can't write event_log_dir apache/datafusion-ballista#2766

Closed

coderfender reviewed Nov 5, 2026

View reviewed changes

comphead reviewed Nov 6, 2026

View reviewed changes

comphead approved these changes Nov 6, 2026

View reviewed changes

fix: make the event-log probe name unique, bound the test's HTTP read…

10defc3

…, and explain the gRPC settings

andygrove enabled auto-merge November 6, 2026 16:12

andygrove added this pull request to the merge queue Nov 6, 2026

Merged via the queue into apache:main with commit eaffbff Nov 6, 2026
26 checks passed

		//! - `COMET_SCHEDULER_EVENT_LOG_DIR`: a directory to write one event log per query to, as
		//! Ballista's scheduler does: the query's plan, a record for each task, and each stage's rows

Stage	`input_rows`	`output_rows`	`elapsed_compute_nanos`
Range stage, reads no other stage	0	1000	0
Reads the range stage's shuffle	1000	1000	3328833

Conversation

andygrove commented Nov 2, 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?

How are these changes tested?

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

andygrove commented Nov 2, 2026 •

edited

Loading