fix(master): page the demand listing; stop treating average size as max

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;

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<String>,
+    /// List mode only: at most this many tables, highest pending first.
+    /// Default 200, max 2000.
+    pub limit: Option<usize>,
+    /// 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,75 +595,111 @@ 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<lance_context_merge::demand::TableDemand>,
 }
 
 #[derive(Debug, Serialize)]
 pub struct DemandReport {
-    /// Whether this master currently holds planner leadership.
+    /// Whether some master currently holds planner leadership.
     pub leader: Option<crate::planner::LeaderRecord>,
+    /// 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<DemandView>,
 }
 
-/// `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<Arc<MasterState>>,
     Query(query): Query<DemandQuery>,
 ) -> Result<Json<DemandReport>, 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::<lance_context_merge::demand::TableDemand>(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(),
             oldest_pending_ms: record.oldest_pending_ms(),
             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<DemandView> = got
+            .kvs()
+            .iter()
+            .filter_map(|kv| {
+                serde_json::from_slice::<TableDemand>(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<DemandView> = kvs
+        .iter()
+        .filter_map(|kv| {
+            serde_json::from_slice::<TableDemand>(&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

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