package gudusoft.gsqlparser.lineage.v2.merge;

import gudusoft.gsqlparser.lineage.v2.json.ArrayRegistry;
import gudusoft.gsqlparser.lineage.v2.json.CanonicalJson;
import gudusoft.gsqlparser.lineage.v2.model.Document;
import gudusoft.gsqlparser.lineage.v2.model.Producer;
import gudusoft.gsqlparser.lineage.v2.validation.DocumentIndex;
import gudusoft.gsqlparser.lineage.v2.validation.LineageValidator;
import gudusoft.gsqlparser.lineage.v2.validation.SchemaValidator;
import gudusoft.gsqlparser.lineage.v2.validation.SourceTextProvider;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
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.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import static gudusoft.gsqlparser.lineage.v2.validation.DocumentIndex.intRef;
import static gudusoft.gsqlparser.lineage.v2.validation.DocumentIndex.items;

/**
 * The minimal L1 merger of milestone M1 (specification Sections 9 and 15.2; contract: section 7 of
 * {@code docs/designs/data-lineage/05-sqllr-m1-core-library-plan-zh.md}).
 *
 * <p>Each input must be a valid document, and so is the result: a merge that would produce an invalid document,
 * or that cannot keep what an input asserts, is refused with a {@link MergeException}; nothing is dropped
 * silently.</p>
 *
 * <p>Source units merge by (unit id, analysis context id). The content of such a unit (statements, scopes,
 * Occurrences, writes, diagnostics, ExplainGraph content, module objects) is carried from one input whose content
 * includes every other input's, compared with references replaced by identities; a {@code LOCAL} unit is carried
 * once per input. Endpoints merge by portable identity, Facts by {@code FactIdV1}, Occurrences by record id,
 * analysis contexts and objects without an identity that no unit owns by their content. References are renumbered
 * in the order of the inputs.</p>
 *
 * <p>The merge module of the specification is not defined yet (Section 20, item 16), so the input digests and the
 * ledger are returned in the {@link Result}, not written into the document. Because the document cannot yet carry
 * two endpoints of one identity, a kind difference other than a generic {@code RELATION} refined by a specific
 * relation kind is refused (Section 9.4 asks to keep both).</p>
 *
 * @since 4.2.17
 */
public final class LineageMerger {

    /** The inputs cannot be merged by this merger; nothing was merged. */
    public static class MergeException extends RuntimeException {
        private static final long serialVersionUID = 1L;

        public MergeException(String message) {
            super(message);
        }
    }

    /** One entry of the ledger: an identity whose generic kind an input refined. */
    public static final class Conflict {
        private final String identity;
        private final String detail;

        Conflict(String identity, String detail) {
            this.identity = identity;
            this.detail = detail;
        }

        /** The identity (an {@code ep1} id). */
        public String getIdentity() {
            return identity;
        }

        public String getDetail() {
            return detail;
        }

        @Override
        public String toString() {
            return identity + ": " + detail;
        }
    }

    /** The merged document and what the merge module would carry. */
    public static final class Result {
        private final Document document;
        private final List<String> inputDigests;
        private final List<Conflict> conflicts;
        private final List<String> notes;

        Result(Document document, List<String> inputDigests, List<Conflict> conflicts, List<String> notes) {
            this.document = document;
            this.inputDigests = Collections.unmodifiableList(inputDigests);
            this.conflicts = Collections.unmodifiableList(conflicts);
            this.notes = Collections.unmodifiableList(notes);
        }

        public Document getDocument() {
            return document;
        }

        /** The sha256 of each distinct input's canonical bytes, in the order the inputs were first given. */
        public List<String> getInputDigests() {
            return inputDigests;
        }

        /** The ledger: identities whose generic kind an input refined. */
        public List<Conflict> getConflicts() {
            return conflicts;
        }

        public List<String> getNotes() {
            return notes;
        }
    }

    private static final List<String> LEVELS = Arrays.asList("L1", "L2", "L3");
    /** The specific endpoint kinds of identity class RELATION (identity-v1, Section 5.2.1). */
    private static final Set<String> RELATION_KINDS = new HashSet<String>(Arrays.asList("TABLE", "VIEW",
            "MATERIALIZED_VIEW", "EXTERNAL_TABLE", "SYNONYM"));
    /** The root members the merger carries, besides the module arrays. */
    private static final Set<String> ROOT = new HashSet<String>(Arrays.asList("specUri", "specVersion",
            "declaredLevel", "modules", "header", "analysisContexts", "sourceUnits", "statements", "scopes", "endpoints",
            "facts", "factCandidates", "candidateSets", "occurrences", "targetFieldOccurrences", "analysisCompleteness",
            "featureCompleteness", "servingCompleteness", "diagnostics", "explainGraph"));
    private static final Set<String> GRAPH = new HashSet<String>(Arrays.asList("completeness", "statementCompleteness",
            "nodes", "microEdges", "candidateSets", "supportPaths"));
    /** The collections of the output, by kind of object. */
    private static final String[][] COLLECTIONS = {
            {"analysisContext", "analysisContexts"}, {"sourceUnit", "sourceUnits"}, {"statement", "statements"},
            {"scope", "scopes"}, {"endpoint", "endpoints"}, {"fact", "facts"}, {"factCandidate", "factCandidates"},
            {"rootCandidateSet", "candidateSets"}, {"occurrence", "occurrences"}, {"tfo", "targetFieldOccurrences"}, {"diagnostic", "diagnostics"},
            {"node", "nodes"}, {"microEdge", "microEdges"}, {"candidateSet", "candidateSets"},
            {"supportPath", "supportPaths"}};
    private static final Set<String> GRAPH_KINDS = new HashSet<String>(Arrays.asList("node", "microEdge",
            "candidateSet", "supportPath"));
    /** Members that modules add to objects of the core (registry group modules); a carrier may add them. */
    private static final Set<String> MODULE_MEMBERS_OF_OBJECTS = new HashSet<String>(Semantics.MODULE_ADDED);
    /** The kinds whose objects belong to the unit of their statement or source unit. */
    private static final Set<String> UNIT_CONTENT = new HashSet<String>(Arrays.asList("statement", "scope",
            "occurrence", "tfo", "diagnostic", "node", "microEdge", "endpoint", "rootCandidateSet", "supportPath",
            "candidateSet"));

    private final Producer producer;
    private final String createdAt;
    private final LineageValidator validator;

    /**
     * @param producer  the producer of the merged document: the merger; must be a valid {@code Producer}
     * @param createdAt the {@code header.createdAt} of the merged document, or null for none; never read from a clock
     */
    public LineageMerger(Producer producer, String createdAt) {
        this(producer, createdAt, LineageValidator.standard(SourceTextProvider.NONE));
    }

    public LineageMerger(Producer producer, String createdAt, LineageValidator validator) {
        if (producer == null || !SchemaValidator.lineageDocument().isValid(producer.toJson(), "Producer")) {
            throw new IllegalArgumentException("the merged document names the merger as its producer, which must be"
                    + " a valid Producer: " + producer);
        }
        this.producer = producer;
        this.createdAt = createdAt;
        this.validator = validator;
    }

    public Result merge(List<Document> inputs) {
        if (inputs == null || inputs.isEmpty()) {
            throw new MergeException("nothing to merge");
        }
        List<byte[]> distinct = new ArrayList<byte[]>();
        List<String> digests = new ArrayList<String>();
        for (Document input : inputs) {
            byte[] bytes = input.toBytes();
            String digest = sha256(bytes);
            if (!digests.contains(digest)) {
                digests.add(digest);
                distinct.add(bytes);
            }
        }
        for (int i = 0; i < distinct.size(); i++) {
            LineageValidator.Result check = validator.validate(distinct.get(i));
            if (!check.isValid()) {
                throw new MergeException("input " + i + " is not valid: " + check.getStage() + " " + check.getCodes());
            }
        }
        Result result = new Run(distinct).run(digests);
        LineageValidator.Result check = validator.validate(result.getDocument().toBytes());
        if (!check.isValid()) {
            throw new MergeException("the merged document would not be valid: " + check.getStage() + " "
                    + check.getFindings());
        }
        return result;
    }

    static String sha256(byte[] data) {
        try {
            StringBuilder out = new StringBuilder();
            for (byte b : MessageDigest.getInstance("SHA-256").digest(data)) {
                out.append(Character.forDigit((b >> 4) & 0xf, 16)).append(Character.forDigit(b & 0xf, 16));
            }
            return out.toString();
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException(e);
        }
    }

    /** Analysis completeness of a merge (Section 4.4): one status when all agree, otherwise PARTIAL. */
    static Object combined(Object a, Object b) {
        return a == null ? b : (a.equals(b) ? a : "PARTIAL");
    }

    // ---------------------------------------------------------------------------------------------

    private final class Run {
        private final List<byte[]> bytes;
        private final List<Map<String, Object>> docs = new ArrayList<Map<String, Object>>();
        private final List<Semantics> semantics = new ArrayList<Semantics>();
        private final List<DocumentIndex> indexes = new ArrayList<DocumentIndex>();
        private final List<Map<Long, Long>> remaps = new ArrayList<Map<Long, Long>>();
        /** For each input, the ref of every keyed object by key, and of every object by kind and form. */
        private final List<Map<String, Long>> keyed = new ArrayList<Map<String, Long>>();
        private final List<Map<String, Long>> byForm = new ArrayList<Map<String, Long>>();
        /** For each input, the refs of the candidate sets of its ExplainGraph (not those inside FactCandidates). */
        private final List<Set<Long>> graphCsets = new ArrayList<Set<Long>>();
        /** For each input, the FactCandidate and end ({@code source}, {@code target}) of each embedded candidate set. */
        private final List<Map<Long, Object[]>> embeddedIn = new ArrayList<Map<Long, Object[]>>();
        /** The new ref of the candidate set at each end of each carried FactCandidate ("new FactCandidate ref|end"). */
        private final Map<String, Long> embeddedSlots = new HashMap<String, Long>();
        /** The input that carries the content of each (unit id, analysis context id). */
        private final Map<String, Integer> carrier = new HashMap<String, Integer>();
        private final Map<String, Long> byKey = new HashMap<String, Long>();
        private final Map<String, TreeMap<Long, Map<String, Object>>> out = new HashMap<String, TreeMap<Long, Map<String, Object>>>();
        private final Deque<Object[]> pending = new ArrayDeque<Object[]>();
        private final Map<String, String> kindOfIdentity = new HashMap<String, String>();
        private final Map<Long, String> kindOverride = new HashMap<Long, String>();
        private final List<Conflict> conflicts = new ArrayList<Conflict>();
        private final List<String> notes = new ArrayList<String>();
        private final TreeMap<Long, Object[]> statementGraph = new TreeMap<Long, Object[]>();
        private long nextRef = 1;

        @SuppressWarnings("unchecked")
        Run(List<byte[]> bytes) {
            this.bytes = bytes;
            for (String[] collection : COLLECTIONS) {
                out.put(collection[0], new TreeMap<Long, Map<String, Object>>());
            }
            for (String member : DocumentIndex.MODULE_MEMBERS) {
                out.put("module:" + member, new TreeMap<Long, Map<String, Object>>());
            }
            for (byte[] data : bytes) {
                docs.add((Map<String, Object>) CanonicalJson.parse(data));
            }
        }

        Result run(List<String> digests) {
            checkCompatible();
            for (int i = 0; i < bytes.size(); i++) {
                Semantics forms = new Semantics(bytes.get(i), true, String.valueOf(i));
                semantics.add(forms);
                indexes.add(new DocumentIndex(docs.get(i)));
                remaps.add(new HashMap<Long, Long>());
                Map<String, Long> keys = new HashMap<String, Long>();
                Map<String, Long> formed = new HashMap<String, Long>();
                for (Long ref : allRefs(i)) {
                    String key = keyOf(i, ref);
                    if (key != null && !keys.containsKey(key)) {
                        keys.put(key, ref);
                    }
                    String form = formKey(i, ref);
                    if (!formed.containsKey(form)) {
                        formed.put(form, ref);
                    }
                }
                keyed.add(keys);
                byForm.add(formed);
                Set<Long> csets = new HashSet<Long>();
                for (Map<?, ?> cset : items(indexes.get(i).graph().get("candidateSets"))) {
                    csets.add(intRef(cset.get("ref")));
                }
                graphCsets.add(csets);
                Map<Long, Object[]> owners = new HashMap<Long, Object[]>();
                for (Map<?, ?> candidate : indexes.get(i).all("factCandidate")) {
                    for (String end : new String[] {"source", "target"}) {
                        Object side = candidate.get(end);
                        Object cset = side instanceof Map<?, ?> ? ((Map<?, ?>) side).get("candidateSet") : null;
                        Long ref = cset instanceof Map<?, ?> ? intRef(((Map<?, ?>) cset).get("ref")) : null;
                        if (ref != null) {
                            owners.put(ref, new Object[] {candidate.get("ref"), end});
                        }
                    }
                }
                embeddedIn.add(owners);
            }
            chooseCarriers();
            for (int i = 0; i < bytes.size(); i++) {
                mergeInput(i);
            }
            return new Result(buildDocument(), new ArrayList<String>(digests), conflicts, notes);
        }

        private TreeSet<Long> allRefs(int i) {
            List<Map<?, ?>> objects = new ArrayList<Map<?, ?>>(items(docs.get(i).get("analysisContexts")));
            for (String[] collection : COLLECTIONS) {
                objects.addAll(semantics.get(i).index().all(collection[0]));
            }
            for (String member : DocumentIndex.MODULE_MEMBERS) {
                objects.addAll(semantics.get(i).index().all("module:" + member));
            }
            TreeSet<Long> refs = new TreeSet<Long>();
            for (Map<?, ?> obj : objects) {
                Long ref = intRef(obj.get("ref"));
                if (ref != null) {
                    refs.add(ref);
                }
            }
            return refs;
        }

        private void checkCompatible() {
            Map<String, Object> first = docs.get(0);
            for (int i = 0; i < docs.size(); i++) {
                Map<String, Object> doc = docs.get(i);
                for (String member : doc.keySet()) {
                    if (!ROOT.contains(member) && !DocumentIndex.MODULE_MEMBERS.contains(member)) {
                        throw new MergeException("input " + i + ": member " + member + " is not supported by this merger");
                    }
                }
                Map<?, ?> graph = doc.get("explainGraph") instanceof Map<?, ?> ? (Map<?, ?>) doc.get("explainGraph")
                        : Collections.emptyMap();
                for (Object member : graph.keySet()) {
                    if (!GRAPH.contains(member)) {
                        throw new MergeException("input " + i + ": explainGraph." + member + " is not supported");
                    }
                }
                for (String member : new String[] {"specUri", "specVersion"}) {
                    if (!same(first.get(member), doc.get(member))) {
                        throw new MergeException("input " + i + ": " + member + " differs from input 0");
                    }
                }
                Map<?, ?> header = (Map<?, ?>) doc.get("header");
                Map<?, ?> header0 = (Map<?, ?>) first.get("header");
                for (String member : new String[] {"registryVersion", "sourceEmbedding"}) {
                    if (!same(header0.get(member), header.get(member))) {
                        throw new MergeException("input " + i + ": header." + member + " differs from input 0");
                    }
                }
                if (header.containsKey("extensions")) {
                    throw new MergeException("input " + i + ": header extensions are not supported by this merger");
                }
            }
        }

        private boolean same(Object a, Object b) {
            return CanonicalJson.jcsText(a).equals(CanonicalJson.jcsText(b));
        }

        // -----------------------------------------------------------------------------------------

        private String kindOf(int i, Object ref) {
            return semantics.get(i).kindOf(ref);
        }

        /** The (unit id, analysis context id) key of a unit of input i (read with identifiers filled). */
        private String unitKey(int i, Object unitRef) {
            Map<?, ?> unit = semantics.get(i).index().get("sourceUnit", unitRef);
            if (unit == null) {
                return null;
            }
            Object id = unit.get("unitId");
            Object context = unit.get("analysisContextId");
            return id instanceof String && context instanceof String ? "unit|" + id + "|" + context : null;
        }

        /** The unit ref an object of unit content belongs to, or null when no unit owns it. */
        private Long owningUnit(int i, Long ref) {
            Long current = ref;
            for (int guard = 0; guard < 4096 && current != null; guard++) {
                String kind = kindOf(i, current);
                Map<?, ?> obj = semantics.get(i).object(current);
                if (obj == null) {
                    return null;
                }
                if ("sourceUnit".equals(kind)) {
                    return current;
                }
                if (!UNIT_CONTENT.contains(kind) && !kind.startsWith("module:")) {
                    return null;
                }
                DocumentIndex index = semantics.get(i).index();
                if ("statement".equals(kind)) {
                    return intRef(obj.get("sourceUnitRef"));
                }
                Map<?, ?> statement = index.get("statement", "rootCandidateSet".equals(kind)
                        ? semantics.get(i).rootSetStatement(current) : "candidateSet".equals(kind)
                        ? semantics.get(i).graphSetStatement(current) : obj.get("statementRef"));
                if (statement != null) {
                    return intRef(statement.get("sourceUnitRef"));
                }
                Long unit = intRef(obj.get("sourceUnitRef"));
                for (String member : new String[] {"span", "location"}) {
                    if (unit == null && obj.get(member) instanceof Map<?, ?>) {
                        unit = intRef(((Map<?, ?>) obj.get(member)).get("sourceUnitRef"));
                    }
                }
                if (unit != null) {
                    return unit;
                }
                if ("scope".equals(kind)) {
                    current = intRef(obj.get("parentRef"));
                } else if ("supportPath".equals(kind)) {
                    // a support path is its micro-edges' statement's: two paths of one Occurrence through two LOCAL
                    // endpoints have one shape, and are still two objects (plan 09, ruling f)
                    Object edges = obj.get("microEdgeRefs");
                    current = edges instanceof List<?> && !((List<?>) edges).isEmpty()
                            ? intRef(((List<?>) edges).get(0)) : null;
                } else if ("endpoint".equals(kind)) {
                    // a LOCAL endpoint belongs where its scope or its owner (a QUERY_RESULT, a temporary table) is
                    current = obj.get("scopeRef") != null ? intRef(obj.get("scopeRef")) : intRef(obj.get("ownerRef"));
                } else {
                    current = null;
                }
            }
            return null;
        }

        /** The unit key whose carrier carries an object, or null. */
        private String owningKey(int i, Long ref) {
            Long unit = owningUnit(i, ref);
            return unit == null ? null : unitKey(i, unit);
        }

        /** The merge key of an object of input i, or null when it has none. */
        private String keyOf(int i, Long ref) {
            String kind = kindOf(i, ref);
            Map<?, ?> obj = semantics.get(i).object(ref);
            if (obj == null) {
                return null;
            }
            if ("analysisContext".equals(kind)) {
                return semantics.get(i).keyedIdentity(ref);
            }
            if ("sourceUnit".equals(kind)) {
                return unitKey(i, ref);
            }
            if ("statement".equals(kind)) {
                String unit = unitKey(i, obj.get("sourceUnitRef"));
                return unit == null ? null : "statement|" + unit + "|" + obj.get("ordinal");
            }
            if ("endpoint".equals(kind)) {
                Map<?, ?> identity = obj.get("identity") instanceof Map<?, ?> ? (Map<?, ?>) obj.get("identity") : null;
                if (identity != null && "PORTABLE".equals(identity.get("status")) && identity.get("id") instanceof String) {
                    return "endpoint|" + identity.get("id");
                }
                return null;
            }
            if ("fact".equals(kind)) {
                return obj.get("id") instanceof String ? "fact|" + obj.get("id") : null;
            }
            if ("occurrence".equals(kind) || "node".equals(kind)) {
                return obj.get("recordId") instanceof String ? kind + "|" + obj.get("recordId") : null;
            }
            return null;
        }

        /** The analysis contexts of an input, as one key. */
        private String analysisKey(int i) {
            TreeSet<String> contexts = new TreeSet<String>();
            for (Map<?, ?> context : items(docs.get(i).get("analysisContexts"))) {
                contexts.add(String.valueOf(semantics.get(i).keyedIdentity(context.get("ref"))));
            }
            return sha256(CanonicalJson.jcsBytes(new ArrayList<String>(contexts)));
        }

        private String formKey(int i, Long ref) {
            return "form|" + kindOf(i, ref) + "|"
                    + CanonicalJson.jcsText(semantics.get(i).formOf(ref));
        }

        /** One object of a unit's content: its form without the members modules add, and those members. */
        private final class Item {
            final String core;
            final Map<String, String> added = new TreeMap<String, String>();

            Item(int i, Long ref) {
                Map<String, Object> form = new TreeMap<String, Object>();
                Object full = semantics.get(i).formOf(ref);
                for (Map.Entry<?, ?> entry : ((Map<?, ?>) full).entrySet()) {
                    String member = String.valueOf(entry.getKey());
                    if (MODULE_MEMBERS_OF_OBJECTS.contains(member)) {
                        added.put(member, CanonicalJson.jcsText(entry.getValue()));
                    } else {
                        form.put(member, entry.getValue());
                    }
                }
                core = kindOf(i, ref) + " " + CanonicalJson.jcsText(form);
            }

            /** True when this item says at least what another says. */
            boolean covers(Item other) {
                for (Map.Entry<String, String> entry : other.added.entrySet()) {
                    if (!entry.getValue().equals(added.get(entry.getKey()))) {
                        return false;
                    }
                }
                return true;
            }
        }

        /**
         * Section 9.2: the records of one unit under one context are unioned. This merger carries the content of
         * such a unit from one input, which must include every other input's content: compared with references
         * replaced by identities, every statement, scope, Occurrence, write, diagnostic, ExplainGraph node and
         * micro-edge, endpoint and module object the unit owns, each with at least the members modules add to it
         * (support paths, evidence). Among such inputs it takes the highest declared level, then the most content,
         * then the first; when there is none it refuses.
         */
        private void chooseCarriers() {
            Map<String, List<Integer>> holders = new LinkedHashMap<String, List<Integer>>();
            List<Map<String, List<Item>>> contents = new ArrayList<Map<String, List<Item>>>();
            for (int i = 0; i < bytes.size(); i++) {
                Map<String, List<Item>> mine = new HashMap<String, List<Item>>();
                for (Long ref : allRefs(i)) {
                    String unit = owningKey(i, ref);
                    if (unit == null) {
                        continue;
                    }
                    if ("sourceUnit".equals(kindOf(i, ref))) {
                        List<Integer> list = holders.get(unit);
                        if (list == null) {
                            list = new ArrayList<Integer>();
                            holders.put(unit, list);
                        }
                        if (!list.contains(i)) {
                            list.add(i);
                        }
                    }
                    List<Item> items = mine.get(unit);
                    if (items == null) {
                        items = new ArrayList<Item>();
                        mine.put(unit, items);
                    }
                    items.add(new Item(i, ref));
                }
                contents.add(mine);
            }
            for (Map.Entry<String, List<Integer>> entry : holders.entrySet()) {
                String unit = entry.getKey();
                Integer chosen = null;
                for (Integer i : entry.getValue()) {
                    boolean all = true;
                    for (Integer j : entry.getValue()) {
                        all &= j.intValue() == i.intValue() || contains(unitItems(contents, i, unit), unitItems(contents, j, unit));
                    }
                    if (all && (chosen == null || better(i, chosen, unitItems(contents, i, unit).size(),
                            unitItems(contents, chosen, unit).size()))) {
                        chosen = i;
                    }
                }
                if (chosen == null) {
                    throw new MergeException("inputs " + entry.getValue() + " assert different content for one source"
                            + " unit under one analysis context and none includes the others: " + unit);
                }
                carrier.put(unit, chosen);
            }
        }

        private List<Item> unitItems(List<Map<String, List<Item>>> contents, int i, String unit) {
            List<Item> items = contents.get(i).get(unit);
            return items == null ? Collections.<Item>emptyList() : items;
        }

        /** True when every item of a part is covered by a distinct item of a whole with the same core. */
        private boolean contains(List<Item> whole, List<Item> part) {
            Map<String, List<Item>> byCore = new HashMap<String, List<Item>>();
            for (Item item : whole) {
                List<Item> list = byCore.get(item.core);
                if (list == null) {
                    list = new ArrayList<Item>();
                    byCore.put(item.core, list);
                }
                list.add(item);
            }
            for (Item item : part) {
                List<Item> list = byCore.get(item.core);
                boolean found = false;
                for (Iterator<Item> it = list == null ? Collections.<Item>emptyList().iterator() : list.iterator();
                     it.hasNext(); ) {
                    if (it.next().covers(item)) {
                        it.remove();
                        found = true;
                        break;
                    }
                }
                if (!found) {
                    return false;
                }
            }
            return true;
        }

        private boolean better(int i, int chosen, int size, int chosenSize) {
            int level = LEVELS.indexOf(docs.get(i).get("declaredLevel")) - LEVELS.indexOf(docs.get(chosen).get("declaredLevel"));
            return level != 0 ? level > 0 : size > chosenSize;
        }

        private boolean carries(int i, Long ref) {
            String unit = owningKey(i, ref);
            return unit == null || carrier.get(unit) == i;
        }

        /** The new ref of a ref of input i; the object is queued to be carried when it is new. */
        private Long map(int i, Object value) {
            Long ref = intRef(value);
            if (ref == null) {
                return null;
            }
            Long known = remaps.get(i).get(ref);
            if (known != null) {
                return known;
            }
            String kind = kindOf(i, ref);
            if (kind == null) {
                throw new MergeException("input " + i + ": ref " + ref + " names nothing");
            }
            String key = keyOf(i, ref);
            String unit = owningKey(i, ref);
            Integer from = unit == null ? null : carrier.get(unit);
            if (from != null && from != i) {
                // the object belongs to a unit another input carries: use that input's equal object
                Long there = key != null ? keyed.get(from).get(key) : null;
                if (there == null) {
                    there = byForm.get(from).get(formKey(i, ref));
                }
                if (there == null) {
                    throw new MergeException("input " + i + ": " + kind + " ref " + ref + " belongs to " + unit
                            + ", which input " + from + " carries without it");
                }
                Long mapped = map(from, there);
                remaps.get(i).put(ref, mapped);
                return mapped;
            }
            boolean embedded = "candidateSet".equals(kind) && !graphCsets.get(i).contains(ref);
            if (embedded && embeddedIn.get(i).containsKey(ref)) {
                // the set at one end of a FactCandidate: when an equal FactCandidate (of this input or another) is
                // carried in its place, a reference to this set names the set of that one
                Object[] owner = embeddedIn.get(i).get(ref);
                String slot = map(i, owner[0]) + "|" + owner[1];
                Long shared = embeddedSlots.get(slot);
                if (shared == null) {
                    shared = nextRef++;
                    embeddedSlots.put(slot, shared);
                }
                remaps.get(i).put(ref, shared);
                return shared;
            }
            if (key == null && unit == null && !embedded) {
                // an object without identity that no unit owns is one object per content within one analysis: two
                // inputs with the same analysis contexts assert it once, inputs with different contexts each keep it
                key = "analysis|" + analysisKey(i) + "|" + formKey(i, ref);
            }
            Long mapped = key == null ? null : byKey.get(key);
            Map<?, ?> obj = "analysisContext".equals(kind) || "rootCandidateSet".equals(kind)
                    ? semantics.get(i).object(ref) : indexes.get(i).get(kind, ref);
            if ("endpoint".equals(kind) && key != null && key.startsWith("endpoint|")) {
                checkKind(i, key.substring("endpoint|".length()), String.valueOf(obj.get("kind")), mapped);
            }
            if (mapped == null) {
                mapped = nextRef++;
                if (key != null) {
                    byKey.put(key, mapped);
                }
                // a candidate set inside a FactCandidate end is copied with its FactCandidate, not on its own
                if (!embedded) {
                    pending.add(new Object[] {i, kind, obj, mapped});
                }
            }
            remaps.get(i).put(ref, mapped);
            return mapped;
        }

        /**
         * The endpoint kind is not an identity input (identity-v1, Section 5.2.1): FROM orders does not say table or
         * view. A generic RELATION is refined by a specific relation kind; any other difference cannot be kept in one
         * document until the merge module carries MergeConflict, so it is refused.
         */
        private void checkKind(int i, String identity, String kind, Long mapped) {
            String seen = kindOfIdentity.get(identity);
            if (seen == null || seen.equals(kind)) {
                kindOfIdentity.put(identity, kind);
                return;
            }
            if ("RELATION".equals(seen) && RELATION_KINDS.contains(kind)) {
                kindOfIdentity.put(identity, kind);
                kindOverride.put(mapped, kind);
                conflicts.add(new Conflict(identity, "kind RELATION refined to " + kind + " by input " + i));
            } else if (!("RELATION".equals(kind) && RELATION_KINDS.contains(seen))) {
                throw new MergeException("input " + i + ": endpoint " + identity + " is a " + kind
                        + " where an earlier input has a " + seen + "; both cannot be kept (Section 9.4)");
            }
        }

        private void mergeInput(int i) {
            DocumentIndex index = indexes.get(i);
            for (Long ref : allRefs(i)) {
                String kind = kindOf(i, ref);
                boolean embedded = "candidateSet".equals(kind) && !graphCsets.get(i).contains(ref);
                if (!embedded && carries(i, ref) && !"supportPath".equals(kind) && !"factCandidate".equals(kind)
                        && !"fact".equals(kind)) {
                    map(i, ref);
                }
            }
            Map<?, ?> graph = index.graph();
            Map<Long, Map<?, ?>> listed = new HashMap<Long, Map<?, ?>>();
            for (Map<?, ?> entry : items(graph.get("statementCompleteness"))) {
                listed.put(intRef(entry.get("statementRef")), entry);
            }
            for (Map<?, ?> statement : index.all("statement")) {
                Long ref = intRef(statement.get("ref"));
                if (!carries(i, ref)) {
                    continue;
                }
                Map<?, ?> entry = listed.get(ref);
                Object completeness = entry != null ? entry.get("completeness")
                        : index.hasGraph() ? graph.get("completeness") : "PARTIAL";
                Long out = remaps.get(i).get(ref);
                Object[] prior = statementGraph.get(out);
                // a statement several inputs carry has the graph of the one that has the most of it (an L1 input has
                // none of it), and of equals the one that says more, whatever the order of the inputs: merging stays
                // associative
                if (prior == null || graphRank(completeness) > graphRank(prior[0])
                        || (graphRank(completeness) == graphRank(prior[0]) && prior[1] == null && entry != null)) {
                    statementGraph.put(out, new Object[] {completeness, entry == null ? null : copy(i, entry, false)});
                }
            }
            drain();
        }

        /** Carries every queued object, which may queue the objects it references. */
        private void drain() {
            while (!pending.isEmpty()) {
                Object[] job = pending.poll();
                int i = (Integer) job[0];
                String kind = (String) job[1];
                @SuppressWarnings("unchecked")
                Map<String, Object> copied = (Map<String, Object>) copy(i, job[2], true);
                copied.put("ref", job[3]);
                out.get(kind).put((Long) job[3], copied);
            }
        }

        /** One container being copied. */
        private final class Frame {
            final Iterator<?> items;
            final Map<String, Object> map;
            final List<Object> list;
            String key;

            Frame(Object container) {
                if (container instanceof Map<?, ?>) {
                    items = ((Map<?, ?>) container).entrySet().iterator();
                    map = new LinkedHashMap<String, Object>();
                    list = null;
                } else {
                    items = ((List<?>) container).iterator();
                    map = null;
                    list = new ArrayList<Object>();
                }
            }

            void add(Object value) {
                if (map != null) {
                    map.put(key, value);
                } else {
                    list.add(value);
                }
            }
        }

        /**
         * A copy of a JSON value of input i with every reference renumbered, made with an explicit stack. The
         * {@code ref} of the top object is its own and is left for the caller; a {@code ref} below it (a candidate
         * set inside a FactCandidate) is renumbered.
         */
        private Object copy(int i, Object value, boolean top) {
            if (!(value instanceof Map<?, ?> || value instanceof List<?>)) {
                return value;
            }
            Deque<Frame> stack = new ArrayDeque<Frame>();
            stack.push(new Frame(value));
            Object result = null;
            while (!stack.isEmpty()) {
                Frame frame = stack.peek();
                if (!frame.items.hasNext()) {
                    stack.pop();
                    Object built = frame.map != null ? frame.map : frame.list;
                    if (stack.isEmpty()) {
                        result = built;
                    } else {
                        stack.peek().add(built);
                    }
                    continue;
                }
                Object item = frame.items.next();
                Object child;
                if (frame.map != null) {
                    Map.Entry<?, ?> entry = (Map.Entry<?, ?>) item;
                    String member = (String) entry.getKey();
                    child = entry.getValue();
                    frame.key = member;
                    if ("extensions".equals(member)) {
                        frame.add(child);
                        continue;
                    }
                    if ("ref".equals(member)) {
                        frame.add(top && stack.size() == 1 ? child : map(i, child));
                        continue;
                    }
                    if (member.endsWith("Refs") && child instanceof List<?>) {
                        List<Object> refs = new ArrayList<Object>();
                        for (Object r : (List<?>) child) {
                            refs.add(map(i, r));
                        }
                        frame.add(refs);
                        continue;
                    }
                    if (member.endsWith("Ref") && intRef(child) != null) {
                        frame.add(map(i, child));
                        continue;
                    }
                } else {
                    child = item;
                }
                if (child instanceof Map<?, ?> || child instanceof List<?>) {
                    stack.push(new Frame(child));
                } else {
                    frame.add(child);
                }
            }
            return result;
        }

        // -----------------------------------------------------------------------------------------

        private Document buildDocument() {
            Map<String, Object> first = docs.get(0);
            Map<String, Object> doc = new LinkedHashMap<String, Object>();
            doc.put("specUri", first.get("specUri"));
            doc.put("specVersion", first.get("specVersion"));
            String level = "L1";
            Set<Object> modules = new TreeSet<Object>();
            Object analysis = null;
            List<Object> serving = new ArrayList<Object>();
            Map<Object, List<Object>> features = new TreeMap<Object, List<Object>>();
            for (int i = 0; i < docs.size(); i++) {
                Map<String, Object> input = docs.get(i);
                if (LEVELS.indexOf(input.get("declaredLevel")) > LEVELS.indexOf(level)) {
                    level = (String) input.get("declaredLevel");
                }
                modules.addAll((List<?>) input.get("modules"));
                analysis = combined(analysis, input.get("analysisCompleteness"));
                serving.add(copy(i, input.get("servingCompleteness"), false));
                for (Map<?, ?> feature : items(input.get("featureCompleteness"))) {
                    List<Object> list = features.get(feature.get("feature"));
                    if (list == null) {
                        list = new ArrayList<Object>();
                        features.put(feature.get("feature"), list);
                    }
                    list.add(copy(i, feature, false));
                }
            }
            drain();
            for (Map.Entry<Long, String> entry : kindOverride.entrySet()) {
                out.get("endpoint").get(entry.getKey()).put("kind", entry.getValue());
            }
            doc.put("declaredLevel", level);
            doc.put("modules", new ArrayList<Object>(modules));
            Map<String, Object> header = new LinkedHashMap<String, Object>();
            header.put("producer", producer.toJson());
            Map<?, ?> header0 = (Map<?, ?>) first.get("header");
            header.put("registryVersion", header0.get("registryVersion"));
            header.put("sourceEmbedding", header0.get("sourceEmbedding"));
            if (createdAt != null) {
                header.put("createdAt", createdAt);
            }
            doc.put("header", header);
            for (String[] collection : COLLECTIONS) {
                // the root candidate sets are present only when an Occurrence of a Fact names one (spec 4.3, rule 1)
                if (!GRAPH_KINDS.contains(collection[0])
                        && !("rootCandidateSet".equals(collection[0]) && out.get(collection[0]).isEmpty())) {
                    doc.put(collection[1], new ArrayList<Object>(out.get(collection[0]).values()));
                }
            }
            for (String member : DocumentIndex.MODULE_MEMBERS) {
                if (!out.get("module:" + member).isEmpty()) {
                    doc.put(member, new ArrayList<Object>(out.get("module:" + member).values()));
                }
            }
            doc.put("analysisCompleteness", analysis);
            doc.put("servingCompleteness", mergedServing(serving));
            if (!features.isEmpty()) {
                List<Object> list = new ArrayList<Object>();
                for (List<Object> entries : features.values()) {
                    list.add(mergedFeature(entries));
                }
                doc.put("featureCompleteness", list);
            }
            if (modules.contains("explain")) {
                doc.put("explainGraph", graph());
            }
            dropRepeats(doc);
            return Document.fromJson(doc);
        }

        /**
         * Equal objects without an identity are one object of the analysis, so two elements of a SET can become
         * equal once their refs are renumbered (two equal diagnostics of one statement, two equal FactCandidates
         * among the inputs of one target field); the SET keeps such an element once. Two elements with an equal
         * sort key and different content stay, and canonical JSON refuses them.
         */
        @SuppressWarnings("unchecked")
        private void dropRepeats(Map<String, Object> doc) {
            Deque<Object[]> stack = new ArrayDeque<Object[]>();
            stack.push(new Object[] {doc, ArrayRegistry.standard().root()});
            while (!stack.isEmpty()) {
                Object[] top = stack.pop();
                ArrayRegistry.Position position = (ArrayRegistry.Position) top[1];
                if (top[0] instanceof Map<?, ?>) {
                    for (Map.Entry<?, ?> entry : ((Map<?, ?>) top[0]).entrySet()) {
                        if (!"extensions".equals(entry.getKey())) {
                            stack.push(new Object[] {entry.getValue(),
                                    position == null ? null : position.member(String.valueOf(entry.getKey()))});
                        }
                    }
                } else if (top[0] instanceof List<?>) {
                    List<Object> list = (List<Object>) top[0];
                    if (position != null && position.registration() != null
                            && position.registration().getOrder() == ArrayRegistry.Order.SET) {
                        Set<String> seen = new HashSet<String>();
                        for (Iterator<Object> it = list.iterator(); it.hasNext();) {
                            if (!seen.add(CanonicalJson.jcsText(it.next()))) {
                                it.remove();
                            }
                        }
                    }
                    ArrayRegistry.Position element = position == null ? null : position.element();
                    for (Object item : list) {
                        stack.push(new Object[] {item, element});
                    }
                }
            }
        }

        /** Equal objects stay as they are; otherwise TRUNCATED unless all are COMPLETE, the truncations unioned. */
        @SuppressWarnings("unchecked")
        private Map<String, Object> mergedServing(List<Object> objects) {
            List<Object> distinct = distinct(objects);
            if (distinct.size() == 1) {
                return (Map<String, Object>) distinct.get(0);
            }
            Map<String, Object> merged = new LinkedHashMap<String, Object>();
            boolean complete = true;
            List<Object> truncations = new ArrayList<Object>();
            for (Object object : distinct) {
                Map<String, Object> serving = (Map<String, Object>) object;
                rejectExtensions(serving, "servingCompleteness");
                complete &= "COMPLETE".equals(serving.get("status"));
                if (serving.get("truncations") instanceof List<?>) {
                    truncations.addAll((List<Object>) serving.get("truncations"));
                }
            }
            merged.put("status", complete ? "COMPLETE" : "TRUNCATED");
            if (!truncations.isEmpty()) {
                merged.put("truncations", distinct(truncations));
            }
            return merged;
        }

        /** Equal entries stay as they are; otherwise the status is combined and the diagnostics unioned. */
        @SuppressWarnings("unchecked")
        private Map<String, Object> mergedFeature(List<Object> objects) {
            List<Object> distinct = distinct(objects);
            if (distinct.size() == 1) {
                return (Map<String, Object>) distinct.get(0);
            }
            Map<String, Object> merged = new LinkedHashMap<String, Object>();
            Object status = null;
            Set<Object> diagnostics = new LinkedHashSet<Object>();
            for (Object object : distinct) {
                Map<String, Object> entry = (Map<String, Object>) object;
                rejectExtensions(entry, "featureCompleteness");
                merged.put("feature", entry.get("feature"));
                status = combined(status, entry.get("status"));
                if (entry.get("diagnosticRefs") instanceof List<?>) {
                    diagnostics.addAll((List<Object>) entry.get("diagnosticRefs"));
                }
            }
            merged.put("status", status);
            if (!diagnostics.isEmpty()) {
                merged.put("diagnosticRefs", new ArrayList<Object>(diagnostics));
            }
            return merged;
        }

        private void rejectExtensions(Map<String, Object> object, String what) {
            if (object.containsKey("extensions")) {
                throw new MergeException(what + " objects with extensions differ between inputs and cannot be combined");
            }
        }

        /** How much of a statement's graph an input has: COMPLETE above PARTIAL above anything else. */
        private int graphRank(Object completeness) {
            return "COMPLETE".equals(completeness) ? 2 : "PARTIAL".equals(completeness) ? 1 : 0;
        }

        private List<Object> distinct(List<Object> values) {
            Map<String, Object> seen = new LinkedHashMap<String, Object>();
            for (Object value : values) {
                seen.put(CanonicalJson.jcsText(value), value);
            }
            return new ArrayList<Object>(seen.values());
        }

        private Map<String, Object> graph() {
            Map<String, Object> graph = new LinkedHashMap<String, Object>();
            boolean allComplete = true;
            for (Object[] completeness : statementGraph.values()) {
                allComplete &= "COMPLETE".equals(completeness[0]);
            }
            String byDefault = allComplete ? "COMPLETE" : "PARTIAL";
            graph.put("completeness", byDefault);
            List<Object> listed = new ArrayList<Object>();
            for (Map.Entry<Long, Object[]> entry : statementGraph.entrySet()) {
                Object[] value = entry.getValue();
                if (value[1] != null && (!byDefault.equals(value[0]) || ((Map<?, ?>) value[1]).get("diagnosticRefs") != null)) {
                    // only a statement that differs from the graph's completeness is listed (fold-v1, Section 2)
                    @SuppressWarnings("unchecked")
                    Map<String, Object> copied = (Map<String, Object>) value[1];
                    copied.put("statementRef", entry.getKey());
                    listed.add(copied);
                } else if (!byDefault.equals(value[0])) {
                    Map<String, Object> item = new LinkedHashMap<String, Object>();
                    item.put("statementRef", entry.getKey());
                    item.put("completeness", value[0]);
                    listed.add(item);
                }
            }
            if (!listed.isEmpty()) {
                graph.put("statementCompleteness", listed);
            }
            graph.put("nodes", new ArrayList<Object>(out.get("node").values()));
            graph.put("microEdges", new ArrayList<Object>(out.get("microEdge").values()));
            if (!out.get("candidateSet").isEmpty()) {
                graph.put("candidateSets", new ArrayList<Object>(out.get("candidateSet").values()));
            }
            graph.put("supportPaths", new ArrayList<Object>(out.get("supportPath").values()));
            return graph;
        }
    }
}