package gudusoft.gsqlparser.ir.semantic.diff;

import gudusoft.gsqlparser.EDbVendor;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.InputSource;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import java.io.StringReader;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;

/**
 * Project the existing {@code DataFlowAnalyzer} XML output into the same
 * canonical form as the Semantic IR projector.
 *
 * <p>The projector consumes only the four XML shapes seen in the captured
 * 5-SQL corpus baselines (see {@code phase0/golden/*.dlineage.xml}). Any
 * input that doesn't fit produces a {@link ProjectorResult} carrying an
 * {@link ProjectorResult.UnsupportedReason}; the harness translates it
 * into a single {@link DivergenceClass#UNSUPPORTED_BY_DLINEAGE} divergence.
 *
 * <p>Aggregate handling — see slice-7 plan §"DlineageXmlProjector". The
 * COUNT(*) heuristic detects function-resultset fdd sources that include
 * a {@code column="*"} reference and suppresses SELECT fan-out from that
 * function. {@code SUM(col)}/{@code AVG(col)}/{@code COUNT(col)} retain
 * their fan-out.
 */
public final class DlineageXmlProjector {

    /**
     * Lower-cased function names whose presence on the immediate source-side
     * of an output's fdd marks that output as aggregate. Mirrors the
     * whitelist in {@code SemanticIRBuilder.AGGREGATE_FUNCTION_NAMES} so that
     * scalar functions like {@code UPPER}, {@code COALESCE}, or {@code TRIM}
     * do not cause a false {@link DivergenceClass#AGGREGATION_MISMATCH}
     * against the IR's {@code OutputColumn.isAggregate()}.
     *
     * <p>Slice 30 added {@code mode} (PostgreSQL ordered-set aggregate). See
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES} for the matching
     * window-vs-aggregate discriminator override.
     */
    private static final Set<String> AGGREGATE_FUNCTION_NAMES;
    static {
        Set<String> s = new HashSet<>(Arrays.asList(
                "count", "sum", "avg", "min", "max",
                "stddev", "variance", "var_samp", "var_pop",
                "stddev_samp", "stddev_pop",
                "listagg", "string_agg", "group_concat", "array_agg",
                // Slice 30: PostgreSQL ordered-set aggregate. Mirrors
                // SemanticIRBuilder.AGGREGATE_FUNCTION_NAMES; see also
                // ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES below.
                "mode",
                // Slice 42: hypothetical-set ordered-set aggregates
                // (Oracle / MSSQL {@code RANK(100) WITHIN GROUP (ORDER
                // BY x)} family). On Oracle / MSSQL the dlineage XML for
                // this shape emits NO {@code clauseType="orderby"} fdr —
                // the slice-13 windowed discriminator
                // {@link #isWindowFunctionResultset} returns false and
                // these names mark the output aggregate=true. The OVER
                // form on Oracle / MSSQL ({@code RANK() OVER (ORDER BY
                // x)}) emits {@code clauseType="orderby"} and the
                // discriminator returns true (windowed) — adding these
                // names to the aggregate set therefore does not affect
                // OVER-form classification. The PG direct-attachment
                // hypothetical-set form ({@code rank(0.5) WITHIN GROUP
                // (...)}) and PG OVER form share the same XML shape;
                // both are classified as windowed (rank is NOT in
                // ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES) — slice 42
                // does NOT lift PG, the IR builder rejects PG
                // hypothetical-set so the divergence harness never
                // sees that case. Probe-confirmed
                // {@code /tmp/probe42/Probe42.java}.
                "rank", "dense_rank", "percent_rank", "cume_dist"));
        AGGREGATE_FUNCTION_NAMES = Collections.unmodifiableSet(s);
    }

    /**
     * Function names whose {@code clauseType="orderby"} fdr may be the
     * WITHIN GROUP order-by rather than a window {@code OVER ORDER BY}.
     *
     * <p>Slice 30 introduced {@code mode}. Slice 35 added PostgreSQL
     * {@code listagg}; slice 36 adds PostgreSQL {@code string_agg}. Probes
     * show dlineage emits the WITHIN GROUP order-by for both as an fdr with
     * {@code clauseType="orderby"} structurally identical to LISTAGG (probe
     * /tmp/ProbeStringAggXml.java verified the XML for aliased and unaliased
     * STRING_AGG WG: a {@code resultset type="function"} STRING_AGG wrapper,
     * an {@code fdd effectType="function"} target/source pair for the arg,
     * and a single {@code fdr clauseType="orderby"} for the WG key — no
     * {@code clauseType="selectList"} fdr is emitted, so the slice-13
     * projector otherwise mistakes the orderby fdr for a window discriminator
     * and reports {@link DivergenceClass#AGGREGATION_MISMATCH}). This set
     * does NOT blindly short-circuit window classification:
     * {@link #isWindowFunctionResultset} still returns {@code true} when a
     * {@code selectList} fdr is present (PARTITION BY / analytic dependency),
     * and only suppresses the order-by-only shape.
     *
     * <p><b>Subset constraint</b>: must be a strict subset of
     * {@link #AGGREGATE_FUNCTION_NAMES} so the IR-side / projector-side
     * aggregate-flag invariants stay symmetric. The class-init check below
     * enforces this.
     *
     * <p><b>Why not include {@code percentile_cont} / {@code percentile_disc}</b>:
     * these have windowed forms in BigQuery / Vertica / Redshift / Oracle /
     * SQL Server (e.g. {@code PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY x)
     * OVER (PARTITION BY y)}). Lifting them requires a stronger structural
     * discriminator. Similarly, hypothetical-set {@code rank} /
     * {@code dense_rank} remain excluded because their PG XML is structurally
     * identical to window form {@code RANK() OVER (ORDER BY ...)}.
     */
    private static final Set<String> ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES;
    static {
        Set<String> s = new HashSet<>();
        s.add("mode");
        s.add("listagg");
        s.add("string_agg");
        ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES = Collections.unmodifiableSet(s);
        // Subset assertion: must be enforced at runtime (not via assert,
        // which can be disabled). A future contributor adding a name to
        // ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES without also adding it to
        // AGGREGATE_FUNCTION_NAMES would produce silently-wrong projector
        // output (the function would be "non-aggregate but
        // not-windowed-either" — first-hop check skips it entirely).
        if (!AGGREGATE_FUNCTION_NAMES.containsAll(ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES)) {
            Set<String> missing = new HashSet<>(ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES);
            missing.removeAll(AGGREGATE_FUNCTION_NAMES);
            throw new IllegalStateException(
                    "ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES must be a subset of "
                            + "AGGREGATE_FUNCTION_NAMES; missing=" + missing);
        }
    }

    /**
     * Functions whose {@code (*)} argument is a row-counting marker rather
     * than a real value dependency, so SELECT fan-out from them should be
     * suppressed (otherwise the dlineage XML's expansion to every joined
     * column would produce false {@link DivergenceClass#IR_MISSING_DEPENDENCY}
     * edges relative to the IR's "aggregate, no sources" representation).
     * Today the only widely-supported example is {@code COUNT(*)}.
     */
    private static final Set<String> NULL_ARG_AGGREGATES;
    static {
        Set<String> s = new HashSet<>();
        s.add("count");
        NULL_ARG_AGGREGATES = Collections.unmodifiableSet(s);
    }

    /**
     * Slice 48 — wrapper-function names through which the projector will
     * descend ONE fdd hop to look for a known ordered-set aggregate
     * inner.
     *
     * <p>Slice 48 introduced {@code lower}. Slice 50 widened the set to
     * include {@code upper} — the natural case-folding sibling. Slice 51
     * widened the set further to include {@code concat} — the canonical
     * multi-argument scalar concatenation wrapper. Slice 52 widened the
     * set again to include {@code trim} / {@code ltrim} / {@code rtrim} —
     * the canonical whitespace-trimming scalar wrappers. Slice 53 widened
     * it to include {@code substring} / {@code substr} — the
     * three-argument string-extraction wrappers. Slice 54 widened it once
     * more to include {@code replace} — the canonical three-argument
     * pattern-replacement wrapper. Slice 55 widens it again to include
     * {@code lpad} / {@code rpad} — the canonical three-argument string
     * padding wrappers. Slice 51 probes ({@code /tmp/Probe51.java},
     * {@code /tmp/Probe51b.java}), slice 52 probe
     * ({@code /tmp/Probe52.java}), slice 53 probe
     * ({@code /tmp/Probe53.java}), slice 54 probe (deleted
     * post-implementation; see {@code Slice54Test} javadoc) and slice 55
     * probe (transient {@code Slice55LpadRpadProbeTest} inside the Maven
     * test JVM, deleted post-confirmation; see {@code Slice55Test}
     * javadoc) confirmed the dlineage XML for {@code CONCAT(...)} /
     * {@code TRIM(...)} / {@code LTRIM(...)} / {@code RTRIM(...)} /
     * {@code SUBSTRING(...)} / {@code SUBSTR(...)} / {@code REPLACE(...)}
     * / {@code LPAD(...)} / {@code RPAD(...)} differs from
     * {@code LOWER(...)} / {@code UPPER(...)} only in the wrapper
     * resultset's {@code name} attribute. The literal-string /
     * character-class / direction-marker / numeric start+length /
     * pattern+replacement / pad-length / pad-character arguments do NOT
     * generate extra {@code fdd} edges on the wrapping column
     * (constants and BOTH/LEADING/TRAILING/FROM/FOR markers do not
     * produce fdd edges in the dlineage XML), so the structural shape is
     * byte-identical to the case-folding wrapper shape: one
     * {@code fdd effectType="select"} from the SELECT-list resultset to
     * the wrapper column, then one {@code fdd effectType="function"}
     * from the wrapper column to the inner ordered-set aggregate's
     * resultset, then the inner aggregate's own argument fdd and
     * orderby fdr exactly as in the case-folding wrapper shape. The
     * IR-side classification (aggregate=true via the slice-31 collector
     * descending into the wrapping function body) and the dlineage-side
     * aggregate classification lift symmetrically once the wrapper
     * allowlist contains the new wrapper name.
     *
     * <p>The {@code ||} (string-concatenation) operator is a special
     * non-case: PG / Snowflake parsers fold {@code expr || ' '} into
     * the projection root WITHOUT emitting a {@code concat} or
     * {@code ||}-named function-resultset wrapper. The dlineage XML for
     * {@code LISTAGG WG || '|'} is byte-identical to the unwrapped
     * {@code LISTAGG WG} form — no descent is needed because the
     * existing first-hop aggregate detection
     * (slice-30 / 35 / 36 carve-outs in
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES}) already
     * classifies that direct path as aggregate=true. Slice 51 / 52 / 53
     * / 54 therefore do NOT add {@code ||} to this allowlist; the
     * wrapper widen is only required for explicit named function calls.
     *
     * <p>{@code TRIM(BOTH ' ' FROM x)} (PG / Snowflake special grammar)
     * is a syntactic surface only — slice 52 probe confirmed the
     * dlineage XML still emits a {@code resultset type="function"}
     * named {@code TRIM} with the same one-fdd-edge connection chain
     * as canonical {@code TRIM(x)}. The descent therefore fires for
     * both forms unchanged. Likewise, the PG SQL-standard
     * {@code SUBSTRING(x FROM 1 FOR 5)} grammar form (with
     * {@code FROM}/{@code FOR} keyword markers) emits the same chain
     * shape as the canonical {@code SUBSTRING(x, 1, 5)} comma form —
     * slice 53 probe confirmed parity. The two-argument
     * {@code SUBSTRING(x, 1)} call (no length) emits the same chain
     * because constants do not produce fdd edges. Slice 54
     * {@code REPLACE(x, pattern, replacement)} probe confirmed the
     * pattern and replacement string literals likewise produce no fdd
     * edges on the wrapping {@code REPLACE.REPLACE} column, so the
     * descent fires identically regardless of the literal values.
     * Slice 55 {@code LPAD(x, length, padchar)} /
     * {@code RPAD(x, length, padchar)} probe confirmed the integer
     * pad-length and literal-string pad-character arguments likewise
     * produce no fdd edges on the wrapping {@code LPAD.LPAD} /
     * {@code RPAD.RPAD} column, so the descent fires identically
     * regardless of the pad-length / pad-character values.
     *
     * <p>Slice 56 {@code LEFT(x, length)} / {@code RIGHT(x, length)}
     * probe confirmed the integer length argument (2nd arg) likewise
     * produces no fdd edges on the wrapping {@code LEFT.LEFT} /
     * {@code RIGHT.RIGHT} column — only the inner ordered-set
     * aggregate's resultset is the fdd source — so the descent fires
     * identically regardless of the literal length value.
     *
     * <p>Slice 57 {@code REGEXP_REPLACE(x, pattern, replacement [,
     * flags] [, position, occurrence, parameters])} probe (transient
     * test inside the Maven test JVM, deleted post-confirmation)
     * confirmed the regex pattern, replacement string, optional
     * flag, and Snowflake positional / occurrence / parameter
     * literals likewise produce no fdd edges on the wrapping
     * {@code REGEXP_REPLACE.REGEXP_REPLACE} column — only the inner
     * ordered-set aggregate's resultset is the fdd source — so the
     * descent fires identically regardless of the constant
     * pattern / replacement / flag / positional values. The PG
     * 3-arg form, PG 4-arg flag form, SF 3-arg form, and SF 6-arg
     * (position / occurrence / parameters) form all produce the
     * same chain shape as slice-49 {@code LOWER(...)}.
     *
     * <p>The wrapper allowlist is intentionally NOT vendor-gated:
     * PostgreSQL and Snowflake are the validated scope per the
     * slice-57 probe, but any dialect whose dlineage XML emits a
     * {@code REGEXP_REPLACE}-named function-resultset wrapper around
     * a {@code mode} / {@code listagg} / {@code string_agg} inner
     * with a non-window fdr (typically the WITHIN GROUP order-by
     * fdr admitted by the slice-30 / 35 / 36 carve-outs in
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES}) will lift
     * transparently. This matches the slice-48 → 56 posture.
     *
     * <p>Other scalar wrappers ({@code COALESCE},
     * {@code INITCAP}, {@code REVERSE}, {@code LENGTH} /
     * {@code CHAR_LENGTH}, arithmetic) stay out of scope. Each
     * carries a different argument-shape signature in the dlineage
     * XML and would need an independent probe pass before being
     * widened — slice 57 stays narrowly focused on the canonical
     * case-folding / concat / trim / substring / replace / lpad /
     * rpad / left / right / regexp_replace family.
     */
    private static final Set<String> SLICE48_DESCENT_WRAPPER_NAMES;
    static {
        Set<String> s = new HashSet<>();
        s.add("lower");
        s.add("upper");
        s.add("concat");
        s.add("trim");
        s.add("ltrim");
        s.add("rtrim");
        s.add("substring");
        s.add("substr");
        s.add("replace");
        s.add("lpad");
        s.add("rpad");
        s.add("left");
        s.add("right");
        s.add("regexp_replace");
        SLICE48_DESCENT_WRAPPER_NAMES = Collections.unmodifiableSet(s);
    }

    /**
     * Slice 48 — inner aggregate names that the {@link
     * #SLICE48_DESCENT_WRAPPER_NAMES} descent is allowed to lift.
     *
     * <p>Slice 48 introduced {@code mode}. Slice 49 widens the set to
     * also include {@code listagg} / {@code string_agg} — the natural
     * sibling lift along the same {@code LOWER(...) WITHIN GROUP (...)}
     * embedded form. Probe {@code /tmp/Probe49.java} confirmed PG /
     * Snowflake {@code LOWER(LISTAGG(x.id, ',') WG (ORDER BY x.region))}
     * dlineage XML matches the slice-48 descent shape exactly: a
     * stacked {@code RS-1.s} → {@code LOWER} → {@code LISTAGG} chain of
     * {@code resultset type="function"} elements linked by
     * {@code fdd effectType="function"} edges, plus a single
     * {@code fdr clauseType="orderby"} on the inner resultset for the
     * WG ORDER BY column. The slice-30 / 35 / 36 carve-outs in
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES} include
     * {@code listagg} and {@code string_agg}, so the defense-in-depth
     * {@link #isWindowFunctionResultset} check at the bottom of the
     * descent (which would otherwise mis-classify the orderby fdr as a
     * window discriminator) returns false for these inner names — the
     * descent is therefore trivially safe.
     *
     * <p>The extra {@code fdd effectType="function"} from
     * {@code LISTAGG.LISTAGG} → base-table column (the function
     * argument) is structurally orthogonal to the descent: the descent
     * only flips {@code result.aggregate = true} and never alters the
     * SELECT-terminal BFS, so the existing per-output SELECT lineage
     * edges are unchanged.
     *
     * <p>Plain aggregates ({@code SUM} / {@code COUNT} / etc.) inside
     * scalar wrappers stay outside the descent (slice-48 boundary
     * preserved): they are NOT in
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES} so the descent's
     * {@code isWindowFunctionResultset} carve-out would NOT apply if
     * they were added here. Lifting them needs a separate slice.
     * Hypothetical-set / {@code percentile_cont} / {@code percentile_disc}
     * also stay out for the same reason — slice 47 / 48 probes
     * confirmed their dlineage XML is structurally indistinguishable
     * from window-OVER form for the inner resultset, so widening the
     * descent allowlist would mis-classify the OVER form as aggregate.
     *
     * <p>Constraint: must be a strict subset of
     * {@link #AGGREGATE_FUNCTION_NAMES}; the runtime check below
     * mirrors the same constraint on
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES}.
     */
    private static final Set<String> SLICE48_DESCENT_INNER_NAMES;
    static {
        Set<String> s = new HashSet<>();
        s.add("mode");
        s.add("listagg");
        s.add("string_agg");
        SLICE48_DESCENT_INNER_NAMES = Collections.unmodifiableSet(s);
        if (!AGGREGATE_FUNCTION_NAMES.containsAll(SLICE48_DESCENT_INNER_NAMES)) {
            Set<String> missing = new HashSet<>(SLICE48_DESCENT_INNER_NAMES);
            missing.removeAll(AGGREGATE_FUNCTION_NAMES);
            throw new IllegalStateException(
                    "SLICE48_DESCENT_INNER_NAMES must be a subset of "
                            + "AGGREGATE_FUNCTION_NAMES; missing=" + missing);
        }
    }

    /** Policy for {@link #resolveBaseSources(String, Document, FanoutPolicy)}. */
    public enum FanoutPolicy {
        /** SELECT projection: function fdd whose argument is `*` or constant suppresses fan-out. */
        SUPPRESS_AGGREGATE_NULL_ARG,
        /** Row-influence: every fdd terminal kept regardless of intervening function nodes. */
        INCLUDE_ALL
    }

    private DlineageXmlProjector() {}

    /**
     * Legacy entry point — equivalent to {@link #project(String, EDbVendor)}
     * with {@code vendor=null}. Retained for callers that don't know or
     * don't care about the source dialect (Phase-0 harness, divergence
     * goldens, slice-7 comparison harness).
     */
    public static ProjectorResult project(String xml) {
        return project(xml, null);
    }

    /**
     * Slice 43 — vendor-scoped projection entry point. The {@code vendor}
     * parameter pins the per-vendor API contract so a future cautious
     * projector override can apply per-vendor name-whitelist exceptions
     * (e.g. PG hypothetical-set {@code rank} / {@code dense_rank} /
     * {@code percent_rank} / {@code cume_dist} where the WG form's XML
     * is structurally indistinguishable from the OVER form). Slice 43
     * deliberately makes <b>no behavior change</b> in the projector —
     * today the {@link #AGGREGATE_FUNCTION_NAMES} /
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES} sets are vendor-
     * agnostic and the {@code vendor} argument is recorded but not
     * consulted. The byte-for-byte equivalence is enforced by
     * {@code Slice43Test.projectorVendorScopeOverloadAccepts*Vendor}.
     *
     * <p>Future slices that wire vendor-specific overrides MUST keep
     * this equivalence for callers that pass {@code vendor=null} (the
     * legacy {@link #project(String)} delegates here with {@code null}).
     * The follow-up slice for PG hypothetical-set top-level admission
     * is the first expected consumer of the {@code vendor} parameter.
     *
     * @param xml dlineage XML output, must not be null/empty
     * @param vendor source SQL dialect; may be {@code null} when the
     *               caller has no vendor information
     */
    public static ProjectorResult project(String xml, EDbVendor vendor) {
        if (xml == null || xml.isEmpty()) {
            return ProjectorResult.unsupported(
                    ProjectorResult.UnsupportedReason.MALFORMED_XML, "empty xml");
        }
        Document doc;
        try {
            DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
            dbf.setNamespaceAware(false);
            // Defensive XML hardening: disable DTDs, external entities, and
            // network access by default. The projector is a public API in
            // src/main and may be called with untrusted input; matching the
            // OWASP XXE-prevention recipe for JDK DocumentBuilderFactory.
            applyXmlSecurity(dbf);
            DocumentBuilder db = dbf.newDocumentBuilder();
            // Belt-and-braces: even with the features above, refuse to
            // resolve any external entity that slips through.
            db.setEntityResolver((publicId, systemId) ->
                    new InputSource(new java.io.StringReader("")));
            db.setErrorHandler(new org.xml.sax.ErrorHandler() {
                @Override public void warning(org.xml.sax.SAXParseException e) {}
                @Override public void error(org.xml.sax.SAXParseException e) throws org.xml.sax.SAXException { throw e; }
                @Override public void fatalError(org.xml.sax.SAXParseException e) throws org.xml.sax.SAXException { throw e; }
            });
            doc = db.parse(new InputSource(new StringReader(xml)));
        } catch (Exception ex) {
            return ProjectorResult.unsupported(
                    ProjectorResult.UnsupportedReason.MALFORMED_XML, ex.getMessage());
        }

        // Identify the final resultset: exactly one <resultset> of a
        // terminal-candidate type that is never referenced as the parent
        // of any fdd OR fdr <source>. Slice 12 lifted set-op programs, so
        // {@code select_union}, {@code select_intersect}, {@code select_minus}
        // are also valid terminal types — dlineage emits one
        // {@code RESULT_OF_<OP>-N} resultset of the matching set-op type
        // whose columns each have an fdd edge with N sources (one per
        // branch's per-position {@code select_list} column). Slice 23
        // widened the parent-id collection to include fdr sources too,
        // so the inner EXISTS-body resultset (referenced only via
        // fdr clause="on") is correctly excluded from terminal candidates.
        List<Element> terminalCandidates = new ArrayList<>();
        NodeList resultsets = doc.getElementsByTagName("resultset");
        for (int i = 0; i < resultsets.getLength(); i++) {
            Element e = (Element) resultsets.item(i);
            if (isTerminalResultsetType(e.getAttribute("type"))) {
                terminalCandidates.add(e);
            }
        }
        if (terminalCandidates.isEmpty()) {
            return ProjectorResult.unsupported(
                    ProjectorResult.UnsupportedReason.NO_RELATIONSHIPS,
                    "no terminal-candidate <resultset> found "
                            + "(select_list / select_union / select_intersect / "
                            + "select_minus / select_except)");
        }
        Set<String> sourceParentIds = collectFddOrFdrSourceParentIds(doc);
        List<Element> terminals = new ArrayList<>();
        for (Element rs : terminalCandidates) {
            if (!sourceParentIds.contains(rs.getAttribute("id"))) {
                terminals.add(rs);
            }
        }
        if (terminals.size() != 1) {
            return ProjectorResult.unsupported(
                    ProjectorResult.UnsupportedReason.MULTIPLE_TERMINAL_SELECTS,
                    "expected exactly one terminal resultset, got " + terminals.size());
        }
        Element finalResultset = terminals.get(0);

        Set<CanonicalLineageEdge> edges = new LinkedHashSet<>();
        Set<String> outputNames = new LinkedHashSet<>();
        Map<String, Boolean> aggregateByOutput = new LinkedHashMap<>();

        // 1) SELECT projection per output column of the final resultset.
        // Skip system pseudo-columns (e.g. RelationRows). They are the implicit
        // fdr-target for row-influence, never user-visible projected columns.
        NodeList finalCols = finalResultset.getElementsByTagName("column");
        for (int i = 0; i < finalCols.getLength(); i++) {
            Element col = (Element) finalCols.item(i);
            // Only direct children of finalResultset count — getElementsByTagName
            // returns descendants too in theory, but our XML schema has columns
            // as direct children of resultset/table, so this is fine in practice.
            if (col.getParentNode() != finalResultset) continue;
            if ("system".equals(col.getAttribute("source"))) continue;

            String outName = col.getAttribute("name").toLowerCase(Locale.ROOT);
            outputNames.add(outName);
            // Default false; flipped to true if BFS visits a function resultset.
            aggregateByOutput.put(outName, false);

            BfsResult br = projectColumn(col.getAttribute("id"), doc,
                    FanoutPolicy.SUPPRESS_AGGREGATE_NULL_ARG);
            if (br.aggregate) {
                aggregateByOutput.put(outName, true);
            }
            for (TableColumn tc : br.terminals) {
                edges.add(new CanonicalLineageEdge(EdgeRole.SELECT, outName, tc.table, tc.column));
            }
        }

        // 2) Row-influence from fdr edges whose target is a column of the final
        // resultset (typically the system RelationRows column).
        String finalRsId = finalResultset.getAttribute("id");
        // Walk the fdr graph: follow chains that target the final resultset's
        // RelationRows, including indirect ones (e.g. RS-1.RelationRows -> sub.RelationRows
        // in 04_nested_subquery). For each fdr source with clauseType where/joinCondition
        // (or clause where/on), resolve to base columns via fdd closure.
        NodeList fdrs = doc.getElementsByTagName("relationship");
        // Collect fdr edges keyed by the parent_id of the target so we can
        // chain RelationRows-targeting fdrs.
        Map<String, List<Element>> fdrsByTargetParentId = new HashMap<>();
        for (int i = 0; i < fdrs.getLength(); i++) {
            Element rel = (Element) fdrs.item(i);
            if (!"fdr".equals(rel.getAttribute("type"))) continue;
            Element target = firstChildElement(rel, "target");
            if (target == null) continue;
            fdrsByTargetParentId.computeIfAbsent(target.getAttribute("parent_id"), k -> new ArrayList<>())
                    .add(rel);
        }

        Deque<String> rowSetParents = new ArrayDeque<>();
        Set<String> visitedRowSetParents = new HashSet<>();
        rowSetParents.add(finalRsId);
        visitedRowSetParents.add(finalRsId);
        while (!rowSetParents.isEmpty()) {
            String parentId = rowSetParents.removeFirst();
            List<Element> rels = fdrsByTargetParentId.get(parentId);
            if (rels == null) continue;
            for (Element rel : rels) {
                Element target = firstChildElement(rel, "target");
                if (target == null) continue;
                // Only follow fdrs whose target is a system column (RelationRows).
                // Other fdrs (e.g. effectType="function" with target on a function
                // resultset) carry aggregation semantics, not row-influence.
                if (!"system".equals(target.getAttribute("source"))) continue;
                // Process each <source> on this fdr.
                NodeList sources = rel.getElementsByTagName("source");
                for (int si = 0; si < sources.getLength(); si++) {
                    Element src = (Element) sources.item(si);
                    if (src.getParentNode() != rel) continue;
                    String clauseType = src.getAttribute("clauseType");
                    String relClause = rel.getAttribute("clause");
                    EdgeRole role = clauseTypeToRole(clauseType, relClause);
                    if (role == null) {
                        // Source is itself a system RelationRows pointing at
                        // another resultset — chain through it.
                        if ("system".equals(src.getAttribute("source"))) {
                            String nextParent = src.getAttribute("parent_id");
                            if (visitedRowSetParents.add(nextParent)) {
                                rowSetParents.add(nextParent);
                            }
                        }
                        continue;
                    }
                    // Source identifies a column on a base table or a non-base
                    // resultset. Resolve via fdd closure with INCLUDE_ALL so
                    // function-arg suppression doesn't skew row influence.
                    String srcParentId = src.getAttribute("parent_id");
                    Element parent = elementById(doc, srcParentId);
                    if (parent == null) continue;
                    if (isBaseTable(parent)) {
                        // Skip constantTable for row influence too.
                        if ("constantTable".equals(parent.getAttribute("type"))) continue;
                        String col = src.getAttribute("column");
                        if ("*".equals(col)) continue;
                        edges.add(new CanonicalLineageEdge(role, null,
                                parent.getAttribute("name").toLowerCase(Locale.ROOT),
                                col.toLowerCase(Locale.ROOT)));
                    } else if (isResultset(parent)) {
                        String srcId = src.getAttribute("id");
                        BfsResult inner = resolveBaseSources(srcId, doc, FanoutPolicy.INCLUDE_ALL);
                        for (TableColumn tc : inner.terminals) {
                            edges.add(new CanonicalLineageEdge(role, null, tc.table, tc.column));
                        }
                    } // else: constantTable / function — already filtered
                }
            }
        }

        return ProjectorResult.ok(
                new CanonicalLineageModel(edges, outputNames, aggregateByOutput));
    }

    /**
     * Project one final-resultset output column. The aggregate flag is set
     * <b>only when an immediate fdd source of this column lives on a
     * function resultset whose function name is in
     * {@link #AGGREGATE_FUNCTION_NAMES}</b>. This matches Semantic IR's
     * per-output (non-transitive) aggregate semantics — see slice-6
     * design notes on flag locality. The BFS itself walks transitively
     * to collect base-table SELECT terminals.
     */
    private static BfsResult projectColumn(String columnId, Document doc, FanoutPolicy policy) {
        BfsResult result = new BfsResult();

        // First-hop aggregate detection. Local rule: this output is aggregate
        // iff *any* immediate fdd source lives on a function resultset whose
        // function name is in AGGREGATE_FUNCTION_NAMES AND the function
        // resultset is NOT a window function (slice 13). We must scan all
        // immediate function sources, not stop at the first one — a fdd whose
        // sources include both `UPPER(...)` and `SUM(...)` columns must still
        // mark the output aggregate.
        //
        // Slice 13 window-vs-aggregate discriminator: a function resultset is
        // windowed iff at least one fdr targeting it carries a source with
        // clauseType="selectList" (PARTITION BY) or clauseType="orderby"
        // (OVER ORDER BY). Plain aggregates have only the RelationRows fdr;
        // window functions add the PARTITION BY / OVER ORDER BY fdrs. Empty
        // OVER () is not discriminable in dlineage XML (looks identical to a
        // plain aggregate); the IR builder rejects empty OVER () to avoid
        // manufacturing AGGREGATION_MISMATCH divergences.
        outer:
        for (Element fdd : findFddByTarget(doc, columnId)) {
            for (Element fnRs : immediateFunctionSourceResultsets(fdd, doc)) {
                String fnName = fnRs.getAttribute("name");
                if (fnName == null || fnName.isEmpty()) continue;
                if (!AGGREGATE_FUNCTION_NAMES.contains(fnName.toLowerCase(Locale.ROOT))) continue;
                if (isWindowFunctionResultset(fnRs, doc)) continue;
                result.aggregate = true;
                break outer;
            }
        }

        // Slice 48 — single-step descent through a known scalar wrapper.
        // The first-hop check above mirrors the IR builder's per-output
        // (non-transitive) aggregate semantics, but for a small set of
        // case-folding scalar wrappers used to format the output of an
        // ordered-set aggregate, the IR side classifies the projection
        // as aggregate via the descendant ordered-set name. The
        // projector must mirror that classification or the divergence
        // harness reports a pre-existing
        // {@link DivergenceClass#AGGREGATION_MISMATCH}.
        //
        // Lift surface as of slice 57:
        //   wrapper-name ∈ SLICE48_DESCENT_WRAPPER_NAMES
        //                = {lower, upper, concat, trim, ltrim, rtrim,
        //                   substring, substr, replace, lpad, rpad,
        //                   left, right, regexp_replace}
        //   ∧ inner-name ∈ SLICE48_DESCENT_INNER_NAMES
        //                = {mode, listagg, string_agg}
        //   ∧ !isWindowFunctionResultset(inner).
        //
        // History:
        //   - slice 48 introduced the descent with wrapper={lower},
        //     inner={mode}.
        //   - slice 49 widened inner to {mode, listagg, string_agg}.
        //   - slice 50 widened wrapper to {lower, upper}.
        //   - slice 51 widened wrapper to {lower, upper, concat}.
        //   - slice 52 widened wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim}.
        //   - slice 53 widened wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim, substring,
        //      substr}.
        //   - slice 54 widened wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim, substring,
        //      substr, replace}.
        //   - slice 55 widened wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim, substring,
        //      substr, replace, lpad, rpad}.
        //   - slice 56 widened wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim, substring,
        //      substr, replace, lpad, rpad, left, right}.
        //   - slice 57 widens wrapper to
        //     {lower, upper, concat, trim, ltrim, rtrim, substring,
        //      substr, replace, lpad, rpad, left, right,
        //      regexp_replace}.
        // The {@code ||} string-concatenation operator does NOT emit a
        // wrapper resultset on PG / Snowflake (the parser folds
        // {@code expr || ' '} into the projection root directly), so no
        // descent is needed for that shape — the existing first-hop
        // aggregate detection already handles it. Other scalar wrappers
        // ({@code COALESCE}, {@code INITCAP}, {@code REVERSE},
        // {@code LENGTH} / {@code CHAR_LENGTH}, arithmetic) stay
        // outside the descent — their argument-shape signatures
        // differ and require independent probes.
        //
        // The descent is also one-level only — {@code LOWER(LOWER(mode WG))}
        // remains divergent because the wrapper's immediate function
        // source is {@code lower}, not {@code mode}. This keeps the
        // descent O(immediate-fdds) and avoids any risk of
        // double-counting through arbitrarily deep wrappings.
        if (!result.aggregate) {
            // The first-hop loop above scans the column's fdds for an
            // immediate function-resultset source whose own column was
            // the projection's value (e.g. {@code RS-1.s} fdd-source
            // {@code LOWER.LOWER}). To descend one level we walk through
            // each scalar-wrapper source's <i>column</i> and look at
            // that column's own fdd targets — the wrapper-internal fdd
            // (e.g. fdd targeting {@code LOWER.LOWER} with source
            // {@code mode.mode}). The {@code findFddByTarget} helper
            // matches on the target column's {@code id}, NOT the
            // resultset id, so we must look up the wrapper's child
            // column id (slice-48 descent's distinguishing detail vs
            // the existing first-hop check).
            outerDescent:
            for (Element fdd : findFddByTarget(doc, columnId)) {
                NodeList sources = fdd.getElementsByTagName("source");
                for (int si = 0; si < sources.getLength(); si++) {
                    Element src = (Element) sources.item(si);
                    if (src.getParentNode() != fdd) continue;
                    String wrapperParentId = src.getAttribute("parent_id");
                    Element wrapperRs = elementById(doc, wrapperParentId);
                    if (wrapperRs == null
                            || !isResultset(wrapperRs)
                            || !"function".equals(wrapperRs.getAttribute("type"))) continue;
                    String wrapperName = wrapperRs.getAttribute("name");
                    if (wrapperName == null) continue;
                    if (!SLICE48_DESCENT_WRAPPER_NAMES.contains(
                            wrapperName.toLowerCase(Locale.ROOT))) continue;
                    String wrapperColumnId = src.getAttribute("id");
                    if (wrapperColumnId == null || wrapperColumnId.isEmpty()) continue;
                    for (Element wrapperFdd : findFddByTarget(doc, wrapperColumnId)) {
                        for (Element innerRs : immediateFunctionSourceResultsets(wrapperFdd, doc)) {
                            String innerName = innerRs.getAttribute("name");
                            if (innerName == null) continue;
                            if (!SLICE48_DESCENT_INNER_NAMES.contains(
                                    innerName.toLowerCase(Locale.ROOT))) continue;
                            // Defense-in-depth: even though `mode` is in
                            // ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES (so
                            // an `orderby`-only fdr is not classified as
                            // windowed), a `selectList` fdr (PARTITION
                            // BY) on the inner resultset still indicates
                            // the OVER form. Skip those PARTITION-BY
                            // windowed inners so the slice-48 descent does
                            // not lift that OVER projection. The known
                            // OVER-only-ORDER-BY carve-out remains
                            // classified as plain aggregate on the
                            // projector side; IR rejects that form, and
                            // Slice48Test documents the limitation.
                            if (isWindowFunctionResultset(innerRs, doc)) continue;
                            result.aggregate = true;
                            break outerDescent;
                        }
                    }
                }
            }
        }

        // BFS for SELECT terminals. The aggregate flag is no longer touched
        // inside the BFS; the only thing the function nodes still control is
        // SELECT-fan-out suppression for COUNT(*)-shaped calls.
        Set<String> visited = new HashSet<>();
        Deque<String> q = new ArrayDeque<>();
        q.add(columnId);
        visited.add(columnId);
        while (!q.isEmpty()) {
            String cur = q.removeFirst();
            for (Element fdd : findFddByTarget(doc, cur)) {
                Element targetEl = firstChildElement(fdd, "target");
                if (targetEl == null) continue;
                Element targetParent = elementById(doc, targetEl.getAttribute("parent_id"));
                boolean targetIsFunction = targetParent != null
                        && isResultset(targetParent)
                        && "function".equals(targetParent.getAttribute("type"));
                if (targetIsFunction
                        && policy == FanoutPolicy.SUPPRESS_AGGREGATE_NULL_ARG
                        && shouldSuppressFunctionFanout(fdd, targetParent)) {
                    continue;
                }
                NodeList sources = fdd.getElementsByTagName("source");
                for (int si = 0; si < sources.getLength(); si++) {
                    Element src = (Element) sources.item(si);
                    if (src.getParentNode() != fdd) continue;
                    String col = src.getAttribute("column");
                    if ("*".equals(col)) continue;
                    String parentId = src.getAttribute("parent_id");
                    Element parent = elementById(doc, parentId);
                    if (parent == null) continue;
                    if (isBaseTable(parent)) {
                        if ("constantTable".equals(parent.getAttribute("type"))) continue;
                        result.terminals.add(new TableColumn(
                                parent.getAttribute("name").toLowerCase(Locale.ROOT),
                                col.toLowerCase(Locale.ROOT)));
                    } else if (isResultset(parent)) {
                        String nextId = src.getAttribute("id");
                        if (visited.add(nextId)) {
                            q.add(nextId);
                        }
                    }
                    // constantTable: skipped already; function resultset
                    // descent is handled when nextId is dequeued and its fdd
                    // is matched above.
                }
            }
        }
        return result;
    }

    /**
     * Public-ish helper used by the row-influence path. Same BFS as
     * {@link #projectColumn} but the aggregate flag is meaningless here.
     */
    static BfsResult resolveBaseSources(String columnId, Document doc, FanoutPolicy policy) {
        return projectColumn(columnId, doc, policy);
    }

    /**
     * Return every function-resultset name that appears as the parent of an
     * immediate {@code <source>} on this fdd. Multiple are possible when an
     * expression projects something like {@code UPPER(name) || SUM(salary)}
     * and dlineage produces sources from both function nodes.
     */
    private static List<String> immediateFunctionSourceNames(Element fdd, Document doc) {
        List<String> names = new ArrayList<>();
        NodeList sources = fdd.getElementsByTagName("source");
        for (int si = 0; si < sources.getLength(); si++) {
            Element src = (Element) sources.item(si);
            if (src.getParentNode() != fdd) continue;
            Element parent = elementById(doc, src.getAttribute("parent_id"));
            if (parent != null && isResultset(parent)
                    && "function".equals(parent.getAttribute("type"))) {
                String name = parent.getAttribute("name");
                if (name != null && !name.isEmpty()) names.add(name);
            }
        }
        return names;
    }

    /**
     * Slice 13: parallel to {@link #immediateFunctionSourceNames} but
     * returns the function-resultset {@link Element}s themselves so the
     * caller can inspect them with {@link #isWindowFunctionResultset}.
     */
    private static List<Element> immediateFunctionSourceResultsets(Element fdd, Document doc) {
        List<Element> els = new ArrayList<>();
        NodeList sources = fdd.getElementsByTagName("source");
        for (int si = 0; si < sources.getLength(); si++) {
            Element src = (Element) sources.item(si);
            if (src.getParentNode() != fdd) continue;
            Element parent = elementById(doc, src.getAttribute("parent_id"));
            if (parent != null && isResultset(parent)
                    && "function".equals(parent.getAttribute("type"))) {
                els.add(parent);
            }
        }
        return els;
    }

    /**
     * Slice 13: window-vs-aggregate discriminator. A function resultset is
     * windowed iff it has at least one {@code fdr} edge whose target's
     * {@code parent_id} equals the function resultset's id AND whose source
     * carries {@code clauseType="selectList"} (PARTITION BY) or
     * {@code clauseType="orderby"} (OVER ORDER BY). Plain (non-windowed)
     * aggregates have only the RelationRows fdr; the addition of PARTITION
     * BY / OVER ORDER BY fdrs is dlineage's way of recording the analytic
     * dependencies of a windowed call.
     *
     * <p>Empty {@code OVER ()} window functions are NOT discriminable here
     * (their XML is byte-identical to a plain aggregate). Slice 13's IR
     * builder rejects empty {@code OVER ()} so the divergence harness
     * never sees that case in practice.
     *
     * <p>Probed against Oracle (slice-13 corpus); cross-vendor parity
     * (PostgreSQL, BigQuery, SparkSQL) is unproven and deferred.
     *
     * <p>Slice 35 override: function names in
     * {@link #ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES} suppress an
     * order-by-only discriminator. PostgreSQL emits a
     * {@code clauseType="orderby"} fdr for the WITHIN GROUP ORDER BY of
     * {@code LISTAGG(... ) WITHIN GROUP (...)}, which the slice-13
     * discriminator would otherwise mis-classify as windowed. A
     * {@code selectList} fdr still wins and keeps analytic/window shapes
     * classified as windowed.
     */
    private static boolean isWindowFunctionResultset(Element fnRs, Document doc) {
        if (fnRs == null) return false;
        String fnName = fnRs.getAttribute("name");
        boolean orderByWithinGroupCandidate = fnName != null
                && ORDER_BY_WITHIN_GROUP_AGGREGATE_NAMES.contains(
                        fnName.toLowerCase(Locale.ROOT));
        String fnRsId = fnRs.getAttribute("id");
        if (fnRsId == null || fnRsId.isEmpty()) return false;
        boolean sawOrderByDiscriminator = false;
        NodeList rels = doc.getElementsByTagName("relationship");
        for (int i = 0; i < rels.getLength(); i++) {
            Element rel = (Element) rels.item(i);
            if (!"fdr".equals(rel.getAttribute("type"))) continue;
            Element target = firstChildElement(rel, "target");
            if (target == null) continue;
            if (!fnRsId.equals(target.getAttribute("parent_id"))) continue;
            NodeList sources = rel.getElementsByTagName("source");
            for (int si = 0; si < sources.getLength(); si++) {
                Element src = (Element) sources.item(si);
                if (src.getParentNode() != rel) continue;
                String ct = src.getAttribute("clauseType");
                if ("selectList".equals(ct)) return true;
                if ("orderby".equals(ct)) sawOrderByDiscriminator = true;
            }
        }
        return sawOrderByDiscriminator && !orderByWithinGroupCandidate;
    }

    /**
     * Decide whether the SELECT fan-out from a function-targeted fdd should
     * be suppressed. Two conditions must hold:
     *
     * <ol>
     *   <li>The function name is in {@link #NULL_ARG_AGGREGATES}. Today
     *       only {@code COUNT}; this prevents a hypothetical
     *       {@code SUM(t.*)} (some dialects allow it) from silently losing
     *       lineage.</li>
     *   <li>Either at least one source has {@code column="*"} (the
     *       {@code COUNT(*)} shape) or every source is a constant-table
     *       column ({@code COUNT(1)} / {@code COUNT(literal)}).</li>
     * </ol>
     */
    private static boolean shouldSuppressFunctionFanout(Element functionFdd, Element functionResultset) {
        if (functionResultset == null) return false;
        String fnName = functionResultset.getAttribute("name");
        if (fnName == null || fnName.isEmpty()) return false;
        if (!NULL_ARG_AGGREGATES.contains(fnName.toLowerCase(Locale.ROOT))) return false;

        boolean anyStar = false;
        boolean anyNonConstant = false;
        boolean anySource = false;
        NodeList sources = functionFdd.getElementsByTagName("source");
        for (int si = 0; si < sources.getLength(); si++) {
            Element src = (Element) sources.item(si);
            if (src.getParentNode() != functionFdd) continue;
            anySource = true;
            if ("*".equals(src.getAttribute("column"))) {
                anyStar = true;
                continue;
            }
            String parentId = src.getAttribute("parent_id");
            Element parent = elementById(functionFdd.getOwnerDocument(), parentId);
            if (parent != null && isBaseTable(parent)
                    && "constantTable".equals(parent.getAttribute("type"))) {
                continue;
            }
            anyNonConstant = true;
        }
        if (!anySource) return false;
        if (anyStar) return true;
        return !anyNonConstant;
    }

    private static List<Element> findFddByTarget(Document doc, String columnId) {
        List<Element> out = new ArrayList<>();
        NodeList rels = doc.getElementsByTagName("relationship");
        for (int i = 0; i < rels.getLength(); i++) {
            Element rel = (Element) rels.item(i);
            if (!"fdd".equals(rel.getAttribute("type"))) continue;
            Element target = firstChildElement(rel, "target");
            if (target == null) continue;
            if (columnId.equals(target.getAttribute("id"))) {
                out.add(rel);
            }
        }
        return out;
    }

    /**
     * Collect parent IDs of fdd <b>and</b> fdr sources. The terminal-resultset
     * detection treats any resultset that is referenced as a source of either
     * data-dependency (fdd) <b>or</b> row-influence (fdr) as non-terminal.
     *
     * <p>Slice 23 motivation: an EXISTS subquery in JOIN ON contributes its
     * inner SELECT-list resultset only via an {@code fdr clause="on"} edge
     * (the inner result is plumbed into outer's RelationRows as a row-
     * influence source, NOT as a data dependency on any outer-projected
     * column). Without considering fdr sources, both the outer SELECT-list
     * resultset AND the inner EXISTS-body resultset would qualify as
     * terminal candidates, producing a false MULTIPLE_TERMINAL_SELECTS.
     *
     * <p>Pre-slice-23 corpus entries (FROM-subquery, scalar-subquery,
     * set-op) all referenced inner resultsets via fdd, so the original
     * fdd-only check sufficed for them; widening to fdr is additive and
     * cannot mis-classify their inner resultsets as referenced when they
     * weren't already.
     */
    private static Set<String> collectFddOrFdrSourceParentIds(Document doc) {
        Set<String> out = new HashSet<>();
        NodeList rels = doc.getElementsByTagName("relationship");
        for (int i = 0; i < rels.getLength(); i++) {
            Element rel = (Element) rels.item(i);
            String relType = rel.getAttribute("type");
            if (!"fdd".equals(relType) && !"fdr".equals(relType)) continue;
            NodeList sources = rel.getElementsByTagName("source");
            for (int si = 0; si < sources.getLength(); si++) {
                Element src = (Element) sources.item(si);
                if (src.getParentNode() != rel) continue;
                String parentId = src.getAttribute("parent_id");
                if (parentId != null && !parentId.isEmpty()) {
                    out.add(parentId);
                }
            }
        }
        return out;
    }

    /**
     * Apply OWASP-recommended XXE-prevention features to a
     * {@link DocumentBuilderFactory}. Each set is best-effort: a parser
     * that does not expose a feature swallows the failure but the others
     * still apply.
     */
    private static void applyXmlSecurity(DocumentBuilderFactory dbf) {
        String[] disable = {
                // Disallow inline DTDs entirely — strongest mitigation.
                "http://apache.org/xml/features/disallow-doctype-decl",
        };
        String[] enable = {
                javax.xml.XMLConstants.FEATURE_SECURE_PROCESSING,
        };
        String[] disableExt = {
                "http://xml.org/sax/features/external-general-entities",
                "http://xml.org/sax/features/external-parameter-entities",
                "http://apache.org/xml/features/nonvalidating/load-external-dtd",
        };
        for (String f : disable) {
            try { dbf.setFeature(f, true); } catch (Exception ignore) {}
        }
        for (String f : enable) {
            try { dbf.setFeature(f, true); } catch (Exception ignore) {}
        }
        for (String f : disableExt) {
            try { dbf.setFeature(f, false); } catch (Exception ignore) {}
        }
        try {
            dbf.setAttribute(javax.xml.XMLConstants.ACCESS_EXTERNAL_DTD, "");
        } catch (Exception ignore) {}
        try {
            dbf.setAttribute(javax.xml.XMLConstants.ACCESS_EXTERNAL_SCHEMA, "");
        } catch (Exception ignore) {}
        dbf.setXIncludeAware(false);
        dbf.setExpandEntityReferences(false);
    }

    private static Element elementById(Document doc, String id) {
        if (id == null || id.isEmpty()) return null;
        // The XML uses string ids on table/resultset/column elements; not
        // declared as XML IDs, so we lookup by attribute.
        for (String tag : new String[]{"table", "resultset", "column"}) {
            NodeList nl = doc.getElementsByTagName(tag);
            for (int i = 0; i < nl.getLength(); i++) {
                Element e = (Element) nl.item(i);
                if (id.equals(e.getAttribute("id"))) return e;
            }
        }
        return null;
    }

    private static Element firstChildElement(Element parent, String tag) {
        NodeList nl = parent.getChildNodes();
        for (int i = 0; i < nl.getLength(); i++) {
            Node n = nl.item(i);
            if (n.getNodeType() != Node.ELEMENT_NODE) continue;
            if (tag.equals(n.getNodeName())) return (Element) n;
        }
        return null;
    }

    private static boolean isBaseTable(Element parent) {
        return "table".equals(parent.getTagName());
    }

    private static boolean isResultset(Element parent) {
        return "resultset".equals(parent.getTagName());
    }

    /**
     * True iff the resultset {@code type} is a candidate to be the
     * program's terminal resultset. {@code select_list} is the standard
     * shape (slice 1+). Slice 12 added the set-op result types
     * ({@code select_union}, {@code select_intersect}, {@code select_minus},
     * {@code select_except}): dlineage emits one {@code RESULT_OF_<OP>-N}
     * resultset of the matching type per top-level set-op program; that
     * resultset's columns have one fdd edge each with N sources (one per
     * branch's per-position {@code select_list} column), and the existing
     * {@code projectColumn} BFS walks those fdd chains transparently to
     * reach base-table terminals. {@code select_except} (PostgreSQL /
     * SQL Server EXCEPT) is distinct from {@code select_minus} (Oracle /
     * Spark / Hive MINUS) at the dlineage layer even though they are
     * semantically equivalent.
     */
    private static boolean isTerminalResultsetType(String type) {
        if (type == null) return false;
        return "select_list".equals(type)
                || "select_union".equals(type)
                || "select_intersect".equals(type)
                || "select_minus".equals(type)
                || "select_except".equals(type);
    }

    private static EdgeRole clauseTypeToRole(String clauseType, String relClause) {
        if ("where".equals(clauseType) || "where".equals(relClause)) return EdgeRole.FILTER;
        if ("joinCondition".equals(clauseType) || "on".equals(relClause)) return EdgeRole.JOIN;
        return null;
    }

    /** Result of projecting one column: terminals plus the aggregate flag. */
    static final class BfsResult {
        final List<TableColumn> terminals = new ArrayList<>();
        boolean aggregate = false;
    }

    /** Lower-cased base (table, column) pair. */
    static final class TableColumn {
        final String table;
        final String column;
        TableColumn(String table, String column) {
            this.table = table;
            this.column = column;
        }
    }
}