having_semantics.hpp

Go to the documentation of this file.

/**
 * @file having_semantics.hpp
 * @brief Provenance evaluation helper for HAVING-clause circuits.
 *
 * When a query includes a HAVING clause, ProvSQL creates a special
 * sub-circuit that encodes the aggregate predicate.  Before the main
 * provenance circuit can be evaluated over a semiring, these HAVING
 * sub-circuits must first be evaluated to determine which groups
 * pass the filter.
 *
 * The single public entry point @c provsql_having() rewrites HAVING
 * comparison gates in the circuit, for any compiled semiring, by
 * enumerating possible worlds -- or, for @c MIN / @c MAX against a
 * constant, by the equivalent single-scan closed form: in every
 * absorptive semiring for the existential comparisons (@c MIN below,
 * @c MAX above a constant), and when ⊗ moreover distributes over ⊖ for
 * the others.  If the HAVING gate type is incompatible with the
 * requested semiring, the function is a no-op.
 */
#ifndef PROVSQL_HAVING_SEMANTICS_HPP
#define PROVSQL_HAVING_SEMANTICS_HPP
 
#include <algorithm>
#include <cstdint>
#include <cmath>
#include <functional>
#include <map>
#include <stdexcept>
#include <string>
#include <vector>
 
#include "GenericCircuit.hpp"
#include "provsql_utils.h"
#include "subset.hpp"
#include "Aggregation.h"
 
/** @cond INTERNAL */
namespace provsql_having_detail {
std::vector<gate_t> collect_sp_cmp_gates(GenericCircuit &c, gate_t start);
bool extract_constant_string(GenericCircuit &c, gate_t x, std::string &C_out);
bool semimod_extract_string_and_K(GenericCircuit &c, gate_t semimod_gate, std::string &m_out, gate_t &k_gate_out);
bool aggtype_is_text(unsigned oid);
bool aggtype_is_integer(unsigned oid);
bool aggtype_is_boolean(unsigned oid);
// True if @p oid is boolean, or an array whose element type is boolean.  Used
// by the array_agg comparison to reconcile PostgreSQL's two boolean text forms
// (scalar 'true'/'false' vs array-element 't'/'f').
bool aggtype_elem_is_boolean(unsigned oid);
bool aggtype_is_numeric(unsigned oid);
 
/**
 * @brief Map values of type @p typoid, given as text, and a threshold to
 *        their dense ranks under the type's comparison function.
 *
 * @return @c false when the type has no comparison function.
 */
bool rank_values_by_type(unsigned typoid, const std::vector<std::string> &vals,
                         const std::string &threshold,
                         std::vector<long> &ranks, long &threshold_rank);
/** @brief @c PROVSQL_GATE_NULL_UUID, the value gate of the NULL value. */
constexpr const char *GATE_NULL_UUID = PROVSQL_GATE_NULL_UUID;
 
/**
 * @brief The text standing for a NULL array element, on both sides of an
 *        @c array_agg comparison.
 *
 * It starts with a NUL byte, which no PostgreSQL text value contains, so it
 * equals itself and no actual element: exactly how array equality treats
 * NULL elements.
 */
inline const std::string &array_null_element() {
  static const std::string null_element("\0NULL", 5);
  return null_element;
}
 
bool parse_array_literal(const std::string &s, std::vector<std::string> &out);
bool parse_decimal_scaled(const std::string &s, long &mantissa, int &scale);
bool rescale_to(long mantissa, int scale, int target_scale, long &out);
ComparisonOperator map_cmp_op(GenericCircuit &c, gate_t cmp_gate, bool &ok);
ComparisonOperator flip_op(ComparisonOperator op);
}
/** @endcond */
 
/**
 * @brief Rewrite HAVING comparison gates in the circuit into their
 *        possible-world provenance (world enumeration, or the MIN / MAX
 *        single-scan closed form where the semiring allows it).
 *
 * @tparam SemiringT  The semiring type used for evaluation.
 * @tparam MapT       The provenance mapping type (gate_t → semiring value).
 * @param c        Circuit to rewrite.
 * @param g        Root gate of the sub-circuit to inspect.
 * @param mapping  Provenance mapping updated in place.
 * @param S        Semiring instance (default-constructed for stateless semirings).
 */
template <typename SemiringT, typename MapT>
void provsql_having(
  GenericCircuit &c,
  gate_t g,
  MapT &mapping,
  SemiringT S = SemiringT{})
{
  using namespace provsql_having_detail;
 
  std::vector<gate_t> cmp_gates = collect_sp_cmp_gates(c, g);
 
  if (cmp_gates.empty())
    return;
 
  // Whether the world enumeration over these contributor annotations can
  // be built *certified*: the semiring persists d-DNNF certificates
  // (circuit-building BoolExpr), every contributor is an independent
  // literal (base Bernoulli or constant) so world terms are decomposable,
  // no literal repeats (a repeat would couple two contributors), and the
  // contributor count is small enough that the complete-worlds
  // enumeration the certificate requires stays affordable.
  auto certifiable_contributors =
    [&](const std::vector<typename SemiringT::value_type> &kv) -> bool {
      if (!S.certifying())
        return false;
      if (kv.size() > 16)
        return false;
      for (size_t i = 0; i < kv.size(); ++i) {
        if (!S.independent_literal(kv[i]))
          return false;
        // Distinctness among non-constant literals (constants are
        // variable-free, so repeats of one() / zero() are harmless).
        if (kv[i] == S.one() || kv[i] == S.zero())
          continue;
        for (size_t j = 0; j < i; ++j)
          if (kv[j] == kv[i])
            return false;
      }
      return true;
    };
 
  // Combine a set of valid-world bitmasks (from an exhaustive enumeration) into
  // the m-semiring predicate-provenance.  Shared by every HAVING domain that
  // enumerates exact worlds (the numeric comparison, the boolean aggregates
  // and array_agg): a certifying semiring gets one certified world term per
  // world under a deterministic OR; otherwise each world contributes
  // ∏present ⊗ (1 ⊖ ⊕missing).  The monus factor is dropped when the
  // enumeration is an @p upset (an absorptive semiring pruned the family to
  // a generating subfamily of minimal worlds) or when the family is
  // @p monotone -- closed under supersets, i.e. the complete valid worlds of
  // a comparison preserved as the group grows -- and the semiring is
  // idempotent: over such a family the monus cancels from the possible-world
  // sum, ⊕_W ann_U(W) = ⊕_W ∏present (Lean Having.witness_identity, which
  // needs idempotence only; absorptivity further collapses the sum to the
  // minimal worlds, Having.witness_minimal, which is what the upset prune
  // exploits).
  auto combine_exhaustive_worlds =
    [&](const std::vector<mask_t> &worlds,
        const std::vector<typename SemiringT::value_type> &kvals,
        bool upset, bool monotone) -> typename SemiringT::value_type {
      if (worlds.empty())
        return S.zero();
 
      if (certifiable_contributors(kvals)) {
        std::vector<typename SemiringT::value_type> disjuncts;
        disjuncts.reserve(worlds.size());
        for (const auto &mask : worlds) {
          std::vector<typename SemiringT::value_type> present, missing;
          for (size_t i = 0; i < kvals.size(); ++i)
            (mask[i] ? present : missing).push_back(kvals[i]);
          disjuncts.push_back(S.certified_world_term(present, missing));
        }
        return S.certified_exclusive_plus(disjuncts);
      }
 
      std::vector<typename SemiringT::value_type> disjuncts;
      disjuncts.reserve(worlds.size());
      const size_t n = kvals.size();
      const bool no_monus = upset || (monotone && S.idempotent());
 
      for (const auto &mask : worlds) {
        std::vector<typename SemiringT::value_type> present, missing;
        present.reserve(n);
        missing.reserve(n);
 
        for (size_t i = 0; i < n; ++i) {
          if (mask[i]) {
            if (kvals[i] != S.one())
              present.push_back(kvals[i]);
          } else if (!no_monus) {
            if (kvals[i] != S.zero())
              missing.push_back(kvals[i]);
          }
        }
 
        auto present_prod = S.times(present);
 
        if (missing.empty()) {
          disjuncts.push_back(std::move(present_prod));
        } else {
          auto monus_factor = S.monus(S.one(), S.plus(missing));
          auto term = monus_factor;
          if (present_prod != S.one())
            term = S.times(std::vector<typename SemiringT::value_type>{
              present_prod, monus_factor});
          disjuncts.push_back(std::move(term));
        }
      }
 
      return S.plus(disjuncts);
    };
 
  // Single-scan closed form for MIN / MAX against a constant in an
  // absorptive m-semiring.  Writing L, L', G, G', E for the ⊕-sums of
  // the contributors whose value is <, <=, >=, >, = C, the possible-world
  // provenance of MIN(a) op C is
  //   MIN <  C : L               MIN >= C : (1 ⊖ L)  ⊗ G
  //   MIN <= C : L'              MIN >  C : (1 ⊖ L') ⊗ G'
  //   MIN =  C : (1 ⊖ L) ⊗ E     MIN <> C : L ⊕ (1 ⊖ L') ⊗ G'
  // and MAX is the mirror image (exchange < and >).  The existential
  // comparisons (MIN <, <=; MAX >, >=) hold in the worlds meeting the set
  // of witnesses, an upward-closed family whose minimal elements are the
  // singletons: the monus cancels from the possible-world sum and
  // absorptivity alone makes the ⊕-sum over the witnesses exact (Lean
  // Having.sum_ann_meet, HavingMonotone.minLe_site_rewrite and its three
  // siblings).  The other comparisons carry the 1 ⊖ factor excluding
  // every world containing a spoiler, and their closed form is proved
  // only when ⊗ moreover distributes over ⊖, which combines that factor
  // with the witnesses by one ⊗ (Lean Having.minScan_correct /
  // Having.maxScan_correct); without distributivity they are not
  // proved, and not taken.  The empty world never satisfies the
  // comparison (MIN / MAX of an empty group is NULL), so scalar
  // aggregation needs no special case.
  auto min_max_scan =
    [&](const std::vector<long> &mvals, long C,
        const std::vector<typename SemiringT::value_type> &kvals,
        ComparisonOperator op,
        AggregationOperator agg_kind) -> typename SemiringT::value_type {
      using V = typename SemiringT::value_type;
      std::vector<V> lt, le, ge, gt, eq;
      for (size_t i = 0; i < kvals.size(); ++i) {
        if (kvals[i] == S.zero()) continue;
        if (mvals[i] < C) lt.push_back(kvals[i]);
        if (mvals[i] <= C) le.push_back(kvals[i]);
        if (mvals[i] >= C) ge.push_back(kvals[i]);
        if (mvals[i] > C) gt.push_back(kvals[i]);
        if (mvals[i] == C) eq.push_back(kvals[i]);
      }
      auto sum = [&](const std::vector<V> &v) -> V {
        return v.empty() ? S.zero() : S.plus(v);
      };
      // (1 ⊖ ⊕spoilers) ⊗ ⊕witnesses
      auto guarded = [&](const std::vector<V> &spoilers,
                         const std::vector<V> &witnesses) -> V {
        if (witnesses.empty()) return S.zero();
        V w = S.plus(witnesses);
        if (spoilers.empty()) return w;
        return S.times(std::vector<V>{S.monus(S.one(), S.plus(spoilers)), w});
      };
 
      // The MAX table is the MIN table with < and > exchanged: flip the
      // operator and swap the roles of the below / above sets.
      const bool is_max = (agg_kind == AggregationOperator::MAX);
      const std::vector<V> &below = is_max ? gt : lt;       // strictly on the "losing" side
      const std::vector<V> &below_eq = is_max ? ge : le;    // losing side or equal
      const std::vector<V> &above = is_max ? lt : gt;       // strictly on the "winning" side
      const std::vector<V> &above_eq = is_max ? le : ge;    // winning side or equal
      if (is_max) op = flip_op(op);
 
      switch (op) {
      case ComparisonOperator::LT: return sum(below);
      case ComparisonOperator::LE: return sum(below_eq);
      case ComparisonOperator::GE: return guarded(below, above_eq);
      case ComparisonOperator::GT: return guarded(below_eq, above);
      case ComparisonOperator::EQ: return guarded(below, eq);
      case ComparisonOperator::NE: {
        V a = sum(below);
        V b = guarded(below_eq, above);
        if (a == S.zero()) return b;
        if (b == S.zero()) return a;
        return S.plus(std::vector<V>{a, b});
      }
      }
      return S.zero();
    };
 
  // Provenance of a predicate decided by the *first present* occurrence:
  // the valid worlds partition into fibers, one per candidate first
  // occurrence i, each fixing i present, a set A_i of occurrences absent
  // (for choose(): every occurrence before i; for bool_or / bool_and: the
  // earlier candidates and the class that must be wholly absent) and
  // leaving the rest free.  In an absorptive m-semiring whose ⊗
  // distributes over ⊖ the free suffix sums to 𝟙 and each fiber is
  // (1 ⊖ ⊕_{A_i} k) ⊗ k_i (Lean Having.firstScan_correct; the same
  // hypotheses as the MIN / MAX scan).  Elsewhere -- Why, Which, How,
  // Formula, counting, the security semiring -- the free part does not
  // collapse, so every world of every fiber is enumerated and annotated
  // ∏present ⊗ (1 ⊖ ⊕missing), exponential in the number of free
  // occurrences; @p monotone says the union of the fibers is closed under
  // supersets (no class has to be wholly absent), which lets an
  // idempotent semiring drop the monus factors.
  using fiber_t = std::pair<size_t, std::vector<size_t>>;   // (present, absent set)
  auto first_present_provenance =
    [&](const std::vector<fiber_t> &fibers,
        const std::vector<typename SemiringT::value_type> &kvals,
        bool monotone) -> typename SemiringT::value_type {
      using V = typename SemiringT::value_type;
      if (fibers.empty()) return S.zero();
 
      if (S.absorptive() && S.mul_sub_left_distributive()) {
        std::vector<V> terms;
        for (const auto &fb : fibers) {
          const V &k = kvals[fb.first];
          if (k == S.zero()) continue;
          std::vector<V> absent;
          for (size_t j : fb.second)
            if (kvals[j] != S.zero()) absent.push_back(kvals[j]);
          if (absent.empty()) { terms.push_back(k); continue; }
          V guard = S.monus(S.one(), S.plus(absent));
          terms.push_back(k == S.one() ? guard : S.times(std::vector<V>{guard, k}));
        }
        return terms.empty() ? S.zero() : S.plus(terms);
      }
 
      const size_t n = kvals.size();
      std::vector<mask_t> worlds;
      for (const auto &fb : fibers) {
        std::vector<bool> fixed(n, false);
        fixed[fb.first] = true;
        for (size_t j : fb.second) fixed[j] = true;
        std::vector<size_t> free_idx;
        for (size_t j = 0; j < n; ++j)
          if (!fixed[j]) free_idx.push_back(j);
        const size_t m = free_idx.size();
        for (size_t sub = 0; sub < (size_t(1) << m); ++sub) {
          mask_t mask(n, false);
          mask[fb.first] = true;
          for (size_t b = 0; b < m; ++b)
            if (sub & (size_t(1) << b)) mask[free_idx[b]] = true;
          worlds.push_back(std::move(mask));
        }
      }
      return combine_exhaustive_worlds(worlds, kvals, /*upset=*/false, monotone);
    };
 
  auto pw_from_cmp_gate = [&](gate_t cmp_gate, typename SemiringT::value_type &pw_out) -> bool {
    const auto &cw = c.getWires(cmp_gate);
    if (cw.size() != 2) return false;
 
    gate_t L = cw[0];
    gate_t R = cw[1];
 
    bool okop = false;
    ComparisonOperator op = map_cmp_op(c, cmp_gate, okop);
    if (!okop) return false;
 
    auto build_from = [&](gate_t agg_side, gate_t const_side, ComparisonOperator effective_op) -> bool {
      if (c.getGateType(agg_side) != gate_agg) return false;
 
      // info2 of the gate_agg is the aggregate's result type -- the
      // comparison domain (int / numeric / float / text) -- in its low 31 bits;
      // the high bit is the scalar-aggregation flag, masked off here.
      const unsigned aggtype = c.getInfos(agg_side).second & PROVSQL_AGG_TYPE_MASK;
      const bool is_scalar =
        (c.getInfos(agg_side).second & PROVSQL_AGG_SCALAR_FLAG) != 0;
      AggregationOperator agg_kind = getAggregationOperator(c.getInfos(agg_side).first);
      const auto &children = c.getWires(agg_side);
 
      // ---- Value-as-text domain: choose() over a non-numeric scalar constant.
      //      Covers text and any other type whose values round-trip through
      //      their text representation (bool, date, uuid, enum, ...); numeric
      //      choose() goes to the numeric domain below (which also supports the
      //      ordering comparisons).  Only = / <> are exposed here. ----
      if ((aggtype_is_text(aggtype) &&
           agg_kind != AggregationOperator::MIN &&
           agg_kind != AggregationOperator::MAX) ||
          (agg_kind == AggregationOperator::CHOOSE &&
           !aggtype_is_numeric(aggtype))) {
        std::string C_str;
        if (!extract_constant_string(c, const_side, C_str)) return false;
 
        std::vector<std::string> mvals_str;
        std::vector<typename SemiringT::value_type> kvals;
        mvals_str.reserve(children.size());
        kvals.reserve(children.size());
        for (gate_t ch : children) {
          if (c.getGateType(ch) != gate_semimod) return false;
          std::string m_str;
          gate_t k_gate{};
          if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
          mvals_str.push_back(m_str);
          kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
        }
 
        // Only choose() is supported (the only text-valued aggregate whose
        // possible-world value is decided occurrence-by-occurrence), and
        // only = / <> are exposed for text.
        if (agg_kind != AggregationOperator::CHOOSE)
          throw std::runtime_error(
            "comparing an aggregate with a text constant in HAVING "
            "is only implemented for choose()");
        if (effective_op != ComparisonOperator::EQ &&
            effective_op != ComparisonOperator::NE)
          throw std::runtime_error(
            "only = and <> are supported when comparing choose() "
            "with a text constant");
 
        // choose() is PICKFIRST: in a world W its value is that of the
        // lowest-index present element.  So choose(W) op C holds iff the
        // first present element matches: one fiber per matching index i,
        // with every occurrence before i absent and the suffix free.
        //
        // The fibers are mutually exclusive by construction (they differ
        // on the first present index): when the semiring certifies
        // enumerations, build them as certified world terms under a
        // certified deterministic OR instead.
        if (certifiable_contributors(kvals)) {
          std::vector<typename SemiringT::value_type> disjuncts;
          std::vector<typename SemiringT::value_type> before;
          for (size_t i = 0; i < kvals.size(); ++i) {
            bool match = (effective_op == ComparisonOperator::EQ)
                           ? (mvals_str[i] == C_str)
                           : (mvals_str[i] != C_str);
            if (match)
              disjuncts.push_back(S.certified_world_term(
                std::vector<typename SemiringT::value_type>{kvals[i]},
                before));
            before.push_back(kvals[i]);
          }
          pw_out = disjuncts.empty() ? S.zero()
                   : S.certified_exclusive_plus(disjuncts);
          return true;
        }
        std::vector<fiber_t> fibers;
        std::vector<size_t> before;
        for (size_t i = 0; i < kvals.size(); ++i) {
          bool match = (effective_op == ComparisonOperator::EQ)
                         ? (mvals_str[i] == C_str)
                         : (mvals_str[i] != C_str);
          if (match) fibers.emplace_back(i, before);
          before.push_back(i);
        }
        pw_out = first_present_provenance(fibers, kvals, /*monotone=*/false);
        return true;
      }
 
      // ---- Boolean comparison domain: bool_or / bool_and (and the every
      //      alias for bool_and) compared with a boolean constant.  A boolean
      //      aggregate has only two possible values, so the worlds yielding the
      //      wanted value are characterised directly in the m-semiring rather
      //      than by a 2^n enumeration.  The rows partition by their value into
      //      a class that must have at least one present member ("someE") and a
      //      class that must be wholly absent ("noneF"):
      //        bool_or  = true  : someE = true-rows                (false free)
      //        bool_or  = false : someE = false-rows, noneF = true-rows
      //        bool_and = true  : someE = true-rows,  noneF = false-rows
      //        bool_and = false : someE = false-rows               (true free)
      //      "at least one of someE present" splits by the first present
      //      someE index (the choose pattern), each fiber also requiring
      //      every noneF row absent; see first_present_provenance for when
      //      that is a closed form and when it is enumerated.  Non-empty
      //      groups are enforced by someE being non-empty. ----
      if (aggtype_is_boolean(aggtype)) {
        if (agg_kind != AggregationOperator::OR &&
            agg_kind != AggregationOperator::AND)
          return false;
        if (effective_op != ComparisonOperator::EQ &&
            effective_op != ComparisonOperator::NE)
          throw std::runtime_error(
            "only = and <> are supported when comparing a boolean aggregate "
            "(bool_or / bool_and / every) with a constant in HAVING");
 
        auto parse_bool = [](const std::string &s, bool &ok) -> bool {
          ok = true;
          if (s == "t" || s == "true"  || s == "1") return true;
          if (s == "f" || s == "false" || s == "0") return false;
          ok = false; return false;
        };
 
        std::string C_str;
        if (!extract_constant_string(c, const_side, C_str)) return false;
        bool okc = false;
        const bool C = parse_bool(C_str, okc);
        if (!okc) return false;
        // The aggregate value the satisfying worlds must produce.
        const bool target = (effective_op == ComparisonOperator::EQ) ? C : !C;
 
        std::vector<bool> vals;
        std::vector<typename SemiringT::value_type> kvals;
        vals.reserve(children.size());
        kvals.reserve(children.size());
        for (gate_t ch : children) {
          if (c.getGateType(ch) != gate_semimod) return false;
          std::string m_str;
          gate_t k_gate{};
          if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
          bool okv = false;
          const bool b = parse_bool(m_str, okv);
          if (!okv) return false;
          vals.push_back(b);
          kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
        }
 
        const bool want_or = (agg_kind == AggregationOperator::OR);
        std::vector<size_t> someE;   // at least one of these must be present
        std::vector<size_t> noneF;   // all of these must be absent
        if (want_or == target) {
          // bool_or=true / bool_and=false: one trigger row suffices, the other
          // class is free.
          for (size_t i = 0; i < vals.size(); ++i)
            if (vals[i] == want_or) someE.push_back(i);
        } else {
          // bool_or=false / bool_and=true: the trigger class must be wholly
          // absent and the opposite class must have at least one present row.
          for (size_t i = 0; i < vals.size(); ++i)
            (vals[i] == want_or ? noneF : someE).push_back(i);
        }
 
        if (someE.empty()) { pw_out = S.zero(); return true; }
 
        if (certifiable_contributors(kvals)) {
          std::vector<typename SemiringT::value_type> disjuncts;
          std::vector<typename SemiringT::value_type> before; // someE rows before i
          for (size_t e : someE) {
            std::vector<typename SemiringT::value_type> missing = before;
            for (size_t f : noneF) missing.push_back(kvals[f]);
            disjuncts.push_back(S.certified_world_term(
              std::vector<typename SemiringT::value_type>{kvals[e]}, missing));
            before.push_back(kvals[e]);
          }
          pw_out = S.certified_exclusive_plus(disjuncts);
          return true;
        }
 
        // With no class to exclude (bool_or = true, bool_and = false) the
        // predicate is existential: it holds in the worlds meeting someE,
        // an upward-closed family whose minimal elements are the
        // singletons, so in an absorptive semiring the ⊕-sum of the
        // someE annotations is exact, with no monus and no distributivity
        // (Lean Having.sum_ann_meet, the MIN <= C argument).  An
        // idempotent semiring still enumerates the family but, it being
        // monotone, without the monus factors.
        if (noneF.empty() && S.absorptive()) {
          std::vector<typename SemiringT::value_type> witnesses;
          for (size_t e : someE)
            if (kvals[e] != S.zero()) witnesses.push_back(kvals[e]);
          pw_out = witnesses.empty() ? S.zero() : S.plus(witnesses);
          return true;
        }
 
        std::vector<fiber_t> fibers;
        std::vector<size_t> absent = noneF;    // grows with the earlier someE rows
        for (size_t e : someE) {
          fibers.emplace_back(e, absent);
          absent.push_back(e);
        }
        pw_out = first_present_provenance(fibers, kvals,
                                          /*monotone=*/noneF.empty());
        return true;
      }
 
      // ---- Array comparison domain: array_agg(x) against a constant array.
      //      No aggregate-specific optimization (the general pipeline): scan the
      //      non-empty worlds whose ordered present elements equal (=) or differ
      //      (<>) from the constant array, then combine those worlds in the
      //      m-semiring.  Elements are compared as their text representations,
      //      so any element type works. ----
      if (agg_kind == AggregationOperator::ARRAY_AGG) {
        if (effective_op != ComparisonOperator::EQ &&
            effective_op != ComparisonOperator::NE)
          throw std::runtime_error(
            "only = and <> are supported when comparing array_agg() with a "
            "constant array in HAVING");
 
        std::string C_str;
        if (!extract_constant_string(c, const_side, C_str)) return false;
        std::vector<std::string> target;
        if (!parse_array_literal(C_str, target)) return false;
 
        std::vector<std::string> vals;
        std::vector<typename SemiringT::value_type> kvals;
        vals.reserve(children.size());
        kvals.reserve(children.size());
        for (gate_t ch : children) {
          if (c.getGateType(ch) != gate_semimod) return false;
          std::string m_str;
          gate_t k_gate{};
          if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
          // A NULL input of the aggregate: its value gate is gate_null().
          if (c.getUUID(c.getWires(ch)[1]) == GATE_NULL_UUID)
            m_str = array_null_element();
          vals.push_back(m_str);
          kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
        }
 
        // Boolean elements: the row values carry the scalar bool text
        // ('true'/'false') while the constant array's elements come back in
        // PostgreSQL's array form ('t'/'f').  Canonicalise both sides so the
        // text comparison below sees the same representation.
        if (aggtype_elem_is_boolean(aggtype)) {
          auto canon_bool = [](std::string &s) {
            if (s == "t" || s == "true"  || s == "1") s = "true";
            else if (s == "f" || s == "false" || s == "0") s = "false";
          };
          for (auto &e : target) canon_bool(e);
          for (auto &v : vals)   canon_bool(v);
        }
 
        auto worlds = enumerate_array_agg_worlds(
          vals, target, effective_op == ComparisonOperator::EQ);
        pw_out = combine_exhaustive_worlds(worlds, kvals, /*upset=*/false,
                                           /*monotone=*/false);
        return true;
      }
 
      // The possible worlds of an aggregate over integer-mapped values
      // compared with an integer threshold: the closed forms, or the
      // enumeration.
      auto finish_domain = [&](const std::vector<long> &mvals, long C,
                               const std::vector<typename SemiringT::value_type> &kvals)
        -> bool {
        // A certified enumeration needs the *complete* valid worlds (every
        // contributor present or explicitly negated): the upset shortcut
        // and the monotone MIN/MAX skips below produce overlapping,
        // non-exclusive disjuncts, sound for absorptive evaluation but
        // unmarkable.  When certifying, request the full enumeration --
        // the same one non-absorptive semirings already use.
        const bool certify = certifiable_contributors(kvals);
 
        // MIN / MAX in an absorptive semiring: the single-scan closed form
        // replaces the 2^n world enumeration -- for the existential
        // comparisons (MIN below, MAX above the constant) in every
        // absorptive semiring, for the others only when ⊗ distributes over
        // ⊖ (see min_max_scan).  A certifying semiring keeps the
        // enumeration, whose exclusive world terms its certificate needs.
        if ((agg_kind == AggregationOperator::MIN ||
             agg_kind == AggregationOperator::MAX) &&
            S.absorptive() && !certify) {
          const bool existential =
            (agg_kind == AggregationOperator::MIN)
              ? (effective_op == ComparisonOperator::LT ||
                 effective_op == ComparisonOperator::LE)
              : (effective_op == ComparisonOperator::GT ||
                 effective_op == ComparisonOperator::GE);
          if (existential || S.mul_sub_left_distributive()) {
            pw_out = min_max_scan(mvals, C, kvals, effective_op, agg_kind);
            return true;
          }
        }
 
        bool upset = false;
        auto worlds = enumerate_valid_worlds(mvals, C, effective_op, agg_kind,
                                             certify ? false : S.absorptive(),
                                             upset, is_scalar);
 
        // Comparisons preserved as the group grows, whose valid worlds are
        // therefore closed under supersets: MIN below and MAX above the
        // constant, COUNT above it (count(col) carries 0/1 values), SUM
        // above it when no contributor is negative.
        const bool above = (effective_op == ComparisonOperator::GT ||
                            effective_op == ComparisonOperator::GE);
        const bool below = (effective_op == ComparisonOperator::LT ||
                            effective_op == ComparisonOperator::LE);
        bool monotone = false;
        switch (agg_kind) {
        case AggregationOperator::MIN:   monotone = below; break;
        case AggregationOperator::MAX:   monotone = above; break;
        case AggregationOperator::COUNT: monotone = above; break;
        case AggregationOperator::SUM:
          monotone = above &&
            std::all_of(mvals.begin(), mvals.end(), [](long v) { return v >= 0; });
          break;
        default: break;
        }
 
        pw_out = combine_exhaustive_worlds(worlds, kvals, upset, monotone);
        return true;
      };
 
      // ---- Ordered domain: MIN / MAX over a type that is not numeric (text,
      //      date, timestamp, ...).  Their value in a world only depends on
      //      the order of the values, so values and threshold are mapped to
      //      their ranks under the type's own comparison function (with its
      //      default collation, for text), and the integer machinery below
      //      applies. ----
      if (!aggtype_is_numeric(aggtype) &&
          (agg_kind == AggregationOperator::MIN ||
           agg_kind == AggregationOperator::MAX)) {
        std::string C_str;
        if (!extract_constant_string(c, const_side, C_str)) return false;
        std::vector<std::string> m_strs;
        std::vector<typename SemiringT::value_type> kvals;
        for (gate_t ch : children) {
          if (c.getGateType(ch) != gate_semimod) return false;
          std::string m_str;
          gate_t k_gate{};
          if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
          m_strs.push_back(m_str);
          kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
        }
        std::vector<long> ranks;
        long C_rank = 0;
        if (!rank_values_by_type(aggtype, m_strs, C_str, ranks, C_rank))
          return false;
        return finish_domain(ranks, C_rank, kvals);
      }
 
      // ---- Numeric comparison domain (int / numeric / float): unify by
      //      scaling every value and the threshold to a common integer
      //      grid by their decimal text, so a numeric(p,d) / finite-decimal
      //      float column is evaluated exactly and fractional thresholds
      //      work.  Integer is the scale-0 case. ----
      std::string C_str;
      if (!extract_constant_string(c, const_side, C_str)) return false;
      long C_mant = 0; int C_scale = 0;
      if (!parse_decimal_scaled(C_str, C_mant, C_scale)) return false;
 
      std::vector<long> m_mant;
      std::vector<int> m_scale;
      std::vector<typename SemiringT::value_type> kvals;
      m_mant.reserve(children.size());
      m_scale.reserve(children.size());
      kvals.reserve(children.size());
      for (gate_t ch : children) {
        if (c.getGateType(ch) != gate_semimod) return false;
        std::string m_str;
        gate_t k_gate{};
        if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
        long mm = 0; int ms = 0;
        if (!parse_decimal_scaled(m_str, mm, ms)) return false;
        m_mant.push_back(mm);
        m_scale.push_back(ms);
        kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
      }
 
      // Common scale: rescale every value and the threshold to integers.
      int target_scale = C_scale;
      for (int ms : m_scale) target_scale = std::max(target_scale, ms);
      long C = 0;
      if (!rescale_to(C_mant, C_scale, target_scale, C)) return false;
      std::vector<long> mvals(m_mant.size());
      for (size_t i = 0; i < m_mant.size(); ++i)
        if (!rescale_to(m_mant[i], m_scale[i], target_scale, mvals[i])) return false;
 
      return finish_domain(mvals, C, kvals);
    };
 
    // General possible-worlds evaluation for a comparison over an *arithmetic
    // expression of one or more aggregates* (sum(x)*sum(y) > k, sum(x) > sum(y),
    // 100/sum(x) > 5, ...), which the single-aggregate path above cannot fold.
    // We enumerate the joint possible worlds over the union of the aggregates'
    // contributors, evaluate the arithmetic numerically in each world, and
    // combine the worlds where the comparison holds in the semiring exactly as
    // the single-aggregate path does (present_prod ⊗ (1 ⊖ missing_sum)).  This
    // is exponential in the number of distinct contributors, so it bails out
    // (returning false, leaving the gate unresolved) beyond a small bound.
    auto build_general = [&](gate_t Lx, gate_t Rx,
                             ComparisonOperator opx) -> bool {
      struct AggInfo {
        AggregationOperator kind;
        bool is_int;                                    // integer-typed result?
        bool is_scalar;                                 // aggregation without GROUP BY?
        std::vector<std::pair<int, double> > contribs;  // (distinct-K index, value)
      };
      // A decimal text denotes an integer iff it has no fractional/exponent part.
      auto text_is_int = [](const std::string &s) -> bool {
        if (s.empty()) return false;
        for (char ch : s)
          if (!((ch >= '0' && ch <= '9') || ch == '+' || ch == '-'))
            return false;
        return true;
      };
      std::map<gate_t, AggInfo> aggs;
      std::map<gate_t, int> kindex;
      std::vector<gate_t> kgates;
 
      // A number, the whole text of it: a date such as 2022-08-05 is not
      // 2022.
      auto parse_number = [](const std::string &s, double &out) -> bool {
        try {
          std::size_t pos = 0;
          out = std::stod(s, &pos);
          return pos == s.size();
        } catch (...) {
          return false;
        }
      };
 
      std::function<bool(gate_t)> collect = [&](gate_t gx) -> bool {
        gate_type gt = c.getGateType(gx);
        if (gt == gate_agg) {
          if (aggs.count(gx))
            return true;
          AggInfo ai;
          ai.kind = getAggregationOperator(c.getInfos(gx).first);
          ai.is_int = aggtype_is_integer(c.getInfos(gx).second & PROVSQL_AGG_TYPE_MASK);
          ai.is_scalar =
            (c.getInfos(gx).second & PROVSQL_AGG_SCALAR_FLAG) != 0;
          // Only the kinds eval reads a value of: any other declines the
          // route here, rather than reading there as "no value" in every
          // world -- which is a comparison that never holds, a probability of
          // zero with nothing to say so.  bool_and / bool_or / array_agg were
          // kept out only by their values failing to parse as numbers.
          switch (ai.kind) {
          case AggregationOperator::SUM:
          case AggregationOperator::COUNT:
          case AggregationOperator::AVG:
          case AggregationOperator::MIN:
          case AggregationOperator::MAX:
          case AggregationOperator::CHOOSE:
            break;
          default:
            return false;
          }
          for (gate_t ch : c.getWires(gx)) {
            if (c.getGateType(ch) != gate_semimod)
              return false;
            std::string ms;
            gate_t kg{};
            if (!semimod_extract_string_and_K(c, ch, ms, kg))
              return false;
            int idx;
            auto it = kindex.find(kg);
            if (it == kindex.end()) {
              idx = static_cast<int>(kgates.size());
              kindex[kg] = idx;
              kgates.push_back(kg);
            } else
              idx = it->second;
            double m;
            if (!parse_number(ms, m)) return false;
            ai.contribs.emplace_back(idx, m);
          }
          aggs.emplace(gx, std::move(ai));
          return true;
        }
        if (gt == gate_arith) {
          for (gate_t ch : c.getWires(gx))
            if (!collect(ch))
              return false;
          return true;
        }
        if (gt == gate_value)
          return true;
        if (gt == gate_semimod) {       // a constant threshold: semimod(1, value)
          std::string ms;
          gate_t kg{};
          if (!semimod_extract_string_and_K(c, gx, ms, kg))
            return false;
          return c.getGateType(kg) == gate_one;
        }
        return false;
      };
 
      if (!collect(Lx) || !collect(Rx))
        return false;
      if (aggs.empty())
        return false;
      const size_t n = kgates.size();
      if (n > 24)                      // 2^n enumeration: keep it bounded
        return false;
 
      // Numeric value of a subexpression in a given world, tracking whether it
      // is integer-valued so that division floors as SQL does (NULL -> false).
      std::function<bool(gate_t, uint64_t, double &, bool &)> eval =
        [&](gate_t gx, uint64_t world, double &out, bool &is_int) -> bool {
        gate_type gt = c.getGateType(gx);
        if (gt == gate_value) {
          std::string s = c.getExtra(gx);
          if (!parse_number(s, out)) return false;
          is_int = text_is_int(s);
          return true;
        }
        if (gt == gate_semimod) {       // constant threshold
          std::string ms; gate_t kg{};
          if (!semimod_extract_string_and_K(c, gx, ms, kg)) return false;
          if (!parse_number(ms, out)) return false;
          is_int = text_is_int(ms);
          return true;
        }
        if (gt == gate_agg) {
          const AggInfo &ai = aggs.at(gx);
          double acc = 0, mn = 0, mx = 0, fst = 0;
          long cnt = 0;
          bool first = true;
          for (const auto &pr : ai.contribs)
            if (world & (uint64_t(1) << pr.first)) {
              double m = pr.second;
              acc += m; ++cnt;
              if (first) { mn = mx = fst = m; first = false; }
              else { mn = std::min(mn, m); mx = std::max(mx, m); }
            }
          is_int = ai.is_int;
          // An aggregate with no surviving contributor in this world declines
          // the world (returning false), which the caller reads as "the
          // comparison does not hold here".  Two situations produce cnt == 0
          // and both want that answer: the group has no row present, so the
          // group's row does not exist; or the group is non-empty but every
          // contributed value was NULL, so the aggregate is SQL NULL and the
          // comparison is NULL, hence false.  (A NULL-valued row never reaches
          // contribs -- the aggregation rewriting drops it -- which is why the
          // two are indistinguishable here, and why they need not be.)
          //
          // COUNT is the exception: an empty set genuinely counts 0, so the
          // world has a value whenever a row exists to carry it -- which is
          // the case for a scalar aggregation, and not for a grouped one whose
          // empty group is no row at all.  The gate records COUNT for both
          // count(*) and count(expr) (see make_aggregation_expression), so no
          // guessing from the values is needed here.
          switch (ai.kind) {
          case AggregationOperator::SUM:   if (cnt == 0) return false; out = acc; return true;
          case AggregationOperator::COUNT:                       // values 1 or 0/1
            if (cnt == 0 && !ai.is_scalar) return false;
            out = acc; return true;
          case AggregationOperator::AVG:   if (cnt == 0) return false; out = acc / cnt; return true;
          case AggregationOperator::MIN:   if (cnt == 0) return false; out = mn; return true;
          case AggregationOperator::MAX:   if (cnt == 0) return false; out = mx; return true;
          // choose() is PICKFIRST: the value of the first present contributor,
          // in the order of the gate's wires (see ChooseAgg).
          case AggregationOperator::CHOOSE: if (cnt == 0) return false; out = fst; return true;
          default:
            // collect admits no other kind: one reaching here would read as
            // "no value" in every world, so it raises instead.
            throw CircuitException(
              "having_semantics: aggregate kind not read per world");
          }
        }
        if (gt == gate_arith) {
          const auto &w = c.getWires(gx);
          unsigned aop = static_cast<unsigned>(c.getInfos(gx).first);
          if (aop == PROVSQL_ARITH_PLUS || aop == PROVSQL_ARITH_TIMES) {
            double r = (aop == PROVSQL_ARITH_PLUS) ? 0 : 1;
            bool all_int = true;
            for (gate_t ch : w) {
              double v; bool vi;
              if (!eval(ch, world, v, vi)) return false;
              if (aop == PROVSQL_ARITH_PLUS) r += v; else r *= v;
              all_int = all_int && vi;
            }
            out = r; is_int = all_int; return true;
          }
          if (aop == PROVSQL_ARITH_MINUS) {
            if (w.size() != 2) return false;
            double a, b; bool ai, bi;
            if (!eval(w[0], world, a, ai) || !eval(w[1], world, b, bi)) return false;
            out = a - b; is_int = ai && bi; return true;
          }
          if (aop == PROVSQL_ARITH_DIV || aop == PROVSQL_ARITH_INTDIV) {
            if (w.size() != 2) return false;
            double a, b; bool ai, bi;
            if (!eval(w[0], world, a, ai) || !eval(w[1], world, b, bi)) return false;
            if (b == 0) return false;
            if (aop == PROVSQL_ARITH_INTDIV) {  // SQL division of integers
              out = std::trunc(a / b);
              is_int = true;
            } else {
              /* A real division, whatever the operands happen to be in this
               * world.  Which of the two divisions SQL means is the gate's to
               * say -- the rewriting emits INTDIV exactly where the
               * expression's type is an integer one -- and deciding it again
               * from the values truncated a division the query wrote over
               * numbers: count(*) OVER w / count(*) OVER w0 came out 1 or 0
               * per world, so a comparison over it answered the probability
               * that the two counts are equal.  The other five evaluators
               * (Monte Carlo, RangeCheck, Expectation, the simplifier, the
               * collapsed moment) all read the op and not the values. */
              out = a / b;
              is_int = false;
            }
            return true;
          }
          if (aop == PROVSQL_ARITH_NEG) {
            if (w.size() != 1) return false;
            double a; bool ai;
            if (!eval(w[0], world, a, ai)) return false;
            out = -a; is_int = ai; return true;
          }
          if (aop == PROVSQL_ARITH_ASFLOAT8 ||
              aop == PROVSQL_ARITH_ASFLOAT4) {
            /* The value read in its own type: a double is what this carries,
             * and a real rounds it to seven digits. */
            double a;
            bool ai;
            if (w.size() != 1 || !eval(w[0], world, a, ai)) return false;
            out = aop == PROVSQL_ARITH_ASFLOAT8
                    ? a : static_cast<double>(static_cast<float>(a));
            is_int = ai;
            return true;
          }
          if (aop == PROVSQL_ARITH_ROUND || aop == PROVSQL_ARITH_FLOOR ||
              aop == PROVSQL_ARITH_CEIL || aop == PROVSQL_ARITH_ABS) {
            /* Read in the world, on the value the world gives: rounding and
             * absolute value of a value, never of an expectation. */
            double a;
            bool ai;
            if (w.empty() || !eval(w[0], world, a, ai)) return false;
            if (aop == PROVSQL_ARITH_FLOOR) out = std::floor(a);
            else if (aop == PROVSQL_ARITH_CEIL) out = std::ceil(a);
            else if (aop == PROVSQL_ARITH_ABS) out = std::fabs(a);
            else if (w.size() == 1) out = std::round(a);
            else {
              double d;
              bool di;
              if (!eval(w[1], world, d, di)) return false;
              const double f = std::pow(10.0, d);
              out = std::round(a * f) / f;
            }
            /* An integer in, an integer out, except a rounding to digits of a
             * value that was not one. */
            is_int = ai || aop != PROVSQL_ARITH_ROUND || w.size() == 1;
            return true;
          }
          if (aop == PROVSQL_ARITH_POW || aop == PROVSQL_ARITH_LN ||
              aop == PROVSQL_ARITH_EXP) {
            /* The transform of the value this world gives.  Without these
             * three the switch fell through to "no value in this world", which
             * the caller reads as "the comparison does not hold here" -- in
             * every world, so a HAVING over sqrt(sum(x)) (a POW of one half),
             * over ln or exp, or over an expression holding one, answered
             * probability 0 whatever the threshold.  The other five evaluators
             * read them, or decline the shape and leave it to this one.
             *
             * Where the transform is undefined ON the value the world gives --
             * the logarithm of a nonpositive number, a negative base raised to
             * a fraction -- the world has no value, the reading a divisor of
             * zero gets (see the DIV arm) rather than the error SQL raises
             * there: raising in one world would take every other world's
             * answer with it. */
            double a;
            bool ai;
 
            if (w.empty() || !eval(w[0], world, a, ai)) return false;
            if (aop == PROVSQL_ARITH_LN) {
              if (!(a > 0)) return false;
              out = std::log(a);
            } else if (aop == PROVSQL_ARITH_EXP) {
              out = std::exp(a);
            } else {
              double e;
              bool ei;
 
              if (w.size() != 2 || !eval(w[1], world, e, ei)) return false;
              out = std::pow(a, e);
              if (std::isnan(out)) return false;
            }
            if (!std::isfinite(out)) return false;
            is_int = false;
            return true;
          }
          if (aop == PROVSQL_ARITH_MAX || aop == PROVSQL_ARITH_MIN) {
            if (w.empty()) return false;
            double r = 0; bool all_int = true, first = true;
            for (gate_t ch : w) {
              double v; bool vi;
              if (!eval(ch, world, v, vi)) return false;
              if (first) { r = v; first = false; }
              else r = (aop == PROVSQL_ARITH_MAX) ? std::max(r, v)
                                                  : std::min(r, v);
              all_int = all_int && vi;
            }
            out = r; is_int = all_int; return true;
          }
          if (aop == PROVSQL_ARITH_PERCENTILE) {
            /* A continuous percentile over the group's rows: the wires are
             * interleaved [ind_1, x_1, ...] with ind_i the row's 0/1 presence,
             * and the fraction is the gate's own text.  The values the world
             * holds are sorted and interpolated at the fraction, as
             * MonteCarloSampler reads them; a world holding none of them has no
             * value, like an aggregate with no contributing row.  No comparison
             * reaches this today -- the gate is built for percentile_cont over
             * a random variable, whose circuits the sampler reads -- so it is
             * written to the sampler's reading rather than to a measurement. */
            std::vector<double> members;
            double fraction, pos, fp;
            std::size_t lo;
 
            if (w.empty() || w.size() % 2 != 0) return false;
            try {
              fraction = std::stod(c.getExtra(gx));
            } catch (const std::exception &) {
              return false;
            }
            for (std::size_t i = 0; i < w.size(); i += 2) {
              double ind, x;
              bool ii, xi;
 
              if (!eval(w[i], world, ind, ii)) return false;
              if (ind < 0.5) continue;              /* the row is not here */
              if (!eval(w[i + 1], world, x, xi)) return false;
              members.push_back(x);
            }
            if (members.empty()) return false;
            std::sort(members.begin(), members.end());
            pos = fraction * static_cast<double>(members.size() - 1);
            lo = static_cast<std::size_t>(pos);
            fp = pos - static_cast<double>(lo);
            out = (lo + 1 < members.size())
                    ? members[lo] + fp * (members[lo + 1] - members[lo])
                    : members[lo];
            is_int = false;
            return true;
          }
          /* An arithmetic this evaluator does not know.  Falling through to
           * "this world has no value" is what made a HAVING over sqrt, ln or
           * exp answer probability 0 in every world, silently and with
           * confidence, for as long as those three arms were missing: the
           * caller reads no value as "the comparison does not hold here".  An
           * unimplemented operator has to be told apart from a world that
           * genuinely has no value, so it raises, as every other evaluator's
           * default does (MonteCarloSampler, Expectation).  A new gate_arith
           * operator therefore fails loudly here until it is read. */
          throw CircuitException(
                  "having_semantics: unknown gate_arith operator tag: " +
                  std::to_string(aop));
        }
        return false;
      };
 
      std::vector<typename SemiringT::value_type> kval(n);
      for (size_t i = 0; i < n; ++i)
        kval[i] = c.evaluate<SemiringT>(kgates[i], mapping, S);
 
      // The joint enumeration below is over complete worlds already:
      // certify the disjuncts when the semiring and contributors allow.
      const bool certify = certifiable_contributors(kval);
 
      // Whether the world in which no contributor is present is itself a
      // valid world.  For a grouped aggregation it is not: an empty group is
      // no row, so there is nothing for the comparison to hold of.  A scalar
      // aggregation always yields its single row, empty input included, so
      // that world is real and a predicate true on it (count(*) = 0, and
      // anything the empty-input values satisfy) must pick it up.  Mixed
      // shapes follow the grouped reading: whichever side is grouped has no
      // row there, so the joined row does not exist either.
      const bool empty_world_valid =
        !aggs.empty() &&
        std::all_of(aggs.begin(), aggs.end(),
                    [](const std::pair<const gate_t, AggInfo> &e) {
        return e.second.is_scalar;
      });
      // No contributor at all: the empty world is the only one, and it is
      // only a world for scalar aggregations (a window frame that may be
      // empty while its row exists).
      if (n == 0 && !empty_world_valid)
        return false;
 
      std::vector<typename SemiringT::value_type> disjuncts;
      const uint64_t total = uint64_t(1) << n;
      for (uint64_t world = empty_world_valid ? 0 : 1; world < total; ++world) {
        double lv, rv;
        bool lint, rint;
        if (!eval(Lx, world, lv, lint) || !eval(Rx, world, rv, rint))
          continue;                                       // NULL comparison: false
        bool holds = false;
        switch (opx) {
        case ComparisonOperator::EQ: holds = (lv == rv); break;
        case ComparisonOperator::NE: holds = (lv != rv); break;
        case ComparisonOperator::LT: holds = (lv <  rv); break;
        case ComparisonOperator::LE: holds = (lv <= rv); break;
        case ComparisonOperator::GT: holds = (lv >  rv); break;
        case ComparisonOperator::GE: holds = (lv >= rv); break;
        }
        if (!holds)
          continue;
 
        std::vector<typename SemiringT::value_type> present, missing;
        for (size_t i = 0; i < n; ++i) {
          if (world & (uint64_t(1) << i)) {
            if (certify || kval[i] != S.one()) present.push_back(kval[i]);
          } else {
            if (certify || kval[i] != S.zero()) missing.push_back(kval[i]);
          }
        }
        if (certify) {
          disjuncts.push_back(S.certified_world_term(present, missing));
          continue;
        }
        auto present_prod = S.times(present);
        if (missing.empty())
          disjuncts.push_back(std::move(present_prod));
        else {
          auto monus_factor = S.monus(S.one(), S.plus(missing));
          disjuncts.push_back(
            present_prod == S.one()
              ? monus_factor
              : S.times(std::vector<typename SemiringT::value_type>{
                  present_prod, monus_factor}));
        }
      }
 
      pw_out = disjuncts.empty() ? S.zero()
               : certify ? S.certified_exclusive_plus(disjuncts)
               : S.plus(disjuncts);
      return true;
    };
 
    // ---- Two array_agg() aggregates compared with each other (a join on
    //      two aggregated arrays): the worlds of the tuples of both, which
    //      may be shared, where the arrays are equal (=) or differ (<>).
    //      Elements are compared as their texts, as against a constant. ----
    auto build_array_pair = [&](gate_t Lx, gate_t Rx,
                                ComparisonOperator opx) -> bool {
      if (opx != ComparisonOperator::EQ && opx != ComparisonOperator::NE)
        return false;
      std::map<gate_t, std::size_t> bit;
      std::vector<typename SemiringT::value_type> kvals;
      std::vector<std::size_t> bits[2];
      std::vector<std::string> vals[2];
      const gate_t sides[2] = {Lx, Rx};
 
      for (int side = 0; side < 2; ++side) {
        if (getAggregationOperator(c.getInfos(sides[side]).first) !=
            AggregationOperator::ARRAY_AGG)
          return false;
        for (gate_t ch : c.getWires(sides[side])) {
          if (c.getGateType(ch) != gate_semimod) return false;
          std::string m_str;
          gate_t k_gate{};
          if (!semimod_extract_string_and_K(c, ch, m_str, k_gate)) return false;
          if (c.getUUID(c.getWires(ch)[1]) == GATE_NULL_UUID)
            m_str = array_null_element();
          auto it = bit.find(k_gate);
          if (it == bit.end()) {
            if (kvals.size() >= 20)
              return false;             // 2^n enumeration: keep it bounded
            it = bit.emplace(k_gate, kvals.size()).first;
            kvals.push_back(c.evaluate<SemiringT>(k_gate, mapping, S));
          }
          bits[side].push_back(it->second);
          vals[side].push_back(m_str);
        }
      }
      if (kvals.empty())
        return false;
 
      auto worlds = enumerate_array_agg_pair_worlds(
        bits[0], vals[0], bits[1], vals[1], kvals.size(),
        opx == ComparisonOperator::EQ);
      pw_out = combine_exhaustive_worlds(worlds, kvals, /*upset=*/false,
                                         /*monotone=*/false);
      return true;
    };
 
    if (c.getGateType(L) == gate_agg && build_from(L, R, op))
      return true;
    if (c.getGateType(R) == gate_agg && build_from(R, L, flip_op(op)))
      return true;
    if (c.getGateType(L) == gate_agg && c.getGateType(R) == gate_agg &&
        build_array_pair(L, R, op))
      return true;
 
    return build_general(L, R, op);
  };
 
  // Each comparison gate is resolved on its own: its possible-world
  // enumeration reads the circuit and writes only its own mapping entry, so
  // one gate the enumeration cannot handle -- an agg-vs-agg shape past the
  // contributor cap, an unsupported aggregate -- says nothing about the
  // others.  Skip it and carry on rather than abandoning the gates that
  // follow: those are left for the semiring's own cmp() to deal with, which
  // renders them (Formula) or raises, and a gate this loop could have
  // resolved should not be demoted to that just because of its position in
  // collect_sp_cmp_gates' traversal.
  for (gate_t cmp_gate : cmp_gates) {
    typename SemiringT::value_type pw;
    if (!pw_from_cmp_gate(cmp_gate, pw))
      continue;
 
    mapping[cmp_gate] = std::move(pw);
  }
}
 
#endif