//! Automatic projection-pushdown desugar — the Rust port of the Julia
//! reference (`pkg/EarthSciAST.jl/src/pushdown_rewrite.jl`) and its Python
//! mirror (`earthsci_ast/pushdown_rewrite.py`), current as of the Phase-1
//! clean consolidation (idempotency guard + record/gate helpers) and the
//! Phase-3 deterministic `applies_to`.
//!
//! A pre-build model-transform pass that recognises the ISRM-shaped
//! `+`-semiring "apply a provider-backed full-domain array to a sparsely
//! supported binned factor" pattern in a CLEAN model and AUTO-GENERATES the
//! four hand-authored constructs (derived IndexSet + `distinct` producer +
//! `member_factor` + `gated_select` record) so the existing value-invention /
//! gated-provider pipeline runs unchanged. The author writes NO derived set,
//! NO producer, and NO gated_select — only the natural math.
//!
//! This is a NARROW desugarer (a pattern recogniser), NOT a general optimizer.
//! It fires ONLY when the reduction's semiring is the additive `(+, 0)`
//! monoid; a `max_product` / `min_sum` / etc. aggregate of the SAME shape is
//! left untouched (the soundness guard).
//!
//! DESIGN DECISION (raw-document path): like the Python port, this pass both
//! detects and emits on the RAW `serde_json::Value` document. The rewrite runs
//! BEFORE `parse::load` (the typed parse) — the [`crate::prepare`] entry point
//! keeps the rewritten raw document in hand so every record consumer
//! ([`pushdown_record`], [`pushdown_provider_gates`]) reads the raw side; the
//! typed pipeline downstream only ever sees the generated *constructs* (index
//! set, producer equation, member variables), which the parser preserves.
//! `serde_json`'s `preserve_order` feature is on, so object iteration order
//! matches the Python `dict` order and detection is deterministic across the
//! bindings.
//!
//! Output parity with Julia/Python is pinned by the shared conformance corpus
//! (`tests/conformance/pushdown/`): for each committed input the rewritten
//! document must deep-equal the Julia-emitted golden as parsed JSON (numbers
//! by value, key order free — see the corpus README).
use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use serde_json::{Map, Value, json};
use crate::faq::{ReduceKind, Semiring};
/// A malformed gate/record encountered while deriving provider gates (mirrors
/// the Julia `RefreshError` sites in pushdown_rewrite.jl and the Python
/// `PushdownRewriteError`).
#[derive(Debug, Clone)]
pub struct PushdownRewriteError(pub String);
impl fmt::Display for PushdownRewriteError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "PushdownRewriteError: {}", self.0)
}
}
impl std::error::Error for PushdownRewriteError {}
// --------------------------------------------------------------------------- //
// Record / model-selection helpers
// --------------------------------------------------------------------------- //
/// The rewrite's provenance record `metadata.x_esd.pushdown` (written by
/// [`desugar_pushdown`]), or `None` when `doc` carries none. This is the
/// record the engine reads BACK to derive provider gates.
pub fn pushdown_record(doc: &Value) -> Option<&Map<String, Value>> {
doc.get("metadata")?
.get("x_esd")?
.get("pushdown")?
.as_object()
}
fn pd_model_name(doc: &Value, model_name: Option<&str>) -> Option<String> {
if let Some(m) = model_name {
return Some(m.to_string());
}
let models = doc.get("models")?.as_object()?;
if models.len() == 1 {
models.keys().next().cloned()
} else {
None
}
}
// --------------------------------------------------------------------------- //
// Raw-AST leaf helpers (the Julia typed-IR helpers, on raw JSON)
// --------------------------------------------------------------------------- //
fn pd_varname(e: &Value) -> Option<&str> {
e.as_str()
}
fn op_of(e: &Value) -> Option<&str> {
e.get("op")?.as_str()
}
/// `index(F, sym)` with EXACTLY one index → `(F, sym)`; else `None`.
fn pd_index_split(e: &Value) -> Option<(&str, &str)> {
if op_of(e)? != "index" {
return None;
}
let a = e.get("args")?.as_array()?;
if a.len() != 2 {
return None;
}
Some((pd_varname(&a[0])?, pd_varname(&a[1])?))
}
/// `index(F, sym…)` with ≥1 index → `(F, [syms…])`; else `None`.
fn pd_index_syms(e: &Value) -> Option<(&str, Vec<&str>)> {
if op_of(e)? != "index" {
return None;
}
let a = e.get("args")?.as_array()?;
if a.len() < 2 {
return None;
}
let f = pd_varname(&a[0])?;
let mut syms = Vec::with_capacity(a.len() - 1);
for x in &a[1..] {
syms.push(pd_varname(x)?);
}
Some((f, syms))
}
/// Classify an aggregate BODY `A[c, out…] · E[c]` — a two-factor `⊗=·`
/// product of a rank-(1+|out|) array factor `A` subscripted `[c, out…]` and a
/// rank-1 factor `E` subscripted `[c]` — into `(Aname, Ename)`, or `None`
/// when `body` is not that exact shape. PURE STRUCTURAL check on index
/// symbols (the caller applies the semiring guard).
fn pd_matvec_factors(body: &Value, c_sym: &str, out_syms: &[&str]) -> Option<(String, String)> {
if op_of(body)? != "*" {
return None;
}
let args = body.get("args")?.as_array()?;
if args.len() != 2 || out_syms.is_empty() {
return None;
}
let mut a_syms: Vec<&str> = vec![c_sym];
a_syms.extend_from_slice(out_syms);
let e_syms: Vec<&str> = vec![c_sym];
let mut aname: Option<&str> = None;
let mut ename: Option<&str> = None;
for arg in args {
let (f, syms) = pd_index_syms(arg)?;
if syms == a_syms {
aname = Some(f);
} else if syms == e_syms {
ename = Some(f);
}
}
Some((aname?.to_string(), ename?.to_string()))
}
/// The `reduce` spelling of a registry ⊕ — how the guard's string comparison
/// names each [`ReduceKind`].
fn reduce_spelling(rk: ReduceKind) -> &'static str {
match rk {
ReduceKind::Sum => "+",
ReduceKind::Product => "*",
ReduceKind::Max => "max",
ReduceKind::Min => "min",
ReduceKind::Or => "or",
ReduceKind::And => "and",
}
}
/// (⊕ spelling, 0̄) — mirrors the Julia `_aggregate_oplus_identity` used by
/// the semiring guard; only the `("+", 0.0)` comparison matters to this pass.
/// DERIVED from the closed §5.1 registry ([`Semiring::from_name`] →
/// [`Semiring::oplus`] → [`ReduceKind::identity`]) — the same source
/// `value_invention::vi_oplus` reduces through — so a semiring added to the
/// registry cannot be missed here.
fn semiring_oplus(semiring: &str) -> Option<(&'static str, f64)> {
let oplus = Semiring::from_name(semiring)?.oplus();
Some((reduce_spelling(oplus), oplus.identity()))
}
/// 0̄ for a `reduce` spelling; identities come from the registry
/// ([`ReduceKind::identity`]), the string→⊕ domain stays this pass's own.
///
/// The domain is the schema's CLOSED `reduce` enum, `+ * max min`, and nothing
/// else. `"or"` used to be admitted here as a sixth spelling; the schema has
/// never allowed it, so the arm was unreachable from any valid file, and
/// [`crate::faq::effective_reduce_kind`] now rejects it too. `or` remains
/// a perfectly reachable ⊕ — but only as `bool_and_or`'s, through the
/// `semiring` field, which [`pd_oplus`] resolves via [`semiring_oplus`] before
/// it ever consults `reduce`.
///
/// An unrecognized spelling still returns `None` (the pass declines to rewrite)
/// rather than raising: this is an optimization, and the diagnostic for an
/// out-of-enum file belongs to `aggregate::validate_oplus_spellings`.
fn oplus_identity(reduce: &str) -> Option<f64> {
let rk = match reduce {
"+" => ReduceKind::Sum,
"*" => ReduceKind::Product,
"max" => ReduceKind::Max,
"min" => ReduceKind::Min,
_ => return None,
};
Some(rk.identity())
}
fn pd_oplus(agg: &Value) -> Option<(String, f64)> {
if let Some(semiring) = agg.get("semiring") {
let s = value_to_display_string(semiring);
return semiring_oplus(&s).map(|(o, i)| (o.to_string(), i));
}
let r = match agg.get("reduce") {
None | Some(Value::Null) => "+".to_string(),
Some(v) => value_to_display_string(v),
};
let id = oplus_identity(&r)?;
Some((r, id))
}
/// The Python `str(v)` coercion the reference applies to record fields.
fn value_to_display_string(v: &Value) -> String {
match v {
Value::String(s) => s.clone(),
other => other.to_string(),
}
}
fn is_faq_op(op: Option<&str>) -> bool {
op == Some("faq")
}
fn pd_flip(op: &str) -> &'static str {
match op {
"<" => ">",
"<=" => ">=",
">" => "<",
_ => "<=",
}
}
/// Condition of the first `ifelse(cond, then, else)` in a raw subtree.
/// Object-value iteration is insertion-ordered (`preserve_order`), matching
/// the Python `dict.values()` walk exactly.
///
/// Hand-rolled rather than `crate::json_visit`: a first-match search returning
/// a borrowed reference with an early return, which the shared visitors don't
/// provide.
fn pd_find_ifelse_cond(e: &Value) -> Option<&Value> {
match e {
Value::Object(m) => {
if m.get("op").and_then(Value::as_str) == Some("ifelse")
&& let Some(a) = m.get("args").and_then(Value::as_array)
&& a.len() == 3
{
return Some(&a[0]);
}
for v in m.values() {
if let Some(r) = pd_find_ifelse_cond(v) {
return Some(r);
}
}
None
}
Value::Array(xs) => {
for x in xs {
if let Some(r) = pd_find_ifelse_cond(x) {
return Some(r);
}
}
None
}
_ => None,
}
}
/// The §5.5.6 overlap-gate envelopes of a containment predicate. `tgt_env` is
/// always the four cell bounds; `src_env` is two point coordinates or four
/// record-envelope bounds, per the shape matched.
struct Containment {
src_env: Vec<String>,
tgt_env: [String; 4],
}
/// Parse a containment predicate — an `and`/`*` of comparisons, each between a
/// factor subscripted by `c_sym` and one subscripted by `r_sym` — into the
/// §5.5.6 overlap-gate envelopes, `src_env` the `r_sym` side. TWO shapes are
/// read, told apart by how many DISTINCT `r_sym`-side factors appear and how
/// many `c_sym`-side bounds each carries:
///
/// * POINT-IN-RECT — 2 factors, each with a min AND a max bound:
/// `src_env = [Px, Py]`, `tgt_env = [xmin, ymin, xmax, ymax]`;
/// * ENVELOPE-OVERLAP — 4 factors, each with EXACTLY ONE bound, i.e. the AABB
/// test `cxmin ≤ rxmax ∧ rxmin ≤ cxmax ∧ cymin ≤ rymax ∧ rymin ≤ cymax`:
/// `src_env = [rxmin, rymin, rxmax, rymax]`, `tgt_env = [cxmin, cymin, cxmax,
/// cymax]`.
///
/// A bound's KIND comes from the ORIENTATION of its comparison, so the
/// comparisons may be authored in any order and either direction.
///
/// Which factors share an axis is decided by appearance order, and that choice
/// is FREE: the envelope predicate is a perfect matching between the four cell
/// and four record factors, and nothing in it says which two comparisons are
/// the x pair — but §5.5.6's broad phase is the conjunction of the same four
/// inequalities under ANY pairing that puts one lower and one upper bound in
/// each axis, because each emitted inequality pairs an envelope entry with the
/// partner matched here. The axis labels are a relabelling the AABB test is
/// invariant under. Appearance order is used because it is deterministic.
///
/// `None` on any other shape.
fn pd_parse_containment(pred: &Value, c_sym: &str, r_sym: &str) -> Option<Containment> {
if !pred.is_object() {
return None;
}
let single = [pred.clone()];
let comps: &[Value] = if matches!(op_of(pred), Some("and") | Some("*")) {
pred.get("args")?.as_array()?.as_slice()
} else {
&single
};
// record-side factor -> {"min": cell bound, "max": cell bound}, in
// first-seen order.
let mut bounds: HashMap<String, HashMap<&'static str, String>> = HashMap::new();
let mut rec_order: Vec<String> = Vec::new();
for cmp in comps {
let opn0 = op_of(cmp)?;
if !matches!(opn0, "<" | "<=" | ">" | ">=") {
return None;
}
let args = cmp.get("args")?.as_array()?;
if args.len() != 2 {
return None;
}
let (f1, sym1) = pd_index_split(&args[0])?;
let (f2, sym2) = pd_index_split(&args[1])?;
let (fc, fp, cell_on_left) = if sym1 == c_sym && sym2 == r_sym {
(f1, f2, true)
} else if sym1 == r_sym && sym2 == c_sym {
(f2, f1, false)
} else {
return None;
};
let opn = if cell_on_left { opn0 } else { pd_flip(opn0) };
let kind = if matches!(opn, "<" | "<=") {
"min"
} else {
"max"
};
let entry = bounds.entry(fp.to_string()).or_insert_with(|| {
rec_order.push(fp.to_string());
HashMap::new()
});
if entry.insert(kind, fc.to_string()).is_some() {
return None; // one bound of each kind per factor
}
}
if rec_order.len() == 2 && rec_order.iter().all(|p| bounds[p.as_str()].len() == 2) {
let (px, py) = (rec_order[0].clone(), rec_order[1].clone());
return Some(Containment {
tgt_env: [
bounds[&px]["min"].clone(),
bounds[&py]["min"].clone(),
bounds[&px]["max"].clone(),
bounds[&py]["max"].clone(),
],
src_env: vec![px, py],
});
}
if rec_order.len() == 4 && rec_order.iter().all(|p| bounds[p.as_str()].len() == 1) {
// A record factor carrying a LOWER cell bound (`cmin ≤ r`) is that
// axis's record MAXIMUM, and vice versa — the AABB test compares each
// side's min against the other side's max.
let his: Vec<String> = rec_order
.iter()
.filter(|p| bounds[p.as_str()].contains_key("min"))
.cloned()
.collect();
let los: Vec<String> = rec_order
.iter()
.filter(|p| bounds[p.as_str()].contains_key("max"))
.cloned()
.collect();
if his.len() != 2 || los.len() != 2 {
return None;
}
return Some(Containment {
tgt_env: [
bounds[his[0].as_str()]["min"].clone(),
bounds[his[1].as_str()]["min"].clone(),
bounds[los[0].as_str()]["max"].clone(),
bounds[los[1].as_str()]["max"].clone(),
],
src_env: vec![
los[0].clone(),
los[1].clone(),
his[0].clone(),
his[1].clone(),
],
});
}
None
}
fn ranges_of(agg: &Value) -> Option<&Map<String, Value>> {
agg.get("ranges")?.as_object()
}
fn range_from(v: Option<&Value>) -> Option<&str> {
v?.get("from")?.as_str()
}
/// A matched binning aggregate — a `+`-semiring reduction over TWO 1-D index
/// sets whose body carries a containment predicate between CELL-indexed and
/// RECORD-indexed factors (either shape [`pd_parse_containment`] reads). BOTH
/// orientations are recognised (CONFORMANCE_SPEC.md §5.5.7):
///
/// ```text
/// FORWARD E[c] = Σ_r [contains(cell_c, rec_r)] · … (the cell axis is output)
/// MIRROR P[r] = Σ_c [contains(cell_c, rec_r)] · … (the record axis is output)
/// ```
struct Binning {
/// The loop symbol carrying the CELL side (the four rect bounds).
c_sym: String,
/// The loop symbol carrying the RECORD side (the point coordinates, or the
/// record envelope's four bounds).
r_sym: String,
/// The index set `c_sym` ranges over.
c_set: String,
/// The index set `r_sym` ranges over.
r_set: String,
/// `true` when the aggregate's own output axis is the CELL one (FORWARD).
out_is_cell: bool,
src_env: Vec<String>,
tgt_env: [String; 4],
}
/// The RHS of the equation whose LHS is the bare variable `name` — the esm
/// 1.0.0 home of the 0.x `variables[name].expression` (esm-spec §6.3.1). An
/// unknown defined by a bare-variable LHS IS the observed this pass matches on,
/// so "has a definition here" and "is observed" are the same question.
fn pd_def<'a>(model: &'a Value, name: &str) -> Option<&'a Value> {
model
.get("equations")?
.as_array()?
.iter()
.find(|eq| eq.get("lhs").and_then(Value::as_str) == Some(name))?
.get("rhs")
}
/// Mutable counterpart of [`pd_def`].
fn pd_def_mut<'a>(model: &'a mut Value, name: &str) -> Option<&'a mut Value> {
model
.get_mut("equations")?
.as_array_mut()?
.iter_mut()
.find(|eq| eq.get("lhs").and_then(Value::as_str) == Some(name))?
.get_mut("rhs")
}
/// Give `name` a defining equation with RHS `rhs`, replacing any existing one.
fn pd_set_def(model: &mut Value, name: &str, rhs: Value) {
if let Some(existing) = pd_def_mut(model, name) {
*existing = rhs;
return;
}
let eqs = model
.as_object_mut()
.expect("model is an object")
.entry("equations")
.or_insert_with(|| Value::Array(Vec::new()));
if !eqs.is_array() {
*eqs = Value::Array(Vec::new());
}
if let Some(list) = eqs.as_array_mut() {
list.push(json!({"lhs": name, "rhs": rhs}));
}
}
/// Is `v` a declared UNKNOWN (esm-spec §6.3)?
fn pd_is_unknown(v: &Value) -> bool {
v.get("type").and_then(Value::as_str) == Some("unknown")
}
/// Is `ev` a binning aggregate whose OUTPUT axis is `out_set`? Returns the
/// binding, or `None`.
///
/// The gate is IDENTICAL either way — the enumeration driver binds its two
/// symbols from the join clause's declared envelopes and knows nothing about
/// cells vs records, and the aggregate's own `output_idx` decides the result's
/// orientation. So the guards here are on the aggregate's SHAPE, not on which
/// axis is which: `out_set` is the index set the observed is shaped on, the
/// single other range supplies the opposite side, and the CONTAINMENT PREDICATE
/// itself says which symbol is the cell (it carries the four rect BOUND
/// factors) and which is the record (the two point coordinates).
/// `want_out_is_cell` lets a caller that already knows the orientation say so.
/// A POINT-IN-RECT predicate parses only one way — the rect side cannot pass
/// for the point side — but an ENVELOPE-OVERLAP predicate is SYMMETRIC and
/// parses BOTH, so the predicate alone no longer decides. The forward arm's
/// cell set comes from the mat-vec array's first axis and mirrors are collected
/// only once `c_set`/`r_set` are fixed, so both callers do know. `None` prefers
/// the out-as-cell reading, which is what the point case always gave.
fn pd_detect_binning(
ev: &Value,
agg: &Value,
out_set: &str,
want_out_is_cell: Option<bool>,
) -> Option<Binning> {
if !pd_is_unknown(ev) {
return None;
}
let shape = ev.get("shape")?.as_array()?;
if shape.len() != 1 || shape[0].as_str()? != out_set {
return None;
}
if !is_faq_op(op_of(agg)) {
return None;
}
let (oplus, ident) = pd_oplus(agg)?;
if !(oplus == "+" && ident == 0.0) {
return None; // SEMIRING GUARD
}
let oi = agg.get("output_idx")?.as_array()?;
if oi.len() != 1 {
return None;
}
let out_sym = oi[0].as_str()?;
let ranges = ranges_of(agg)?;
if ranges.len() != 2 || range_from(ranges.get(out_sym)) != Some(out_set) {
return None;
}
let in_sym = ranges.keys().find(|k| k.as_str() != out_sym)?.clone();
let in_set = range_from(ranges.get(&in_sym))?.to_string();
let body = agg.get("expr")?;
if !body.is_object() {
return None;
}
let pred = pd_find_ifelse_cond(body)?;
if want_out_is_cell != Some(false)
&& let Some(env) = pd_parse_containment(pred, out_sym, &in_sym)
{
return Some(Binning {
c_sym: out_sym.to_string(),
r_sym: in_sym,
c_set: out_set.to_string(),
r_set: in_set,
out_is_cell: true,
src_env: env.src_env,
tgt_env: env.tgt_env,
});
}
if want_out_is_cell == Some(true) {
return None;
}
let env = pd_parse_containment(pred, &in_sym, out_sym)?;
Some(Binning {
c_sym: in_sym,
r_sym: out_sym.to_string(),
c_set: in_set,
r_set: out_set.to_string(),
out_is_cell: false,
src_env: env.src_env,
tgt_env: env.tgt_env,
})
}
// --------------------------------------------------------------------------- //
// Detection-time template-reference expansion (esm-spec §9.6.4 rule 2).
//
// Under Option B (§9.6.4) `load` PRESERVES `apply_expression_template`
// references: they ride to the build boundary, where they are expanded ONCE with
// site recording. `prepare` therefore hands
// `desugar_pushdown` a document whose binning body may be a surviving reference
// rather than the containment `ifelse` the recogniser looks for.
//
// §9.6.4 rule 4 ("patterns do not see through surviving references") governs the
// §9.6.3 REWRITE-RULE ENGINE. This desugar is a different consumer and rule 2
// governs it: a reference DENOTES its expansion. So whether the pushdown fires
// MUST NOT depend on whether the author factored the body through a template —
// detection runs on the EXPANDED view.
//
// EMISSION does not: `pd_apply` edits the call site's `bindings` (and the
// aggregate's own `ranges` / `args` / `shape` / `join`), never the shared
// template body, so the body stays shared and singly-lowered and Option B
// survives the rewrite. `pd_assert_rects_rebound` is the post-condition.
// --------------------------------------------------------------------------- //
/// The op name of a surviving expression-template reference.
const APPLY_OP: &str = "apply_expression_template";
/// The component template registry of `model`, or `None`.
///
/// Only the component-level `expression_templates` block is consulted, which is
/// what the Julia reference reads (`coerce_esm_file` fills `component_templates`
/// from exactly these blocks) — a top-level authored registry is a DECLARATION
/// that load materialises into the components, so on the `prepare` input form
/// the per-component block is the registry.
fn pd_templates(model: &Value) -> Option<&Map<String, Value>> {
model
.get("expression_templates")
.and_then(Value::as_object)
.filter(|m| !m.is_empty())
}
/// Does `node` carry a surviving `apply_expression_template` reference?
/// Descends every object value, `bindings` included.
///
/// Hand-rolled rather than `crate::json_visit`: a short-circuiting existence
/// test, and the read-only shared visitor cannot stop early.
fn pd_has_apply(node: &Value) -> bool {
match node {
Value::Object(m) => {
m.get("op").and_then(Value::as_str) == Some(APPLY_OP) || m.values().any(pd_has_apply)
}
Value::Array(xs) => xs.iter().any(pd_has_apply),
_ => false,
}
}
/// The `name` of the first surviving reference in `node` (pre-order), for the
/// residual diagnostic; `None` when it carries none.
///
/// Hand-rolled rather than `crate::json_visit`: short-circuits at the first
/// reference node and does not descend into it.
fn pd_first_apply_name(node: &Value) -> Option<String> {
match node {
Value::Object(m) => {
if m.get("op").and_then(Value::as_str) == Some(APPLY_OP) {
return m.get("name").and_then(Value::as_str).map(str::to_string);
}
m.values().find_map(pd_first_apply_name)
}
Value::Array(xs) => xs.iter().find_map(pd_first_apply_name),
_ => None,
}
}
/// `Expand(node)` against `templates` — DETECTION ONLY; nothing of the result is
/// emitted. Returns `None` when there is nothing to expand, and `None` when
/// expansion FAILS: the pass's contract is to leave a document it cannot
/// recognise alone, and an unexpandable reference is then reported by
/// [`pd_binning_refusal`] if the variable is join-shaped.
fn pd_expand_for_detection(node: &Value, templates: Option<&Map<String, Value>>) -> Option<Value> {
let templates = templates?;
if !pd_has_apply(node) {
return None;
}
crate::lower_expression_templates::expand_against_registry(node, templates, "pushdown_rewrite")
.ok()
}
/// The definitions ([`pd_def`]) that carried a surviving
/// `apply_expression_template` reference, EXPANDED — the `Expand(tree)` view the
/// pattern matcher must see (§9.6.4 rule 2). From esm 1.0.0 an observed
/// unknown's body is its defining EQUATION's right-hand side, so this — not the
/// variable table — is what the detector matches against.
///
/// Only the expanded definitions are returned, as OVERRIDES for [`pd_def_view`]
/// to consult: a template-free document builds an empty map, allocates no
/// `Value`, and takes the byte-identical pre-existing path.
///
/// DETECTION ONLY. The emission side reads [`pd_def`] / [`pd_def_mut`] so it
/// edits the AUTHORED body (and, for a template-factored one, the call site's
/// `bindings`) rather than a detached expansion.
fn pd_detection_defs(model: &Value) -> Map<String, Value> {
let mut out = Map::new();
let templates = pd_templates(model);
if templates.is_none() {
return out;
}
let Some(eqs) = model.get("equations").and_then(Value::as_array) else {
return out;
};
for eq in eqs {
let Some(lhs) = eq.get("lhs").and_then(Value::as_str) else {
continue;
};
let Some(rhs) = eq.get("rhs") else {
continue;
};
// FIRST definition wins, matching `pd_def`.
if out.contains_key(lhs) {
continue;
}
if let Some(expanded) = pd_expand_for_detection(rhs, templates) {
out.insert(lhs.to_string(), expanded);
}
}
out
}
/// [`pd_def`] through the detection view: the EXPANDED body when
/// [`pd_detection_defs`] produced one for `name`, else the authored body.
fn pd_def_view<'a>(
model: &'a Value,
defs: &'a Map<String, Value>,
name: &str,
) -> Option<&'a Value> {
defs.get(name).or_else(|| pd_def(model, name))
}
// --------------------------------------------------------------------------- //
// Residual diagnostics.
//
// A pattern recogniser that declines SILENTLY is indistinguishable from one that
// fired — until, hours later, an ungated provider fetch runs the machine out of
// memory. These keep the two cases apart:
//
// NOT A JOIN — a `+`-aggregate with no containment predicate is a
// legitimately dense factor. Nothing to gate, no
// diagnostic.
// A JOIN I CANNOT READ — the aggregate bins records into cells of the SAME set
// that indexes a provider-backed rank-2 array it feeds,
// but the containment could not be recovered. Reported.
//
// WARNING, not error: the pass's contract (CONFORMANCE_SPEC §5.5.7) is that an
// unrecognised document comes back unchanged, and the residue is a PERFORMANCE
// defect — the numbers stay right, the fetch gets big. The one hard error in
// this pass is `pd_assert_rects_rebound`, where the rewrite HAS fired and a rect
// factor could not be re-pointed: wrong numbers, not slow ones.
// --------------------------------------------------------------------------- //
/// The fixed, cross-binding `consequence` string of a residual diagnostic.
pub const PD_UNGATED_CONSEQUENCE: &str = "the provider-backed array is fetched WHOLESALE — no derived support set \
is produced and no gate is emitted";
/// Why [`pd_detect_binning`] refused `ev`, for a caller that has ALREADY
/// established `ev` sits in the join position. `agg` is `ev`'s defining
/// equation RHS from the detection view, exactly as [`pd_detect_binning`]
/// received it.
///
/// `None` ⇒ `ev` is simply not join-shaped (no diagnostic warranted). Otherwise
/// `(reason, template)`: `("surviving_template_reference", Some(name))` when the
/// body carries a reference that could not be expanded for matching,
/// `("predicate_unparsed", None)` when a containment `ifelse` was found but did
/// not read as a rectangle containment in either orientation.
fn pd_binning_refusal(
ev: &Value,
agg: &Value,
out_set: &str,
) -> Option<(&'static str, Option<String>)> {
let shape = ev.get("shape")?.as_array()?;
if shape.len() != 1 || shape[0].as_str() != Some(out_set) {
return None;
}
if !is_faq_op(op_of(agg)) {
return None;
}
let (oplus, ident) = pd_oplus(agg)?;
if !(oplus == "+" && ident == 0.0) {
return None;
}
let oi = agg.get("output_idx")?.as_array()?;
if oi.len() != 1 {
return None;
}
let out_sym = oi[0].as_str()?;
let ranges = ranges_of(agg)?;
if ranges.len() != 2 || range_from(ranges.get(out_sym)) != Some(out_set) {
return None;
}
let in_sym = ranges.keys().find(|k| k.as_str() != out_sym)?;
range_from(ranges.get(in_sym))?;
let body = agg.get("expr")?;
if !body.is_object() {
return None;
}
if pd_find_ifelse_cond(body).is_none() {
// No predicate at all ⇒ genuinely dense, unless a surviving reference
// is hiding one.
let tname = pd_first_apply_name(body)?;
return Some(("surviving_template_reference", Some(tname)));
}
Some(("predicate_unparsed", None))
}
/// The human-readable rendering of one diagnostic record: what was recognised,
/// what could not be read, and what it costs.
pub fn pd_diagnostic_message(d: &Value) -> String {
let g = |k: &str| d.get(k).and_then(Value::as_str).unwrap_or("");
let why = if g("reason") == "surviving_template_reference" {
let tpl = d.get("template").and_then(Value::as_str);
format!(
"its body carries a surviving `apply_expression_template` reference{} \
that could not be expanded for matching",
match tpl {
Some(t) => format!(" to '{t}'"),
None => String::new(),
}
)
} else {
"its containment predicate did not read as a point-in-rectangle containment \
(four cell-indexed rect bounds against two record-indexed point coordinates) nor as an \
envelope-overlap one (four bounds on each side)"
.to_string()
};
format!(
"projection-pushdown desugar: '{}' is join-shaped — it bins records into the \
cells of index set '{}' and feeds the provider-backed array '{}' through '{}' — but \
{}, so the rewrite does NOT fire for it and {}. Bind the containment's factors \
through the template's params, or write the predicate longhand.",
g("variable"),
g("index_set"),
g("array"),
g("consumer"),
why,
PD_UNGATED_CONSEQUENCE
)
}
/// The MIRRORED-orientation binning aggregates of a model: per-RECORD observeds
/// `P[r] = Σ_{c∈C} [contains(cell_c, pt_r)] · …` over the plan's cell set
/// `c_set` and record set `r_set`. Returned as `(name, src_env, tgt_env)`
/// triples, SORTED by name so the emitted document is identical across
/// bindings and hash seeds.
///
/// A mirror needs NOTHING but the gate (CONFORMANCE_SPEC.md §5.5.7, "MIRRORED
/// arm"). Its cell axis stays the FULL `c_set`, so its envelope factors are the
/// document's own const-array rects, unrewritten.
fn pd_mirror_specs(
model: &Value,
defs: &Map<String, Value>,
c_set: &str,
r_set: &str,
forward_names: &[String],
) -> Vec<(String, Vec<String>, [String; 4])> {
let mut out: Vec<(String, Vec<String>, [String; 4])> = Vec::new();
let Some(variables) = model.get("variables").and_then(Value::as_object) else {
return out;
};
for (name, v) in variables {
if forward_names.iter().any(|f| f == name) {
continue;
}
let Some(agg) = pd_def_view(model, defs, name) else {
continue;
};
let Some(bind) = pd_detect_binning(v, agg, r_set, Some(false)) else {
continue;
};
if bind.out_is_cell || bind.c_set != c_set || bind.r_set != r_set {
continue;
}
// Never stack a second gate on an aggregate that already declares a join.
if agg.get("join").is_some() {
continue;
}
out.push((name.clone(), bind.src_env, bind.tgt_env));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
out
}
/// One matched FORWARD binning observed (`E[c] = Σ_r …`) in the plan.
///
/// The envelopes are the ones [`pd_parse_containment`] read out of THIS
/// aggregate's own containment predicate — the gate emitted onto the rewritten
/// `E` is derived, not authored.
struct ForwardBinSpec {
/// The binning observed's name (`E`).
ename: String,
/// Its binding's cell symbol.
c_sym: String,
/// The gate's source envelope factor names.
src_env: Vec<String>,
/// The gate's target envelope factor names.
tgt_env: [String; 4],
}
/// The detected pushdown plan (the Julia/Python plan dict, typed).
struct Plan {
c_set: String,
rcv_set: String,
r_set: String,
/// `(conc name, contracted symbol)` per matched reduction.
conc_specs: Vec<(String, String)>,
/// Provider-backed array factors to gate, SORTED (deterministic
/// `applies_to` — the one collection-order-dependent list that leaks into
/// the emitted document; mirrors the Julia `sort!(A_names)`).
a_names: Vec<String>,
/// The matched FORWARD binning observeds.
e_specs: Vec<ForwardBinSpec>,
/// `(P name, gate src_env, gate tgt_env)` per MIRRORED binning observed,
/// sorted by name. These receive ONLY the gate (§5.5.7 "MIRRORED arm").
mirror_specs: Vec<(String, Vec<String>, [String; 4])>,
src_env: Vec<String>,
tgt_env: [String; 4],
/// EVERY array any binning body reads on the cell axis, name => declared
/// shape. The envelope factors of `tgt_env` are a SUBSET: a binning body is
/// free to read the cell's geometry, its area, its ring stack, and all of
/// them ride the same re-pointing. See [`pd_cell_factors`].
cell_factors: BTreeMap<String, Vec<String>>,
/// The cell-axis reads that CANNOT be re-pointed, name => subscript sketch.
cell_bad: BTreeMap<String, String>,
rep_ename: String,
rep_csym: String,
rep_rsym: String,
}
/// One consumer whose reduction matched the pushdown pattern (see
/// [`pd_match_conc`]), together with the FORWARD binning observed it reads.
struct ConcMatch<'a> {
/// The reduction's contracted symbol.
s_sym: &'a String,
/// The cell index set (the mat-vec array's first axis).
c_set: &'a str,
/// The record index set the reduction is shaped on.
r_set: &'a str,
/// The provider-backed rank-2 `[c_set, r_set]` array factor (`A`).
aname: String,
/// The FORWARD binning observed (`E`).
ename: String,
/// `E`'s aggregate body on the detection view.
eagg: &'a Value,
/// `E`'s binning binding.
bind: Binning,
}
/// Match ONE declared unknown against the pushdown consumer pattern: an
/// additive-`(+, 0)` mat-vec reduction over a declared rank-2 parameter and a
/// FORWARD binning observed. Returns `None` when any guard refuses; a binning
/// factor that is binning-shaped but unreadable additionally records a
/// `pushdown_join_unrecognised` diagnostic into `diags`.
fn pd_match_conc<'a>(
model: &'a Value,
defs: &'a Map<String, Value>,
variables: &'a Map<String, Value>,
cname: &str,
cv: &Value,
diags: &mut Vec<Value>,
) -> Option<ConcMatch<'a>> {
if !pd_is_unknown(cv) {
return None;
}
// An OBSERVED unknown is one with a bare-variable-LHS defining
// equation (esm-spec §6.3.1); one without is a state or algebraic
// unknown and matches nothing here.
let agg = pd_def_view(model, defs, cname)?;
if !is_faq_op(op_of(agg)) {
return None;
}
let (oplus, ident) = pd_oplus(agg)?;
if !(oplus == "+" && ident == 0.0) {
return None; // SEMIRING GUARD
}
let oi = agg.get("output_idx").and_then(Value::as_array)?;
if oi.len() != 1 {
return None;
}
let rcv_sym = oi[0].as_str()?;
let ranges = ranges_of(agg)?;
if ranges.len() != 2 || !ranges.contains_key(rcv_sym) {
return None;
}
let s_sym = ranges.keys().find(|k| k.as_str() != rcv_sym)?;
let c_set = range_from(ranges.get(s_sym))?;
let r_set = range_from(ranges.get(rcv_sym))?;
let body = agg.get("expr")?;
let (aname, ename) = pd_matvec_factors(body, s_sym, &[rcv_sym])?;
// A must be a declared rank-2 parameter [c_set, r_set].
let av = variables.get(&aname)?;
let a_ok = av.get("type").and_then(Value::as_str) == Some("parameter")
&& av
.get("shape")
.and_then(Value::as_array)
.map(|s| s.len() == 2 && s[0].as_str() == Some(c_set) && s[1].as_str() == Some(r_set))
.unwrap_or(false);
if !a_ok {
return None;
}
let ev = variables.get(&ename)?;
let eagg = pd_def_view(model, defs, &ename)?;
let Some(bind) = pd_detect_binning(ev, eagg, c_set, None) else {
// `ev` is the rank-1 factor of a `+`-mat-vec against a
// provider-backed `[c_set, r_set]` array: the join position. If it
// is ALSO binning-shaped but unreadable, say so — silence here is
// the ungated whole-array fetch that surfaces hours later.
if let Some((reason, template)) = pd_binning_refusal(ev, eagg, c_set) {
diags.push(json!({
"code": "pushdown_join_unrecognised",
"variable": ename,
"consumer": cname,
"array": aname,
"index_set": c_set,
"reason": reason,
"template": match template {
Some(t) => Value::String(t),
None => Value::Null,
},
"consequence": PD_UNGATED_CONSEQUENCE,
}));
}
return None;
};
if !bind.out_is_cell {
return None; // FORWARD arm only
}
Some(ConcMatch {
s_sym,
c_set,
r_set,
aname,
ename,
eagg,
bind,
})
}
/// Deterministic, deduplicated diagnostic order: `variables` is a
/// key-ordered map but the same E can be reached from several `conc`
/// consumers.
fn pd_order_diags(diags: &mut Vec<Value>) {
diags.sort_by(|a, b| {
let k = |v: &Value| {
(
v["variable"].as_str().unwrap_or("").to_string(),
v["consumer"].as_str().unwrap_or("").to_string(),
v["array"].as_str().unwrap_or("").to_string(),
)
};
k(a).cmp(&k(b))
});
diags.dedup_by(|a, b| {
a["variable"] == b["variable"] && a["consumer"] == b["consumer"] && a["array"] == b["array"]
});
}
/// Detect the pushdown pattern across a model's observeds.
///
/// `defs` is the DETECTION view ([`pd_detection_defs`]): the observed
/// definitions with surviving template references expanded, so a binning body
/// factored through a template matches exactly as its expansion would. `model`
/// supplies the declarations beside them — shapes and types, which no expansion
/// touches.
///
/// Returns `(plan, diagnostics)` — `plan` `None` when nothing matches / the
/// semiring guard fails, `diagnostics` the residual "a join I could not read"
/// records (see [`pd_binning_refusal`]).
fn pd_detect(model: &Value, defs: &Map<String, Value>) -> (Option<Plan>, Vec<Value>) {
let Some(variables) = model.get("variables").and_then(Value::as_object) else {
return (None, Vec::new());
};
let mut diags: Vec<Value> = Vec::new();
let mut conc_specs: Vec<(String, String)> = Vec::new();
let mut a_names: Vec<String> = Vec::new();
let mut e_specs: Vec<ForwardBinSpec> = Vec::new();
let mut cell_factors: BTreeMap<String, Vec<String>> = BTreeMap::new();
let mut cell_bad: BTreeMap<String, String> = BTreeMap::new();
let mut plan: Option<Plan> = None;
for (cname, cv) in variables {
let Some(m) = pd_match_conc(model, defs, variables, cname, cv, &mut diags) else {
continue;
};
match &plan {
None => {
plan = Some(Plan {
c_set: m.c_set.to_string(),
rcv_set: m.r_set.to_string(),
r_set: m.bind.r_set.clone(),
conc_specs: Vec::new(),
a_names: Vec::new(),
e_specs: Vec::new(),
mirror_specs: Vec::new(),
src_env: m.bind.src_env.clone(),
tgt_env: m.bind.tgt_env.clone(),
cell_factors: BTreeMap::new(),
cell_bad: BTreeMap::new(),
rep_ename: m.ename.clone(),
rep_csym: m.bind.c_sym.clone(),
rep_rsym: m.bind.r_sym.clone(),
});
}
Some(p) if !(m.c_set == p.c_set && m.r_set == p.rcv_set) => {
continue; // narrow: one cell set
}
Some(_) => {}
}
conc_specs.push((cname.clone(), m.s_sym.clone()));
if !a_names.contains(&m.aname) {
a_names.push(m.aname);
}
if !e_specs.iter().any(|s| s.ename == m.ename) {
// Collected on the DETECTION view, like every other pattern read: a
// body factored through a template must yield the same gather set
// as its longhand twin (esm-spec §9.6.4 rule 2). Emission still
// edits the authored body / call-site bindings, and
// `pd_assert_rects_rebound` proves the substitution landed.
pd_cell_factors(
m.eagg,
&m.bind.c_sym,
variables,
m.c_set,
&mut cell_factors,
&mut cell_bad,
);
e_specs.push(ForwardBinSpec {
ename: m.ename,
c_sym: m.bind.c_sym,
src_env: m.bind.src_env,
tgt_env: m.bind.tgt_env,
});
}
}
pd_order_diags(&mut diags);
let Some(mut plan) = plan else {
return (None, diags);
};
if conc_specs.is_empty() {
return (None, diags);
}
// Deterministic plan order (mirrors the Julia `sort!(A_names)`).
a_names.sort();
// MIRRORED-orientation binning aggregates (`P[r] = Σ_c […]`) over the SAME
// cell/record sets. They are collected only once the forward pattern has
// fixed `c_set`/`r_set`: the mirror is a RIDER on the rewrite, never its
// trigger, so a document holding only mirrored binning aggregates is not
// rewritten at all (§5.5.7).
let forward_names: Vec<String> = e_specs.iter().map(|s| s.ename.clone()).collect();
plan.mirror_specs = pd_mirror_specs(model, defs, &plan.c_set, &plan.r_set, &forward_names);
plan.conc_specs = conc_specs;
plan.a_names = a_names;
plan.e_specs = e_specs;
plan.cell_factors = cell_factors;
plan.cell_bad = cell_bad;
(Some(plan), diags)
}
// --------------------------------------------------------------------------- //
// Emission
// --------------------------------------------------------------------------- //
/// Walk a binning body and record into `out` EVERY array it reads at a position
/// on the cell axis — not only the envelope factors of the containment
/// predicate — mapping each to its declared shape. Ungatherable reads land in
/// `bad`.
///
/// The rewrite re-points the aggregate's reduction range onto the compact
/// derived support set, so from that moment `c_sym` counts support positions,
/// not grid cells. Any array still indexed by it that was NOT re-pointed then
/// reads full-grid values at support positions: WRONG NUMBERS, no diagnostic.
/// `out` is exactly the set that must be gathered.
///
/// Membership is decided by the DECLARATION, not by the subscript: an array is
/// on the cell axis iff its declared `shape[0]` is the cell index set `c_set`.
/// That is what keeps a flat-offset gather into a DIFFERENT axis — the
/// `index(Temperature, layer * N_SRC + c)` spelling a layered met read wants,
/// whose base is declared over the full `[all_cells]` axis — out of the map: it
/// is not on the cell axis, it stays full-grid, and it is still correct after
/// the rewrite because nothing about it moved.
///
/// Three subscript shapes on a cell-axis array, and they are not alike:
///
/// * `index(F, c)` — the whole trailing slice at cell `c`, a rank-3 ring stack
/// read as a polygon operand. Gatherable.
/// * `index(F, c, v, d)` — a fully-subscripted scalar read. Gatherable, and by
/// the SAME gather: the generated array keeps `F`'s rank, so both spellings
/// survive the substitution of the name.
/// * `index(F, <expr in c>)` — arithmetic on the cell position. NOT gatherable:
/// the compact axis is a renumbering, so `c + 1` means nothing in it.
/// Recorded in `bad`.
///
/// A cell-axis array indexed WITHOUT the cell symbol is deliberately left
/// alone: it still reads the full-grid array at a full-grid position, which the
/// rewrite does not disturb.
fn pd_cell_factors(
node: &Value,
c_sym: &str,
variables: &Map<String, Value>,
c_set: &str,
out: &mut BTreeMap<String, Vec<String>>,
bad: &mut BTreeMap<String, String>,
) {
crate::json_visit::visit_values(node, &mut |_path, v| {
let Some(o) = v.as_object() else { return };
if o.get("op").and_then(Value::as_str) == Some("index")
&& let Some(a) = o.get("args").and_then(Value::as_array)
&& a.len() >= 2
&& let Some(f) = a[0].as_str()
&& let Some(shape) = variables
.get(f)
.and_then(|v| v.get("shape"))
.and_then(Value::as_array)
&& shape.first().and_then(Value::as_str) == Some(c_set)
{
if a[1].as_str() == Some(c_sym) {
out.entry(f.to_string()).or_insert_with(|| {
shape
.iter()
.map(|t| t.as_str().unwrap_or_default().to_string())
.collect()
});
} else if pd_mentions_sym(&a[1], c_sym) {
bad.entry(f.to_string())
.or_insert_with(|| pd_subscript_sketch(&a[1]));
}
}
});
}
/// Does `node` reference the loop symbol `sym` anywhere?
///
/// Hand-rolled rather than `crate::json_visit`: descent is key-dependent
/// (`name` entries are labels, not references, and are skipped) and the test
/// short-circuits.
fn pd_mentions_sym(node: &Value, sym: &str) -> bool {
match node {
Value::String(s) => s == sym,
Value::Object(o) => o
.iter()
.any(|(k, v)| k != "name" && pd_mentions_sym(v, sym)),
Value::Array(xs) => xs.iter().any(|x| pd_mentions_sym(x, sym)),
_ => false,
}
}
/// A one-line rendering of a subscript expression, for the refusal message.
fn pd_subscript_sketch(node: &Value) -> String {
match node {
Value::String(s) => s.clone(),
Value::Number(n) => n.to_string(),
Value::Object(o) => {
let op = o.get("op").and_then(Value::as_str).unwrap_or("?");
let empty: Vec<Value> = Vec::new();
let args = o.get("args").and_then(Value::as_array).unwrap_or(&empty);
if op == "index" && !args.is_empty() {
let inner: Vec<String> = args[1..].iter().map(pd_subscript_sketch).collect();
format!("{}[{}]", pd_subscript_sketch(&args[0]), inner.join(", "))
} else {
let inner: Vec<String> = args.iter().map(pd_subscript_sketch).collect();
format!("{}({})", op, inner.join(", "))
}
}
_ => "?".to_string(),
}
}
/// The `(variable declaration, defining aggregate)` for one per-support cell
/// gather `pd_cell__C__F`, RANK-PRESERVING.
///
/// Rank 1 — the envelope factors — emits exactly what it always did:
///
/// ```text
/// pd_cell__C__F[c] = F[member_factor[c]]
/// ```
///
/// Rank k keeps every trailing axis, so a `[cells, vertex, xy]` ring stack comes
/// out as a `[support, vertex, xy]` ring stack and every use of it survives the
/// rename unchanged — the sliced polygon-operand form `index(F, c)` and the
/// fully-subscripted scalar form alike:
///
/// ```text
/// pd_cell__C__F[c, t0, t1] = F[member_factor[c], t0, t1]
/// ```
///
/// This is a map, not a reduction: every range appears in `output_idx`. The
/// trailing loop symbols are named `pd_t0…` rather than reusing the document's
/// own, because the gather is generated in its own scope and a collision with an
/// authored symbol would be a silent capture.
fn pd_gather_defn(
f: &str,
shape: &[String],
setname: &str,
mfactor: &str,
index_set_names: &HashSet<String>,
c_set: &str,
) -> Result<(Value, Value), PushdownRewriteError> {
let mut decl_shape: Vec<Value> = vec![Value::String(setname.to_string())];
let mut output_idx: Vec<Value> = vec![Value::String("c".into())];
let mut ranges = Map::new();
ranges.insert("c".into(), json!({"from": setname}));
let mut subs: Vec<Value> = Vec::new();
for (i, t) in shape.iter().skip(1).enumerate() {
if !index_set_names.contains(t) {
return Err(PushdownRewriteError(format!(
"projection-pushdown desugar: cannot gather '{f}' onto the derived \
support set of '{c_set}'. Its declared shape is {shape:?}, whose \
trailing entry '{t}' is not a named index set, so the generated \
gather has no range to iterate it over. Declare the array's \
trailing axes as index sets, or keep the value off the cell axis."
)));
}
let sym = format!("pd_t{i}");
decl_shape.push(Value::String(t.clone()));
output_idx.push(Value::String(sym.clone()));
ranges.insert(sym.clone(), json!({"from": t}));
subs.push(Value::String(sym));
}
let mut expr_args: Vec<Value> = vec![
Value::String(f.to_string()),
pd_ix(mfactor.to_string(), "c"),
];
expr_args.extend(subs);
Ok((
json!({"type": "unknown", "shape": decl_shape}),
json!({
"op": "faq",
"output_idx": output_idx,
"ranges": Value::Object(ranges),
"args": [f, mfactor],
"expr": {"op": "index", "args": expr_args},
}),
))
}
/// Refuse, loudly, when a binning body reads a cell-axis array at a computed
/// cell position. See [`pd_cell_factors`] for why it cannot be re-pointed.
///
/// A hard error, not a warning, and not a silent decline. The pattern HAS
/// matched: an aggregate reading a cell-axis array at `c + 1` cannot be gated at
/// all without renumbering arithmetic the compact axis does not admit, and
/// emitting the rewrite anyway is the silent-wrong-numbers failure this pass
/// exists to prevent.
fn pd_refuse_ungatherable(
bad: &BTreeMap<String, String>,
c_set: &str,
setname: &str,
) -> Result<(), PushdownRewriteError> {
if bad.is_empty() {
return Ok(());
}
let detail = bad
.iter()
.map(|(f, s)| format!("'{f}' at [{s}]"))
.collect::<Vec<_>>()
.join("; ");
Err(PushdownRewriteError(format!(
"projection-pushdown desugar: the binning aggregate reads a cell-axis array \
at a COMPUTED cell position ({detail}). The rewrite re-points the reduction \
onto the derived support set '{setname}', which renumbers '{c_set}' — \
support position i is grid cell member_factor[i], and no arithmetic on i \
survives that renumbering. A gather can carry `F[c]`, and `F[c, …]`, but \
not `F[f(c)]`. Index the array with the bare cell symbol, or move the value \
off the '{c_set}' axis (declare it over the axis it is really indexed by, \
so it stays full-grid and is left alone)."
)))
}
/// In-place: rewrite every `index(F, …)` whose factor `F` is a key of
/// `rectmap` to `index(rectmap[F], …)` throughout a raw AST subtree.
///
/// This walk descends EVERY object value, `bindings` included, so a rect factor
/// that reaches the binning body through an `apply_expression_template` call
/// site is reached AT THE CALL SITE — which is exactly where the rewrite must
/// land, so the shared template body stays untouched and singly-lowered
/// (esm-spec §9.6.4 Option B). Two binding spellings carry a rect factor and
/// both are handled: a subscripted binding (`{"F": index(src_W, "c")}`) by the
/// `index` arm, and a BARE FACTOR-NAME binding (`{"F": "src_W"}`, substituted
/// into the body's own `index(F, c)`) by the `bindings` arm. A bare string is
/// rewritten ONLY inside `bindings` — elsewhere a string is an `output_idx`
/// entry, a range key, a scalar field or a template `name`, none of which are
/// variable references.
fn pd_rewrite_rects(node: &mut Value, rectmap: &HashMap<String, String>) {
crate::json_visit::visit_values_mut(node, &mut |v| {
let Some(m) = v.as_object_mut() else { return };
if m.get("op").and_then(Value::as_str) == Some("index")
&& let Some(a) = m.get_mut("args").and_then(Value::as_array_mut)
&& let Some(first) = a.first_mut()
&& let Some(f) = first.as_str()
&& let Some(g) = rectmap.get(f)
{
*first = Value::String(g.clone());
}
if m.get("op").and_then(Value::as_str) == Some(APPLY_OP)
&& let Some(b) = m.get_mut("bindings").and_then(Value::as_object_mut)
{
for v in b.values_mut() {
if let Some(g) = v.as_str().and_then(|sv| rectmap.get(sv)) {
*v = Value::String(g.clone());
}
}
}
});
}
fn pd_ix(f: impl Into<Value>, idx: impl Into<Value>) -> Value {
json!({"op": "index", "args": [f.into(), idx.into()]})
}
/// One dict-form `join.overlap` clause (CONFORMANCE_SPEC.md §5.5.6 wire form).
/// `eps` is always `0.0`: the rewrite derives the envelopes from an EXACT
/// rectangle-containment predicate that stays on as the narrow `filter`, so no
/// FP slack is wanted.
fn pd_overlap_clause(src_env: &[String], tgt_env: &[String]) -> Value {
json!({
"overlap": {
"src_env": src_env.to_vec(),
"tgt_env": tgt_env.to_vec(),
"eps": 0.0,
}
})
}
/// Collect every factor name in `rectmap` that still appears in an
/// `index(F, …)` position — every occurrence [`pd_rewrite_rects`] targets but
/// did not reach.
fn pd_collect_stale_rects(
node: &Value,
rectmap: &HashMap<String, String>,
out: &mut std::collections::BTreeSet<String>,
) {
crate::json_visit::visit_values(node, &mut |_path, v| {
if let Some(m) = v.as_object()
&& m.get("op").and_then(Value::as_str) == Some("index")
&& let Some(f) = m
.get("args")
.and_then(Value::as_array)
.and_then(|a| a.first())
.and_then(Value::as_str)
&& rectmap.contains_key(f)
{
out.insert(f.to_string());
}
});
}
/// POST-CONDITION of the forward arm's rect re-pointing, discharged on the
/// EXPANDED form of the rewritten aggregate (esm-spec §9.6.4 rule 2: what the
/// evaluator sees is `Expand(tree)`).
///
/// `E`'s reduction axis now ranges over the COMPACT derived support set, so
/// every rect reference in its body must have become the corresponding
/// `pd_cell__*` gather. The rewrite achieves that by editing the CALL SITE,
/// which is what keeps the shared template body untouched. A rect factor named
/// FREE inside a template body is therefore unreachable: rewriting it would mean
/// rewriting the shared body, corrupting every other call site (the generated
/// producer `filter` among them, which must keep full-grid references). Left
/// alone it would index a compact per-support gather with full-grid positions —
/// WRONG NUMBERS, silently. Hence a hard error, whose remedy is the one the
/// template machinery already prescribes: bind the value through the params.
fn pd_assert_rects_rebound(
expr: &Value,
ename: &str,
rectmap: &HashMap<String, String>,
templates: Option<&Map<String, Value>>,
) -> Result<(), PushdownRewriteError> {
if rectmap.is_empty() {
return Ok(());
}
let expanded = pd_expand_for_detection(expr, templates);
let view = expanded.as_ref().unwrap_or(expr);
let mut stale = std::collections::BTreeSet::new();
pd_collect_stale_rects(view, rectmap, &mut stale);
if stale.is_empty() {
return Ok(());
}
let names: Vec<&str> = stale.iter().map(String::as_str).collect();
Err(PushdownRewriteError(format!(
"[template_body_references_pushdown_rewritten_variable] projection-pushdown desugar: the binning aggregate '{ename}' still reads '{}' after its reduction axis was re-pointed onto the generated derived support set. Those references live in an expression-template BODY, not in the call site's `bindings`, so the rewrite — which edits call sites only, to keep the template body shared and singly-lowered (esm-spec §9.6.4 Option B) — cannot re-point them, and they would index the compact per-support cell gathers with full-grid positions. Bind the value through the template's params, or write the binning body longhand.",
names.join("', '")
)))
}
/// Put a rewritten model's `equations` into the §5.5.7 canonical order:
///
/// 1. every equation whose LHS is **not** a bare variable — the derivative
/// equations and the generated `distinct` producer — each keeping its
/// relative input order;
/// 2. every **definition** (bare-variable LHS), sorted by the defined name,
/// lexicographically by UTF-8 code point.
///
/// This is NORMATIVE, not cosmetic, and it applies even though Rust's own
/// emission order is deterministic. The rewrite generates the member buffers
/// and the per-rect cell gathers while walking the model's variable collection,
/// which is a hash map in this binding and a `Dict` in the Julia reference —
/// so without canonicalizing, the emitted document varies with the hash seed
/// from run to run, and the `tests/conformance/pushdown/` goldens could not be
/// compared as ordered arrays at all. Appending in a fixed order would be
/// deterministic HERE while still disagreeing with the other bindings, which is
/// why the spec asks for the ordering rather than for stable iteration.
///
/// Sorting only the definitions is what keeps this safe for an authored
/// document: a definition is identified by the name it defines, so reordering
/// two of them cannot change the system (the evaluator dependency-orders
/// observeds itself), while a derivative equation's position among its peers is
/// left exactly as the author wrote it.
fn pd_canonicalize_equations(model: &mut Value) {
let Some(eqs) = model.get_mut("equations").and_then(Value::as_array_mut) else {
return;
};
// A stable partition, so group (1) keeps its relative input order.
let (mut structural, mut definitions): (Vec<Value>, Vec<Value>) = std::mem::take(eqs)
.into_iter()
.partition(|eq| !eq.get("lhs").is_some_and(Value::is_string));
definitions.sort_by(|a, b| {
let key = |e: &Value| {
e.get("lhs")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string()
};
key(a).cmp(&key(b))
});
structural.append(&mut definitions);
*eqs = structural;
}
/// The emission jobs' shared state: the plan under emission, the generated
/// construct names, the gather set, and its re-pointing map — derived once in
/// [`PdEmit::new`]. One method per emission job; [`pd_apply`] is the job list.
struct PdEmit<'p> {
plan: &'p Plan,
/// The derived index set's name (`pd_support__<c>`).
setname: String,
/// The generated producer aggregate's id (`pd_faq__<c>`).
faqid: String,
/// The member state variable's name (`pd_members__<c>`).
memvar: String,
/// The member_factor const parameter's name (`pd_member_factor__<c>`).
mfactor: String,
/// The gather set: the envelope factors PLUS every other array a binning
/// body reads on the cell axis. Envelopes lead, so a document whose bodies
/// read nothing else emits exactly what it emitted before; the rest follow in
/// sorted order (`cell_factors` is a BTreeMap).
rects: Vec<String>,
/// Rect factor name → its cell-gather observed's name.
rectmap: HashMap<String, String>,
/// The document's OWN index-set names, snapshotted before the emission
/// adds the derived one. A gather's trailing ranges are author-declared
/// axes, so this is the right set to check them against.
index_set_names: HashSet<String>,
}
impl<'p> PdEmit<'p> {
fn new(esm: &Value, plan: &'p Plan) -> Result<Self, PushdownRewriteError> {
let c = &plan.c_set;
let setname = format!("pd_support__{c}");
pd_refuse_ungatherable(&plan.cell_bad, c, &setname)?;
let index_set_names: HashSet<String> = esm
.get("index_sets")
.and_then(Value::as_object)
.map(|m| m.keys().cloned().collect())
.unwrap_or_default();
let mut rects: Vec<String> = Vec::new();
for f in &plan.tgt_env {
if !rects.contains(f) {
rects.push(f.clone());
}
}
for f in plan.cell_factors.keys() {
if !rects.contains(f) {
rects.push(f.clone());
}
}
let rectmap: HashMap<String, String> = rects
.iter()
.map(|f| (f.clone(), format!("pd_cell__{c}__{f}")))
.collect();
Ok(PdEmit {
plan,
setname,
faqid: format!("pd_faq__{c}"),
memvar: format!("pd_members__{c}"),
mfactor: format!("pd_member_factor__{c}"),
rects,
rectmap,
index_set_names,
})
}
/// The cell-gather observed's name for rect factor `f`.
fn cellgath(&self, f: &str) -> String {
format!("pd_cell__{}__{f}", self.plan.c_set)
}
/// The derived index set.
fn add_derived_index_set(
&self,
root: &mut Map<String, Value>,
) -> Result<(), PushdownRewriteError> {
let index_sets = root
.entry("index_sets")
.or_insert_with(|| Value::Object(Map::new()));
index_sets
.as_object_mut()
.ok_or_else(|| PushdownRewriteError("index_sets is not an object".into()))?
.insert(
self.setname.clone(),
json!({
"kind": "derived",
"from_faq": self.faqid,
"member_factor": self.mfactor,
}),
);
Ok(())
}
/// Producer filter comparisons, deep-copied from the representative E
/// BEFORE E is rewritten (they must keep full-grid rect factor refs).
/// Read off the DEFINING EQUATION: esm 1.0.0 has no variable
/// `expression` field.
fn producer_filter(
&self,
model: &Value,
templates: Option<&Map<String, Value>>,
) -> Result<Value, PushdownRewriteError> {
let repexpr = pd_def(model, &self.plan.rep_ename)
.cloned()
.ok_or_else(|| {
PushdownRewriteError("representative E lost its defining equation".into())
})?;
// When the call site hides the predicate behind a template reference, read it
// off the EXPANDED body (§9.6.4 rule 2) — the producer wants the FULL-GRID
// rect references, which is exactly what the pre-rewrite expansion yields.
// The expansion is a scratch value: nothing of it is emitted except these
// comparisons, so the document's template block and call sites are untouched.
// A template-free document never builds one, so its emitted filter is
// byte-identical to before.
let rep_expanded = repexpr
.get("expr")
.and_then(|e| pd_expand_for_detection(e, templates));
let ifcond = repexpr
.get("expr")
.and_then(pd_find_ifelse_cond)
.or_else(|| rep_expanded.as_ref().and_then(pd_find_ifelse_cond))
.ok_or_else(|| {
PushdownRewriteError(
"pushdown desugar: representative E lost its containment ifelse".into(),
)
})?;
let comps: Vec<Value> = if matches!(op_of(ifcond), Some("and") | Some("*")) {
ifcond
.get("args")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default()
} else {
vec![ifcond.clone()]
};
Ok(json!({"op": "*", "args": comps}))
}
/// Member state var + member_factor param.
fn add_member_vars(&self, model: &mut Value) -> Result<(), PushdownRewriteError> {
let mv = model
.get_mut("variables")
.and_then(Value::as_object_mut)
.ok_or_else(|| PushdownRewriteError("model variables is not an object".into()))?;
mv.insert(
self.memvar.clone(),
json!({"type": "unknown", "shape": [self.setname.clone()]}),
);
mv.insert(
self.mfactor.clone(),
json!({"type": "parameter", "default": 0.0, "shape": [self.setname.clone()]}),
);
Ok(())
}
/// Per-rect cell-gather observeds.
///
/// The DECLARATION is a bare `unknown`; what makes it observed is the
/// bare-variable-LHS equation added alongside (esm-spec §6.3.1).
fn add_cell_gathers(&self, model: &mut Value) -> Result<(), PushdownRewriteError> {
let c = &self.plan.c_set;
let mv = model
.get_mut("variables")
.and_then(Value::as_object_mut)
.ok_or_else(|| PushdownRewriteError("model variables is not an object".into()))?;
let mut cellgath_defs: Vec<(String, Value)> = Vec::new();
let mut sorted_rects: Vec<&String> = self.rects.iter().collect();
sorted_rects.sort();
for f in sorted_rects {
let fallback: Vec<String> = mv
.get(f)
.and_then(|v| v.get("shape"))
.and_then(Value::as_array)
.map(|a| {
a.iter()
.map(|t| t.as_str().unwrap_or_default().to_string())
.collect()
})
.unwrap_or_else(|| vec![c.clone()]);
let shape: &[String] = self.plan.cell_factors.get(f).unwrap_or(&fallback);
let (decl, defn) = pd_gather_defn(
f,
shape,
&self.setname,
&self.mfactor,
&self.index_set_names,
c,
)?;
mv.insert(self.cellgath(f), decl);
cellgath_defs.push((self.cellgath(f), defn));
}
// The cell-gather DEFINITIONS, now that the `variables` borrow is done.
for (name, rhs) in cellgath_defs {
pd_set_def(model, &name, rhs);
}
Ok(())
}
/// Gate the provider-backed arrays onto the derived axis.
fn gate_provider_arrays(&self, model: &mut Value) -> Result<(), PushdownRewriteError> {
let mv = model
.get_mut("variables")
.and_then(Value::as_object_mut)
.ok_or_else(|| PushdownRewriteError("model variables is not an object".into()))?;
for a in &self.plan.a_names {
if let Some(av) = mv.get_mut(a).and_then(Value::as_object_mut) {
av.insert(
"shape".to_string(),
json!([self.setname.clone(), self.plan.rcv_set.clone()]),
);
}
}
Ok(())
}
/// Rewrite E: axis -> derived set, rect factors -> cell gathers, + GATE.
///
/// The rewritten `E` still reduces over the FULL record axis, so without a
/// gate it visits |support|*|records| pairs -- 1520*43650 on isrm.esm.
/// Attach the SAME overlap clause the producer carries, re-pointed at the
/// generated cell gathers, and the enumeration driver (§5.5.6) walks one
/// candidate partner list per output cell instead. The clause is derived,
/// not authored: its envelopes are exactly the ones `pd_parse_containment`
/// read out of this aggregate's own containment predicate.
fn rewrite_forward_es(
&self,
model: &mut Value,
templates: Option<&Map<String, Value>>,
) -> Result<(), PushdownRewriteError> {
for spec in &self.plan.e_specs {
let ename = &spec.ename;
let expr = pd_def_mut(model, ename).ok_or_else(|| {
PushdownRewriteError(format!("E '{ename}' lost its defining equation"))
})?;
if let Some(from) = expr
.get_mut("ranges")
.and_then(|r| r.get_mut(&spec.c_sym))
.and_then(Value::as_object_mut)
{
from.insert("from".to_string(), Value::String(self.setname.clone()));
}
pd_rewrite_rects(expr, &self.rectmap);
if let Some(args) = expr.get_mut("args").and_then(Value::as_array_mut) {
for s in args.iter_mut() {
if let Some(name) = s.as_str()
&& let Some(g) = self.rectmap.get(name)
{
*s = Value::String(g.clone());
}
}
}
if expr.get("join").is_none()
&& let Some(eo) = expr.as_object_mut()
{
let gathered: Vec<String> = spec
.tgt_env
.iter()
.map(|f| self.rectmap.get(f).cloned().unwrap_or_else(|| f.clone()))
.collect();
eo.insert(
"join".to_string(),
Value::Array(vec![pd_overlap_clause(&spec.src_env, &gathered)]),
);
}
let expr_snapshot = expr.clone();
if let Some(evo) = model
.get_mut("variables")
.and_then(|v| v.get_mut(ename))
.and_then(Value::as_object_mut)
{
evo.insert("shape".to_string(), json!([self.setname.clone()]));
}
pd_assert_rects_rebound(&expr_snapshot, ename, &self.rectmap, templates)?;
}
Ok(())
}
/// MIRRORED orientation: gate only.
///
/// A per-record binning aggregate `P[r] = SUM_{c in C} [contains(cell_c, pt_r)]*...`
/// is the same join read the other way round. It gets ONLY the gate -- no
/// derived index set, no `distinct` producer, no `member_factor`, no
/// provider gating -- because it wants the FULL record axis: every record
/// must produce a value, and a record outside the grid must come out as the
/// semiring identity (the driver leaves such a position with no term and the
/// identity fill emits 0). There is nothing to compact, so a mirrored
/// value-invention would derive a support set nobody reads. Its envelopes
/// stay the document's own const-array factors (the cell axis is not
/// re-pointed), so the mirror also needs no rect gathers.
fn gate_mirrors(&self, model: &mut Value) {
for (pname, p_src, p_tgt) in &self.plan.mirror_specs {
let Some(pexpr) = pd_def_mut(model, pname) else {
continue;
};
if pexpr.get("join").is_none()
&& let Some(po) = pexpr.as_object_mut()
{
po.insert(
"join".to_string(),
Value::Array(vec![pd_overlap_clause(p_src, p_tgt)]),
);
}
}
}
/// Restrict the conc reductions to the derived axis.
fn restrict_conc(&self, model: &mut Value) {
for (cname, ssym) in &self.plan.conc_specs {
if let Some(from) = pd_def_mut(model, cname)
.and_then(|e| e.get_mut("ranges"))
.and_then(|r| r.get_mut(ssym))
.and_then(Value::as_object_mut)
{
from.insert("from".to_string(), Value::String(self.setname.clone()));
}
}
}
/// Generated `distinct` producer (reuses E's containment + geometry).
fn append_producer(&self, model: &mut Value, prod_filter: Value) {
let mut prod_args: Vec<String> = Vec::new();
for s in self.plan.src_env.iter().chain(self.plan.tgt_env.iter()) {
if !prod_args.contains(s) {
prod_args.push(s.clone());
}
}
let mut prod_ranges = Map::new();
prod_ranges.insert(
self.plan.rep_rsym.clone(),
json!({"from": self.plan.r_set.clone()}),
);
prod_ranges.insert(
self.plan.rep_csym.clone(),
json!({"from": self.plan.c_set.clone()}),
);
let producer = json!({
"lhs": pd_ix(self.memvar.clone(), "m"),
"rhs": {
"op": "faq",
"output_idx": ["m"],
"ranges": prod_ranges,
"expr": {"op": "true", "args": []},
"distinct": true,
"semiring": "bool_and_or",
"id": self.faqid.clone(),
"join": [pd_overlap_clause(&self.plan.src_env, &self.plan.tgt_env)],
"filter": prod_filter,
"key": {"op": "skolem", "label": "cell", "args": [self.plan.rep_csym.clone()]},
"args": prod_args,
},
});
let eqs = model
.as_object_mut()
.expect("model is an object")
.entry("equations")
.or_insert_with(|| Value::Array(Vec::new()));
match eqs.as_array_mut() {
Some(list) => list.push(producer),
None => {
*eqs = Value::Array(vec![producer]);
}
}
}
/// Inspectable pushdown provenance / gated_select record.
fn record_provenance(self, root: &mut Map<String, Value>) -> Result<(), PushdownRewriteError> {
let md = root
.entry("metadata")
.or_insert_with(|| Value::Object(Map::new()));
let md = md
.as_object_mut()
.ok_or_else(|| PushdownRewriteError("metadata is not an object".into()))?;
let xesd = md
.entry("x_esd")
.or_insert_with(|| Value::Object(Map::new()));
xesd.as_object_mut()
.ok_or_else(|| PushdownRewriteError("metadata.x_esd is not an object".into()))?
.insert(
"pushdown".to_string(),
json!({
"derived_set": self.setname,
"producer_id": self.faqid,
"member_factor": self.mfactor,
"member_var": self.memvar,
"gated_select": {
"gated_by": self.setname,
"applies_to": self.plan.a_names.clone(),
"gated_axis": 0,
},
}),
);
Ok(())
}
}
fn pd_apply(
esm: &Value,
mname: &str,
plan: &Plan,
templates: Option<&Map<String, Value>>,
) -> Result<Value, PushdownRewriteError> {
let mut d = esm.clone(); // fresh, mutable (input purity)
let emit = PdEmit::new(esm, plan)?;
let root = d
.as_object_mut()
.ok_or_else(|| PushdownRewriteError("document is not an object".into()))?;
emit.add_derived_index_set(root)?;
let model = root
.get_mut("models")
.and_then(|m| m.get_mut(mname))
.filter(|m| m.is_object())
.ok_or_else(|| PushdownRewriteError(format!("model '{mname}' is not an object")))?;
let prod_filter = emit.producer_filter(model, templates)?;
emit.add_member_vars(model)?;
emit.add_cell_gathers(model)?;
emit.gate_provider_arrays(model)?;
emit.rewrite_forward_es(model, templates)?;
emit.gate_mirrors(model);
emit.restrict_conc(model);
emit.append_producer(model, prod_filter);
// --- canonical equation order (CONFORMANCE_SPEC.md §5.5.7) ---------------
pd_canonicalize_equations(model);
emit.record_provenance(root)?;
Ok(d)
}
/// Recognise the projection-pushdown pattern in `esm`'s named model and, when
/// it matches, return a NEW document (`Cow::Owned`) with the four constructs
/// desugared in (a `kind:"derived"` index set, a `distinct:true` overlap-gated
/// producer aggregate, a `member_factor` const parameter, and an inspectable
/// `gated_select` record) plus the reduction axis of the matched E / A / conc
/// nodes re-pointed onto the generated derived set. Returns the input
/// UNCHANGED (`Cow::Borrowed`) when no model is selected, the pattern does not
/// match, or the reduction's semiring is not the additive `(+, 0)` monoid
/// (the soundness guard).
///
/// IDEMPOTENT: a document already carrying the provenance record
/// `metadata.x_esd.pushdown` is returned unchanged — the generated constructs
/// would otherwise re-match and stack a second `pd_support__pd_support__…`
/// layer.
pub fn desugar_pushdown<'a>(
esm: &'a Value,
model_name: Option<&str>,
) -> Result<Cow<'a, Value>, PushdownRewriteError> {
if !esm.is_object() || pushdown_record(esm).is_some() {
return Ok(Cow::Borrowed(esm));
}
if esm.get("models").and_then(Value::as_object).is_none() {
return Ok(Cow::Borrowed(esm));
}
let Some(mname) = pd_model_name(esm, model_name) else {
return Ok(Cow::Borrowed(esm));
};
let Some(model) = esm.get("models").and_then(|m| m.get(&mname)) else {
return Ok(Cow::Borrowed(esm));
};
if !model.is_object() {
return Ok(Cow::Borrowed(esm));
}
let Some((plan, diags, templates)) = pd_analyze(model) else {
return Ok(Cow::Borrowed(esm));
};
// RESIDUAL DIAGNOSTICS (CONFORMANCE_SPEC §5.5.7): a join-shaped aggregate the
// recogniser could NOT read is reported here, not swallowed. See
// `pd_binning_refusal` for the "not a join" / "a join I could not read"
// split, and `pushdown_diagnostics` for the inspectable form.
for d in &diags {
eprintln!("warning: {}", pd_diagnostic_message(d));
}
let Some(plan) = plan else {
return Ok(Cow::Borrowed(esm));
};
pd_apply(esm, &mname, &plan, templates.as_ref()).map(Cow::Owned)
}
/// The residual diagnostics [`desugar_pushdown`] would emit for `esm`.
///
/// One record per aggregate that IS join-shaped (it bins records into the cells
/// of an index set and feeds a provider-backed rank-2 array through a
/// `+`-semiring mat-vec) but whose containment predicate the recogniser could
/// not read, so the rewrite does not fire for it and that array is fetched
/// WHOLESALE.
///
/// Inspectable, side-effect-free counterpart of the warning stream: same
/// records, same order (sorted by `variable`/`consumer`/`array`), stable field
/// set (`code`, `variable`, `consumer`, `array`, `index_set`, `reason`,
/// `template`, `consequence`), pinned across bindings by the
/// `tests/conformance/pushdown/` corpus. Empty for a document that already
/// carries the rewrite record, for one with no model selected, and —
/// deliberately — for one that simply is NOT join-shaped: "no join here" is not
/// a defect.
pub fn pushdown_diagnostics(esm: &Value, model_name: Option<&str>) -> Vec<Value> {
if !esm.is_object() || pushdown_record(esm).is_some() {
return Vec::new();
}
let Some(mname) = pd_model_name(esm, model_name) else {
return Vec::new();
};
let Some(model) = esm.get("models").and_then(|m| m.get(&mname)) else {
return Vec::new();
};
pd_analyze(model).map(|(_, d, _)| d).unwrap_or_default()
}
/// The ONE detection entry point shared by [`desugar_pushdown`] (which then
/// emits) and [`pushdown_diagnostics`] (which only reports): run the matcher on
/// the EXPANDED view and hand back the plan (`None` ⇒ the pattern did not
/// match), the residual diagnostics, and the component template registry the
/// emission side needs.
#[allow(clippy::type_complexity)]
fn pd_analyze(model: &Value) -> Option<(Option<Plan>, Vec<Value>, Option<Map<String, Value>>)> {
model.get("variables")?.as_object()?;
let defs = pd_detection_defs(model);
let (plan, diags) = pd_detect(model, &defs);
Some((plan, diags, pd_templates(model).cloned()))
}
// --------------------------------------------------------------------------- //
// RECORD-DERIVED PROVIDER GATING (the Julia Phase-1 helpers, raw-JSON side).
// --------------------------------------------------------------------------- //
/// One native axis of a loader selection — the vocabulary shared by a data
/// loader's declared `select.axes` (esm-spec §8.9) and by the pushdown record's
/// gate template. JSON spellings, in the same order as the variants:
///
/// ```text
/// "all" every index of the axis
/// {"fixed": 0} / {"fixed": [0]} index 0, and the axis is DROPPED
/// {"range": {"start": 0, "stop": 52411}} a strided prefix/window (step ≥ 1)
/// {"gated_by": "<derived index set>"} the set's materialised members
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GateAxis {
/// Full native axis.
All,
/// Take native index `i` (0-based) and DROP the axis.
Fixed(usize),
/// The half-open strided range `[start, stop)` by `step`, as the new axis.
/// Length `ceil((stop - start) / step)`; the axis is kept.
Range {
/// Inclusive first index (0-based).
start: usize,
/// Exclusive last index.
stop: usize,
/// Stride (>= 1).
step: usize,
},
/// The named derived set's materialised members, in the set's canonical
/// (sorted) member order, as the new compact axis.
GatedBy(String),
}
/// Parse one JSON axis selector of the [`GateAxis`] vocabulary.
///
/// `ctx` names the declaring site for the error message. `gated_by_override`,
/// when given, replaces the declared set name — the pushdown path substitutes
/// its GENERATED set name into a loader's authored `{"gated_by": …}` slot.
pub fn parse_select_axis(
ctx: &str,
ax: &Value,
gated_by_override: Option<&str>,
) -> Result<GateAxis, PushdownRewriteError> {
let bad = |detail: String| PushdownRewriteError(format!("{ctx}: {detail}"));
if ax.as_str() == Some("all") || ax.is_null() {
return Ok(GateAxis::All);
}
let Some(m) = ax.as_object() else {
return Err(bad(format!(
"unrecognised axis selector {ax}; expected \"all\", {{\"fixed\": i}}, \
{{\"range\": {{\"start\": s, \"stop\": e}}}} or {{\"gated_by\": \"<set>\"}}"
)));
};
if let Some(g) = m.get("gated_by") {
let name = match gated_by_override {
Some(o) => o.to_string(),
None => g
.as_str()
.ok_or_else(|| bad("\"gated_by\" must name a derived index set".into()))?
.to_string(),
};
return Ok(GateAxis::GatedBy(name));
}
if let Some(fx) = m.get("fixed") {
let fi = match fx {
Value::Array(a) => a.first().and_then(Value::as_u64),
other => other.as_u64(),
}
.ok_or_else(|| bad("\"fixed\" must be a non-negative integer index".into()))?;
return Ok(GateAxis::Fixed(fi as usize));
}
if let Some(r) = m.get("range") {
let obj = r
.as_object()
.ok_or_else(|| bad("\"range\" must be an object {start, stop, step?}".into()))?;
let start = obj.get("start").and_then(Value::as_u64).unwrap_or(0) as usize;
let stop = obj
.get("stop")
.and_then(Value::as_u64)
.ok_or_else(|| bad("\"range\" needs an integer \"stop\"".into()))?
as usize;
let step = obj.get("step").and_then(Value::as_u64).unwrap_or(1) as usize;
if step == 0 {
return Err(bad("\"range.step\" must be >= 1".into()));
}
if stop < start {
return Err(bad(format!(
"\"range\" is empty: stop {stop} precedes start {start}"
)));
}
return Ok(GateAxis::Range { start, stop, step });
}
let mut keys: Vec<&str> = m.keys().map(String::as_str).collect();
keys.sort_unstable();
Err(bad(format!(
"unrecognised axis selector keys {keys:?}; expected one of fixed, range, gated_by"
)))
}
/// Parse a whole `axes` array of the [`GateAxis`] vocabulary (see
/// [`parse_select_axis`]).
pub fn parse_select_axes(
ctx: &str,
axes: &[Value],
gated_by_override: Option<&str>,
) -> Result<Vec<GateAxis>, PushdownRewriteError> {
axes.iter()
.map(|ax| parse_select_axis(ctx, ax, gated_by_override))
.collect()
}
/// A provider-key ⇒ engine gate: per-NATIVE-axis selection plus the LOADER
/// variable tails the gate applies to.
#[derive(Debug, Clone)]
pub struct ProviderGate {
pub axes: Vec<GateAxis>,
pub applies_to: Vec<String>,
}
/// The `(external name, model variable)` routings of a raw document: which
/// external field feeds which declared array.
///
/// The `from` side is `"<source>.<file_variable>"` and the `to` side is the
/// consuming parameter's namespaced name. A PROVIDER is keyed by the `to` side —
/// the consuming parameter is the only spelling that names one field and every
/// field, since a source declares no variables — so [`pushdown_provider_gates`]
/// matches on `to`. These pairs remain the routing that lets an array supplied
/// under the SOURCE's spelling still reach its consumer.
///
/// Two spellings are read. A data source binds to the consuming parameter
/// through that parameter's own
/// `update: {kind: "data", source, from: {file_variable}}` (esm-spec §5.4,
/// §8.5); a NON-source producer may still feed a parameter through a
/// `variable_map` coupling edge. Either way the pair is "what feeds what".
pub fn pushdown_coupling_pairs(doc: &Value) -> Vec<(String, String)> {
let mut out = Vec::new();
// (1) A parameter bound to a data source by its own `update` (1.0.0).
if let Some(models) = doc.get("models").and_then(Value::as_object) {
let mut mnames: Vec<&String> = models.keys().collect();
mnames.sort();
for mname in mnames {
let Some(vars) = models[mname].get("variables").and_then(Value::as_object) else {
continue;
};
let mut vnames: Vec<&String> = vars.keys().collect();
vnames.sort();
for vname in vnames {
let Ok(var) =
serde_json::from_value::<crate::types::ModelVariable>(vars[vname].clone())
else {
continue;
};
for rule in var.update.iter().flat_map(|spec| spec.rules()) {
let (Some(source), Some(binding)) = (
rule.data_source(),
rule.value().and_then(|v| v.from.as_ref()),
) else {
continue;
};
out.push((
format!("{source}.{}", binding.file_variable),
format!("{mname}.{vname}"),
));
}
}
}
}
// (2) A `variable_map` coupling edge (the 0.x spelling, still admissible
// between two ordinary components).
if let Some(cp) = doc.get("coupling").and_then(Value::as_array) {
for c in cp {
if c.get("type").and_then(Value::as_str) != Some("variable_map") {
continue;
}
let frm = c.get("from").and_then(Value::as_str).unwrap_or("");
let to = c.get("to").and_then(Value::as_str).unwrap_or("");
if !frm.is_empty() && !to.is_empty() {
out.push((frm.to_string(), to.to_string()));
}
}
}
out
}
/// Rank of the (rewritten) gated model arrays — the fallback native rank when
/// a loader declares no axes template (2 for the ISRM shape; read from the
/// document rather than hard-coded).
fn pushdown_gated_rank(doc: &Value, applies: &[String]) -> usize {
if let Some(models) = doc.get("models").and_then(Value::as_object) {
for m in models.values() {
let Some(mv) = m.get("variables").and_then(Value::as_object) else {
continue;
};
for a in applies {
if let Some(shp) = mv
.get(a)
.and_then(|v| v.get("shape"))
.and_then(Value::as_array)
&& !shp.is_empty()
{
return shp.len();
}
}
}
}
2
}
/// Per-NATIVE-axis gate `axes` for `loader`: the loader's declared
/// `metadata.x_esd.gated_select.axes` template with the GENERATED set name
/// substituted into its `gated_by` slot (validated against the record's
/// `gated_axis`); else a rank-`mrank` all-axes gate with `gated_by` at
/// `gaxis`.
fn pushdown_gate_axes(
doc: &Value,
loader: &str,
gset: &str,
gaxis: i64,
mrank: usize,
) -> Result<Vec<GateAxis>, PushdownRewriteError> {
let tpl = doc
.get("data_sources")
.and_then(|d| d.get(loader))
.and_then(|l| l.get("metadata"))
.and_then(|m| m.get("x_esd"))
.and_then(|x| x.get("gated_select"))
.and_then(|g| g.get("axes"))
.and_then(Value::as_array);
if let Some(tpl) = tpl {
let axes = parse_select_axes(
&format!("data_sources.{loader} gated_select template"),
tpl,
Some(gset),
)?;
// The gated axis's position among the axes the fetch KEEPS (a `fixed`
// axis is dropped, so it does not shift the model's axis numbering).
let mut nonfixed: i64 = 0;
let mut gpos: i64 = -1;
for ax in &axes {
match ax {
GateAxis::Fixed(_) => {}
GateAxis::GatedBy(_) => {
gpos = nonfixed;
nonfixed += 1;
}
_ => nonfixed += 1,
}
}
if gpos != gaxis {
return Err(PushdownRewriteError(format!(
"data_sources.{loader} gated_select template puts the gated axis at \
non-fixed position {gpos}, but the rewrite record gates model axis \
{gaxis} — the loader template and the rewritten arrays disagree"
)));
}
return Ok(axes);
}
if gaxis < 0 || gaxis as usize >= mrank {
return Err(PushdownRewriteError(format!(
"rewrite record gated_axis {gaxis} out of range for rank-{mrank} gated arrays"
)));
}
let mut axes = vec![GateAxis::All; mrank];
axes[gaxis as usize] = GateAxis::GatedBy(gset.to_string());
Ok(axes)
}
/// Provider-key ⇒ engine gate, derived from `doc`'s rewrite record
/// (`metadata.x_esd.pushdown.gated_select`).
///
/// A provider is GATED when its key names a `data_sources` variable
/// (`"<Loader>"` or `"<ModelPath>.<param>"`) that the document's `update`
/// rules — or a coupling `variable_map` — route
/// onto one of the record's `applies_to` model arrays. The gate's per-NATIVE-
/// axis `axes` come from the loader's own `metadata.x_esd.gated_select.axes`
/// template when it declares one (with the record's GENERATED set name
/// substituted), else from the model array's rank with `gated_by` at the
/// record's `gated_axis`. `applies_to` carries the LOADER-variable tails.
/// Empty when `doc` carries no record or no coupling routes a provider onto a
/// gated array.
pub fn pushdown_provider_gates(
doc: &Value,
provider_keys: &[String],
) -> Result<HashMap<String, ProviderGate>, PushdownRewriteError> {
let mut gates = HashMap::new();
let Some(rec) = pushdown_record(doc) else {
return Ok(gates);
};
let Some(gs) = rec.get("gated_select").and_then(Value::as_object) else {
return Ok(gates);
};
let applies: Vec<String> = gs
.get("applies_to")
.and_then(Value::as_array)
.map(|a| a.iter().map(value_to_display_string).collect())
.unwrap_or_default();
let gset = gs
.get("gated_by")
.map(value_to_display_string)
.unwrap_or_default();
let gaxis = gs.get("gated_axis").and_then(Value::as_i64).unwrap_or(0);
if applies.is_empty() || gset.is_empty() {
return Ok(gates);
}
// routing: "<Loader>.<var>" => the gated model array (its `to` side is the
// consuming parameter, which is what a provider is keyed by).
let mut fed: Vec<(String, String)> = Vec::new();
for (frm, to) in pushdown_coupling_pairs(doc) {
if !frm.contains('.') {
continue;
}
let tail = to.rsplit('.').next().unwrap_or(&to);
if applies.iter().any(|a| a == tail) {
fed.push((frm, to));
}
}
if fed.is_empty() {
return Ok(gates);
}
let mrank = pushdown_gated_rank(doc, &applies);
for k in provider_keys {
let (loader, lvars) = if let Some((frm, to)) = fed.iter().find(|(_, to)| to == k) {
// A provider for ONE data-fed parameter: its key IS the consumer, so
// the SOURCE comes from the routing rather than from the key prefix
// (which names the consuming MODEL).
let loader = frm.split_once('.').map(|(l, _)| l).unwrap_or(frm.as_str());
let tail = to.rsplit('.').next().unwrap_or(to);
(loader.to_string(), vec![tail.to_string()])
} else {
// A provider for a WHOLE source, serving several of its columns.
let mut lvars: Vec<String> = fed
.iter()
.filter_map(|(f, to)| {
let (l, _) = f.split_once('.')?;
(l == k).then(|| to.rsplit('.').next().unwrap_or(to).to_string())
})
.collect();
if lvars.is_empty() {
continue;
}
lvars.sort();
lvars.dedup();
(k.clone(), lvars)
};
let axes = pushdown_gate_axes(doc, &loader, &gset, gaxis, mrank)?;
gates.insert(
k.clone(),
ProviderGate {
axes,
applies_to: lvars,
},
);
}
Ok(gates)
}