diff --git a/crates/lance-context-master/src/planner.rs b/crates/lance-context-master/src/planner.rs index 4587ab8..04a4cd7 100644 --- a/crates/lance-context-master/src/planner.rs +++ b/crates/lance-context-master/src/planner.rs @@ -634,32 +634,30 @@ async fn shadow_pass( } // Estimated cost. ROLLOUT_APPEND_MAX_BYTES is the *target* for one // pass, not a ceiling: the append path reads a whole generation - // before checking it, so one oversized generation (decoded) sets - // the real floor, and buffers cost ~2x decoded. The planner does - // not know per-generation sizes, so the estimate is - // 2 x max(max_bytes, largest_generation_seen_for_table) - // where the largest generation is approximated from the table's - // sealed-bytes watermark growth per generation when the scan has - // reported bytes, and the configured local budget otherwise - an - // unknown must reserve the worst case this executor could be asked - // to hold, or the capacity check validates nothing. + // before checking it, so the largest single generation sets the + // real floor, and buffers cost ~2x decoded. The planner does not + // know per-generation sizes - a scan reports bytes-through-sealed, + // from which only an *average* per generation follows, and an + // average is not a maximum. Until per-generation sizes are + // reported the only honest reservation for a table whose largest + // generation is unknown is the executor's whole local budget. The + // average is recorded as a diagnostic so the gap between "what we + // reserve" and "what we would reserve with real sizes" is visible. let per_pass_target = state.config.append.rollout_append_max_bytes as u64; - let largest_generation = if s.pending_bytes > 0 && s.pending_generations > 0 { + let local_budget = state.config.append.rollout_append_local_memory_bytes as u64; + let average_generation = if s.pending_bytes > 0 && s.pending_generations > 0 { s.pending_bytes / s.pending_generations } else { 0 }; - let worst_known = per_pass_target.max(largest_generation); - let expected_bytes = if s.pending_bytes == 0 { - // No size data at all: the whole local budget, because that is - // what a single oversized generation can consume. - (state.config.append.rollout_append_local_memory_bytes as u64).max(per_pass_target * 2) - } else { - worst_known.saturating_mul(2).max(1) - }; - if s.pending_bytes == 0 { - metrics::counter!("planner_shadow_bytes_unknown_total").increment(1); - } + metrics::histogram!("planner_shadow_average_generation_bytes") + .record(average_generation as f64); + let expected_bytes = local_budget.max(per_pass_target * 2).max(1); + metrics::counter!( + "planner_shadow_reservation_basis_total", + "basis" => if s.pending_bytes == 0 { "no_size_data" } else { "average_only" } + ) + .increment(1); let pick = headroom .values_mut() .filter(|h| h.fits(TaskKind::MergeWal, expected_bytes)) @@ -1347,6 +1345,7 @@ mod tests { axum::extract::State(first.clone()), axum::extract::Query(crate::routes::DemandQuery { target: Some("hot".into()), + ..Default::default() }), ) .await @@ -1358,6 +1357,26 @@ mod tests { assert_eq!(report.tables.len(), 1); assert_eq!(report.tables[0].pending_generations, 42); assert_eq!(report.tables[0].shards, 2); + assert!( + report.tables[0].record.is_some(), + "single-target view carries the record" + ); + // List mode: paged, bounded, pending-sorted, summary only. + let axum::Json(list) = crate::routes::scheduler_demand( + axum::extract::State(first.clone()), + axum::extract::Query(crate::routes::DemandQuery { + target: None, + limit: Some(1), + include_idle: false, + }), + ) + .await + .unwrap(); + assert_eq!(list.total_tables, 2, "hot and cold are both cached"); + assert_eq!(list.tables_with_pending, 1); + assert_eq!(list.tables.len(), 1, "limit applied"); + assert_eq!(list.tables[0].target, "hot"); + assert!(list.tables[0].record.is_none(), "list view is summary only"); // Shadow placement: with this master's heartbeat live, the hot table // (42 pending >= 8) gets exactly one Shadow assignment against it; diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 7e0d486..5e1b60a 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -575,9 +575,16 @@ pub async fn list_cooldowns( #[derive(Deserialize, Default)] pub struct DemandQuery { pub target: Option, + /// List mode only: at most this many tables, highest pending first. + /// Default 200, max 2000. + pub limit: Option, + /// List mode only: include tables with zero pending generations. + #[serde(default)] + pub include_idle: bool, } -/// One table's demand as the planner sees it. +/// One table's demand as the planner sees it. The full record is included +/// only for a single-target query; the list view carries the summary. #[derive(Debug, Serialize)] pub struct DemandView { pub target: String, @@ -588,63 +595,50 @@ pub struct DemandView { pub shards: usize, pub observed_revision: i64, pub updated_ms: i64, - pub record: lance_context_merge::demand::TableDemand, + #[serde(skip_serializing_if = "Option::is_none")] + pub record: Option, } #[derive(Debug, Serialize)] pub struct DemandReport { - /// Whether this master currently holds planner leadership. + /// Whether some master currently holds planner leadership. pub leader: Option, + /// Tables in the cache (list mode), before `limit`/`include_idle`. + pub total_tables: usize, + /// Tables with pending > 0 in the cache (list mode). + pub tables_with_pending: usize, pub tables: Vec, } -/// `GET /api/v1/scheduler/demand[?target=]` — the planner's demand cache, -/// folded from per-shard watermark events. Read-only; this is what the -/// planner would schedule from. Absent while no planner has run. +/// `GET /api/v1/scheduler/demand[?target=|?limit=&include_idle=]` — the +/// planner's demand cache, folded from per-shard watermark events. +/// Read-only. With `target`, one table with its full record. Without, a +/// paged read of the whole cache (13k tables is tens of pages) reduced to a +/// bounded, pending-sorted summary so the response is never the size of +/// the cache. Absent while no planner has run. pub async fn scheduler_demand( State(state): State>, Query(query): Query, ) -> Result, MasterError> { + use lance_context_merge::demand::TableDemand; let keys = crate::planner::Keys::new(&state.config.etcd.etcd_prefix); - let mut client = state.task_store.etcd_client().clone(); + let client = state.task_store.etcd_client(); let leader = client + .clone() .get(keys.leader(), None) .await .map_err(|e| MasterError::Internal(e.to_string()))? .kvs() .first() .and_then(|kv| serde_json::from_slice(kv.value()).ok()); - let response = match &query.target { - Some(target) => { - if target.is_empty() || target.len() > 512 { - return Err(MasterError::InvalidRequest("invalid target".into())); - } - client.get(keys.demand(target), None).await - } - None => { - client - .get( - keys.demand_prefix(), - Some(etcd_client::GetOptions::new().with_prefix()), - ) - .await - } - } - .map_err(|e| MasterError::Internal(e.to_string()))?; let prefix = keys.demand_prefix(); - let mut tables = Vec::new(); - for kv in response.kvs() { - let Ok(record) = - serde_json::from_slice::(kv.value()) - else { - continue; - }; - let key = String::from_utf8_lossy(kv.key()); + let view = |key: &[u8], record: TableDemand, full: bool| { + let key = String::from_utf8_lossy(key); let target = key .strip_prefix(prefix.as_str()) .and_then(crate::planner::unhex) .unwrap_or_else(|| key.to_string()); - tables.push(DemandView { + DemandView { target, pending_generations: record.pending_generations(), pending_bytes: record.pending_bytes(), @@ -652,11 +646,60 @@ pub async fn scheduler_demand( shards: record.shards.len(), observed_revision: record.observed_revision, updated_ms: record.updated_ms, - record, - }); + record: full.then_some(record), + } + }; + if let Some(target) = &query.target { + if target.is_empty() || target.len() > 512 { + return Err(MasterError::InvalidRequest("invalid target".into())); + } + let got = client + .clone() + .get(keys.demand(target), None) + .await + .map_err(|e| MasterError::Internal(e.to_string()))?; + let tables: Vec = got + .kvs() + .iter() + .filter_map(|kv| { + serde_json::from_slice::(kv.value()) + .ok() + .map(|r| view(kv.key(), r, true)) + }) + .collect(); + let with_pending = tables.iter().filter(|t| t.pending_generations > 0).count(); + return Ok(Json(DemandReport { + leader, + total_tables: tables.len(), + tables_with_pending: with_pending, + tables, + })); + } + let limit = query.limit.unwrap_or(200).clamp(1, 2000); + let (kvs, _) = crate::planner::read_prefix_paged(client, &prefix) + .await + .map_err(MasterError::Internal)?; + let total_tables = kvs.len(); + let mut tables: Vec = kvs + .iter() + .filter_map(|kv| { + serde_json::from_slice::(&kv.value) + .ok() + .map(|r| view(&kv.key, r, false)) + }) + .collect(); + let tables_with_pending = tables.iter().filter(|t| t.pending_generations > 0).count(); + if !query.include_idle { + tables.retain(|t| t.pending_generations > 0); } tables.sort_by_key(|t| std::cmp::Reverse(t.pending_generations)); - Ok(Json(DemandReport { leader, tables })) + tables.truncate(limit); + Ok(Json(DemandReport { + leader, + total_tables, + tables_with_pending, + tables, + })) } /// `GET /api/v1/scheduler/executors` — every live executor heartbeat with diff --git a/docs/design/scheduler-p1-data-model.md b/docs/design/scheduler-p1-data-model.md index 0f6c7d6..ea88b7f 100644 --- a/docs/design/scheduler-p1-data-model.md +++ b/docs/design/scheduler-p1-data-model.md @@ -219,19 +219,19 @@ says no demand events are lost; `placement_gap` is expected to stay non‑zero until legacy workers and catch‑up Jobs heartbeat as executors. Byte estimates: `ROLLOUT_APPEND_MAX_BYTES` is a per‑pass *target*, not a -ceiling — the append path reads a whole generation before checking it, so -one oversized generation sets the real floor and buffers cost ~2× decoded. -The estimate is `2 × max(max_bytes, largest generation seen)`, where the -largest generation is approximated from the table's bytes‑per‑generation -when the scan has reported bytes. When no size is known at all the reservation -is the executor's **whole local budget**, because that is what a single -oversized generation can consume. Writers do not measure generation sizes, so -this is the common case today; `planner_shadow_bytes_unknown_total` counts it. -Measured per‑generation sizes come with the scan reading generation data -statistics, a later change. Compaction and -index scoring, commit‑ready class 2 and `max_consecutive_turns` come with P2's -preparation integration; catch‑up Job heartbeats come when Jobs become an -executor kind. +ceiling — the append path reads a whole generation before checking it, so the +largest single generation sets the real floor and buffers cost ~2× decoded. +The planner does not know per‑generation sizes: a scan reports +bytes‑through‑sealed, from which only an **average** per generation follows, +and an average is not a maximum. Until per‑generation sizes are reported +(a later change to the scan), every shadow reservation is the executor's +**whole local budget**, which is the only honest answer for an unknown +maximum. That makes the shadow capacity model deliberately pessimistic — +it answers "could this executor safely take one unit" and **cannot** answer +"how many units in parallel"; do not read `placement_gap` as a parallelism +measurement. `planner_shadow_reservation_basis_total{basis}` and +`planner_shadow_average_generation_bytes` record what information each +reservation was based on so the gap to a size‑aware model is visible. ## 3. Leader