Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 0 additions & 7 deletions crates/asap-aware-mapping/src/cost_model.rs
Original file line number Diff line number Diff line change
Expand Up @@ -761,13 +761,6 @@ pub trait CostModel {
f64::NAN
}

/// Physical feasibility evidence for a complete summary candidate.
/// `None` defers admission to physical/deployment compilation; `Some(false)`
/// excludes the candidate without changing its computation or parameters.
fn summary_support_evidence(&self, _summary: &OperatorNode) -> Option<bool> {
None
}

/// Which mixed exact/summary execution shapes the downstream runtime
/// advertises (issue #171). Gates candidate *generation* in
/// [`crate::exact_composition::ExactCompositionStrategy`]: a shape the
Expand Down
139 changes: 139 additions & 0 deletions crates/asap-aware-mapping/src/deployment_capabilities.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
//! What the deployment's executor can run and keep (#509's third
//! deployment input, beside the cost and accuracy models).
//!
//! [`DeploymentCapabilities`] says three things:
//!
//! * which sketch families the executor supports, and which readouts it
//! supports for each. A family is supported when the executor can build
//! its state, and a readout of it when the executor can answer that
//! statistic ([`StatisticKind`]) from the state. Stage 1 (Pass 1) offers
//! no summary candidate whose family or readout is unsupported, so no
//! Pass 2 candidate has one either (Pass 2 rules share, split into panes
//! or reuse existing builds; none adds a family or a readout). Exact
//! accumulators and plain exact execution are always supported;
//! * whether raw samples are retained. If they are, Stage 3 counts them on
//! disk once per source, at the longest window a query-time scan of it
//! reads. If they are not, raw samples are not kept after they arrive, so
//! a candidate that scans raw data at query time is invalid (the
//! rejection lists every such scan): only ingestion-time work reads raw
//! rows. Stored data (at rest or mixed) with no raw data retained is a
//! contradiction, refused as bad input ([`DeploymentCapabilities::validate`]);
//! * whether query-time results are retained across evaluations. If they
//! are, Stage 2 also offers a tumbling-pane chain whose readers evaluate
//! together at its pane width as `MatAtQueryKeptAcrossEvaluations`: each
//! evaluation builds one new pane and keeps the others for the next ones,
//! covering the readers' longest look-back. A chain two or more queries
//! read keeps its `MatAtQuery` option too (shared within one evaluation),
//! and Stage 3 picks the cheaper. If they are not, a query-time result is
//! reused only within one evaluation.
//!
//! The default supports everything the planner can build, retains raw data
//! and keeps no query-time result across evaluations, which is the
//! planner's behavior without deployment inputs.

use asap_types::post_asap::{SketchAlgorithm, StatisticKind};
use asap_types::workload::{DataArrival, DataWorkload};
use serde::{Deserialize, Serialize};
use thiserror::Error;

/// What the deployment's executor can run and keep. Planning reads it in
/// Stage 1 (`supported_summaries`), Stage 2 (`query_time_results_retained`)
/// and Stage 3 (`raw_data_retained`). It is data only; `stage_pipeline` reads it from
/// `--deployment-capabilities <json>` and shows it in the stage document.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct DeploymentCapabilities {
/// The sketch families the executor can build, each with the readouts
/// it can answer from one; `None` (the default) supports every family
/// and readout the planner can build.
pub supported_summaries: Option<Vec<SupportedSummary>>,
/// Whether raw samples are kept for query-time scans; default `true`.
pub raw_data_retained: bool,
/// Whether a query-time result may be kept across evaluations
/// (`MatAtQueryKeptAcrossEvaluations`); default `false`.
pub query_time_results_retained: bool,
}

/// One sketch family the executor can build, with the readouts it can
/// answer from the family's state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SupportedSummary {
/// The family the executor can build.
pub family: SketchAlgorithm,
/// The statistics it can answer from one state of the family.
pub readouts: Vec<StatisticKind>,
}

impl Default for DeploymentCapabilities {
fn default() -> Self {
Self {
supported_summaries: None,
raw_data_retained: true,
query_time_results_retained: false,
}
}
}

impl DeploymentCapabilities {
/// Whether the executor can build `family` and answer `readout` from it.
pub fn supports(&self, family: &SketchAlgorithm, readout: StatisticKind) -> bool {
self.supported_summaries.as_ref().is_none_or(|supported| {
supported
.iter()
.any(|s| s.family == *family && s.readouts.contains(&readout))
})
}

/// Checks these capabilities against the data workload: stored data
/// (at rest, or mixed) is raw data the deployment keeps, so it
/// contradicts `raw_data_retained: false`. Plan selection refuses such
/// inputs before measuring any candidate.
pub fn validate(&self, data: &DataWorkload) -> Result<(), DeploymentCapabilitiesError> {
match &data.arrival {
DataArrival::AtRest | DataArrival::Mixed { .. } if !self.raw_data_retained => {
Err(DeploymentCapabilitiesError::StoredDataNotRetained {
arrival: data.arrival,
})
}
_ => Ok(()),
}
}
}

/// Deployment capabilities that contradict the other planning inputs
/// ([`DeploymentCapabilities::validate`]); always bad input.
#[derive(Debug, Clone, PartialEq, Error)]
pub enum DeploymentCapabilitiesError {
/// Bad input: the data workload declares stored raw data (`AtRest` or
/// `Mixed`), but the deployment says it retains none.
#[error("the data is stored ({arrival:?}), but the deployment does not retain raw data")]
StoredDataNotRetained {
/// The declared arrival that stores raw data.
arrival: DataArrival,
},
}

#[cfg(test)]
mod tests {
use super::*;

/// The default supports every family and readout, retains raw data and
/// keeps no query-time result across evaluations; a JSON object with
/// only some fields keeps the defaults for the rest.
#[test]
fn default_supports_everything_and_json_fills_the_rest() {
let default = DeploymentCapabilities::default();
assert!(default.supports(&SketchAlgorithm::Theta, StatisticKind::TopK));
assert!(default.raw_data_retained && !default.query_time_results_retained);
let parsed: DeploymentCapabilities = serde_json::from_str(
r#"{"supported_summaries": [{"family": "Kll", "readouts": ["Quantile"]}]}"#,
)
.unwrap();
assert!(parsed.supports(&SketchAlgorithm::Kll, StatisticKind::Quantile));
assert!(!parsed.supports(&SketchAlgorithm::Kll, StatisticKind::Total));
assert!(!parsed.supports(&SketchAlgorithm::DDSketch, StatisticKind::Quantile));
assert!(parsed.raw_data_retained && !parsed.query_time_results_retained);
assert!(serde_json::from_str::<DeploymentCapabilities>(r#"{"raw_data": false}"#).is_err());
}
}
5 changes: 5 additions & 0 deletions crates/asap-aware-mapping/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,8 @@
pub mod accuracy;
pub mod analytical_cost;
pub mod cost_model;
/// The deployment's capabilities: supported summaries and what it retains.
pub mod deployment_capabilities;
pub mod empirical_cost;
pub mod empirical_resources;
pub mod exact_composition;
Expand Down Expand Up @@ -179,6 +181,9 @@ pub use cost_model::{
CostModel, CostProvenance, CostUnit, DefaultCostModel, ExactCompositionCostInputs,
ExactCompositionCostRequest, ValueOperationCapabilities,
};
pub use deployment_capabilities::{
DeploymentCapabilities, DeploymentCapabilitiesError, SupportedSummary,
};
pub use exact_composition::{ExactComposition, ExactCompositionStrategy, OperationPlacement};
pub use explanation::{
explain_replacements, explain_replacements_with, ExplanationKind, ReplacementExplanation,
Expand Down
29 changes: 25 additions & 4 deletions crates/asap-aware-mapping/src/pass/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ use crate::accuracy::{
AccuracyEvidenceProvider, AccuracyModel, DefaultAccuracyModel, NoAccuracyEvidence,
};
use crate::cost_model::{CostModel, DefaultCostModel};
use crate::deployment_capabilities::DeploymentCapabilities;
use crate::per_computation_candidates::PerComputationCandidateError;
use crate::stage3_plan_selection::{PlanSelectionSettings, Selection, SelectionError};

Expand All @@ -45,16 +46,19 @@ static DEFAULT_ACCURACY_MODEL: DefaultAccuracyModel = DefaultAccuracyModel;
static NO_ACCURACY_EVIDENCE: NoAccuracyEvidence = NoAccuracyEvidence;
static DEFAULT_PLAN_SELECTION: std::sync::LazyLock<PlanSelectionSettings> =
std::sync::LazyLock::new(PlanSelectionSettings::default);
static DEFAULT_DEPLOYMENT: std::sync::LazyLock<DeploymentCapabilities> =
std::sync::LazyLock::new(DeploymentCapabilities::default);

// ── Input ────────────────────────────────────────────────────────────────

/// Planning logic, as opposed to the scoped facts it consumes: a model can have
/// a built-in default, evidence about a particular deployment cannot.
///
/// Plan selection reads `accuracy`, `evidence` and `plan_selection`. It
/// measures candidates with its own analytical model and compares them
/// under `plan_selection`'s cost model, so [`Stage1ToStage3Pass`] does not
/// read `cost`.
/// Plan selection reads `accuracy`, `evidence`, `plan_selection` and
/// `deployment_capabilities`. It measures candidates with its own
/// analytical model and compares them under `plan_selection`'s cost model,
/// so [`Stage1ToStage3Pass`] does not read `cost`. Stage 1 and Stage 2 read
/// `deployment_capabilities` too.
#[derive(Clone, Copy)]
#[non_exhaustive]
pub struct PlanningModels<'a> {
Expand All @@ -65,6 +69,11 @@ pub struct PlanningModels<'a> {
/// cost model 1 with illustrative unit costs), the resource speed and
/// the non-repeating query horizon.
pub plan_selection: &'a PlanSelectionSettings,
/// What the deployment's executor can run and keep: the summaries it
/// supports (Stage 1), whether it retains raw data (Stage 3) and
/// whether it keeps query-time results across evaluations (Stage 2);
/// default [`DeploymentCapabilities::default`].
pub deployment_capabilities: &'a DeploymentCapabilities,
}

impl<'a> PlanningModels<'a> {
Expand All @@ -78,6 +87,7 @@ impl<'a> PlanningModels<'a> {
accuracy,
evidence,
plan_selection: &DEFAULT_PLAN_SELECTION,
deployment_capabilities: &DEFAULT_DEPLOYMENT,
}
}

Expand All @@ -88,6 +98,7 @@ impl<'a> PlanningModels<'a> {
accuracy: &DEFAULT_ACCURACY_MODEL,
evidence: &NO_ACCURACY_EVIDENCE,
plan_selection: &DEFAULT_PLAN_SELECTION,
deployment_capabilities: &DEFAULT_DEPLOYMENT,
}
}

Expand All @@ -112,6 +123,16 @@ impl<'a> PlanningModels<'a> {
self.plan_selection = plan_selection;
self
}

/// These models with the deployment's capabilities (supported
/// summaries, raw-data and query-time retention).
pub fn with_deployment_capabilities(
mut self,
deployment_capabilities: &'a DeploymentCapabilities,
) -> Self {
self.deployment_capabilities = deployment_capabilities;
self
}
}

#[derive(Clone, Copy)]
Expand Down
39 changes: 21 additions & 18 deletions crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,7 @@ use std::rc::Rc;
use asap_types::ir::flat::flatten;
use asap_types::ir::schema_support::with_promql_series_identity;
use asap_types::ir::{OperatorNode, QueryRoot};
use asap_types::workload::{
DataWorkload, QueryLanguage, QueryRecurrence, QueryWorkloadEntry, RepeatedDemand,
};
use asap_types::workload::{DataWorkload, QueryLanguage, QueryRecurrence, QueryWorkloadEntry};

use super::{OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, QueryPlan};
use crate::asap_aware_cse::{
Expand All @@ -61,7 +59,9 @@ use crate::per_computation_candidates::{
compose_logical_candidate, generate_per_computation_candidates, PerComputationCandidateError,
PerComputationCandidates,
};
use crate::stage2_physical_asap_optimization::{optimize_physical_asap, PhysicalCandidate};
use crate::stage2_physical_asap_optimization::{
fixed_interval_ms, optimize_physical_asap, PhysicalCandidate,
};
use crate::stage3_plan_selection::{select_plan, Selection, SelectionError};
use crate::PlanningModels;

Expand Down Expand Up @@ -295,17 +295,6 @@ impl Pass2 {
}
}

/// The fixed interval at which a query repeats, in milliseconds.
fn fixed_interval_ms(recurrence: &QueryRecurrence) -> Option<u64> {
match recurrence {
QueryRecurrence::Repeated(
RepeatedDemand::FixedInterval(interval)
| RepeatedDemand::FixedIntervalAt { interval, .. },
) => Some(u64::from(interval.0)).filter(|&ms| ms > 0),
_ => None,
}
}

/// The Pass 2 candidates of the composed candidate with roots `roots`, in
/// enumeration order, each distinct candidate once. `interval_ms[q]` is the
/// fixed interval of query `q`, if it repeats at one.
Expand Down Expand Up @@ -488,8 +477,10 @@ impl Pass2List {

/// [`run_stage1_to_stage3`] after Pass 1: enumerates every Stage 1
/// candidate of `per_computation` (the Pass 1 inventory of the frontends'
/// roots, [`stage1_logical_asap_optimization`]), so a caller can read the
/// inventory, e.g. its
/// roots, [`stage1_logical_asap_optimization`]) without the summaries the
/// deployment does not support
/// ([`PerComputationCandidates::without_unsupported_summaries`]), so a
/// caller can read the inventory, e.g. its
/// [`combinations`](PerComputationCandidates::combinations), before
/// enumerating.
pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
Expand All @@ -498,6 +489,8 @@ pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
data: &DataWorkload,
models: PlanningModels<'_>,
) -> Stage1ToStage3Run<Id> {
let per_computation =
per_computation.without_unsupported_summaries(models.deployment_capabilities);
let recurrence: Vec<_> = entries.iter().map(|e| e.recurrence.clone()).collect();
let stage1 = stage1_logical_asap_optimization(&per_computation, &recurrence).collect();
run_stage2_and_stage3(per_computation, stage1, entries, data, models)
Expand All @@ -508,6 +501,15 @@ pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
/// enumerated): candidate `n` (from 1) is `L<n>`, implemented as `P<n>`
/// with nothing materialized and as `P<n>-m<k>` (`k` from 1) for each other
/// materialization assignment.
///
/// It does not prune: `per_computation` must already be pruned for
/// `models.deployment_capabilities`
/// ([`PerComputationCandidates::without_unsupported_summaries`]), with
/// `stage1` enumerated from that pruned inventory, as
/// [`run_stage1_to_stage3_from_pass1`] does. Pruning here would come too
/// late: each of `stage1`'s choices indexes the inventory it was enumerated
/// from, so a candidate with an unsupported summary would already be in
/// `stage1`.
pub fn run_stage2_and_stage3<Id: Clone + std::fmt::Debug>(
per_computation: PerComputationCandidates<Id>,
stage1: Vec<Stage1Candidate>,
Expand All @@ -523,7 +525,8 @@ pub fn run_stage2_and_stage3<Id: Clone + std::fmt::Debug>(
let number = index + 1;
let id = format!("L{number}");
if let Ok(roots) = &candidate.roots {
match optimize_physical_asap(&id, roots, entries, data) {
match optimize_physical_asap(&id, roots, entries, data, models.deployment_capabilities)
{
Ok(implemented) => {
for (k, mut p) in implemented.into_iter().enumerate() {
p.id = match k {
Expand Down
Loading
Loading