package gudusoft.gsqlparser.sqlenv;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.TBaseType;
import gudusoft.gsqlparser.sqlenv.IdentifierRules.CaseCompare;
import gudusoft.gsqlparser.sqlenv.IdentifierRules.CaseFold;
import gudusoft.gsqlparser.sqlenv.unicode.GspCaseData;
import gudusoft.gsqlparser.util.SQLUtil;

import java.text.Collator;
import java.util.ArrayList;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;

/**
 * 标识符服务（Identifier Service）
 *
 * <p>提供所有标识符规范化和比较的统一入口。</p>
 *
 * <p><strong>为什么需要它（面向入门者）：</strong><br>
 * 不同数据库对名称的大小写和引号行为差别很大（是否区分大小写、是否折叠为大/小写、是否受 collation 影响）。
 * 如果在业务代码里到处手写 {@code toLowerCase/equalsIgnoreCase}，很容易出错且难以维护。
 * 本服务将“如何折叠与如何比较”的规则集中在一个地方，确保全局一致、可扩展、易测试。</p>
 *
 * <p><strong>与其它类的关系：</strong>
 * <ul>
 *   <li>{@link IdentifierRules}：一张“规则卡片”（策略），描述未引号/带引号的折叠（fold）与比较（compare）。</li>
 *   <li>{@link IdentifierProfile}：一个“厂商档案”，为某个数据库厂商打包不同对象组（表/列/函数）的规则，并携带厂商开关（如 MySQL lower_case_table_names、SQL Server collation）。</li>
 *   <li>IdentifierService：基于 Profile 执行规范化与比较，向外提供统一的 {@code normalize/areEqual/keyForMap} 接口（门面）。</li>
 * </ul>
 *
 * <p><strong>设计理念与收益：</strong>
 * <ul>
 *   <li>一致性：所有地方都通过同一入口生成 Map 键与做比较，避免行为分裂。</li>
 *   <li>可扩展：新增厂商仅需补一份规则工厂；旧调用不变。</li>
 *   <li>可测试：规则与服务可单测覆盖，验证各厂商/对象类型是否符合预期。</li>
 *   <li>性能友好：{@link #keyForMap} 先做规范化，Map 查找 O(1)；SQL Server 经 {@link CollatorProvider} 做 collation 比较。</li>
 * </ul>
 *
 * <p><strong>使用了哪些设计模式：</strong>
 * <ul>
 *   <li>策略（Strategy）：{@link IdentifierRules} 即为可替换的规则策略。</li>
 *   <li>工厂（Factory）：{@code IdentifierRules.forOracle()/forPostgreSQL()}、{@code IdentifierProfile.forVendor(...)} 产出预设策略组合。</li>
 *   <li>门面（Facade）：本类用少量方法对外隐藏折叠/比较/引号/Collator 细节。</li>
 *   <li>依赖注入（DI）：构造时注入 {@link IdentifierProfile} 与可选 {@link CollatorProvider}，便于替换与测试。</li>
 *   <li>不可变值对象（Immutable）：规则/flags 不可变，线程安全、可缓存。</li>
 *   <li>Provider：{@link CollatorProvider} 解耦 SQL Server 的 collation 依赖。</li>
 * </ul>
 *
 * <p><strong>关键约束（必须遵守）：</strong>
 * <ul>
 *   <li>所有索引键构造必须通过 {@link #keyForMap}</li>
 *   <li>所有标识符比较必须通过 {@link #areEqual}</li>
 *   <li>禁止在业务代码中直接调用 {@code String.toUpperCase/toLowerCase/equalsIgnoreCase}</li>
 * </ul>
 *
 * <p><strong>使用示例：</strong>
 * <pre>
 * IdentifierService service = new IdentifierService(profile, collatorProvider);
 *
 * // 规范化标识符（用于索引键）
 * String key = service.normalize("MyTable", ESQLDataObjectType.dotTable);
 * // Oracle: "MYTABLE", PostgreSQL: "mytable", ClickHouse: "MyTable"
 *
 * // 比较两个标识符
 * boolean eq = service.areEqual("MyTable", "MYTABLE", ESQLDataObjectType.dotTable);
 * // Oracle: true, ClickHouse: false
 *
 * // 构造索引键
 * String mapKey = service.keyForMap("MyTable", ESQLDataObjectType.dotTable);
 * </pre>
 *
 * @since 3.1.0.9
 */
public class IdentifierService {

    // ===== Static Cache for High-Performance Normalization =====

    /**
     * Static cache: one IdentifierService instance per database vendor
     *
     * <p>Pre-populated at class loading time for maximum performance.
     * EnumMap provides O(1) lookup with minimal memory overhead.</p>
     *
     * <p>Thread-safe: all cached instances are immutable.</p>
     */
    private static final Map<EDbVendor, IdentifierService> VENDOR_CACHE = new EnumMap<>(EDbVendor.class);

    static {
        // Pre-populate cache for all vendors at startup
        for (EDbVendor vendor : EDbVendor.values()) {
            IdentifierProfile profile = IdentifierProfile.forVendor(
                vendor,
                IdentifierProfile.VendorFlags.defaults()
            );
            CollatorProvider collatorProvider = (vendor == EDbVendor.dbvmssql || vendor == EDbVendor.dbvazuresql)
                ? new CollatorProvider()
                : null;
            VENDOR_CACHE.put(vendor, new IdentifierService(profile, collatorProvider));
        }
    }

    /**
     * High-performance static normalize method with caching
     *
     * <p>This method provides a convenient static interface for identifier normalization
     * while leveraging a pre-populated cache of IdentifierService instances.</p>
     *
     * <p><strong>Performance characteristics:</strong>
     * <ul>
     *   <li>O(1) cache lookup using EnumMap
     *   <li>No object creation - services are pre-created and reused
     *   <li>Thread-safe - all cached instances are immutable
     *   <li>Handles all vendor-specific rules (collation, per-object-type rules, etc.)
     * </ul>
     *
     * <p><strong>Usage example:</strong>
     * <pre>
     * String normalized = IdentifierService.normalizeStatic(
     *     EDbVendor.dbvoracle,
     *     ESQLDataObjectType.dotTable,
     *     "MyTable"
     * );
     * // Result: "MYTABLE"
     * </pre>
     *
     * @param dbVendor database vendor
     * @param objectType object type (table, column, schema, etc.)
     * @param identifier identifier to normalize (may be quoted)
     * @return normalized identifier (unquoted and case-folded)
     */
    public static String normalizeStatic(EDbVendor dbVendor, ESQLDataObjectType objectType, String identifier) {
        if (identifier == null || identifier.isEmpty()) {
            return identifier;
        }

        // Get cached instance (O(1) lookup in EnumMap)
        IdentifierService service = VENDOR_CACHE.get(dbVendor);
        if (service == null) {
            // Fallback for unexpected null (should never happen with pre-populated cache)
            // Create on-demand instance
            IdentifierProfile profile = IdentifierProfile.forVendor(dbVendor, IdentifierProfile.VendorFlags.defaults());
            CollatorProvider collatorProvider = (dbVendor == EDbVendor.dbvmssql || dbVendor == EDbVendor.dbvazuresql)
                ? new CollatorProvider()
                : null;
            service = new IdentifierService(profile, collatorProvider);
        }

        return service.normalize(identifier, objectType);
    }

    /**
     * High-performance static identifier comparison method with caching
     *
     * <p>This method provides a convenient static interface for identifier comparison
     * while leveraging a pre-populated cache of IdentifierService instances.</p>
     *
     * <p><strong>Performance characteristics:</strong>
     * <ul>
     *   <li>O(1) cache lookup using EnumMap
     *   <li>No object creation - services are pre-created and reused
     *   <li>Thread-safe - all cached instances are immutable
     *   <li>Handles all vendor-specific rules (case-sensitive, case-insensitive, collation-based)
     * </ul>
     *
     * <p><strong>Usage example:</strong>
     * <pre>
     * boolean equal = IdentifierService.areEqualStatic(
     *     EDbVendor.dbvoracle,
     *     ESQLDataObjectType.dotTable,
     *     "MyTable",
     *     "MYTABLE"
     * );
     * // Result: true (Oracle is case-insensitive for unquoted identifiers)
     * </pre>
     *
     * @param dbVendor database vendor
     * @param objectType object type (table, column, schema, etc.)
     * @param ident1 first identifier to compare (may be quoted)
     * @param ident2 second identifier to compare (may be quoted)
     * @return true if identifiers are equal according to vendor rules
     */
    public static boolean areEqualStatic(EDbVendor dbVendor, ESQLDataObjectType objectType, String ident1, String ident2) {
        // Handle null cases
        if (ident1 == null || ident2 == null) {
            return ident1 == ident2;
        }

        // Fast path: reference equality
        if (ident1 == ident2) {
            return true;
        }

        // Get cached instance (O(1) lookup in EnumMap)
        IdentifierService service = VENDOR_CACHE.get(dbVendor);
        if (service == null) {
            // Fallback for unexpected null (should never happen with pre-populated cache)
            // Create on-demand instance
            IdentifierProfile profile = IdentifierProfile.forVendor(dbVendor, IdentifierProfile.VendorFlags.defaults());
            CollatorProvider collatorProvider = (dbVendor == EDbVendor.dbvmssql || dbVendor == EDbVendor.dbvazuresql)
                ? new CollatorProvider()
                : null;
            service = new IdentifierService(profile, collatorProvider);
        }

        return service.areEqual(ident1, ident2, objectType);
    }

    /**
     * Static {@link #canonKey} accessor over the cached default-flags service per vendor
     * (same cache {@link #areEqualStatic} uses, so key equality mirrors that relation).
     */
    public static CanonKey canonKeyStatic(EDbVendor dbVendor, ESQLDataObjectType objectType, String identifier) {
        IdentifierService service = VENDOR_CACHE.get(dbVendor);
        if (service == null) {
            IdentifierProfile profile = IdentifierProfile.forVendor(dbVendor, IdentifierProfile.VendorFlags.defaults());
            CollatorProvider collatorProvider = (dbVendor == EDbVendor.dbvmssql || dbVendor == EDbVendor.dbvazuresql)
                ? new CollatorProvider()
                : null;
            service = new IdentifierService(profile, collatorProvider);
        }
        return service.canonKey(identifier, objectType);
    }

    // ===== Instance Fields =====

    private final IdentifierProfile profile;
    private final CollatorProvider collatorProvider;  // SQL Server 专用（可选）

    // ===== 构造函数 =====

    /**
     * 构造标识符服务
     *
     * @param profile 厂商标识符配置档案
     * @param collatorProvider Collator 提供者（SQL Server 专用，可为 null）
     */
    public IdentifierService(IdentifierProfile profile, CollatorProvider collatorProvider) {
        this.profile = profile;
        this.collatorProvider = collatorProvider;
    }

    // ===== 规范化（用于索引键构造） =====

    /**
     * 规范化标识符（去引号 + 大小写折叠）
     *
     * <p>用于构造索引键，确保相同语义的标识符生成相同的键。
     *
     * @param identifier 原始标识符（可能带引号）
     * @param objectType 对象类型
     * @return 规范化后的标识符
     */
    public String normalize(String identifier, ESQLDataObjectType objectType) {
        if (identifier == null || identifier.isEmpty()) {
            return identifier;
        }
        return normalize(identifier, objectType, isQuoted(identifier, objectType));
    }

    /** Quote-state-threaded body: recognition runs ONCE per operand per public
     *  entry point — areEqual previously re-recognized each operand up to four
     *  times through the normalize/keyForMap/foldPayload chain (hot-path cost
     *  surfaced by the U5 fork-pair benchmark). */
    private String normalize(String identifier, ESQLDataObjectType objectType, boolean quoted) {
        if (identifier == null || identifier.isEmpty()) {
            return identifier;
        }
        IdentifierRules rules = profile.getRules(objectType);

        // 1. 去引号
        String unquoted = quoted ? removeQuotes(identifier, objectType) : identifier;

        // 2. 大小写折叠
        CaseFold fold = quoted ? rules.quotedFold : rules.unquotedFold;
        return applyFold(unquoted, fold);
    }

    /**
     * 应用大小写折叠
     *
     * <p>Since U5, folding is pinned to {@link GspCaseData} (UCD
     * {@value GspCaseData#UNICODE_VERSION}) instead of the JVM's case tables, so canonical
     * keys are identical on every JDK. Locale-conditional rules (tr/az/lt) are excluded by
     * construction — the pinned tables carry only ROOT-locale mappings plus the
     * Final_Sigma context rule.
     */
    private String applyFold(String str, CaseFold fold) {
        switch (fold) {
            case UPPER:
                return GspCaseData.toUpperFull(str);
            case LOWER:
                return GspCaseData.toLowerFull(str);
            case NONE:
            default:
                return str;
        }
    }

    // ===== 比较（用于查找匹配） =====

    /**
     * 比较两个标识符是否相等（P0d.1：canonical-key equality）。
     *
     * <p>Defined as {@code canonKey(ident1).equals(canonKey(ident2))}, which makes this an
     * equivalence relation by construction. The previous rule selection ("quoted compare if
     * EITHER operand is quoted" over quote-stripped-but-unfolded text) was an OR-relation and
     * non-transitive on quoted-case-preserving dialects: Oracle {@code foo ~ "foo"} and
     * {@code foo ~ "FOO"} but {@code "foo" ≁ "FOO"}. Under the canonical relation each operand
     * folds by its own quote state (unquoted {@code foo} → {@code FOO} matches quoted
     * {@code "FOO"}, not quoted {@code "foo"}), matching how the names are actually stored.
     *
     * @param ident1 标识符 1
     * @param ident2 标识符 2
     * @param objectType 对象类型
     * @return true 如果相等
     */
    public boolean areEqual(String ident1, String ident2, ESQLDataObjectType objectType) {
        if (ident1 == null || ident2 == null) {
            return ident1 == ident2;
        }

        // 快速路径：引用相同
        if (ident1 == ident2) {
            return true;
        }

        // Same service ⇒ same rule domain, so compare canonical payloads directly instead
        // of allocating CanonKeys (this sits on the resolver2 matcher hot path). The
        // payloads come from the SAME helpers canonKey uses, so
        // areEqual(a,b) == canonKey(a).equals(canonKey(b)) holds by construction.
        // Recognition once per operand (quote-state threading — see normalize).
        IdentifierRules rules = profile.getRules(objectType);
        boolean q1 = isQuoted(ident1, objectType);
        boolean q2 = isQuoted(ident2, objectType);
        boolean collation1 = effectiveCompare(rules, q1) == CaseCompare.COLLATION_BASED;
        boolean collation2 = effectiveCompare(rules, q2) == CaseCompare.COLLATION_BASED;
        if (collation1 != collation2) {
            // Mixed payload kinds (possible only in custom profiles) never key-match.
            return false;
        }
        if (collation1) {
            String s1 = q1 ? removeQuotes(ident1, objectType) : ident1;
            String s2 = q2 ? removeQuotes(ident2, objectType) : ident2;
            if (collatorProvider != null) {
                Collator collator = collatorProvider.getCollator(profile.getFlags().defaultCollation);
                return collator.compare(s1, s2) == 0;
            }
            return caseCollapse(s1).equals(caseCollapse(s2));
        }
        return foldPayload(ident1, objectType, rules, q1)
                .equals(foldPayload(ident2, objectType, rules, q2));
    }

    /** Effective compare rule for one operand's quote state (SAME_AS_UNQUOTED resolved). */
    static CaseCompare effectiveCompare(IdentifierRules rules, boolean isQuoted) {
        CaseCompare compare = isQuoted ? rules.quotedCompare : rules.unquotedCompare;
        return compare == CaseCompare.SAME_AS_UNQUOTED ? rules.unquotedCompare : compare;
    }

    /**
     * Canonical fold payload of one operand: the stored-form text ({@link #keyForMap}),
     * case-collapsed when the effective compare is INSENSITIVE but the applied fold is
     * NONE (MySQL/BigQuery columns keep the spelling yet compare case-insensitively).
     */
    private String foldPayload(String identifier, ESQLDataObjectType objectType, IdentifierRules rules) {
        return foldPayload(identifier, objectType, rules, isQuoted(identifier, objectType));
    }

    private String foldPayload(String identifier, ESQLDataObjectType objectType,
                               IdentifierRules rules, boolean quoted) {
        String storedText = keyForMap(identifier, objectType, quoted);
        CaseFold appliedFold = quoted ? rules.quotedFold : rules.unquotedFold;
        if (effectiveCompare(rules, quoted) == CaseCompare.INSENSITIVE && appliedFold == CaseFold.NONE) {
            storedText = caseCollapse(storedText);
        }
        return storedText;
    }

    // ===== 索引键构造（段级，用于分层索引） =====

    /**
     * 为 Map 索引构造键（单个标识符段）
     *
     * <p><strong>注意：</strong>此方法仅用于分层索引的单段键，不用于复合键。
     *
     * <p>对于 COLLATION_BASED（SQL Server），不做 fold，返回原始标识符
     * （后续通过桶+Collator 比较）。
     *
     * @param identifier 标识符
     * @param objectType 对象类型
     * @return 索引键
     */
    public String keyForMap(String identifier, ESQLDataObjectType objectType) {
        if (identifier == null || identifier.isEmpty()) {
            return identifier;
        }
        return keyForMap(identifier, objectType, isQuoted(identifier, objectType));
    }

    /** Quote-state-threaded body (see {@link #normalize(String, ESQLDataObjectType, boolean)}). */
    private String keyForMap(String identifier, ESQLDataObjectType objectType, boolean isQuoted) {
       if (identifier == null || identifier.isEmpty()) {
           return identifier;
       }

       // 强制单段输入校验（Phase 0: 默认启用）
       assertSingleSegmentOrThrow(identifier,objectType);

       IdentifierRules rules = profile.getRules(objectType);
       CaseCompare compare = isQuoted ? rules.quotedCompare : rules.unquotedCompare;

       // 处理 SAME_AS_UNQUOTED
       if (compare == CaseCompare.SAME_AS_UNQUOTED) {
           compare = rules.unquotedCompare;
       }

       // 对于 COLLATION_BASED，不折叠，返回原始标识符（去引号）。
       // Return the ONCE-decoded payload directly — routing it through
       // normalize() would decode a second time when the stored payload itself
       // looks like a quoted lexical form (MSSQL [[foo]]] decodes to [foo],
       // which normalize would re-decode to foo, colliding with the key of
       // [foo] even though areEqual keeps the two distinct — Codex U5 round-1
       // finding 3).
       if (compare == CaseCompare.COLLATION_BASED) {
           return isQuoted ? removeQuotes(identifier, objectType) : identifier;
       }

       // 其他情况：规范化后作为键
       return normalize(identifier, objectType, isQuoted);
    }

    /**
     * Build the canonical key of a single identifier segment (unification plan stage P0c).
     *
     * <p>Fold-based rules use the {@link #keyForMap} text as the payload; collation-based
     * rules (SQL Server / Azure) use {@link java.text.CollationKey} bytes from the SAME
     * collator {@link #areEqual} compares with, so key equality mirrors that collator. When
     * a collation-based profile has no {@link CollatorProvider}, {@code areEqual} falls back
     * to comparing {@link #caseCollapse} payloads (pinned Unicode simple case folding since
     * U5); the key mirrors that exactly by carrying the collapsed decoded text.
     *
     * <p>NOTE: since P0d.1, {@link #areEqual} IS canonical-key equality. The legacy
     * {@code TSQLEnv.compareIdentifier} façade migrates to this relation in P0d.2; until
     * then the P0c equivalence proof measures that remaining engine's distance from it.
     *
     * @param identifier single identifier segment (may be quoted); null returns null
     * @param objectType kind of database object the name refers to
     * @return the canonical key, or null for a null identifier
     */
    public CanonKey canonKey(String identifier, ESQLDataObjectType objectType) {
        if (identifier == null) {
            return null;
        }

        IdentifierProfile.ObjectGroup group = profile.groupOf(objectType);
        long fingerprint = profile.getFingerprint();
        IdentifierRules rules = profile.getRules(objectType);

        boolean quoted = isQuoted(identifier, objectType);
        if (effectiveCompare(rules, quoted) == CaseCompare.COLLATION_BASED) {
            String stripped = quoted ? removeQuotes(identifier, objectType) : identifier;
            String collation = profile.getFlags().defaultCollation;
            if (collatorProvider != null) {
                String collatorId = collation + ":" + collatorProvider.getClass().getName();
                byte[] keyBytes = collatorProvider.getCollator(collation)
                        .getCollationKey(stripped).toByteArray();
                return new CanonKey(profile.getVendor(), group, fingerprint, collatorId, stripped, keyBytes);
            }
            // No provider: areEqual falls back to comparing caseCollapse payloads
            // (pinned C+S simple fold since U5) — mirror it.
            return new CanonKey(profile.getVendor(), group, fingerprint,
                    collation + ":none", caseCollapse(stripped), null);
        }

        // Same payload construction areEqual compares directly (see foldPayload).
        return new CanonKey(profile.getVendor(), group, fingerprint, null,
                foldPayload(identifier, objectType, rules, quoted), null);
    }

    /**
     * Case-insensitive collapse of an identifier payload.
     *
     * <p>Since U5 this is defined by Unicode simple case folding (CaseFolding.txt
     * statuses C+S at pinned UCD {@value GspCaseData#UNICODE_VERSION}), applied
     * code-point-wise — NO LONGER by {@link String#equalsIgnoreCase}'s per-{@code char}
     * upper-then-lower fold. The deltas are deliberate corrections: supplementary-plane
     * case pairs (Deseret 𐐀/𐐨) now collapse equal, and {@code İ}/{@code ı} no longer
     * collapse to ASCII {@code i} (CASEFOLD_ALGORITHM_CHANGE in the U0 allowlist).
     */
    private static String caseCollapse(String s) {
        return GspCaseData.simpleFoldCollapse(s);
    }

    // ===== Persistent canonical-identity API (plan §6, slice U3) =====

    /** Lazily computed V1 policy ids per object type (profile is immutable, so
     *  each id is a constant of this service instance). The holder itself is
     *  created on the FIRST V1 API call: IdentifierService instances exist
     *  per catalog object (e.g. TSQLTable), and eagerly allocating two maps
     *  on every one of them would bill unused U3 functionality to the whole
     *  catalog (Codex U3 round-1 finding 5). */
    private static final class PolicyIdCachesV1 {
        final java.util.concurrent.ConcurrentHashMap<ESQLDataObjectType, String> persistent =
                new java.util.concurrent.ConcurrentHashMap<>();
        final java.util.concurrent.ConcurrentHashMap<ESQLDataObjectType, String> approximate =
                new java.util.concurrent.ConcurrentHashMap<>();
    }

    private volatile PolicyIdCachesV1 policyIdCachesV1;

    private PolicyIdCachesV1 policyIdCachesV1() {
        PolicyIdCachesV1 caches = policyIdCachesV1;
        if (caches == null) {
            synchronized (this) {
                caches = policyIdCachesV1;
                if (caches == null) {
                    caches = new PolicyIdCachesV1();
                    policyIdCachesV1 = caches;
                }
            }
        }
        return caches;
    }

    /**
     * Mint the persistable canonical key of one identifier segment.
     *
     * <p><b>Contract (no carve-out).</b> For two
     * {@link IdentifierInputForm#SQL_LEXICAL} inputs {@code a}, {@code b}
     * under the same service and object type:
     * {@code persistentKeyV1(type, a, SQL_LEXICAL).equals(persistentKeyV1(type, b, SQL_LEXICAL))}
     * ⟺ {@link #areEqual areEqual(a, b, type)}.
     * {@link IdentifierInputForm#CATALOG_STORED} inputs join the same
     * equivalence domain: a stored name keys equal to the lexical spelling
     * that produced it (Oracle stored {@code MixedCase} ≡ lexical
     * {@code "MixedCase"}; stored {@code FOO} ≡ lexical {@code foo}; MySQL
     * mode-1 stored {@code straße} ≡ bare lexical {@code straße}). In cells
     * whose unquoted branch folds while the quoted branch compares
     * insensitively without folding (MySQL table names, Presto family), the
     * unquoted fold and the quoted collapse are DIFFERENT maps wherever full
     * folding and simple folding disagree (final sigma: {@code ΟΣ} lowers to
     * {@code ος} while the collapse maps {@code ς} to {@code σ}) — that
     * asymmetry is {@link #areEqual}'s own U5-certified branch semantics, and
     * stored names side with the fold-image (the form the catalog actually
     * holds), not with the quoted collapse. Keys are stable across JVMs and
     * JDK versions — folding is pinned ({@link GspCaseData}), never JVM case
     * tables — and the {@link PersistentIdentifierKey#getPolicyId() policyId}
     * rotates exactly when the resolved equivalence behavior changes.
     *
     * <p><b>Payload semantics.</b> SQL_LEXICAL: strict codec decode for
     * recognized quoted spellings ({@link IdentifierCodec#decodeLexical} —
     * malformed input throws {@link IdentifierCodec.MalformedIdentifierException};
     * this API never guesses identity), then the quote-state branch's fold,
     * then collapse when the branch compares INSENSITIVE without folding.
     * CATALOG_STORED: no recognition, no decode. SENSITIVE stored-name
     * domains keep the stored text; INSENSITIVE domains take the cell's fold
     * when any branch folds (idempotent on real catalog data — stored names
     * ARE fold-images) and the case-collapse only in never-folding cells.
     * (Refines the plan's "INSENSITIVE cells collapse / SENSITIVE cells keep
     * the stored text": in folding cells the collapse is the wrong
     * canonicalizer for stored names — see persistentPayloadV1 — and for
     * folding quoted branches — Snowflake
     * {@code QUOTED_IDENTIFIERS_IGNORE_CASE} — the fold keeps cross-form
     * equality intact.)
     *
     * <p><b>Rejections — before descriptor construction AND before any
     * decode/payload work.</b> Cells with both branches resolving to
     * COLLATION_BASED throw {@link UnsupportedOperationException} naming the
     * collation (collator weights are not portably reproducible); mixed-kind
     * custom profiles (one branch collation-based, one fold-based — the
     * {@link #areEqual} mixed-payload-kind guard case) throw the dedicated
     * subtype {@link MixedKindProfileException} — the biconditional's scope
     * is fold/fold cells. {@link #approximateKeyV1} serves both.
     *
     * <p>Null {@code name} returns null. The empty string returns a defined
     * key (empty payload, real policyId): degenerate but stable.
     *
     * @param objectType kind of database object the name refers to
     * @param name single identifier segment
     * @param form how the string reached the caller — lexical SQL text or a
     *             catalog-stored name; the forms are not interchangeable
     * @return the persistent key, or null for a null name
     */
    public PersistentIdentifierKey persistentKeyV1(ESQLDataObjectType objectType, String name,
                                                   IdentifierInputForm form) {
        java.util.Objects.requireNonNull(objectType, "objectType");
        java.util.Objects.requireNonNull(form, "form");
        if (name == null) {
            return null;
        }
        IdentifierRules rules = profile.getRules(objectType);
        rejectNonFoldCellV1(rules);
        String payload = persistentPayloadV1(name, objectType, form, rules);
        String policyId = policyIdCachesV1().persistent.computeIfAbsent(objectType,
                t -> IdentifierPolicyV1.persistentPolicyId(profile, t,
                        GspCaseData.UNICODE_VERSION, GspCaseData.ALGORITHM_REV));
        return new PersistentIdentifierKey(policyId, profile.groupOf(objectType), payload);
    }

    /**
     * Mint the best-effort key that serves EVERY rule cell, including the
     * collation-based and mixed-kind cells {@link #persistentKeyV1} rejects:
     * quote-strip (lenient — a malformed quoted spelling keeps its identity
     * as written, mirroring {@link #areEqual}'s totality) plus pinned simple
     * case-fold collapse. CATALOG_STORED input is collapsed as-is.
     *
     * <p>Claims NO {@code areEqual} biconditional — see
     * {@link ApproximateIdentifierKey}. The distinct type and the distinct
     * {@code ak1-} policy namespace guarantee these keys can never be stored
     * or matched where exact keys are expected.
     *
     * @param objectType kind of database object the name refers to
     * @param name single identifier segment
     * @param form lexical SQL text or catalog-stored name
     * @return the approximate key, or null for a null name
     */
    public ApproximateIdentifierKey approximateKeyV1(ESQLDataObjectType objectType, String name,
                                                     IdentifierInputForm form) {
        java.util.Objects.requireNonNull(objectType, "objectType");
        java.util.Objects.requireNonNull(form, "form");
        if (name == null) {
            return null;
        }
        String text = name;
        if (form == IdentifierInputForm.SQL_LEXICAL
                && IdentifierCodec.isQuoted(profile, objectType, name)) {
            text = removeQuotes(name, objectType);
        }
        String policyId = policyIdCachesV1().approximate.computeIfAbsent(objectType,
                t -> IdentifierPolicyV1.approximatePolicyId(profile, t,
                        GspCaseData.UNICODE_VERSION, GspCaseData.ALGORITHM_REV));
        return new ApproximateIdentifierKey(policyId, profile.groupOf(objectType),
                caseCollapse(text));
    }

    /**
     * Human-readable rendering of the V1 policies for one object type — a
     * diagnostic, explicitly NOT the identity ({@code policyId} is the hash
     * of the binary descriptor), so this formatting can improve without
     * rotating any stored key.
     *
     * @param objectType kind of database object
     * @return one-line description of the persistent and approximate policies
     */
    public String describePolicyV1(ESQLDataObjectType objectType) {
        java.util.Objects.requireNonNull(objectType, "objectType");
        IdentifierRules rules = profile.getRules(objectType);
        CaseCompare effU = effectiveCompare(rules, false);
        CaseCompare effQ = effectiveCompare(rules, true);
        StringBuilder sb = new StringBuilder(160);
        sb.append("policyV1{vendor=").append(IdentifierPolicyV1.vendorWire(profile.getVendor()))
                .append(", group=").append(IdentifierPolicyV1.groupWire(profile.groupOf(objectType)))
                .append(", cell=").append(rules.unquotedFold).append('/').append(effU)
                .append('|').append(rules.quotedFold).append('/').append(effQ)
                .append(", codec=").append(IdentifierCodec.CODEC_REVISION);
        if (IdentifierCodec.consultsAnsiQuotes(profile.getVendor())) {
            sb.append(", ansiQuotes=").append(profile.getFlags().mysqlAnsiQuotes);
        }
        sb.append(", mapRev=").append(IdentifierPolicyV1.mapRevision(objectType));
        boolean collation = effU == CaseCompare.COLLATION_BASED || effQ == CaseCompare.COLLATION_BASED;
        if (collation) {
            sb.append(", persistent=UNSUPPORTED(")
                    .append(effU == effQ ? "collation " + profile.getFlags().defaultCollation
                            : "mixed-kind cell")
                    .append(')');
        } else {
            sb.append(", ucd=").append(IdentifierPolicyV1.cellUsesUcd(rules)
                            ? GspCaseData.UNICODE_VERSION + "/alg" + GspCaseData.ALGORITHM_REV
                            : "excluded")
                    .append(", persistent=")
                    .append(policyIdCachesV1().persistent.computeIfAbsent(objectType,
                            t -> IdentifierPolicyV1.persistentPolicyId(profile, t,
                                    GspCaseData.UNICODE_VERSION, GspCaseData.ALGORITHM_REV)));
        }
        sb.append(", approximate=")
                .append(policyIdCachesV1().approximate.computeIfAbsent(objectType,
                        t -> IdentifierPolicyV1.approximatePolicyId(profile, t,
                                GspCaseData.UNICODE_VERSION, GspCaseData.ALGORITHM_REV)))
                .append('}');
        return sb.toString();
    }

    /**
     * Typed rejection for mixed-kind custom profiles (one branch
     * collation-based, one fold-based — the cell shape {@link #areEqual}'s
     * mixed-payload-kind guard never key-matches). A distinct TYPE, not just
     * a distinct message, so callers can tell this rejection apart from the
     * all-collation one programmatically (Codex U3 round-1 finding 2). It
     * still extends {@link UnsupportedOperationException} so a caller
     * handling "persistent keys unavailable for this cell" generically
     * catches both.
     */
    public static final class MixedKindProfileException extends UnsupportedOperationException {

        private static final long serialVersionUID = 1L;

        private final CaseCompare unquotedKind;
        private final CaseCompare quotedKind;

        MixedKindProfileException(CaseCompare unquotedKind, CaseCompare quotedKind) {
            super("persistentKeyV1: mixed-kind custom profile (unquoted=" + unquotedKind
                    + ", quoted=" + quotedKind + ") rejected before descriptor construction - "
                    + "the biconditional's scope is fold/fold cells; use approximateKeyV1");
            this.unquotedKind = unquotedKind;
            this.quotedKind = quotedKind;
        }

        /** Effective compare kind of the unquoted branch. */
        public CaseCompare getUnquotedKind() {
            return unquotedKind;
        }

        /** Effective compare kind of the quoted branch. */
        public CaseCompare getQuotedKind() {
            return quotedKind;
        }
    }

    /** Typed rejection of the cells outside the fold/fold biconditional scope
     *  — BEFORE any descriptor or payload work (plan §6), including strict
     *  decoding: a malformed spelling on a rejected cell gets the cell
     *  rejection, never a decode error. */
    private void rejectNonFoldCellV1(IdentifierRules rules) {
        CaseCompare effU = effectiveCompare(rules, false);
        CaseCompare effQ = effectiveCompare(rules, true);
        boolean collationU = effU == CaseCompare.COLLATION_BASED;
        boolean collationQ = effQ == CaseCompare.COLLATION_BASED;
        if (collationU && collationQ) {
            throw new UnsupportedOperationException(
                    "persistentKeyV1: this cell compares via collation '"
                            + profile.getFlags().defaultCollation
                            + "' - collator weights are not portably reproducible; "
                            + "use approximateKeyV1 (no areEqual biconditional)");
        }
        if (collationU != collationQ) {
            throw new MixedKindProfileException(effU, effQ);
        }
    }

    /** Canonical persistent payload — byte-identical with {@link #foldPayload}
     *  for well-formed lexical input, so key equality IS {@link #areEqual}.
     *
     *  <p>CATALOG_STORED (Codex U3 round-1 finding 1): a stored name under an
     *  INSENSITIVE cell whose unquoted branch FOLDS is the fold-image of the
     *  spelling that created it, so its canonical payload is the (idempotent)
     *  unquoted fold of the stored text — NOT the case-collapse. Collapsing
     *  broke the primary catalog↔SQL join wherever the fold and the simple
     *  C+S collapse disagree: MySQL mode-1 stored {@code ος} (created by bare
     *  {@code ΟΣ}, contextual lower) collapsed to {@code οσ} and stopped
     *  keying with bare {@code ΟΣ}'s payload {@code ος}. (ß is NOT such a
     *  case — U+00DF carries only an F mapping in CaseFolding, so the
     *  collapse preserves it like the lower fold does; Codex round 2.)
     *  Collapse remains correct for INSENSITIVE cells that never fold (MySQL
     *  columns, BigQuery columns): there collapse IS the cell's canonical
     *  payload on every branch. */
    private String persistentPayloadV1(String name, ESQLDataObjectType objectType,
                                       IdentifierInputForm form, IdentifierRules rules) {
        if (form == IdentifierInputForm.CATALOG_STORED) {
            if (effectiveCompare(rules, true) != CaseCompare.INSENSITIVE) {
                // SENSITIVE stored-name domain: the stored text IS the identity.
                return name;
            }
            CaseFold storedFold = rules.unquotedFold != CaseFold.NONE
                    ? rules.unquotedFold : rules.quotedFold;
            if (storedFold != CaseFold.NONE) {
                return applyFold(name, storedFold);
            }
            return caseCollapse(name);
        }
        boolean quoted = IdentifierCodec.isQuoted(profile, objectType, name);
        // Strict: MalformedIdentifierException propagates (unlike removeQuotes,
        // which keeps malformed spellings' as-written identity for the total
        // areEqual/keyForMap facade).
        String text = quoted ? IdentifierCodec.decodeLexical(profile, objectType, name) : name;
        CaseFold fold = quoted ? rules.quotedFold : rules.unquotedFold;
        String payload = applyFold(text, fold);
        if (effectiveCompare(rules, quoted) == CaseCompare.INSENSITIVE && fold == CaseFold.NONE) {
            payload = caseCollapse(payload);
        }
        return payload;
    }

    /** Test hook (anti-false-rotation, plan §2 U0.2): the persistent policy id
     *  under a caller-chosen UCD version/algorithm revision. Same rejections
     *  as {@link #persistentKeyV1}. */
    String persistentPolicyIdV1(ESQLDataObjectType objectType, String ucdVersion,
                                int ucdAlgorithmRev) {
        rejectNonFoldCellV1(profile.getRules(objectType));
        return IdentifierPolicyV1.persistentPolicyId(profile, objectType, ucdVersion,
                ucdAlgorithmRev);
    }

    /** Test hook: approximate policy id under a caller-chosen UCD version. */
    String approximatePolicyIdV1(ESQLDataObjectType objectType, String ucdVersion,
                                 int ucdAlgorithmRev) {
        return IdentifierPolicyV1.approximatePolicyId(profile, objectType, ucdVersion,
                ucdAlgorithmRev);
    }

    // ===== 复合键构造（全限定名级，可选） =====

    /**
     * 判断是否可以使用复合键快速路径
     *
     * <p>条件（必须全部满足）：
     * <ol>
     * <li>所有对象组都是 SENSITIVE
     * <li>输入没有引号字符
     * <li>没有 COLLATION_BASED 类型
     * </ol>
     *
     * @param qualifiedName 完整限定名（如 "db.schema.table"）
     * @return true 如果可以使用复合键
     */
    public boolean canUseCompositeKey(String qualifiedName) {
        // 条件 1: 检查是否包含引号字符
        if (containsQuoteChar(qualifiedName)) {
            return false;
        }

        // 条件 2: 检查所有对象组是否全部 SENSITIVE
        for (IdentifierProfile.ObjectGroup group : IdentifierProfile.ObjectGroup.values()) {
            IdentifierRules rules = profile.getRulesByGroup(group);
            if (rules.unquotedCompare != CaseCompare.SENSITIVE) {
                return false;
            }
        }

        return true;
    }

    /**
     * 构造复合键（使用长度前缀编码避免冲突）
     *
     * <p>格式: {@code len1#segment1|len2#segment2|len3#segment3|objectType}
     *
     * <p>例如: {@code "3#db1|6#schema|5#table|dotTable"}
     *
     * <p><strong>优势：</strong>避免分隔符冲突（标识符可能包含 '.' 或 '|'）
     *
     * @param segments 标识符段列表
     * @param objectType 对象类型
     * @return 复合键
     */
    public String buildCompositeKey(List<String> segments, ESQLDataObjectType objectType) {
        StringBuilder sb = new StringBuilder(segments.size() * 20);
        for (int i = 0; i < segments.size(); i++) {
            String seg = segments.get(i);
            if (i > 0) {
                sb.append('|');
            }
            sb.append(seg.length()).append('#').append(seg);
        }
        sb.append('|').append(objectType.name());
        return sb.toString();
    }

    /**
     * 构造复合键（从完整限定名）
     *
     * @param qualifiedName 完整限定名（如 "db.schema.table"）
     * @param objectType 对象类型
     * @return 复合键
     */
    public String buildCompositeKey(String qualifiedName, ESQLDataObjectType objectType) {
        List<String> segments = SQLUtil.parseNames(qualifiedName);
        return buildCompositeKey(segments, objectType);
    }

    // ===== 完整限定名处理（多段处理） =====

    /**
     * 解析完整限定名为段列表
     *
     * <p>包装 {@link SQLUtil#parseNames(String, EDbVendor)} 以支持厂商特定解析：
     * <ul>
     * <li>MSSQL: 保留 ".." 以便后续展开
     * <li>BigQuery: 反引号内的点号仍视为层级分隔
     * <li>其它: 按 '.' 分段，处理引号包裹的段
     * </ul>
     *
     * @param qualifiedName 完整限定名（如 "db.schema.table" 或 "db..table"）
     * @return 段列表
     */
    public List<String> parseQualifiedName(String qualifiedName) {
        return SQLUtil.parseNames(qualifiedName, profile.getVendor());
    }

    /**
     * 厂商级预处理（展开特殊语法）
     *
     * <p>处理厂商特定语法：
     * <ul>
     * <li>MSSQL/Azure SQL: 将 "db..table" 展开为 "db.&lt;global or dbo&gt;.table"
     * </ul>
     *
     * @param segments 原始段列表
     * @param vendor 数据库厂商
     * @return 展开后的段列表
     */
    public List<String> expandVendorSpecific(List<String> segments, EDbVendor vendor) {
        return expandVendorSpecific(segments, vendor, null);
    }

    /**
     * 厂商级预处理（展开特殊语法）
     *
     * <p>处理厂商特定语法：
     * <ul>
     * <li>MSSQL/Azure SQL: 将 "db..table" 展开为 "db.&lt;defaultSchema or dbo&gt;.table"
     * </ul>
     *
     * @param segments 原始段列表
     * @param vendor 数据库厂商
     * @param defaultSchema 默认 schema，如果为 null 则使用 "dbo"
     * @return 展开后的段列表
     */
    public List<String> expandVendorSpecific(List<String> segments, EDbVendor vendor, String defaultSchema) {
        if (vendor != EDbVendor.dbvmssql && vendor != EDbVendor.dbvazuresql) {
            return segments;
        }

        // MSSQL: 展开 ".." 为 ".<defaultSchema or dbo>."
        // Note: We no longer use ModelBindingManager.getGlobalSchema() here because that is
        // designed for DataFlowAnalyzer context. Using it here causes test pollution when
        // tests fail without cleaning up the ThreadLocal state.
        String schemaToUse = (defaultSchema != null && !defaultSchema.isEmpty()) ? defaultSchema : "dbo";

        List<String> expanded = new ArrayList<>();
        for (int i = 0; i < segments.size(); i++) {
            String seg = segments.get(i);
            if (seg.isEmpty() && i > 0 && i < segments.size() - 1) {
                // 这是 ".." 中的空段
                expanded.add(schemaToUse);
            } else {
                expanded.add(seg);
            }
        }
        return expanded;
    }

    /**
     * 规范化单个段（去引号 + 大小写折叠）
     *
     * <p>与 {@link #normalize(String, ESQLDataObjectType)} 类似，但语义明确为"单段"处理。
     *
     * @param segment 单个段标识符
     * @param objectType 对象类型
     * @return 规范化后的段
     */
    public String normalizeSegment(String segment, ESQLDataObjectType objectType) {
        return normalize(segment, objectType);
    }

    /**
     * 规范化完整限定名（等价于 SQLUtil.getIdentifierNormalName）
     *
     * <p>处理流程：
     * <ol>
     * <li>解析为段列表：{@link #parseQualifiedName(String)}
     * <li>展开厂商特定语法：{@link #expandVendorSpecific(List, EDbVendor)}
     * <li>根据 supportCatalog/supportSchema 决定各段类型
     * <li>逐段规范化：{@link #normalizeSegment(String, ESQLDataObjectType)}
     * <li>重新拼接为 "catalog.schema.table.column" 格式
     * </ol>
     *
     * @param qualifiedName 完整限定名
     * @param objectType 最终对象类型（dotTable/dotColumn/dotSchema 等）
     * @return 规范化后的完整限定名
     */
    public String normalizeQualifiedName(String qualifiedName, ESQLDataObjectType objectType) {
        if (qualifiedName == null || qualifiedName.isEmpty()) {
            return qualifiedName;
        }

        List<String> segments = expandVendorSpecific(parseQualifiedName(qualifiedName), profile.getVendor());
        if (segments.isEmpty()) {
            return qualifiedName;
        }

        boolean supportCatalog = TSQLEnv.supportCatalog(profile.getVendor());
        boolean supportSchema = TSQLEnv.supportSchema(profile.getVendor());

        StringBuilder builder = new StringBuilder();

        // 根据 supportCatalog/supportSchema 与 objectType 决定各段类型
        if (supportCatalog && supportSchema) {
            if (objectType == ESQLDataObjectType.dotColumn) {
                if (segments.size() > 4) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 3), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 4) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(3), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotColumn));
                }
            } else if (objectType == ESQLDataObjectType.dotTable
                    || objectType == ESQLDataObjectType.dotOraclePackage
                    || objectType == ESQLDataObjectType.dotFunction
                    || objectType == ESQLDataObjectType.dotProcedure
                    || objectType == ESQLDataObjectType.dotTrigger) {
                if (segments.size() > 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 2), objectType));
                } else if (segments.size() == 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), objectType));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), objectType));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), objectType));
                }
            } else if (objectType == ESQLDataObjectType.dotSchema) {
                if (segments.size() > 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 1), objectType));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), objectType));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), objectType));
                }
            } else if (objectType == ESQLDataObjectType.dotCatalog) {
                if (segments.size() > 1) {
                    builder.append(normalizeSegment(SQLUtil.mergeSegments(segments, 0), objectType));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), objectType));
                }
            }
        } else if (supportCatalog) {
            if (objectType == ESQLDataObjectType.dotColumn) {
                if (segments.size() > 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 2), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotColumn));
                }
            } else if (objectType == ESQLDataObjectType.dotTable
                    || objectType == ESQLDataObjectType.dotOraclePackage
                    || objectType == ESQLDataObjectType.dotFunction
                    || objectType == ESQLDataObjectType.dotProcedure
                    || objectType == ESQLDataObjectType.dotTrigger) {
                if (segments.size() > 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 1), objectType));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), objectType));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), objectType));
                }
            } else if (objectType == ESQLDataObjectType.dotSchema || objectType == ESQLDataObjectType.dotCatalog) {
                if (segments.size() > 1) {
                    builder.append(normalizeSegment(SQLUtil.mergeSegments(segments, 0), ESQLDataObjectType.dotCatalog));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotCatalog));
                }
            }
        } else if (supportSchema) {
            if (objectType == ESQLDataObjectType.dotColumn) {
                if (segments.size() > 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 2), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 3) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(2), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotTable))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), ESQLDataObjectType.dotColumn));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotColumn));
                }
            } else if (objectType == ESQLDataObjectType.dotTable
                    || objectType == ESQLDataObjectType.dotOraclePackage
                    || objectType == ESQLDataObjectType.dotFunction
                    || objectType == ESQLDataObjectType.dotProcedure
                    || objectType == ESQLDataObjectType.dotTrigger) {
                if (segments.size() > 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(SQLUtil.mergeSegments(segments, 1), objectType));
                } else if (segments.size() == 2) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema))
                            .append(".")
                            .append(normalizeSegment(segments.get(1), objectType));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), objectType));
                }
            } else if (objectType == ESQLDataObjectType.dotSchema || objectType == ESQLDataObjectType.dotCatalog) {
                if (segments.size() > 1) {
                    builder.append(normalizeSegment(SQLUtil.mergeSegments(segments, 0), ESQLDataObjectType.dotSchema));
                } else if (segments.size() == 1) {
                    builder.append(normalizeSegment(segments.get(0), ESQLDataObjectType.dotSchema));
                }
            }
        }

        return builder.toString();
    }

    /**
     * 生成层级索引所需的段级键列表
     *
     * <p>用于分层索引（hierarchical index），确保各层键由 {@link #keyForMap} 统一生成。
     *
     * <p><strong>注意：</strong>调用前应先调用 {@link #parseQualifiedName(String)} 和
     * {@link #expandVendorSpecific(List, EDbVendor)} 完成解析与展开。
     *
     * @param qualifiedName 完整限定名
     * @param partTypes 各段对应的对象类型列表（长度应与段数一致）
     * @return 段级键列表
     * @throws IllegalArgumentException 如果段数与类型数不匹配
     */
    public List<String> keysForHierarchy(String qualifiedName, List<ESQLDataObjectType> partTypes) {
        List<String> segments = expandVendorSpecific(parseQualifiedName(qualifiedName), profile.getVendor());
        if (segments.size() != partTypes.size()) {
            throw new IllegalArgumentException("Segment count (" + segments.size()
                    + ") does not match type count (" + partTypes.size() + ")");
        }

        List<String> keys = new ArrayList<>();
        for (int i = 0; i < segments.size(); i++) {
            keys.add(keyForMap(segments.get(i), partTypes.get(i)));
        }
        return keys;
    }

    /**
     * 断言标识符为单段，否则抛出异常
     *
     * <p>用于在 {@link #keyForMap} 入口处确保输入为单段标识符（不包含 '.' 分隔符）。
     *
     * <p>如果 {@link TBaseType#ALLOW_MULTI_SEGMENT_IN_KEY} 为 true（兼容模式），
     * 则仅记录警告日志而不抛异常。
     *
     * @param identifier 待检查的标识符
     * @throws IllegalArgumentException 如果标识符包含多段且未启用兼容模式
     */
    public void assertSingleSegmentOrThrow(String identifier, ESQLDataObjectType objectType) {
        if (identifier == null || identifier.isEmpty()) {
            return;
        }

        // Compat mode is a no-op (the warning below is disabled), but this method sits on
        // the per-comparison hot path since P0d — segmenting just to discard the result
        // was ~15% of dlineage wall on comparison-dense files. Strict mode is unchanged.
        if (TBaseType.ALLOW_MULTI_SEGMENT_IN_KEY) {
            return;
        }

        List<String> segments = parseQualifiedName(identifier);
        if (segments.size() > 1) {
            String message = "keyForMap requires single segment, but got " + segments.size()
                    + " segments: " + identifier;
            if (TBaseType.ALLOW_MULTI_SEGMENT_IN_KEY) {
                // 兼容模式：仅记录警告
               // System.err.println("[WARN] " + message);
               // System.err.println("[WARN] return by SQLUtil.getIdentifierNormalName:" + SQLUtil.getIdentifierNormalName(this.getProfile().getVendor(), identifier, objectType));
            } else {
                // 严格模式：抛出异常
                throw new IllegalArgumentException(message);
            }
        }
    }

    // ===== 辅助方法 =====

    /**
     * 判断标识符是否被引号包围
     *
     * <p>Since U5, recognition is the codec's per-vendor/per-role delimiter table
     * ({@link IdentifierCodec#isQuoted}) instead of the legacy first-char switch
     * ({@code TSQLEnv.isDelimitedIdentifier}). The deltas are deliberate
     * corrections: vendor-documented forms the legacy switch missed are now
     * recognized (Spark/Databricks backticks, PostgreSQL {@code U&"…"},
     * Sybase/SQLite brackets, ClickHouse backticks, MDX brackets, Power Query
     * {@code #"…"}, DAX per-role bracket vs apostrophe, MySQL {@code ANSI_QUOTES}
     * double quotes). SQL Server keeps apostrophe recognition in the COLUMN role
     * ({@code AS 'REGKEY'} names a real column; dropping it lost lineage edges —
     * see {@code IdentifierCodec.RULES_MSSQL_COLUMN}).
     */
    private boolean isQuoted(String identifier, ESQLDataObjectType objectType) {
        if (objectType != null) {
            return IdentifierCodec.isQuoted(profile, objectType, identifier);
        }
        // "Any type" callers (CatalogStore.getByName(name, null)): no single codec
        // role can represent "any" — DAX quotes table names with apostrophes but
        // columns/measures with brackets — so probe the role rule-sets and accept
        // either (Codex U5 round-1 finding 2). For every other vendor the two
        // probes hit the same table and the second is never reached.
        return IdentifierCodec.isQuoted(profile, ESQLDataObjectType.dotTable, identifier)
                || IdentifierCodec.isQuoted(profile, ESQLDataObjectType.dotColumn, identifier);
    }

    /** The codec role that recognizes this spelling for a null ("any type") caller. */
    private ESQLDataObjectType recognizingRole(String identifier, ESQLDataObjectType objectType) {
        if (objectType != null) {
            return objectType;
        }
        return IdentifierCodec.isQuoted(profile, ESQLDataObjectType.dotTable, identifier)
                ? ESQLDataObjectType.dotTable : ESQLDataObjectType.dotColumn;
    }

    /**
     * 移除标识符的引号
     *
     * <p>Since U5, a codec-recognized quoted spelling is DECODED
     * ({@link IdentifierCodec#decodeLexical}: delimiters removed, escapes resolved —
     * {@code [a]]b]} → {@code a]b}, {@code U&"d\0061ta"} → {@code data}) instead of
     * having its first/last code unit stripped. When strict decoding REJECTS the
     * input ({@code [a.b[} unterminated, {@code "EMPLOYEES"@"dblink"} trailing
     * text), the ORIGINAL spelling is returned unchanged: the string is not a
     * well-formed single quoted identifier, and guessing a payload by stripping
     * the first/last code units fabricates identity ({@code "EMPLOYEES"@"LD.X"}
     * strips to the nonsense {@code EMPLOYEES"@"LD.X}, which pre-U5 survived only
     * because a multi-segment safety net in {@code SQLUtil.normalizeIdentifier}
     * happened to discard it). Totality holds either way —
     * {@link #areEqual}/{@link #keyForMap} accept arbitrary text and malformed
     * spellings simply keep their identity as written.
     */
    private String removeQuotes(String identifier, ESQLDataObjectType objectType) {
        try {
            return IdentifierCodec.decodeLexical(profile,
                    recognizingRole(identifier, objectType), identifier);
        } catch (IdentifierCodec.MalformedIdentifierException e) {
            return identifier;
        }
    }

    /**
     * 检查字符串是否包含引号字符
     *
     * <p>检查常见引号字符：", ', `, [, ]
     */
    private boolean containsQuoteChar(String str) {
        for (int i = 0; i < str.length(); i++) {
            char c = str.charAt(i);
            if (c == '"' || c == '\'' || c == '`' || c == '[' || c == ']') {
                return true;
            }
        }
        return false;
    }

    // ===== Getter 方法 =====

    /**
     * 获取标识符配置档案
     *
     * @return 配置档案
     */
    public IdentifierProfile getProfile() {
        return profile;
    }

    /**
     * 获取 Collator 提供者
     *
     * @return Collator 提供者（可能为 null）
     */
    public CollatorProvider getCollatorProvider() {
        return collatorProvider;
    }
}