diff --git a/crates/cost/src/execution_demand.rs b/crates/cost/src/execution_demand.rs index d966d03bd..c65b9b02e 100644 --- a/crates/cost/src/execution_demand.rs +++ b/crates/cost/src/execution_demand.rs @@ -10,6 +10,7 @@ use super::analytical::AnalyticalCostError; use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId}; use asap_types::ir::properties::ExecutionTiming; +use asap_types::ir::{ASAPOp, Operator}; use asap_types::physical::Lifetime; use asap_types::workload::{QueryRecurrence, RangeEvaluation, RepeatedDemand}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; @@ -335,22 +336,28 @@ pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap = dag + .nodes + .iter() + .filter_map(|node| match node.payload { + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + tumbling_window_width_ms, + .. + }) => Some((node.id, tumbling_window_width_ms)), + _ => None, + }) + .collect(); let inputs = |id: PhysicalASAPNodeId| { dag.edges .iter() @@ -389,10 +408,12 @@ pub fn computations( let mut rate_of_materialized = HashMap::new(); for (&id, &(timing, lifetime)) in &materialized_nodes { let at_ingestion = timing == ExecutionTiming::IngestionTime; - // A query-time chain materialized across evaluations is built once per request. + // A query-time result materialized across evaluations is computed + // once per request; a query-time chain builds one window when it ends. let per_request = !at_ingestion && matches!(lifetime, Lifetime::AcrossEvaluations(_)); - let rate = match at_ingestion { - true => 1.0, + let rate = match (at_ingestion, chain_width_ms.get(&id)) { + (true, _) => 1.0, + (false, Some(&width_ms)) if per_request => 1_000.0 / width_ms.max(1) as f64, _ => evaluation_rate( reached.get(&id).into_iter().flatten().filter_map(|&q| { let range = if per_request { None } else { range_of(q) }; @@ -421,23 +442,48 @@ pub fn computations( .filter(|(_, (timing, _))| *timing == ExecutionTiming::IngestionTime) .map(|(&id, _)| id), ); + // Each query's evaluation grid: the queries on one grid are one + // execution context, evaluated together. + let grid_ms = |query: usize| { + recurrence + .get(query) + .and_then(|r| r.evaluation_grid_ms(range_of(query))) + .unwrap_or(1) + }; let mut count: HashMap<_, (u64, f64)> = HashMap::new(); for (id, of_node) in consumers(dag) { let entry = count.entry(id).or_default(); // The refresh windows whose ingestion run already computes the node. let mut runs = HashSet::new(); + // The queries reaching it, by evaluation grid. + let mut queries_by_grid: BTreeMap> = BTreeMap::new(); for consumer in of_node { - if let Consumer::Materialized(node) = consumer { - if let Some(Some(window)) = refresh_windows.get(&node) { - if !runs.insert(*window) { - continue; + match consumer { + Consumer::Query(query) => queries_by_grid + .entry(grid_ms(query)) + .or_default() + .push(query), + Consumer::Materialized(node) => { + if let Some(Some(window)) = refresh_windows.get(&node) { + if !runs.insert(*window) { + continue; + } } + entry.0 += 1; + entry.1 += rate_of_materialized[&node]; } } + } + for queries in queries_by_grid.into_values() { entry.0 += 1; - entry.1 += match consumer { - Consumer::Query(query) => query_rates[query], - Consumer::Materialized(node) => rate_of_materialized[&node], + entry.1 += match root_rates { + Some(_) => queries.iter().map(|&query| query_rates[query]).sum(), + None => evaluation_rate( + queries + .iter() + .filter_map(|&query| recurrence.get(query).map(|r| (r, range_of(query)))), + non_repeating_query_horizon_seconds, + )?, }; } } @@ -624,6 +670,134 @@ mod tests { ); } + /// A plan node `id` running `payload` with `timing` and `lifetime`. + fn node( + id: PhysicalASAPNodeId, + payload: Operator, + timing: ExecutionTiming, + lifetime: Lifetime, + ) -> asap_types::ir::physical_export::PhysicalASAPDAGNode { + use asap_types::ir::properties::{DataPrimitive, ExecutionDataState}; + use asap_types::ir::schema::{DataType, Field, FieldDataType, Schema}; + asap_types::ir::physical_export::PhysicalASAPDAGNode { + id, + output_state: ExecutionDataState { + timing, + primitive: DataPrimitive::of( + &payload, + asap_types::ir::OperatorResultKind::RangeVector, + ), + }, + parameters: matches!(payload, Operator::ASAP(_)) + .then_some(asap_types::ir::schema::SketchParams::Kll { k: 200 }), + payload, + lifetime, + output_schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::new("value", FieldDataType::Plain(DataType::Float64), false), + ], + 0, + vec![], + ), + guarantee: None, + retention_ms: None, + summary_coverage: None, + selected_physical_candidate: None, + } + } + + fn scan() -> Operator { + use asap_types::ir::operator::operator_properties::Source; + use asap_types::ir::schema::{DataType, Field, Schema}; + Operator::NonASAP(asap_types::ir::NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + ], + 0, + vec![], + ), + }) + } + + fn last(seconds: u64, child: PhysicalASAPNodeId) -> Operator { + Operator::NonASAP(asap_types::ir::NonASAPOp::TimeRange { + range: std::time::Duration::from_secs(seconds), + kind: asap_types::ir::TimeRangeKind::Range, + child, + }) + } + + /// `m[5m]` read by three queries: two every 10 s at phase 0, one every + /// 30 s. The two on the 10 s grid are one execution context, which + /// computes the not-materialized range and scan once for both: two + /// contexts, 1/10 + 1/30 per second. + #[test] + fn a_not_materialized_node_is_computed_once_per_evaluation_grid() { + let query_time = ExecutionTiming::QueryTime; + let dag = PhysicalASAPDAG::new( + vec![ + node(0, scan(), query_time, Lifetime::None), + node(1, last(300, 0), query_time, Lifetime::None), + ], + vec![1, 1, 1], + vec![], + ) + .unwrap(); + let recurrence = [ + fixed(10_000, Some(0)), + fixed(10_000, Some(0)), + fixed(30_000, Some(0)), + ]; + let rates = computations(&dag, &recurrence, &[], 3600.0, None).unwrap(); + for id in [0, 1] { + assert_eq!(rates[&id].0, 2); + close(rates[&id].1, 1.0 / 10.0 + 1.0 / 30.0); + } + } + + /// A query-time chain of 10 s windows materialized across evaluations, + /// read through its last minute by a query every 60 s: it builds one + /// window when each ends, 1/10 per second, and so does the scan it + /// builds from; the range over it runs at the query's rate. + #[test] + fn a_query_time_chain_builds_one_window_per_width() { + use asap_types::ir::operator::operator_properties::Reduction; + use asap_types::ir::schema::{FieldDataType, SketchAlgorithm, SummaryUpdate}; + let query_time = ExecutionTiming::QueryTime; + let chain = Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: 0, + summary_type: FieldDataType::Sketch(SketchAlgorithm::Kll), + input: SummaryUpdate::column(asap_types::ir::scalar::expr_ir::ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + filter: None, + tumbling_window_width_ms: 10_000, + }); + let dag = PhysicalASAPDAG::new( + vec![ + node(0, scan(), query_time, Lifetime::None), + node( + 1, + chain, + query_time, + Lifetime::AcrossEvaluations(asap_types::physical::StorageMedium::Memory), + ), + node(2, last(60, 1), query_time, Lifetime::None), + ], + vec![2], + vec![], + ) + .unwrap(); + let rates = computations(&dag, &[fixed(60_000, Some(0))], &[], 3600.0, None).unwrap(); + close(rates[&0].1, 1.0 / 10.0); + close(rates[&1].1, 1.0 / 10.0); + close(rates[&2].1, 1.0 / 60.0); + } + // Invalid fixed intervals and estimated rates still produce the evaluator's error. #[test] fn invalid_recurrence_is_rejected() { diff --git a/crates/cost/src/usage.rs b/crates/cost/src/usage.rs index 8f0bbad9f..1a43799f9 100644 --- a/crates/cost/src/usage.rs +++ b/crates/cost/src/usage.rs @@ -994,7 +994,8 @@ pub fn node_usage<'a>( // One window's states: each input row updates the window its // time falls in. At ingestion time the rows of one window // arrive at the rate of one second's `input.rows`; at query - // time its input holds the rows of every retained window. + // time each build reads the rows of the window that just ended, + // its input. let group_count = groups(reduction)?; let rows_per_tumbling_window = match ingestion { true => u64::try_from( @@ -1002,7 +1003,7 @@ pub fn node_usage<'a>( .div_ceil(1_000), ) .map_err(|_| AnalyticalCostError::Overflow)?, - false => input.rows.div_ceil(retained_tumbling_window_count), + false => input.rows, }; let input_rows = StateInputRows { rows_per_state: Some(rows_per_tumbling_window.div_ceil(group_count.max(1))), diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 2fdff66df..c0f81d537 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -22,6 +22,7 @@ use asap_planner::{ SelectedPlanReport, UserInput, }; use asap_types::ir::operator::Source; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator}; use futures::{executor::block_on, StreamExt}; @@ -34,11 +35,19 @@ const EVALUATION_MS: i64 = 3_600_000; /// δ 0.01, over 1,000 ingesting `latency_ms` series sampled every second: /// #509 Example 4b when the one query is its p99. fn workload(queries: &[&str]) -> PlanningWorkload { + let every_10_minutes: Vec<(&str, u64)> = + queries.iter().map(|&query| (query, 600_000)).collect(); + workload_every(&every_10_minutes) +} + +/// [`workload`] with each query repeating every `interval_ms` of its own, at +/// phase 0. +fn workload_every(queries: &[(&str, u64)]) -> PlanningWorkload { let entries: Vec = queries .iter() - .map(|query| { + .map(|(query, interval_ms)| { format!( - r#"{{"query": {query:?}, "demand": {{"fixed_interval_at": {{"interval": 600000, "evaluation_phase": 0}}}}, + r#"{{"query": {query:?}, "demand": {{"fixed_interval_at": {{"interval": {interval_ms}, "evaluation_phase": 0}}}}, "requirements": {{"accuracy": {{"accuracy_metric": "rank", "max_error": 0.01, "failure_probability": 0.01, "accuracy_scope": "whole_result"}}, "max_latency_ms": 200.0}}, @@ -249,6 +258,53 @@ async fn the_plan_carries_stage3s_predicted_usage() { assert!(usage.per_node.iter().any(|node| node.node_id == root)); } +/// Example 4b's p99 every 10 minutes and its p50 every 30 minutes: two +/// evaluation contexts. The chain of windows maintained at ingestion time +/// runs once, whoever reads it, and retains the hour both read back +/// ($h_v$, constraint (R)); each context reads it with query-time nodes of +/// its own. +#[tokio::test] +async fn one_chain_serves_two_evaluation_contexts() { + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let (plan, _) = plan_with_report_for( + workload_every(&[ + ("quantile_over_time(0.99, latency_ms[60m])", 600_000), + ("quantile_over_time(0.5, latency_ms[60m])", 1_800_000), + ]), + &capabilities, + ) + .await; + let chains: Vec<_> = plan + .dag + .nodes + .iter() + .filter(|node| { + matches!( + node.payload, + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. }) + ) + }) + .collect(); + assert_eq!(chains.len(), 1, "{:#?}", plan.dag.nodes); + assert_eq!(chains[0].retention_ms, Some(3_600_000)); + let ranges_over_the_chain = plan + .dag + .nodes + .iter() + .filter(|node| { + matches!( + node.payload, + Operator::NonASAP(NonASAPOp::TimeRange { child, .. }) if child == chains[0].id + ) + }) + .count(); + assert_eq!(ranges_over_the_chain, 2); + assert_ne!(plan.dag.roots[0], plan.dag.roots[1]); +} + /// `sum(latency_ms)` reads an instant selector: per series, only the latest /// sample in its 1 s lookback counts. Series `a` has samples at T-900 ms (1) /// and T-100 ms (10), series `b` at T-200 ms (100): the sum is 10 + 100. diff --git a/crates/stage3-plan-selection/src/cost.rs b/crates/stage3-plan-selection/src/cost.rs index 5749058ec..1f15cb2c0 100644 --- a/crates/stage3-plan-selection/src/cost.rs +++ b/crates/stage3-plan-selection/src/cost.rs @@ -1,32 +1,38 @@ -//! The baseline objective's coefficients (concepts doc, section 3, "Stage -//! 3: baseline selection and plan output"): $w_v$ of every physical -//! candidate and $c_{u,c,i}$ of every edge a plan may select, from -//! `asap-cost` under cost model 1. +//! The objective's coefficients (concepts doc, sections 3 and 5, "Cost +//! and retention") under cost model 1, from `asap-cost`: per execution of +//! a candidate in a context ([`crate::milp::execution_contexts`]), per +//! edge in a context, and per amount a candidate retains: +//! +//! $$\sum_v W^{\mathrm{mat}}_v x^{\mathrm{mat}}_v + \sum_{\kappa,v} +//! \rho_\kappa c_{\kappa,v} x_{\kappa,v} + \sum_{\kappa,u,c,i} +//! \rho_\kappa c_{\kappa,u,c,i} y_{\kappa,u,c,i} + \sum_v \pi_v h_v.$$ //! //! Each candidate is costed as one plan node ([`candidate_node`], //! [`asap_cost::node_usage`]) over the logical output of each input group: //! the most any of the group's costed candidates outputs (rows, bytes and, -//! for a state, its θ). That is $w_v$, which does not depend on the -//! producers chosen. The part of a consumer's cost that changes with the -//! producer read is on the edge: $c_{u,c,i}$ is the consumer's cost with -//! input $i$ read from $u$'s own output, minus $w_c$. It is exact when a -//! consumer's cost changes with one input's producer only (every consumer -//! of a state reads one state input); a state candidate without θ of its -//! own (a merge, a `TimeRange` over a chain) outputs the most of its input -//! group's θ, so a consumer two hops above a sketch build is costed at the -//! largest θ below it. +//! for a state, its θ), at its execution's rate $\rho_\kappa$ +//! (`executions_per_second`). That does not depend on the producers +//! chosen. The part of a consumer's cost that changes with the producer +//! read is on the edge: $\rho_\kappa c_{\kappa,u,c,i}$ is the consumer's +//! cost in $\kappa$ with input $i$ read from $u$'s own output, minus its +//! own. It is exact when a consumer's cost changes with one input's +//! producer only (every consumer of a state reads one state input); a state +//! candidate without θ of its own (a merge, a `TimeRange` over a chain) +//! outputs the most of its input group's θ, so a consumer two hops above a +//! sketch build is costed at the largest θ below it. //! -//! How often a candidate runs is the baseline's fixed rate: at ingestion -//! time, one second of arriving rows per second; at query time, once per -//! evaluation of the queries reaching its group (once per request when it -//! is materialized across evaluations). A materialized candidate and a -//! query-time `Scan` retain the most their selectable consumers read back -//! ([`retention_need_ms`]). Execution contexts (#1207 PR 7) make both per -//! context. +//! A candidate that retains time-indexed rows ([`retains_rows`]) is costed +//! retaining the most its consumers read back ([`retention_need_ms`]), the +//! largest $h_v$ (R) allows. Retention is linear in that amount: $\pi_v$ is +//! what retaining one more window (or second) costs there, and $W^{\mathrm{mat}}_v$ +//! of a materialized candidate is its cost less $\pi_v h_v$; a query-time +//! scan's executions keep reading what it retains, and only its raw data on +//! disk is in $\pi_v h_v$. A candidate one of whose consumers reads it +//! unbounded (a `Scan` read whole, a `TimeShift` anchored with `@`) has no +//! $h_v$: it is costed retaining everything. use std::collections::{BTreeSet, HashMap}; -use asap_cost::execution_demand::evaluation_rate; use asap_cost::{node_usage, CostInputs, CostModel, NodeUsage, ProducerOutput}; use asap_types::ir::materialized_view::ViewCoverage; use asap_types::ir::physical_export::{ @@ -40,14 +46,26 @@ use asap_types::stages::{ }; use asap_types::workload::{DataWorkload, QueryWorkloadEntry}; -/// $w_v$ and $c_{u,c,i}$. -pub(crate) struct SelectionCosts { - /// $w_v$ per second, by [`PhysicalCandidateId`], or why `asap-cost` - /// could not cost it (the MILP leaves such a candidate out). - pub(crate) candidate: Vec>, - /// $c_{u,c,i}$ per second of each selectable edge whose producer and - /// consumer are costable; an edge missing here cannot be selected. - pub(crate) edge: HashMap, +use crate::milp::execution_contexts::{executions_per_second, Execution, ExecutionContexts}; +use crate::milp::model::QueryId; + +/// The objective's coefficients (concepts doc, section 5, "Cost and +/// retention"), per second. +pub(crate) struct ExecutionCosts { + /// $\rho_\kappa c_{\kappa,v}$ of each execution, $W^{\mathrm{mat}}_v$ + /// of an execution in `Materialized(v)`, by position in + /// `ExecutionContexts::executions`; or why `asap-cost` could not cost + /// it (the MILP leaves it out). + pub(crate) per_execution: Vec>, + /// $\rho_\kappa c_{\kappa,u,c,i}$ of each edge, by position in + /// `ExecutionContexts::edges`; `None` when it could not be costed (the + /// MILP leaves it out). + pub(crate) per_edge: Vec>, + /// $\pi_v$, per ms retained, of each candidate that retains rows + /// ([`retains_rows`]) only within a bound: it has $h_v$ and + /// constraint (R). One whose consumer reads it unbounded is costed + /// retaining everything, in its executions' costs. + pub(crate) retention_cost_per_second: HashMap, } /// Candidate `v` as a plan node with id `id`, its input `i` read from node @@ -126,8 +144,9 @@ pub(crate) fn retains_rows(physical: &PhysicalASAPMemoCandidates, v: PhysicalCan /// offset plus range of the `TimeShift`s and the `TimeRange` above it on /// the consumers' side (`consumers_of` gives each candidate's edges to its /// own consumers), plus the longest range-evaluation span of the queries -/// reaching the `TimeRange`; a tumbling-window chain reads its newest -/// window back. A consumer that reads it otherwise reads it at the +/// reaching the `TimeRange`; a tumbling-window chain materialized across +/// evaluations reads its newest window back, and one that is not what its +/// own consumers read. A consumer that reads it otherwise reads it at the /// evaluation time (only the span), except a `Scan`, which it reads whole: /// `None`, unbounded, as for a `TimeShift` anchored with `@`. pub(crate) fn retention_need_ms( @@ -186,11 +205,28 @@ pub(crate) fn retention_need_ms( continue; } Operator::NonASAP(NonASAPOp::TimeShift { .. }) => return None, - // A chain builds its newest window from it. + // A materialized chain builds its newest window from it; one + // that is not builds, in each run, every window its own + // consumers read. Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { tumbling_window_width_ms, .. - }) => offset_ms.checked_add(i64::try_from(*tumbling_window_width_ms).ok()?)?, + }) => match physical.physical_candidates[consumer.0] + .physical_properties + .lifetime() + { + Lifetime::AcrossEvaluations(_) => { + offset_ms.checked_add(i64::try_from(*tumbling_window_width_ms).ok()?)? + } + Lifetime::None | Lifetime::OneEvaluation(_) => { + pending.extend( + consumers_of(consumer) + .into_iter() + .map(|edge| (physical.edge(edge).consumer_candidate_id, offset_ms)), + ); + continue; + } + }, _ if reads_raw_data => return None, _ => offset_ms.checked_add(span_ms(consumer))?, }; @@ -199,38 +235,59 @@ pub(crate) fn retention_need_ms( u64::try_from(need_ms).ok() } -/// [`SelectionCosts`] over `selectable` (each admissible edge a plan may -/// select, by consumer and input). Errs on nothing: a candidate -/// `asap-cost` cannot cost is reported in [`SelectionCosts::candidate`]. -pub(crate) fn selection_costs( +/// The objective's coefficients of every execution and edge of `contexts` +/// ([`ExecutionCosts`]; module doc). `query_outputs` are the executions +/// that may answer each query output. Errs on nothing: what `asap-cost` +/// cannot cost is reported in [`ExecutionCosts::per_execution`] or left +/// out of [`ExecutionCosts::per_edge`]. +pub(crate) fn execution_costs( physical: &PhysicalASAPMemoCandidates, - selectable: &[PhysicalEdgeId], + contexts: &ExecutionContexts, + query_outputs: &[(QueryId, Vec)], entries: &[QueryWorkloadEntry], data: &DataWorkload, cost_inputs: &CostInputs, formula_scale: f64, -) -> SelectionCosts { +) -> ExecutionCosts { let memo = &physical.logical_candidates.memo; let count = physical.physical_candidates.len(); - let mut out_edges: Vec> = vec![Vec::new(); count]; - for &edge in selectable { - out_edges[physical.edge(edge).producer_candidate_id.0].push(edge); + // The most any consumer of each candidate that retains rows reads back + // ($h_v$ at its largest): `None` when one reads it unbounded. + let mut most_read_ms: HashMap> = (0..count) + .map(PhysicalCandidateId) + .filter(|&v| retains_rows(physical, v)) + .map(|v| (v, Some(0))) + .collect(); + let mut read = |v: PhysicalCandidateId, need_ms: Option| { + if let Some(most) = most_read_ms.get_mut(&v) { + *most = most.zip(need_ms).map(|(most, need)| most.max(need)); + } + }; + for &(_, edge, producer) in contexts.edges() { + read( + producer.candidate, + edge_retention_need_ms(physical, entries, edge), + ); + } + for (query, executions) in query_outputs { + for execution in executions { + read( + execution.candidate, + Some(output_retention_need_ms(&entries[query.0])), + ); + } } // Each candidate's node, its inputs numbered 0.. (the ids do not - // matter to `node_usage`), with the retention it would need. + // matter to `node_usage`), retaining the most its consumers read back. + let node_retaining = |v: PhysicalCandidateId, retention_ms: Option| { + let inputs: Vec = (0..input_count(physical, v)).collect(); + let mut node = candidate_node(physical, v, v.0, &inputs); + node.retention_ms = retention_ms; + node + }; let nodes: Vec = (0..count) .map(PhysicalCandidateId) - .map(|v| { - let inputs: Vec = (0..input_count(physical, v)).collect(); - let mut node = candidate_node(physical, v, v.0, &inputs); - if retains_rows(physical, v) { - node.retention_ms = - retention_need_ms(physical, entries, v, &out_edges[v.0], &|consumer| { - out_edges[consumer.0].clone() - }); - } - node - }) + .map(|v| node_retaining(v, most_read_ms.get(&v).copied().flatten())) .collect(); // Each group's expression as a payload, for its logical output. let group_payloads: Vec = memo @@ -247,21 +304,35 @@ pub(crate) fn selection_costs( let order = memo .topological_order() .unwrap_or_else(|| (0..memo.groups().len() as u32).map(MemoGroupId).collect()); + let cost = |v: PhysicalCandidateId, + node: &PhysicalASAPDAGNode, + inputs: &[ProducerOutput<'_>], + per_second: f64| { + cost_node( + physical, + v, + node, + inputs, + per_second, + data, + cost_inputs, + formula_scale, + ) + .and_then(|usage| cost_per_second(&usage, cost_inputs)) + }; + // Each candidate's usage of one computation (what it outputs does not + // depend on how often it runs) over its input groups' logical outputs. let mut usage: Vec, String>>> = (0..count).map(|_| None).collect(); let mut logical: Vec>> = vec![None; memo.groups().len()]; for group in order { for &v in &candidates_of_group[group.0 as usize] { - let inputs: Option>> = input_groups(memo, physical, v) - .iter() - .map(|input| logical[input.0 as usize].clone()) - .collect(); - usage[v.0] = Some(match inputs { + usage[v.0] = Some(match logical_inputs(memo, physical, &logical, v) { Some(inputs) => cost_node( physical, v, &nodes[v.0], &inputs, - entries, + 1.0, data, cost_inputs, formula_scale, @@ -277,45 +348,163 @@ pub(crate) fn selection_costs( .filter_map(|v| usage[v.0].as_ref()?.as_ref().ok()), ); } - let candidate: Vec> = usage + // Each execution's (own, raw data) cost per second at its rate, over + // its input groups' logical outputs. + let at_rate: Vec> = contexts + .executions() .iter() - .map(|usage| match usage { - Some(Ok(usage)) => cost_per_second(usage, cost_inputs), - Some(Err(error)) => Err(error.clone()), - None => Err("it reads a group on a cycle of candidates".to_string()), + .map(|&execution| { + let v = execution.candidate; + if let Some(Err(error)) = &usage[v.0] { + return Err(error.clone()); + } + let per_second = executions_per_second( + physical, + entries, + contexts, + execution, + cost_inputs.non_repeating_query_horizon_seconds, + )?; + let inputs = logical_inputs(memo, physical, &logical, v) + .ok_or("it reads a group on a cycle of candidates")?; + let (own, raw_data) = cost(v, &nodes[v.0], &inputs, per_second)?; + Ok((own, raw_data, per_second)) + }) + .collect(); + // π_v of a candidate retaining rows every consumer reads within a bound: + // what one more Δ ms of retention costs, over Δ (a window of a chain, + // else a second), at its retention's largest; its whole own cost when + // it is materialized, its raw data otherwise (each of its runs reads + // what it reads at that retention, ρ_κ c_{κ,v}). + // The one execution of each materialized candidate, by position. + let own_execution: HashMap = contexts + .executions() + .iter() + .enumerate() + .filter(|(_, execution)| { + matches!( + physical.physical_candidates[execution.candidate.0] + .physical_properties + .lifetime(), + Lifetime::AcrossEvaluations(_) + ) }) + .map(|(index, execution)| (execution.candidate, index)) .collect(); - let mut edge = HashMap::new(); - for &id in selectable { - let e = physical.edge(id); - let (u, c) = (e.producer_candidate_id, e.consumer_candidate_id); - let (Some(Ok(produced)), Ok(own)) = (&usage[u.0], &candidate[c.0]) else { + let mut retention_cost_per_second = HashMap::new(); + for (&v, &most_read_ms) in &most_read_ms { + let Some(retention_ms) = most_read_ms else { continue; }; - let Some(mut inputs) = input_groups(memo, physical, c) - .iter() - .map(|input| logical[input.0 as usize].clone()) - .collect::>>() - else { - continue; + let candidate = &physical.physical_candidates[v.0]; + let materialized = matches!( + candidate.physical_properties.lifetime(), + Lifetime::AcrossEvaluations(_) + ); + let delta_ms = candidate.chain_width_ms(physical).unwrap_or(1_000).max(1); + let longer = node_retaining(v, Some(retention_ms + delta_ms)); + let per_ms = match materialized { + true => own_execution.get(&v).and_then(|&index| { + let (own, raw_data, per_second) = at_rate[index].as_ref().ok()?; + let inputs = logical_inputs(memo, physical, &logical, v)?; + let (longer_own, longer_raw_data) = cost(v, &longer, &inputs, *per_second).ok()?; + Some((longer_own + longer_raw_data - own - raw_data) / delta_ms as f64) + }), + false => usage[v.0] + .as_ref() + .and_then(|usage| usage.as_ref().ok()) + .and_then(|_| { + let inputs = logical_inputs(memo, physical, &logical, v)?; + let (_, raw_data) = cost(v, &nodes[v.0], &inputs, 1.0).ok()?; + let (_, longer_raw_data) = cost(v, &longer, &inputs, 1.0).ok()?; + Some((longer_raw_data - raw_data) / delta_ms as f64) + }), }; - inputs[e.input_index_of_consumer] = producer_output(&nodes[u.0], produced); - let with_u = cost_node( - physical, - c, - &nodes[c.0], - &inputs, - entries, - data, - cost_inputs, - formula_scale, - ) - .and_then(|usage| cost_per_second(&usage, cost_inputs)); - if let Ok(with_u) = with_u { - edge.insert(id, with_u - own); + if let Some(per_ms) = per_ms { + retention_cost_per_second.insert(v, per_ms.max(0.0)); } } - SelectionCosts { candidate, edge } + let per_execution: Vec> = contexts + .executions() + .iter() + .zip(&at_rate) + .map(|(execution, at_rate)| { + let &(own, raw_data, _) = at_rate.as_ref().map_err(Clone::clone)?; + let v = execution.candidate; + let materialized = matches!( + physical.physical_candidates[v.0] + .physical_properties + .lifetime(), + Lifetime::AcrossEvaluations(_) + ); + Ok(match (retention_cost_per_second.get(&v), materialized) { + (None, _) => own + raw_data, + (Some(per_ms), true) => { + let retention_ms = most_read_ms[&v].unwrap_or(0) as f64; + own + raw_data - per_ms * retention_ms + } + (Some(_), false) => own, + }) + }) + .collect(); + let per_edge = contexts + .edges() + .iter() + .map(|&(consumer, edge, producer)| { + let e = physical.edge(edge); + let (u, c) = (e.producer_candidate_id, e.consumer_candidate_id); + let consumer_index = contexts.index_of(consumer); + per_execution[contexts.index_of(producer)].as_ref().ok()?; + let &(own, raw_data, per_second) = at_rate[consumer_index].as_ref().ok()?; + let produced = usage[u.0].as_ref()?.as_ref().ok()?; + let mut inputs = logical_inputs(memo, physical, &logical, c)?; + inputs[e.input_index_of_consumer] = producer_output(&nodes[u.0], produced); + let (with_u_own, with_u_raw_data) = cost(c, &nodes[c.0], &inputs, per_second).ok()?; + Some(with_u_own + with_u_raw_data - own - raw_data) + }) + .collect(); + ExecutionCosts { + per_execution, + per_edge, + retention_cost_per_second, + } +} + +/// $\operatorname{need}_{\kappa,u,c,i}$ of `edge`: how far back its +/// producer's rows are read through it ([`retention_need_ms`] over the +/// admissible edges above its consumer); `None`, unbounded. +pub(crate) fn edge_retention_need_ms( + physical: &PhysicalASAPMemoCandidates, + entries: &[QueryWorkloadEntry], + edge: PhysicalEdgeId, +) -> Option { + retention_need_ms( + physical, + entries, + physical.edge(edge).producer_candidate_id, + &[edge], + &|consumer| physical.admissible_edges_out_of(consumer).to_vec(), + ) +} + +/// How far back a query reads the output it reads directly: its +/// range-evaluation span (0 without one). +pub(crate) fn output_retention_need_ms(entry: &QueryWorkloadEntry) -> u64 { + entry.range_evaluation.map_or(0, |range| range.span_ms) +} + +/// The logical outputs of candidate `v`'s input groups, by input index; +/// `None` when one has no costed candidate. +fn logical_inputs<'a>( + memo: &Memo, + physical: &PhysicalASAPMemoCandidates, + logical: &[Option>], + v: PhysicalCandidateId, +) -> Option>> { + input_groups(memo, physical, v) + .iter() + .map(|input| logical[input.0 as usize].clone()) + .collect() } /// The input groups of candidate `v`, by input index. @@ -381,55 +570,27 @@ fn logical_output<'a: 'b, 'b>( output } -/// Candidate `v`'s usage as `node` over `inputs`, at its rate (module doc). +/// Candidate `v`'s usage as `node` over `inputs`, computed `per_second` +/// times per second. #[allow(clippy::too_many_arguments)] fn cost_node<'a>( physical: &PhysicalASAPMemoCandidates, v: PhysicalCandidateId, node: &'a PhysicalASAPDAGNode, inputs: &[ProducerOutput<'a>], - entries: &[QueryWorkloadEntry], + per_second: f64, data: &DataWorkload, cost_inputs: &'a CostInputs, formula_scale: f64, ) -> Result, String> { let memo = &physical.logical_candidates.memo; let candidate = &physical.physical_candidates[v.0]; - let group = candidate.logical_candidate_id.group; let ingestion = candidate.physical_properties.timing() == ExecutionTiming::IngestionTime; - let per_second = match ingestion { - true => 1.0, - false => { - let per_request = matches!( - candidate.physical_properties.lifetime(), - Lifetime::AcrossEvaluations(_) - ); - evaluation_rate( - memo.groups()[group.0 as usize] - .reached_by_queries() - .iter() - .filter_map(|&query| entries.get(query)) - .map(|entry| { - ( - &entry.recurrence, - (!per_request).then_some(entry.range_evaluation).flatten(), - ) - }), - cost_inputs.non_repeating_query_horizon_seconds, - ) - .map_err(|error| error.to_string())? - } - }; - let ingestion_window_ms = match (&node.payload, ingestion) { - (_, false) => None, - ( - Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { - tumbling_window_width_ms, - .. - }), - true, - ) => Some(*tumbling_window_width_ms), - (_, true) => window_ms_below(memo, group), + let ingestion_window_ms = match ingestion { + false => None, + true => candidate + .chain_width_ms(physical) + .or_else(|| window_ms_below(memo, candidate.logical_candidate_id.group)), }; node_usage( node, @@ -443,9 +604,9 @@ fn cost_node<'a>( .map_err(|error| error.to_string()) } -/// $w_v$ from a candidate's usage: its cost per second under cost model 1 -/// plus the raw data a query-time scan retains on disk. -fn cost_per_second(usage: &NodeUsage<'_>, cost_inputs: &CostInputs) -> Result { +/// A candidate's cost per second from its usage under cost model 1: its +/// own, and the raw data a scan retains on disk. +fn cost_per_second(usage: &NodeUsage<'_>, cost_inputs: &CostInputs) -> Result<(f64, f64), String> { let CostModel::UsageOverTime(unit_cost) = &cost_inputs.cost_model else { return Err("the selection MILP costs under cost model 1 only".into()); }; @@ -456,11 +617,13 @@ fn cost_per_second(usage: &NodeUsage<'_>, cost_inputs: &CostInputs) -> Result= 0.0 { - true => Ok(cost), + match [own, raw_data] + .iter() + .all(|cost| cost.is_finite() && *cost >= 0.0) + { + true => Ok((own, raw_data)), false => Err(format!( - "its cost per second {cost} is not a finite non-negative number" + "its cost per second {own} + {raw_data} is not a finite non-negative number" )), } } diff --git a/crates/stage3-plan-selection/src/lib.rs b/crates/stage3-plan-selection/src/lib.rs index 81702d51f..8eae208dd 100644 --- a/crates/stage3-plan-selection/src/lib.rs +++ b/crates/stage3-plan-selection/src/lib.rs @@ -3,15 +3,17 @@ //! selection and realization", and sections 4 and 5). //! //! [`select_plan`] reads Stage 2's physical candidates $\mathcal C$ and the -//! admissible part of the physical edges $\mathcal E$, selects with a MILP -//! over $x_v$, $y_{u,c,i}$, $z_{q,v}$ and $t_v$ (constraints (A)–(I); -//! (G) and (H) hold because $y$ exists only on admissible edges), and -//! builds the plan (`PhysicalASAPDAG`) from $\{x = 1\}$ and $\{y = 1\}$ -//! as its last step. The extensions add execution contexts $\kappa$ with $x^{\mathrm{mat}}_v$, -//! $x_{\kappa,v}$, $y_{\kappa,u,c,i}$ and the retention amount $h_v$ -//! (section 5), and accuracy at each query root, constraint (K), with -//! per-candidate error bounds $E_c$ and $D_c$ (section 4). Costs come from -//! `asap-cost`. +//! admissible part of the physical edges $\mathcal E$, and selects with a +//! MILP over execution contexts $\kappa$ (section 5): $x_{\kappa,v}$ per +//! execution of a candidate in a context (an `AcrossEvaluations` candidate +//! runs once, $x^{\mathrm{mat}}_v$), $y_{\kappa,u,c,i}$, $z_{q,v}$, $t$ and the +//! retention $h_v$, under constraints (A)–(I) per context and (R) ((G) and +//! (H) hold because $y$ exists only on admissible edges whose producer +//! delivers on the consumer's context). It builds the plan +//! (`PhysicalASAPDAG`) from the selected executions and edges as its last +//! step. Accuracy at each query root, constraint (K), with per-candidate +//! error bounds $E_c$ and $D_c$ (section 4), is the remaining extension. +//! Costs come from `asap-cost`. mod cost; mod milp; diff --git a/crates/stage3-plan-selection/src/milp/execution_contexts.rs b/crates/stage3-plan-selection/src/milp/execution_contexts.rs new file mode 100644 index 000000000..24518487b --- /dev/null +++ b/crates/stage3-plan-selection/src/milp/execution_contexts.rs @@ -0,0 +1,334 @@ +//! Execution contexts $\kappa$ (concepts doc, section 5, "Execution +//! contexts, by example"): the distinct contexts of a workload, every +//! execution of a candidate the selection may choose, and the edges +//! $y_{\kappa,u,c,i}$ between executions. +//! +//! - A query is evaluated in the evaluation context of its grid +//! ([`evaluation_grid`]); queries on one grid share it, a range query's +//! steps included. +//! - A candidate with $\ell$ = None or `OneEvaluation` runs once per context +//! that selects it ($x_{\kappa,v}$), the context of its consumer. +//! - A candidate with $\ell$ = `AcrossEvaluations` runs once, in its own +//! context `Materialized(v)` ($x^{\mathrm{mat}}_v = x_{\mathrm{Materialized}(v),v}$). +//! Its inputs run in that context at query time, and in the ingestion +//! pipeline of its refresh window at ingestion time, which the other +//! ingestion-time builds of that window share. +//! - The input of a `TimeShift` anchored with `@ end()` or `@ start()` runs +//! on the grid of the request end (start) times: once per request of a +//! range query, as an instant query with its recurrence is evaluated. +//! - An input of a consumer in $\kappa$ reads a producer, over an admissible +//! edge, only where the producer delivers on $\kappa$ +//! (`PhysicalCandidate::delivers_on`). + +use std::collections::BTreeSet; + +use asap_types::ir::materialized_view::RefreshWindow; +use asap_types::ir::operator::AtModifier; +use asap_types::ir::{NonASAPOp, Operator}; +use asap_types::physical::{ExecutionTiming, Lifetime}; +use asap_types::stages::{ + producer_satisfies_physical_requirements, EvaluationGrid, ExecutionContext, ExecutionContextId, + MemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidateId, PhysicalEdgeId, +}; +use asap_types::workload::{greatest_common_divisor, QueryWorkloadEntry}; + +use crate::cost::input_count; +use crate::milp::model::QueryId; + +/// One run of candidate `candidate` in context `context`: a node of the +/// plan. $x^{\mathrm{mat}}_v$ is the execution of $v$ in +/// `Materialized(v)`, $x_{\kappa,v}$ that of $v$ in $\kappa$. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub(crate) struct Execution { + pub(crate) context: ExecutionContextId, + pub(crate) candidate: PhysicalCandidateId, +} + +/// The distinct contexts of a workload and every execution and edge the +/// selection may choose ([`Self::new`]). +pub(crate) struct ExecutionContexts { + /// The distinct contexts, by [`ExecutionContextId`]. + contexts: Vec, + /// Every execution reached from a query output, ascending. + executions: Vec, + /// Every $y_{\kappa,u,c,i}$ as (consumer execution, edge $(u,c,i)$, + /// producer execution), ascending; $\kappa$ is the consumer's context. + edges: Vec<(Execution, PhysicalEdgeId, Execution)>, +} + +impl ExecutionContexts { + /// The contexts, executions and edges of `physical` for the queries + /// `entries` (by query index), reached from `outputs` (each query output + /// reference, a query and the group it reads), and per output the + /// executions that may answer it: a candidate of its group a query-time + /// consumer may read (`producer_satisfies_physical_requirements`) that + /// delivers on the query's context. + pub(crate) fn new( + physical: &PhysicalASAPMemoCandidates, + entries: &[QueryWorkloadEntry], + outputs: &[(QueryId, MemoGroupId)], + ) -> (Self, Vec<(QueryId, Vec)>) { + let mut contexts = Self { + contexts: Vec::new(), + executions: Vec::new(), + edges: Vec::new(), + }; + let mut reached: BTreeSet = BTreeSet::new(); + let mut pending: Vec = Vec::new(); + let mut output_executions = Vec::with_capacity(outputs.len()); + for &(query, group) in outputs { + let context = contexts.context_id(ExecutionContext::Evaluation { + grid: evaluation_grid(&entries[query.0]), + }); + let mut answering = Vec::new(); + for (v, candidate) in physical.physical_candidates.iter().enumerate() { + if candidate.logical_candidate_id.group != group + || !producer_satisfies_physical_requirements( + candidate, + ExecutionTiming::QueryTime, + ) + { + continue; + } + if let Some(execution) = + contexts.execution_reading(physical, context, PhysicalCandidateId(v)) + { + answering.push(execution); + if reached.insert(execution) { + pending.push(execution); + } + } + } + output_executions.push((query, answering)); + } + let mut edges = BTreeSet::new(); + while let Some(consumer) = pending.pop() { + let c = consumer.candidate; + let context = contexts.input_context(physical, entries, consumer); + for input in 0..input_count(physical, c) { + for &edge in physical.admissible_edges_into(c, input) { + let u = physical.edge(edge).producer_candidate_id; + if let Some(producer) = contexts.execution_reading(physical, context, u) { + edges.insert((consumer, edge, producer)); + if reached.insert(producer) { + pending.push(producer); + } + } + } + } + } + contexts.executions = reached.into_iter().collect(); + contexts.edges = edges.into_iter().collect(); + (contexts, output_executions) + } + + /// The id of `context`, added to the list when new. + fn context_id(&mut self, context: ExecutionContext) -> ExecutionContextId { + match self.contexts.iter().position(|known| *known == context) { + Some(index) => ExecutionContextId(index), + None => { + self.contexts.push(context); + ExecutionContextId(self.contexts.len() - 1) + } + } + } + + /// The execution of producer `u` a consumer in `context` reads, when + /// `u` delivers on it: its own for an `AcrossEvaluations` producer, + /// otherwise its run in `context`. + fn execution_reading( + &mut self, + physical: &PhysicalASAPMemoCandidates, + context: ExecutionContextId, + u: PhysicalCandidateId, + ) -> Option { + let producer = &physical.physical_candidates[u.0]; + if !producer.delivers_on(physical, &self.contexts[context.0]) { + return None; + } + let context = match producer.physical_properties.lifetime() { + Lifetime::AcrossEvaluations(_) => self.context_id(ExecutionContext::Materialized(u)), + Lifetime::None | Lifetime::OneEvaluation(_) => context, + }; + Some(Execution { + context, + candidate: u, + }) + } + + /// The context the inputs of `consumer` run in: an `AcrossEvaluations` + /// candidate's at ingestion time, the ingestion pipeline of its refresh + /// window (its chain width, else its group's evaluation grid); + /// a `TimeShift` anchored at `@ end()` or `@ start()` in an evaluation + /// context, the grid of its queries' request end or start times; else + /// the consumer's own context. + fn input_context( + &mut self, + physical: &PhysicalASAPMemoCandidates, + entries: &[QueryWorkloadEntry], + consumer: Execution, + ) -> ExecutionContextId { + let candidate = &physical.physical_candidates[consumer.candidate.0]; + let memo = &physical.logical_candidates.memo; + let group = candidate.logical_candidate_id.group; + let at_ingestion = candidate.physical_properties.timing() == ExecutionTiming::IngestionTime; + if at_ingestion + && matches!( + candidate.physical_properties.lifetime(), + Lifetime::AcrossEvaluations(_) + ) + { + let refresh_window = match candidate.chain_width_ms(physical) { + Some(width_ms) => Some(RefreshWindow { + width_ms, + phase_ms: 0, + }), + None => physical.logical_candidates.evaluation_grid_of_group[group.0 as usize].map( + |grid| RefreshWindow { + width_ms: grid.period_ms, + phase_ms: grid.phase_ms, + }, + ), + }; + return match refresh_window { + Some(window) => self.context_id(ExecutionContext::IngestionPipeline(window)), + None => consumer.context, + }; + } + let at = match memo + .candidate(candidate.logical_candidate_id) + .expression + .operator + { + Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }) => shift.at, + _ => None, + }; + let ( + Some(at @ (AtModifier::End | AtModifier::Start)), + ExecutionContext::Evaluation { grid }, + ) = (at, self.contexts[consumer.context.0]) + else { + return consumer.context; + }; + let period_ms = memo.groups()[group.0 as usize] + .reached_by_queries() + .iter() + .filter_map(|&query| entries.get(query)) + .filter(|entry| evaluation_grid(entry) == grid) + .fold(0, |period_ms, entry| { + greatest_common_divisor(period_ms, request_time_grid_ms(entry, at)) + }); + match period_ms { + 0 => consumer.context, + period_ms => self.context_id(ExecutionContext::Evaluation { + grid: EvaluationGrid { + period_ms, + phase_ms: 0, + }, + }), + } + } + + /// Every execution reached from a query output, ascending. + pub(crate) fn executions(&self) -> &[Execution] { + &self.executions + } + + /// Every $y_{\kappa,u,c,i}$ as (consumer execution, edge, producer + /// execution), ascending. + pub(crate) fn edges(&self) -> &[(Execution, PhysicalEdgeId, Execution)] { + &self.edges + } + + /// The position of `execution` in [`Self::executions`]. + pub(crate) fn index_of(&self, execution: Execution) -> usize { + self.executions + .binary_search(&execution) + .expect("a reached execution") + } +} + +/// The evaluation grid of `entry`'s query, the grid of its context: the +/// times its evaluations fall on (`QueryRecurrence::evaluation_grid_ms` of +/// its range evaluation), anchored at the epoch. Times that are not known +/// may fall on any millisecond: a 1 ms grid, on which only producers stored +/// at every time deliver. +pub(crate) fn evaluation_grid(entry: &QueryWorkloadEntry) -> EvaluationGrid { + EvaluationGrid { + period_ms: entry + .recurrence + .evaluation_grid_ms(entry.range_evaluation) + .unwrap_or(1), + phase_ms: 0, + } +} + +/// The grid, in ms, of the times `entry`'s requests anchor `@ end()` or +/// `@ start()` at: its request end times, shifted back by its +/// range-evaluation span for `@ start()`; 1 ms when not known. +fn request_time_grid_ms(entry: &QueryWorkloadEntry, at: AtModifier) -> u64 { + let end_ms = entry.recurrence.evaluation_grid_ms(None).unwrap_or(1); + match (at, entry.range_evaluation) { + (AtModifier::Start, Some(range)) => greatest_common_divisor(end_ms, range.span_ms), + _ => end_ms, + } +} + +/// $\rho$ of `execution`'s context, as how often it runs, from the queries +/// it serves: an ingestion-time execution is costed over one second of +/// arriving rows (1); in an evaluation context, the combined evaluation +/// rate (`asap_cost::execution_demand::evaluation_rate`) of the queries on +/// its grid that reach its group, with their range evaluations, and of +/// the range queries whose request end (start) times are the grid, once per +/// request (an input of a `TimeShift` anchored with `@`); in +/// `Materialized(m)`, one +/// build per window of a query-time chain $m$, else once per request of +/// the queries reaching $m$'s group. +pub(crate) fn executions_per_second( + physical: &PhysicalASAPMemoCandidates, + entries: &[QueryWorkloadEntry], + contexts: &ExecutionContexts, + execution: Execution, + non_repeating_query_horizon_seconds: f64, +) -> Result { + use asap_cost::execution_demand::evaluation_rate; + let candidate = &physical.physical_candidates[execution.candidate.0]; + if candidate.physical_properties.timing() == ExecutionTiming::IngestionTime { + return Ok(1.0); + } + let memo = &physical.logical_candidates.memo; + let reaching = |v: PhysicalCandidateId| { + let group = physical.physical_candidates[v.0].logical_candidate_id.group; + memo.groups()[group.0 as usize] + .reached_by_queries() + .iter() + .filter_map(|&query| entries.get(query)) + }; + let rate = match contexts.contexts[execution.context.0] { + ExecutionContext::Evaluation { grid } => evaluation_rate( + reaching(execution.candidate).filter_map(|entry| { + if evaluation_grid(entry) == grid { + return Some((&entry.recurrence, entry.range_evaluation)); + } + let request_time = [AtModifier::End, AtModifier::Start] + .into_iter() + .any(|at| grid.period_ms == request_time_grid_ms(entry, at)); + (entry.range_evaluation.is_some() && request_time) + .then_some((&entry.recurrence, None)) + }), + non_repeating_query_horizon_seconds, + ), + ExecutionContext::Materialized(m) => { + match physical.physical_candidates[m.0].chain_width_ms(physical) { + Some(width_ms) => Ok(1_000.0 / width_ms.max(1) as f64), + None => evaluation_rate( + reaching(m).map(|entry| (&entry.recurrence, None)), + non_repeating_query_horizon_seconds, + ), + } + } + // Only ingestion-time candidates run in an ingestion pipeline. + ExecutionContext::IngestionPipeline(_) => Ok(1.0), + }; + rate.map_err(|error| error.to_string()) +} diff --git a/crates/stage3-plan-selection/src/milp/mod.rs b/crates/stage3-plan-selection/src/milp/mod.rs index 9a2d7f7e3..8a535ee7b 100644 --- a/crates/stage3-plan-selection/src/milp/mod.rs +++ b/crates/stage3-plan-selection/src/milp/mod.rs @@ -1,3 +1,4 @@ //! Stage 3's selection MILP (concepts doc, sections 3–5): its variables, //! one function per constraint family, and the execution contexts. +pub(crate) mod execution_contexts; pub(crate) mod model; diff --git a/crates/stage3-plan-selection/src/milp/model.rs b/crates/stage3-plan-selection/src/milp/model.rs index 0a4f687d2..30c7bad66 100644 --- a/crates/stage3-plan-selection/src/milp/model.rs +++ b/crates/stage3-plan-selection/src/milp/model.rs @@ -3,12 +3,11 @@ //! //! | Family | Variables | Constraints | //! |---|---|---| -//! | baseline (section 3) | $x_v$, $y_{u,c,i}$, $z_{q,v}$, $t_v$ ([`SelectionVariables`]) | (A) [`input_satisfaction_constraints`], (B) [`producer_selection_constraints`], (C)/(D) [`query_coverage_constraints`], (E) [`acyclicity_constraints`], (F) [`utilization_constraints`], one consumer of a query-time result that is not materialized [`unshared_result_constraints`]; (G), (H): $y$ exists only on admissible edges | -//! | execution contexts (section 5) | $x^{\mathrm{mat}}_v$, $x_{\kappa,v}$, $y_{\kappa,u,c,i}$, $h_v$ ([`ExecutionContextVariables`]) | (R) [`retention_constraints`] | +//! | baseline (section 3) over execution contexts (section 5) | $x_{\kappa,v}$ (with $x^{\mathrm{mat}}_v = x_{\mathrm{Materialized}(v),v}$), $y_{\kappa,u,c,i}$, $z_{q,v}$, $t$, $h_v$ ([`SelectionVariables`]) | (A) [`input_satisfaction_constraints`], (B) [`producer_selection_constraints`], (C)/(D) [`query_coverage_constraints`], (E) [`acyclicity_constraints`], (F) [`utilization_constraints`], one consumer per context of a query-time result that is not materialized [`unshared_result_constraints`], (R) [`retention_constraints`]; (G), (H): $y$ exists only on admissible edges where the producer delivers on the consumer's context | //! | accuracy (section 4) | $E_{c,m}$, $F_{c,i,m}$, $D_c$, $H_{c,i}$ ([`AccuracyVariables`]) | (K) [`accuracy_constraints`] | //! -//! The baseline is solved by [`solve_selection`]; the extensions' bodies -//! are `todo!()` until #1207 PRs 7 (contexts) and 8 (accuracy). +//! [`solve_selection`] solves it without (K), whose body is `todo!()` until +//! #1207 PR 8. use std::collections::HashMap; use std::time::Instant; @@ -20,137 +19,187 @@ use good_lp::{ use asap_types::ir::properties::AccuracyMetric; use asap_types::physical::{ExecutionTiming, Lifetime}; -use asap_types::stages::{ - ExecutionContextId, PhysicalASAPMemoCandidates, PhysicalCandidateId, PhysicalEdgeId, -}; +use asap_types::stages::{PhysicalASAPMemoCandidates, PhysicalCandidateId, PhysicalEdgeId}; +use asap_types::workload::QueryWorkloadEntry; -use crate::cost::{input_count, SelectionCosts}; +use crate::cost::{edge_retention_need_ms, output_retention_need_ms, ExecutionCosts}; +use crate::milp::execution_contexts::{Execution, ExecutionContexts}; /// A query, by index in the workload. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct QueryId(pub usize); -/// What the baseline MILP is built over. +/// What the MILP is built over. pub(crate) struct SelectionInputs<'a> { pub(crate) physical: &'a PhysicalASAPMemoCandidates, - /// $w_v$ and $c_{u,c,i}$: $y$ exists only on the edges - /// [`SelectionCosts::edge`] costs, $x_v$ is 0 for a candidate it could - /// not cost. - pub(crate) costs: &'a SelectionCosts, + /// Each query's workload entry, for the retention needs of (R). + pub(crate) entries: &'a [QueryWorkloadEntry], + /// The executions $x$ and edges $y$ are over. + pub(crate) contexts: &'a ExecutionContexts, + /// The objective's coefficients: $x$ is 0 for an execution it could + /// not cost, $y$ exists only on an edge it costs, $h_v$ only for a + /// candidate with a retention cost. + pub(crate) costs: &'a ExecutionCosts, /// Each query output reference (a query, or one operator a scalar query - /// reads), with the candidates that may answer it: $z_{q,v}$ exists + /// reads), with the executions that may answer it: $z_{q,v}$ exists /// only for them. - pub(crate) query_outputs: &'a [(QueryId, Vec)], + pub(crate) query_outputs: &'a [(QueryId, Vec)], } -/// The baseline variables (section 3). +/// The variables (sections 3 and 5). pub(crate) struct SelectionVariables { - /// $x_v$, by [`PhysicalCandidateId`]. + /// $x_{\kappa,v}$, by position in `ExecutionContexts::executions`. x: Vec, - /// $y_{u,c,i}$, on admissible edges only. - y: HashMap, - /// $z_{q,v}$, for $v \in G_q$. - z: HashMap<(QueryId, PhysicalCandidateId), Variable>, - /// $t_v$, by [`PhysicalCandidateId`]: only for candidates on a cycle of - /// admissible edges (a strongly connected component); `None` elsewhere. + /// $y_{\kappa,u,c,i}$, by position in `ExecutionContexts::edges`; `None` + /// on an edge that could not be costed. + y: Vec>, + /// $z_{q,v}$, for $v \in G_q$, by the execution answering $q$. + z: HashMap<(QueryId, Execution), Variable>, + /// $t$, by position in `ExecutionContexts::executions`: only for + /// executions on a cycle of selectable edges (a strongly connected + /// component); `None` elsewhere. t: Vec>, + /// $h_v$ in seconds, for a candidate with a retention cost. + h: HashMap, } -/// $x$, $y$, $z$ and $t$ over `inputs`, added to `problem`. +/// $x$, $y$, $z$, $t$ and $h$ over `inputs`, added to `problem`. fn selection_variables( inputs: &SelectionInputs<'_>, problem: &mut ProblemVariables, ) -> SelectionVariables { - let count = inputs.physical.physical_candidates.len(); - let x = (0..count) - .map(|v| match inputs.costs.candidate[v].is_ok() { + let x = inputs + .costs + .per_execution + .iter() + .map(|cost| match cost.is_ok() { true => problem.add(variable().binary()), false => problem.add(variable().min(0).max(0)), }) .collect(); - let mut edges: Vec = inputs.costs.edge.keys().copied().collect(); - edges.sort_unstable(); - let y = edges - .into_iter() - .map(|edge| (edge, problem.add(variable().binary()))) + let y = inputs + .costs + .per_edge + .iter() + .map(|cost| cost.map(|_| problem.add(variable().binary()))) .collect(); let mut z = HashMap::new(); - for (query, candidates) in inputs.query_outputs { - for &v in candidates { - z.entry((*query, v)) + for (query, executions) in inputs.query_outputs { + for &execution in executions { + z.entry((*query, execution)) .or_insert_with(|| problem.add(variable().binary())); } } - // t_v only inside a strongly connected component of the selectable + // t only inside a strongly connected component of the selectable // edges, bounded by its size. let t = component_sizes(inputs) .into_iter() .map(|size| (size > 1).then(|| problem.add(variable().min(0).max((size - 1) as f64)))) .collect(); - SelectionVariables { x, y, z, t } + let mut retaining: Vec = inputs + .costs + .retention_cost_per_second + .keys() + .copied() + .collect(); + retaining.sort_unstable(); + let h = retaining + .into_iter() + .map(|v| (v, problem.add(variable().min(0)))) + .collect(); + SelectionVariables { x, y, z, t, h } +} + +/// The selectable edges ($y$ exists), by position in +/// `ExecutionContexts::edges`, with their (consumer, edge, producer). +fn selectable<'a>( + inputs: &'a SelectionInputs<'_>, + variables: &'a SelectionVariables, +) -> impl Iterator + 'a { + inputs + .contexts + .edges() + .iter() + .zip(&variables.y) + .enumerate() + .filter_map(|(k, (&edge, y))| Some((k, edge, (*y)?))) } -/// (A) $\sum_{u \in P_{c,i}} y_{u,c,i} = x_c$. +/// (A) $\sum_{u \in P_{c,i}} y_{\kappa,u,c,i} = x_{\kappa,c}$, per execution +/// of $c$ and input $i$. fn input_satisfaction_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { let physical = inputs.physical; + let mut producers: HashMap<(usize, usize), Expression> = HashMap::new(); + for (_, (consumer, edge, _), y) in selectable(inputs, variables) { + *producers + .entry(( + inputs.contexts.index_of(consumer), + physical.edge(edge).input_index_of_consumer, + )) + .or_default() += y; + } let mut constraints = Vec::new(); - for c in (0..physical.physical_candidates.len()).map(PhysicalCandidateId) { - for i in 0..input_count(physical, c) { - let producers: Expression = physical - .admissible_edges_into(c, i) - .iter() - .filter_map(|edge| variables.y.get(edge)) - .sum(); - constraints.push(constraint!(producers == variables.x[c.0])); + for (index, execution) in inputs.contexts.executions().iter().enumerate() { + for i in 0..crate::cost::input_count(physical, execution.candidate) { + let chosen = producers.remove(&(index, i)).unwrap_or_default(); + constraints.push(constraint!(chosen == variables.x[index])); } } constraints } -/// (B) $y_{u,c,i} \le x_u$. +/// (B) $y_{\kappa,u,c,i} \le x$ of $u$'s execution it reads. fn producer_selection_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { - sorted(&variables.y) - .map(|(edge, y)| { - let producer = inputs.physical.edge(edge).producer_candidate_id; - constraint!(y <= variables.x[producer.0]) + selectable(inputs, variables) + .map(|(_, (_, _, producer), y)| { + constraint!(y <= variables.x[inputs.contexts.index_of(producer)]) }) .collect() } -/// (C) $\sum_{v \in G_q} z_{q,v} = 1$; (D) $z_{q,v} \le x_v$. +/// (C) $\sum_{v \in G_q} z_{q,v} = 1$; (D) $z_{q,v} \le x$ of $v$'s +/// execution answering $q$. fn query_coverage_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { let mut constraints = Vec::new(); - for (query, candidates) in inputs.query_outputs { - let covered: Expression = candidates.iter().map(|&v| variables.z[&(*query, v)]).sum(); + for (query, executions) in inputs.query_outputs { + let covered: Expression = executions + .iter() + .map(|&execution| variables.z[&(*query, execution)]) + .sum(); constraints.push(constraint!(covered == 1)); } - for (&(_, v), &z) in sorted_by_key(&variables.z) { - constraints.push(constraint!(z <= variables.x[v.0])); + for (&(_, execution), &z) in sorted_by_key(&variables.z) { + constraints.push(constraint!( + z <= variables.x[inputs.contexts.index_of(execution)] + )); } constraints } -/// (E) $t_u + 1 \le t_c + n(1 - y_{u,c,i})$, only on edges inside a strongly -/// connected component of the admissible edges, $n$ its size. +/// (E) $t_u + 1 \le t_c + n(1 - y_{\kappa,u,c,i})$ between the producer's +/// and the consumer's executions, only on edges inside a strongly connected +/// component of the selectable edges, $n$ its size. fn acyclicity_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { let components = selectable_components(inputs); let sizes = component_sizes(inputs); - sorted(&variables.y) - .filter_map(|(edge, y)| { - let edge = inputs.physical.edge(edge); - let (u, c) = (edge.producer_candidate_id.0, edge.consumer_candidate_id.0); + selectable(inputs, variables) + .filter_map(|(_, (consumer, _, producer), y)| { + let (u, c) = ( + inputs.contexts.index_of(producer), + inputs.contexts.index_of(consumer), + ); if components[u] != components[c] { return None; } @@ -161,90 +210,116 @@ fn acyclicity_constraints( .collect() } -/// (F) $x_v \le \sum_{(v,c,i)} y_{v,c,i} + \sum_q z_{q,v}$. +/// (F) $x_{\kappa,v} \le \sum y + \sum_q z_{q,v}$ over the edges and query +/// outputs reading that execution. fn utilization_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { - let uses = uses_of(inputs, variables); - uses.into_iter() + uses_of(inputs, variables) + .into_iter() .enumerate() - .map(|(v, used)| constraint!(variables.x[v] <= used)) + .map(|(index, used)| constraint!(variables.x[index] <= used)) .collect() } /// A query-time result that is not materialized ($\ell$ = None) has at most -/// one consumer: $\sum_{(v,c,i)} y_{v,c,i} + \sum_q z_{q,v} \le x_v$. -/// Consumers that share a result read a `OneEvaluation` or a materialized -/// producer (section 5, per context in #1207 PR 7). An ingestion-time -/// producer is exempt: there is no `OneEvaluation` at ingestion time, and -/// the consumers of one ingestion run share its rows. +/// one consumer per context: $\sum y + \sum_q z_{q,v} \le x_{\kappa,v}$. +/// Consumers in one context that share a result read a `OneEvaluation` or +/// a materialized producer (section 5). An ingestion-time producer is +/// exempt: there is no `OneEvaluation` at ingestion time, and the +/// consumers of one ingestion run share its rows. fn unshared_result_constraints( inputs: &SelectionInputs<'_>, variables: &SelectionVariables, ) -> Vec { - let uses = uses_of(inputs, variables); - uses.into_iter() + uses_of(inputs, variables) + .into_iter() .enumerate() - .filter(|(v, _)| { - let properties = inputs.physical.physical_candidates[*v].physical_properties; + .filter(|(index, _)| { + let v = inputs.contexts.executions()[*index].candidate; + let properties = inputs.physical.physical_candidates[v.0].physical_properties; properties.lifetime() == Lifetime::None && properties.timing() == ExecutionTiming::QueryTime }) - .map(|(v, used)| constraint!(used <= variables.x[v])) + .map(|(index, used)| constraint!(used <= variables.x[index])) .collect() } -/// Per candidate, its selected edges to consumers and query outputs: -/// $\sum_{(v,c,i)} y_{v,c,i} + \sum_q z_{q,v}$. -fn uses_of(inputs: &SelectionInputs<'_>, variables: &SelectionVariables) -> Vec { - let mut uses: Vec = (0..inputs.physical.physical_candidates.len()) - .map(|v| { - inputs - .physical - .admissible_edges_out_of(PhysicalCandidateId(v)) - .iter() - .filter_map(|edge| variables.y.get(edge)) - .sum() - }) - .collect(); - for (&(_, v), &z) in sorted_by_key(&variables.z) { - uses[v.0] += z; +/// (R) $h_v \ge \operatorname{need}_{\kappa,v,c,i}\, y_{\kappa,v,c,i}$ and +/// $h_v \ge \operatorname{span}_q\, z_{q,v}$, for a candidate $v$ with $h_v$: +/// how far back its rows are read through each selected edge (the reading +/// path's offset plus range, `crate::cost::retention_need_ms`) and by each +/// query reading it directly (its range-evaluation span), a constant per +/// edge, so (R) is linear. Query latency is ignored: a result counts as read +/// when its evaluation starts. +fn retention_constraints( + inputs: &SelectionInputs<'_>, + variables: &SelectionVariables, +) -> Vec { + let mut constraints = Vec::new(); + for (_, (_, edge, producer), y) in selectable(inputs, variables) { + let Some(&h) = variables.h.get(&producer.candidate) else { + continue; + }; + let need_ms = edge_retention_need_ms(inputs.physical, inputs.entries, edge) + .expect("a candidate with h_v is read within a bound"); + constraints.push(constraint!(h >= seconds(need_ms) * y)); } - uses + for (&(query, execution), &z) in sorted_by_key(&variables.z) { + if let Some(&h) = variables.h.get(&execution.candidate) { + let need_ms = output_retention_need_ms(&inputs.entries[query.0]); + constraints.push(constraint!(h >= seconds(need_ms) * z)); + } + } + constraints } -/// `y` in edge order, so the model is the same in every run. -fn sorted( - y: &HashMap, -) -> impl Iterator { - let mut y: Vec<(PhysicalEdgeId, Variable)> = y.iter().map(|(&e, &v)| (e, v)).collect(); - y.sort_unstable_by_key(|(edge, _)| *edge); - y.into_iter() +/// `ms` in seconds, the unit of $h_v$ in the MILP. +fn seconds(ms: u64) -> f64 { + ms as f64 / 1_000.0 +} + +/// Per execution, its selected edges to consumers and query outputs: +/// $\sum y + \sum_q z_{q,v}$. +fn uses_of(inputs: &SelectionInputs<'_>, variables: &SelectionVariables) -> Vec { + let mut uses: Vec = vec![Expression::default(); variables.x.len()]; + for (_, (_, _, producer), y) in selectable(inputs, variables) { + uses[inputs.contexts.index_of(producer)] += y; + } + for (&(_, execution), &z) in sorted_by_key(&variables.z) { + uses[inputs.contexts.index_of(execution)] += z; + } + uses } -/// `z` in key order. +/// `z` in key order, so the model is the same in every run. fn sorted_by_key( - z: &HashMap<(QueryId, PhysicalCandidateId), Variable>, -) -> Vec<(&(QueryId, PhysicalCandidateId), &Variable)> { + z: &HashMap<(QueryId, Execution), Variable>, +) -> Vec<(&(QueryId, Execution), &Variable)> { let mut z: Vec<_> = z.iter().collect(); z.sort_unstable_by_key(|(key, _)| **key); z } -/// Each candidate's strongly connected component of the selectable edges. +/// Each execution's strongly connected component of the selectable edges. fn selectable_components(inputs: &SelectionInputs<'_>) -> Vec { + let contexts = inputs.contexts; strongly_connected_components( - inputs.physical.physical_candidates.len(), - inputs.costs.edge.keys().map(|&edge| { - let edge = inputs.physical.edge(edge); - (edge.producer_candidate_id.0, edge.consumer_candidate_id.0) - }), + contexts.executions().len(), + contexts + .edges() + .iter() + .zip(&inputs.costs.per_edge) + .filter(|(_, cost)| cost.is_some()) + .map(|(&(consumer, _, producer), _)| { + (contexts.index_of(producer), contexts.index_of(consumer)) + }), ) } -/// The size of each candidate's [`selectable_components`] component: 1 -/// for a candidate on no cycle. +/// The size of each execution's [`selectable_components`] component: 1 +/// for an execution on no cycle. fn component_sizes(inputs: &SelectionInputs<'_>) -> Vec { let components = selectable_components(inputs); let mut size: HashMap = HashMap::new(); @@ -304,37 +379,42 @@ fn strongly_connected_components( component } -/// The baseline MILP's optimum: $\{x = 1\}$, $\{y = 1\}$ and $\{z = 1\}$, -/// and the MILP's size and time. +/// The MILP's optimum: the executions with $x = 1$, the edges with +/// $y = 1$, the execution answering each query output ($z = 1$) and $h_v$ +/// of each selected candidate that has it, and the MILP's size and time. #[derive(Debug, Clone)] pub(crate) struct SolvedSelection { - /// The selected candidates, ascending. - pub(crate) candidates: Vec, - /// The selected edges, ascending. - pub(crate) edges: Vec, - /// The candidate answering each of `SelectionInputs::query_outputs`, in + /// The selected executions, ascending. + pub(crate) executions: Vec, + /// The selected edges as (consumer, edge, producer), ascending. + pub(crate) edges: Vec<(Execution, PhysicalEdgeId, Execution)>, + /// The execution answering each of `SelectionInputs::query_outputs`, in /// its order. - pub(crate) outputs: Vec, + pub(crate) outputs: Vec, + /// $h_v$ in ms of each selected candidate with $h_v$. + pub(crate) retention_ms: HashMap, pub(crate) variables: usize, pub(crate) constraints: usize, pub(crate) construction_ms: u64, pub(crate) solve_ms: u64, } -/// Why the baseline MILP has no optimum. +/// Why the MILP has no optimum. #[derive(Debug, Clone, PartialEq)] pub(crate) enum SelectionMilpError { - /// No plan meets constraints (A)–(F). + /// No plan meets the constraints. Infeasible, /// The solver failed. Solver(String), } -/// Builds and solves the baseline MILP: minimize -/// $\sum_v w_v x_v + \sum_{(u,c,i)} c_{u,c,i} y_{u,c,i}$ under (A)–(F) and -/// [`unshared_result_constraints`], with HiGHS at gap 0 on one thread, the -/// coefficients scaled so the largest is 1,000 (HiGHS's tolerances are -/// absolute). +/// Builds and solves the MILP: minimize +/// $\sum_v W^{\mathrm{mat}}_v x^{\mathrm{mat}}_v + \sum_{\kappa,v} +/// \rho_\kappa c_{\kappa,v} x_{\kappa,v} + \sum \rho_\kappa c_{\kappa,u,c,i} +/// y_{\kappa,u,c,i} + \sum_v \pi_v h_v$ under (A)–(F), +/// [`unshared_result_constraints`] and (R), with HiGHS at gap 0 on one +/// thread, the coefficients scaled so the largest is 1,000 (HiGHS's +/// tolerances are absolute). pub(crate) fn solve_selection( inputs: &SelectionInputs<'_>, ) -> Result { @@ -348,29 +428,46 @@ pub(crate) fn solve_selection( acyclicity_constraints(inputs, &variables), utilization_constraints(inputs, &variables), unshared_result_constraints(inputs, &variables), + retention_constraints(inputs, &variables), ] .into_iter() .flatten() .collect(); - let largest = inputs - .costs - .candidate + let costs = inputs.costs; + // π_v per second of retention, the unit of h. + let retention_cost = |v: &PhysicalCandidateId| costs.retention_cost_per_second[v] * 1_000.0; + let largest = costs + .per_execution .iter() .flatten() - .chain(inputs.costs.edge.values()) - .fold(0.0_f64, |largest, &cost| largest.max(cost.abs())); + .chain(costs.per_edge.iter().flatten()) + .copied() + .chain(costs.retention_cost_per_second.keys().map(retention_cost)) + .fold(0.0_f64, |largest, cost| largest.max(cost.abs())); let scale = match largest > 0.0 && largest.is_finite() { true => 1_000.0 / largest, false => 1.0, }; - let candidate_terms = inputs - .costs - .candidate + let execution_terms = costs + .per_execution .iter() .zip(&variables.x) .filter_map(|(cost, &x)| cost.as_ref().ok().map(|cost| cost * scale * x)); - let edge_terms = sorted(&variables.y).map(|(edge, y)| inputs.costs.edge[&edge] * scale * y); - let objective: Expression = candidate_terms.chain(edge_terms).sum(); + let edge_terms = costs + .per_edge + .iter() + .zip(&variables.y) + .filter_map(|(cost, y)| Some(cost.as_ref()? * scale * (*y)?)); + let mut retaining: Vec<(PhysicalCandidateId, Variable)> = + variables.h.iter().map(|(&v, &h)| (v, h)).collect(); + retaining.sort_unstable_by_key(|(v, _)| *v); + let retention_terms = retaining + .iter() + .map(|(v, h)| retention_cost(v) * scale * *h); + let objective: Expression = execution_terms + .chain(edge_terms) + .chain(retention_terms) + .sum(); let variable_count = problem.len(); // HiGHS's columns are the variables in the order they were added. let column_of: HashMap = problem @@ -410,27 +507,39 @@ pub(crate) fn solve_selection( } other => return Err(SelectionMilpError::Solver(format!("{other:?}"))), }; - let chosen = |variable: Variable| columns[column_of[&variable]] > 0.5; + let value = |variable: Variable| columns[column_of[&variable]]; + let chosen = |variable: Variable| value(variable) > 0.5; + let executions: Vec = inputs + .contexts + .executions() + .iter() + .zip(&variables.x) + .filter(|(_, &x)| chosen(x)) + .map(|(&execution, _)| execution) + .collect(); + let retention_ms = retaining + .iter() + .filter(|(v, _)| executions.iter().any(|execution| execution.candidate == *v)) + .map(|&(v, h)| (v, (value(h) * 1_000.0).round().max(0.0) as u64)) + .collect(); Ok(SolvedSelection { - candidates: (0..variables.x.len()) - .filter(|&v| chosen(variables.x[v])) - .map(PhysicalCandidateId) - .collect(), - edges: sorted(&variables.y) - .filter(|&(_, y)| chosen(y)) - .map(|(edge, _)| edge) + executions, + edges: selectable(inputs, &variables) + .filter(|&(_, _, y)| chosen(y)) + .map(|(_, edge, _)| edge) .collect(), outputs: inputs .query_outputs .iter() - .map(|(query, candidates)| { - candidates + .map(|(query, executions)| { + executions .iter() .copied() - .find(|&v| chosen(variables.z[&(*query, v)])) + .find(|&execution| chosen(variables.z[&(*query, execution)])) .expect("(C): one output per query output reference") }) .collect(), + retention_ms, variables: variable_count, constraints: constraint_count, construction_ms, @@ -438,20 +547,6 @@ pub(crate) fn solve_selection( }) } -/// The execution-context variables (section 5), which replace $x$ and the -/// unindexed $y$ of the baseline. -#[allow(dead_code)] // #1207 PR 7. -struct ExecutionContextVariables { - /// $x^{\mathrm{mat}}_v$, for an `AcrossEvaluations` candidate. - x_materialized: HashMap, - /// $x_{\kappa,v}$, for a candidate with $\ell$ = None or `OneEvaluation`. - x_in_execution_context: HashMap<(ExecutionContextId, PhysicalCandidateId), Variable>, - /// $y_{\kappa,u,c,i}$, indexed by the consumer's context. - y_in_execution_context: HashMap<(ExecutionContextId, PhysicalEdgeId), Variable>, - /// $h_v$, for an `AcrossEvaluations` producer. - retention_amount: HashMap, -} - /// The accuracy variables of constraint (K) (section 4), all continuous. #[allow(dead_code)] // #1207 PR 8. struct AccuracyVariables { @@ -466,20 +561,6 @@ struct AccuracyVariables { failure_probability_of_input: HashMap<(PhysicalCandidateId, usize), Variable>, } -/// (R) $h_v \ge \operatorname{need}_{\kappa,v,c,i}\, y_{\kappa,v,c,i}$: for a -/// time-indexed producer (raw data, a chain, results stored per evaluation -/// time) the reading path's offset plus range, a constant per edge, so (R) -/// is linear. Query latency is ignored: a result counts as read when its -/// evaluation starts. -#[allow(dead_code)] // #1207 PR 7. -fn retention_constraints( - physical: &PhysicalASAPMemoCandidates, - variables: &ExecutionContextVariables, -) -> Vec { - let _ = (physical, variables); - todo!("#1207 PR 7: constraint (R)") -} - /// (K): $E_{c,m}$ per candidate and metric and $D_c$ per candidate, big-M on /// selected edges, the metric rule on the selected path, and the root /// checks $E_{v,m_q} \le \varepsilon_q + M_q(1 - z_{q,v})$, diff --git a/crates/stage3-plan-selection/src/milp/model/tests.rs b/crates/stage3-plan-selection/src/milp/model/tests.rs index e87291292..212030051 100644 --- a/crates/stage3-plan-selection/src/milp/model/tests.rs +++ b/crates/stage3-plan-selection/src/milp/model/tests.rs @@ -1,12 +1,14 @@ -//! The baseline MILP over small hand-built candidate graphs with given -//! costs: which candidates and edges its optimum selects, and the plan -//! built from it. +//! The selection MILP over small hand-built candidate graphs with given +//! costs: which executions and edges its optimum selects, its retention, +//! and the plan built from it. -use std::collections::HashMap; use std::time::Duration; +use asap_types::ir::operator::operator_properties::Reduction; use asap_types::ir::operator::TimeShift; -use asap_types::ir::{NonASAPOp, Operator, QueryRoot, TimeRangeKind}; +use asap_types::ir::scalar::expr_ir::ColumnRef; +use asap_types::ir::schema::{ExactKind, ExactParams, FieldDataType, SummaryUpdate}; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator, QueryRoot, TimeRangeKind}; use asap_types::physical::{ExecutionTiming, Lifetime, PhysicalProperties, StorageMedium}; use asap_types::stages::{ CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, MemoGroupId, MemoQuery, @@ -16,13 +18,18 @@ use asap_types::stages::{ use asap_types::test_support::{ add_candidate, group_with_own_candidate, series_schema, time_series_scan, }; +use asap_types::workload::{ + Predictability, Query, QueryRecurrence, QueryRequirements, RepeatedDemand, RepetitionInterval, + TimestampMs, +}; use super::*; -use crate::cost::SelectionCosts; +use crate::cost::ExecutionCosts; use crate::plan::plan_of; const NONE: Lifetime = Lifetime::None; const ONE_EVALUATION: Lifetime = Lifetime::OneEvaluation(StorageMedium::Memory); +const ACROSS_EVALUATIONS: Lifetime = Lifetime::AcrossEvaluations(StorageMedium::Memory); fn scan(memo: &mut Memo) -> MemoGroupId { group_with_own_candidate(memo, time_series_scan("m", series_schema(&["job"]), vec![])) @@ -119,75 +126,119 @@ fn edge(physical: &PhysicalASAPMemoCandidates, u: usize, c: usize) -> PhysicalEd .0 } -/// Costs `w` per candidate and `edge_costs` per (producer, consumer), 0 for -/// every other edge. -fn costs( - physical: &PhysicalASAPMemoCandidates, - w: &[f64], - edge_costs: &[((usize, usize), f64)], -) -> SelectionCosts { - let mut edge_cost: HashMap = - physical.edges().map(|(id, _)| (id, 0.0)).collect(); - for &((u, c), cost) in edge_costs { - edge_cost.insert(edge(physical, u, c), cost); - } - SelectionCosts { - candidate: w.iter().map(|&w| Ok(w)).collect(), - edge: edge_cost, +/// Every query on a 10 s grid, `count` of them. +fn every_10s(count: usize) -> Vec { + (0..count).map(|_| every(10_000)).collect() +} + +/// A query every `interval_ms` at phase 0. +fn every(interval_ms: u32) -> QueryWorkloadEntry { + QueryWorkloadEntry { + query: Query("q".into()), + requirements: QueryRequirements::default(), + predictability: Predictability::default(), + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval_ms), + evaluation_phase: TimestampMs(0), + }), + range_evaluation: None, } } -/// Each query output reference of `physical`'s memo with every candidate of -/// its group. -fn outputs(physical: &PhysicalASAPMemoCandidates) -> Vec<(QueryId, Vec)> { +/// Each query output reference of `physical`'s memo: query `q` reads its +/// operator root. +fn outputs(physical: &PhysicalASAPMemoCandidates) -> Vec<(QueryId, MemoGroupId)> { physical .logical_candidates .memo .queries .iter() .enumerate() - .map(|(query, memo_query)| { - let QueryRoot::Operator(group) = memo_query.root else { - unreachable!("operator roots only") - }; - let candidates = (0..physical.physical_candidates.len()) - .filter(|&v| physical.physical_candidates[v].logical_candidate_id.group == group) - .map(PhysicalCandidateId) - .collect(); - (QueryId(query), candidates) + .map(|(query, memo_query)| match memo_query.root { + QueryRoot::Operator(group) => (QueryId(query), group), + QueryRoot::Scalar(_) => unreachable!("operator roots only"), }) .collect() } +/// Solves over `physical` for `entries` with every execution of candidate +/// `v` costing `w[v]`, every edge from `u` to `c` `edge_costs[(u, c)]` (0 +/// when missing) and candidate `v` retaining at `retention[v]` per ms; +/// returns the solution and the contexts it is over. fn solve( physical: &PhysicalASAPMemoCandidates, - costs: &SelectionCosts, -) -> Result { - solve_selection(&SelectionInputs { + entries: &[QueryWorkloadEntry], + w: &[f64], + edge_costs: &[((usize, usize), f64)], + retention: &[(usize, f64)], +) -> ( + Result, + ExecutionContexts, +) { + let (contexts, query_outputs) = ExecutionContexts::new(physical, entries, &outputs(physical)); + let costs = ExecutionCosts { + per_execution: contexts + .executions() + .iter() + .map(|execution| Ok(w[execution.candidate.0])) + .collect(), + per_edge: contexts + .edges() + .iter() + .map(|&(_, id, _)| { + let edge = physical.edge(id); + let key = (edge.producer_candidate_id.0, edge.consumer_candidate_id.0); + Some( + edge_costs + .iter() + .find(|(pair, _)| *pair == key) + .map_or(0.0, |(_, cost)| *cost), + ) + }) + .collect(), + retention_cost_per_second: retention + .iter() + .map(|&(v, per_ms)| (PhysicalCandidateId(v), per_ms)) + .collect(), + }; + let solved = solve_selection(&SelectionInputs { physical, - costs, - query_outputs: &outputs(physical), - }) + entries, + contexts: &contexts, + costs: &costs, + query_outputs: &query_outputs, + }); + (solved, contexts) +} + +/// The candidates of `solved`'s executions, one per execution, ascending. +fn candidates(solved: &SolvedSelection) -> Vec { + let mut candidates: Vec = solved + .executions + .iter() + .map(|execution| execution.candidate.0) + .collect(); + candidates.sort_unstable(); + candidates } -fn ids(ids: &[usize]) -> Vec { - ids.iter().copied().map(PhysicalCandidateId).collect() +fn first(group: MemoGroupId) -> CandidateForMemoGroupId { + CandidateForMemoGroupId { + group, + index_in_group: 0, + } } /// `m[5m]`, its scan with two physical candidates. Through v0 the read /// costs 1 + 5, through v1 2 + 0: the optimum takes v1, and the plan built /// from it is the scan v1 below the range v2, the scan retaining the 5 -/// minutes the range reads. +/// minutes the range reads (R). #[test] fn selects_the_producer_whose_own_and_edge_costs_are_lowest() { let mut memo = Memo::default(); let m = scan(&mut memo); let last_5m = range(&mut memo, 300, m); query(&mut memo, last_5m); - let first = |group: MemoGroupId| CandidateForMemoGroupId { - group, - index_in_group: 0, - }; let physical = physical( memo, &[ @@ -196,45 +247,48 @@ fn selects_the_producer_whose_own_and_edge_costs_are_lowest() { (first(last_5m), NONE), ], ); - let costs = costs(&physical, &[1.0, 2.0, 1.0], &[((0, 2), 5.0)]); - let solved = solve(&physical, &costs).unwrap(); - assert_eq!(solved.candidates, ids(&[1, 2])); - assert_eq!(solved.edges, vec![edge(&physical, 1, 2)]); - assert_eq!(solved.outputs, ids(&[2])); + let (solved, _) = solve( + &physical, + &every_10s(1), + &[1.0, 2.0, 1.0], + &[((0, 2), 5.0)], + &[(1, 1e-9)], + ); + let solved = solved.unwrap(); + assert_eq!(candidates(&solved), vec![1, 2]); + assert_eq!(solved.edges.len(), 1); + assert_eq!(solved.edges[0].1, edge(&physical, 1, 2)); + assert_eq!(solved.outputs[0].candidate, PhysicalCandidateId(2)); - let outputs = vec![(QueryId(0), last_5m)]; - let plan = plan_of(&physical, &[], &solved, &outputs).unwrap(); + let plan = plan_of(&physical, &solved, &outputs(&physical)).unwrap(); assert_eq!(plan.roots, vec![1]); assert_eq!(plan.edges.len(), 1); assert_eq!((plan.edges[0].producer, plan.edges[0].consumer), (0, 1)); - let candidates: Vec<_> = plan + let selected: Vec<_> = plan .nodes .iter() .map(|node| node.selected_physical_candidate) .collect(); assert_eq!( - candidates, + selected, vec![Some(PhysicalCandidateId(1)), Some(PhysicalCandidateId(2))] ); assert_eq!(plan.nodes[0].retention_ms, Some(300_000)); assert_eq!(plan.nodes[1].retention_ms, None); } -/// `m[5m]` and `m[1m]` both read the scan. Its candidate that is not -/// materialized (cost 1) may have one consumer, so the optimum reads the -/// scan's `OneEvaluation` candidate (cost 10) twice rather than both. +/// `m[5m]` and `m[1m]` both read the scan, both queries every 10 s: one +/// context. Its candidate that is not materialized (cost 1) may have one +/// consumer there, so the optimum reads the scan's `OneEvaluation` +/// candidate (cost 10) twice rather than both; without it no plan exists. #[test] -fn a_query_time_result_that_is_not_materialized_has_one_consumer() { +fn a_query_time_result_that_is_not_materialized_has_one_consumer_per_context() { let mut memo = Memo::default(); let m = scan(&mut memo); let last_5m = range(&mut memo, 300, m); let last_1m = range(&mut memo, 60, m); query(&mut memo, last_5m); query(&mut memo, last_1m); - let first = |group: MemoGroupId| CandidateForMemoGroupId { - group, - index_in_group: 0, - }; let physical = physical( memo, &[ @@ -244,9 +298,8 @@ fn a_query_time_result_that_is_not_materialized_has_one_consumer() { (first(m), ONE_EVALUATION), ], ); - let with_one_evaluation = costs(&physical, &[1.0, 1.0, 1.0, 10.0], &[]); - let solved = solve(&physical, &with_one_evaluation).unwrap(); - assert_eq!(solved.candidates, ids(&[1, 2, 3])); + let (solved, _) = solve(&physical, &every_10s(2), &[1.0, 1.0, 1.0, 10.0], &[], &[]); + assert_eq!(candidates(&solved.unwrap()), vec![1, 2, 3]); let without_one_evaluation = PhysicalASAPMemoCandidates::new( physical.logical_candidates.clone(), @@ -258,11 +311,110 @@ fn a_query_time_result_that_is_not_materialized_has_one_consumer() { .collect(), vec![], ); - let only_unshared = costs(&without_one_evaluation, &[0.0, 1.0, 1.0], &[]); + let (solved, _) = solve( + &without_one_evaluation, + &every_10s(2), + &[0.0, 1.0, 1.0], + &[], + &[], + ); + assert_eq!(solved.unwrap_err(), SelectionMilpError::Infeasible); +} + +/// The same two queries every 10 s and every 15 s: two contexts, each +/// running the scan that is not materialized once for its own query +/// (cost 1 each), cheaper than one `OneEvaluation` run (cost 10), which +/// could not be shared across contexts anyway. +#[test] +fn a_result_that_is_not_materialized_runs_once_per_context_that_reads_it() { + let mut memo = Memo::default(); + let m = scan(&mut memo); + let last_5m = range(&mut memo, 300, m); + let last_1m = range(&mut memo, 60, m); + query(&mut memo, last_5m); + query(&mut memo, last_1m); + let physical = physical( + memo, + &[ + (first(m), NONE), + (first(last_5m), NONE), + (first(last_1m), NONE), + (first(m), ONE_EVALUATION), + ], + ); + let (solved, contexts) = solve( + &physical, + &[every(10_000), every(15_000)], + &[1.0, 1.0, 1.0, 10.0], + &[], + &[], + ); + let solved = solved.unwrap(); + assert_eq!(candidates(&solved), vec![0, 0, 1, 2]); + let scan_contexts: Vec<_> = solved + .executions + .iter() + .filter(|execution| execution.candidate.0 == 0) + .map(|execution| execution.context) + .collect(); + assert_ne!(scan_contexts[0], scan_contexts[1]); + let plan = plan_of(&physical, &solved, &outputs(&physical)).unwrap(); + assert_eq!(plan.nodes.len(), 4); + assert_eq!(contexts.executions().len(), 6); +} + +/// `m[1m]` every 10 s and `m[5m]` every 20 s, both over a chain of 10 s +/// windows of the scan, materialized across evaluations: the chain runs +/// once, in its own context, and delivers on both grids; it retains the +/// longest range its selected consumers read, 5 minutes (R). +#[test] +fn a_materialized_chain_runs_once_and_retains_what_its_consumers_read_back() { + let mut memo = Memo::default(); + let m = scan(&mut memo); + let chain = group_with_own_candidate( + &mut memo, + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: m, + summary_type: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + filter: None, + tumbling_window_width_ms: 10_000, + }), + ); + let last_1m = range(&mut memo, 60, chain); + let last_5m = range(&mut memo, 300, chain); + query(&mut memo, last_1m); + query(&mut memo, last_5m); + let physical = physical( + memo, + &[ + (first(m), NONE), + (first(chain), ACROSS_EVALUATIONS), + (first(last_1m), NONE), + (first(last_5m), NONE), + ], + ); + let (solved, _) = solve( + &physical, + &[every(10_000), every(20_000)], + &[1.0, 1.0, 1.0, 1.0], + &[], + &[(1, 1e-6)], + ); + let solved = solved.unwrap(); + assert_eq!(candidates(&solved), vec![0, 1, 2, 3]); assert_eq!( - solve(&without_one_evaluation, &only_unshared).unwrap_err(), - SelectionMilpError::Infeasible + solved.retention_ms.get(&PhysicalCandidateId(1)), + Some(&300_000) ); + let plan = plan_of(&physical, &solved, &outputs(&physical)).unwrap(); + let chain_node = plan + .nodes + .iter() + .find(|node| node.selected_physical_candidate == Some(PhysicalCandidateId(1))) + .unwrap(); + assert_eq!(chain_node.retention_ms, Some(300_000)); } /// Ga = `m offset 1s` and Gb = `m offset 2s`, each also computed from the @@ -280,10 +432,6 @@ fn acyclicity_forbids_two_groups_reading_each_other() { let b_from_a = add_candidate(&mut memo, b, shift(1_000, a), None).unwrap(); query(&mut memo, a); query(&mut memo, b); - let first = |group: MemoGroupId| CandidateForMemoGroupId { - group, - index_in_group: 0, - }; // v0 scan; v1 Ga①, v2 Gb①, v3 Ga②, v4 Gb②. let physical = physical( memo, @@ -295,13 +443,55 @@ fn acyclicity_forbids_two_groups_reading_each_other() { (b_from_a, ONE_EVALUATION), ], ); - let costs = costs(&physical, &[0.0, 10.0, 10.0, 1.0, 1.0], &[]); - let solved = solve(&physical, &costs).unwrap(); + let (solved, _) = solve( + &physical, + &every_10s(2), + &[0.0, 10.0, 10.0, 1.0, 1.0], + &[], + &[], + ); + let solved = solved.unwrap(); + let selected = candidates(&solved); assert!( - solved.candidates == ids(&[0, 1, 4]) || solved.candidates == ids(&[0, 2, 3]), - "{:?}", - solved.candidates + selected == vec![0, 1, 4] || selected == vec![0, 2, 3], + "{selected:?}" ); - let outputs = vec![(QueryId(0), a), (QueryId(1), b)]; - plan_of(&physical, &[], &solved, &outputs).unwrap(); + plan_of(&physical, &solved, &outputs(&physical)).unwrap(); +} + +/// `m[5m]` over a chain of 10 s windows of the scan: a chain materialized +/// across evaluations builds only its newest window from the scan (10 s +/// read back); one rebuilt in each evaluation builds every window the range +/// over it reads (5 minutes). +#[test] +fn a_chain_rebuilt_in_each_evaluation_reads_back_what_its_consumers_read() { + let mut memo = Memo::default(); + let m = scan(&mut memo); + let chain = group_with_own_candidate( + &mut memo, + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: m, + summary_type: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + filter: None, + tumbling_window_width_ms: 10_000, + }), + ); + let last_5m = range(&mut memo, 300, chain); + query(&mut memo, last_5m); + let physical = physical( + memo, + &[ + (first(m), NONE), + (first(chain), ACROSS_EVALUATIONS), + (first(chain), NONE), + (first(last_5m), NONE), + ], + ); + let need_ms = |chain: usize| { + crate::cost::edge_retention_need_ms(&physical, &every_10s(1), edge(&physical, 0, chain)) + }; + assert_eq!(need_ms(1), Some(10_000)); + assert_eq!(need_ms(2), Some(300_000)); } diff --git a/crates/stage3-plan-selection/src/plan.rs b/crates/stage3-plan-selection/src/plan.rs index 2260c8e73..b96c906b9 100644 --- a/crates/stage3-plan-selection/src/plan.rs +++ b/crates/stage3-plan-selection/src/plan.rs @@ -1,7 +1,8 @@ -//! The plan Stage 3 builds from the baseline MILP's optimum as its last -//! step (concepts doc, section 3, "Plan output"): one `PhysicalASAPDAG` -//! node per selected candidate, its inputs from the selected edges, each -//! node's `selected_physical_candidate` set. +//! The plan Stage 3 builds from the MILP's optimum as its last step +//! (concepts doc, section 3, "Plan output"): one `PhysicalASAPDAG` node +//! per selected execution, its inputs from the selected edges, each node's +//! `selected_physical_candidate` set. Several nodes may come from one +//! candidate, one per context it runs in. use std::collections::{BTreeMap, HashMap}; @@ -9,106 +10,78 @@ use asap_types::ir::physical_export::{ PhysicalASAPDAG, PhysicalASAPDAGValidationError, PhysicalASAPNodeId, }; use asap_types::ir::QueryRoot; -use asap_types::stages::{ - MemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidateId, PhysicalEdgeId, -}; -use asap_types::workload::QueryWorkloadEntry; +use asap_types::stages::{MemoGroupId, PhysicalASAPMemoCandidates}; -use crate::cost::{candidate_node, input_count, retains_rows, retention_need_ms}; +use crate::cost::{candidate_node, input_count}; +use crate::milp::execution_contexts::Execution; use crate::milp::model::{QueryId, SolvedSelection}; /// The plan of `solved`, whose output `j` answers `outputs[j]` (a query and /// the group its root, or one operator its scalar root, reads). Nodes are -/// numbered producers first. A node that retains rows (materialized, or a -/// query-time `Scan`) has `retention_ms` the most its selected consumers -/// read back ([`retention_need_ms`]), and at least the range-evaluation -/// span of a query it answers directly. +/// numbered producers first. A node whose candidate has $h_v$ retains that +/// much (`retention_ms`); one that retains rows a consumer reads unbounded +/// has `retention_ms` `None` and keeps every row. pub(crate) fn plan_of( physical: &PhysicalASAPMemoCandidates, - entries: &[QueryWorkloadEntry], solved: &SolvedSelection, outputs: &[(QueryId, MemoGroupId)], ) -> Result { - let mut producer_of: HashMap<(PhysicalCandidateId, usize), PhysicalCandidateId> = - HashMap::new(); - let mut consumers_of: BTreeMap> = BTreeMap::new(); - for &id in &solved.edges { - let edge = physical.edge(id); + let mut producer_of: HashMap<(Execution, usize), Execution> = HashMap::new(); + let mut consumers_of: BTreeMap> = BTreeMap::new(); + for &(consumer, edge, producer) in &solved.edges { producer_of.insert( - (edge.consumer_candidate_id, edge.input_index_of_consumer), - edge.producer_candidate_id, + (consumer, physical.edge(edge).input_index_of_consumer), + producer, ); - consumers_of - .entry(edge.producer_candidate_id) - .or_default() - .push(id); + consumers_of.entry(producer).or_default().push(consumer); } - let consumers = |v: PhysicalCandidateId| consumers_of.get(&v).cloned().unwrap_or_default(); - let inputs = |v: PhysicalCandidateId| -> Vec { - (0..input_count(physical, v)) - .map(|i| producer_of[&(v, i)]) - .collect() - }; + let input_count = |execution: Execution| input_count(physical, execution.candidate); // Producers first: Kahn's order over the selected edges, which (E) // keeps acyclic. - let mut waiting: HashMap = solved - .candidates + let mut waiting: HashMap = solved + .executions .iter() - .map(|&v| (v, input_count(physical, v))) + .map(|&execution| (execution, input_count(execution))) .collect(); - let mut ready: Vec = solved - .candidates + let mut ready: Vec = solved + .executions .iter() .rev() .copied() - .filter(|v| waiting[v] == 0) + .filter(|execution| waiting[execution] == 0) .collect(); - let mut node_of: HashMap = HashMap::new(); - let mut order = Vec::with_capacity(solved.candidates.len()); - while let Some(v) = ready.pop() { - node_of.insert(v, order.len()); - order.push(v); - for edge in consumers(v) { - let consumer = physical.edge(edge).consumer_candidate_id; - let left = waiting.get_mut(&consumer).expect("a selected consumer"); + let mut node_of: HashMap = HashMap::new(); + let mut order = Vec::with_capacity(solved.executions.len()); + while let Some(execution) = ready.pop() { + node_of.insert(execution, order.len()); + order.push(execution); + for consumer in consumers_of.get(&execution).into_iter().flatten() { + let left = waiting.get_mut(consumer).expect("a selected consumer"); *left -= 1; if *left == 0 { - ready.push(consumer); + ready.push(*consumer); } } } - if order.len() != solved.candidates.len() { + if order.len() != solved.executions.len() { return Err(PhysicalASAPDAGValidationError::Cycle); } - // The span of the queries each candidate answers directly. - let mut output_span_ms: HashMap = HashMap::new(); - for (&(query, _), &v) in outputs.iter().zip(&solved.outputs) { - let span_ms = entries - .get(query.0) - .and_then(|entry| entry.range_evaluation) - .map_or(0, |range| range.span_ms); - let known = output_span_ms.entry(v).or_default(); - *known = (*known).max(span_ms); - } let nodes = order .iter() .enumerate() - .map(|(id, &v)| { - let input_ids: Vec = - inputs(v).iter().map(|producer| node_of[producer]).collect(); - let mut node = candidate_node(physical, v, id, &input_ids); - if retains_rows(physical, v) { - node.retention_ms = - retention_need_ms(physical, entries, v, &consumers(v), &consumers) - .map(|need_ms| need_ms.max(output_span_ms.get(&v).copied().unwrap_or(0))); - } + .map(|(id, &execution)| { + let input_ids: Vec = (0..input_count(execution)) + .map(|i| node_of[&producer_of[&(execution, i)]]) + .collect(); + let mut node = candidate_node(physical, execution.candidate, id, &input_ids); + node.retention_ms = solved.retention_ms.get(&execution.candidate).copied(); node }) .collect(); let output_node: HashMap<(QueryId, MemoGroupId), PhysicalASAPNodeId> = outputs .iter() .zip(&solved.outputs) - .map(|(&output, v)| (output, node_of[v])) + .map(|(&output, execution)| (output, node_of[execution])) .collect(); let mut roots = Vec::new(); let mut scalar_roots = Vec::new(); diff --git a/crates/stage3-plan-selection/src/select.rs b/crates/stage3-plan-selection/src/select.rs index a82161ee0..418771b66 100644 --- a/crates/stage3-plan-selection/src/select.rs +++ b/crates/stage3-plan-selection/src/select.rs @@ -1,8 +1,9 @@ -//! Stage 3, plan selection: [`select_plan`] costs every physical candidate -//! and selectable edge ([`crate::cost`]), builds and solves the baseline -//! selection MILP over Stage 2's candidates and admissible edges +//! Stage 3, plan selection: [`select_plan`] finds every execution of a +//! candidate in a context and every edge between executions the plan may +//! select ([`crate::milp::execution_contexts`]), costs them +//! ([`crate::cost`]), builds and solves the selection MILP over them //! ([`crate::milp::model`]), and builds the plan from the selected -//! candidates and edges as its last step ([`crate::plan`]). +//! executions and edges as its last step ([`crate::plan`]). use std::collections::BTreeSet; @@ -10,22 +11,20 @@ use asap_cost::{summary_type_name, CostInputs, CostModel}; use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId}; use asap_types::ir::schema::FieldDataType; use asap_types::ir::{ASAPOp, Operator, QueryRoot}; -use asap_types::physical::ExecutionTiming; use asap_types::stages::{ - output_references, producer_satisfies_physical_requirements, EvaluationGrid, ExecutionContext, - MemoGroupId, PhysicalASAPMemoCandidates, PhysicalCandidate, PhysicalCandidateId, PhysicalEdge, - PhysicalEdgeId, PruneReason, PrunedCandidate, + output_references, MemoGroupId, PhysicalASAPMemoCandidates, PruneReason, PrunedCandidate, }; use asap_types::workload::{DataWorkload, QueryWorkloadEntry}; -use crate::cost::{selection_costs, SelectionCosts}; +use crate::cost::{execution_costs, ExecutionCosts}; +use crate::milp::execution_contexts::{Execution, ExecutionContexts}; use crate::milp::model::{solve_selection, QueryId, SelectionInputs, SelectionMilpError}; use crate::plan::plan_of; use crate::report::{AcceptedAccuracy, ErrorSelectingPlan, Rejection, RejectionCause, Selection}; /// Stage 3, plan selection: select one plan among `physical`'s candidates /// (module documentation) under `cost_inputs`, and build it: one -/// `PhysicalASAPDAG` node per selected candidate, its inputs from the +/// `PhysicalASAPDAG` node per selected execution, its inputs from the /// selected edges, each node's `selected_physical_candidate` set. /// `entries[q]` is query `q`'s workload entry; `data` the data workload /// with its bounds. @@ -55,19 +54,6 @@ pub fn select_plan( }); } let formula_scale = cost_inputs.frozen_calibration_factor.unwrap_or(1.0); - let selectable: Vec = physical - .edges() - .filter(|(_, edge)| edge.is_admissible() && delivers_to_consumer(physical, entries, edge)) - .map(|(id, _)| id) - .collect(); - let costs = selection_costs( - physical, - &selectable, - entries, - data, - cost_inputs, - formula_scale, - ); let roots: Vec> = queries.iter().map(|query| query.root.clone()).collect(); let outputs: Vec<(QueryId, MemoGroupId)> = output_references(&roots) @@ -76,35 +62,36 @@ pub fn select_plan( .collect::>() .into_iter() .collect(); - let query_outputs: Vec<(QueryId, Vec)> = outputs - .iter() - .map(|&(query, group)| { - let candidates = physical - .physical_candidates - .iter() - .enumerate() - .filter(|(v, candidate)| { - candidate.logical_candidate_id.group == group - && costs.candidate[*v].is_ok() - && producer_satisfies_physical_requirements( - candidate, - ExecutionTiming::QueryTime, - ) - && candidate.delivers_on(physical, &evaluation_context(&entries[query.0])) - }) - .map(|(v, _)| PhysicalCandidateId(v)) + let (contexts, output_executions) = ExecutionContexts::new(physical, entries, &outputs); + let costs = execution_costs( + physical, + &contexts, + &output_executions, + entries, + data, + cost_inputs, + formula_scale, + ); + let query_outputs: Vec<(QueryId, Vec)> = output_executions + .into_iter() + .map(|(query, executions)| { + let costed = executions + .into_iter() + .filter(|&execution| costs.per_execution[contexts.index_of(execution)].is_ok()) .collect(); - (query, candidates) + (query, costed) }) .collect(); if query_outputs .iter() - .any(|(_, candidates)| candidates.is_empty()) + .any(|(_, executions)| executions.is_empty()) { return Err(infeasible(&costs)); } let solved = solve_selection(&SelectionInputs { physical, + entries, + contexts: &contexts, costs: &costs, query_outputs: &query_outputs, }) @@ -112,16 +99,21 @@ pub fn select_plan( SelectionMilpError::Infeasible => infeasible(&costs), SelectionMilpError::Solver(reason) => ErrorSelectingPlan::Solver(reason), })?; - let plan = plan_of(physical, entries, &solved, &outputs).map_err(|error| { + let plan = plan_of(physical, &solved, &outputs).map_err(|error| { ErrorSelectingPlan::Solver(format!("the selected plan is invalid: {error}")) })?; + // A candidate none of whose executions could be costed. + let mut costed = vec![None; physical.physical_candidates.len()]; + for (execution, cost) in contexts.executions().iter().zip(&costs.per_execution) { + let known = costed[execution.candidate.0].get_or_insert(false); + *known |= cost.is_ok(); + } let selection = Selection { accepted_accuracy_per_query: accepted_accuracy(physical, &plan), - pruned_candidates: costs - .candidate + pruned_candidates: costed .iter() .zip(&physical.physical_candidates) - .filter(|(cost, _)| cost.is_err()) + .filter(|(costed, _)| **costed == Some(false)) .map(|(_, &candidate)| PrunedCandidate { candidate, reason: PruneReason::UsageNotCostable, @@ -135,50 +127,12 @@ pub fn select_plan( Ok((plan, selection)) } -/// The evaluation context of `entry`'s query: the grid its evaluations -/// fall on (`QueryRecurrence::evaluation_grid_ms`), anchored at the epoch. -/// Times that are not known may fall on any millisecond: a 1 ms grid, on -/// which only producers stored at every time deliver. -fn evaluation_context(entry: &QueryWorkloadEntry) -> ExecutionContext { - ExecutionContext::Evaluation { - grid: EvaluationGrid { - period_ms: entry - .recurrence - .evaluation_grid_ms(entry.range_evaluation) - .unwrap_or(1), - phase_ms: 0, - }, - } -} - -/// Whether `edge`'s producer delivers on every evaluation its query-time -/// consumer serves: on the evaluation context of each query reaching the -/// consumer's group ([`PhysicalCandidate::delivers_on`]). The baseline has -/// no other context: an ingestion-time consumer's (its pipeline) is checked -/// in #1207 PR 7. -fn delivers_to_consumer( - physical: &PhysicalASAPMemoCandidates, - entries: &[QueryWorkloadEntry], - edge: &PhysicalEdge, -) -> bool { - let producer: &PhysicalCandidate = &physical.physical_candidates[edge.producer_candidate_id.0]; - let consumer = &physical.physical_candidates[edge.consumer_candidate_id.0]; - if consumer.physical_properties.timing() != ExecutionTiming::QueryTime { - return true; - } - physical.logical_candidates.memo.groups()[consumer.logical_candidate_id.group.0 as usize] - .reached_by_queries() - .iter() - .filter_map(|&query| entries.get(query)) - .all(|entry| producer.delivers_on(physical, &evaluation_context(entry))) -} - -/// The selection MILP is infeasible, with the reasons candidates could not +/// The selection MILP is infeasible, with the reasons executions could not /// be costed. -fn infeasible(costs: &SelectionCosts) -> ErrorSelectingPlan { +fn infeasible(costs: &ExecutionCosts) -> ErrorSelectingPlan { let mut reasons: Vec<&str> = Vec::new(); for reason in costs - .candidate + .per_execution .iter() .filter_map(|cost| cost.as_ref().err()) { @@ -187,11 +141,12 @@ fn infeasible(costs: &SelectionCosts) -> ErrorSelectingPlan { } } let mut reason = "no plan gives every query output a candidate whose every input has an \ - admissible producer (the selection MILP is infeasible)" + admissible producer that delivers on its context (the selection MILP is \ + infeasible)" .to_string(); if !reasons.is_empty() { reason += &format!( - "; candidates left out because they could not be costed: {}", + "; executions left out because they could not be costed: {}", reasons.join("; ") ); } diff --git a/crates/types/src/ir/operator/tumbling_window_chain.rs b/crates/types/src/ir/operator/tumbling_window_chain.rs index 065a3a310..02ebeb049 100644 --- a/crates/types/src/ir/operator/tumbling_window_chain.rs +++ b/crates/types/src/ir/operator/tumbling_window_chain.rs @@ -3,8 +3,8 @@ //! with a time index. A chain has no range and no window count: a //! `TimeRange` (and `TimeShift`s) over it select the windows a query reads, //! a `SummaryMerge` over that selection is the state of the chain's build -//! over the same range, and how many windows it retains is $h_v$ (#1207 -//! PR 7). +//! over the same range, and how many windows it retains is $h_v$, the +//! longest its selected consumers read back (Stage 3's constraint (R)). use super::asap::ASAPOp; use super::input::OperatorInput; use super::node::Operator; diff --git a/crates/types/src/stages/physical/memo_candidates.rs b/crates/types/src/stages/physical/memo_candidates.rs index 99e250135..b449ddb21 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -41,7 +41,11 @@ impl PhysicalCandidate { /// phase; any other result is stored on its group's evaluation grid /// (`evaluation_grid_of_group`), which must hold every time of the /// context, unless the group does not depend on the evaluation time - /// (then it delivers on every context). + /// (then it delivers on every context). The times of a + /// `Materialized(v)` context are $v$'s own: every window end of a chain + /// $v$ (its width at phase 0), else its group's evaluation grid; with + /// neither they are not known, and no time-dependent producer delivers + /// on it. pub fn delivers_on( &self, candidates: &PhysicalASAPMemoCandidates, @@ -53,18 +57,7 @@ impl PhysicalCandidate { Lifetime::AcrossEvaluations(_) => { let logical = &candidates.logical_candidates; let group = self.logical_candidate_id.group; - let chain_width_ms = match logical - .memo - .candidate(self.logical_candidate_id) - .expression - .operator - { - Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { - tumbling_window_width_ms, - .. - }) => Some(tumbling_window_width_ms), - _ => None, - }; + let chain_width_ms = self.chain_width_ms(candidates); let depends_on_evaluation_time = logical.memo.groups()[group.0 as usize] .group_properties .depends_on_evaluation_time; @@ -76,10 +69,17 @@ impl PhysicalCandidate { ExecutionContext::IngestionPipeline(window) => { (window.width_ms, window.phase_ms) } - ExecutionContext::Materialized(_) => { - todo!( - "#1207 PR 7: the times of a Materialized(v) context, v's own schedule" - ) + ExecutionContext::Materialized(v) => { + let materialized = &candidates.physical_candidates[v.0]; + match materialized.chain_width_ms(candidates) { + Some(width_ms) => (width_ms, 0), + None => match logical.evaluation_grid_of_group + [materialized.logical_candidate_id.group.0 as usize] + { + Some(grid) => (grid.period_ms, grid.phase_ms), + None => return false, + }, + } } }; match chain_width_ms { @@ -96,6 +96,24 @@ impl PhysicalCandidate { } } } + + /// Its tumbling-window width when it is a chain + /// (`SummaryTumblingWindowChain`). + pub fn chain_width_ms(&self, candidates: &PhysicalASAPMemoCandidates) -> Option { + match candidates + .logical_candidates + .memo + .candidate(self.logical_candidate_id) + .expression + .operator + { + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + tumbling_window_width_ms, + .. + }) => Some(tumbling_window_width_ms), + _ => None, + } + } } /// $\alpha^{\mathrm{phys}}_{u,c,i}$: timing and lifetime only (the concepts @@ -494,6 +512,70 @@ mod tests { } } + /// In `Materialized(v)` a producer delivers on $v$'s own times: a + /// 5 s chain inside the computation of a 10 s chain (every 10 s) and of + /// X = the scan's last 5 minutes stored on its 60 s grid, but not a + /// 10 s chain inside X's computation every 25 s, nor anything + /// time-dependent inside a build with neither a width nor a grid. + #[test] + fn delivers_on_the_own_times_of_a_materialized_computation() { + let grid = |period_ms| { + Some(EvaluationGrid { + period_ms, + phase_ms: 0, + }) + }; + let materialized = |grid_of_x| { + let mut candidates = candidates(memo(), vec![None, None, None, None, grid_of_x]); + candidates.physical_candidates = vec![ + candidate(WINDOWS_10S, QueryTime, Lifetime::AcrossEvaluations(MEMORY)), + candidate(LAST_5M, QueryTime, Lifetime::AcrossEvaluations(MEMORY)), + candidate(BUILD, QueryTime, Lifetime::AcrossEvaluations(MEMORY)), + ]; + candidates + }; + let across_5s = candidate(WINDOWS_5S, QueryTime, Lifetime::AcrossEvaluations(MEMORY)); + let across_10s = candidate( + WINDOWS_10S, + IngestionTime, + Lifetime::AcrossEvaluations(MEMORY), + ); + let in_computation_of = |v| ExecutionContext::Materialized(PhysicalCandidateId(v)); + for (candidates, producer, context, delivers) in [ + ( + materialized(grid(60_000)), + &across_5s, + in_computation_of(0), + true, + ), + ( + materialized(grid(60_000)), + &across_5s, + in_computation_of(1), + true, + ), + ( + materialized(grid(25_000)), + &across_10s, + in_computation_of(1), + false, + ), + ( + materialized(grid(60_000)), + &across_5s, + in_computation_of(2), + false, + ), + ] { + assert_eq!( + producer.delivers_on(&candidates, &context), + delivers, + "{:?} in {context:?}", + producer.logical_candidate_id + ); + } + } + /// `producer_satisfies_physical_requirements` over every valid producer state and every valid /// consumer state: a query-time producer serves only query-time /// consumers, an ingestion-time producer that is not materialized only diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index bf90a36c9..230b19a86 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -501,10 +501,17 @@ then the Pass 2 transformations), and expand shared sizes, tumbling-window horiz (planned) raw windows into concrete groups, so that choosing a threshold is choosing a group; choose one candidate per needed group; compose the choice into one plan and check and cost it with Stage 2 and Stage 3. The search is -the group selection MILP (mixed-integer linear program, #1095): it chooses one -candidate per memo group and execution context, with each shareable candidate -charged once, and is solved once; it prices cost model 1 only (cost models 2 -and 3 are an `UnsupportedCostModel` error, #888). Stage 3 builds the chosen +the selection MILP (mixed-integer linear program, #1204, concepts doc +sections 3 and 5) over execution contexts: a candidate with lifetime None or +`OneEvaluation` runs once per context that selects it ($x_{\kappa,v}$; a +context is an evaluation grid shared by the queries on it, an ingestion +pipeline's refresh window, or a materialized candidate's own computation), an +`AcrossEvaluations` candidate once ($x^{\mathrm{mat}}_v$), every edge per +context of its consumer ($y_{\kappa,u,c,i}$, only where `delivers_on` +holds), and a candidate that retains time-indexed rows retains what its +selected consumers read back ($h_v$, constraint (R)). It is solved once and +costs under cost model 1 only (cost models 2 and 3 are an +`UnsupportedCostModel` error, #888). Stage 3 builds the chosen plan, a `PhysicalASAPDAG`, as its last step, so it does not depend on the Stage 2 crate. The facade runs Pass 2, Stage 2 and Stage 3 once, over the full tumbling-window-width domain, with no time limit. @@ -515,7 +522,7 @@ The memo rules are in `crates/stage1-pass2-reuse-candidates/src/` (module `memo`, crate-private: the crate root re-exports each entry point); the memo types Stage 1 hands Stage 2 are in `crates/types/src/stages/` (`asap_types::stages`); the facade that runs the stages is in `crates/planner/src/run.rs`; the search over Stage 2's -candidates is in `crates/stage3-plan-selection/src/` (`select.rs`, `groups/`, `milp/`). +candidates is in `crates/stage3-plan-selection/src/` (`select.rs`, `cost.rs`, `milp/`, `plan.rs`). | Step | Entry point | Path | |---|---|---| @@ -524,7 +531,7 @@ candidates is in `crates/stage3-plan-selection/src/` (`select.rs`, `groups/`, `m | Memo types | `DagOfLogicalMemoGroups`, `PerComputationLogicalMemoCandidates`, `HowToComputeAggregateMeasure`, `Memo`, `CandidateForMemoGroup`, `OriginOfCandidateForMemoGroup`, `LogicalASAPMemoCandidates`, `CandidateGenerationTimeout`, `root_groups`, `queries_rooted_at` | `crates/types/src/stages/` (`asap_types::stages`) | | Pass 2 | `generate_logical_asap_memo_candidates` → `LogicalASAPMemoCandidates`, from Pass 1's output. It calls in this order: 1. rollup from the finest common group; 2. top-k reuse; 3. reach, caps and readout requirements; 4. dead-candidate elimination; 5. chain groups; 6. the logical edges | `stage1-pass2-reuse-candidates/src/` (`logical_memo_candidates.rs`, `rules/{rollup,topk,chains}.rs`, `requirements.rs`, `dead_candidates.rs`; `LogicalASAPMemoCandidates` itself: `asap_types::stages`) | | Memo to candidate DAGs | `enumerate`, `build_logical_asap_dags_from_chosen_candidates`, `ChosenCandidatePerMemoGroup`, `AllChosenCandidatesPerMemoGroup`, `ErrorBuildingLogicalASAPDagsFromChosenCandidates` | `pass2/memo/enumeration.rs` (re-exported at the crate root) | -| Production memo build and search | The facade's `run_planning` builds the DAG of logical memo groups and runs Pass 1 (`pass1_memo`) once; then `generate_logical_asap_memo_candidates`, `generate_physical_asap_memo_candidates` and Stage 3's `select_plan`, once each; `select_plan` builds the plan. Behind `select_plan`: the group selection MILP (#1095), `groups/` and `milp/` | `crates/planner/src/run.rs`; `crates/stage3-plan-selection/src/select.rs` | +| Production memo build and search | The facade's `run_planning` builds the DAG of logical memo groups and runs Pass 1 (`pass1_memo`) once; then `generate_logical_asap_memo_candidates`, `generate_physical_asap_memo_candidates` and Stage 3's `select_plan`, once each; `select_plan` builds the plan. Behind `select_plan`: the selection MILP over execution contexts (#1204), `cost.rs` and `milp/` (`execution_contexts.rs`, `model.rs`) | `crates/planner/src/run.rs`; `crates/stage3-plan-selection/src/select.rs` | ## ASAPPlanner Detailed Design: Planning Stages and their decisions/Strategies @@ -1290,11 +1297,9 @@ not check executor support. It then removes the candidates no buildable input or query output can use (`InputHasNoBuildableProducer`, `ReachesNoQueryOutput`), repeating until nothing more is removed. Stage 2 makes no execution-scope -copies: how many times a candidate runs is Stage 3's choice, and a -selection names each run as a node of its `DagOfSelectedCandidates`. -Today's Stage 3 engine, whose MILP runs each candidate at most once, makes -the copies it needs itself before solving and decodes each selected copy -into a node. +copies: how many times a candidate runs is Stage 3's choice. Stage 3 runs +a candidate with lifetime None or `OneEvaluation` once per execution +context that selects it, and builds each run as a node of the plan. *Example for Algorithm 3* ([Example 4, Pattern B](#example-4-materialization-of-window-summaries-in-physical-planning)). The logical candidate is the `Tumbling(1m)` KLL option from Algorithm 2. The