Skip to content

Commit 2d46e96

Browse files
authored
Merge pull request #475 from ProjectASAP/feat/precompute-runtime-contract
feat(precompute): install ingest, schema, and producer contracts
2 parents 800b83f + 11306d0 commit 2d46e96

9 files changed

Lines changed: 593 additions & 84 deletions

File tree

‎control_plane/src/emit/backend_push.rs‎

Lines changed: 36 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -442,41 +442,52 @@ async fn push_cumulative_entries(
442442
// succeeds only when all three are accepted.
443443
let plan_id = PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
444444
let generated_at_unix_ms = now_unix_ms();
445-
let plan_bytes = match crate::backend_plan::from_stage_config(
445+
let backend_plan = match crate::backend_plan::from_stage_config(
446446
&cumulative_be,
447447
monitors,
448448
plan_id,
449449
generated_at_unix_ms,
450450
) {
451-
Ok(plan) => plan.encode_to_vec(),
451+
Ok(plan) => plan,
452452
Err(e) => {
453453
warn!(error = %e, "backend_plan::from_stage_config failed; refusing partial publication");
454454
return PushOutcome::EmitFailed;
455455
}
456456
};
457-
let precompute_plan = PrecomputePlan {
458-
envelope: PlanEnvelope {
459-
plan_id,
460-
plan_version: 1,
461-
generated_at_unix_ms,
462-
activation_unix_ms: generated_at_unix_ms,
463-
expiry_unix_ms: None,
464-
backend_compat: "asap-query-backend.v1".into(),
465-
planner_revision: crate::physical::compiler::PLANNER_REVISION.into(),
466-
capability_snapshot_id: "replanner".into(),
467-
},
468-
materializations: match cumulative_be
469-
.aggregations
470-
.iter()
471-
.map(crate::backend_plan::aggregation_config_for_materialization)
472-
.collect::<anyhow::Result<Vec<_>>>()
473-
{
474-
Ok(materializations) => materializations,
475-
Err(error) => {
476-
warn!(%error, "failed to build typed PrecomputePlan");
477-
return PushOutcome::EmitFailed;
478-
}
479-
},
457+
let plan_bytes = backend_plan.encode_to_vec();
458+
let precompute_envelope = PlanEnvelope {
459+
plan_id,
460+
plan_version: 1,
461+
generated_at_unix_ms,
462+
activation_unix_ms: generated_at_unix_ms,
463+
expiry_unix_ms: None,
464+
backend_compat: "asap-query-backend.v1".into(),
465+
planner_revision: crate::physical::compiler::PLANNER_REVISION.into(),
466+
capability_snapshot_id: "replanner".into(),
467+
};
468+
let materializations = match cumulative_be
469+
.aggregations
470+
.iter()
471+
.map(crate::backend_plan::aggregation_config_for_materialization)
472+
.collect::<anyhow::Result<Vec<_>>>()
473+
{
474+
Ok(materializations) => materializations,
475+
Err(error) => {
476+
warn!(%error, "failed to build typed PrecomputePlan");
477+
return PushOutcome::EmitFailed;
478+
}
479+
};
480+
let precompute_plan = match PrecomputePlan::build(
481+
precompute_envelope,
482+
materializations,
483+
&backend_plan,
484+
&["legacy-replanner".into()],
485+
) {
486+
Ok(plan) => plan,
487+
Err(error) => {
488+
warn!(%error, "failed to validate typed PrecomputePlan");
489+
return PushOutcome::EmitFailed;
490+
}
480491
};
481492

482493
// Storage-routing: the routing classifier (`build_routing_entry` in

‎control_plane/src/emit/stage_config.rs‎

Lines changed: 21 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -3142,24 +3142,33 @@ fn sketch_params_to_json(p: &SketchParams) -> JsonValue {
31423142
SketchParams::Kll { k } => json!({ "k": k }),
31433143
SketchParams::DDSketch { alpha } => json!({ "alpha": alpha }),
31443144
SketchParams::Hll { precision } => json!({ "precision": precision }),
3145-
// Cms/CmsWithHeap: matches the pre-split shape exactly — the old
3146-
// `SketchParams::Cms` arm never emitted `with_heap` in JSON even
3147-
// though `CmsParams.with_heap` existed as a field; heap-bearing
3148-
// and bare CMS produced identical wire JSON. `heap_size` is a
3149-
// new field with no wire representation here (nothing on the
3150-
// real backend wire path reads it — see `bind_cms_topk.rs`).
3151-
SketchParams::Cms { width, depth } | SketchParams::CmsWithHeap { width, depth, .. } => {
3152-
json!({ "w": width, "d": depth })
3153-
}
3145+
SketchParams::Cms { width, depth } => json!({ "w": width, "d": depth }),
3146+
SketchParams::CmsWithHeap {
3147+
width,
3148+
depth,
3149+
heap_size,
3150+
} => json!({
3151+
"w": width,
3152+
"d": depth,
3153+
"with_heap": true,
3154+
"heap_size": heap_size,
3155+
}),
31543156
// CountSketch/CountSketchWithHeap: the old arm always emitted
31553157
// `with_heap` (from `CountSketchParams.with_heap: bool`);
31563158
// that boolean is now the kind identity itself.
31573159
SketchParams::CountSketch { width, depth } => {
31583160
json!({ "w": width, "d": depth, "with_heap": false })
31593161
}
3160-
SketchParams::CountSketchWithHeap { width, depth, .. } => {
3161-
json!({ "w": width, "d": depth, "with_heap": true })
3162-
}
3162+
SketchParams::CountSketchWithHeap {
3163+
width,
3164+
depth,
3165+
heap_size,
3166+
} => json!({
3167+
"w": width,
3168+
"d": depth,
3169+
"with_heap": true,
3170+
"heap_size": heap_size,
3171+
}),
31633172
// Exact accumulators never reach here -- see
31643173
// `sketch_kind_to_backend_type`'s doc.
31653174
SketchParams::Kmv { .. } | SketchParams::Theta { .. } => unreachable!(

‎control_plane/src/opamp/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1016,6 +1016,7 @@ mod tests {
10161016
},
10171017
materializations: vec![crate::physical::compiler::CollectorMaterialization {
10181018
query_id: "q".into(),
1019+
materialization: asap_types::PolicyFingerprint(1),
10191020
metric: "requests".into(),
10201021
algorithm: "hll".into(),
10211022
parameters: serde_json::json!({"precision": 14}),

0 commit comments

Comments
 (0)