From a3f3abd66bda25be15bb0219f76f4fb115e96160 Mon Sep 17 00:00:00 2001 From: Beinan Date: Fri, 9 Oct 2026 08:05:31 +0000 Subject: [PATCH] fix(master): page the demand listing; stop treating average size as max Review round 7, two code findings. GET /scheduler/demand without a target still did one unpaged prefix read of the whole cache, so at 13k tables it failed the same 4 MiB cap the planner was fixed for in #364. The listing is now a paged read reduced to a bounded summary: pending-sorted, default 200 and at most 2000 rows, idle tables excluded unless include_idle=true, with total_tables and tables_with_pending alongside. The full TableDemand record is returned only for a single-target query. Response size is now independent of cache size. The shadow byte estimate used pending_bytes / pending_generations - an average - where a maximum belongs. The append path reads a whole generation before checking the per-pass target, so the largest single generation is the real floor, and no publisher reports per-generation sizes yet. The estimate is now the executor's whole local budget for every unit, which is the only honest reservation for an unknown maximum. The spec says plainly that this makes the shadow capacity model answer "can one unit fit" and not "how many in parallel", and that placement_gap must not be read as a parallelism measurement. planner_shadow_reservation_basis_total{basis} and planner_shadow_average_generation_bytes record what each reservation was based on. Tests: the planner etcd test now also exercises the list view - paged, limit=1 returns only the pending table, record omitted, totals correct. 18 planner/routes tests pass; clippy -D warnings, fmt, typos clean. Refs #333. Review round 7. Co-Authored-By: Claude Fable 5 --- crates/lance-context-master/src/planner.rs | 61 +++++++---- crates/lance-context-master/src/routes.rs | 115 ++++++++++++++------- docs/design/scheduler-p1-data-model.md | 26 ++--- 3 files changed, 132 insertions(+), 70 deletions(-) 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