feat(cost): add cache evidence to physical plan estimates

crates/asap-aware-mapping/src/physical_plan_cost_model.rs

@@ -4,6 +4,7 @@ use std::{cell::RefCell, rc::Rc};
 
 use asap_types::post_asap::{SketchAlgorithm, SummaryExpr, SummaryNode};
 use asap_types::pre_asap::{AggIntent, QueryExpr};
+use asap_types::resources::CacheProfile;
 
 use crate::analytical_cost::{
     estimate_physical_dag_comparison, AnalyticalCostError,
@@ -27,6 +28,7 @@ use crate::replacement::{Replacement, ReplacementSubDAG, TargetSubDAG};
 pub struct PhysicalEvidenceSnapshot {
     pub version: String,
     pub scope: ComparisonScope,
+    pub cache_profile: CacheProfile,
 }
 
 /// Deployment evidence needed to price one planner alternative.
@@ -90,6 +92,11 @@ impl<'a> PhysicalPlanCostModel<'a> {
         provider: &'a dyn PlannerPhysicalPlanProvider,
         calibration: ResourceCalibration,
     ) -> Result<Self, AnalyticalCostError> {
+        if calibration.version.trim().is_empty() {
+            return Err(AnalyticalCostError::MissingOrStale(
+                "resource_calibration.version",
+            ));
+        }
         calibration.validate()?;
         Ok(Self {
             provider,
@@ -159,12 +166,14 @@ impl<'a> PhysicalPlanCostModel<'a> {
                 root: &raw.root,
                 scope,
                 statistics: &raw,
+                cache_profile: &snapshot.cache_profile,
             },
             PhysicalDagEstimateRequest {
                 nodes: &replacement.nodes,
                 root: &replacement.root,
                 scope,
                 statistics: &replacement,
+                cache_profile: &snapshot.cache_profile,
             },
         )?;
         let raw_cost = Cost(resources.raw.calibrated_cost(&self.calibration)?);
@@ -413,6 +422,7 @@ mod tests {
             Ok(PhysicalEvidenceSnapshot {
                 version: "test-snapshot-1".into(),
                 scope: scope(),
+                cache_profile: CacheProfile::no_cache(),
             })
         }
 
@@ -482,6 +492,99 @@ mod tests {
         );
     }
 
+    // Ranking must retain the uncovered byte of a nearly resident buffer cache.
+    #[test]
+    fn tiny_buffer_misses_still_affect_global_selection() {
+        use crate::analytical_cost::{CacheCapacityEvidence, CacheEvidence};
+        const WORKING_SET: u64 = 1_u64 << 63;
+        struct AlmostResident(TestProvider);
+        impl PlannerPhysicalPlanProvider for AlmostResident {
+            fn capture_evidence_snapshot(
+                &self,
+                target: &TargetSubDAG<'_>,
+            ) -> Result<PhysicalEvidenceSnapshot, AnalyticalCostError> {
+                let mut snapshot = self.0.capture_evidence_snapshot(target)?;
+                snapshot.cache_profile = CacheProfile::Evidence(CacheEvidence {
+                    version: "almost-resident-v1".into(),
+                    distinct_evaluations: 10,
+                    repeated_identical_evaluations: 0,
+                    result_invalidation_ratio: None,
+                    result_cache: CacheCapacityEvidence {
+                        working_set_bytes: 1,
+                        capacity_bytes: 0,
+                    },
+                    buffer_cache: CacheCapacityEvidence {
+                        working_set_bytes: WORKING_SET,
+                        capacity_bytes: WORKING_SET - 1,
+                    },
+                });
+                Ok(snapshot)
+            }
+            fn query_node_evidence(
+                &self,
+                snapshot: &PhysicalEvidenceSnapshot,
+                request: PhysicalNodeRequest<'_>,
+            ) -> Result<PhysicalNodeEvidence, AnalyticalCostError> {
+                let mut evidence = self.0.query_node_evidence(snapshot, request)?;
+                if let OperatorStatistics::Scan {
+                    source_read_bytes, ..
+                } = &mut evidence.statistics
+                {
+                    *source_read_bytes = WORKING_SET;
+                }
+                Ok(evidence)
+            }
+            fn summary_physical_dag(
+                &self,
+                snapshot: &PhysicalEvidenceSnapshot,
+                summary: &Rc<SummaryNode>,
+                target: &TargetSubDAG<'_>,
+            ) -> Result<PhysicalDag, AnalyticalCostError> {
+                self.0.summary_physical_dag(snapshot, summary, target)
+            }
+        }
+        let space = crate::replacement::search_workload_with(
+            vec![("q", query())],
+            &crate::replacement::default_strategies(),
+        );
+        let provider = AlmostResident(TestProvider::new(true, WORKING_SET));
+        let model = PhysicalPlanCostModel::new(
+            &provider,
+            ResourceCalibration {
+                cost_per_cpu_op: 0.0,
+                cost_per_scan_byte: 1.0,
+                cost_per_retained_byte: 0.0,
+                version: "disk-only-v1".into(),
+            },
+        )
+        .unwrap();
+        let selected = space.global_selection(&model);
+        assert!(
+            selected
+                .for_target(&space.roots[0].1)
+                .unwrap()
+                .chosen
+                .is_some(),
+            "one remaining build read must rank ahead of ten remaining raw reads"
+        );
+    }
+
+    // Versioned physical ranking cannot accept an anonymous calibration generation.
+    #[test]
+    fn blank_calibration_version_is_rejected_before_ranking() {
+        let provider = TestProvider::new(true, 800);
+        for version in ["", " \t\n"] {
+            let mut coefficients = calibration();
+            coefficients.version = version.into();
+            assert!(matches!(
+                PhysicalPlanCostModel::new(&provider, coefficients),
+                Err(AnalyticalCostError::MissingOrStale(
+                    "resource_calibration.version"
+                ))
+            ));
+        }
+    }
+
     #[test]
     fn missing_summary_evidence_keeps_the_raw_target() {
         let root = query();
@@ -557,6 +660,7 @@ mod tests {
                 Ok(PhysicalEvidenceSnapshot {
                     version: "  \t".into(),
                     scope: scope(),
+                    cache_profile: CacheProfile::no_cache(),
                 })
             }
 

crates/asap-aware-mapping/src/query_physical_lowering.rs

@@ -1647,6 +1647,7 @@ mod tests {
             })
         };
         let shared_scope = scope(vec![source_coverage]);
+        let no_cache = crate::analytical_cost::CacheProfile::no_cache();
         let shared_dag = lower_query_physical_dag(&root, &shared_scope, &shared_provider).unwrap();
         assert_eq!(shared_dag.nodes.len(), 2);
         assert_eq!(
@@ -1659,12 +1660,14 @@ mod tests {
                 root: &dag.root,
                 scope: &independent_scope,
                 statistics: &dag,
+                cache_profile: &no_cache,
             },
             PhysicalDagEstimateRequest {
                 nodes: &shared_dag.nodes,
                 root: &shared_dag.root,
                 scope: &shared_scope,
                 statistics: &shared_dag,
+                cache_profile: &no_cache,
             },
         )
         .unwrap();