package gudusoft.gsqlparser.resolver2;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.ETableSource;
import gudusoft.gsqlparser.TSourceToken;
import gudusoft.gsqlparser.nodes.TObjectName;
import gudusoft.gsqlparser.nodes.TResultColumn;
import gudusoft.gsqlparser.nodes.TResultColumnList;
import gudusoft.gsqlparser.nodes.TTable;
import gudusoft.gsqlparser.nodes.TTableList;
import gudusoft.gsqlparser.stmt.TSelectSqlStatement;
import gudusoft.gsqlparser.resolver2.matcher.INameMatcher;
import gudusoft.gsqlparser.resolver2.model.AmbiguousColumnSource;
import gudusoft.gsqlparser.resolver2.model.ColumnSource;
import gudusoft.gsqlparser.resolver2.model.FieldPath;
import gudusoft.gsqlparser.resolver2.model.ResolutionContext;
import gudusoft.gsqlparser.resolver2.model.ResolutionResult;
import gudusoft.gsqlparser.resolver2.namespace.INamespace;
import gudusoft.gsqlparser.resolver2.namespace.UnnestNamespace;
import gudusoft.gsqlparser.resolver2.scope.IScope;
import gudusoft.gsqlparser.resolver2.scope.ResolvedImpl;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.IdentityHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

/**
 * Core component for resolving column references to their sources.
 *
 * Key responsibilities:
 * 1. Resolve TObjectName (column references) using scope tree
 * 2. Handle qualified and unqualified names
 * 3. Detect and report ambiguities
 * 4. Apply GUESS_COLUMN_STRATEGY for ambiguous columns
 * 5. Update TObjectName with resolution results
 * 6. Update ResolutionContext for global querying
 */
public class NameResolver {

    private final INameMatcher nameMatcher;
    private final ResolutionContext context;
    private final TSQLResolverConfig config;

    /**
     * Create a NameResolver with full configuration.
     *
     * @param config The resolver configuration (includes name matcher and strategy)
     * @param context The resolution context for tracking results
     */
    public NameResolver(TSQLResolverConfig config, ResolutionContext context) {
        this.config = config;
        this.nameMatcher = config.getNameMatcher();
        this.context = context;
    }

    /**
     * Create a NameResolver with just name matcher (backward compatibility).
     * Uses default configuration for GUESS_COLUMN_STRATEGY.
     *
     * @deprecated Use NameResolver(TSQLResolverConfig, ResolutionContext) instead
     */
    @Deprecated
    public NameResolver(INameMatcher nameMatcher, ResolutionContext context) {
        this.nameMatcher = nameMatcher;
        this.context = context;
        this.config = null; // Will use TBaseType.GUESS_COLUMN_STRATEGY
    }

    /**
     * Get the effective GUESS_COLUMN_STRATEGY.
     * Returns config value if available, otherwise TBaseType.GUESS_COLUMN_STRATEGY.
     */
    private int getGuessColumnStrategy() {
        if (config != null) {
            return config.getGuessColumnStrategy();
        }
        return gudusoft.gsqlparser.TBaseType.GUESS_COLUMN_STRATEGY;
    }

    private static final boolean DEBUG_RESOLUTION = false;

    /**
     * Resolve a column reference (TObjectName) within a given scope.
     *
     * @param objName The column reference to resolve
     * @param scope The scope where the reference appears
     * @return Resolution result
     */
    public ResolutionResult resolve(TObjectName objName, IScope scope) {
        if (objName == null || scope == null) {
            return ResolutionResult.notFound("<null>");
        }

        // Extract name parts from TObjectName
        List<String> nameParts = extractNameParts(objName);
        if (nameParts.isEmpty()) {
            return ResolutionResult.notFound("<empty>");
        }

        if (DEBUG_RESOLUTION) {
            System.out.println("[DEBUG-RESOLVE] Resolving: " + objName +
                " nameParts=" + nameParts + " scopeType=" + scope.getScopeType());
        }

        // Use scope to resolve the name
        ResolvedImpl resolved = new ResolvedImpl();
        scope.resolve(nameParts, nameMatcher, false, resolved);

        if (DEBUG_RESOLUTION) {
            System.out.println("[DEBUG-RESOLVE]   Resolved matches: " + resolved.getCount());
            if (resolved.getCount() > 1) {
                for (ResolvedImpl.Match m : resolved.getMatches()) {
                    System.out.println("[DEBUG-RESOLVE]     Match: " + m.namespace.getDisplayName() +
                        " type=" + m.namespace.getClass().getSimpleName() +
                        " id=" + System.identityHashCode(m.namespace) +
                        " remaining=" + m.remainingNames +
                        " scope=" + m.scope.getScopeType());
                }
            }
        }

        // Process resolution results
        ResolutionResult result = processResolvedMatches(objName, nameParts, resolved);

        // Delta 3: Struct-field fallback for BigQuery/Snowflake
        // If resolution failed and we have a 2-part qualified name like "customer.customer_id",
        // try interpreting it as column.field (struct field access) instead of table.column.
        // Only for 2-part names; 3+ part no-alias case (e.g., customer.address.city) is not
        // handled here to avoid changing resolution paths for existing 3-part BigQuery patterns
        // (e.g., purchases.first.msts) that are correctly handled by DataFlowAnalyzer heuristics.
        // The alias case (o.customer.address.city) IS handled by resolveColumnPath() in
        // processResolvedMatches() and does not depend on this fallback.
        if (!result.isExactMatch() && nameParts.size() == 2 && isStructFieldVendor()) {
            ResolutionResult structFieldResult = tryStructFieldFallback(objName, nameParts, scope);
            if (structFieldResult != null && structFieldResult.isExactMatch()) {
                result = structFieldResult;
                if (DEBUG_RESOLUTION) {
                    System.out.println("[DEBUG-RESOLVE]   Struct-field fallback succeeded for: " + objName);
                }
            }
        }

        // Delta 4: Side-channel hint for 3+ part no-alias struct access (BigQuery only).
        // When resolution failed and we have 3+ parts (e.g., customer.address.city),
        // set a StructFieldHint WITHOUT changing the resolution result or sourceTable.
        // This provides struct path info to DataFlowAnalyzer without altering lineage topology.
        if (!result.isExactMatch() && nameParts.size() >= 3 && isStructFieldHintVendor()) {
            trySetStructFieldHint(objName, nameParts, scope);
        }

        if (DEBUG_RESOLUTION) {
            System.out.println("[DEBUG-RESOLVE]   Result: " + result.getStatus() +
                (result.isExactMatch() && result.getColumnSource() != null ?
                    " source=" + result.getColumnSource().getExposedName() : ""));
        }

        // Update TObjectName and context
        updateObjectNameWithResult(objName, result);

        return result;
    }

    /**
     * Check if the current vendor supports struct-field access syntax (column.field).
     * Currently supported: BigQuery, Snowflake
     */
    private boolean isStructFieldVendor() {
        if (config == null) {
            return false;
        }
        EDbVendor vendor = config.getVendor();
        return vendor == EDbVendor.dbvbigquery || vendor == EDbVendor.dbvsnowflake;
    }

    /**
     * Check if the current vendor supports struct-field hint annotations.
     * Currently BigQuery only (Snowflake uses schema.table.column for 3-part names).
     */
    private boolean isStructFieldHintVendor() {
        if (config == null) {
            return false;
        }
        return config.getVendor() == EDbVendor.dbvbigquery;
    }

    /**
     * Delta 4: Try to set a StructFieldHint for 3+ part no-alias struct access.
     *
     * For "customer.address.city" (3 parts, no alias):
     * - Treat the first part ("customer") as a potential base column
     * - If found in any visible namespace, set a hint with fieldPath=["address", "city"]
     * - Does NOT change ResolutionResult or sourceTable
     *
     * @param objName The TObjectName to annotate with a hint
     * @param nameParts The extracted name parts (e.g., ["customer", "address", "city"])
     * @param scope The scope to search in
     */
    private void trySetStructFieldHint(TObjectName objName, List<String> nameParts, IScope scope) {
        String baseColumnName = nameParts.get(0);
        List<String> fieldPathSegments = nameParts.subList(1, nameParts.size());

        // Try to find the base column as an unqualified name
        List<String> singlePartName = Collections.singletonList(baseColumnName);
        ResolvedImpl resolved = new ResolvedImpl();
        scope.resolve(singlePartName, nameMatcher, false, resolved);

        if (resolved.isEmpty()) {
            return; // Base column not found in any namespace
        }

        // Check if any namespace actually has this column
        for (ResolvedImpl.Match match : resolved.getMatches()) {
            gudusoft.gsqlparser.resolver2.model.ColumnSource baseColumnSource =
                match.namespace.resolveColumn(baseColumnName);
            if (baseColumnSource != null) {
                // Found the base column - create hint
                gudusoft.gsqlparser.resolver2.model.FieldPath fieldPath =
                    gudusoft.gsqlparser.resolver2.model.FieldPath.of(fieldPathSegments);
                gudusoft.gsqlparser.resolver2.model.StructFieldHint hint =
                    new gudusoft.gsqlparser.resolver2.model.StructFieldHint(
                        baseColumnName, fieldPath,
                        "struct_field_hint_no_alias", 0.7);
                objName.setStructFieldHint(hint);

                if (DEBUG_RESOLUTION) {
                    System.out.println("[DEBUG-RESOLVE]   StructFieldHint set for: " + objName +
                        " -> base=" + baseColumnName + ", fieldPath=" + fieldPathSegments);
                }
                return;
            }
        }
    }

    /**
     * Delta 3: Try to resolve a qualified name as struct-field access (column.field).
     *
     * In BigQuery/Snowflake, "customer.customer_id" might be:
     * 1. Table "customer" with column "customer_id" (standard interpretation)
     * 2. Column "customer" (STRUCT type) with field "customer_id" (struct-field access)
     *
     * If the standard interpretation failed, try the struct-field interpretation:
     * - Treat the first part as an unqualified column name
     * - If found, return the base column as the source (the STRUCT column)
     * - Preserve the field path (segments beyond the base column) for downstream use
     *
     * @param objName The original TObjectName
     * @param nameParts The extracted name parts (e.g., ["customer", "customer_id"])
     * @param scope The scope to search in
     * @return Resolution result if struct-field interpretation succeeds, null otherwise
     */
    private ResolutionResult tryStructFieldFallback(TObjectName objName, List<String> nameParts, IScope scope) {
        // The base column is the first part (e.g., "customer" in "customer.customer_id")
        String baseColumnName = nameParts.get(0);

        // The field path is everything after the base column
        // For "customer.customer_id", fieldPath = ["customer_id"]
        // For "customer.address.city", fieldPath = ["address", "city"]
        List<String> fieldPathSegments = nameParts.size() > 1
            ? nameParts.subList(1, nameParts.size())
            : Collections.emptyList();

        // Try to resolve the base column as an unqualified name
        List<String> singlePartName = Collections.singletonList(baseColumnName);
        ResolvedImpl resolved = new ResolvedImpl();
        scope.resolve(singlePartName, nameMatcher, false, resolved);

        if (resolved.isEmpty()) {
            return null; // Base column not found
        }

        // Found potential matches - check if any namespace has this column
        for (ResolvedImpl.Match match : resolved.getMatches()) {
            INamespace namespace = match.namespace;

            // Try to resolve the base column name in this namespace
            ColumnSource baseColumnSource = namespace.resolveColumn(baseColumnName);
            if (baseColumnSource != null) {
                // Found the base column - return it as the source with field path preserved
                if (DEBUG_RESOLUTION) {
                    System.out.println("[DEBUG-RESOLVE]   Struct-field: found base column '" +
                        baseColumnName + "' in " + namespace.getDisplayName() +
                        ", fieldPath=" + fieldPathSegments);
                }

                // Create a new ColumnSource with:
                // 1. struct_field_access evidence marker (for backward compatibility)
                // 2. fieldPath preserved (new in Improvement B)
                FieldPath fieldPath = FieldPath.of(fieldPathSegments);
                ColumnSource structFieldSource = baseColumnSource.withFieldPath(fieldPath, "struct_field_access");

                return ResolutionResult.exactMatch(structFieldSource);
            }
        }

        return null;
    }

    /**
     * Extract name parts from TObjectName.
     * Examples:
     * - "col" -> ["col"]
     * - "t.col" -> ["t", "col"]
     * - "schema.table.col" -> ["schema", "table", "col"]
     *
     * For BigQuery/Snowflake struct field access like "customer.customer_id":
     * - If schema/table tokens are not set, but toString() contains dots,
     *   extract all parts from toString() to capture field paths.
     */
    private List<String> extractNameParts(TObjectName objName) {
        List<String> parts = new ArrayList<>();

        // Add schema if present, but NOT when databaseToken is also set.
        // When databaseToken is present, the name is fully qualified (db.schema.table.column)
        // and the schema position IS the schema, not a table alias. Including it would cause
        // the resolver to incorrectly match the schema against table aliases. (Mantis #4268)
        if (objName.getSchemaToken() != null && objName.getDatabaseToken() == null) {
            parts.add(objName.getSchemaString());
        }

        // Add table/qualifier if present
        if (objName.getTableToken() != null) {
            parts.add(objName.getTableString());
        }

        // Add column name
        String columnName = objName.getColumnNameOnly();
        if (columnName != null) {
            parts.add(columnName);
        }

        // Improvement B: Handle BigQuery/Snowflake struct field access
        // If we only got a single part from standard extraction,
        // but toString() contains more segments (dots), extract them
        // This handles cases like "customer.customer_id" where:
        // - getColumnNameOnly() returns "customer"
        // - toString() returns "customer.customer_id"
        if (isStructFieldVendor()) {
            String fullName = objName.toString();
            if (fullName != null && fullName.contains(".")) {
                // Check if fullName has more segments than what we extracted
                String[] fullParts = splitNameParts(fullName);
                if (fullParts.length > parts.size() || hasConsecutiveDuplicates(parts)) {
                    // Replace parts with full extracted segments
                    parts.clear();
                    Collections.addAll(parts, fullParts);
                }
            }
        }

        return parts;
    }

    /**
     * Check if a list has consecutive duplicate elements.
     * This detects parser bugs where segments are duplicated.
     * Example: ["customer", "customer", "address"] returns true.
     */
    private boolean hasConsecutiveDuplicates(List<String> list) {
        if (list == null || list.size() < 2) {
            return false;
        }
        for (int i = 1; i < list.size(); i++) {
            if (list.get(i) != null && list.get(i).equals(list.get(i - 1))) {
                return true;
            }
        }
        return false;
    }


    /**
     * Split a dotted name into parts, handling quoted identifiers.
     * Examples:
     * - "a.b.c" -> ["a", "b", "c"]
     * - "`a.b`.c" -> ["`a.b`", "c"]
     * - "a.`b.c`" -> ["a", "`b.c`"]
     *
     * @param name The dotted name string
     * @return Array of name parts
     */
    private String[] splitNameParts(String name) {
        if (name == null || name.isEmpty()) {
            return new String[0];
        }

        // Simple case: no quotes, just split on dots
        if (!name.contains("`") && !name.contains("\"") && !name.contains("[")) {
            return name.split("\\.");
        }

        // Complex case: handle quoted identifiers
        List<String> parts = new ArrayList<>();
        StringBuilder current = new StringBuilder();
        char quoteChar = 0;

        for (int i = 0; i < name.length(); i++) {
            char c = name.charAt(i);

            if (quoteChar != 0) {
                // Inside a quoted identifier
                current.append(c);
                if (c == quoteChar) {
                    // Check for escaped quote (doubled)
                    if (i + 1 < name.length() && name.charAt(i + 1) == quoteChar) {
                        current.append(name.charAt(++i));
                    } else {
                        // End of quoted identifier
                        quoteChar = 0;
                    }
                }
            } else if (c == '`' || c == '"' || c == '[') {
                // Start of quoted identifier
                quoteChar = (c == '[') ? ']' : c;
                current.append(c);
            } else if (c == '.') {
                // Separator
                if (current.length() > 0) {
                    parts.add(current.toString());
                    current.setLength(0);
                }
            } else {
                current.append(c);
            }
        }

        // Add last part
        if (current.length() > 0) {
            parts.add(current.toString());
        }

        return parts.toArray(new String[0]);
    }

    /**
     * Process resolution matches and determine final result.
     */
    private ResolutionResult processResolvedMatches(TObjectName objName,
                                                    List<String> nameParts,
                                                    ResolvedImpl resolved) {
        String columnName = nameParts.get(nameParts.size() - 1);

        if (resolved.isEmpty()) {
            // No matches found
            return ResolutionResult.notFound(columnName);
        }

        // Deduplicate matches by namespace identity
        // The same namespace can be found through different scope paths (e.g., CTE scope and SELECT scope)
        // but it's still the same source - not truly ambiguous
        List<ResolvedImpl.Match> uniqueMatches = deduplicateMatchesByNamespace(resolved.getMatches());

        if (uniqueMatches.size() == 1) {
            // Exactly one unique namespace - success!
            ResolvedImpl.Match match = uniqueMatches.get(0);
            INamespace namespace = match.namespace;

            // For qualified names like "t.col", we need to resolve the column
            // For unqualified names like "col", we need to find it in the namespace
            ColumnSource columnSource;

            if (!match.remainingNames.isEmpty()) {
                // Still have parts to resolve (e.g., found table, need to find column)
                // Phase B3: Support multi-segment paths (table.column.field...)
                if (match.remainingNames.size() > 1 && isStructFieldVendor()) {
                    // Multiple segments: use resolveColumnPath for deep field access
                    // e.g., remainingNames = ["customer", "address", "city"]
                    // -> base column "customer", fieldPath ["address", "city"]
                    columnSource = namespace.resolveColumnPath(match.remainingNames);

                    if (columnSource == null) {
                        // Base column not found
                        String baseColName = match.remainingNames.get(0);
                        return ResolutionResult.notFound(baseColName,
                            "Column '" + baseColName + "' not found in " + namespace.getDisplayName());
                    }

                    if (DEBUG_RESOLUTION) {
                        System.out.println("[DEBUG-RESOLVE]   Deep path resolved: " +
                            match.remainingNames + " -> base=" + columnSource.getExposedName() +
                            ", fieldPath=" + (columnSource.hasFieldPath() ? columnSource.getFieldPath() : "none"));
                    }
                } else {
                    // Single segment: regular column resolution
                    String remainingColName = match.remainingNames.get(match.remainingNames.size() - 1);
                    columnSource = namespace.resolveColumn(remainingColName);

                    if (columnSource == null) {
                        // Column not found in the namespace
                        return ResolutionResult.notFound(remainingColName,
                            "Column '" + remainingColName + "' not found in " + namespace.getDisplayName());
                    }
                }
            } else {
                // Name resolved to a table/namespace directly with no remaining parts
                // This happens when the column name matches the table alias exactly.
                //
                // For UNNEST tables (e.g., "UNNEST(arr) AS x"), the alias "x" is ALSO
                // the implicit column name. When user writes "x" as a column reference,
                // the scope resolution matches the table alias "x", leaving remainingNames empty.
                // We should still try to resolve "x" as a column in the namespace.
                //
                // Example: SELECT x FROM UNNEST([1,2,3]) AS x
                // Here "x" in SELECT refers to the implicit column, not the table.
                if (namespace instanceof UnnestNamespace) {
                    // For UNNEST, try to resolve the matched name as a column
                    columnSource = namespace.resolveColumn(columnName);
                    if (columnSource != null) {
                        if (DEBUG_RESOLUTION) {
                            System.out.println("[DEBUG-RESOLVE]   UNNEST implicit column resolved: " +
                                columnName + " in " + namespace.getDisplayName());
                        }
                        return ResolutionResult.exactMatch(columnSource);
                    }
                }

                // For other namespaces or if column not found, this is an error
                return ResolutionResult.notFound(columnName,
                    "Name '" + columnName + "' resolves to a table, not a column");
            }

            return ResolutionResult.exactMatch(columnSource);
        }

        // Multiple unique namespaces - truly ambiguous
        List<ColumnSource> candidates = new ArrayList<>();
        // Mantis 4550: set when we synthesize a placeholder candidate for an
        // ambiguous-star subquery (see below). Such a candidate means the column
        // is genuinely ambiguous across derived tables and must NOT be collapsed
        // by GUESS_COLUMN_STRATEGY into a single guessed match.
        boolean synthesizedAmbStar = false;

        for (ResolvedImpl.Match match : uniqueMatches) {
            INamespace namespace = match.namespace;

            // Resolve column in this namespace
            // Phase B3: Support multi-segment paths
            ColumnSource columnSource;
            if (!match.remainingNames.isEmpty()) {
                if (match.remainingNames.size() > 1 && isStructFieldVendor()) {
                    // Multi-segment: use resolveColumnPath
                    columnSource = namespace.resolveColumnPath(match.remainingNames);
                } else {
                    // Single segment: use resolveColumn
                    String remainingColName = match.remainingNames.get(match.remainingNames.size() - 1);
                    columnSource = namespace.resolveColumn(remainingColName);
                }
            } else {
                columnSource = namespace.resolveColumn(columnName);
            }

            // Mantis 4550: an ambiguous-star subquery (SELECT * over a multi-table
            // FROM) cannot pin an unqualified column to a single base table and,
            // under the default NOT_PICKUP strategy, resolveColumn() returns null.
            // SelectScope now still reports it as an outer-scope candidate. Anchor a
            // placeholder ColumnSource to the subquery so it counts as a distinct
            // candidate (getSourceNamespace().getSourceTable() -> the derived table).
            // finalTable is left null; candidateTables reconciliation expands the
            // derived table into its base tables. The placeholder is NOT cached as an
            // inferred column on the namespace, so resolveColumn() itself is unchanged.
            // Only single-segment column references qualify (no deep struct path);
            // qualified single-table refs never reach this multi-namespace branch
            // (they resolve via the uniqueMatches.size()==1 path above), so this
            // cannot make "T0_2.col" falsely resolve.
            boolean singleSegment = match.remainingNames.isEmpty()
                    || match.remainingNames.size() == 1;
            if (columnSource == null
                    && singleSegment
                    && namespace instanceof gudusoft.gsqlparser.resolver2.namespace.SubqueryNamespace
                    && ((gudusoft.gsqlparser.resolver2.namespace.SubqueryNamespace) namespace).hasAmbiguousStar()) {
                String placeholderName = match.remainingNames.isEmpty()
                        ? columnName
                        : match.remainingNames.get(match.remainingNames.size() - 1);
                columnSource = new ColumnSource(namespace, placeholderName, null, 0.3,
                        "ambiguous_star_outer_candidate");
                synthesizedAmbStar = true;
            }

            if (columnSource != null) {
                candidates.add(columnSource);
            }
        }

        if (candidates.isEmpty()) {
            // Resolved to tables, but none have the column
            return ResolutionResult.notFound(columnName,
                "Column '" + columnName + "' not found in any of " + resolved.getCount() + " tables");
        }

        if (candidates.size() == 1) {
            // Only one table actually has the column
            return ResolutionResult.exactMatch(candidates.get(0));
        }

        // Mantis 4550: when a placeholder ambiguous-star candidate was synthesized,
        // the column is genuinely ambiguous across the derived tables. Force an
        // AMBIGUOUS result regardless of GUESS_COLUMN_STRATEGY (nearest/farthest),
        // which would otherwise collapse a definite sibling + maybe-star sibling
        // back into a single EXACT match.
        if (synthesizedAmbStar) {
            sortCandidatesByTablePosition(candidates);
            AmbiguousColumnSource ambiguousResult = new AmbiguousColumnSource(columnName, candidates);
            return ResolutionResult.ambiguous(ambiguousResult);
        }

        // Sort candidates by their table's position in the SQL text (FROM clause order)
        // This ensures consistent ordering for both ambiguous results and GUESS_COLUMN_STRATEGY
        sortCandidatesByTablePosition(candidates);

        // Check if all candidates are "inferred" (from tables without DDL metadata)
        // If so, return AMBIGUOUS regardless of GUESS_COLUMN_STRATEGY
        // Only apply GUESS_COLUMN_STRATEGY when we have definite knowledge about the columns
        //
        // The determination uses configurable thresholds:
        // - minDefiniteConfidence: minimum confidence to be considered "definite" (default 0.9)
        // - allowGuessWhenAllInferred: if true, allow guessing even when all candidates are inferred
        boolean hasDefiniteCandidate = false;
        double highestConfidence = 0.0;
        double minDefiniteConf = config != null ? config.getMinDefiniteConfidence() : 0.9;

        for (ColumnSource candidate : candidates) {
            // A candidate is "definite" if it has high confidence and is not just inferred from usage
            String evidence = candidate.getEvidence();
            double confidence = candidate.getConfidence();

            // Track highest confidence among candidates
            if (confidence > highestConfidence) {
                highestConfidence = confidence;
            }

            // Use structured evidence if available, otherwise check legacy evidence
            boolean isInferred;
            if (candidate.getEvidenceDetail() != null) {
                // Use the structured evidence's own determination
                isInferred = !candidate.getEvidenceDetail().isHighConfidence() ||
                             candidate.getEvidenceDetail().isInferred();
            } else {
                // Fallback to legacy logic
                // "inferred_from_usage" means table without DDL metadata - not definite
                // Low confidence (< minDefiniteConf) means we're guessing - not definite
                // Null evidence with high confidence is also considered inferred
                isInferred = (evidence == null || evidence.equals("inferred_from_usage")) ||
                             confidence < minDefiniteConf;
            }

            if (!isInferred) {
                hasDefiniteCandidate = true;
                break;
            }
        }

        // Check if we should allow guessing when all candidates are inferred
        boolean allowGuessInferred = config != null && config.isAllowGuessWhenAllInferred();

        if (!hasDefiniteCandidate && !allowGuessInferred) {
            // All candidates are from tables without DDL metadata (all inferred)
            // Don't guess - return as ambiguous so formatter can handle appropriately
            if (DEBUG_RESOLUTION) {
                System.out.println("[DEBUG-RESOLVE]   AMBIGUOUS (all inferred): " + columnName + " with " + candidates.size() + " candidates");
                for (ColumnSource c : candidates) {
                    System.out.println("[DEBUG-RESOLVE]     - " + (c.getSourceNamespace() != null ? c.getSourceNamespace().getDisplayName() : "null") +
                        " evidence=" + c.getEvidence() + " confidence=" + c.getConfidence());
                }
            }
            AmbiguousColumnSource ambiguous = new AmbiguousColumnSource(columnName, candidates);
            return ResolutionResult.ambiguous(ambiguous);
        }

        // Additional check: even if we have definite candidates or allow inferred guessing,
        // require at least one candidate to meet the minConfidenceToGuess threshold
        double minConfToGuess = config != null ? config.getMinConfidenceToGuess() : 0.95;
        if (highestConfidence < minConfToGuess && !allowGuessInferred) {
            // No candidate has sufficient confidence for guessing
            if (DEBUG_RESOLUTION) {
                System.out.println("[DEBUG-RESOLVE]   AMBIGUOUS (confidence too low): " + columnName +
                    " highest=" + highestConfidence + " required=" + minConfToGuess);
            }
            AmbiguousColumnSource ambiguous = new AmbiguousColumnSource(columnName, candidates);
            return ResolutionResult.ambiguous(ambiguous);
        }

        // Multiple tables have the column - apply GUESS_COLUMN_STRATEGY
        // Candidates are already sorted by table position (done earlier)
        int strategy = getGuessColumnStrategy();

        if (strategy == TSQLResolverConfig.GUESS_COLUMN_STRATEGY_NEAREST) {
            // Pick the first candidate (nearest table in FROM clause order)
            return ResolutionResult.exactMatch(candidates.get(0));
        } else if (strategy == TSQLResolverConfig.GUESS_COLUMN_STRATEGY_FARTHEST) {
            // Pick the last candidate (farthest table in FROM clause order)
            return ResolutionResult.exactMatch(candidates.get(candidates.size() - 1));
        }

        // GUESS_COLUMN_STRATEGY_NOT_PICKUP: leave as ambiguous
        if (DEBUG_RESOLUTION) {
            System.out.println("[DEBUG-RESOLVE]   AMBIGUOUS (NOT_PICKUP strategy): " + columnName + " with " + candidates.size() + " candidates");
        }
        AmbiguousColumnSource ambiguous = new AmbiguousColumnSource(columnName, candidates);
        return ResolutionResult.ambiguous(ambiguous);
    }

    /**
     * Deduplicate matches by namespace identity.
     * The same namespace can be found through different scope paths (e.g., CTE scope and SELECT scope)
     * but it's still the same source - not truly ambiguous.
     *
     * @param matches All matches from scope resolution
     * @return List of unique matches by namespace identity
     */
    private List<ResolvedImpl.Match> deduplicateMatchesByNamespace(List<ResolvedImpl.Match> matches) {
        if (matches == null || matches.size() <= 1) {
            return matches;
        }

        // Use identity-based set to track unique namespaces
        java.util.IdentityHashMap<INamespace, ResolvedImpl.Match> uniqueByNamespace =
            new java.util.IdentityHashMap<>();

        for (ResolvedImpl.Match match : matches) {
            if (match.namespace != null && !uniqueByNamespace.containsKey(match.namespace)) {
                uniqueByNamespace.put(match.namespace, match);
            }
        }

        return new ArrayList<>(uniqueByNamespace.values());
    }

    /**
     * Sort candidates by their source table's position in the SQL text.
     * This ensures that when GUESS_COLUMN_STRATEGY_NEAREST or FARTHEST is applied,
     * the candidates are ordered according to their actual position in the FROM clause.
     *
     * Uses the table's start token (lineNo, columnNo) to determine position.
     *
     * @param candidates List of ColumnSource candidates to sort in place
     */
    private void sortCandidatesByTablePosition(List<ColumnSource> candidates) {
        if (candidates == null || candidates.size() <= 1) {
            return;
        }

        candidates.sort(new Comparator<ColumnSource>() {
            @Override
            public int compare(ColumnSource c1, ColumnSource c2) {
                TTable t1 = c1.getFinalTable();
                TTable t2 = c2.getFinalTable();

                // If either table is null, maintain relative order
                if (t1 == null && t2 == null) return 0;
                if (t1 == null) return 1;  // null tables go to the end
                if (t2 == null) return -1;

                TSourceToken token1 = t1.getStartToken();
                TSourceToken token2 = t2.getStartToken();

                // If either token is null, maintain relative order
                if (token1 == null && token2 == null) return 0;
                if (token1 == null) return 1;
                if (token2 == null) return -1;

                // Compare by line number first
                int lineCmp = Long.compare(token1.lineNo, token2.lineNo);
                if (lineCmp != 0) {
                    return lineCmp;
                }

                // If same line, compare by column number
                return Long.compare(token1.columnNo, token2.columnNo);
            }
        });
    }

    /**
     * Update TObjectName with the resolution result.
     * Also registers with ResolutionContext.
     *
     * For ambiguous columns (multiple candidate tables), this also populates
     * TObjectName.candidateTables with all possible source tables.
     *
     * IMPORTANT: When resolution fails (notFound) and the column's sourceTable
     * was set during Phase 1 to an UNNEST table, we clear the sourceTable.
     * This is because UNNEST tables have a fixed set of columns (implicit column,
     * offset, struct fields) and should NOT have arbitrary columns inferred.
     * Clearing sourceTable allows the formatter to treat these as "missed" columns.
     */
    private void updateObjectNameWithResult(TObjectName objName, ResolutionResult result) {
        // Update TObjectName with resolution result
        objName.setResolution(result);

        // For notFound results, check if Phase 1 incorrectly linked to UNNEST table
        // UNNEST tables have a fixed column set - don't allow inferred columns
        if (!result.isExactMatch() && !result.isAmbiguous()) {
            TTable currentSourceTable = objName.getSourceTable();
            if (currentSourceTable != null &&
                currentSourceTable.getTableType() == ETableSource.unnest) {
                // Clear the incorrectly set sourceTable from Phase 1
                // This column wasn't found in the UNNEST namespace, so it shouldn't
                // be attributed to the UNNEST table
                objName.setSourceTable(null);
                if (DEBUG_RESOLUTION) {
                    System.out.println("[DEBUG-RESOLVE]   Cleared incorrect UNNEST sourceTable for: " +
                        objName + " (not found in UNNEST namespace)");
                }
            }
        }

        // For ambiguous results, populate candidateTables with all candidate tables
        // IMPORTANT: Clear existing candidateTables first, as Phase 1 (linkColumnToTable)
        // may have added candidates from incorrect scopes (e.g., MERGE target table for
        // columns inside USING subquery). Phase 2 (NameResolver) has proper scope awareness
        // and produces the authoritative candidate list.
        if (result.isAmbiguous() && result.getAmbiguousSource() != null) {
            AmbiguousColumnSource ambiguous = result.getAmbiguousSource();
            // Clear Phase 1 candidates before adding Phase 2's scope-aware candidates
            objName.getCandidateTables().clear();
            for (ColumnSource candidate : ambiguous.getCandidates()) {
                gudusoft.gsqlparser.nodes.TTable candidateTable = candidate.getFinalTable();
                if (candidateTable != null) {
                    objName.getCandidateTables().addTable(candidateTable);
                    continue;
                }
                // Mantis 4550: an ambiguous-star subquery placeholder has no single
                // finalTable. Add its immediate derived table (the subquery) so the
                // reconcile pass below expands it into the physical base tables that
                // can actually supply the column (e.g. T0_2 -> A.A2, A.A3).
                gudusoft.gsqlparser.nodes.TTable derivedTable =
                    candidate.getSourceNamespace() != null
                        ? candidate.getSourceNamespace().getSourceTable() : null;
                if (derivedTable != null) {
                    objName.getCandidateTables().addTable(derivedTable);
                    continue;
                }
                // A candidate that cannot collapse to a single physical table
                // (e.g. a UNION-backed derived table) would otherwise be dropped
                // here. Trace its set-operation branches projection-aware, by the
                // requested column, so only branches that actually supply the
                // column contribute base tables (a constant branch like
                // "SELECT 1 AS a" adds nothing).
                addBranchBaseTablesForColumn(objName, candidate);
            }
        }

        // Mantis 4550: reconcile candidateTables so they point to the physical
        // base tables that could actually supply the column, rather than the
        // intermediate derived tables. Phase 1 (linkColumnToTable) may leave
        // stale candidates that include derived tables which cannot supply the
        // column (e.g. a subquery with an explicit column list that omits it),
        // and the EXACT/guessed-match path above does not clear them. See
        // reconcileCandidateTablesToBaseTables() for the full rationale.
        reconcileCandidateTablesToBaseTables(objName);

        // Register with context for global querying
        context.registerResolution(objName, result);
    }

    /**
     * Mantis 4550: rewrite {@code candidateTables} so each entry is a physical
     * base table that could supply the column, expanding any intermediate
     * derived table (subquery / CTE) into the underlying tables it exposes.
     *
     * <p>Background: for an unqualified column whose source is ambiguous across
     * several derived tables in the FROM clause, Phase 1 ({@code linkColumnToTable})
     * may populate {@code candidateTables} with the derived tables themselves.
     * When Phase 2 resolves the column to a guessed EXACT match (rather than a
     * truly AMBIGUOUS result) the scope-aware ambiguous branch that normally
     * clears and rebuilds {@code candidateTables} is skipped, leaving those
     * stale derived-table candidates in place. Some of them cannot even supply
     * the column (e.g. a {@code SELECT x, ds FROM ...} subquery for column
     * {@code S311}).</p>
     *
     * <p>This pass walks each candidate: physical tables are kept as-is, while
     * a derived table is replaced by the set of physical base tables reachable
     * through its projection that could supply the column. A derived table whose
     * closed projection does not expose the column contributes nothing. Results
     * are de-duplicated by qualified table name, preserving first-seen order, so
     * the same physical table reached via two derived paths (or two aliases of a
     * self-join) collapses to a single candidate.</p>
     */
    private void reconcileCandidateTablesToBaseTables(TObjectName objName) {
        TTableList candidates = objName.getCandidateTables();
        if (candidates == null || candidates.size() == 0) {
            return;
        }

        String columnName = objName.getColumnNameOnly();
        if (columnName == null || columnName.isEmpty()) {
            return;
        }

        List<TTable> expanded = new ArrayList<TTable>();
        Set<String> seenKeys = new LinkedHashSet<String>();
        IdentityHashMap<TSelectSqlStatement, Boolean> visiting =
            new IdentityHashMap<TSelectSqlStatement, Boolean>();
        for (int i = 0; i < candidates.size(); i++) {
            collectBaseTablesSupplyingColumn(candidates.getTable(i), columnName,
                expanded, seenKeys, visiting);
        }

        // Only replace when expansion produced concrete base tables; otherwise
        // keep the existing list rather than wiping resolution information.
        if (expanded.isEmpty()) {
            return;
        }

        candidates.clear();
        for (TTable t : expanded) {
            candidates.addTable(t);
        }
    }

    /**
     * Add, to {@code objName}'s candidate tables, the base tables that supply the
     * column for an ambiguous candidate whose {@link ColumnSource#getFinalTable()}
     * is {@code null} (typically a set-operation/UNION-backed namespace). The
     * namespace's SELECT statement is traced projection-aware by the requested
     * column, so a branch whose projection at that position is a constant or
     * expression contributes no base table. Falls back to every reachable final
     * table only when projection-aware tracing yields nothing.
     */
    private void addBranchBaseTablesForColumn(TObjectName objName, ColumnSource candidate) {
        INamespace ns = candidate.getSourceNamespace();
        if (ns == null) {
            return;
        }

        List<TTable> base = new ArrayList<TTable>();
        Set<String> seenKeys = new LinkedHashSet<String>();
        IdentityHashMap<TSelectSqlStatement, Boolean> visiting =
            new IdentityHashMap<TSelectSqlStatement, Boolean>();

        String columnName = objName.getColumnNameOnly();
        TSelectSqlStatement nsSelect = ns.getSelectStatement();
        boolean traced = false;
        if (nsSelect != null && columnName != null && !columnName.isEmpty()) {
            collectFromSelectBody(nsSelect, columnName, -1, base, seenKeys, visiting);
            traced = true;
        }

        if (!traced) {
            // We could not trace the namespace projection-aware (no SELECT body to
            // inspect): report every reachable final table rather than dropping
            // the candidate entirely. When tracing DID run, an empty result means
            // "no physical table supplies this column" (e.g. all branches project
            // a constant), so we must not fall back and re-add those tables.
            List<TTable> all = ns.getAllFinalTables();
            if (all != null) {
                for (TTable t : all) {
                    if (t != null) {
                        base.add(t);
                    }
                }
            }
        }

        for (TTable t : base) {
            objName.getCandidateTables().addTable(t);
        }
    }

    /**
     * Collect, into {@code out}, the physical base tables reachable from
     * {@code table} that could supply {@code columnName}. Derived tables
     * (subqueries / CTEs) are expanded through their projection; physical
     * tables are added directly. De-duplicated by qualified table name via
     * {@code seenKeys}. {@code visiting} guards against cyclic CTE/subquery
     * references.
     */
    private void collectBaseTablesSupplyingColumn(TTable table, String columnName,
            List<TTable> out, Set<String> seenKeys,
            IdentityHashMap<TSelectSqlStatement, Boolean> visiting) {
        if (table == null) {
            return;
        }

        TSelectSqlStatement body = getDerivedBody(table);
        if (body == null) {
            // Physical base table (or a reference we cannot trace further).
            addBaseTable(table, out, seenKeys);
            return;
        }

        if (visiting.containsKey(body)) {
            return;
        }
        visiting.put(body, Boolean.TRUE);
        try {
            // A CTE with an explicit column list (e.g. WITH cte(a) AS (SELECT x
            // FROM t)) renames its outputs, so the requested name must be mapped
            // positionally onto the body's projection rather than matched by the
            // body's own display names.
            int forcedIndex = -1;
            gudusoft.gsqlparser.nodes.TObjectNameList cteColumns =
                (table.isCTEName() && table.getCTE() != null) ? table.getCTE().getColumnList() : null;
            if (cteColumns != null && cteColumns.size() > 0) {
                forcedIndex = indexOfMatchingName(cteColumns, columnName);
                if (forcedIndex < 0) {
                    // The CTE's column list does not expose this name.
                    return;
                }
            }

            collectFromSelectBody(body, columnName, forcedIndex, out, seenKeys, visiting);
        } finally {
            visiting.remove(body);
        }
    }

    /**
     * Collect base tables from a derived table's SELECT body. Handles set
     * operations (UNION / EXCEPT / INTERSECT) by flattening to leaf SELECTs and
     * mapping the column onto each leaf positionally, and plain SELECTs via
     * {@link #collectFromSimpleSelect}.
     *
     * @param forcedIndex when {@code >= 0}, the column is identified by position
     *     (e.g. a CTE explicit column list); when {@code -1}, by output name.
     */
    private void collectFromSelectBody(TSelectSqlStatement body, String columnName,
            int forcedIndex, List<TTable> out, Set<String> seenKeys,
            IdentityHashMap<TSelectSqlStatement, Boolean> visiting) {
        if (body == null) {
            return;
        }

        if (body.isCombinedQuery()) {
            List<TSelectSqlStatement> leaves = new ArrayList<TSelectSqlStatement>();
            flattenSetOperationLeaves(body, leaves);
            if (leaves.isEmpty()) {
                return;
            }
            // All set operands share one output shape by position. Determine the
            // position from the first leaf (or the forced/CTE index) and trace
            // that ordinal in every operand.
            int idx = forcedIndex;
            if (idx < 0) {
                idx = indexOfNamedProjection(leaves.get(0).getResultColumnList(), columnName);
            }
            if (idx < 0) {
                // The output ordinal is unknown (e.g. the leftmost operand is
                // SELECT *), so the column cannot be mapped onto each operand by
                // position. Matching later operands by name would silently drop a
                // renamed branch, so over-report every FROM table of every operand
                // instead.
                for (TSelectSqlStatement leaf : leaves) {
                    expandAllFromTables(leaf, columnName, out, seenKeys, visiting);
                }
                return;
            }
            for (TSelectSqlStatement leaf : leaves) {
                collectFromSimpleSelect(leaf, columnName, idx, out, seenKeys, visiting);
            }
            return;
        }

        collectFromSimpleSelect(body, columnName, forcedIndex, out, seenKeys, visiting);
    }

    /**
     * Collect base tables from a single (non-combined) SELECT.
     *
     * <p>When {@code forcedIndex >= 0} the column is resolved by position onto
     * the projection (used for CTE column lists and set-operation operands);
     * otherwise it is matched by output name. A {@code SELECT *} or an
     * out-of-range/star position falls back to expanding every FROM table, since
     * any of them may supply the column.</p>
     */
    private void collectFromSimpleSelect(TSelectSqlStatement select, String columnName,
            int forcedIndex, List<TTable> out, Set<String> seenKeys,
            IdentityHashMap<TSelectSqlStatement, Boolean> visiting) {
        if (select == null) {
            return;
        }
        TResultColumnList rcl = select.getResultColumnList();
        if (rcl == null) {
            return;
        }

        if (forcedIndex >= 0) {
            boolean hasStar = false;
            for (int i = 0; i < rcl.size(); i++) {
                if (starQualifier(rcl.getResultColumn(i)) != null) {
                    hasStar = true;
                    break;
                }
            }
            if (hasStar || forcedIndex >= rcl.size()) {
                // Positional mapping is unreliable: any FROM table may supply it.
                expandAllFromTables(select, columnName, out, seenKeys, visiting);
                return;
            }
            TResultColumn rc = rcl.getResultColumn(forcedIndex);
            TObjectName underlying = rc != null ? rc.getColumnFullname() : null;
            if (underlying != null) {
                traceNamedColumnIntoFrom(select, underlying, out, seenKeys, visiting);
            }
            // A computed projection at this position has no physical base table.
            return;
        }

        // Match by output name. Collect base tables from EVERY projection that
        // could expose the column: each matching named projection (there may be
        // several with the same output name, e.g. "SELECT t1.c, t2.c") AND every
        // star projection. We do not early-return on the first match, so
        // genuinely ambiguous derived tables surface all of their suppliers.
        boolean unqualifiedStar = false;
        List<String> qualifiedStarPrefixes = new ArrayList<String>();
        for (int i = 0; i < rcl.size(); i++) {
            TResultColumn rc = rcl.getResultColumn(i);
            if (rc == null) {
                continue;
            }
            String starPrefix = starQualifier(rc);
            if (starPrefix != null) {
                if (starPrefix.isEmpty()) {
                    unqualifiedStar = true;
                } else {
                    qualifiedStarPrefixes.add(starPrefix);
                }
                continue;
            }

            String outName = rc.getDisplayName();
            if (outName != null && nameMatcher.matches(outName, columnName)) {
                TObjectName underlying = rc.getColumnFullname();
                if (underlying != null) {
                    traceNamedColumnIntoFrom(select, underlying, out, seenKeys, visiting);
                }
                // A computed/expression column has no physical base table.
            }
        }

        if (unqualifiedStar) {
            expandAllFromTables(select, columnName, out, seenKeys, visiting);
        }
        for (String prefix : qualifiedStarPrefixes) {
            TTable matched = findTableByAliasOrName(select, prefix);
            if (matched != null) {
                collectBaseTablesSupplyingColumn(matched, columnName, out, seenKeys, visiting);
            }
        }
        // No star and no named match: this SELECT cannot supply the column.
    }

    /**
     * Expand every table in {@code select}'s FROM clause as a possible supplier
     * of {@code columnName}.
     */
    private void expandAllFromTables(TSelectSqlStatement select, String columnName,
            List<TTable> out, Set<String> seenKeys,
            IdentityHashMap<TSelectSqlStatement, Boolean> visiting) {
        if (select == null || select.tables == null) {
            return;
        }
        for (int i = 0; i < select.tables.size(); i++) {
            collectBaseTablesSupplyingColumn(select.tables.getTable(i),
                columnName, out, seenKeys, visiting);
        }
    }

    /**
     * Iteratively flatten a set-operation tree into the leaf SELECTs that
     * actually supply output column values, left to right. Uses an explicit
     * stack to avoid StackOverflowError on deeply nested set-operation chains.
     *
     * <p>The operator matters: UNION and INTERSECT take values from both
     * operands, but EXCEPT / MINUS take output values from the LEFT operand only
     * (the right operand merely filters rows), so its right branch is not a
     * source of the selected column and is skipped.</p>
     */
    private void flattenSetOperationLeaves(TSelectSqlStatement stmt, List<TSelectSqlStatement> leaves) {
        java.util.Deque<TSelectSqlStatement> stack = new java.util.ArrayDeque<TSelectSqlStatement>();
        stack.push(stmt);
        while (!stack.isEmpty()) {
            TSelectSqlStatement cur = stack.pop();
            if (cur == null) {
                continue;
            }
            if (cur.isCombinedQuery()) {
                gudusoft.gsqlparser.ESetOperatorType op = cur.getSetOperatorType();
                boolean rightSuppliesValues =
                    op != gudusoft.gsqlparser.ESetOperatorType.except
                    && op != gudusoft.gsqlparser.ESetOperatorType.minus;
                // Push right first (when it supplies values) so the left operand
                // is processed first.
                if (rightSuppliesValues && cur.getRightStmt() != null) {
                    stack.push(cur.getRightStmt());
                }
                if (cur.getLeftStmt() != null) {
                    stack.push(cur.getLeftStmt());
                }
            } else {
                leaves.add(cur);
            }
        }
    }

    /**
     * Index of the first projection in {@code rcl} whose output name matches
     * {@code columnName}, or {@code -1} if none matches or a star projection
     * makes positional indexing unreliable.
     */
    private int indexOfNamedProjection(TResultColumnList rcl, String columnName) {
        if (rcl == null) {
            return -1;
        }
        for (int i = 0; i < rcl.size(); i++) {
            TResultColumn rc = rcl.getResultColumn(i);
            if (rc == null) {
                continue;
            }
            if (starQualifier(rc) != null) {
                // A star shifts subsequent positions unpredictably.
                return -1;
            }
            String name = rc.getDisplayName();
            if (name != null && nameMatcher.matches(name, columnName)) {
                return i;
            }
        }
        return -1;
    }

    /**
     * Index of the first name in {@code names} that matches {@code columnName}.
     */
    private int indexOfMatchingName(gudusoft.gsqlparser.nodes.TObjectNameList names, String columnName) {
        if (names == null) {
            return -1;
        }
        for (int i = 0; i < names.size(); i++) {
            TObjectName name = names.getObjectName(i);
            String n = name != null ? name.getColumnNameOnly() : null;
            if (n != null && nameMatcher.matches(n, columnName)) {
                return i;
            }
        }
        return -1;
    }

    /**
     * Trace a named projection column (e.g. {@code t.col} or {@code col}) into
     * the FROM clause of {@code body} and collect the base tables that supply
     * the referenced column.
     */
    private void traceNamedColumnIntoFrom(TSelectSqlStatement body, TObjectName underlying,
            List<TTable> out, Set<String> seenKeys,
            IdentityHashMap<TSelectSqlStatement, Boolean> visiting) {
        if (body == null || body.tables == null || underlying == null) {
            return;
        }
        String underlyingName = underlying.getColumnNameOnly();
        if (underlyingName == null || underlyingName.isEmpty()) {
            return;
        }
        String tableQualifier = columnTableQualifier(underlying);
        if (tableQualifier != null && !tableQualifier.isEmpty()) {
            TTable matched = findTableByAliasOrName(body, tableQualifier);
            if (matched != null) {
                collectBaseTablesSupplyingColumn(matched, underlyingName, out,
                    seenKeys, visiting);
                return;
            }
        }

        // If resolver2 already resolved this inner projection column to a single
        // source (e.g. via DDL/sqlenv metadata), trust that source rather than
        // broadening to every FROM table. This keeps candidateTables accurate
        // for "SELECT c FROM t1 JOIN t2" when only t1 actually exposes c.
        TTable resolvedSource = underlying.getSourceTable();
        if (resolvedSource != null && isInFromClause(body, resolvedSource)) {
            collectBaseTablesSupplyingColumn(resolvedSource, underlyingName, out,
                seenKeys, visiting);
            return;
        }

        // Unqualified and unresolved: the underlying column may originate from
        // any FROM table.
        for (int i = 0; i < body.tables.size(); i++) {
            collectBaseTablesSupplyingColumn(body.tables.getTable(i),
                underlyingName, out, seenKeys, visiting);
        }
    }

    /**
     * Whether {@code target} is one of the tables in {@code body}'s FROM clause
     * (by object identity). Used to ensure a resolved {@code sourceTable} really
     * belongs to this derived table's scope before trusting it.
     */
    private boolean isInFromClause(TSelectSqlStatement body, TTable target) {
        if (body == null || body.tables == null || target == null) {
            return false;
        }
        for (int i = 0; i < body.tables.size(); i++) {
            if (body.tables.getTable(i) == target) {
                return true;
            }
        }
        return false;
    }

    /**
     * Return the SELECT body of a derived table (inline subquery or CTE
     * reference), or {@code null} if {@code table} is a physical table.
     */
    private TSelectSqlStatement getDerivedBody(TTable table) {
        if (table == null) {
            return null;
        }
        if (table.getSubquery() != null) {
            return table.getSubquery();
        }
        if (table.isCTEName() && table.getCTE() != null) {
            return table.getCTE().getSubquery();
        }
        return null;
    }

    /**
     * Add a physical base table to {@code out}, de-duplicated by qualified name.
     */
    private void addBaseTable(TTable table, List<TTable> out, Set<String> seenKeys) {
        if (table == null) {
            return;
        }
        String key = baseTableKey(table);
        if (seenKeys.add(key)) {
            out.add(table);
        }
    }

    private String baseTableKey(TTable table) {
        String name = table.getFullName();
        if (name == null || name.isEmpty()) {
            name = table.getName();
        }
        if (name == null) {
            name = table.toString();
        }
        return name == null ? "" : name.toLowerCase();
    }

    /**
     * If {@code rc} is a star result column, return its qualifier: an empty
     * string for an unqualified {@code *}, or the table prefix for {@code q.*}.
     * Returns {@code null} when {@code rc} is not a star column.
     */
    private String starQualifier(TResultColumn rc) {
        if (rc == null || rc.getExpr() == null) {
            return null;
        }
        TObjectName operand = rc.getExpr().getObjectOperand();
        String text = operand != null ? operand.toString() : rc.toString();
        if (text == null) {
            return null;
        }
        text = text.trim();
        if ("*".equals(text)) {
            return "";
        }
        if (text.endsWith(".*")) {
            String prefix = text.substring(0, text.length() - 2).trim();
            // Keep only the last segment of a multi-part prefix (e.g. a.b.* -> b).
            int dot = prefix.lastIndexOf('.');
            if (dot >= 0) {
                prefix = prefix.substring(dot + 1).trim();
            }
            return prefix;
        }
        return null;
    }

    /**
     * Return the table qualifier of a (possibly) qualified column reference,
     * e.g. {@code "t"} for {@code t.col} or {@code "t"} for {@code s.t.col}.
     * Returns {@code null} for an unqualified column.
     */
    private String columnTableQualifier(TObjectName column) {
        if (column == null) {
            return null;
        }
        String full = column.toString();
        if (full == null) {
            return null;
        }
        int lastDot = full.lastIndexOf('.');
        if (lastDot < 0) {
            return null;
        }
        String prefix = full.substring(0, lastDot).trim();
        int prevDot = prefix.lastIndexOf('.');
        if (prevDot >= 0) {
            prefix = prefix.substring(prevDot + 1).trim();
        }
        return prefix.isEmpty() ? null : prefix;
    }

    /**
     * Find a table in {@code body}'s FROM clause by alias or (last segment of)
     * name, matching case-insensitively.
     */
    private TTable findTableByAliasOrName(TSelectSqlStatement body, String qualifier) {
        if (body == null || body.tables == null || qualifier == null) {
            return null;
        }
        for (int i = 0; i < body.tables.size(); i++) {
            TTable t = body.tables.getTable(i);
            if (t == null) {
                continue;
            }
            String alias = t.getAliasName();
            if (alias != null && alias.equalsIgnoreCase(qualifier)) {
                return t;
            }
            String name = t.getName();
            if (name != null) {
                int dot = name.lastIndexOf('.');
                String shortName = dot >= 0 ? name.substring(dot + 1) : name;
                if (shortName.equalsIgnoreCase(qualifier)) {
                    return t;
                }
            }
        }
        return null;
    }

    /**
     * Resolve a column within a specific namespace (for direct lookups).
     */
    public ResolutionResult resolveInNamespace(String columnName, INamespace namespace) {
        if (columnName == null || namespace == null) {
            return ResolutionResult.notFound("<null>");
        }

        ColumnSource source = namespace.resolveColumn(columnName);
        if (source != null) {
            return ResolutionResult.exactMatch(source);
        }

        return ResolutionResult.notFound(columnName);
    }

    /**
     * Find all namespaces that contain a given column.
     * Used for implementing full candidate collection in ambiguous scenarios.
     */
    public List<INamespace> findNamespacesWithColumn(String columnName, IScope scope) {
        List<INamespace> result = new ArrayList<>();

        for (INamespace ns : scope.getVisibleNamespaces()) {
            if (ns.hasColumn(columnName) == ColumnLevel.EXISTS) {
                result.add(ns);
            }
        }

        return result;
    }

    public INameMatcher getNameMatcher() {
        return nameMatcher;
    }

    public ResolutionContext getContext() {
        return context;
    }
}