fix(engine): bound table gap analysis memory, analyse expression columns

core/engine/src/analysis/table/merge.rs

@@ -86,7 +86,7 @@ impl VerifyTable<'_> {
             .map(|mut row| {
                 for &col in &row.stale {
                     if let CellConstraint::Known(set) = &row.cells[col] {
-                        if let Some(text) = CellText::of(set, &ValueSet::all(), false) {
+                        if let Some(text) = self.cell_text(col, set, &ValueSet::all(), false) {
                             row.rule
                                 .insert(self.inputs[col].id.clone(), Arc::from(text.as_str()));
                         }
@@ -366,7 +366,8 @@ impl VerifyTable<'_> {
             return false;
         }
         let union = a.union(&b);
-        let Some(text) = CellText::of(&union, &ValueSet::all(), self.inputs[col].dated) else {
+        let Some(text) = self.cell_text(col, &union, &ValueSet::all(), self.inputs[col].dated)
+        else {
             return false;
         };
         let id = self.inputs[col].id.clone();

core/engine/src/analysis/table/missing.rs

@@ -258,9 +258,10 @@ impl VerifyTable<'_> {
                     if !dim.input {
                         example = None;
                     }
-                    if let Some(cells) = cells.as_mut() {
-                        if let Some(&first) = dim.columns.first() {
-                            cells.push((self.inputs[first].id.clone(), text.clone()));
+                    if let (Some(list), Some(&first)) = (cells.as_mut(), dim.columns.first()) {
+                        match self.cell_text(first, set, &dim.domain, dim.dated) {
+                            Some(cell) => list.push((self.inputs[first].id.clone(), cell)),
+                            None => cells = None,
                         }
                     }
                     parts.push((dim.label.clone(), text));
@@ -389,7 +390,11 @@ impl Gaps<'_> {
             if sets.iter().all(|set| set == bounds) {
                 continue;
             }
-            let groups = Partition::groups(bounds, &sets);
+            let partition = Partition::new(bounds, &sets);
+            if partition.touches(&sets) > self.work {
+                return None;
+            }
+            let groups = partition.groups(&sets);
             if groups.len() < 2 {
                 continue;
             }

core/engine/src/analysis/table/mod.rs

@@ -78,6 +78,7 @@ impl TableColumn {
             prefer: resolved.and_then(Self::kind_of),
             label,
             domain: field_type.and_then(Self::domain_of),
+            subject: None,
         }
     }
 

core/engine/src/analysis/table/partition.rs

@@ -94,8 +94,49 @@ pub(super) struct Partition {
 }
 
 impl Partition {
-    pub(super) fn groups(region: &ValueSet, sets: &[ValueSet]) -> Vec<(ValueSet, Vec<usize>)> {
-        let partition = Self::new(region, sets);
+    pub(super) fn touches(&self, sets: &[ValueSet]) -> usize {
+        let mut present = Vec::with_capacity(self.pieces.len() + 1);
+        present.push(0usize);
+        for piece in &self.pieces {
+            present.push(present[present.len() - 1] + usize::from(piece.is_some()));
+        }
+        sets.iter()
+            .map(|set| {
+                let strings = match &set.strings {
+                    StringSet::Finite(keys) => keys
+                        .iter()
+                        .filter(|k| self.strings.contains_key(*k))
+                        .count(),
+                    StringSet::CoFinite(excluded) => {
+                        let listed_excluded = excluded
+                            .iter()
+                            .filter(|k| self.strings.contains_key(*k))
+                            .count();
+                        self.strings.len() - listed_excluded
+                            + usize::from(self.string_rest.is_some())
+                    }
+                };
+                let numbers: usize = set
+                    .numbers
+                    .intervals()
+                    .iter()
+                    .filter_map(|interval| self.points.range(interval))
+                    .map(|(first, last)| present[last + 1] - present[first])
+                    .sum();
+                let bools = [ValueSet::FALSE, ValueSet::TRUE]
+                    .into_iter()
+                    .enumerate()
+                    .filter(|(bit, slot)| set.bools & slot != 0 && self.bools[*bit].is_some())
+                    .count();
+                let null = usize::from(set.null && self.null.is_some());
+                let other = usize::from(set.other && self.other.is_some());
+                strings + numbers + bools + null + other
+            })
+            .sum()
+    }
+
+    pub(super) fn groups(&self, sets: &[ValueSet]) -> Vec<(ValueSet, Vec<usize>)> {
+        let partition = self;
         let mut touched: Vec<Vec<usize>> = vec![Vec::new(); partition.atoms.len()];
         for (id, set) in sets.iter().enumerate() {
             for atom in partition.touching(set) {
@@ -120,7 +161,7 @@ impl Partition {
             .collect()
     }
 
-    fn new(region: &ValueSet, sets: &[ValueSet]) -> Self {
+    pub(super) fn new(region: &ValueSet, sets: &[ValueSet]) -> Self {
         let mut partition = Self {
             atoms: Vec::new(),
             strings: HashMap::default(),
@@ -246,3 +287,34 @@ impl Partition {
         out
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn above(n: i64) -> ValueSet {
+        ValueSet::numbers(NumberSet::from_intervals(vec![Interval::new(
+            Bound::Exclusive(Decimal::from(n)),
+            Bound::Unbounded,
+        )]))
+    }
+
+    #[test]
+    fn touches_counts_what_groups_list() {
+        let mut sets: Vec<ValueSet> = (0..40).map(above).collect();
+        sets.extend((0..10).map(|i| ValueSet::string(&format!("s{i}"))));
+        sets.push(ValueSet::string("s1").complement());
+        sets.push(ValueSet::bool(true).union(&ValueSet::null()));
+        sets.push(above(5).union(&ValueSet::string("s2")));
+        let region = ValueSet::all();
+        let partition = Partition::new(&region, &sets);
+        let listed: usize = sets.iter().map(|set| partition.touching(set).len()).sum();
+        assert_eq!(partition.touches(&sets), listed);
+        let grouped: usize = partition
+            .groups(&sets)
+            .iter()
+            .map(|(_, ids)| ids.len())
+            .sum();
+        assert!(grouped <= partition.touches(&sets));
+    }
+}

core/engine/src/analysis/table/verify.rs

@@ -7,7 +7,7 @@ use zen_expression::intellisense::IntelliSense;
 
 use super::cell::CellConstraint;
 use super::index::RowIndex;
-use super::print::DateDay;
+use super::print::{CellText, DateDay};
 use super::value_set::{ValueKind, ValueSet};
 
 pub(crate) const MAX_INPUTS: usize = 30;
@@ -35,6 +35,7 @@ pub(crate) struct VerifyInput {
     pub(crate) prefer: Option<ValueKind>,
     pub(crate) label: Arc<str>,
     pub(crate) domain: Option<ValueSet>,
+    pub(crate) subject: Option<Vec<Rc<str>>>,
 }
 
 #[derive(Debug, Clone)]
@@ -116,8 +117,77 @@ impl Budget {
     }
 }
 
-impl VerifyTable<'_> {
+impl<'a> VerifyTable<'a> {
     pub(crate) fn verify(&self, is: &mut IntelliSense) -> Vec<Finding> {
+        match self.with_expression_subjects(is) {
+            Some(resolved) => resolved.verify_resolved(is),
+            None => self.verify_resolved(is),
+        }
+    }
+
+    fn with_expression_subjects(&self, is: &mut IntelliSense) -> Option<VerifyTable<'a>> {
+        let mut inputs = self.inputs.clone();
+        let mut changed = false;
+        for input in inputs.iter_mut().filter(|input| input.field.is_none()) {
+            let mut subject: Option<Vec<Rc<str>>> = None;
+            let mut consistent = true;
+            for rule in self.rules {
+                let source = rule.get(&input.id).map(|c| c.as_ref()).unwrap_or("");
+                if source.trim().is_empty() {
+                    continue;
+                }
+                match (CellConstraint::subject_of(is, source), &subject) {
+                    (Some(found), None) => subject = Some(found),
+                    (Some(found), Some(known)) if found == *known => {}
+                    _ => {
+                        consistent = false;
+                        break;
+                    }
+                }
+            }
+            let (true, Some(subject)) = (consistent, subject) else {
+                continue;
+            };
+            let field: Arc<str> = Arc::from(subject.join("."));
+            let plain = subject.iter().all(|segment| {
+                let mut chars = segment.chars();
+                chars
+                    .next()
+                    .is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
+                    && chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
+            });
+            input.analyzable = true;
+            input.path = plain.then(|| field.clone());
+            if input.label.trim().is_empty() {
+                input.label = field.clone();
+            }
+            input.field = Some(field);
+            input.subject = Some(subject);
+            changed = true;
+        }
+        changed.then(|| VerifyTable {
+            mode: self.mode,
+            inputs,
+            outputs: self.outputs.clone(),
+            rules: self.rules,
+            fallible: self.fallible.clone(),
+        })
+    }
+
+    pub(super) fn cell_text(
+        &self,
+        col: usize,
+        set: &ValueSet,
+        domain: &ValueSet,
+        dated: bool,
+    ) -> Option<String> {
+        match &self.inputs[col].subject {
+            Some(subject) => CellText::standard(set, domain, dated, &subject.join(".")),
+            None => CellText::of(set, domain, dated),
+        }
+    }
+
+    fn verify_resolved(&self, is: &mut IntelliSense) -> Vec<Finding> {
         let mut findings = Vec::new();
         if self.inputs.is_empty() {
             return findings;
@@ -130,7 +200,18 @@ impl VerifyTable<'_> {
                     .iter()
                     .map(|col| {
                         let source = rule.get(&col.id).map(|c| c.as_ref()).unwrap_or("");
-                        CellConstraint::parse(is, source, col.unary, col.analyzable, col.dated)
+                        match &col.subject {
+                            Some(subject) => {
+                                CellConstraint::parse_subject(is, source, subject, col.dated)
+                            }
+                            None => CellConstraint::parse(
+                                is,
+                                source,
+                                col.unary,
+                                col.analyzable,
+                                col.dated,
+                            ),
+                        }
                     })
                     .collect()
             })

core/engine/tests/table_verification.rs

@@ -686,12 +686,17 @@ fn opaque_cells_never_create_gaps() {
     }
     .assert_gaps(
         Severity::Hint,
-        "no row matches 1 input case: In 1 false",
-        json!([{
-            "cells": { "i1": "false" },
-            "description": "In 1 false",
-            "example": { "applicant": { "vip": false } }
-        }]),
+        "no row matches 2 input cases: In 0 <= 10; In 0 > 10 and In 1 false",
+        json!([
+            {
+                "cells": { "i0": "applicant.age <= 10" },
+                "description": "In 0 <= 10"
+            },
+            {
+                "cells": { "i0": "applicant.age > 10", "i1": "false" },
+                "description": "In 0 > 10 and In 1 false"
+            }
+        ]),
     );
 }
 
@@ -1831,3 +1836,72 @@ async fn collect_compression_keeps_the_order_of_results() {
         assert_eq!(before, after, "{input}");
     }
 }
+
+#[test]
+fn expression_columns_on_one_field_compress_like_field_columns() {
+    Table {
+        hit: "first",
+        inputs: &[""],
+        outputs: &["applicant.discount"],
+        rows: &[
+            ("r1", &["applicant.age > 50"], &["true"]),
+            ("r2", &["applicant.age > 30"], &["true"]),
+        ],
+    }
+    .assert_compressed(Some((2, &["i0=applicant.age > 30 o0=true"])));
+}
+
+#[test]
+fn expression_columns_on_one_field_report_rows_that_never_fire() {
+    let table = Table {
+        hit: "first",
+        inputs: &[""],
+        outputs: &["applicant.discount"],
+        rows: &[
+            ("r1", &["applicant.age > 30"], &["1"]),
+            ("r2", &["applicant.age > 50 and applicant.age < 60"], &["2"]),
+        ],
+    };
+    let policy = table.policy_findings();
+    assert_eq!(policy, table.graph_findings());
+    assert_eq!(policy.len(), 1, "{policy:?}");
+    assert!(policy[0].contains("r2"), "{policy:?}");
+}
+
+#[test]
+fn expression_columns_write_missing_cases_as_expressions() {
+    let table = Table {
+        hit: "first",
+        inputs: &[""],
+        outputs: &["applicant.discount"],
+        rows: &[("r1", &["applicant.age > 30"], &["1"])],
+    };
+    for (kind, diagnostics) in [
+        ("policy", table.policy_diagnostics()),
+        ("graph", table.graph_diagnostics()),
+    ] {
+        let gaps = Table::gaps(diagnostics).unwrap_or_else(|| panic!("{kind}: no MissingCases"));
+        assert_eq!(
+            gaps.cases[0]["cells"]["i0"],
+            json!("applicant.age <= 30"),
+            "{kind}"
+        );
+    }
+}
+
+#[test]
+fn expression_columns_mixing_fields_stay_opaque() {
+    let table = Table {
+        hit: "first",
+        inputs: &[""],
+        outputs: &["applicant.discount"],
+        rows: &[
+            ("r1", &["applicant.age > 30"], &["1"]),
+            ("r2", &["applicant.tier == \"gold\""], &["1"]),
+            ("r3", &["applicant.age > 50"], &["1"]),
+        ],
+    };
+    table.assert_compressed(None);
+    assert!(table.policy_findings().is_empty());
+    assert!(table.graph_findings().is_empty());
+}