andygrove commented on code in PR #7299: URL: https://github.com/apache/datafusion-comet/pull/7299#discussion_r4403667098
########## native/native-executors/src/bin/comet-scheduler.rs: ########## @@ -0,0 +1,76 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A Comet scheduler: Ballista's scheduler, which Comet executors register with, wrapped in the +//! Arrow Flight service that Spark drivers submit queries to. +//! +//! Configured through environment variables, with Ballista's defaults otherwise: +//! - `COMET_SCHEDULER_BIND_HOST` (default `127.0.0.1`) +//! - `COMET_SCHEDULER_PORT`: Ballista's gRPC port, for executors (default `50050`) +//! - `COMET_SCHEDULER_FLIGHT_PORT`: the Flight port, for drivers (default `50060`) + +use std::net::SocketAddr; +use std::sync::Arc; + +use arrow_flight::flight_service_server::FlightServiceServer; +use ballista_scheduler::cluster::BallistaCluster; +use ballista_scheduler::config::SchedulerConfig; +use ballista_scheduler::scheduler_process::{create_scheduler, start_grpc_service}; +use comet_native_executors::codec::CometPhysicalCodec; +use comet_native_executors::frontend::CometFlightService; +use comet_native_executors::{env_var, init_logging}; +use datafusion_proto::protobuf::{LogicalPlanNode, PhysicalPlanNode}; +use log::info; + +#[tokio::main] +async fn main() -> Result<(), Box<dyn std::error::Error>> { + init_logging(); + let mut config = SchedulerConfig { + override_physical_codec: Some(Arc::new(CometPhysicalCodec::default())), + ..Default::default() Review Comment: Confirmed in the Ballista 55.0.0 source, thanks. 16f004ab2a makes the lifetime something an operator can set and a client can see, rather than keeping results until release. `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` sets Ballista's `finished_job_data_clean_up_interval_seconds` (300 by default, and 0 keeps results), documented in the scheduler binary and the README. Each endpoint's `expiration_time` is the job's `ended_at` plus that retention, and since Ballista records `ended_at` before it starts the cleanup timer, a partition is still there until its endpoint expires. `endpoints_say_when_their_results_are_deleted` in `tests/cluster.rs` runs with a five-second retention: it checks the expiration time, reads the first endpoint, waits until it expires, and checks that the second, which nothing read, is gone. Releasing results once the driver has read them needs the driver to do it, so I've added that to the driver-side sub-task in #7288. -- 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]
