/**
 * @file reachability_evaluate.cpp
 * @brief SQL entry points for decomposition-aligned reachability
 *        compilation over bounded-treewidth data.
 *
 * Exposes @c ReachabilityCompiler (see @c ReachabilityCompiler.h) to SQL:
 *
 * - @c reachability_evaluate(): exact probability that the target vertex
 *   is reachable from the source vertex (two-terminal network
 *   reliability), in time linear in the number of edges for data of
 *   bounded treewidth;
 * - @c reachability_compile_stats(): same compilation, returning the
 *   probability together with structural statistics (data treewidth,
 *   number of bags, maximum DP state count, d-D size) that
 *   substantiate the linear-size claim in tests and benchmarks.
 *
 * Both take the edge relation in columnar form (parallel arrays of
 * source vertices, target vertices, provenance tokens, and
 * probabilities); the user-facing wrappers in @c provsql.sql gather
 * those arrays from an arbitrary provenance-tracked edge relation.
 * Vertices are dense integer IDs (the wrappers map arbitrary vertex
 * values onto them).
 */
extern "C" {
#include "postgres.h"
#include "fmgr.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "access/htup_details.h"
#include "catalog/pg_type.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/tuplestore.h"
#include "utils/uuid.h"

#include "provsql_utils.h"
#include "provsql_mmap.h"
#include "provsql_shmem.h"

PG_FUNCTION_INFO_V1(reachability_evaluate);
PG_FUNCTION_INFO_V1(reachability_compile_stats);
PG_FUNCTION_INFO_V1(reachability_materialize);
PG_FUNCTION_INFO_V1(reachability_materialize_hops);
PG_FUNCTION_INFO_V1(reachability_materialize_any);
PG_FUNCTION_INFO_V1(reachability_materialize_cover);
}

#include "c_cpp_compatibility.h"
#include "ReachabilityCompiler.h"
#include "CertifiedDDMaterialize.h"
#include "provsql_utils_cpp.h"

#include <map>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>

namespace {

/**
 * @brief Validate a 1-D, NULL-free array argument and return its length.
 * @param arr   The array (may be null for "no edges").
 * @param what  Argument name for error messages.
 * @return      Number of elements (0 when @p arr is null).
 */
int checkedArrayLength(ArrayType *arr, const char *what)
{
  if (arr == NULL)
    return 0;
  if (ARR_NDIM(arr) > 1)
    provsql_error("reachability: %s must be a one-dimensional array", what);
  if (ARR_HASNULL(arr))
    provsql_error("reachability: %s must not contain NULLs", what);
  return ARR_NDIM(arr) == 0 ? 0 : ARR_DIMS(arr)[0];
}

/**
 * @brief Decode the four columnar edge arrays (arguments 0..3 of every
 *        entry point) into edge rows.
 *
 * @param fcinfo  PostgreSQL function-call info.
 * @return        The decoded edge rows.
 */
std::vector<ReachabilityCompiler::EdgeRow> edgesFromArgs(
  FunctionCallInfo fcinfo, int block_args_at = -1)
{
  ArrayType *srcs   = PG_ARGISNULL(0) ? NULL : PG_GETARG_ARRAYTYPE_P(0);
  ArrayType *dsts   = PG_ARGISNULL(1) ? NULL : PG_GETARG_ARRAYTYPE_P(1);
  ArrayType *tokens = PG_ARGISNULL(2) ? NULL : PG_GETARG_ARRAYTYPE_P(2);
  ArrayType *probs  = PG_ARGISNULL(3) ? NULL : PG_GETARG_ARRAYTYPE_P(3);
  ArrayType *bkeys  = NULL;
  ArrayType *bidx   = NULL;
  if (block_args_at >= 0) {
    bkeys = PG_ARGISNULL(block_args_at) ? NULL
            : PG_GETARG_ARRAYTYPE_P(block_args_at);
    bidx  = PG_ARGISNULL(block_args_at+1) ? NULL
            : PG_GETARG_ARRAYTYPE_P(block_args_at+1);
  }

  const int n = checkedArrayLength(srcs, "sources");
  if (checkedArrayLength(dsts, "destinations") != n ||
      checkedArrayLength(tokens, "tokens") != n ||
      checkedArrayLength(probs, "probabilities") != n)
    provsql_error("reachability: edge arrays must have the same length");
  if (bkeys != NULL &&
      (checkedArrayLength(bkeys, "block keys") != n ||
       checkedArrayLength(bidx, "block indices") != n))
    provsql_error("reachability: edge arrays must have the same length");

  std::vector<ReachabilityCompiler::EdgeRow> rows;
  rows.reserve(n);

  if (n > 0) {
    /* All element types are fixed-length and NULL-free (checked above),
     * so the data areas are packed and can be read directly, as
     * create_gate does for uuid[]. */
    const int32 *src_data = (const int32 *) ARR_DATA_PTR(srcs);
    const int32 *dst_data = (const int32 *) ARR_DATA_PTR(dsts);
    const pg_uuid_t *token_data = (const pg_uuid_t *) ARR_DATA_PTR(tokens);
    const float8 *prob_data = (const float8 *) ARR_DATA_PTR(probs);
    const pg_uuid_t *bkey_data =
      bkeys ? (const pg_uuid_t *) ARR_DATA_PTR(bkeys) : NULL;
    const int32 *bidx_data = bidx ? (const int32 *) ARR_DATA_PTR(bidx) : NULL;

    for (int i = 0; i < n; ++i) {
      ReachabilityCompiler::EdgeRow row;
      row.src = static_cast<unsigned long>(src_data[i]);
      row.dst = static_cast<unsigned long>(dst_data[i]);
      row.token = uuid2string(token_data[i]);
      row.prob = prob_data[i];
      if (row.prob < 0. || row.prob > 1.)
        provsql_error("reachability: edge probability %f out of [0,1]",
                      row.prob);
      if (bkey_data) {
        bool nil = true;
        for (int b = 0; b < 16; ++b)
          if (bkey_data[i].data[b] != 0)
            nil = false;
        if (!nil) {
          row.block_key = uuid2string(bkey_data[i]);
          row.block_index = static_cast<unsigned>(bidx_data[i]);
        }
      }
      rows.push_back(std::move(row));
    }
  }

  return rows;
}

/**
 * @brief Decode the source-arc arrays (three consecutive arguments
 *        starting at @p base): vertices, tokens (nil = certain) and
 *        probabilities.
 *
 * @param fcinfo  PostgreSQL function-call info.
 * @param base    Index of the source-vertices argument.
 * @return        The decoded source arcs.
 */
std::vector<ReachabilityCompiler::SourceArc> sourcesFromArgs(
  FunctionCallInfo fcinfo, int base)
{
  std::vector<ReachabilityCompiler::SourceArc> sources;
  ArrayType *sv = PG_ARGISNULL(base) ? NULL : PG_GETARG_ARRAYTYPE_P(base);
  ArrayType *st = PG_ARGISNULL(base+1) ? NULL : PG_GETARG_ARRAYTYPE_P(base+1);
  ArrayType *sp = PG_ARGISNULL(base+2) ? NULL : PG_GETARG_ARRAYTYPE_P(base+2);
  const int ns = checkedArrayLength(sv, "source vertices");
  if (checkedArrayLength(st, "source tokens") != ns ||
      checkedArrayLength(sp, "source probabilities") != ns)
    provsql_error("reachability: source arrays must have the same length");
  if (ns == 0)
    provsql_error("reachability: at least one source is required");
  const int32 *v_data = (const int32 *) ARR_DATA_PTR(sv);
  const pg_uuid_t *t_data = (const pg_uuid_t *) ARR_DATA_PTR(st);
  const float8 *p_data = (const float8 *) ARR_DATA_PTR(sp);
  sources.reserve(ns);
  for (int i = 0; i < ns; ++i) {
    ReachabilityCompiler::SourceArc sa;
    sa.vertex = static_cast<unsigned long>(v_data[i]);
    bool nil = true;
    for (int b = 0; b < 16; ++b)
      if (t_data[i].data[b] != 0)
        nil = false;
    sa.certain = nil;
    if (!nil)
      sa.token = uuid2string(t_data[i]);
    sa.prob = p_data[i];
    if (sa.prob < 0. || sa.prob > 1.)
      provsql_error("reachability: source probability %f out of [0,1]",
                    sa.prob);
    sources.push_back(std::move(sa));
  }
  return sources;
}

/**
 * @brief Decode the single-target argument layout and run the compilation.
 *
 * Argument layout: @c srcs @c int[], @c dsts @c int[], @c tokens
 * @c uuid[], @c probs @c float8[], @c source @c int, @c target @c int,
 * @c directed @c boolean.
 *
 * @param fcinfo  PostgreSQL function-call info.
 * @return        The compiled d-D and statistics.
 */
ReachabilityCompiler::Result compileFromArgs(FunctionCallInfo fcinfo)
{
  for (int i = 4; i < 7; ++i)
    if (PG_ARGISNULL(i))
      provsql_error("reachability: source, target and directed must not be NULL");

  auto rows = edgesFromArgs(fcinfo);
  const unsigned long source = static_cast<unsigned long>(PG_GETARG_INT32(4));
  const unsigned long target = static_cast<unsigned long>(PG_GETARG_INT32(5));
  const bool directed = PG_GETARG_BOOL(6);

  try {
    return ReachabilityCompiler::compile(rows, source, target, directed);
  } catch (TreeDecompositionException &) {
    provsql_error(
      "reachability: data treewidth exceeds the supported limit (%d)",
      TreeDecomposition::MAX_TREEWIDTH);
    throw; /* unreachable; placate the compiler */
  }
}


} // namespace

/**
 * @brief PostgreSQL-callable entry point: exact reachability probability.
 *
 * Arguments: see @c compileFromArgs().
 * Returns: the probability that @c target is reachable from @c source.
 */
Datum reachability_evaluate(PG_FUNCTION_ARGS)
{
  try {
    auto result = compileFromArgs(fcinfo);
    PG_RETURN_FLOAT8(result.dd.probabilityEvaluation());
  } catch (const std::exception &e) {
    provsql_error("reachability: %s", e.what());
  } catch (...) {
    provsql_error("reachability: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief PostgreSQL-callable entry point: probability plus compilation
 *        statistics.
 *
 * Arguments: see @c compileFromArgs().
 * Returns: composite @c (probability, data_treewidth, nb_bags,
 * max_states, nb_gates, nb_variables).
 */
Datum reachability_compile_stats(PG_FUNCTION_ARGS)
{
  try {
    auto result = compileFromArgs(fcinfo);

    TupleDesc tupdesc;
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
      provsql_error("reachability_compile_stats: expected composite return type");
    tupdesc = BlessTupleDesc(tupdesc);

    Datum values[6];
    bool nulls[6] = {false, false, false, false, false, false};
    values[0] = Float8GetDatum(result.dd.probabilityEvaluation());
    values[1] = Int32GetDatum(static_cast<int32>(result.stats.data_treewidth));
    values[2] = Int64GetDatum(static_cast<int64>(result.stats.nb_bags));
    values[3] = Int64GetDatum(static_cast<int64>(result.stats.max_states));
    values[4] = Int64GetDatum(static_cast<int64>(result.stats.nb_gates));
    values[5] = Int64GetDatum(static_cast<int64>(result.stats.nb_variables));

    PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
  } catch (const std::exception &e) {
    provsql_error("reachability_compile_stats: %s", e.what());
  } catch (...) {
    provsql_error("reachability_compile_stats: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief PostgreSQL-callable entry point: all-targets compilation and
 *        materialisation.
 *
 * Arguments: @c srcs @c int[], @c dsts @c int[], @c tokens @c uuid[],
 * @c probs @c float8[], @c block_keys @c uuid[] (nil = independent
 * tuple; otherwise the BID key variable grouping mutually exclusive
 * alternatives) and @c block_indices @c int[], @c source_vertices
 * @c int[], @c source_tokens @c uuid[] (the nil UUID marking a certain
 * source), @c source_probs @c float8[], @c directed @c boolean.
 * Returns: one @c (vertex, token) row per vertex reachable in the
 * all-edges-present world, @c token being the materialised certified
 * provenance circuit of "some present source reaches the vertex",
 * wrapped in the @c 'absorptive' assumption marker (see
 * @c wrapAssumedAbsorptive).  This is the engine behind the rewriter's
 * recursive-reachability route.
 */
Datum reachability_materialize(PG_FUNCTION_ARGS)
{
  try {
    if (PG_ARGISNULL(9))
      provsql_error("reachability: directed must not be NULL");

    auto rows = edgesFromArgs(fcinfo, 4);
    const bool directed = PG_GETARG_BOOL(9);

    /* Sources: parallel arrays of vertices, tokens and probabilities; the
     * nil UUID marks a *certain* source (an untracked source relation, or
     * the constant base arm of the recursive shape). */
    const auto sources = sourcesFromArgs(fcinfo, 6);

    ReachabilityCompiler::AllResult all;
    try {
      all = ReachabilityCompiler::compileAll(rows, sources, directed);
    } catch (TreeDecompositionException &) {
      provsql_error(
        "reachability: data treewidth exceeds the supported limit (%d)",
        TreeDecomposition::MAX_TREEWIDTH);
    }

    std::vector<gate_t> roots;
    roots.reserve(all.roots.size());
    for (const auto &vr : all.roots)
      roots.push_back(vr.root);
    const auto uuid_of = materializeCertifiedDD(all.dd, roots);

    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    MemoryContext per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    MemoryContext oldcontext = MemoryContextSwitchTo(per_query_ctx);

    TupleDesc tupdesc;
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
      MemoryContextSwitchTo(oldcontext);
      provsql_error("reachability_materialize: function must return a row type");
    }
    tupdesc = BlessTupleDesc(tupdesc);

    Tuplestorestate *tupstore = tuplestore_begin_heap(
      rsinfo->allowedModes & SFRM_Materialize_Random, false, work_mem);
    rsinfo->returnMode = SFRM_Materialize;
    rsinfo->setResult = tupstore;
    rsinfo->setDesc = tupdesc;

    for (const auto &vr : all.roots) {
      Datum values[2];
      bool nulls[2] = {false, false};
      values[0] = Int32GetDatum(static_cast<int32>(vr.vertex));
      pg_uuid_t *u = (pg_uuid_t *) palloc(sizeof(pg_uuid_t));
      *u = wrapAssumedAbsorptive(uuid_of.at(vr.root));
      values[1] = UUIDPGetDatum(u);
      tuplestore_putvalues(tupstore, tupdesc, values, nulls);
    }

    MemoryContextSwitchTo(oldcontext);
    return (Datum) 0;
  } catch (const std::exception &e) {
    provsql_error("reachability_materialize: %s", e.what());
  } catch (...) {
    provsql_error("reachability_materialize: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief PostgreSQL-callable entry point: bounded-hop all-targets
 *        compilation and materialisation.
 *
 * Arguments 0..9 as @c reachability_materialize, then @c hop_bound
 * @c int (maximum walk length) and @c hop_seed @c int (the recursive
 * CTE's base-arm hop constant, added to the reported lengths).
 * Returns: one @c (vertex, hops, token) row per (vertex, walk length)
 * pair achievable in the all-edges-present world -- matching the rows
 * the generic fixpoint derives for the hop-counting CTE shape, with
 * @c token the materialised certified circuit of "some present source
 * reaches the vertex by a walk of exactly this many edges", wrapped in
 * the @c 'absorptive' assumption marker.
 *
 * Additionally pre-creates, for every vertex with at least two length
 * rows, the gate a hop-discarding query's deduplication will mint --
 * @c uuid5('plus{sorted tokens}') over the vertex's length tokens
 * (multiset, as @c provenance_plus aggregates them) -- as a certified
 * single-child @c plus over the DP's native within-bound root, which
 * computes the same Boolean function as that OR but *deterministically
 * by construction*.  The natural "is the vertex within k hops" query
 * thus evaluates through the linear certified route instead of falling
 * back to generic knowledge compilation over correlated per-length
 * tokens; the pre-created gate is content-addressed, so the rewriter's
 * later @c create_gate of the same UUID is an idempotent no-op.
 */
Datum reachability_materialize_hops(PG_FUNCTION_ARGS)
{
  try {
    if (PG_ARGISNULL(9) || PG_ARGISNULL(10) || PG_ARGISNULL(11))
      provsql_error(
        "reachability: directed, hop_bound and hop_seed must not be NULL");

    auto rows = edgesFromArgs(fcinfo, 4);
    const bool directed = PG_GETARG_BOOL(9);
    const int32 hop_bound = PG_GETARG_INT32(10);
    const int32 hop_seed = PG_GETARG_INT32(11);
    if (hop_bound < 0 ||
        static_cast<unsigned>(hop_bound) > ReachabilityCompiler::MAX_HOP_BOUND)
      provsql_error("reachability: hop bound %d out of [0,%u]",
                    hop_bound, ReachabilityCompiler::MAX_HOP_BOUND);

    const auto sources = sourcesFromArgs(fcinfo, 6);

    ReachabilityCompiler::AllHopsResult all;
    try {
      all = ReachabilityCompiler::compileAllHops(
        rows, sources, directed, static_cast<unsigned>(hop_bound));
    } catch (TreeDecompositionException &) {
      provsql_error(
        "reachability: data treewidth exceeds the supported limit (%d)",
        TreeDecomposition::MAX_TREEWIDTH);
    }

    std::vector<gate_t> roots;
    roots.reserve(all.roots.size() + all.within_roots.size());
    for (const auto &vr : all.roots)
      roots.push_back(vr.root);
    for (const auto &vr : all.within_roots)
      roots.push_back(vr.root);
    const auto uuid_of = materializeCertifiedDD(all.dd, roots);

    /* Dedup pre-creation: per vertex, the multiset of its length tokens
     * *as the work table carries them* (i.e. wrapped in the 'absorptive'
     * assumption marker, like every materialised root), sorted as text,
     * addressed in the dedicated "plus-canonical" recipe namespace that
     * provenance_plus probes (and never creates under, so a hit there is
     * always a deliberate pre-creation).  The aliased child is the DP's
     * native within-bound root, wrapped in the same marker so the
     * aggregated token refuses non-absorptive evaluation too. */
    {
      std::unordered_map<unsigned long, std::vector<std::string> > by_vertex;
      for (const auto &vr : all.roots)
        by_vertex[vr.vertex].push_back(
          uuid2string(wrapAssumedAbsorptive(uuid_of.at(vr.root))));
      for (const auto &vr : all.within_roots) {
        auto it = by_vertex.find(vr.vertex);
        if (it == by_vertex.end() || it->second.size() < 2)
          continue;
        std::vector<std::string> texts = it->second;
        std::sort(texts.begin(), texts.end());
        std::string name = "plus-canonical{";
        for (std::size_t i = 0; i < texts.size(); ++i) {
          if (i)
            name += ",";
          name += texts[i];
        }
        name += "}";
        const pg_uuid_t dedup = provsqlUuidV5(name);
        const pg_uuid_t within =
          wrapAssumedAbsorptive(uuid_of.at(vr.root));
        provsql_internal_create_gate(&dedup, gate_plus, 1, &within);
        /* Route tag in info2, like every materialised root: this alias is a
         * user-visible root too, and must report 'reachability' rather than
         * the generic 'independent'. */
        provsql_internal_set_infos(&dedup, DNNF_CERT_INFO,
                                   PROVSQL_ROUTE_REACHABILITY);
      }
    }

    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    MemoryContext per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    MemoryContext oldcontext = MemoryContextSwitchTo(per_query_ctx);

    TupleDesc tupdesc;
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
      MemoryContextSwitchTo(oldcontext);
      provsql_error(
        "reachability_materialize_hops: function must return a row type");
    }
    tupdesc = BlessTupleDesc(tupdesc);

    Tuplestorestate *tupstore = tuplestore_begin_heap(
      rsinfo->allowedModes & SFRM_Materialize_Random, false, work_mem);
    rsinfo->returnMode = SFRM_Materialize;
    rsinfo->setResult = tupstore;
    rsinfo->setDesc = tupdesc;

    for (const auto &vr : all.roots) {
      Datum values[3];
      bool nulls[3] = {false, false, false};
      values[0] = Int32GetDatum(static_cast<int32>(vr.vertex));
      values[1] = Int32GetDatum(hop_seed + static_cast<int32>(vr.hops));
      pg_uuid_t *u = (pg_uuid_t *) palloc(sizeof(pg_uuid_t));
      *u = wrapAssumedAbsorptive(uuid_of.at(vr.root));
      values[2] = UUIDPGetDatum(u);
      tuplestore_putvalues(tupstore, tupdesc, values, nulls);
    }

    MemoryContextSwitchTo(oldcontext);
    return (Datum) 0;
  } catch (const std::exception &e) {
    provsql_error("reachability_materialize_hops: %s", e.what());
  } catch (...) {
    provsql_error("reachability_materialize_hops: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief PostgreSQL-callable entry point: per-group "some member
 *        reachable" compilation and materialisation.
 *
 * Arguments 0..9 as @c reachability_materialize, then two parallel
 * arrays flattening the groups: @c group_ids @c int[] and
 * @c member_vertices @c int[] (dense vertex IDs).  For each distinct
 * group, compiles the certified circuit of "some member vertex is
 * reachable from a present source" (@c compileAnyReach: the
 * set-reachability bit folded through the decomposition DP, so the
 * disjunction over the group's *correlated* per-vertex events is
 * deterministic by construction) and materialises it; returns one
 * @c (group_id, token) row per group, each token wrapped in the
 * @c 'absorptive' assumption marker.  The caller plants each token
 * under the canonical address of the group's per-vertex reach tokens,
 * keeping cross-vertex aggregations ("is some vertex of this region
 * reachable") on the linear certified route.
 */
Datum reachability_materialize_any(PG_FUNCTION_ARGS)
{
  try {
    if (PG_ARGISNULL(9))
      provsql_error("reachability: directed must not be NULL");

    auto rows = edgesFromArgs(fcinfo, 4);
    const bool directed = PG_GETARG_BOOL(9);
    const auto sources = sourcesFromArgs(fcinfo, 6);

    ArrayType *gids = PG_ARGISNULL(10) ? NULL : PG_GETARG_ARRAYTYPE_P(10);
    ArrayType *gverts = PG_ARGISNULL(11) ? NULL : PG_GETARG_ARRAYTYPE_P(11);
    const int ng = checkedArrayLength(gids, "group ids");
    if (checkedArrayLength(gverts, "group member vertices") != ng)
      provsql_error("reachability: group arrays must have the same length");
    if (ng == 0)
      provsql_error("reachability: at least one group member is required");

    const int32 *gid_data = (const int32 *) ARR_DATA_PTR(gids);
    const int32 *gv_data = (const int32 *) ARR_DATA_PTR(gverts);
    std::map<int32, std::vector<unsigned long> > groups;
    for (int i = 0; i < ng; ++i)
      groups[gid_data[i]].push_back(static_cast<unsigned long>(gv_data[i]));
    std::vector<int32> group_ids;
    std::vector<std::vector<unsigned long> > sets;
    group_ids.reserve(groups.size());
    sets.reserve(groups.size());
    for (auto &[gid, members] : groups) {
      group_ids.push_back(gid);
      sets.push_back(std::move(members));
    }

    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    MemoryContext per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    MemoryContext oldcontext = MemoryContextSwitchTo(per_query_ctx);

    TupleDesc tupdesc;
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
      MemoryContextSwitchTo(oldcontext);
      provsql_error(
        "reachability_materialize_any: function must return a row type");
    }
    tupdesc = BlessTupleDesc(tupdesc);

    Tuplestorestate *tupstore = tuplestore_begin_heap(
      rsinfo->allowedModes & SFRM_Materialize_Random, false, work_mem);
    rsinfo->returnMode = SFRM_Materialize;
    rsinfo->setResult = tupstore;
    rsinfo->setDesc = tupdesc;

    // One shared compilation for all groups (the prelude and the
    // seed-independent parts of the circuit are computed and emitted
    // once), then one materialisation pass over all the roots (shared
    // gates are walked once).
    ReachabilityCompiler::AnyReachAllResult all;
    try {
      all = ReachabilityCompiler::compileAnyReachAll(rows, sources, sets,
                                                     directed);
    } catch (TreeDecompositionException &) {
      MemoryContextSwitchTo(oldcontext);
      provsql_error(
        "reachability: data treewidth exceeds the supported limit (%d)",
        TreeDecomposition::MAX_TREEWIDTH);
    }
    const auto uuid_of = materializeCertifiedDD(all.dd, all.roots);

    for (std::size_t i = 0; i < group_ids.size(); ++i) {
      Datum values[2];
      bool nulls[2] = {false, false};
      values[0] = Int32GetDatum(group_ids[i]);
      pg_uuid_t *u = (pg_uuid_t *) palloc(sizeof(pg_uuid_t));
      *u = wrapAssumedAbsorptive(uuid_of.at(all.roots[i]));
      values[1] = UUIDPGetDatum(u);
      tuplestore_putvalues(tupstore, tupdesc, values, nulls);
    }

    MemoryContextSwitchTo(oldcontext);
    return (Datum) 0;
  } catch (const std::exception &e) {
    provsql_error("reachability_materialize_any: %s", e.what());
  } catch (...) {
    provsql_error("reachability_materialize_any: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief PostgreSQL-callable entry point: "every member vertex
 *        reachable" (k-terminal / coverage) compilation and
 *        materialisation.
 *
 * Arguments 0..9 as @c reachability_materialize, then
 * @c member_vertices @c int[] (dense vertex IDs).  Compiles the
 * certified circuit of "every member vertex is reachable from a
 * present source" (@c compileCoverReach: the pending rescuer-set
 * antichain folded through the decomposition DP, so the conjunction
 * over the members' *correlated* per-vertex events is deterministic
 * by construction), materialises it, and returns its token, wrapped
 * in the @c 'absorptive' assumption marker.  The caller plants the
 * token under the times-canonical address of the members' per-vertex
 * reach tokens, keeping reachability self-join conjunctions ("are
 * these k vertices all reachable") on the linear certified route --
 * with the joint-worlds semantics: under nonnegative min-plus the
 * token evaluates to the cost of the cheapest covering subgraph
 * (directed Steiner cost), shared edges paid once.
 */
Datum reachability_materialize_cover(PG_FUNCTION_ARGS)
{
  try {
    if (PG_ARGISNULL(9))
      provsql_error("reachability: directed must not be NULL");

    auto rows = edgesFromArgs(fcinfo, 4);
    const bool directed = PG_GETARG_BOOL(9);
    const auto sources = sourcesFromArgs(fcinfo, 6);

    ArrayType *mverts = PG_ARGISNULL(10) ? NULL : PG_GETARG_ARRAYTYPE_P(10);
    const int nm = checkedArrayLength(mverts, "member vertices");
    if (nm == 0)
      provsql_error("reachability: at least one member vertex is required");
    const int32 *mv_data = (const int32 *) ARR_DATA_PTR(mverts);
    std::vector<unsigned long> set;
    set.reserve(nm);
    for (int i = 0; i < nm; ++i)
      set.push_back(static_cast<unsigned long>(mv_data[i]));

    ReachabilityCompiler::AnyReachAllResult all;
    try {
      all = ReachabilityCompiler::compileCoverReachAll(rows, sources, {set},
                                                       directed);
    } catch (TreeDecompositionException &) {
      provsql_error(
        "reachability: data treewidth exceeds the supported limit (%d)",
        TreeDecomposition::MAX_TREEWIDTH);
    }
    const auto uuid_of = materializeCertifiedDD(all.dd, all.roots);

    pg_uuid_t *u = (pg_uuid_t *) palloc(sizeof(pg_uuid_t));
    *u = wrapAssumedAbsorptive(uuid_of.at(all.roots[0]));
    PG_RETURN_UUID_P(u);
  } catch (const std::exception &e) {
    provsql_error("reachability_materialize_cover: %s", e.what());
  } catch (...) {
    provsql_error("reachability_materialize_cover: unknown exception");
  }
  PG_RETURN_NULL();
}