///|
/// The crossing-soundness rules the predicate pass applies on its descent:
/// the per-expression analyses that decide whether
/// dropping rows before a stage can change its output, its errors, or the
/// predicate's own verdicts. `row_stable` (row-local, non-value-raising),
/// `signed_zero_safe` (signed-zero safety on a `Float` group key),
/// `group_cell_stable` (an aggregation's per-group cell is neither null nor
/// value-raising), and the pass-through tests (bare-column coverage,
/// row-count preservation, `defines_none_of`).

///|
/// Whether evaluating `expr` is invariant under dropping *other* rows of
/// its scope: each output cell depends only on its own row (literals
/// broadcast), and no cell can raise on a value. Aggregations fail the
/// first half — their value is computed over the whole scope, so a
/// filtered scope changes it. `Cast` fails the second half — its parse /
/// range rejection raises on the *values* it meets, so dropping a row
/// can drop the very cell that would have raised. `Map` and `MapBatches` fail
/// it the same way — an opaque closure can `raise` on the values it meets — so
/// both are value barriers, exactly like `Cast`. A `LitSeries` is a *shape*
/// barrier: a frame-tall literal column is positional (cell `i` is row `i`),
/// so dropping rows before the stage would leave it mismatched against the
/// now-shorter frame — pushing a filter past it could turn a working plan
/// into a `LengthMismatch`, so no filter crosses it either. Everything else
/// (column reads, scalar literals, the operators, the total unary probes,
/// the string-namespace transforms, aliasing, ternary selection) is
/// row-local and raises only on **scope-independent** facts, which is the
/// property that matters here rather than totality: the dtype-level
/// `TypeMismatch` of a non-String operand, and — for the regex `Str` ops — the
/// `InvalidOperation` of a pattern that does not compile. That compile happens
/// once per evaluation, hoisted out of the per-cell map,
/// so it depends on the pattern and not on the rows
/// present: dropping rows cannot make an invalid pattern compile, or a valid one
/// fail. Only a per-*value* raise would be unsound to sink across, since the
/// dropped row could be the one that raised.
fn row_stable(expr : @ir.ExprNode) -> Bool {
  // Iterative tree walk (explicit stack, not the call stack) so a deeply nested
  // predicate cannot overflow and abort during optimization — total at any
  // depth, as this function's doc promises. `children()` drives the *descent*,
  // but the classification below is per variant and the `match` is exhaustive by
  // design (no wildcard): a new `ExprNode` variant does not inherit a verdict
  // here, it fails to compile until someone decides whether it is row-local.
  let stack : Array[@ir.ExprNode] = [expr]
  for ;; {
    match stack.pop() {
      None => break
      Some(e) =>
        match e {
          @ir.ExprNode::Agg(_, _)
          | @ir.ExprNode::Cast(_, _)
          | @ir.ExprNode::Map(_, _, _)
          | @ir.ExprNode::MapBatches(_, _, _, _)
          | @ir.ExprNode::LitSeries(_) => return false
          @ir.ExprNode::Col(_)
          | @ir.ExprNode::Lit(_)
          | @ir.ExprNode::Binary(_, _, _)
          | @ir.ExprNode::Unary(_, _)
          | @ir.ExprNode::Str(_, _)
          | @ir.ExprNode::Alias(_, _)
          | @ir.ExprNode::Ternary(_, _, _)
          | @ir.ExprNode::FillNull(_, _)
          | @ir.ExprNode::FillNan(_, _)
          | @ir.ExprNode::IsIn(_, _)
          | @ir.ExprNode::IsBetween(_, _, _, _) =>
            for child in e.children() {
              stack.push(child)
            }
        }
    }
  }
  true
}

///|
/// Whether `expr` cannot observe the sign of a zero — the extra gate the
/// `Aggregate` predicate pushdown needs to stay sound on a `Float` group key.
/// `KeyCell` folds `-0.0` and `+0.0` into one group, but the group's output row
/// keeps its *representative* row's signed key, so a predicate that can tell the
/// two apart disagrees between a group's individual rows (filtered below the
/// aggregate) and that representative (filtered above it). Division and a
/// negative exponent are the operators in the vocabulary that make a zero's sign
/// observable — `1.0 / +0.0 = +inf` but `1.0 / -0.0 = -inf`, and
/// `(+0.0) ** -1.0 = +inf` but `(-0.0) ** -1.0 = -inf`, while `+` / `-` / `*`
/// / `%` and every comparison treat `-0.0` and `+0.0` as equal (remainder can
/// carry a zero's sign into its result but no comparison downstream can read it
/// back out) — so a signed-zero-safe predicate gives every row of a folded group
/// the same verdict and may sink below the aggregate, while one that divides or
/// exponentiates could drop only part of the group and must stay above. Walks
/// `children()` like `row_stable`, so a wrapper node kind added later is walked
/// without this rule naming it — but the `BinOp` match below is exhaustive:
/// a new operator fails compilation here until it is classified, because the
/// silent default would be the unsound one (a sign-reading operator passing as
/// safe could drop half a folded group).
fn signed_zero_safe(expr : @ir.ExprNode) -> Bool {
  // Iterative walk like `row_stable`: a deeply nested predicate stays total
  // instead of overflowing the call stack during optimization.
  let stack : Array[@ir.ExprNode] = [expr]
  for ;; {
    match stack.pop() {
      None => break
      Some(e) =>
        match e {
          @ir.ExprNode::Binary(op, _, _) =>
            match op {
              // The sign-observing operators fence the descent.
              @ir.BinOp::Div | @ir.BinOp::FloorDiv | @ir.BinOp::Pow =>
                return false
              // Everything else in the operator vocabulary treats `-0.0` and
              // `+0.0` as equal (arithmetic that cannot carry the sign into a
              // readable position, and every comparison), so the walk descends.
              @ir.BinOp::Add
              | @ir.BinOp::Sub
              | @ir.BinOp::Mul
              | @ir.BinOp::Mod
              | @ir.BinOp::Eq
              | @ir.BinOp::Ne
              | @ir.BinOp::Lt
              | @ir.BinOp::Le
              | @ir.BinOp::Gt
              | @ir.BinOp::Ge
              | @ir.BinOp::And
              | @ir.BinOp::Or =>
                for child in e.children() {
                  stack.push(child)
                }
            }
          _ =>
            for child in e.children() {
              stack.push(child)
            }
        }
    }
  }
  true
}

///|
/// Whether an aggregation expression's per-group cell is provably
/// independent of *which other groups exist*: never changing value,
/// null-ness, or raised error when groups appear or disappear around it.
/// The gate is the group-invariance argument stated once in the
/// crossing-rules header of `push_predicates`, the pass driver this rule
/// serves; per op it says:
/// `sum` / `count` / `n_unique` always produce a non-null value, and
/// `mean` / `min` / `max` are pure functions of their group's cells
/// (null exactly when an all-null group makes them so) — all four cross
/// **only over a `row_stable` operand**, because a `sum(col.cast(Int))` or
/// `sum(col.map(…))` still raises on the cells its operand meets, so a
/// value-raising operand (a nested `Cast` / `Map`, or an `Agg` /
/// `LitSeries`) fences the filter above exactly as a top-level cast does.
/// The sample statistics (`std` / `var` / `median`) and `first` / `last`
/// stay fenced as a conservative bound — loosening them wants the same
/// argument per op and no consumer has asked for it.
/// A `cast`'s rejection is value-dependent, as is a
/// `map`'s opaque closure (null- or raise-valued on the cells it meets);
/// the null literal is a null cell by definition; a bare column — or a
/// `lit_series`, equally row-wise — is not reduction-shaped at all (the
/// executor raises structurally — fencing keeps that error exactly where the
/// as-built plan reports it). The
/// combinators preserve the property cell-wise: arithmetic, comparison,
/// and the Kleene connectives over non-null cells stay non-null, as do
/// the unary wrappers, the string-namespace transforms, and a ternary over
/// a non-null condition.
fn group_cell_stable(expr : @ir.ExprNode) -> Bool {
  // Iterative walk like `row_stable`: total at any predicate depth. An `Agg`
  // node is terminal here — its verdict is its op's (the reducing ops defer
  // to `row_stable` of the operand) — so its
  // operand is never pushed for further `group_cell_stable` descent.
  let stack : Array[@ir.ExprNode] = [expr]
  for ;; {
    match stack.pop() {
      None => break
      Some(e) =>
        match e {
          @ir.ExprNode::Col(_) => return false
          @ir.ExprNode::Lit(value) =>
            match value {
              @types.Scalar::Null => return false
              @types.Scalar::Int(_)
              | @types.Scalar::Float(_)
              | @types.Scalar::Bool(_)
              | @types.Scalar::String(_)
              | @types.Scalar::Date(_) => ()
            }
          @ir.ExprNode::Agg(op, operand) =>
            match op {
              @series.ReduceOp::Sum
              | @series.ReduceOp::Count
              | @series.ReduceOp::NUnique
              | @series.ReduceOp::Mean
              | @series.ReduceOp::Min
              | @series.ReduceOp::Max =>
                if !row_stable(operand) {
                  return false
                }
              @series.ReduceOp::Std
              | @series.ReduceOp::Var
              | @series.ReduceOp::Median
              | @series.ReduceOp::First
              | @series.ReduceOp::Last => return false
            }
          @ir.ExprNode::Cast(_, _)
          | @ir.ExprNode::Map(_, _, _)
          | @ir.ExprNode::MapBatches(_, _, _, _)
          | @ir.ExprNode::LitSeries(_) => return false
          @ir.ExprNode::Binary(_, _, _)
          | @ir.ExprNode::Unary(_, _)
          | @ir.ExprNode::Str(_, _)
          | @ir.ExprNode::Alias(_, _)
          | @ir.ExprNode::FillNull(_, _)
          | @ir.ExprNode::FillNan(_, _)
          | @ir.ExprNode::Ternary(_, _, _)
          | @ir.ExprNode::IsIn(_, _)
          | @ir.ExprNode::IsBetween(_, _, _, _) =>
            for child in e.children() {
              stack.push(child)
            }
        }
    }
  }
  true
}

///|
/// Whether the selection list contains `name` as a bare `col(name)` item.
fn has_bare_column(exprs : Array[@expr.Expr], name : String) -> Bool {
  for expr in exprs {
    match expr.node() {
      @ir.ExprNode::Col(n) => if n == name { return true }
      _ => ()
    }
  }
  false
}

///|
/// Whether the stage defines none of the predicate's columns — the
/// `WithColumns` crossing rule. A defined name is served by the stage
/// itself: below it, the name either does not exist or (a replacement)
/// holds the original values the stage was about to overwrite, so a
/// predicate reading one must stay above.
fn defines_none_of(exprs : Array[@expr.Expr], predicate : @expr.Expr) -> Bool {
  let refs = predicate.referenced_columns()
  for expr in exprs {
    if refs.contains(expr.output_name()) {
      return false
    }
  }
  true
}