refactor(logical-optimizer): move Stage 1's realization helpers out of the legacy search

crates/executor/src/summary_kernels/univmon.rs

@@ -280,7 +280,7 @@ mod tests {
     /// sizing it is within 1% of a Zipf stream's true L2 norm.
     #[test]
     fn l2_is_layer0_f2_within_the_certified_bound() {
-        use asap_logical_optimizer::pass1::replacement::default_size_params;
+        use asap_logical_optimizer::pass1::realization::default_size_params;
         use planner_types::ir::operator::agg_intent::default_cardinality;
         use planner_types::ir::schema::{SketchAlgorithm, SketchParams};
         let SketchParams::UnivMon {

crates/frontend-sql/tests/frequency_l2.rs

@@ -91,7 +91,7 @@ async fn frequency_l2_preserves_accuracy_target() {
 #[tokio::test]
 async fn stage1_offers_exact_and_summary_l2_alternatives() {
     use asap_logical_optimizer::pass1::logical_candidates::enumerate_local_logical_candidates;
-    use asap_logical_optimizer::pass1::replacement::Realization;
+    use asap_logical_optimizer::pass1::realization::Realization;
     use asap_types::ir::QueryRoot;
     let target = AccuracyTarget::EpsilonDelta {
         epsilon: 0.01,

crates/integration-tests/tests/planner_layering_example2.rs

@@ -4,7 +4,7 @@ mod executor_models;
 mod physical_common;
 use asap_executor::values::Value;
 use asap_frontend_sql::{lower_sql, SqlCatalog};
-use asap_logical_optimizer::pass1::replacement::Realization;
+use asap_logical_optimizer::pass1::realization::Realization;
 use asap_plan_selection::plan_stages;
 use asap_types::ir::operator::AggIntent;
 use asap_types::ir::schema::{DataType, Field, Schema, SketchAlgorithm};

crates/logical-optimizer/src/accuracy/estimators/cms.rs

@@ -41,7 +41,7 @@ mod tests {
 
     #[test]
     fn local_guarantee_inverts_frequency_sizing() {
-        use crate::pass1::replacement::default_size_params;
+        use crate::pass1::realization::default_size_params;
         use asap_types::ir::schema::{GroupingStrategy, SketchKind};
         let c = asap_types::ir::operator::agg_intent::default_cardinality();
         let params = default_size_params(SketchAlgorithm::Cms, &c, 0.01, 0.001);
@@ -69,7 +69,7 @@ mod tests {
     /// for ε/2 and δ/2 (Pass 1's split) it meets it.
     #[test]
     fn hydra_guarantee_adds_the_shared_grid_term() {
-        use crate::pass1::replacement::default_size_params;
+        use crate::pass1::realization::default_size_params;
         use asap_types::ir::schema::{
             default_hydra_params, GroupingStrategy, HydraKind, SketchKind,
         };

crates/logical-optimizer/src/accuracy/estimators/count_sketch.rs

@@ -55,7 +55,7 @@ mod tests {
 
     #[test]
     fn count_sketch_uses_an_l2_guarantee() {
-        use crate::pass1::replacement::default_size_params;
+        use crate::pass1::realization::default_size_params;
         use asap_types::ir::operator::agg_intent::default_cardinality;
         use asap_types::ir::schema::{GroupingStrategy, SketchKind};
         let intent = default_cardinality();

crates/logical-optimizer/src/accuracy/estimators/hll.rs

@@ -230,7 +230,7 @@ mod tests {
     }
     #[test]
     fn generic_rse_sizing_does_not_certify_confidence() {
-        use crate::pass1::replacement::default_size_params;
+        use crate::pass1::realization::default_size_params;
         use asap_types::ir::operator::agg_intent::default_cardinality;
         use asap_types::ir::schema::{GroupingStrategy, SketchKind};
         let c = default_cardinality();

crates/logical-optimizer/src/accuracy/estimators/kll.rs

@@ -42,7 +42,7 @@ mod tests {
 
     #[test]
     fn local_guarantee_inverts_rank_sizing() {
-        use crate::pass1::replacement::default_size_params;
+        use crate::pass1::realization::default_size_params;
         use asap_types::ir::operator::agg_intent::default_quantile;
         use asap_types::ir::schema::{GroupingStrategy, SketchKind};
         let q = default_quantile(0.99);

crates/logical-optimizer/src/accuracy/estimators/mod.rs

@@ -91,7 +91,7 @@ pub(super) fn local_guarantee(
                 hydra_shared_grid_failure_probability: Some((-f64::from(*shared_rows)).exp()),
                 ..Default::default()
             };
-            Some(crate::pass1::grouping::hydra_guarantee(&inner, &stats))
+            Some(hydra_guarantee(&inner, &stats))
         }
         // No accuracy model for the other shared groupings.
         FieldDataType::Sketch(..) => None,
@@ -186,7 +186,7 @@ impl<'a> EstimatorAccuracy<'a> {
         target: Option<&AccuracyTarget>,
     ) -> Self {
         let (epsilon, delta) = target
-            .map(crate::pass1::replacement::accuracy_budget)
+            .map(crate::pass1::realization::accuracy_budget)
             .unwrap_or((0.0, 0.0));
         Self {
             base,
@@ -257,3 +257,93 @@ impl AccuracyModel for EstimatorAccuracy<'_> {
         self.base.answers(statistic, guarantee)
     }
 }
+
+/// Compose the inner per-subpopulation guarantee with Hydra's outer shared
+/// grid. The paper's collision term depends on deployment/data statistics;
+/// keeping those leaves symbolic makes the formula explicit while ensuring
+/// target satisfaction fails closed until a caller supplies them.
+pub(crate) fn hydra_guarantee(
+    inner: &ResultGuarantee,
+    stats: &PropagationStats,
+) -> ResultGuarantee {
+    let mut provenance = inner.provenance.clone();
+    provenance.extend(stats.evidence_provenance.clone());
+    provenance.push(GuaranteeSource::ChildGuarantee {
+        input_index: 0,
+        guarantee: Box::new(inner.clone()),
+    });
+    if stats.hydra_shared_grid_collision_bound.is_none() {
+        provenance.push(GuaranteeSource::UnavailableStatistic {
+            statistic: "hydra_shared_grid_collision_bound".into(),
+        });
+    }
+    if stats.hydra_shared_grid_failure_probability.is_none() {
+        provenance.push(GuaranteeSource::UnavailableStatistic {
+            statistic: "hydra_shared_grid_failure_probability".into(),
+        });
+    }
+    provenance.push(GuaranteeSource::CompositionStep {
+        operator: CompositionOperator::ApproximateAggregate,
+        rule: "hydra_shared_grid_union_bound".into(),
+    });
+    ResultGuarantee {
+        metric: inner.metric,
+        bound: BoundExpr::Sum {
+            terms: vec![
+                inner.bound.clone(),
+                stats.hydra_shared_grid_collision_bound.map_or_else(
+                    || BoundExpr::Unknown {
+                        statistic: "hydra_shared_grid_collision_bound".into(),
+                    },
+                    |value| BoundExpr::Constant { value },
+                ),
+            ],
+        },
+        failure_probability: ProbabilityExpr::UnionBound {
+            terms: vec![
+                inner.failure_probability.clone(),
+                stats.hydra_shared_grid_failure_probability.map_or_else(
+                    || ProbabilityExpr::Unknown {
+                        statistic: "hydra_shared_grid_failure_probability".into(),
+                    },
+                    |value| ProbabilityExpr::Constant { value },
+                ),
+            ],
+        },
+        provenance,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn hydra_composes_inner_and_shared_grid_error_symbolically() {
+        let inner = ResultGuarantee {
+            metric: ErrorMetric::Frequency,
+            bound: BoundExpr::Constant { value: 0.01 },
+            failure_probability: ProbabilityExpr::Constant { value: 0.02 },
+            provenance: vec![],
+        };
+        let composed = hydra_guarantee(&inner, &PropagationStats::default());
+
+        assert_eq!(composed.metric, ErrorMetric::Frequency);
+        assert!(matches!(
+            composed.bound,
+            BoundExpr::Sum { ref terms }
+                if matches!(terms.as_slice(), [
+                    BoundExpr::Constant { value },
+                    BoundExpr::Unknown { statistic },
+                ] if *value == 0.01 && statistic == "hydra_shared_grid_collision_bound")
+        ));
+        assert!(matches!(
+            composed.failure_probability,
+            ProbabilityExpr::UnionBound { ref terms }
+                if matches!(terms.as_slice(), [
+                    ProbabilityExpr::Constant { value },
+                    ProbabilityExpr::Unknown { statistic },
+                ] if *value == 0.02 && statistic == "hydra_shared_grid_failure_probability")
+        ));
+    }
+}

crates/logical-optimizer/src/accuracy/mod.rs

@@ -45,7 +45,7 @@ pub trait AccuracyModel {
     /// The guarantee of reading `query` out of a summary of family `family`
     /// built over an **exact** input — derived from the family's committed
     /// parameters by inverting the same sizing formulas
-    /// [`crate::pass1::replacement::default_size_params`] uses. `None` when this
+    /// [`crate::pass1::realization::default_size_params`] uses. `None` when this
     /// model has no error model for the family (the default has none for
     /// `Sample`/`Wavelet`/`StatModel`).
     fn local_guarantee(

crates/logical-optimizer/src/lib.rs

@@ -66,14 +66,14 @@ pub use pass1::exact_composition::{
 pub use pass1::explanation::{
     explain_replacements, explain_replacements_with, ExplanationKind, ReplacementExplanation,
 };
-pub use pass1::grouping::{has_subpopulations, HydraGroupingStrategy};
+pub use pass1::grouping::HydraGroupingStrategy;
+pub use pass1::realization::{has_subpopulations, summary_candidates, Realization};
 pub use pass1::replacement::{
     default_strategies, is_logical_rewrite, search_workload, search_workload_with,
-    search_workload_with_targets, summary_candidates, ASAPStrategies, CandidateLogicalASAPDAGs,
-    GlobalSelection, Matcher, Proposals, Realization, RealizationError, RejectedCandidate,
-    Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG,
-    SharedSubDAGStrategy, TargetSubDAG, TargetSubDAGCandidates, TargetSubDAGSelection,
-    MAX_SEARCH_ITERATIONS,
+    search_workload_with_targets, ASAPStrategies, CandidateLogicalASAPDAGs, GlobalSelection,
+    Matcher, Proposals, RealizationError, RejectedCandidate, Replacement, ReplacementProvenance,
+    ReplacementStrategy, ReplacementSubDAG, SharedSubDAGStrategy, TargetSubDAG,
+    TargetSubDAGCandidates, TargetSubDAGSelection, MAX_SEARCH_ITERATIONS,
 };
 pub use pass1::rewrite::{AvgToSumOverCountStrategy, SemanticEquivalentRewriteStrategy};
 pub use pass2::reconciliation::AccuracyReconciliationStrategy;

crates/logical-optimizer/src/pass1/exact_composition.rs

@@ -77,9 +77,10 @@ use asap_types::physical::execution_data_state::lift_plain;
 use asap_types::physical::ExactOperationSchemaError;
 use asap_types::types::AccuracyTarget;
 
+use crate::pass1::realization::Realization;
 use crate::pass1::replacement::{
-    bindable_intent, describe_intent, realizations_for_intent, Realization, RealizationError,
-    Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG,
+    bindable_intent, describe_intent, realizations_for_intent, RealizationError, Replacement,
+    ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG,
 };
 use crate::{AccuracyModel, DefaultAccuracyModel, PropagationStats};