package gudusoft.gsqlparser.dlineage.util;

import gudusoft.gsqlparser.dlineage.dataflow.model.xml.table;
import gudusoft.gsqlparser.sqlenv.QualifiedCanonKey;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Merge keys of the physical tables and views of one output merge.
 *
 * <p>Two tables are one table exactly when their identities under the analysis's identifier rules agree,
 * whatever their legacy folded names. A table carries its identity when its producer minted one (a table
 * analyzed from SQL, a catalog table read by {@code SQLEnvMetadataAnalyzer}), or vouches that its published
 * name is a SQL spelling (the SQLDep, Grabit and sqlflow readers), in which case the name read under this
 * merge's rules is its identity. A table from a serialized document carries none: its published name is
 * then the only evidence, and a weak one, since a stored name is published bare and reads as another
 * name.</p>
 *
 * <p>Decided after every table of the merge is known, and independently of the order they arrive in:</p>
 * <ul>
 * <li>a table carrying an identity is keyed by that identity;</li>
 * <li>a table carrying none joins the one identity it can be shown to be: the identity of a table that
 * published exactly the same name (the table reloaded, or its copy), else the identity its own published
 * name reads as, if some table carries it; under a configured identifier profile its published name is its
 * identity, as the configured-profile merge always read it;</li>
 * <li>otherwise it keeps the legacy merge when no table of its legacy bucket has an identity, and stays
 * apart when one does, rather than attaching to a guess.</li>
 * </ul>
 *
 * <p>An identity is keyed by the first legacy bucket it was seen in: the bucket's first identity by the
 * legacy key itself, as before identities existed, so a merge that groups tables as it always did also
 * orders its output as it always did; another identity of the bucket by the legacy key and its position.</p>
 */
public final class PhysicalTableMergeKeys {

    private final boolean configuredRules;
    private final Set<QualifiedCanonKey> carried = new HashSet<QualifiedCanonKey>();
    /** The carried identities each published name was seen with, across the merge. */
    private final Map<String, Set<QualifiedCanonKey>> carriedByPublishedName =
            new HashMap<String, Set<QualifiedCanonKey>>();
    private final Map<QualifiedCanonKey, String> firstBucket = new HashMap<QualifiedCanonKey, String>();
    private final Map<String, List<QualifiedCanonKey>> identitiesByBucket = new HashMap<String, List<QualifiedCanonKey>>();

    /**
     * @param configuredRules whether the analysis runs under identifier rules of its own, rather than its
     *        vendor's defaults
     */
    public PhysicalTableMergeKeys(boolean configuredRules) {
        this.configuredRules = configuredRules;
    }

    /**
     * The identity a table carries: the one its producer minted, else its published name read under this
     * analysis's rules when the producer vouches the name is a SQL spelling. Null for a table that carries
     * none.
     */
    public static QualifiedCanonKey carriedIdentity(table table, String legacyKey, String qualifiedTableName) {
        if (table.getMergeIdentity() != null) {
            return table.getMergeIdentity();
        }
        return table.isPublishedNameSqlSpelling()
                ? DlineageUtil.publishedTableIdentity(qualifiedTableName, legacyKey) : null;
    }

    /**
     * First pass for one physical table or view: {@link #add} for the table under its own name and, for a merged
     * table that carries an identity, under each exact name the carried tables merged into it published
     * ({@code table.getCarriedPublishedNames()}), so a merge in stages matches a table that carries no identity
     * on the evidence one merge of all the inputs has.
     */
    public void addTable(table table, String legacyKey, String qualifiedTableName) {
        QualifiedCanonKey identity = carriedIdentity(table, legacyKey, qualifiedTableName);
        add(legacyKey, identity, qualifiedTableName);
        if (identity == null || table.getCarriedPublishedNames() == null) {
            return;
        }
        for (Map.Entry<String, String> name : table.getCarriedPublishedNames().entrySet()) {
            add(name.getValue(), identity, name.getKey());
        }
    }

    /** First pass: one call per physical table or view of the merge, or per name {@link #addTable} registers. */
    public void add(String legacyKey, QualifiedCanonKey carriedIdentity, String qualifiedTableName) {
        if (legacyKey == null) {
            return;
        }
        QualifiedCanonKey identity = carriedIdentity;
        if (identity != null) {
            carried.add(identity);
            if (qualifiedTableName != null) {
                Set<QualifiedCanonKey> identities = carriedByPublishedName.get(qualifiedTableName);
                if (identities == null) {
                    identities = new LinkedHashSet<QualifiedCanonKey>();
                    carriedByPublishedName.put(qualifiedTableName, identities);
                }
                identities.add(identity);
            }
        } else if (configuredRules) {
            identity = DlineageUtil.publishedTableIdentity(qualifiedTableName, legacyKey);
        }
        if (identity == null) {
            return;
        }
        if (!firstBucket.containsKey(identity)) {
            firstBucket.put(identity, legacyKey);
        }
        List<QualifiedCanonKey> identities = identitiesByBucket.get(legacyKey);
        if (identities == null) {
            identities = new ArrayList<QualifiedCanonKey>();
            identitiesByBucket.put(legacyKey, identities);
        }
        if (!identities.contains(identity)) {
            identities.add(identity);
        }
    }

    /** Second pass: the merge key of a table given to {@link #add}. */
    public String keyOf(String legacyKey, QualifiedCanonKey carriedIdentity, String qualifiedTableName) {
        if (legacyKey == null) {
            return null;
        }
        QualifiedCanonKey identity = carriedIdentity != null ? carriedIdentity
                : uncarriedIdentity(legacyKey, qualifiedTableName);
        String bucket = identity == null ? null : firstBucket.get(identity);
        if (bucket != null) {
            int index = identitiesByBucket.get(bucket).indexOf(identity);
            return index == 0 ? bucket : bucket + "\u0001" + index;
        }
        return identitiesByBucket.containsKey(legacyKey) ? legacyKey + "\u0002" + qualifiedTableName : legacyKey;
    }

    private QualifiedCanonKey uncarriedIdentity(String legacyKey, String qualifiedTableName) {
        Set<QualifiedCanonKey> byName = qualifiedTableName == null ? null : carriedByPublishedName.get(qualifiedTableName);
        if (byName != null) {
            return byName.size() == 1 ? byName.iterator().next() : null;
        }
        QualifiedCanonKey read = DlineageUtil.publishedTableIdentity(qualifiedTableName, legacyKey);
        if (read == null) {
            return null;
        }
        return carried.contains(read) || configuredRules ? read : null;
    }
}