Skip to content

Commit dae148f

Browse files
zzylolclaude
andauthored
Execute Planner physical operators through typed deployment bindings (#774)
* docs: clarify Planner physical plan and SDS architecture * docs: specify executable subplan materialization boundaries * docs: scope migration to backend precompute and query plans * docs: clarify window terminology migration * Revert "docs: clarify window terminology migration" This reverts commit daa5281. * docs: focus physical plan and SDS designs * docs: add physical compiler input example * docs: add physical compiler output example * docs: include query expression in compiler example * docs: reorganize physical plan integration design * docs: reorganize SDS and migration designs * docs: define maintenance inputs before plan split example * docs: use plan version consistently in backend design * docs: remove standalone catalog materialization abstraction * docs: add concise planner backend glossary * refactor: split installed maintenance DAGs from query execution * refactor: remove backend Collector dependency and normalize legacy DAGs * test: restore whole-backend process coverage without Collector * fix: migrate Planner main and reject legacy runtime artifacts * test: send full sketch envelope in whole-backend E2E * refactor: bind backend summary state through versioned SDS slots * test: send complete sketch envelopes in process fixtures * docs: clarify selected deployment guarantee terminology * docs: explain missing planner maintenance guarantee * docs: motivate selected producer maintenance decision * docs: label catalog reads and SDS metadata ownership * docs: align SDS ownership and lifecycle terminology * docs: use current planner and plan-version names consistently * docs: align integration diagram with SDS ownership * docs: clarify instance identity and shared producer wording * refactor: align plan bindings with current Planner and SDS contract * docs: distinguish summary definitions from runtime stores * docs: model one runtime summary store for DAG bindings * docs: scope SDS lifecycle to read eligibility * docs: tie stored summary examples directly to DAG outputs * docs: name summary tables, stored records, and output references by role * docs: illustrate summary definitions, stored records, and output references * docs: limit v1 summary storage to definitions and stored summaries * refactor: align plan bindings with summary store v1 * fix: validate selected DAG provenance * fix: retain neutral sketch codec dependencies when syncing main * fix: align derived DAG validation with current schema versions * fix: keep maintenance document version distinct from complete DAG version * refactor: adopt costed Planner selection without legacy API adapters * test: verify exact process routing for uncertified Planner candidates * docs: remove redundant Planner selection contract * deps: pin Planner bounded HLL confidence model * test: size transmitted KLL state from a certified accuracy contract * test: retain KLL collector capability when using theoretical confidence * refactor: implement typed summary semantics and physical plan lowering * docs: separate Planner physical computation from backend deployment * refactor: name the backend orchestration entry DeploymentPlanCompiler * Restack PR #770 with implementation before standalone acceptance * chore: consume Planner physical precompute candidate interfaces * chore: consume Planner materialization frontier enumeration * chore: consume exact temporal ranking physical candidates * chore: use shared Planner candidate winner selection * chore: consume sparse counter shared readout contracts * test: declare collector ranking fixture capabilities explicitly * Use Planner counter-window candidate execution * Use shared keyed-counter omission contract * Use Planner exact-counter population omission * Construct fixture evidence for the standalone operator foundation * build: pin Planner exact-state scratch merge implementation * build: pin shared finalized-pane reconstruction fix * docs: bind Planner physical DAGs without backend re-lowering * docs: describe summary inputs with groups and pane duration * docs: define summary semantic completeness beyond input scope * docs: define SDS identity through canonical Planner computation * docs: decouple SDS semantic identity from executable Planner IR * docs: define Planner-owned SDS discovery for future ad hoc queries * docs: streamline SDS design around definitions and stored results * docs: track bound-query SDS migration across implementation PRs * docs: identify active shared-library PR in bound-query migration * build: align shared Planner dependencies with remote PR 462 * docs: separate bound SDS range lookup from state validation * feat(sds): separate bound output routing from persisted semantic identity * docs: state SDS migration responsibilities without stale implementation claims * refactor(sds): keep semantic variants compact without changing wire format * test: align evidence fixtures with selected exact count state * docs: align precompute SDS description with semantic catalog * test: bind imported-state fixtures to their actual semantic definition * test: distinguish imported CMS transport from total-count planning * docs: clarify Planner physical plan and SDS architecture * docs: specify executable subplan materialization boundaries * docs: scope migration to backend precompute and query plans * docs: clarify window terminology migration * Revert "docs: clarify window terminology migration" This reverts commit daa5281. * docs: focus physical plan and SDS designs * docs: add physical compiler input example * docs: add physical compiler output example * docs: include query expression in compiler example * docs: reorganize physical plan integration design * docs: reorganize SDS and migration designs * docs: define maintenance inputs before plan split example * docs: use plan version consistently in backend design * docs: remove standalone catalog materialization abstraction * docs: add concise planner backend glossary * docs: clarify selected deployment guarantee terminology * docs: explain missing planner maintenance guarantee * docs: motivate selected producer maintenance decision * docs: label catalog reads and SDS metadata ownership * docs: align SDS ownership and lifecycle terminology * docs: use current planner and plan-version names consistently * docs: align integration diagram with SDS ownership * docs: clarify instance identity and shared producer wording * docs: distinguish summary definitions from runtime stores * docs: model one runtime summary store for DAG bindings * docs: scope SDS lifecycle to read eligibility * docs: tie stored summary examples directly to DAG outputs * docs: name summary tables, stored records, and output references by role * docs: illustrate summary definitions, stored records, and output references * docs: limit v1 summary storage to definitions and stored summaries * docs: separate Planner physical computation from backend deployment * docs: bind Planner physical DAGs without backend re-lowering * docs: describe summary inputs with groups and pane duration * docs: define summary semantic completeness beyond input scope * docs: define SDS identity through canonical Planner computation * docs: decouple SDS semantic identity from executable Planner IR * docs: define Planner-owned SDS discovery for future ad hoc queries * docs: streamline SDS design around definitions and stored results * docs: track bound-query SDS migration across implementation PRs * docs: identify active shared-library PR in bound-query migration * docs: separate bound SDS range lookup from state validation * docs: state SDS migration responsibilities without stale implementation claims * docs: preserve design index after rebasing onto main * docs: clarify SDS source identity and version-scoped recovery * docs: illustrate SDS identity and recovery decisions * feat: bind Planner dataset semantics to deployment input identity * fix: recognize shared native batch encoding at dependency boundary * test: reject pre-dataset planning snapshot versions * refactor: version dataset-bound catalog and update empty-plan fixtures * docs: describe dataset-bound planning and installation inputs * docs: clarify candidate selection and deployment ownership * refactor: split backend deployment plans and bind stored outputs * fix: restrict foundation SDS recovery to the installed generation * fix: allocate fresh physical series when a plan version changes * test: record foundation rebase and recovery regression evidence * fix: complete shared state encoding adoption at the dependency boundary * style: satisfy workspace formatting after the compiler rename * feat(sds): separate bound output routing from persisted semantic identity * feat(sds): require dataset-bound Planner definitions and versioned catalogs * refactor: implement typed summary semantics and physical plan lowering * fix: validate final SDS semantics across Planner adapters and recovery * test: verify final SDS identity installation, serving and restart * test: align shared execution fixtures with final #749 SDS foundation * docs: record shared execution directly after the SDS foundation * fix(control-plane): supply dataset_identity in the API test fixtures CompileAndPublishPhysicalPlanRequest gained a required `dataset_identity` field in this branch, but two api_tests fixtures build their request body as JSON by hand and were never updated, so both failed deserialization with `missing field dataset_identity` before reaching the handler. Take the value from the planning snapshot's own `environment.dataset_identity` rather than inventing one, so the fixture keeps describing the same dataset the rest of the snapshot describes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Remove obsolete SDS wire aliases and persisted metadata migration * fix(data-plane): initialize dataset_identity in the bootstrap ingest contract IngestContract gained an optional `dataset_identity`, but the bootstrap envelope in data_plane's entry point was not updated, so the binary failed to compile with E0063 and took the whole workspace build with it. The bootstrap envelope describes the state before any plan is installed, where every other field is a placeholder, so there is no dataset to bind to yet; `None` is the accurate value. A published plan supplies the identity. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Reject obsolete state-column syntax in compiler roundtrip coverage * Remove unused materialization wire adapter and propagate recovery errors * Keep projection fixtures on the canonical typed encoding * Remove superseded deployment API field aliases * fix(control-plane): supply dataset_identity in the API test fixtures CompileAndPublishPhysicalPlanRequest gained a required `dataset_identity` field in this branch, but two api_tests fixtures build their request body as JSON by hand and were never updated, so both failed deserialization with `missing field dataset_identity` before reaching the handler. Take the value from the planning snapshot's own `environment.dataset_identity` rather than inventing one, so the fixture keeps describing the same dataset the rest of the snapshot describes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> (cherry picked from commit e627dbb) * Document obsolete-format removals and cleanup validation * Record passing restacked plan, storage and serving checks * Remove generated evaluation archives and keep validation summaries * Normalize validation summary formatting * Remove PR process reports and defer execution design to shared-runtime PR * Keep discovery and calibration on dataset-bound snapshot version 3 * Use typed deployment configuration in process E2E fixtures * Align process assertions with dataset-bound SDS and cold successor activation * fix: execute typed Planner fragments and preserve terminal resource errors * fix: bind exact integer samples without losing input types * test: verify exact integer protocol input binding * refactor: name per-boundary physical fragments explicitly * fix: consume finalized Planner query candidate outputs * test: require Planner filters for bound protocol vectors * test: use Planner schema lifting module * fix: bind Planner filters and finalized query results * docs: describe bound Planner filter execution * fix: retain state binding and sharing beneath explicit query readouts * fix: pin Planner query finalization for every candidate entry point * docs: distinguish query values from stored accumulator boundaries * test: assert exact Count state beneath its query readout --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent b08690b commit dae148f

97 files changed

Lines changed: 2201 additions & 13677 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Cargo.lock‎

Lines changed: 33 additions & 8 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ resolver = "2"
33
members = [
44
"crates/asap_otel_proto",
55
"crates/asap_types",
6-
"crates/asap_sketch_codec",
76
"data_plane",
87
"control_plane",
98
]
@@ -15,10 +14,10 @@ version = "0.1.0"
1514
[workspace.dependencies]
1615
# Keep Planner frontends, selection, and IR on the same immutable revision.
1716
# Alias upstream asap-types because this workspace also defines asap_types.
18-
planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
19-
asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
20-
asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
21-
asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
17+
planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
18+
asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
19+
asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
20+
asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
2221

2322
# Shared external deps (used by 2+ crates)
2423
serde = { version = "1.0", features = ["derive"] }
@@ -38,6 +37,8 @@ arc-swap = "1.7"
3837
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
3938

4039
# Internal crates
40+
asap-physical-operators = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
41+
asap_sketch_codec = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
4142
asap_types = { path = "crates/asap_types" }
4243
asap_otel_proto = { path = "crates/asap_otel_proto" }
4344
indexmap = { version = "2.0", features = ["serde"] }

‎control_plane/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ name = "control_plane"
1212
path = "src/main.rs"
1313

1414
[dependencies]
15+
asap-physical-operators.workspace = true
1516
tokio = { version = "1", features = ["full"] }
1617
axum = { version = "0.7", features = ["ws"] }
1718
futures-util = "0.3"

‎control_plane/examples/calibration_candidates.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
3737
vec![lhs, rhs],
3838
json!({"operator_debug":format!("{operator:?}"),"timing_debug":format!("{timing:?}")}),
3939
),
40+
4041
SummaryExpr::ValueOperation {
4142
child,
4243
operation,
@@ -73,11 +74,11 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
7374
right,
7475
kind,
7576
pred,
76-
pruning,
77+
..
7778
} => (
7879
"RelationalJoin",
7980
vec![left, right],
80-
json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}"), "pruning_debug":format!("{pruning:?}")}),
81+
json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}")}),
8182
),
8283
SummaryExpr::SummarySubtract { left, right } => {
8384
("SummarySubtract", vec![left, right], json!({}))

‎control_plane/src/emit/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ fn extract_from_node(node: &Rc<SummaryNode>) -> Option<SketchAlgorithm> {
5656
SummaryExpr::SummaryEstimate { summary_input, .. } => extract_from_node(summary_input),
5757
SummaryExpr::SummaryMerge { children, .. } => children.iter().find_map(extract_from_node),
5858
SummaryExpr::ValueOperation { child, .. } => extract_from_node(child),
59+
5960
// Not surfaced by any `Bind*` path yet (gated on rules that
6061
// haven't landed — see `deployment_expr.rs`'s module docs).
6162
SummaryExpr::BinaryOp { .. }

‎control_plane/src/physical/compiler.rs‎

Lines changed: 18 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -2105,6 +2105,7 @@ fn summary_agg_metric(node: &SummaryNode) -> Option<String> {
21052105
}
21062106
}
21072107
SummaryExpr::SummaryAgg { child, .. } => walk(child, metrics),
2108+
21082109
SummaryExpr::ValueOperation { child, .. } => walk(child, metrics),
21092110
SummaryExpr::SummaryEstimate { summary_input, .. } => walk(summary_input, metrics),
21102111
SummaryExpr::SummaryMerge { children, .. } => {
@@ -2173,11 +2174,12 @@ fn observed_population_matches_root(
21732174
};
21742175
if measures.is_empty()
21752176
|| !measures.iter().all(|intent| {
2176-
matches!(
2177-
intent,
2178-
AggIntent::FrequencyL2 { col: None, .. }
2179-
| AggIntent::FrequencyEntropy { col: None, .. }
2180-
) || matches!(intent, AggIntent::Cardinality { cols, .. } if cols.is_empty())
2177+
matches!(intent, AggIntent::Cardinality { cols, .. } if cols.is_empty())
2178+
|| matches!(
2179+
intent,
2180+
AggIntent::FrequencyL2 { col: None, .. }
2181+
| AggIntent::FrequencyEntropy { col: None, .. }
2182+
)
21812183
})
21822184
{
21832185
return false;
@@ -2380,6 +2382,7 @@ fn requires_exact_erp_fallback(
23802382
walk(left, out);
23812383
walk(right, out);
23822384
}
2385+
23832386
SummaryExpr::KeepPreAsap(_) => {}
23842387
}
23852388
}
@@ -3612,16 +3615,14 @@ fn collect_selected_materializations(
36123615
}
36133616
match &node.expr {
36143617
SummaryExpr::RelationalJoin {
3615-
right: candidates,
36163618
left: values,
3619+
right: candidates,
36173620
kind: planner_types::pre_asap::JoinKind::Semi,
36183621
pruning: Some(_),
36193622
..
36203623
} => {
36213624
walk(candidates, readout, composable, grouping.clone(), selected)?;
3622-
// In a hybrid TopK, the sketch is only a candidate-membership
3623-
// sidecar. Prometheus owns the authoritative value subtree;
3624-
// provisioning local exact state here duplicates that work.
3625+
// Explicit external authoritative values do not need duplicate local state.
36253626
if !composable {
36263627
walk(values, readout, composable, grouping.clone(), selected)?;
36273628
}
@@ -4490,6 +4491,7 @@ pub(crate) mod tests {
44904491
}
44914492
}
44924493

4494+
// The explicit rate-value frontier cannot be rebound as raw counter deltas.
44934495
#[test]
44944496
fn unsupported_rate_heap_uses_explicit_exact_route() {
44954497
let query = "topk(2, sum by (job) (rate(m[1m])))";
@@ -4517,6 +4519,7 @@ pub(crate) mod tests {
45174519
assert!(plan.precompute_plan.materializations.is_empty());
45184520
}
45194521

4522+
// Native exact values remain explicit; unsupported heaps publish no stored state.
45204523
#[test]
45214524
fn hybrid_rate_topk_preserves_the_original_exact_subquery() {
45224525
let query = "topk(2, sum by (job) (rate(m[1m])))";
@@ -4541,7 +4544,7 @@ pub(crate) mod tests {
45414544
assert!(matches!(
45424545
&entry.nodes[&entry.root],
45434546
QueryPlanNode::Logical {
4544-
operator: ResidualQueryOperator::TopKSelection { k: 2, .. },
4547+
operator: ResidualQueryOperator::Limit { n: 2, .. },
45454548
..
45464549
}
45474550
));
@@ -5764,6 +5767,11 @@ pub(crate) mod tests {
57645767
let selected = match &selected_root.expr {
57655768
SummaryExpr::SummaryEstimate { summary_input, .. } => summary_input.clone(),
57665769
SummaryExpr::SummaryAgg { .. } => selected_root.clone(),
5770+
SummaryExpr::ValueOperation {
5771+
child,
5772+
operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
5773+
..
5774+
} => child.clone(),
57675775
_ => panic!("expected maintained aggregate fixture"),
57685776
};
57695777
request.queries[0].selected_plan_root = Rc::new(SummaryNode {
@@ -5967,15 +5975,6 @@ pub(crate) mod tests {
59675975
let (right, _) = right.into_physical_compilation_request().unwrap();
59685976
let left = request.queries[0].selected_plan_root.clone();
59695977
let right = right.queries[0].selected_plan_root.clone();
5970-
let right = Rc::new(SummaryNode {
5971-
expr: SummaryExpr::ValueOperation {
5972-
timing: planner_types::post_asap::ExecutionTiming::QueryTime,
5973-
operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
5974-
child: right.clone(),
5975-
},
5976-
schema: right.schema.clone(),
5977-
guarantee: None,
5978-
});
59795978
request.queries[0].selected_plan_root = Rc::new(SummaryNode {
59805979
expr: SummaryExpr::BinaryOp {
59815980
timing: planner_types::post_asap::ExecutionTiming::QueryTime,

‎control_plane/src/physical/plan_dot.rs‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,16 @@ fn escape(value: &str) -> String {
139139

140140
fn query_node_label(node: &QueryPlanNode) -> String {
141141
match node {
142+
QueryPlanNode::PhysicalFragment { dag, .. } => {
143+
asap_physical_operators::physical_planner::CompiledPhysicalDag::decode(dag)
144+
.map(|plan| {
145+
format!(
146+
"Physical\n{}",
147+
plan.operator_name(plan.roots()[0]).unwrap_or("Input")
148+
)
149+
})
150+
.unwrap_or_else(|_| "Invalid physical DAG".into())
151+
}
142152
QueryPlanNode::RelationalJoin { .. } => "RelationalJoin".into(),
143153
QueryPlanNode::Relational { .. } => "Relational".into(),
144154
QueryPlanNode::Logical { operator, .. } => format!("Logical\n{}", residual_label(operator)),
@@ -154,7 +164,6 @@ fn query_node_label(node: &QueryPlanNode) -> String {
154164
QueryPlanNode::SummaryEstimate { query, .. } => format!("SummaryEstimate\n{query:?}"),
155165
QueryPlanNode::ExactReadout { readout, .. } => format!("ExactReadout\n{readout:?}"),
156166
QueryPlanNode::SummaryMerge { .. } => "SummaryMerge".into(),
157-
QueryPlanNode::MembershipFilter { .. } => "MembershipFilter".into(),
158167
QueryPlanNode::ExternalExact { .. } => "ExternalExact".into(),
159168
QueryPlanNode::ExactFallback { reason } => format!("ExactFallback\n{reason}"),
160169
}
@@ -169,7 +178,7 @@ fn residual_label(operator: &ResidualQueryOperator) -> &'static str {
169178
ResidualQueryOperator::UnaryNegate => "UnaryNegate",
170179
ResidualQueryOperator::VectorToScalar => "VectorToScalar",
171180
ResidualQueryOperator::Aggregate { .. } => "Aggregate",
172-
ResidualQueryOperator::TopKSelection { .. } => "TopKSelection",
181+
ResidualQueryOperator::Limit { .. } => "Limit",
173182
ResidualQueryOperator::Binary { .. } => "Binary",
174183
ResidualQueryOperator::Temporal { .. } => "Temporal",
175184
ResidualQueryOperator::Sort { .. } => "Sort",

‎control_plane/src/physical/post_asap/cost_model.rs‎

Lines changed: 37 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -174,9 +174,20 @@ impl ControlPlaneCostModel {
174174
&self,
175175
candidate: &ReplacementSubDAG,
176176
) -> Option<CandidateCostEstimate> {
177+
if let Replacement::Rewrite(root) = &candidate.replacement {
178+
// Price a semantic rewrite through its executable summary candidates,
179+
// in the same retained-state units as a direct summary candidate.
180+
// Unknown realizations remain uncosted rather than receiving zero.
181+
use asap_aware_mapping::{ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG};
182+
return SketchAlgorithmStrategy::new(self)
183+
.replacements(&TargetSubDAG::new(root))
184+
.iter()
185+
.filter(|candidate| matches!(candidate.replacement, Replacement::Summary(_)))
186+
.filter_map(|candidate| self.candidate_cost_estimate(candidate))
187+
.min_by(|a, b| a.value.total_cmp(&b.value));
188+
}
177189
let Replacement::Summary(root) = &candidate.replacement else {
178-
// Exact compositions have a separate measured rate model. Raw
179-
// rewrites have no retained-state estimate in this model.
190+
// Exact compositions have a separate measured rate model.
180191
return None;
181192
};
182193
let dag = planner_types::post_asap::compile_executable_dag(root).ok()?;
@@ -510,6 +521,23 @@ fn intent_accuracy(intent: &AggIntent) -> AccuracyTarget {
510521
}
511522

512523
impl CostModel for ControlPlaneCostModel {
524+
fn summary_support_evidence(
525+
&self,
526+
summary: &planner_types::post_asap::SummaryNode,
527+
) -> Option<bool> {
528+
use planner_types::post_asap::{NonNegativeWeightProof, WeightDomain};
529+
let dag =
530+
planner_types::post_asap::compile_executable_dag(&std::rc::Rc::new(summary.clone()))
531+
.ok()?;
532+
// Counter-weighted heaps now consume explicit rate values. The raw
533+
// ingestion adapter cannot bind that frontier as counter deltas.
534+
let requires_rate_values = dag.nodes.iter().any(|node| matches!(&node.payload,
535+
ExecutableOperatorPayload::SummaryAgg { input, family: SummaryFamilyType::Sketch(kind, _), .. }
536+
if matches!(kind.algorithm(), SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap)
537+
&& matches!(input.weight_domain, WeightDomain::NonNegative { proof: NonNegativeWeightProof::ResetAwareCounterDerivative })));
538+
requires_rate_values.then_some(false)
539+
}
540+
513541
fn candidate_cost(
514542
&self,
515543
candidate: &ReplacementSubDAG,
@@ -869,6 +897,13 @@ impl ForcedFamilyCostModel {
869897
}
870898

871899
impl CostModel for ForcedFamilyCostModel {
900+
fn summary_support_evidence(
901+
&self,
902+
summary: &planner_types::post_asap::SummaryNode,
903+
) -> Option<bool> {
904+
self.inner.summary_support_evidence(summary)
905+
}
906+
872907
fn candidate_cost(
873908
&self,
874909
candidate: &ReplacementSubDAG,

0 commit comments

Comments
 (0)