From 4d5814fed96e373a939625b68f087c2ce9737c9c Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 31 Oct 2026 14:38:47 -0600 Subject: [PATCH 1/8] feat: let comet-scheduler advertise an external host to its executors 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 fdb4db09b6be672ad382ccc1f2ead9524933c59c) --- native/native-executors/src/bin/comet-scheduler.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index d1506f335ac..3c812b101a3 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -20,6 +20,8 @@ //! //! Configured through environment variables, with Ballista's defaults otherwise: //! - `COMET_SCHEDULER_BIND_HOST` (default `127.0.0.1`) +//! - `COMET_SCHEDULER_EXTERNAL_HOST`: the host executors reach this scheduler at, which they +//! report task status to (default `localhost`) //! - `COMET_SCHEDULER_PORT`: Ballista's gRPC port, for executors (default `50050`) //! - `COMET_SCHEDULER_FLIGHT_PORT`: the Flight port, for drivers (default `50060`) //! - `COMET_SCHEDULER_RESULT_RETENTION_SECONDS`: how long a query's results stay readable after @@ -48,6 +50,9 @@ async fn main() -> Result<(), Box> { if let Some(host) = env_var("COMET_SCHEDULER_BIND_HOST")? { config.bind_host = host; } + if let Some(host) = env_var("COMET_SCHEDULER_EXTERNAL_HOST")? { + config.external_host = host; + } if let Some(port) = env_var("COMET_SCHEDULER_PORT")? { config.bind_port = port; } From 1d8cf07de6d968679656a99b7171084cabbd0c4e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 31 Oct 2026 15:06:06 -0600 Subject: [PATCH 2/8] feat: let the Comet Native Executor binaries raise their gRPC message 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 23277200eaac95174aca6e428758a30efb408a5d) --- native/native-executors/src/bin/comet-executor.rs | 6 ++++++ native/native-executors/src/bin/comet-scheduler.rs | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/native/native-executors/src/bin/comet-executor.rs b/native/native-executors/src/bin/comet-executor.rs index f1b7cec6a99..2e23ef8dfa1 100644 --- a/native/native-executors/src/bin/comet-executor.rs +++ b/native/native-executors/src/bin/comet-executor.rs @@ -26,6 +26,8 @@ //! - `COMET_EXECUTOR_GRPC_PORT` (default `50052`) //! - `COMET_SCHEDULER_HOST` (default `localhost`) and `COMET_SCHEDULER_PORT` (default `50050`) //! - `COMET_EXECUTOR_TASKS`: concurrent tasks (default: the number of CPUs) +//! - `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_WORK_DIR`: where shuffle and spill files go (default: a temporary directory) use std::sync::Arc; @@ -61,6 +63,10 @@ async fn main() -> Result<(), Box> { config.vcores = tasks; } config.work_dir = env_var("COMET_EXECUTOR_WORK_DIR")?; + if let Some(size) = env_var("COMET_GRPC_MAX_MESSAGE_SIZE")? { + config.grpc_max_decoding_message_size = size; + config.grpc_max_encoding_message_size = size; + } start_executor_process(Arc::new(config)).await?; Ok(()) diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index 3c812b101a3..53cc90ee221 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -26,6 +26,8 @@ //! - `COMET_SCHEDULER_FLIGHT_PORT`: the Flight port, for drivers (default `50060`) //! - `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) use std::net::SocketAddr; use std::sync::Arc; @@ -53,6 +55,11 @@ async fn main() -> Result<(), Box> { 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")? { + config.grpc_server_max_decoding_message_size = size; + config.grpc_server_max_encoding_message_size = size; + config.grpc_client_max_message_size = size; + } if let Some(port) = env_var("COMET_SCHEDULER_PORT")? { config.bind_port = port; } From 8aef1c92415308cc4545ff56291d239c828506b2 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 31 Oct 2026 16:19:26 -0600 Subject: [PATCH 3/8] feat: let comet-executor set the size of its memory pool 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 89251149c0f27ec110afc39f39b46d1add66c844) --- native/native-executors/src/bin/comet-executor.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/native/native-executors/src/bin/comet-executor.rs b/native/native-executors/src/bin/comet-executor.rs index 2e23ef8dfa1..9a08a7c62df 100644 --- a/native/native-executors/src/bin/comet-executor.rs +++ b/native/native-executors/src/bin/comet-executor.rs @@ -28,6 +28,9 @@ //! - `COMET_EXECUTOR_TASKS`: concurrent tasks (default: the number of CPUs) //! - `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 +//! in flight between stages, comes on top of it. //! - `COMET_EXECUTOR_WORK_DIR`: where shuffle and spill files go (default: a temporary directory) use std::sync::Arc; @@ -63,6 +66,9 @@ async fn main() -> Result<(), Box> { config.vcores = tasks; } config.work_dir = env_var("COMET_EXECUTOR_WORK_DIR")?; + if let Some(size) = env_var("COMET_EXECUTOR_MEMORY_POOL_SIZE")? { + config.memory_pool_size = Some(size); + } if let Some(size) = env_var("COMET_GRPC_MAX_MESSAGE_SIZE")? { config.grpc_max_decoding_message_size = size; config.grpc_max_encoding_message_size = size; From 55f47335e94dd0da94ad7bc67fd1000e31e9e20b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 1 Nov 2026 12:43:53 -0700 Subject: [PATCH 4/8] feat: write the scheduler's event logs to COMET_SCHEDULER_EVENT_LOG_DIR (cherry picked from commit e9d6e3da06a2244a84590e56f9b204a74576d022) --- native/native-executors/src/bin/comet-scheduler.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index 53cc90ee221..1d8f73f887a 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -28,6 +28,9 @@ //! 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) +//! - `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) use std::net::SocketAddr; use std::sync::Arc; @@ -66,6 +69,9 @@ async fn main() -> Result<(), Box> { if let Some(seconds) = env_var("COMET_SCHEDULER_RESULT_RETENTION_SECONDS")? { config.finished_job_data_clean_up_interval_seconds = seconds; } + if let Some(dir) = env_var::("COMET_SCHEDULER_EVENT_LOG_DIR")? { + config.event_log_dir = Some(dir); + } let flight_port: u16 = env_var("COMET_SCHEDULER_FLIGHT_PORT")?.unwrap_or(50060); let grpc_addr: SocketAddr = format!("{}:{}", config.bind_host, config.bind_port).parse()?; let flight_addr = SocketAddr::new(grpc_addr.ip(), flight_port); From bf08350f5d0a5ae955cef77e62681db657cd81b1 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 2 Nov 2026 12:12:01 -0700 Subject: [PATCH 5/8] fix: compile Ballista's event-log writer into comet-scheduler 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 fc7179509229fbd5bf73e1d057d61cee077b7260) --- native/native-executors/Cargo.lock | 58 ++++++++++++++++++++++++++++-- native/native-executors/Cargo.toml | 4 ++- 2 files changed, 58 insertions(+), 4 deletions(-) diff --git a/native/native-executors/Cargo.lock b/native/native-executors/Cargo.lock index 4d21ac8a3e5..452fe6d373c 100644 --- a/native/native-executors/Cargo.lock +++ b/native/native-executors/Cargo.lock @@ -996,6 +996,16 @@ dependencies = [ "tokio", ] +[[package]] +name = "ballista-api-types" +version = "55.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b97d9eea12c759eda5f828b4f0eba2b42784d96a4e7858afe5624a9522507841" +dependencies = [ + "serde", + "utoipa", +] + [[package]] name = "ballista-core" version = "55.0.0" @@ -1023,6 +1033,7 @@ dependencies = [ "tonic-prost", "tonic-prost-build", "url", + "utoipa", "uuid", ] @@ -1056,6 +1067,19 @@ dependencies = [ "uuid", ] +[[package]] +name = "ballista-history" +version = "55.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "926ed4783d3b4d24fc28e899c7aef2f0127a511238320eadfc06c65ba2772bf8" +dependencies = [ + "ballista-api-types", + "log", + "serde", + "serde_json", + "tokio", +] + [[package]] name = "ballista-scheduler" version = "55.0.0" @@ -1065,7 +1089,9 @@ dependencies = [ "arrow-flight", "async-trait", "axum", + "ballista-api-types", "ballista-core", + "ballista-history", "dashmap", "datafusion", "datafusion-proto", @@ -1079,11 +1105,14 @@ dependencies = [ "prost", "rand 0.10.3", "serde", + "serde_json", "tokio", "tokio-stream", "tonic", "tower-http 0.7.1", + "tracing", "url-escape", + "utoipa", "uuid", ] @@ -3471,7 +3500,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "aes-gcm", "anyhow", @@ -3529,7 +3558,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "proc-macro2", "quote", @@ -3539,7 +3568,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "anyhow", "async-trait", @@ -6880,6 +6909,29 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "utoipa" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8765fe27aeff71012a3f90fa474cf1df863a7d0dd81ed25de4e63fff2544994f" +dependencies = [ + "indexmap 2.14.2", + "serde", + "serde_json", + "utoipa-gen", +] + +[[package]] +name = "utoipa-gen" +version = "6.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d935f1c83fdf8b88f09bbe050d0cea78ea4afee6d7b0317403228c1200b2c8ab" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "uuid" version = "1.29.0" diff --git a/native/native-executors/Cargo.toml b/native/native-executors/Cargo.toml index 117cd7e3df4..388c06189aa 100644 --- a/native/native-executors/Cargo.toml +++ b/native/native-executors/Cargo.toml @@ -37,7 +37,9 @@ publish = false arrow-flight = "59.2.0" ballista-core = "55.0.0" ballista-executor = { version = "55.0.0", default-features = false, features = ["arrow-ipc-optimizations"] } -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"] } datafusion = { version = "55.2.0", default-features = false } # Comet's core crate, under the name of its library. Without the default hdfs-opendal feature, # which links libjvm through libhdfs. From 94010d26d377eccf081e3d0c2a06129cca8b3810 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 2 Nov 2026 13:05:11 -0700 Subject: [PATCH 6/8] fix: keep Ballista's REST API off the Comet scheduler and document the 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. --- .../src/bin/comet-executor.rs | 16 +++-- .../src/bin/comet-scheduler.rs | 17 ++++- native/native-executors/tests/cluster.rs | 62 +++++++++++++++++++ 3 files changed, 87 insertions(+), 8 deletions(-) diff --git a/native/native-executors/src/bin/comet-executor.rs b/native/native-executors/src/bin/comet-executor.rs index 9a08a7c62df..9f0cfc8f78f 100644 --- a/native/native-executors/src/bin/comet-executor.rs +++ b/native/native-executors/src/bin/comet-executor.rs @@ -26,11 +26,17 @@ //! - `COMET_EXECUTOR_GRPC_PORT` (default `50052`) //! - `COMET_SCHEDULER_HOST` (default `localhost`) and `COMET_SCHEDULER_PORT` (default `50050`) //! - `COMET_EXECUTOR_TASKS`: concurrent tasks (default: the number of CPUs) -//! - `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 -//! in flight between stages, comes on top of it. +//! - `COMET_GRPC_MAX_MESSAGE_SIZE`: the largest message, in bytes, the executor sends or accepts +//! on Ballista's gRPC connections with the scheduler, such as a task with many partitions of a +//! scan, and on its Flight service (default: Ballista's 16 MiB). Set it on the scheduler as +//! well. The executor's reads of other executors' shuffle data keep the limit of the query's +//! `ballista.client.grpc_max_message_size` (16 MiB by default), which this does not change. +//! - `COMET_EXECUTOR_MEMORY_POOL_SIZE`: the executor's memory budget in bytes, `0` for no limit. +//! It is not one pool that the tasks share: each task gets its own pool of the budget divided +//! by `COMET_EXECUTOR_TASKS`, times the task slots it takes, even when it runs alone. By +//! default the budget is 70% of the smaller of the container's memory limit and the host's +//! memory, or no limit when neither can be found. Memory outside the pools, such as batches in +//! flight between stages, comes on top of it. //! - `COMET_EXECUTOR_WORK_DIR`: where shuffle and spill files go (default: a temporary directory) use std::sync::Arc; diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index 1d8f73f887a..d99e6142eb1 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -26,11 +26,20 @@ //! - `COMET_SCHEDULER_FLIGHT_PORT`: the Flight port, for drivers (default `50060`) //! - `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) +//! - `COMET_GRPC_MAX_MESSAGE_SIZE`: the largest message, in bytes, of Ballista's gRPC service +//! between the scheduler and its executors, such as a task with many partitions of a scan +//! (default: Ballista's 16 MiB). Set it on the executors as well, for their side of the same +//! messages. The Flight service for drivers has no limit. Shuffle reads between executors keep +//! the limit of the query's `ballista.client.grpc_max_message_size` (16 MiB by default), which +//! this does not change. //! - `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) +//! and compute time (default: no event logs). The task records are best effort: when the +//! writer falls behind, Ballista drops them, so a log can have fewer records than the query had +//! tasks. Logs stay in the directory until something removes them; +//! `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` applies to query results, not to logs. +//! +//! Ballista's REST API, which lists and cancels jobs without authentication, is not served. use std::net::SocketAddr; use std::sync::Arc; @@ -50,6 +59,8 @@ async fn main() -> Result<(), Box> { init_logging(); let mut config = SchedulerConfig { override_physical_codec: Some(Arc::new(CometPhysicalCodec::default())), + // Compiling the event-log writer compiles the REST API in with it, on the gRPC port. + disable_rest_api: true, ..Default::default() }; if let Some(host) = env_var("COMET_SCHEDULER_BIND_HOST")? { diff --git a/native/native-executors/tests/cluster.rs b/native/native-executors/tests/cluster.rs index dee8dcaa7b5..ca93a1a1bc5 100644 --- a/native/native-executors/tests/cluster.rs +++ b/native/native-executors/tests/cluster.rs @@ -21,6 +21,7 @@ mod common; use std::future::Future; +use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; @@ -43,6 +44,8 @@ struct Cluster { processes: Vec, work_dir: PathBuf, flight_url: String, + /// The port of Ballista's gRPC service, for executors. + scheduler_port: u16, } /// Held while a cluster starts, so that two tests don't pick the same free port in between. @@ -67,6 +70,7 @@ impl Cluster { std::process::id() )), flight_url: format!("http://127.0.0.1:{flight_port}"), + scheduler_port, }; let mut env = vec![ ("COMET_SCHEDULER_BIND_HOST", "127.0.0.1".to_string()), @@ -166,6 +170,24 @@ fn free_port() -> u16 { .port() } +/// 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(); + write!( + stream, + "GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n" + ) + .unwrap(); + let mut response = vec![]; + stream.read_to_end(&mut response).unwrap(); + let response = String::from_utf8_lossy(&response); + response + .split_whitespace() + .nth(1) + .and_then(|status| status.parse().ok()) + .unwrap_or_else(|| panic!("Not an HTTP response: {response}")) +} + fn wait_for(port: u16) { let deadline = Instant::now() + Duration::from_secs(60); while TcpStream::connect(("127.0.0.1", port)).is_err() { @@ -263,3 +285,43 @@ async fn endpoints_say_when_their_results_are_deleted() { tokio::time::sleep(Duration::from_millis(200)).await; } } + +#[tokio::test(flavor = "multi_thread")] +async fn writes_event_logs_without_serving_the_rest_api() { + let log_dir = std::env::temp_dir().join(format!( + "comet-native-executors-event-logs-{}-{}", + std::process::id(), + free_port() + )); + let cluster = Cluster::start_with(&[( + "COMET_SCHEDULER_EVENT_LOG_DIR", + log_dir.to_string_lossy().into_owned(), + )]); + cluster.run(&query(vec![range_stage(100, 2)])).await; + + // The scheduler renames a query's log to `.eventlog` once it has written all of it. + let deadline = Instant::now() + Duration::from_secs(60); + let log = loop { + let finished = std::fs::read_dir(&log_dir).ok().and_then(|entries| { + entries + .filter_map(|entry| entry.ok().map(|entry| entry.path())) + .find(|path| { + path.extension() + .is_some_and(|extension| extension == "eventlog") + }) + }); + if let Some(log) = finished { + break log; + } + assert!(Instant::now() < deadline, "No finished event log"); + tokio::time::sleep(Duration::from_millis(100)).await; + }; + assert!(!std::fs::read_to_string(&log).unwrap().is_empty()); + + // Compiling the event-log writer compiles Ballista's REST API in, which lists and cancels + // jobs without authentication on the scheduler's gRPC port. The scheduler doesn't serve it. + assert_eq!(http_status(cluster.scheduler_port, "/api/jobs"), 404); + + drop(cluster); + let _ = std::fs::remove_dir_all(&log_dir); +} From dd77ce75d66f95576302cd7976f06587f4298d5e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 3 Nov 2026 14:25:21 -0700 Subject: [PATCH 7/8] fix: check the event-log directory at startup and document what logs 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. --- .../src/bin/comet-scheduler.rs | 31 +++++++++++--- native/native-executors/tests/cluster.rs | 40 +++++++++++++++++++ 2 files changed, 66 insertions(+), 5 deletions(-) diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index d99e6142eb1..56009a3c7a6 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -33,15 +33,22 @@ //! the limit of the query's `ballista.client.grpc_max_message_size` (16 MiB by default), which //! this does not change. //! - `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). The task records are best effort: when the -//! writer falls behind, Ballista drops them, so a log can have fewer records than the query had -//! tasks. Logs stay in the directory until something removes them; -//! `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` applies to query results, not to logs. +//! 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 +//! block as one operator with no metrics, so the plan shows a block only by its size, input +//! rows count only the rows a stage reads from other stages (not those it reads from files), +//! and a task's compute time covers only that reading. A task's start and end times give how +//! long it ran. +//! The task records are best effort: when the writer falls behind, Ballista drops them, so a +//! log can have fewer records than the query had tasks. Logs stay in the directory until +//! something removes them; `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` applies to query +//! results, not to logs. //! //! Ballista's REST API, which lists and cancels jobs without authentication, is not served. use std::net::SocketAddr; +use std::path::Path; use std::sync::Arc; use arrow_flight::flight_service_server::FlightServiceServer; @@ -81,6 +88,11 @@ async fn main() -> Result<(), Box> { config.finished_job_data_clean_up_interval_seconds = seconds; } if let Some(dir) = env_var::("COMET_SCHEDULER_EVENT_LOG_DIR")? { + // Ballista creates the directory on a background task, and when it can't write there it + // only logs warnings. Fail at startup instead, as for a malformed setting. + check_writable(Path::new(&dir)).map_err(|e| { + format!("Cannot write event logs to COMET_SCHEDULER_EVENT_LOG_DIR {dir:?}: {e}") + })?; config.event_log_dir = Some(dir); } let flight_port: u16 = env_var("COMET_SCHEDULER_FLIGHT_PORT")?.unwrap_or(50060); @@ -108,3 +120,12 @@ async fn main() -> Result<(), Box> { } Ok(()) } + +/// Creates `dir` if it doesn't exist, and checks that the scheduler can create files in it: a +/// 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())); + std::fs::File::create(&probe)?; + std::fs::remove_file(probe) +} diff --git a/native/native-executors/tests/cluster.rs b/native/native-executors/tests/cluster.rs index ca93a1a1bc5..499625c40f0 100644 --- a/native/native-executors/tests/cluster.rs +++ b/native/native-executors/tests/cluster.rs @@ -325,3 +325,43 @@ async fn writes_event_logs_without_serving_the_rest_api() { drop(cluster); let _ = std::fs::remove_dir_all(&log_dir); } + +#[test] +fn refuses_an_event_log_dir_it_cannot_write() { + // No one can create a directory inside a file, not even root, as in CI's container. + let file = std::env::temp_dir().join(format!( + "comet-native-executors-not-a-dir-{}-{}", + std::process::id(), + free_port() + )); + std::fs::write(&file, "").unwrap(); + let mut scheduler = Command::new(env!("CARGO_BIN_EXE_comet-scheduler")) + .env("COMET_SCHEDULER_PORT", free_port().to_string()) + .env("COMET_SCHEDULER_FLIGHT_PORT", free_port().to_string()) + .env("COMET_SCHEDULER_EVENT_LOG_DIR", file.join("logs")) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(60); + let status = loop { + if let Some(status) = scheduler.try_wait().unwrap() { + break status; + } + if Instant::now() > deadline { + let _ = scheduler.kill(); + panic!("The scheduler started with an event-log directory it can't write to"); + } + std::thread::sleep(Duration::from_millis(100)); + }; + let mut stderr = String::new(); + scheduler + .stderr + .take() + .unwrap() + .read_to_string(&mut stderr) + .unwrap(); + let _ = std::fs::remove_file(&file); + assert!(!status.success()); + assert!(stderr.contains("COMET_SCHEDULER_EVENT_LOG_DIR"), "{stderr}"); +} From 10defc3779c4990c241e612d87b3e2bfd19f3e82 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 6 Nov 2026 09:11:21 -0700 Subject: [PATCH 8/8] fix: make the event-log probe name unique, bound the test's HTTP read, and explain the gRPC settings --- native/native-executors/src/bin/comet-scheduler.rs | 12 ++++++++++-- native/native-executors/tests/cluster.rs | 3 +++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/native/native-executors/src/bin/comet-scheduler.rs b/native/native-executors/src/bin/comet-scheduler.rs index 56009a3c7a6..ba68c19a1d6 100644 --- a/native/native-executors/src/bin/comet-scheduler.rs +++ b/native/native-executors/src/bin/comet-scheduler.rs @@ -35,7 +35,7 @@ //! - `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 +//! stage and task, with the task's times and the rows it wrote. The log sees a stage's Comet //! block as one operator with no metrics, so the plan shows a block only by its size, input //! rows count only the rows a stage reads from other stages (not those it reads from files), //! and a task's compute time covers only that reading. A task's start and end times give how @@ -77,6 +77,9 @@ async fn main() -> Result<(), Box> { config.external_host = host; } if let Some(size) = env_var("COMET_GRPC_MAX_MESSAGE_SIZE")? { + // The scheduler is a gRPC server for the executors and also a client of theirs, and has + // a setting for the client. An executor has one pair of settings, for its server and for + // its client of the scheduler. config.grpc_server_max_decoding_message_size = size; config.grpc_server_max_encoding_message_size = size; config.grpc_client_max_message_size = size; @@ -125,7 +128,12 @@ async fn main() -> Result<(), Box> { /// 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())); + // Schedulers in containers are often PID 1, so the name also has the time, for two that share + // a log volume. + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos()); + let probe = dir.join(format!(".comet-scheduler-{}-{nanos}", std::process::id())); std::fs::File::create(&probe)?; std::fs::remove_file(probe) } diff --git a/native/native-executors/tests/cluster.rs b/native/native-executors/tests/cluster.rs index 499625c40f0..2ed61c2cebf 100644 --- a/native/native-executors/tests/cluster.rs +++ b/native/native-executors/tests/cluster.rs @@ -173,6 +173,9 @@ fn free_port() -> u16 { /// 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(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); write!( stream, "GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n"