Skip to content

Commit 9efc252

Browse files
zzylolclaude
andcommitted
perf(asap_types): decode each installed DAG once per validation
PrecomputePlan::validate and validate_catalog_contents decoded every installed DAG inside the per-materialization loop, O(materializations x DAGs) decodes. Decode lazily once per DAG; decode errors still surface only where a DAG is used. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 6c49478 commit 9efc252

2 files changed

Lines changed: 24 additions & 9 deletions

File tree

‎crates/asap_types/src/precompute_plan.rs‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -522,6 +522,12 @@ impl PrecomputePlan {
522522
.chain(input.inputs.iter().copied())
523523
})
524524
.collect();
525+
// Decode each DAG at most once; errors still surface only where used.
526+
let dags = self
527+
.executable_dags
528+
.values()
529+
.map(|installed| (installed, std::cell::OnceCell::new()))
530+
.collect::<Vec<_>>();
525531
for config in &self.materializations {
526532
if !config.population_key_encoding.is_legacy()
527533
&& (self.ingest.protocol != IngestProtocol::PrometheusRemoteWriteV1
@@ -572,11 +578,11 @@ impl PrecomputePlan {
572578
}
573579
validated_source_window_cohort(config, &sources)?;
574580
let mut matched = false;
575-
for installed in self.executable_dags.values() {
576-
let dag = installed
577-
.document
578-
.decode()
579-
.map_err(PrecomputePlanError::CatalogContract)?;
581+
for (installed, dag) in &dags {
582+
let dag = dag
583+
.get_or_init(|| installed.document.decode())
584+
.as_ref()
585+
.map_err(|error| PrecomputePlanError::CatalogContract(error.clone()))?;
580586
for sink in &installed.binding.precompute_sinks {
581587
if !matches!(installed.binding.node(*sink),
582588
Some(crate::executable_plan::BackendNodeBinding::Materialization { stored_output })

‎crates/asap_types/src/precompute_plan/catalog.rs‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,12 @@ impl PrecomputePlan {
5555
"stored output semantics differ from installed writer computation",
5656
));
5757
}
58+
// Decode each DAG at most once; errors still surface only where used.
59+
let dags = self
60+
.executable_dags
61+
.values()
62+
.map(|installed| (installed, std::cell::OnceCell::new()))
63+
.collect::<Vec<_>>();
5864
for config in &self.materializations {
5965
if config.derived_input.is_some() && config.semantic_fragment.is_none() {
6066
return Err(invalid(
@@ -63,16 +69,19 @@ impl PrecomputePlan {
6369
}
6470
if let Some(expected) = &config.semantic_fragment {
6571
let mut found = false;
66-
for installed in self.executable_dags.values() {
67-
let dag = installed.document.decode().map_err(invalid)?;
72+
for (installed, dag) in &dags {
73+
let dag = dag
74+
.get_or_init(|| installed.document.decode())
75+
.as_ref()
76+
.map_err(|error| invalid(error.clone()))?;
6877
for (id, binding) in &installed.binding.nodes {
6978
if matches!(binding, crate::executable_plan::BackendNodeBinding::Materialization { stored_output }
7079
if stored_output.fingerprint() == config.policy_fingerprint())
7180
{
7281
found = true;
7382
let actual = match &self.ingest.dataset_identity {
74-
Some(dataset) => crate::semantic_fragment::SemanticFragment::from_stored_output_in_dataset(&dag, *id, dataset.clone()),
75-
None => crate::semantic_fragment::SemanticFragment::from_stored_output(&dag, *id),
83+
Some(dataset) => crate::semantic_fragment::SemanticFragment::from_stored_output_in_dataset(dag, *id, dataset.clone()),
84+
None => crate::semantic_fragment::SemanticFragment::from_stored_output(dag, *id),
7685
}.map_err(invalid)?;
7786
if &actual != expected {
7887
return Err(invalid(

0 commit comments

Comments
 (0)