diff --git a/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs b/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs
index e18050e22..ea2b891ca 100644
--- a/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs
+++ b/crates/logical-asap-optimization/src/pass1/per_computation_candidates.rs
@@ -113,9 +113,8 @@ use asap_types::ir::scalar::expr_ir::ArithmeticOpKind;
 use asap_types::ir::scalar::{resolve_column_ref, ColumnRef, CompareOpKind, ScalarValue};
 use asap_types::ir::schema::{DataType, Schema};
 use asap_types::ir::schema::{
-    EntityIdentity, ExactKind, ExactParams, FieldDataType, GroupingStrategy,
-    NonNegativeWeightProof, SketchAlgorithm, SketchKind, StatisticKind, SummaryInputExpr,
-    SummaryUpdate, WeightDomain,
+    EntityIdentity, ExactKind, FieldDataType, GroupingStrategy, NonNegativeWeightProof,
+    SketchAlgorithm, SketchKind, StatisticKind, SummaryInputExpr, SummaryUpdate, WeightDomain,
 };
 use asap_types::ir::{
     ASAPOp, BinaryOperator, ExprSemantics, NonASAPOp, Operator, OperatorNode, OperatorResultKind,
@@ -143,21 +142,6 @@ use asap_types::stages::{
     HowToComputeAggregateMeasure, MemoGroupId, PruneReason,
 };
 
-/// `FieldDataType::ExactAggregate` of accumulator `kind`, with its parameters
-/// (none has any, so they mirror `kind`).
-fn exact_family(kind: ExactKind) -> FieldDataType {
-    let params = match kind {
-        ExactKind::Sum => ExactParams::Sum,
-        ExactKind::Count => ExactParams::Count,
-        ExactKind::Min => ExactParams::Min,
-        ExactKind::Max => ExactParams::Max,
-        ExactKind::Rate => ExactParams::Rate,
-        ExactKind::IRate => ExactParams::IRate,
-        ExactKind::Increase => ExactParams::Increase,
-    };
-    FieldDataType::ExactAggregate(kind, params)
-}
-
 /// Unranked candidates of one measure and the candidates pruned for it.
 #[derive(Debug, Clone, PartialEq)]
 pub(super) struct ComputationChoices {
@@ -702,18 +686,6 @@ pub enum PerComputationCandidateError {
     InvalidChoice,
     #[error("unsupported substitution in logical ASAP-aware optimization: {0}")]
     Unsupported(&'static str),
-    /// A Pass 2 rule found more than `max` candidates of one kind, whose
-    /// 2^n subsets it would enumerate: the workload is too large to plan
-    /// this way, not bad input.
-    #[error("found {found} {counted}, more than {max}")]
-    TooManyCandidates {
-        /// What was counted, e.g. "rollup candidates".
-        counted: &'static str,
-        /// How many were found, more than `max`.
-        found: usize,
-        /// The cap that was exceeded.
-        max: usize,
-    },
 }
 
 /// Enumerate exact, rewrite and summary choices of one measure in stable catalog
@@ -1078,7 +1050,7 @@ fn add_nested_sum_targets(targets: &mut [ComputationTarget], logical_nodes: &[Rc
 }
 
 /// The input rows, the inner sum and the outer grouping (moved onto the
-/// input rows by column, [`partitions_over`]) of a nested-sum merge of
+/// input rows by column, [`Reduction::partitions_over`]) of a nested-sum merge of
 /// `target`, or `None` when it does not apply: `target` is a grouped `sum`
 /// (not by `without`) of the value column of a per-series or grouped `sum`
 /// directly beneath it, neither filtered nor with `HAVING`, and not grouped
@@ -1143,7 +1115,7 @@ fn nested_sum_input(target: &OperatorNode) -> Option<(Rc<OperatorNode>, AggInten
     {
         return None;
     }
-    let partitions = partitions_over(reduction, &inner.schema, &rows.schema)?;
+    let partitions = reduction.partitions_over(&inner.schema, &rows.schema)?;
     Some((Rc::clone(rows), inner_sum.clone(), partitions))
 }
 
@@ -1280,7 +1252,7 @@ fn reads_only_time_series(node: &OperatorNode) -> bool {
 /// `target`'s inner aggregate, by the keyed-additive rule, or `None` when it
 /// does not apply. Rows that carry the full series identity rank it as a
 /// column, as [`summary_update`] does. The partitions are `target`'s, moved
-/// onto the input the heap reads ([`partitions_over`]).
+/// onto the input the heap reads ([`Reduction::partitions_over`]).
 fn whole_expression_input(
     target: &OperatorNode,
 ) -> Option<(Rc<OperatorNode>, SummaryUpdate, Reduction)> {
@@ -1326,7 +1298,7 @@ fn whole_expression_input(
             ..
         })
     );
-    let partitions = partitions_over(reduction, &child.schema, &realized.child.schema)?;
+    let partitions = reduction.partitions_over(&child.schema, &realized.child.schema)?;
     if per_series && realized.child.schema.has_promql_series_identity() {
         input.item = Some(asap_types::ir::schema::topk::series_identity_item(
             &realized.child.schema,
@@ -1344,40 +1316,6 @@ fn whole_expression_input(
     Some((realized.child, input, partitions))
 }
 
-/// `reduction`'s keys, which index `inner` (the absorbed aggregate's
-/// output), as indices into `input` (the rows the heap reads), matched by
-/// column. `None` when a key is missing or ambiguous in `input`, or the keys
-/// are given by `without`: `input` carries columns `inner` dropped.
-pub(crate) fn partitions_over(
-    reduction: &Reduction,
-    inner: &Schema,
-    input: &Schema,
-) -> Option<Reduction> {
-    let Reduction::Reduce(keys) = reduction else {
-        return Some(Reduction::PerEntity);
-    };
-    if keys.is_without() {
-        return None;
-    }
-    let column = |schema: &Schema, index: usize| {
-        crate::pass1::realization::column_ref(&schema.fields[index])
-    };
-    let keys = keys
-        .keys()
-        .iter()
-        .map(|&index| {
-            inner.fields.get(index)?;
-            let key = column(inner, index);
-            let mut matches = (0..input.fields.len()).filter(|&i| column(input, i) == key);
-            match (matches.next(), matches.next()) {
-                (Some(i), None) => Some(i),
-                _ => None,
-            }
-        })
-        .collect::<Option<_>>()?;
-    Some(Reduction::by(keys))
-}
-
 /// `update` with the open requirement that its weights are samples of
 /// `metric` ([`WeightDomain::RawDataSamplesOfMetric`]), when it is a keyed heap
 /// update whose weights are not otherwise proven.
@@ -1920,7 +1858,7 @@ pub(crate) fn exact_accumulator(
     input: &Rc<OperatorNode>,
     reduction: Reduction,
 ) -> Result<Rc<OperatorNode>, PerComputationCandidateError> {
-    let family = exact_family(kind);
+    let family = FieldDataType::exact(kind);
     let read = measure_input(intent, input, None)?;
     Ok(OperatorNode::new_shared(Operator::ASAP(
         ASAPOp::SummaryAgg {
@@ -1968,7 +1906,7 @@ fn substitute_measure(
         ))?)
     };
     let finalized = |kind: &ExactKind, intent: &AggIntent, read| {
-        let state = state(exact_family(*kind), intent, read)?;
+        let state = state(FieldDataType::exact(*kind), intent, read)?;
         Ok::<_, PerComputationCandidateError>(OperatorNode::new_shared(Operator::ASAP(
             ASAPOp::FinalizeExactAccumulator { child: state },
         ))?)
@@ -2348,7 +2286,7 @@ fn counted_column(
         schema
             .fields
             .get(id)
-            .map(crate::pass1::realization::column_ref)
+            .map(asap_types::ir::schema::Field::column_ref)
     };
     filter
         .map_or_else(Vec::new, |filter| not_null(&filter.0))
@@ -2410,9 +2348,7 @@ fn summary_update(
                     .iter()
                     .enumerate()
                     .filter(|(index, _)| *index != value && !groups.contains(index))
-                    .map(|(_, field)| {
-                        SummaryInputExpr::Column(crate::pass1::realization::column_ref(field))
-                    })
+                    .map(|(_, field)| SummaryInputExpr::Column(field.column_ref()))
                     .collect();
                 if items.is_empty() {
                     return Err(PerComputationCandidateError::Unsupported(
@@ -2424,7 +2360,7 @@ fn summary_update(
                 let excluding = groups
                     .iter()
                     .filter_map(|&index| child.fields.get(index))
-                    .map(crate::pass1::realization::column_ref)
+                    .map(asap_types::ir::schema::Field::column_ref)
                     .collect();
                 SummaryInputExpr::EntityIdentity(EntityIdentity::PromqlLabelSet { excluding })
             };
@@ -3640,7 +3576,7 @@ mod tests {
         };
         assert_eq!(
             *family,
-            FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum)
+            FieldDataType::ExactAggregate(ExactKind::Sum, asap_types::ir::schema::ExactParams::Sum)
         );
         assert!(matches!(
             child.non_asap(),
@@ -4705,7 +4641,10 @@ mod tests {
         };
         assert_eq!(
             sum_family,
-            &FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum)
+            &FieldDataType::ExactAggregate(
+                ExactKind::Sum,
+                asap_types::ir::schema::ExactParams::Sum
+            )
         );
         assert!(!leaf.contains_asap());
         assert!(child
diff --git a/crates/logical-asap-optimization/src/pass1/realization.rs b/crates/logical-asap-optimization/src/pass1/realization.rs
index 1965c20f3..d5dc1f09c 100644
--- a/crates/logical-asap-optimization/src/pass1/realization.rs
+++ b/crates/logical-asap-optimization/src/pass1/realization.rs
@@ -76,7 +76,7 @@ pub(crate) fn summarised_column(intent: &AggIntent, child_schema: &Schema) -> Co
         .first()
         .and_then(|id| child_schema.fields.get(*id))
     {
-        Some(c) => column_ref(c),
+        Some(c) => c.column_ref(),
         None => ColumnRef::SampleValue,
     }
 }
@@ -87,21 +87,11 @@ pub(crate) fn ranked_value(child: &OperatorNode) -> Result<(ColumnId, ColumnRef)
     let ranked = ranked_column(child)?;
     let column = match resolve_column_ref(&ColumnRef::SampleValue, &child.schema) {
         Ok(value) if value == ranked => ColumnRef::SampleValue,
-        _ => column_ref(&child.schema.fields[ranked]),
+        _ => child.schema.fields[ranked].column_ref(),
     };
     Ok((ranked, column))
 }
 
-pub(crate) fn column_ref(column: &Field) -> ColumnRef {
-    match &column.table {
-        Some(t) => ColumnRef::Qualified {
-            table: t.clone(),
-            name: column.name.clone(),
-        },
-        None => ColumnRef::Named(column.name.clone()),
-    }
-}
-
 /// What the summary consumes per input row. An intent that reads one column (or
 /// none) feeds that column; `COUNT(DISTINCT a, b)` feeds the whole tuple as one
 /// item, so the distinct-count sketch hashes `(a, b)` rather than `a` — the
@@ -123,7 +113,7 @@ pub(crate) fn summarised_input(
     }
     let legs = cols
         .iter()
-        .map(|id| child_schema.fields.get(*id).map(column_ref))
+        .map(|id| child_schema.fields.get(*id).map(Field::column_ref))
         .collect::<Option<Vec<_>>>()?;
     Some(SummaryInputExpr::Tuple(
         legs.into_iter().map(SummaryInputExpr::Column).collect(),
@@ -217,7 +207,7 @@ pub(crate) fn realize_keyed_additive_summary_input(
         }
         [AggIntent::Sum { col }] => SummaryInputExpr::Column(match col {
             None => ColumnRef::SampleValue,
-            Some(index) => match schema_column_ref(raw_child, *index) {
+            Some(index) => match raw_child.schema.fields.get(*index).map(Field::column_ref) {
                 Some(column) => column,
                 None => {
                     // Unsupported: sum-ranked Top-K value column is outside the raw input schema.
@@ -247,7 +237,7 @@ pub(crate) fn realize_keyed_additive_summary_input(
         Reduction::PerEntity => vec![],
         Reduction::Reduce(keys) => keys
             .iter()
-            .filter_map(|index| schema_column_ref(child, *index))
+            .filter_map(|index| child.schema.fields.get(*index).map(Field::column_ref))
             .collect(),
     };
     let item = match reduction {
@@ -257,7 +247,7 @@ pub(crate) fn realize_keyed_additive_summary_input(
         Reduction::Reduce(keys) if !keys.is_without() && !keys.is_empty() => {
             let Some(columns) = keys
                 .iter()
-                .map(|index| schema_column_ref(raw_child, *index))
+                .map(|index| raw_child.schema.fields.get(*index).map(Field::column_ref))
                 .collect::<Option<Vec<_>>>()
             else {
                 // Unsupported: ranked item column is outside the raw input schema.
@@ -357,17 +347,6 @@ fn summed_per_series_samples(grouped: &OperatorNode) -> Option<(Rc<OperatorNode>
     Some((Rc::clone(rows), inner.clone()))
 }
 
-pub(crate) fn schema_column_ref(child: &OperatorNode, index: usize) -> Option<ColumnRef> {
-    let column = child.schema.fields.get(index)?;
-    Some(match &column.table {
-        Some(table) => ColumnRef::Qualified {
-            table: table.clone(),
-            name: column.name.clone(),
-        },
-        None => ColumnRef::Named(column.name.clone()),
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs b/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs
index 55bf79e40..991e96f40 100644
--- a/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs
+++ b/crates/logical-asap-optimization/src/pass2/asap_aware_cse.rs
@@ -110,9 +110,10 @@ use std::rc::Rc;
 
 use asap_types::ir::operator::operator_properties::{Reduction, Source};
 use asap_types::ir::operator::AggIntent;
+use asap_types::ir::scalar::resolve_column_ref;
 use asap_types::ir::schema::{
     ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, MergedError,
-    Schema, SketchKind, PROMQL_SERIES_IDENTITY,
+    Schema, SketchKind, SummaryInputExpr, SummaryUpdate, WeightDomain, PROMQL_SERIES_IDENTITY,
 };
 use asap_types::ir::ScalarExpr;
 use asap_types::ir::SortKey;
@@ -121,10 +122,34 @@ use asap_types::ir::{
     TimeRangeKind,
 };
 
-use crate::pass1::per_computation_candidates::PerComputationCandidateError;
 use asap_accuracy::estimators::size_for_all_readers;
 use asap_accuracy::propagation::recompose_guarantees;
 
+use asap_types::ir::SchemaDerivationError;
+
+/// Why a sharing rewrite builds no candidate. The rules report it as text
+/// ([`Pass2RuleError`](crate::pass2::memo::state_groups::Pass2RuleError)) or
+/// drop the candidate.
+#[derive(Debug, thiserror::Error)]
+pub enum CseError {
+    #[error(transparent)]
+    Structure(#[from] SchemaDerivationError),
+    #[error("unsupported substitution in logical ASAP-aware optimization: {0}")]
+    Unsupported(&'static str),
+    /// A rule found more than `max` candidates of one kind, whose 2^n subsets
+    /// it would enumerate: the workload is too large to plan this way, not
+    /// bad input.
+    #[error("found {found} {counted}, more than {max}")]
+    TooManyCandidates {
+        /// What was counted, e.g. "rollup choices".
+        counted: &'static str,
+        /// How many were found, more than `max`.
+        found: usize,
+        /// The most the rule enumerates.
+        max: usize,
+    },
+}
+
 /// `roots` with the time-series scans of one metric made identical when
 /// they project different labels, so that interning finds them identical: the PromQL frontend projects only the labels each query uses
 /// (`ts, value, job, …` for `quantile by (job) (0.5, data)`), and
@@ -139,9 +164,7 @@ use asap_accuracy::propagation::recompose_guarantees;
 /// and the other not, when a node would change arity, or when a consumer
 /// would have to narrow a summary state or a grouped or filtered tumbling-window chain
 /// would read a `Project`.
-pub fn with_union_scan_labels(
-    roots: &[QueryRoot],
-) -> Result<Option<Vec<QueryRoot>>, PerComputationCandidateError> {
+pub fn with_union_scan_labels(roots: &[QueryRoot]) -> Result<Option<Vec<QueryRoot>>, CseError> {
     let mut nodes = Vec::new();
     let mut seen = HashSet::new();
     for node in roots.iter().flat_map(QueryRoot::operator_refs) {
@@ -310,7 +333,7 @@ fn passes_labels_on(node: &OperatorNode) -> bool {
 fn narrowed(
     wide: &Rc<OperatorNode>,
     schema: &Schema,
-) -> Result<Option<Rc<OperatorNode>>, PerComputationCandidateError> {
+) -> Result<Option<Rc<OperatorNode>>, CseError> {
     use asap_types::ir::ProjectItem;
     if wide.schema.fields == schema.fields {
         return Ok(Some(Rc::clone(wide)));
@@ -370,9 +393,7 @@ fn same_build_family(a: &FieldDataType, b: &FieldDataType) -> bool {
 
 /// One build serving every member: the first member with the members'
 /// shared size ([`size_for_all_readers`]).
-fn sized_for_all(
-    members: &[&Rc<OperatorNode>],
-) -> Result<Option<Rc<OperatorNode>>, PerComputationCandidateError> {
+fn sized_for_all(members: &[&Rc<OperatorNode>]) -> Result<Option<Rc<OperatorNode>>, CseError> {
     // An exact accumulator has one size: the first member serves them all.
     if matches!(
         members[0].asap(),
@@ -468,7 +489,7 @@ pub struct WindowCompositionPass2Candidate {
     /// build has its own chain.
     pub shared_chains: Vec<SharedTumblingWindowChain>,
     /// The candidate's roots with those builds in tumbling windows.
-    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>,
+    pub roots: Result<Vec<QueryRoot>, CseError>,
 }
 
 /// One tumbling-window chain several tumbling-window candidates read (window-composition rule):
@@ -596,9 +617,9 @@ fn splittable_window_ms(build: &OperatorNode) -> Option<u64> {
 pub fn window_composition_pass2_candidates(
     roots: &[QueryRoot],
     candidates: &[TumblingWindowCandidate],
-) -> Result<Vec<WindowCompositionPass2Candidate>, PerComputationCandidateError> {
+) -> Result<Vec<WindowCompositionPass2Candidate>, CseError> {
     if candidates.len() > MAX_TUMBLING_WINDOW_CANDIDATES {
-        return Err(PerComputationCandidateError::TooManyCandidates {
+        return Err(CseError::TooManyCandidates {
             counted: "summary builds that could be built in tumbling windows",
             found: candidates.len(),
             max: MAX_TUMBLING_WINDOW_CANDIDATES,
@@ -1027,7 +1048,7 @@ fn longest(candidates: &[TumblingWindowCandidate], group: &[usize]) -> usize {
 fn shared_chain_build(
     candidates: &[TumblingWindowCandidate],
     group: &[usize],
-) -> Result<Option<Rc<OperatorNode>>, PerComputationCandidateError> {
+) -> Result<Option<Rc<OperatorNode>>, CseError> {
     // Of the longest windows, the finest grouping: a rollup group's host.
     let host = *group
         .iter()
@@ -1053,7 +1074,7 @@ fn in_tumbling_windows(
     roots: &[QueryRoot],
     candidates: &[TumblingWindowCandidate],
     shared_chains: &[(SharedTumblingWindowChain, Rc<OperatorNode>)],
-) -> Result<Vec<QueryRoot>, PerComputationCandidateError> {
+) -> Result<Vec<QueryRoot>, CseError> {
     let mut replaced: HashMap<*const OperatorNode, Rc<OperatorNode>> = HashMap::new();
     for (shared, build) in shared_chains {
         let chain = tumbling_window_chain(&TumblingWindowCandidate {
@@ -1074,7 +1095,7 @@ fn in_tumbling_windows(
             let read = if group_key_count(member) == group_key_count(build) {
                 merge
             } else {
-                regrouped(member, &merge).ok_or(PerComputationCandidateError::Unsupported(
+                regrouped(member, &merge).ok_or(CseError::Unsupported(
                     "a coarser tumbling-window build whose keys the shared chain does not hold",
                 ))?
             };
@@ -1128,7 +1149,7 @@ pub(crate) fn with_tumbling_window_chain(
     family: FieldDataType,
     tumbling_window_width_ms: u64,
     tumbling_window_count: u32,
-) -> Result<Vec<QueryRoot>, PerComputationCandidateError> {
+) -> Result<Vec<QueryRoot>, CseError> {
     let Some(ASAPOp::SummaryMerge {
         children,
         newest_tumbling_windows: own_newest,
@@ -1203,7 +1224,7 @@ pub(crate) fn with_resized_build(
     roots: &[QueryRoot],
     build: &Rc<OperatorNode>,
     family: FieldDataType,
-) -> Result<Vec<QueryRoot>, PerComputationCandidateError> {
+) -> Result<Vec<QueryRoot>, CseError> {
     let Some(ASAPOp::SummaryAgg {
         child,
         input,
@@ -1253,7 +1274,7 @@ pub(crate) fn with_resized_build(
 /// output names.
 fn tumbling_window_chain(
     candidate: &TumblingWindowCandidate,
-) -> Result<Rc<OperatorNode>, PerComputationCandidateError> {
+) -> Result<Rc<OperatorNode>, CseError> {
     let Some(ASAPOp::SummaryAgg {
         child: range,
         family,
@@ -1293,7 +1314,7 @@ fn rebuild(
     node: &Rc<OperatorNode>,
     replaced: &HashMap<*const OperatorNode, Rc<OperatorNode>>,
     rebuilt: &mut HashMap<*const OperatorNode, Rc<OperatorNode>>,
-) -> Result<Rc<OperatorNode>, PerComputationCandidateError> {
+) -> Result<Rc<OperatorNode>, CseError> {
     if let Some(done) = rebuilt.get(&Rc::as_ptr(node)) {
         return Ok(Rc::clone(done));
     }
@@ -1350,7 +1371,7 @@ pub struct RollupPass2Candidate {
     /// The candidate's roots with those builds rolled up and every guarantee
     /// composed again; `Err` when the rebuilt DAG does not derive or
     /// compose, which drops only this candidate.
-    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>,
+    pub roots: Result<Vec<QueryRoot>, CseError>,
 }
 
 /// The most rollup or top-k reuse candidates one candidate may have: every
@@ -1836,7 +1857,6 @@ fn rolled_up(
     finer: &Rc<OperatorNode>,
     coarse_value: &OperatorNode,
 ) -> Option<Rc<OperatorNode>> {
-    use crate::pass1::per_computation_candidates::{exact_accumulator, partitions_over};
     let Some(ASAPOp::SummaryAgg {
         child,
         family,
@@ -1892,7 +1912,7 @@ fn rolled_up(
     {
         return None;
     }
-    let regrouped = partitions_over(reduction, &child.schema, &finer.schema)?;
+    let regrouped = reduction.partitions_over(&child.schema, &finer.schema)?;
     if regrouped
         .group_keys()
         .is_some_and(|keys| keys.iter().any(|&k| k >= value))
@@ -1900,12 +1920,23 @@ fn rolled_up(
         return None;
     }
     let kind = rollup_kind(family)?;
-    let intent = match kind {
-        ExactKind::Min => AggIntent::Min { col: Some(value) },
-        ExactKind::Max => AggIntent::Max { col: Some(value) },
-        _ => AggIntent::Sum { col: Some(value) },
-    };
-    let state = exact_accumulator(kind, &intent, finer, regrouped).ok()?;
+    // An exact accumulator over the finer value column: what Pass 1 builds
+    // for a Sum, Min or Max of that column (no item, signed weights).
+    let weight = finer.schema.fields[value].column_ref();
+    resolve_column_ref(&weight, &finer.schema).ok()?;
+    let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+        child: Rc::clone(finer),
+        input: SummaryUpdate {
+            item: None,
+            weight: SummaryInputExpr::Column(weight),
+            weight_domain: WeightDomain::UnknownOrSigned,
+        },
+        family: FieldDataType::exact(kind),
+        reduction: regrouped,
+        grouping: GroupingStrategy::default(),
+        filter: None,
+    }))
+    .ok()?;
     let mut finalized =
         OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator {
             child: state,
@@ -1981,7 +2012,7 @@ fn is_sketch_build(node: &OperatorNode) -> bool {
 pub fn rollup_pass2_candidates(
     roots: &[QueryRoot],
     candidates: &[RollupCandidate],
-) -> Result<Vec<RollupPass2Candidate>, PerComputationCandidateError> {
+) -> Result<Vec<RollupPass2Candidate>, CseError> {
     Ok(subsets(&rollup_choices(candidates), "rollup")?
         .into_iter()
         .map(|choices| {
@@ -2021,11 +2052,9 @@ pub fn rollup_pass2_candidates(
 /// coarse builds have no shared size, or a node does not derive.
 fn sketch_rollups(
     choices: &[Vec<RollupCandidate>],
-) -> Result<Vec<(*const OperatorNode, Rc<OperatorNode>)>, PerComputationCandidateError> {
+) -> Result<Vec<(*const OperatorNode, Rc<OperatorNode>)>, CseError> {
     let unsupported = || {
-        PerComputationCandidateError::Unsupported(
-            "a sketch rollup whose finest and coarse builds have no shared size",
-        )
+        CseError::Unsupported("a sketch rollup whose finest and coarse builds have no shared size")
     };
     // A finest build can also be the coarse build of another choice, read
     // through a compensating filter. Each build is replaced once: as a
@@ -2075,13 +2104,10 @@ fn sketch_rollups(
 
 /// Every non-empty subset of `candidates`, in binary-counter order (the
 /// first candidate the lowest bit); an error past [`MAX_REUSE_CANDIDATES`].
-fn subsets<T: Clone>(
-    candidates: &[T],
-    rule: &'static str,
-) -> Result<Vec<Vec<T>>, PerComputationCandidateError> {
+fn subsets<T: Clone>(candidates: &[T], rule: &'static str) -> Result<Vec<Vec<T>>, CseError> {
     let n = candidates.len();
     if n > MAX_REUSE_CANDIDATES {
-        return Err(PerComputationCandidateError::TooManyCandidates {
+        return Err(CseError::TooManyCandidates {
             counted: match rule {
                 "rollup" => "rollup choices",
                 _ => "top-k reuse candidates",
@@ -2106,12 +2132,12 @@ fn subsets<T: Clone>(
 fn replace_all(
     roots: &[QueryRoot],
     replaced: &HashMap<*const OperatorNode, Rc<OperatorNode>>,
-) -> Result<Vec<QueryRoot>, PerComputationCandidateError> {
+) -> Result<Vec<QueryRoot>, CseError> {
     fn walk(
         node: &Rc<OperatorNode>,
         replaced: &HashMap<*const OperatorNode, Rc<OperatorNode>>,
         rebuilt: &mut HashMap<*const OperatorNode, Rc<OperatorNode>>,
-    ) -> Result<Rc<OperatorNode>, PerComputationCandidateError> {
+    ) -> Result<Rc<OperatorNode>, CseError> {
         if let Some(done) = rebuilt.get(&Rc::as_ptr(node)) {
             return Ok(Rc::clone(done));
         }
diff --git a/crates/logical-asap-optimization/src/pass2/memo/enumeration.rs b/crates/logical-asap-optimization/src/pass2/memo/enumeration.rs
index 21d4d1e36..d849a5e75 100644
--- a/crates/logical-asap-optimization/src/pass2/memo/enumeration.rs
+++ b/crates/logical-asap-optimization/src/pass2/memo/enumeration.rs
@@ -19,10 +19,9 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::rc::Rc;
 
-use asap_types::ir::{OperatorNode, QueryRoot};
+use asap_types::ir::{OperatorNode, QueryRoot, SchemaDerivationError};
 
 use super::rules::reuse::replace;
-use crate::pass1::per_computation_candidates::PerComputationCandidateError;
 use asap_accuracy::propagation::recompose_guarantees;
 use asap_types::stages::Memo;
 use asap_types::stages::MemoGroupId;
@@ -61,7 +60,7 @@ pub enum ErrorBuildingLogicalASAPDagsFromChosenCandidates {
     Cycle { group: MemoGroupId },
     /// Re-deriving the composed nodes' schemas and guarantees failed.
     #[error("recomposing the composed DAG: {0}")]
-    Recompose(#[from] PerComputationCandidateError),
+    Recompose(#[from] SchemaDerivationError),
 }
 
 /// Every choice of one candidate per needed group, starting from the query
@@ -198,7 +197,7 @@ pub fn build_logical_asap_dags_from_chosen_candidates(
         .collect::<Result<Vec<_>, ErrorBuildingLogicalASAPDagsFromChosenCandidates>>()?
         .into_iter()
         .unzip();
-    let roots = recompose_guarantees(&roots).map_err(PerComputationCandidateError::from)?;
+    let roots = recompose_guarantees(&roots)?;
     Ok(groups
         .into_iter()
         .zip(roots)
diff --git a/crates/logical-asap-optimization/tests/pass_layering.rs b/crates/logical-asap-optimization/tests/pass_layering.rs
index 76f9fd92e..daf0595a7 100644
--- a/crates/logical-asap-optimization/tests/pass_layering.rs
+++ b/crates/logical-asap-optimization/tests/pass_layering.rs
@@ -1,21 +1,26 @@
-//! Pass 2 builds on Pass 1, never the reverse (#1096): Pass 1's production
-//! code does not import `crate::pass2`. Test modules may use Pass 2 to build
-//! fixtures. Pass 2 → Pass 1 is still allowed (#1191).
+//! Stage 1's passes are independent modules (#1096, #1191): neither pass's
+//! production code imports the other. Test modules may use the other pass to
+//! build fixtures. IR helpers both passes need live in `asap-types`.
 
-use std::path::Path;
+use std::path::{Path, PathBuf};
 
-/// Pass 1's production source does not reference `pass2`.
-#[test]
-fn pass1_does_not_import_pass2() {
-    let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
-    let mut files = vec![src.join("pass1.rs")];
-    for entry in std::fs::read_dir(src.join("pass1")).unwrap() {
-        let path = entry.unwrap().path();
-        // `*_tests.rs` files are `#[cfg(test)]` modules.
-        if !path.to_string_lossy().ends_with("_tests.rs") {
-            files.push(path);
+/// Production lines under `src/<pass>.rs` and `src/<pass>/` that name a path
+/// through `needle`'s module.
+fn production_lines_mentioning(pass: &str, needle: &str) -> Vec<String> {
+    fn walk(dir: &Path, files: &mut Vec<PathBuf>) {
+        for entry in std::fs::read_dir(dir).unwrap() {
+            let path = entry.unwrap().path();
+            if path.is_dir() {
+                walk(&path, files);
+            // `*_tests.rs` files are `#[cfg(test)]` modules.
+            } else if !path.to_string_lossy().ends_with("_tests.rs") {
+                files.push(path);
+            }
         }
     }
+    let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
+    let mut files = vec![src.join(format!("{pass}.rs"))];
+    walk(&src.join(pass), &mut files);
     let mut offending = Vec::new();
     for file in files {
         let text = std::fs::read_to_string(&file).unwrap();
@@ -23,14 +28,32 @@ fn pass1_does_not_import_pass2() {
         // parse with `syn` if inline `#[cfg(test)]` items ever need excluding.
         let production = text.split("#[cfg(test)]\nmod tests").next().unwrap();
         for (n, line) in production.lines().enumerate() {
-            if line.contains("pass2") {
+            if line.contains(needle) {
                 offending.push(format!("{}:{}: {}", file.display(), n + 1, line.trim()));
             }
         }
     }
+    offending
+}
+
+/// Pass 1's production source does not reference `pass2`.
+#[test]
+fn pass1_does_not_import_pass2() {
+    let offending = production_lines_mentioning("pass1", "pass2::");
     assert!(
         offending.is_empty(),
         "Pass 1 imports Pass 2:\n{}",
         offending.join("\n")
     );
 }
+
+/// Pass 2's production source does not reference `pass1`.
+#[test]
+fn pass2_does_not_import_pass1() {
+    let offending = production_lines_mentioning("pass2", "pass1::");
+    assert!(
+        offending.is_empty(),
+        "Pass 2 imports Pass 1:\n{}",
+        offending.join("\n")
+    );
+}
diff --git a/crates/types/src/ir/operator/operator_properties.rs b/crates/types/src/ir/operator/operator_properties.rs
index 0edf68a59..e2a6ef1da 100644
--- a/crates/types/src/ir/operator/operator_properties.rs
+++ b/crates/types/src/ir/operator/operator_properties.rs
@@ -447,6 +447,38 @@ pub enum Reduction<C = ColumnId> {
     PerEntity,
 }
 
+impl Reduction {
+    /// These keys, which index `inner`, as indices into `input`, matched by
+    /// column. `None` when a key is missing or ambiguous in `input`, or the
+    /// keys are given by `without`: `input` carries columns `inner` dropped.
+    pub fn partitions_over(
+        &self,
+        inner: &crate::ir::schema::Schema,
+        input: &crate::ir::schema::Schema,
+    ) -> Option<Reduction> {
+        let Reduction::Reduce(keys) = self else {
+            return Some(Reduction::PerEntity);
+        };
+        if keys.is_without() {
+            return None;
+        }
+        let keys = keys
+            .keys()
+            .iter()
+            .map(|&index| {
+                let key = inner.fields.get(index)?.column_ref();
+                let mut matches =
+                    (0..input.fields.len()).filter(|&i| input.fields[i].column_ref() == key);
+                match (matches.next(), matches.next()) {
+                    (Some(i), None) => Some(i),
+                    _ => None,
+                }
+            })
+            .collect::<Option<_>>()?;
+        Some(Reduction::by(keys))
+    }
+}
+
 impl<C> Reduction<C> {
     /// Shorthand for the common case — group by these (possibly empty)
     /// keys, kept rather than excluded.
@@ -475,3 +507,41 @@ impl<C> Reduction<C> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ir::schema::{DataType, Field, Schema};
+
+    /// A key moves to the input's column of the same name and qualifier; a
+    /// key that is missing or ambiguous there, or a `without` grouping, has no
+    /// partitions over the input.
+    #[test]
+    fn partitions_over_matches_keys_by_column() {
+        let field = |name: &str| Field::plain(name, DataType::Utf8, false);
+        let inner = Schema::new(vec![field("job"), field("value")]);
+        let input = Schema::new(vec![field("ts"), field("job"), field("value")]);
+        assert_eq!(
+            Reduction::by(vec![0]).partitions_over(&inner, &input),
+            Some(Reduction::by(vec![1]))
+        );
+        assert_eq!(
+            Reduction::PerEntity.partitions_over(&inner, &input),
+            Some(Reduction::PerEntity)
+        );
+        let ambiguous = Schema::new(vec![field("job"), field("job")]);
+        assert_eq!(
+            Reduction::by(vec![0]).partitions_over(&inner, &ambiguous),
+            None
+        );
+        let qualified = Schema::new(vec![field("job").with_table("t")]);
+        assert_eq!(
+            Reduction::by(vec![0]).partitions_over(&inner, &qualified),
+            None
+        );
+        assert_eq!(
+            Reduction::Reduce(GroupKeys::without(vec![0])).partitions_over(&inner, &input),
+            None
+        );
+    }
+}
diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs
index 6b4196ceb..90796a76b 100644
--- a/crates/types/src/ir/schema/mod.rs
+++ b/crates/types/src/ir/schema/mod.rs
@@ -83,6 +83,18 @@ impl<T> Field<T> {
         self.table = Some(table.into());
         self
     }
+
+    /// A reference to this field: qualified by its table when it has one.
+    pub fn column_ref(&self) -> crate::ir::scalar::ColumnRef {
+        use crate::ir::scalar::ColumnRef;
+        match &self.table {
+            Some(table) => ColumnRef::Qualified {
+                table: table.clone(),
+                name: self.name.clone(),
+            },
+            None => ColumnRef::Named(self.name.clone()),
+        }
+    }
 }
 
 impl Field<FieldDataType> {
@@ -168,6 +180,21 @@ pub enum FieldDataType {
 }
 
 impl FieldDataType {
+    /// The state of exact accumulator `kind`, with its parameters (none has
+    /// any, so they mirror `kind`).
+    pub fn exact(kind: ExactKind) -> Self {
+        let params = match kind {
+            ExactKind::Sum => ExactParams::Sum,
+            ExactKind::Count => ExactParams::Count,
+            ExactKind::Min => ExactParams::Min,
+            ExactKind::Max => ExactParams::Max,
+            ExactKind::Rate => ExactParams::Rate,
+            ExactKind::IRate => ExactParams::IRate,
+            ExactKind::Increase => ExactParams::Increase,
+        };
+        FieldDataType::ExactAggregate(kind, params)
+    }
+
     pub(crate) fn is_plain(&self) -> bool {
         matches!(self, FieldDataType::Plain(_))
     }
diff --git a/crates/types/src/ir/schema/topk.rs b/crates/types/src/ir/schema/topk.rs
index 11ab729f3..1ad5c2186 100644
--- a/crates/types/src/ir/schema/topk.rs
+++ b/crates/types/src/ir/schema/topk.rs
@@ -130,15 +130,7 @@ pub fn series_identity_item(
                 && field.dtype == FieldDataType::Plain(DataType::Utf8)
                 && field.name != PROMQL_SERIES_IDENTITY
         })
-        .map(|(_, field)| {
-            SummaryInputExpr::Column(match &field.table {
-                Some(table) => ColumnRef::Qualified {
-                    table: table.clone(),
-                    name: field.name.clone(),
-                },
-                None => ColumnRef::Named(field.name.clone()),
-            })
-        })
+        .map(|(_, field)| SummaryInputExpr::Column(field.column_ref()))
         .collect();
     items.push(SummaryInputExpr::Column(ColumnRef::Named(
         PROMQL_SERIES_IDENTITY.into(),