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
 *
 * // 构造索引键 (key equality is exactly areEqual; NOT the stored name normalize returns)
 * 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. These entries are the DEFAULT
        // configuration and stay that way: a per-analysis profile never replaces one, so the
        // vendor-only static API cannot change meaning under a caller's feet (design §5.3).
        for (EDbVendor vendor : EDbVendor.values()) {
            IdentifierProfile profile = IdentifierProfile.forVendor(
                vendor,
                IdentifierProfile.VendorFlags.defaults()
            );
            VENDOR_CACHE.put(vendor, new IdentifierService(profile, collatorProviderFor(vendor)));
        }
    }

    /**
     * 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);
    }

    /**
     * Canonical key for one component, with its provenance stated rather than guessed (stage S0R).
     *
     * <p>The two forms are not interchangeable and this is the one place the difference is handled:
     * a {@link IdentifierInputForm#SQL_LEXICAL} text is a spelling the user wrote, so its own quote
     * state decides whether case folds; a {@link IdentifierInputForm#CATALOG_STORED} text is
     * already the name a catalog holds, so it is re-encoded to its canonical quoted spelling first.
     * On Snowflake a stored {@code foo} is the quoted name {@code "foo"}, NOT the unquoted
     * {@code foo} that folds to {@code FOO} — a factory that guessed would merge the two.
     *
     * @param text       the component text; null returns null
     * @param objectType the comparison/quoting type for this component's role
     * @param form       where the text came from
     */
    public CanonKey canonKeyOf(String text, ESQLDataObjectType objectType, IdentifierInputForm form) {
        if (objectType == null) {
            throw new IllegalArgumentException("objectType is null");
        }
        if (form == null) {
            throw new IllegalArgumentException("input form is null: provenance must be stated, "
                    + "because a stored name and a SQL spelling fold differently");
        }
        if (text == null) {
            return null;
        }
        String lexical;
        switch (form) {
            case CATALOG_STORED:
                lexical = IdentifierCodec.encodeStored(profile, objectType, text);
                break;
            case SQL_LEXICAL:
                lexical = text;
                break;
            default:
                // Exhaustive on purpose: a new input form must be handled deliberately, not
                // silently treated as a SQL spelling — which would fold a stored name.
                throw new IllegalStateException("unhandled identifier input form: " + form);
        }
        return canonKey(lexical, objectType);
    }

    /**
     * Static {@link #qualifiedCanonKeyFromLexical} accessor over the same cached
     * default-flags service per vendor, so qualified key equality mirrors the relation
     * {@link #areEqualStatic} defines segment by segment.
     */
    public static QualifiedCanonKey qualifiedCanonKeyFromLexicalStatic(EDbVendor dbVendor,
            ESQLDataObjectType objectType, String qualifiedName) {
        return defaultService(dbVendor).qualifiedCanonKeyFromLexical(qualifiedName, objectType);
    }

    /** Static {@link #qualifiedCanonKeyFromStored(List, ESQLDataObjectType)} accessor. */
    public static QualifiedCanonKey qualifiedCanonKeyFromStoredStatic(EDbVendor dbVendor,
            ESQLDataObjectType objectType, List<String> storedSegments) {
        return defaultService(dbVendor).qualifiedCanonKeyFromStored(storedSegments, objectType);
    }

    /**
     * The cached default-flags service for a vendor, or a fresh one when the cache has no
     * entry — the same lookup-then-build the static accessors above it already perform.
     */
    private static IdentifierService defaultService(EDbVendor dbVendor) {
        IdentifierService service = VENDOR_CACHE.get(dbVendor);
        if (service == null) {
            service = forProfile(IdentifierProfile.forVendor(
                    dbVendor, IdentifierProfile.VendorFlags.defaults()));
        }
        return service;
    }

    /**
     * The vendor's DEFAULT-configuration service — the same instance the vendor-only static
     * methods above use.
     *
     * <p>Explicitly the default policy, not "the current one": nothing switches what this
     * returns, so an independent caller keeps the semantics it has always had. Code inside one
     * analysis must instead carry that analysis's own service (see
     * {@link TSQLEnv#getIdentifierService()}), because a per-analysis
     * {@link IdentifierProfile} is invisible here by design.
     *
     * @param dbVendor database vendor
     * @return the default-configuration service for that vendor; never null
     */
    public static IdentifierService defaultServiceFor(EDbVendor dbVendor) {
        java.util.Objects.requireNonNull(dbVendor, "dbVendor");
        return defaultService(dbVendor);
    }

    /**
     * Build the service for one {@link IdentifierProfile}, attaching the
     * {@link CollatorProvider} the profile's vendor needs.
     *
     * <p>Single place that decides which vendors collate, so a new entry point cannot
     * accidentally build a SQL Server service without a collator — which silently downgrades
     * collation comparison to the pinned case-collapse fallback.
     *
     * @param profile the frozen configuration for one analysis (or a vendor default)
     * @return a service executing the one comparison engine under that profile
     */
    public static IdentifierService forProfile(IdentifierProfile profile) {
        java.util.Objects.requireNonNull(profile, "profile");
        return new IdentifierService(profile, collatorProviderFor(profile.getVendor()));
    }

    /**
     * The {@link CollatorProvider} a vendor's identifier rules require, or null when the vendor
     * folds instead of collating.
     *
     * @param vendor database vendor
     * @return a new collator provider for collation-based vendors, else null
     */
    public static CollatorProvider collatorProviderFor(EDbVendor vendor) {
        return (vendor == EDbVendor.dbvmssql || vendor == EDbVendor.dbvazuresql)
                ? new CollatorProvider()
                : null;
    }

    // ===== 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;
        // 3. The length the database keeps (PostgreSQL: 63 UTF-8 bytes), quoted or not.
        return clipName(applyFold(unquoted, fold), rules, identifier, objectType, quoted);
    }

    /** Cut to the cell's stored length ({@link IdentifierRules#maxUtf8Bytes}); a no-op without one. */
    private static String clip(String text, IdentifierRules rules) {
        return rules.maxUtf8Bytes > 0 ? GspCaseData.clipUtf8Bytes(text, rules.maxUtf8Bytes) : text;
    }

    /**
     * {@link #clip} for a name that arrived as a SQL spelling: the database shortens IDENTIFIER
     * tokens — a well-formed delimited spelling, or a valid undelimited one — and nothing else.
     * Lineage also hands this service texts that are not names at all (an unaliased result column
     * is labelled with its expression text, {@code COALESCE(SUM(...))}); cutting those would only
     * damage the label and could merge two expressions that share a 63-byte prefix. The spelling
     * check runs only when the text is actually longer than the limit.
     */
    private String clipName(String text, IdentifierRules rules, String spelling,
                            ESQLDataObjectType objectType, boolean quoted) {
        if (rules.maxUtf8Bytes <= 0) {
            return text;
        }
        String clipped = GspCaseData.clipUtf8Bytes(text, rules.maxUtf8Bytes);
        if (clipped == text || !isIdentifierToken(spelling, objectType, quoted)) {
            return text;
        }
        return clipped;
    }

    /**
     * Whether a spelling is an identifier token (see {@link #clipName}): a well-formed delimited
     * spelling, or an undelimited one the lexer reads as one identifier. The undelimited rule is
     * PostgreSQL's scanner ({@code ident_start = [A-Za-z\200-\377_]},
     * {@code ident_cont = [A-Za-z\200-\377_0-9$]}): every non-ASCII character is an identifier
     * character, letter or not — {@code xxx😀} is one name and is cut like any other — while an
     * ASCII space, parenthesis or operator ends the token. PostgreSQL is the only vendor whose
     * cell truncates. Deliberately wider than {@link IdentifierCodec#isVendorValidSpelling}, which
     * admits only non-ASCII letters because it answers a different question (is a bare spelling
     * safe to WRITE).
     */
    private boolean isIdentifierToken(String spelling, ESQLDataObjectType objectType, boolean quoted) {
        if (quoted) {
            try {
                IdentifierCodec.decodeLexical(profile, recognizingRole(spelling, objectType), spelling);
                return true;
            } catch (IdentifierCodec.MalformedIdentifierException e) {
                return false;
            }
        }
        if (spelling.isEmpty()) {
            return false;
        }
        char first = spelling.charAt(0);
        if (first < 0x80 && !isAsciiLetter(first) && first != '_') {
            return false;
        }
        for (int i = 1; i < spelling.length(); i++) {
            char c = spelling.charAt(i);
            if (c < 0x80 && !isAsciiLetter(c) && !(c >= '0' && c <= '9') && c != '_' && c != '$') {
                return false;
            }
        }
        return true;
    }

    private static boolean isAsciiLetter(char c) {
        return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z');
    }

    /**
     * 应用大小写折叠
     *
     * <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 UPPER_SIMPLE:
                return GspCaseData.toUpperSimple(str);
            case LOWER_SIMPLE:
                return GspCaseData.toLowerSimple(str);
            case LOWER_ASCII:
                return GspCaseData.toLowerAscii(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));
    }

    /**
     * Whether UNQUOTED names of one object category compare case-sensitively under this profile:
     * the one answer {@link NameService#isCaseSensitive} and the resolver2 name matcher give.
     *
     * <p>A {@link CaseCompare#SENSITIVE SENSITIVE} cell is case-sensitive and an
     * {@link CaseCompare#INSENSITIVE INSENSITIVE} one is not. A
     * {@link CaseCompare#COLLATION_BASED COLLATION_BASED} cell answers from its collation
     * ({@link CollatorProvider#isCaseSensitiveCollation}): {@code _CS_AI} and {@code _BIN}/
     * {@code _BIN2} are case-sensitive although they are compared through a collator, so the
     * compare kind alone is not the answer. Quoted operands may follow a different rule;
     * {@link #areEqual} is the complete relation, this is a one-bit summary of the unquoted branch.
     *
     * @param objectType the object category
     * @return true when unquoted names that differ only in case are different names
     */
    public boolean isCaseSensitive(ESQLDataObjectType objectType) {
        switch (profile.getRules(objectType).unquotedCompare) {
            case SENSITIVE:
                return true;
            case COLLATION_BASED:
                return CollatorProvider.isCaseSensitiveCollation(profile.getFlags().defaultCollation);
            default:
                return false;
        }
    }

    /** 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 #storedForm}),
     * 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 = storedForm(identifier, objectType, quoted);
        CaseFold appliedFold = quoted ? rules.quotedFold : rules.unquotedFold;
        if (effectiveCompare(rules, quoted) == CaseCompare.INSENSITIVE && appliedFold == CaseFold.NONE) {
            storedText = clipName(collapse(storedText, rules), rules, identifier, objectType, quoted);
        }
        return storedText;
    }

    /**
     * The image under which an INSENSITIVE, non-folding branch of {@code rules} compares
     * ({@link IdentifierRules#collapse}): Unicode simple case folding unless the cell names the
     * vendor's own mapping (MySQL: the simple lowercase mapping for tables and columns, the simple
     * uppercase mapping for routines).
     */
    private static String collapse(String s, IdentifierRules rules) {
        switch (rules.collapse) {
            case LOWER_SIMPLE:
                return GspCaseData.toLowerSimple(s);
            case UPPER_SIMPLE:
                return GspCaseData.toUpperSimple(s);
            case SIMPLE_FOLD:
            default:
                return caseCollapse(s);
        }
    }

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

    /**
     * 为 Map 索引构造键（单个标识符段）
     *
     * <p><strong>注意：</strong>此方法仅用于分层索引的单段键，不用于复合键。
     *
     * <p>Key equality is exactly {@link #areEqual}: {@code keyForMap(a).equals(keyForMap(b))} if
     * and only if {@code areEqual(a, b, objectType)}, for every cell of the profile. It is the
     * comparison payload, which is NOT the stored name ({@link #normalize}) wherever the two
     * differ: a branch that compares case-insensitively without folding (MySQL columns, MySQL
     * table names under {@code lower_case_table_names} 1 and 2, quoted or not) keys by its
     * collapse, so {@code `İ`} and {@code i} get one key, as they are one table; a
     * collation-based cell (SQL Server) keys by the collation key of the same collator
     * {@link #areEqual} compares with (or, with no collator, by the case collapse {@link #areEqual}
     * falls back to), so {@code café} and {@code cafe} get one key under {@code _AI} and
     * {@code SalesOrder} and {@code salesorder} one key under {@code _CI}. Up to 2026-10 this
     * returned the stored name, whose equality is neither: two equal names could get two keys.
     *
     * @param identifier 标识符
     * @param objectType 对象类型
     * @return 索引键
     */
    public String keyForMap(String identifier, ESQLDataObjectType objectType) {
        if (identifier == null || identifier.isEmpty()) {
            return identifier;
        }
        boolean quoted = isQuoted(identifier, objectType);
        IdentifierRules rules = profile.getRules(objectType);
        if (effectiveCompare(rules, quoted) == CaseCompare.COLLATION_BASED) {
            assertSingleSegmentOrThrow(identifier, objectType);
            // The ONCE-decoded payload (see storedForm: MSSQL [[foo]]] decodes to [foo], which
            // must not be decoded again into foo).
            String stripped = quoted ? removeQuotes(identifier, objectType) : identifier;
            if (collatorProvider != null) {
                return collationKeyText(collatorProvider.getCollator(profile.getFlags().defaultCollation)
                        .getCollationKey(stripped).toByteArray());
            }
            return caseCollapse(stripped);
        }
        return foldPayload(identifier, objectType, rules, quoted);
    }

    /**
     * A collation key as map-key text: a NUL marker, then the key bytes in hex. Two texts are
     * equal exactly when the keys are, i.e. when the collator compares the names equal.
     */
    private static String collationKeyText(byte[] key) {
        char[] out = new char[1 + 2 * key.length];
        out[0] = '\u0000';
        for (int i = 0; i < key.length; i++) {
            int b = key[i] & 0xFF;
            out[1 + 2 * i] = HEX[b >>> 4];
            out[2 + 2 * i] = HEX[b & 0xF];
        }
        return new String(out);
    }

    private static final char[] HEX = "0123456789abcdef".toCharArray();

    /**
     * The stored form a fold-based payload starts from: the decoded, folded and clipped name
     * ({@link #normalize}); for a collation-based branch the once-decoded spelling.
     */
    private String storedForm(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 (the fold payload) 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);
        IdentifierRules rules = profile.getRules(objectType);
        // The CATEGORY's effective cell, not the group's: with per-category configuration two
        // categories of one group can compare differently, and a key must not equate payloads
        // produced under two different relations (design §7.1).
        IdentifierRules ruleDomain = rules.resolvedCell();

        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, profile, ruleDomain,
                        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, profile, ruleDomain,
                    collation + ":none", caseCollapse(stripped), null);
        }

        // Same payload construction areEqual compares directly (see foldPayload).
        return new CanonKey(profile.getVendor(), group, profile, ruleDomain, 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 (the MySQL forks' 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 σ}; MySQL itself, verified 2026-10-11, uses the simple lowercase
     * mapping for both, so {@code ΟΣ} and {@code `ΟΣ`} are both {@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);
        if (rules.collapse != IdentifierRules.CaseCollapse.SIMPLE_FOLD) {
            sb.append(", collapse=").append(rules.collapse);
        }
        if (rules.maxUtf8Bytes > 0) {
            sb.append(", maxUtf8Bytes=").append(rules.maxUtf8Bytes);
        }
        sb.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: a MySQL-fork 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) {
            // Every result is cut to the cell's stored length last (a stored name already fits,
            // so this only keeps an over-long input consistent with its lexical spelling).
            if (effectiveCompare(rules, true) != CaseCompare.INSENSITIVE) {
                // SENSITIVE stored-name domain: the stored text IS the identity.
                return clip(name, rules);
            }
            CaseFold storedFold = rules.unquotedFold != CaseFold.NONE
                    ? rules.unquotedFold : rules.quotedFold;
            if (storedFold != CaseFold.NONE) {
                return clip(applyFold(name, storedFold), rules);
            }
            return clip(collapse(clip(name, rules), rules), rules);
        }
        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;
        // Same sequence as foldPayload: fold, cut, then (non-folding INSENSITIVE) collapse, cut.
        String payload = clipName(applyFold(text, fold), rules, name, objectType, quoted);
        if (effectiveCompare(rules, quoted) == CaseCompare.INSENSITIVE && fold == CaseFold.NONE) {
            payload = clipName(collapse(payload, rules), rules, name, objectType, quoted);
        }
        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: every object CATEGORY must be SENSITIVE. Iterating the three groups was
        // sufficient only while a group's rule governed all of its categories; with per-category
        // configuration a table-insensitive override inside NAME_GROUP would have left this
        // shortcut on, and a shortcut that cannot be proved correct must close (design §7.1).
        for (ESQLDataObjectType type : ESQLDataObjectType.values()) {
            IdentifierRules rules = profile.getRules(type);
            if (rules.unquotedCompare != CaseCompare.SENSITIVE) {
                return false;
            }
            // buildCompositeKey joins the RAW segments, so it is a key only where a name is its
            // own comparison payload: no fold and no truncation (a cut name equals every longer
            // name it is cut from, and raw segments would key them apart).
            if (rules.unquotedFold != CaseFold.NONE || rules.maxUtf8Bytes > 0) {
                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);
    }

    // ===== Qualified identity keys (S0 of the identity/display separation plan) =====
    //
    // These supersede buildCompositeKey/canUseCompositeKey for identity purposes. That
    // pair cannot represent quote state and refuses (canUseCompositeKey == false) every
    // name containing a quote character, which is exactly the shape that loses lineage
    // (MantisBT 4728). Nothing in production calls these yet; stages S2/S3 migrate the
    // catalog and dlineage indexes onto them.

    /**
     * Identity key for a qualified name <strong>as written in SQL</strong> — delimiters and
     * case exactly as the user spelled them, e.g. {@code db."My Schema".t}.
     *
     * <p>Each segment is keyed under its own object type ({@link QualifiedNameHierarchy}),
     * so a schema segment is compared by schema rules and a column segment by column rules.
     * Segment boundaries and arity are preserved: a name is never joined back into a dotted
     * string, because that would merge the path {@code a.f} with the single column named
     * {@code "A.F"}.
     *
     * <p><strong>No omitted segment is filled in.</strong> SQL Server's {@code db..t} keys as
     * three segments with an empty middle one, NOT as {@code db.dbo.t}: which schema an
     * omission means is a property of the session, and a context-free key that guessed
     * {@code dbo} would silently bind the wrong object whenever the default schema is
     * something else. Resolving an omission is a lookup concern and needs explicit context.
     *
     * @param qualifiedName the name as written; null or empty returns null
     * @param objectType    what the whole name denotes
     * @return the key, or {@code null} when the name is null or empty
     * @throws UnsupportedOperationException when this (vendor, objectType) pair has no
     *         hierarchy mapping — {@code dotUnknown}, {@code dotRoutine}, {@code dotDblink},
     *         {@code dotDataType}, {@code dotParameter}. Probe first with
     *         {@link QualifiedNameHierarchy#hasHierarchyMapping} if the caller can be handed
     *         an arbitrary type. Throwing rather than returning null is deliberate: a
     *         {@code HashMap} accepts a null key, so a forgotten check would quietly merge
     *         every unsupported object into one bucket.
     */
    public QualifiedCanonKey qualifiedCanonKeyFromLexical(String qualifiedName, ESQLDataObjectType objectType) {
        if (qualifiedName == null || qualifiedName.isEmpty()) {
            return null;
        }
        List<String> segments = QualifiedNameHierarchy.split(profile.getVendor(), qualifiedName);
        return build(segments, objectType, false, qualifiedName);
    }

    /**
     * Identity key from stored segments — text already in the form a catalog holds, with no
     * delimiters, one list entry per hierarchy level.
     *
     * <p>Stored text is re-encoded to its canonical lexical spelling
     * ({@link IdentifierCodec#encodeStored}) before canonicalization, so a stored name that
     * merely looks unquoted is not folded as though the user had written it unquoted. On
     * Snowflake a catalog entry stored as {@code foo} is the quoted name {@code "foo"}, not
     * the unquoted {@code foo} that folds to {@code FOO} — which is why this is a separate
     * method and why there is no factory that guesses provenance.
     *
     * <p>There is deliberately no overload taking a dot-joined string: a stored segment may
     * itself contain a dot, so joining and re-splitting is lossy in a way no amount of
     * documentation makes safe.
     *
     * @param storedSegments stored text per level, outermost first; null or empty returns null
     * @param objectType     what the whole name denotes
     * @throws UnsupportedOperationException as documented on {@link #qualifiedCanonKeyFromLexical}
     */
    public QualifiedCanonKey qualifiedCanonKeyFromStored(List<String> storedSegments, ESQLDataObjectType objectType) {
        if (storedSegments == null || storedSegments.isEmpty()) {
            return null;
        }
        return build(storedSegments, objectType, true, String.valueOf(storedSegments));
    }

    /**
     * Identity key of a qualified name whose segments arrived in DIFFERENT forms: one entry per
     * hierarchy level, outermost first, each with the form it arrived in. A segment the SQL spelled
     * is {@link IdentifierInputForm#SQL_LEXICAL}; a name a catalog or a session default holds is
     * {@link IdentifierInputForm#CATALOG_STORED}. Each segment is canonicalized by its own form, so a
     * stored default schema {@code Mixed} and the SQL spelling {@code "Mixed"} give the same key, and
     * neither is re-read through the other.
     *
     * <p>As with {@link #qualifiedCanonKeyFromStored}, there is no dot-joined overload: the forms
     * belong to segments, and joining loses which is which.
     *
     * @param segments text per level, outermost first; null or empty returns null
     * @param forms    the form of each segment, the same length as {@code segments}
     * @param objectType what the whole name denotes
     * @throws IllegalArgumentException when the forms do not match the segments one to one (also when
     *         the vendor's hierarchy would merge segments, since a merged segment has no single form)
     * @throws UnsupportedOperationException as documented on {@link #qualifiedCanonKeyFromLexical}
     */
    public QualifiedCanonKey qualifiedCanonKeyFromSegments(List<String> segments, List<IdentifierInputForm> forms,
                                                           ESQLDataObjectType objectType) {
        if (segments == null || segments.isEmpty()) {
            return null;
        }
        if (forms == null || forms.size() != segments.size()) {
            throw new IllegalArgumentException("one input form per segment: " + segments + " / " + forms);
        }
        List<QualifiedNameHierarchy.Segment> plan =
                QualifiedNameHierarchy.plan(profile.getVendor(), objectType, segments);
        if (plan == null) {
            throw new UnsupportedOperationException("no identifier hierarchy for "
                    + profile.getVendor() + "/" + objectType + " (name: " + segments
                    + "); probe QualifiedNameHierarchy.hasHierarchyMapping first");
        }
        if (plan.size() != segments.size()) {
            throw new IllegalArgumentException("the hierarchy merges segments of " + segments
                    + ", so their input forms no longer line up");
        }
        List<CanonKey> keys = new ArrayList<CanonKey>(plan.size());
        for (int i = 0; i < plan.size(); i++) {
            QualifiedNameHierarchy.Segment segment = plan.get(i);
            String lexical = forms.get(i) == IdentifierInputForm.CATALOG_STORED
                    ? IdentifierCodec.encodeStored(profile, segment.getType(), segment.getText())
                    : segment.getText();
            CanonKey key = canonKey(lexical, segment.getType());
            if (key == null) {
                throw new IllegalStateException("null canonical key for segment of " + segments);
            }
            keys.add(key);
        }
        return new QualifiedCanonKey(keys, objectType);
    }

    private QualifiedCanonKey build(List<String> segments, ESQLDataObjectType objectType,
                                    boolean stored, String forMessage) {
        List<QualifiedNameHierarchy.Segment> plan =
                QualifiedNameHierarchy.plan(profile.getVendor(), objectType, segments);
        if (plan == null) {
            throw new UnsupportedOperationException("no identifier hierarchy for "
                    + profile.getVendor() + "/" + objectType + " (name: " + forMessage
                    + "); probe QualifiedNameHierarchy.hasHierarchyMapping first");
        }
        List<CanonKey> keys = new ArrayList<CanonKey>(plan.size());
        for (QualifiedNameHierarchy.Segment segment : plan) {
            String lexical = stored
                    ? IdentifierCodec.encodeStored(profile, segment.getType(), segment.getText())
                    : segment.getText();
            CanonKey key = canonKey(lexical, segment.getType());
            if (key == null) {
                // canonKey only returns null for a null identifier, which a segment never is.
                throw new IllegalStateException("null canonical key for segment of " + forMessage);
            }
            keys.add(key);
        }
        return new QualifiedCanonKey(keys, 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;
    }
}