package gudusoft.gsqlparser.resolver2;

import gudusoft.gsqlparser.TBaseType;
import gudusoft.gsqlparser.TCustomSqlStatement;
import gudusoft.gsqlparser.IRelation;
import gudusoft.gsqlparser.TLog;
import gudusoft.gsqlparser.TSourceToken;
import gudusoft.gsqlparser.TStatementList;
import gudusoft.gsqlparser.ETableSource;
import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.EDbObjectType;
import gudusoft.gsqlparser.EErrorType;
import gudusoft.gsqlparser.ESqlClause;
import gudusoft.gsqlparser.ESqlStatementType;
import gudusoft.gsqlparser.TSyntaxError;
import gudusoft.gsqlparser.stmt.dax.TDaxStmt;
import gudusoft.gsqlparser.stmt.TAlterTableStatement;
import gudusoft.gsqlparser.stmt.TCreateTableSqlStatement;
import gudusoft.gsqlparser.stmt.TInsertSqlStatement;
import gudusoft.gsqlparser.stmt.TUpdateSqlStatement;
import gudusoft.gsqlparser.stmt.TUseDatabase;
import gudusoft.gsqlparser.stmt.TDeleteSqlStatement;
import gudusoft.gsqlparser.stmt.TSelectSqlStatement;
import gudusoft.gsqlparser.compiler.TContext;
import gudusoft.gsqlparser.nodes.TObjectName;
import gudusoft.gsqlparser.nodes.TObjectNameList;
import gudusoft.gsqlparser.nodes.TTable;
import gudusoft.gsqlparser.nodes.TTableList;
import gudusoft.gsqlparser.nodes.TJoinExpr;
import gudusoft.gsqlparser.nodes.TParseTreeNode;
import gudusoft.gsqlparser.nodes.TParseTreeVisitor;
import gudusoft.gsqlparser.nodes.TResultColumn;
import gudusoft.gsqlparser.nodes.TResultColumnList;
import gudusoft.gsqlparser.nodes.TQualifyClause;
import gudusoft.gsqlparser.nodes.TExpression;
import gudusoft.gsqlparser.nodes.TGroupBy;
import gudusoft.gsqlparser.nodes.TGroupByItem;
import gudusoft.gsqlparser.EExpressionType;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.resolver2.model.AmbiguousColumnSource;
import gudusoft.gsqlparser.resolver2.model.ColumnSource;
import gudusoft.gsqlparser.resolver2.model.FromScopeIndex;
import gudusoft.gsqlparser.resolver2.model.ResolutionContext;
import gudusoft.gsqlparser.resolver2.model.ResolutionResult;
import gudusoft.gsqlparser.resolver2.model.ResolutionStatistics;
import gudusoft.gsqlparser.resolver2.ResolutionStatus;
import gudusoft.gsqlparser.resolver2.result.IResolutionResult;
import gudusoft.gsqlparser.resolver2.result.ResolutionResultImpl;
import gudusoft.gsqlparser.resolver2.scope.FromScope;
import gudusoft.gsqlparser.resolver2.scope.GlobalScope;
import gudusoft.gsqlparser.resolver2.scope.IScope;
import gudusoft.gsqlparser.resolver2.scope.SelectScope;
import gudusoft.gsqlparser.resolver2.scope.CTEScope;
import gudusoft.gsqlparser.resolver2.scope.GroupByScope;
import gudusoft.gsqlparser.resolver2.scope.HavingScope;
import gudusoft.gsqlparser.resolver2.scope.OrderByScope;
import gudusoft.gsqlparser.resolver2.scope.UpdateScope;
import gudusoft.gsqlparser.resolver2.scope.DeleteScope;
import gudusoft.gsqlparser.resolver2.namespace.INamespace;
import gudusoft.gsqlparser.resolver2.namespace.TableNamespace;
import gudusoft.gsqlparser.resolver2.namespace.SubqueryNamespace;
import gudusoft.gsqlparser.resolver2.namespace.CTENamespace;
import gudusoft.gsqlparser.nodes.TCTE;
import gudusoft.gsqlparser.nodes.TCTEList;
import gudusoft.gsqlparser.nodes.TUnnestClause;
import gudusoft.gsqlparser.stmt.TSelectSqlStatement;
import gudusoft.gsqlparser.resolver2.iterative.ConvergenceDetector;
import gudusoft.gsqlparser.resolver2.iterative.ResolutionPass;
import gudusoft.gsqlparser.resolver2.enhancement.NamespaceEnhancer;
import gudusoft.gsqlparser.resolver2.enhancement.EnhancementResult;
import gudusoft.gsqlparser.resolver2.enhancement.CollectedColumnRef;
import gudusoft.gsqlparser.resolver2.namespace.CTENamespace;
import gudusoft.gsqlparser.sqlenv.CanonKey;
import gudusoft.gsqlparser.sqlenv.IdentifierService;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.TSQLEnv;
import gudusoft.gsqlparser.util.SQLUtil;
import gudusoft.gsqlparser.TAttributeNode;
import gudusoft.gsqlparser.resolver2.binding.BindingDiagnostic;
import gudusoft.gsqlparser.resolver2.binding.BindingDiagnosticPostPass;
import gudusoft.gsqlparser.resolver2.binding.BindingMetadataAuthority;
import gudusoft.gsqlparser.resolver2.binding.BindingResult;
import gudusoft.gsqlparser.resolver2.binding.ColumnAuthority;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

// ScopeBuilder for visitor-based scope construction
import gudusoft.gsqlparser.resolver2.ScopeBuilder;
import gudusoft.gsqlparser.resolver2.ScopeBuildResult;

/**
 * New SQL Resolver - Phase 2 Enhanced Framework
 *
 * This is the main entry point for the new resolution architecture.
 * Provides improved column-to-table resolution with:
 * - Clear scope-based name resolution
 * - Full candidate collection for ambiguous cases
 * - Confidence-scored inference
 * - Better tracing and debugging
 *
 * Usage:
 * <pre>
 * TSQLResolver2 resolver = new TSQLResolver2(context, statements);
 * boolean success = resolver.resolve();
 * ResolutionStatistics stats = resolver.getStatistics();
 * </pre>
 *
 * Phase 1 capabilities:
 * - Basic SELECT statement resolution
 * - Table and subquery namespaces
 * - Qualified and unqualified column references
 * - FROM clause scope management
 *
 * Phase 2 capabilities:
 * - JOIN scope handling with nullable semantics
 * - CTE (WITH clause) resolution
 * - Iterative resolution framework (auto-converges after first pass if no iteration needed)
 *
 * Future phases will add:
 * - Evidence-based inference
 * - Star column expansion
 */
public class TSQLResolver2 {

    /**
     * Immutable observation of one {@link #resolve()} invocation.
     *
     * <p>This is deliberately diagnostic-only. The parser continues to ignore
     * the boolean returned by {@code resolve()} exactly as before; callers that
     * need to distinguish "resolver object exists" from "resolver completed"
     * can inspect this value after parsing.</p>
     */
    public static final class RunOutcome {

        private static final RunOutcome NOT_ATTEMPTED =
                new RunOutcome(false, false, null, null);
        private static final RunOutcome ATTEMPTED =
                new RunOutcome(true, false, null, null);

        private final boolean attempted;
        private final boolean successful;
        private final String failureClassName;
        private final String failureMessage;

        private RunOutcome(boolean attempted, boolean successful,
                String failureClassName, String failureMessage) {
            this.attempted = attempted;
            this.successful = successful;
            this.failureClassName = failureClassName;
            this.failureMessage = failureMessage;
        }

        private static RunOutcome notAttempted() {
            return NOT_ATTEMPTED;
        }

        private static RunOutcome attempted() {
            return ATTEMPTED;
        }

        private static RunOutcome completed(boolean successful) {
            return new RunOutcome(true, successful, null, null);
        }

        private static RunOutcome failed(Exception failure) {
            return new RunOutcome(true, false,
                    failure.getClass().getName(), failure.getMessage());
        }

        public boolean isAttempted() {
            return attempted;
        }

        public boolean isSuccessful() {
            return successful;
        }

        public String getFailureClassName() {
            return failureClassName;
        }

        public String getFailureMessage() {
            return failureMessage;
        }
    }

    private final TContext globalContext;
    private final TStatementList sqlStatements;
    private final TSQLResolverConfig config;
    private final ResolutionContext resolutionContext;
    private final NameResolver nameResolver;

    /**
     * Binding traces for the dynamic-SQL publication proof; {@code null}
     * unless {@link TSQLResolverConfig#isCaptureBindingTrace()}.
     */
    private final gudusoft.gsqlparser.resolver2.binding.BindingTraceRegistry bindingTraceRegistry;

    /** @see #bindingTraceRegistry */
    public gudusoft.gsqlparser.resolver2.binding.BindingTraceRegistry getBindingTraceRegistry() {
        return bindingTraceRegistry;
    }

    /** Global scope (root of scope tree) */
    private GlobalScope globalScope;

    /** Convergence detector for iterative resolution */
    private ConvergenceDetector convergenceDetector;

    /** History of all resolution passes */
    private final List<ResolutionPass> passHistory;

    /**
     * Scope cache for iterative resolution.
     * Maps statements to their scope trees to avoid rebuilding scopes on each pass.
     * Key: TCustomSqlStatement, Value: SelectScope (or other scope type)
     */
    private final java.util.Map<Object, IScope> statementScopeCache;

    /**
     * Column-to-Scope mapping for iterative resolution (Principle 1: Scope完全复用).
     * Built once in Pass 1, reused in Pass 2+ to avoid rebuilding scopes.
     * Maps each TObjectName (column reference) to the IScope where it should be resolved.
     */
    private final java.util.Map<TObjectName, IScope> columnToScopeMap;

    /**
     * FromScope index cache for O(1) table/namespace lookups (Performance Optimization B).
     * Maps FromScope instances to their pre-built indexes.
     * Built lazily on first access, cleared at the start of each resolve() call.
     * Uses IdentityHashMap because we need object identity, not equals().
     */
    private final Map<IScope, FromScopeIndex> fromScopeIndexCache;

    /**
     * Cache for Teradata NAMED alias lookup.
     * Maps SELECT statements to their alias index (alias name -> TResultColumn).
     * Uses IdentityHashMap because we need object identity, not equals().
     * Optimization C: Reduces O(cols * select_items) to O(cols) for Teradata.
     */
    private final Map<TSelectSqlStatement, Map<String, TResultColumn>> teradataNamedAliasCache;

    /** Resolver-invocation-local identity index for legacy linked-column lists. */
    private final Map<TObjectNameList, Set<TObjectName>> linkedColumnIdentityCache;

    /**
     * Mantis 4651 — the synthetic clones produced by {@link #createTracedColumnClones()},
     * held by identity.
     *
     * <p>A column referenced through a derived table that selects a star
     * (e.g. {@code SELECT c FROM (SELECT * FROM t)}) can reach the physical table's
     * {@code linkedColumns} through two independent paths in
     * {@link #syncColumnToLegacy(TObjectName)}: the traced clone (whose
     * {@code sourceTable} is the physical table) is appended by the generic add,
     * and the original reference (whose {@code sourceTable} is the derived table) is
     * appended by the subquery fallback. Both used to fire, so
     * {@code TTable.getLinkedColumns()} reported the same column twice. This set and
     * {@link #tracedStarCloneKeys} let each path see what the other contributed.
     */
    private final Set<TObjectName> tracedStarClones;

    /**
     * Mantis 4651 — the physical columns covered by {@link #tracedStarClones},
     * indexed by table identity, so the subquery fallback in
     * {@link #syncColumnToLegacy(TObjectName)} can tell that an equivalent clone
     * already carries the link.
     *
     * <p>Structural on both axes on purpose: the table is the map's identity key
     * rather than its {@code identityHashCode} (two tables can share a 32-bit hash,
     * and a collision here would suppress the second table's only physical link),
     * and the column is a {@link CanonKey} rather than a folded string, so equality
     * is exactly {@link SQLUtil#sameName} — including on the collation-based vendors
     * whose {@code CanonKey} text is diagnostic only.</p>
     */
    private final Map<TTable, Set<CanonKey>> tracedStarCloneKeys;

    /**
     * All column references collected during Pass 1 (Principle 1: Scope完全复用).
     * Used in Pass 2+ to re-resolve names without rebuilding the scope tree.
     */
    private final List<TObjectName> allColumnReferences;

    /**
     * ScopeBuilder for visitor-based scope construction.
     * Replaces manual scope building with proper nested scope handling.
     */
    private final ScopeBuilder scopeBuilder;
    private gudusoft.gsqlparser.resolver2.metadata.StatementMetadata statementMetadata;


    /**
     * Result from ScopeBuilder containing the complete scope tree.
     * This is populated in Pass 1 and reused in Pass 2+.
     */
    private ScopeBuildResult scopeBuildResult;

    /**
     * The namespace member each star-expanded attribute was produced from, by attribute
     * identity. {@code expandQualifiedStar} / {@code expandUnqualifiedStar} build a
     * {@link TAttributeNode} from a namespace entry and keep only its NAME; the
     * {@link ColumnSource} behind the entry would otherwise be gone, and a consumer that wants
     * to know which column an expanded attribute stands for would be left to reparse the name —
     * a stored spelling, so neither {@code SQL_LEXICAL} nor a rendered form matches it reliably.
     * Recorded here at the moment of expansion instead. See
     * {@link #getStarExpansionSource(TAttributeNode)}.
     */
    private final java.util.Map<TAttributeNode, ColumnSource> starExpansionSources =
            new java.util.IdentityHashMap<TAttributeNode, ColumnSource>();

    /**
     * Slice S5 — populated by {@link BindingDiagnosticPostPass} after the
     * iterative resolver loop converges, when at least one binding flag is
     * on. Stays at {@link BindingResult#empty()} when binding is disabled
     * or {@code resolve()} has not yet been called (plan §5.6, §12).
     */
    private BindingResult bindingResult = BindingResult.empty();

    /** Observation state for the most recent {@link #resolve()} call. */
    private RunOutcome runOutcome = RunOutcome.notAttempted();

    /**
     * NamespaceEnhancer for explicit column collection and enhancement.
     * Handles the explicit namespace enhancement phase between resolution passes.
     * Columns are collected during resolution and added to namespaces explicitly.
     */
    private NamespaceEnhancer namespaceEnhancer;

    /**
     * Create resolver with default configuration
     */
    public TSQLResolver2(TContext context, TStatementList statements) {
        this(context, statements, TSQLResolverConfig.createDefault());
    }

    /**
     * Create resolver with custom configuration
     */
    public TSQLResolver2(TContext context, TStatementList statements, TSQLResolverConfig config) {
        this.globalContext = context;
        this.sqlStatements = statements;
        this.config = config;
        this.resolutionContext = new ResolutionContext();
        this.nameResolver = new NameResolver(config, resolutionContext);
        this.passHistory = new ArrayList<>();
        this.statementScopeCache = new java.util.HashMap<>();
        this.columnToScopeMap = new java.util.HashMap<>();
        this.fromScopeIndexCache = new IdentityHashMap<>();
        this.teradataNamedAliasCache = new IdentityHashMap<>();
        this.linkedColumnIdentityCache = new IdentityHashMap<>();
        this.tracedStarClones = java.util.Collections.newSetFromMap(new IdentityHashMap<TObjectName, Boolean>());
        this.tracedStarCloneKeys = new IdentityHashMap<>();
        this.allColumnReferences = new ArrayList<>();

        // Initialize ScopeBuilder for visitor-based scope construction
        this.scopeBuilder = new ScopeBuilder(context, config.getNameMatcher());
        // Pass guessColumnStrategy from config for namespace isolation (prevents test side effects)
        if (config.hasCustomGuessColumnStrategy()) {
            this.scopeBuilder.setGuessColumnStrategy(config.getGuessColumnStrategy());
        }

        // Binding-trace capture (dynamic-SQL publication proof, R4 step 3).
        // Registry exists only when opted in; the capture hooks in
        // NameResolver/ScopeBuilder are a null check otherwise.
        if (config.isCaptureBindingTrace()) {
            this.bindingTraceRegistry =
                    new gudusoft.gsqlparser.resolver2.binding.BindingTraceRegistry();
            this.nameResolver.setBindingTraceRegistry(bindingTraceRegistry);
            this.scopeBuilder.setBindingTraceRegistry(bindingTraceRegistry);
        } else {
            this.bindingTraceRegistry = null;
        }

        // If context is null, try to get TSQLEnv from statements
        // This allows TSQLEnv to flow from parser.setSqlEnv() through statements
        if (statements != null && statements.size() > 0) {
            try {
                TCustomSqlStatement firstStmt = statements.get(0);
                if (firstStmt != null && firstStmt.getGlobalScope() != null &&
                    firstStmt.getGlobalScope().getSqlEnv() != null) {
                    this.scopeBuilder.setSqlEnv(firstStmt.getGlobalScope().getSqlEnv());
                }
            } catch (Exception e) {
                // Silently ignore - SQLEnv is optional enhancement
            }
        }

        // Initialize convergence detector for iterative resolution
        this.convergenceDetector = new ConvergenceDetector(
            config.getMaxIterations(),
            config.getStablePassesForConvergence(),
            config.getMinProgressRate()
        );

        // Initialize namespace enhancer for explicit column collection
        // Debug mode follows the global resolver log setting
        this.namespaceEnhancer = new NamespaceEnhancer(TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE);
    }

    /**
     * Set the TSQLEnv to use for table metadata lookup.
     * This allows external callers to provide TSQLEnv if automatic detection fails.
     *
     * @param sqlEnv the SQL environment containing table metadata
     */
    public void setSqlEnv(gudusoft.gsqlparser.sqlenv.TSQLEnv sqlEnv) {
        if (scopeBuilder != null) {
            scopeBuilder.setSqlEnv(sqlEnv);
        }
    }

    /**
     * Get the TSQLEnv used for table metadata lookup.
     *
     * @return the SQL environment, or null if not set
     */
    public gudusoft.gsqlparser.sqlenv.TSQLEnv getSqlEnv() {
        return scopeBuilder != null ? scopeBuilder.getSqlEnv() : null;
    }

    /**
     * Get the set of virtual trigger tables (deleted/inserted in SQL Server triggers).
     * These tables should be excluded from table output since their columns are
     * resolved to the trigger's target table.
     *
     * @return Set of TTable objects that are virtual trigger tables
     */
    public java.util.Set<gudusoft.gsqlparser.nodes.TTable> getVirtualTriggerTables() {
        return scopeBuilder != null ? scopeBuilder.getVirtualTriggerTables() : java.util.Collections.emptySet();
    }

    /**
     * Get the SQL statements being resolved.
     *
     * @return the list of SQL statements
     */
    public TStatementList getStatements() {
        return sqlStatements;
    }

    // Performance timing fields (instance-level for single resolve() call)
    private long timeScopeBuilder = 0;
    private long timeNameResolution = 0;
    private long timeEnhancement = 0;
    private long timeLegacySync = 0;
    private long timeOther = 0;

    // Global accumulators for profiling across all resolve() calls
    private static long globalTimeScopeBuilder = 0;
    private static long globalTimeNameResolution = 0;
    private static long globalTimeEnhancement = 0;
    private static long globalTimeLegacySync = 0;
    private static long globalTimeOther = 0;
    private static int globalResolveCount = 0;

    /**
     * Reset global timing accumulators.
     */
    public static void resetGlobalTimings() {
        globalTimeScopeBuilder = 0;
        globalTimeNameResolution = 0;
        globalTimeEnhancement = 0;
        globalTimeLegacySync = 0;
        globalTimeOther = 0;
        globalResolveCount = 0;
        // Reset detailed legacy sync timings
        globalTimeClearLinked = 0;
        globalTimeFillAttributes = 0;
        globalTimeSyncColumns = 0;
        globalTimePopulateOrphans = 0;
        globalTimeClearHints = 0;
    }

    /**
     * Get global performance timing breakdown for profiling across all resolve() calls.
     * @return formatted timing information
     */
    public static String getGlobalPerformanceTimings() {
        long total = globalTimeScopeBuilder + globalTimeNameResolution + globalTimeEnhancement + globalTimeLegacySync + globalTimeOther;
        return String.format(
            "TSQLResolver2 Global Timings (across %d resolve() calls):\n" +
            "  ScopeBuilder: %d ms (%.1f%%)\n" +
            "  NameResolution: %d ms (%.1f%%)\n" +
            "  Enhancement: %d ms (%.1f%%)\n" +
            "  LegacySync: %d ms (%.1f%%)\n" +
            "  Other: %d ms (%.1f%%)\n" +
            "  Total: %d ms",
            globalResolveCount,
            globalTimeScopeBuilder, total > 0 ? 100.0 * globalTimeScopeBuilder / total : 0,
            globalTimeNameResolution, total > 0 ? 100.0 * globalTimeNameResolution / total : 0,
            globalTimeEnhancement, total > 0 ? 100.0 * globalTimeEnhancement / total : 0,
            globalTimeLegacySync, total > 0 ? 100.0 * globalTimeLegacySync / total : 0,
            globalTimeOther, total > 0 ? 100.0 * globalTimeOther / total : 0,
            total);
    }

    /**
     * Get performance timing breakdown for profiling.
     * @return formatted timing information
     */
    public String getPerformanceTimings() {
        long total = timeScopeBuilder + timeNameResolution + timeEnhancement + timeLegacySync + timeOther;
        return String.format(
            "TSQLResolver2 Timings:\n" +
            "  ScopeBuilder: %d ms (%.1f%%)\n" +
            "  NameResolution: %d ms (%.1f%%)\n" +
            "  Enhancement: %d ms (%.1f%%)\n" +
            "  LegacySync: %d ms (%.1f%%)\n" +
            "  Other: %d ms (%.1f%%)\n" +
            "  Total: %d ms",
            timeScopeBuilder, total > 0 ? 100.0 * timeScopeBuilder / total : 0,
            timeNameResolution, total > 0 ? 100.0 * timeNameResolution / total : 0,
            timeEnhancement, total > 0 ? 100.0 * timeEnhancement / total : 0,
            timeLegacySync, total > 0 ? 100.0 * timeLegacySync / total : 0,
            timeOther, total > 0 ? 100.0 * timeOther / total : 0,
            total);
    }

    /**
     * Perform resolution on all SQL statements
     */
    public boolean resolve() {
        runOutcome = RunOutcome.attempted();

        // Traces describe exactly one run; a repeated resolve() must not mix
        // stale entries from the previous run.
        if (bindingTraceRegistry != null) {
            bindingTraceRegistry.clear();
        }
        // Reset timing counters
        timeScopeBuilder = 0;
        timeNameResolution = 0;
        timeEnhancement = 0;
        timeLegacySync = 0;
        timeOther = 0;

        // S5: clear any prior binding result so flag-off resolves stay
        // pinned to BindingResult.empty().
        bindingResult = BindingResult.empty();

        // Setup logging
        TLog.clearLogs();
        if (!TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            TLog.disableLog();
        } else {
            TLog.enableAllLevelLog();
        }

        try {
            logInfo("Starting TSQLResolver2.resolve()");

            long startTime = System.currentTimeMillis();

            CircularCteDetector.Finding circularCte =
                CircularCteDetector.find(sqlStatements, config.getVendor());
            if (circularCte != null) {
                reportCircularCte(circularCte);
                throw new IllegalArgumentException(circularCte.getMessage());
            }

            // S3: enable per-TObjectName binding-trace recording when at least
            // one binding flag is on. Allocation stays lazy — flag-off parses
            // never instantiate the trace maps (plan §10.1 perf gate).
            boolean bindingEnabled = config != null
                && (config.isEmitBindingDiagnostics()
                    || config.isBindingIncludeSuccessfulReferences());
            if (bindingEnabled) {
                resolutionContext.enableBindingTrace();
            }

            if (getSqlEnv() == null) {
                EDbVendor vendor = config.getVendor();
                if (vendor == null && sqlStatements != null && sqlStatements.size() > 0) vendor = sqlStatements.get(0).dbvendor;
                if (vendor == null) vendor = EDbVendor.dbvoracle;
                setSqlEnv(new TSQLEnv(vendor) {
                    @Override public void initSQLEnv() { }
                });
            }
            statementMetadata = new gudusoft.gsqlparser.resolver2.metadata.StatementMetadata();
            statementMetadata.collect(sqlStatements, getSqlEnv());
            scopeBuilder.setStatementMetadata(statementMetadata);

            // Phase 1: Build global scope (once for all passes)
            buildGlobalScope();

            timeOther += System.currentTimeMillis() - startTime;

            // Phase 2: Perform iterative resolution
            // (automatically completes after first pass if no second pass is needed)
            boolean ok = performIterativeResolution();

            // S5: run the binding-diagnostic post-pass exactly once after the
            // iterative loop converges, when at least one binding flag is on.
            // Plan §5.6.1: this is a diagnostic interpretation pass — it
            // reads the final resolver state but never re-binds names.
            if (ok && bindingEnabled) {
                BindingDiagnosticPostPass postPass =
                    new BindingDiagnosticPostPass(this, config);
                bindingResult = postPass.run();
            }

            runOutcome = RunOutcome.completed(ok);
            return ok;

        } catch (Exception e) {
            runOutcome = RunOutcome.failed(e);
            logError("Exception in TSQLResolver2.resolve(): " + e.getMessage());
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                e.printStackTrace();
            }
            return false;
        } finally {
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                TBaseType.dumpLogs(false);
            }
        }
    }

    private void reportCircularCte(CircularCteDetector.Finding finding) {
        TCTE cte = finding.getErrorNode();
        if (cte == null || cte.getSubquery() == null || cte.getTableName() == null
                || cte.getTableName().getStartToken() == null) {
            return;
        }

        TCustomSqlStatement statement = cte.getSubquery();
        statement.parseerrormessagehandle(new TSyntaxError(
            cte.getTableName().getStartToken(),
            finding.getMessage(),
            EErrorType.sperror,
            TBaseType.MSG_ERROR_SYNTAX_ERROR,
            statement));
    }

    /**
     * Perform iterative resolution.
     * Automatically converges after first pass if no additional passes are needed.
     *
     * Architecture:
     * - Pass 1: Build scope tree + initial name resolution
     * - Pass 2-N: Reuse scope tree, collect evidence, infer columns, re-resolve names
     *
     * This separation allows:
     * 1. Scopes to accumulate inferred columns across iterations
     * 2. Later scopes to reference earlier scopes' inferred columns
     * 3. Forward references to be resolved in subsequent passes
     */
    private boolean performIterativeResolution() {
        logInfo("Performing iterative resolution (max iterations: " + config.getMaxIterations() + ")");

        int passNumber = 1;
        ResolutionStatistics previousStats = null;
        boolean continueIterating = true;
        boolean scopesBuilt = false;

        while (continueIterating) {
            logInfo("=== Pass " + passNumber + " ===");

            // Create a resolution pass
            ResolutionPass pass = new ResolutionPass(passNumber, previousStats);

            if (passNumber == 1) {
                // ========== PASS 1: Build scope tree + initial resolution ==========
                logInfo("Pass 1: Building scope tree using ScopeBuilder and performing initial resolution");

                // Clear all state for fresh start
                resolutionContext.clear();
                columnToScopeMap.clear();
                fromScopeIndexCache.clear();
                dmlIndexCache.clear();
                teradataNamedAliasCache.clear();
                allColumnReferences.clear();

                // Use ScopeBuilder to build complete scope tree (handles all nesting correctly)
                long scopeBuilderStart = System.currentTimeMillis();
                scopeBuildResult = scopeBuilder.build(sqlStatements);

                // Get global scope from builder
                globalScope = scopeBuildResult.getGlobalScope();

                // Copy column references and scope mappings from ScopeBuildResult
                columnToScopeMap.putAll(scopeBuildResult.getColumnToScopeMap());
                allColumnReferences.addAll(scopeBuildResult.getAllColumnReferences());
                timeScopeBuilder += System.currentTimeMillis() - scopeBuilderStart;

                logInfo("ScopeBuilder complete: " + scopeBuildResult.getStatistics());
                logInfo("Built " + scopeBuildResult.getStatementScopeMap().size() + " SelectScopes");

                // Initialize NamespaceEnhancer with scope tree (caches star namespaces)
                namespaceEnhancer.initialize(scopeBuildResult);
                namespaceEnhancer.startPass(passNumber);

                // Get SET clause target columns that should not be re-resolved
                Set<TObjectName> setClauseTargetColumns = scopeBuilder.getSetClauseTargetColumns();

                // Get INSERT ALL target columns that should not be re-resolved
                Set<TObjectName> insertAllTargetColumns = scopeBuilder.getInsertAllTargetColumns();

                // Get MERGE INSERT VALUES columns that need sourceTable restoration after resolution
                Map<TObjectName, TTable> mergeInsertValuesColumns = scopeBuilder.getMergeInsertValuesColumns();

                // Perform initial name resolution for all collected columns
                logInfo("Performing initial name resolution for " + allColumnReferences.size() + " column references");
                long nameResStart = System.currentTimeMillis();
                for (TObjectName objName : allColumnReferences) {
                    // Skip SET clause target columns - they already have sourceTable correctly set
                    // to the UPDATE target table and should NOT be resolved through star columns
                    if (setClauseTargetColumns.contains(objName)) {
                        continue;
                    }

                    // Skip INSERT ALL target columns - they already have sourceTable correctly set
                    // to the INSERT target table and should NOT be resolved against the subquery scope
                    if (insertAllTargetColumns.contains(objName)) {
                        continue;
                    }

                    // Skip ClickHouse expression-CTE aliases (WITH <expr> AS ident) -
                    // phase 1 resolved them to scalar symbols; they are not columns
                    // of any relation and must not enter namespace matching or the
                    // enhancement pipeline.
                    if (objName.getExpressionCteRef() != null && objName.getSourceTable() == null) {
                        continue;
                    }

                    IScope scope = columnToScopeMap.get(objName);
                    if (scope != null) {
                        nameResolver.resolve(objName, scope);

                        // Handle USING column priority for JOIN...USING syntax
                        handleUsingColumnResolution(objName);

                        // Handle Teradata NAMED alias resolution
                        handleTeradataNamedAliasResolution(objName);
                        handleQualifyClauseAliasResolution(objName);
                        handleSelectListAliasResolution(objName);

                        // Handle subquery aliased/calculated column resolution
                        // Ensures aliased columns don't incorrectly trace to base tables
                        handleSubqueryAliasedColumnResolution(objName);

                        // The correction handlers above may change or clear
                        // sourceTable AFTER the binding trace was captured; a
                        // trace whose table definition no longer matches is
                        // stale and must not survive as complete (fail-closed).
                        if (bindingTraceRegistry != null) {
                            bindingTraceRegistry.invalidateIfInconsistent(objName);
                        }

                        // Collect unresolved references for enhancement
                        collectForEnhancementIfNeeded(objName, scope);
                    }
                }

                // Restore sourceTable for MERGE INSERT VALUES columns after name resolution.
                // Name resolution may have set an AMBIGUOUS resolution (e.g., column 'product'
                // appears in both target and source tables through the ON clause). In MERGE
                // semantics, WHEN NOT MATCHED VALUES columns always reference the USING (source)
                // table.
                //
                // For AMBIGUOUS resolution: clear it so getSourceTable() returns the actual field
                // value (the USING table). AMBIGUOUS means the column was found in both target and
                // source namespaces, but semantically it must reference the source.
                //
                // For EXACT_MATCH resolution: keep it because it contains star column push-down
                // tracing info (e.g., when USING is a subquery with SELECT *, the resolution
                // traces the VALUES column to the physical table inside the subquery).
                for (Map.Entry<TObjectName, TTable> entry : mergeInsertValuesColumns.entrySet()) {
                    TObjectName col = entry.getKey();
                    TTable usingTable = entry.getValue();
                    ResolutionResult res = col.getResolution();
                    if (res != null && res.isAmbiguous()) {
                        col.setResolution(null);
                    }
                    col.setSourceTable(usingTable);
                    // This restore runs AFTER the per-reference consistency
                    // check; a scope trace observed before it is now stale.
                    if (bindingTraceRegistry != null) {
                        bindingTraceRegistry.invalidateIfInconsistent(col);
                    }
                }

                timeNameResolution += System.currentTimeMillis() - nameResStart;

                // Explicit Enhancement Phase: Add collected columns to namespaces
                long enhanceStart = System.currentTimeMillis();
                EnhancementResult enhanceResult = namespaceEnhancer.enhance();
                timeEnhancement += System.currentTimeMillis() - enhanceStart;
                logInfo("Pass 1 enhancement: " + enhanceResult.getTotalAdded() + " columns added to namespaces");

                scopesBuilt = true;
                logInfo("Pass 1 complete. Resolved " + allColumnReferences.size() + " column references.");


            } else {
                // ========== PASS 2+: Explicit Enhancement + Re-resolve ==========
                logInfo("Pass " + passNumber + ": Explicit namespace enhancement and re-resolution");

                // ======== Phase A: Start New Pass ========
                namespaceEnhancer.startPass(passNumber);

                // ======== Phase B: Clear Resolution Results (keep scopes!) ========
                logInfo("Phase B: Clearing resolution results (scopes preserved)");
                resolutionContext.clear();

                // ======== Phase C: Re-resolve with Enhanced Namespaces ========
                logInfo("Phase C: Re-resolving with enhanced namespaces");

                // Get SET clause target columns that should not be re-resolved
                Set<TObjectName> setClauseTargetColumns = scopeBuilder.getSetClauseTargetColumns();

                // Get INSERT ALL target columns that should not be re-resolved
                Set<TObjectName> insertAllTargetColumns = scopeBuilder.getInsertAllTargetColumns();

                // Get MERGE INSERT VALUES columns that need sourceTable restoration after resolution
                Map<TObjectName, TTable> mergeInsertValuesColumns = scopeBuilder.getMergeInsertValuesColumns();

                // Re-resolve all column references using their original scopes
                // Scopes are reused from Pass 1, but namespaces may have been enhanced
                for (TObjectName objName : allColumnReferences) {
                    // Skip SET clause target columns - they already have sourceTable correctly set
                    // to the UPDATE target table and should NOT be resolved through star columns
                    if (setClauseTargetColumns.contains(objName)) {
                        continue;
                    }

                    // Skip INSERT ALL target columns - they already have sourceTable correctly set
                    // to the INSERT target table and should NOT be resolved against the subquery scope
                    if (insertAllTargetColumns.contains(objName)) {
                        continue;
                    }

                    // Skip ClickHouse expression-CTE aliases (WITH <expr> AS ident) -
                    // phase 1 resolved them to scalar symbols; they are not columns
                    // of any relation and must not enter namespace matching or the
                    // enhancement pipeline.
                    if (objName.getExpressionCteRef() != null && objName.getSourceTable() == null) {
                        continue;
                    }

                    IScope scope = columnToScopeMap.get(objName);
                    if (scope != null) {
                        nameResolver.resolve(objName, scope);

                        // Handle USING column priority for JOIN...USING syntax
                        handleUsingColumnResolution(objName);

                        // Handle Teradata NAMED alias resolution
                        handleTeradataNamedAliasResolution(objName);
                        handleQualifyClauseAliasResolution(objName);
                        handleSelectListAliasResolution(objName);

                        // Handle subquery aliased/calculated column resolution
                        // Ensures aliased columns don't incorrectly trace to base tables
                        handleSubqueryAliasedColumnResolution(objName);

                        // The correction handlers above may change or clear
                        // sourceTable AFTER the binding trace was captured; a
                        // trace whose table definition no longer matches is
                        // stale and must not survive as complete (fail-closed).
                        if (bindingTraceRegistry != null) {
                            bindingTraceRegistry.invalidateIfInconsistent(objName);
                        }

                        // Collect for next enhancement pass if still targets star namespace
                        collectForEnhancementIfNeeded(objName, scope);
                    }
                }

                // Restore sourceTable for MERGE INSERT VALUES columns after re-resolution
                for (Map.Entry<TObjectName, TTable> entry : mergeInsertValuesColumns.entrySet()) {
                    ResolutionResult res = entry.getKey().getResolution();
                    if (res != null && res.isAmbiguous()) {
                        entry.getKey().setResolution(null);
                    }
                    entry.getKey().setSourceTable(entry.getValue());
                    // Same staleness rule as the pass-1 restore above.
                    if (bindingTraceRegistry != null) {
                        bindingTraceRegistry.invalidateIfInconsistent(entry.getKey());
                    }
                }

                // ======== Phase D: Explicit Namespace Enhancement ========
                logInfo("Phase D: Explicit namespace enhancement");
                EnhancementResult enhanceResult = namespaceEnhancer.enhance();
                logInfo("Pass " + passNumber + " enhancement: " +
                       enhanceResult.getTotalAdded() + " columns added, " +
                       enhanceResult.getTotalSkipped() + " skipped (existing)");

                // Legacy support: also run old evidence collection (if needed)
                if (config.isEvidenceCollectionEnabled()) {
                    runLegacyEvidenceCollection();
                }
            }

            // Get statistics after this pass
            ResolutionStatistics currentStats = getStatistics();
            pass.complete(currentStats);

            // Record this pass
            convergenceDetector.recordPass(pass);
            passHistory.add(pass);

            logInfo(pass.getSummary());

            // Check convergence
            ConvergenceDetector.ConvergenceResult convergence = convergenceDetector.checkConvergence();
            if (convergence.hasConverged()) {
                logInfo("Convergence detected: " + convergence.getReason());
                pass.setStopReason(convergence.getReason());
                continueIterating = false;
            } else {
                // Prepare for next pass
                previousStats = currentStats;
                passNumber++;
            }
        }

        // Create cloned columns for star column tracing
        // This is a CORE part of TSQLResolver2 - when a column traces through a CTE/subquery
        // with SELECT * to a physical table, we create a cloned TObjectName with sourceTable
        // pointing to the traced physical table. This ensures complete lineage tracking.
        createTracedColumnClones();

        // Sync to legacy structures if enabled
        if (config.isLegacyCompatibilityEnabled()) {
            long syncStart = System.currentTimeMillis();
            syncToLegacyStructures();
            timeLegacySync += System.currentTimeMillis() - syncStart;
        }

        // Print final statistics
        logInfo("Iterative resolution complete after " + passHistory.size() + " passes");
        ResolutionStatistics finalStats = getStatistics();
        logInfo("Final statistics: " + finalStats);

        // Print namespace enhancement summary if in debug mode
        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("=== Namespace Enhancement Summary ===");
            logInfo("Total columns added: " + namespaceEnhancer.getTotalColumnsAdded());
        }

        // Print performance timing breakdown
        logInfo(getPerformanceTimings());

        // Accumulate to global timings for profiling
        globalTimeScopeBuilder += timeScopeBuilder;
        globalTimeNameResolution += timeNameResolution;
        globalTimeEnhancement += timeEnhancement;
        globalTimeLegacySync += timeLegacySync;
        globalTimeOther += timeOther;
        globalResolveCount++;

        return true;
    }

    /**
     * Run legacy evidence collection (deprecated).
     * Kept for backward compatibility.
     */
    @SuppressWarnings("deprecation")
    private void runLegacyEvidenceCollection() {
        logInfo("Running legacy evidence collection (deprecated)");

        gudusoft.gsqlparser.resolver2.inference.EvidenceCollector evidenceCollector =
            new gudusoft.gsqlparser.resolver2.inference.EvidenceCollector();

        int evidenceCount = 0;
        for (int i = 0; i < sqlStatements.size(); i++) {
            Object stmt = sqlStatements.get(i);
            if (stmt instanceof TSelectSqlStatement) {
                List<gudusoft.gsqlparser.resolver2.inference.InferenceEvidence> stmtEvidence =
                    evidenceCollector.collectFromSelect((TSelectSqlStatement) stmt);
                evidenceCount += stmtEvidence.size();
            }
        }

        logInfo("Legacy evidence collection: " + evidenceCount + " items");
    }

    /**
     * Get the namespace enhancer for external access to enhancement history.
     *
     * @return the namespace enhancer
     */
    public NamespaceEnhancer getNamespaceEnhancer() {
        return namespaceEnhancer;
    }

    /**
     * Get a detailed enhancement report.
     *
     * @return detailed report string
     */
    public String getEnhancementReport() {
        return namespaceEnhancer.generateReport();
    }

    /**
     * Re-process a statement for name resolution only (without rebuilding scopes).
     * This is used in Pass 2+ to re-resolve names using enhanced scopes.
     *
     * CRITICAL (Principle 1: Scope完全复用):
     * - Scope tree is built ONCE in Pass 1 and completely reused in Pass 2+
     * - This method MUST NOT call processStatement() which rebuilds scopes
     * - Instead, it iterates through allColumnReferences and re-resolves each
     *   column using its original scope from columnToScopeMap
     *
     * This allows:
     * - Namespaces to be enhanced across iterations (Principle 2)
     * - Star columns to benefit from reverse inference (Principle 3)
     * - All previous inference results to be preserved
     */
    private void reprocessStatementNamesOnly(Object statement) {
        logDebug("Re-resolving column references without rebuilding scopes");

        // Re-resolve all column references using their original scopes
        // The scopes are reused from Pass 1, but their namespaces may have been enhanced
        for (TObjectName objName : allColumnReferences) {
            IScope scope = columnToScopeMap.get(objName);
            if (scope != null) {
                // Re-resolve this column using the (potentially enhanced) scope
                nameResolver.resolve(objName, scope);

                // Handle USING column priority for JOIN...USING syntax
                handleUsingColumnResolution(objName);

                // Handle Teradata NAMED alias resolution
                handleTeradataNamedAliasResolution(objName);
                handleQualifyClauseAliasResolution(objName);
                handleSelectListAliasResolution(objName);

                // Collect for next enhancement pass if still unresolved
                collectForEnhancementIfNeeded(objName, scope);
            } else {
                logError("No scope found for column: " + objName);
            }
        }
    }

    /**
     * Handle special resolution for USING columns in JOIN...USING syntax.
     * In "a JOIN table2 USING (id)", the USING column exists in BOTH tables.
     * - The synthetic column (clone) resolves to the right-side table (table2)
     * - The original USING column resolves to the left-side table (a)
     *
     * @param objName The column reference
     */
    private void handleUsingColumnResolution(TObjectName objName) {
        if (objName == null || scopeBuildResult == null) return;

        // Check if this is a synthetic USING column (should resolve to right table)
        TTable rightTable = scopeBuildResult.getUsingColumnRightTable(objName);
        if (rightTable != null) {
            // This is the synthetic USING column - set its sourceTable to the right-side table
            objName.setSourceTable(rightTable);

            // Create a proper resolution with the right-side table
            gudusoft.gsqlparser.resolver2.model.ColumnSource source =
                new gudusoft.gsqlparser.resolver2.model.ColumnSource(
                    null,  // no namespace for USING columns
                    objName.getColumnNameOnly(),
                    null,  // no definition node
                    1.0,   // high confidence
                    "using_column_right",
                    rightTable  // override table - the right-side table of the JOIN
                );
            gudusoft.gsqlparser.resolver2.model.ResolutionResult result =
                gudusoft.gsqlparser.resolver2.model.ResolutionResult.exactMatch(source);

            // Update the TObjectName's resolution so formatter uses correct finalTable
            objName.setResolution(result);

            // Also register in ResolutionContext so getReferencesTo(table) can find it
            resolutionContext.registerResolution(objName, result);

            logDebug("USING column " + objName.getColumnNameOnly() +
                " -> right-side table " + rightTable.getName());
            return;
        }

        // Check if this is the original USING column (should resolve to left table)
        TTable leftTable = scopeBuildResult.getUsingColumnLeftTable(objName);
        if (leftTable != null) {
            // This is the original USING column - set its sourceTable to the left-side table
            objName.setSourceTable(leftTable);

            // Create a proper resolution with the left-side table
            gudusoft.gsqlparser.resolver2.model.ColumnSource source =
                new gudusoft.gsqlparser.resolver2.model.ColumnSource(
                    null,  // no namespace for USING columns
                    objName.getColumnNameOnly(),
                    null,  // no definition node
                    1.0,   // high confidence
                    "using_column_left",
                    leftTable  // override table - the left-side table of the JOIN
                );
            gudusoft.gsqlparser.resolver2.model.ResolutionResult result =
                gudusoft.gsqlparser.resolver2.model.ResolutionResult.exactMatch(source);

            // Update the TObjectName's resolution so formatter uses correct finalTable
            objName.setResolution(result);

            // Also register in ResolutionContext so getReferencesTo(table) can find it
            resolutionContext.registerResolution(objName, result);

            logDebug("USING column " + objName.getColumnNameOnly() +
                " -> left-side table " + leftTable.getName());
        }
    }

    /**
     * Handle Teradata NAMED alias resolution.
     *
     * <p>In Teradata, NAMED aliases defined in the SELECT list (using the {@code (NAMED alias)} syntax)
     * can be referenced in the WHERE and QUALIFY clauses of the same SELECT statement. This is different
     * from standard SQL where column aliases are only visible in ORDER BY.</p>
     *
     * <p>This method checks if a resolved column matches a NAMED alias from the enclosing SELECT list.
     * If it does, the resolution is updated to indicate this is a calculated column (alias), not a
     * physical column from the table.</p>
     *
     * <p>Example:</p>
     * <pre>
     * SELECT USI_ID, SUBS_ID,
     *        (CAST(:param AS TIMESTAMP(0)))(NAMED REPORT_DTTM)
     * FROM PRD2_ODW.SUBS_USI_HISTORY
     * WHERE stime <= REPORT_DTTM AND etime > REPORT_DTTM
     * </pre>
     * <p>Here, REPORT_DTTM references in WHERE should NOT be linked to PRD2_ODW.SUBS_USI_HISTORY
     * because REPORT_DTTM is a NAMED alias, not a physical column.</p>
     *
     * @param objName The column reference to check
     */
    private void handleTeradataNamedAliasResolution(TObjectName objName) {
        if (objName == null || sqlStatements == null || sqlStatements.size() == 0) return;

        // Only applies to Teradata
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        if (dbVendor != EDbVendor.dbvteradata) return;

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

        // Only apply to UNQUALIFIED column references (no table prefix)
        // If a column has a table qualifier like "CP.CALC_PLATFORM_ID", it's clearly
        // referencing a specific table's column, not a NAMED alias
        if (objName.getTableToken() != null) return;

        // Get the scope for this column reference
        IScope scope = columnToScopeMap.get(objName);
        if (scope == null) return;

        // Find the enclosing SELECT statement from the scope
        TSelectSqlStatement enclosingSelect = findEnclosingSelectFromScope(scope);
        if (enclosingSelect == null) return;

        // Optimization C: Use cached index for O(1) lookup instead of O(N) iteration
        Map<String, TResultColumn> aliasIndex = getTeradataNamedAliasIndex(enclosingSelect);
        if (aliasIndex == null || aliasIndex.isEmpty()) return;

        // S2: look up via the vendor-aware key. Teradata is case-insensitive
        // for unquoted column aliases but case-sensitive for quoted ones, so
        // raw {@code toLowerCase()} drops quoted-alias correctness.
        //
        // CRITICAL — vendor source-of-truth invariant (codex round-2 review):
        // both this lookup and the storage path in
        // {@link #getTeradataNamedAliasIndex} read from the SAME source
        // {@code sqlStatements.get(0).dbvendor}: this method's local
        // {@code dbVendor} variable is initialized from it at line 916
        // above, and the storage path passes the same value. They cannot
        // diverge.
        String aliasKey = normalizeName(
                dbVendor,
                gudusoft.gsqlparser.sqlenv.ESQLDataObjectType.dotColumn,
                columnName);
        TResultColumn resultCol = aliasIndex.get(aliasKey);
        if (resultCol == null) {
            // Fallback: quoted-vs-unquoted mix in the same SELECT (rare).
            // Walk the cached map with a vendor-aware compare so the
            // normalized-key fast probe miss does not become a false negative.
            for (Map.Entry<String, TResultColumn> entry : aliasIndex.entrySet()) {
                if (sameName(
                        dbVendor,
                        gudusoft.gsqlparser.sqlenv.ESQLDataObjectType.dotColumn,
                        entry.getKey(),
                        columnName)) {
                    resultCol = entry.getValue();
                    break;
                }
            }
        }
        if (resultCol == null) return;

        // Skip if objName is part of this result column's expression
        // This handles cases like "CAST(ID AS DECIMAL) AS ID" where the ID inside
        // CAST is the source column, not a reference to the ID alias
        if (isColumnWithinResultColumn(objName, resultCol)) {
            return;
        }

        // Found a matching NAMED alias
        // Clear the source table since this is an alias, not a physical column
        objName.setSourceTable(null);

        // Create a new ColumnSource with the TResultColumn as the definition node
        // This will make isCalculatedColumn() return true
        ColumnSource source = new ColumnSource(
            null,  // namespace - not from a table
            columnName,
            resultCol,  // definition node - the TResultColumn with the alias
            1.0,   // high confidence
            "teradata_named_alias"
        );
        ResolutionResult result = ResolutionResult.exactMatch(source);
        objName.setResolution(result);
        resolutionContext.registerResolution(objName, result);

        logDebug("Teradata NAMED alias: " + columnName + " -> alias from SELECT list");
    }

    /**
     * Handle QUALIFY clause alias resolution for Snowflake, BigQuery, and Databricks.
     *
     * <p>In Snowflake, BigQuery, and Databricks, column aliases defined in the SELECT list
     * can be referenced in the QUALIFY clause. This is different from standard SQL where
     * column aliases are only visible in ORDER BY.</p>
     *
     * <p>This method checks if a column reference in the QUALIFY clause matches an alias
     * from the enclosing SELECT list. If it does, the resolution is updated to indicate
     * this is a calculated column (alias), not a physical column from the table.</p>
     *
     * <p>Example:</p>
     * <pre>
     * SELECT RoomNumber, RoomType, BlockFloor,
     *        ROW_NUMBER() OVER (PARTITION BY RoomType ORDER BY BlockFloor) AS row_num
     * FROM Hospital.Room
     * QUALIFY row_num = 1
     * </pre>
     * <p>Here, row_num in QUALIFY should NOT be linked to Hospital.Room because
     * row_num is an alias for the window function, not a physical column.</p>
     *
     * @param objName The column reference to check
     */
    private void handleQualifyClauseAliasResolution(TObjectName objName) {
        if (objName == null || sqlStatements == null || sqlStatements.size() == 0) return;

        // Only applies to databases that support QUALIFY with alias visibility
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        if (dbVendor != EDbVendor.dbvsnowflake &&
            dbVendor != EDbVendor.dbvbigquery &&
            dbVendor != EDbVendor.dbvdatabricks) return;

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

        // Only apply to UNQUALIFIED column references (no table prefix)
        if (objName.getTableToken() != null) return;

        // Check if this column is within a QUALIFY clause
        if (!isInQualifyClause(objName)) return;

        // Get the scope for this column reference
        IScope scope = columnToScopeMap.get(objName);
        if (scope == null) return;

        // Find the enclosing SELECT statement from the scope
        TSelectSqlStatement enclosingSelect = findEnclosingSelectFromScope(scope);
        if (enclosingSelect == null) return;

        // Look for a matching alias in the SELECT list
        TResultColumnList resultColumns = enclosingSelect.getResultColumnList();
        if (resultColumns == null || resultColumns.size() == 0) return;

        TResultColumn matchingResultCol = null;
        for (int i = 0; i < resultColumns.size(); i++) {
            TResultColumn resultCol = resultColumns.getResultColumn(i);
            if (resultCol == null) continue;

            // Check if this result column has an alias matching the column name
            if (resultCol.getAliasClause() != null &&
                resultCol.getAliasClause().getAliasName() != null) {
                String aliasName = resultCol.getAliasClause().getAliasName().toString();
                if (aliasName != null && sameName(dbVendor, ESQLDataObjectType.dotColumn, aliasName, columnName)) {
                    matchingResultCol = resultCol;
                    break;
                }
            }
        }

        if (matchingResultCol == null) return;

        // Found a matching alias - clear the source table since this is an alias, not a physical column
        objName.setSourceTable(null);

        // Create a new ColumnSource with the TResultColumn as the definition node
        // This will make isCalculatedColumn() return true
        ColumnSource source = new ColumnSource(
            null,  // namespace - not from a table
            columnName,
            matchingResultCol,  // definition node - the TResultColumn with the alias
            1.0,   // high confidence
            "qualify_clause_alias"
        );
        ResolutionResult result = ResolutionResult.exactMatch(source);
        objName.setResolution(result);
        resolutionContext.registerResolution(objName, result);

        logDebug("QUALIFY clause alias: " + columnName + " -> alias from SELECT list");
    }

    /**
     * Handle a reference to a SELECT-list column alias from elsewhere in the same
     * SELECT block — the general "lateral column alias" case, of which the Teradata
     * NAMED and QUALIFY handlers above are two special cases.
     *
     * <p>Snowflake, Teradata, Vertica and other dialects whose
     * {@link TSQLEnv#isAliasReferenceForbidden} entry is {@code false} let a
     * SELECT-list alias be referenced in the same block's WHERE / HAVING:</p>
     * <pre>
     * WITH base AS (SELECT id, val FROM t1),
     *      final AS (SELECT id, TO_CHAR(val,'mm/dd/yyyy')::TIMESTAMP_NTZ AS derived_col
     *                FROM base
     *                WHERE derived_col &gt; CURRENT_DATE)
     * SELECT * FROM final
     * </pre>
     *
     * <p>{@code derived_col} in that WHERE is the alias, not a column of {@code base}
     * — {@code base} projects only {@code id} and {@code val}. Without this handler
     * resolver2 fell through to the namespace's non-authoritative MAYBE path and
     * stamped {@code sourceTable = base}, i.e. a confident link to a table that
     * provably has no such column, while {@code sourceColumn} already pointed at the
     * alias. Nothing anywhere flagged the contradiction (MantisBT 4659).</p>
     *
     * <p><b>What this deliberately does NOT do:</b> it never overrides a link the
     * namespace can PROVE. When the FROM-clause namespace authoritatively exposes a
     * column of that name — real catalog metadata, an explicit CTE column list, a
     * subquery projection — the alias and the column collide and the existing
     * table link is kept. Only a MAYBE/inferred link, which was never proven in the
     * first place, is replaced. See {@link BindingMetadataAuthority#lookup}.</p>
     *
     * <p>The alias-binding rule itself is NOT reimplemented here: it is
     * {@link TResultColumnList#findLateralAliasDefinition}, shared with Phase 1
     * {@code TCustomSqlStatement.linkColumnToTable()}, so the publish path and the
     * lookup path cannot drift apart.</p>
     *
     * @param objName The column reference to check
     */
    private void handleSelectListAliasResolution(TObjectName objName) {
        if (objName == null || sqlStatements == null || sqlStatements.size() == 0) return;

        // Only apply to UNQUALIFIED column references (no table prefix): a qualified
        // "t.derived_col" names that table's column, never a SELECT-list alias.
        if (objName.getTableToken() != null) return;

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

        // A more specific handler (Teradata NAMED alias, QUALIFY) may already have bound
        // this reference to its alias. Leave it alone rather than restamping it with a
        // weaker evidence string.
        ResolutionResult existing = objName.getResolution();
        if (existing != null && existing.getColumnSource() != null
                && existing.getColumnSource().getSourceNamespace() == null
                && existing.getColumnSource().isColumnAlias()) {
            return;
        }

        IScope scope = columnToScopeMap.get(objName);
        if (scope == null) return;

        TSelectSqlStatement enclosingSelect = findEnclosingSelectFromScope(scope);
        if (enclosingSelect == null) return;

        TResultColumnList resultColumns = enclosingSelect.getResultColumnList();

        // Read the vendor from the enclosing statement rather than sqlStatements.get(0):
        // a statement list can mix vendors only in theory, but the enclosing SELECT is
        // always the authority for its own alias visibility rules.
        EDbVendor dbVendor = enclosingSelect.dbvendor;
        if (dbVendor == null) dbVendor = sqlStatements.get(0).dbvendor;

        // Only where the alias is actually visible, per VENDOR and per CLAUSE.
        // Textual position alone is NOT enough: a JOIN ... ON predicate is written
        // after the SELECT list but belongs to the FROM clause, which is evaluated
        // before any alias exists, so a name there is a table column even when it
        // happens to match an alias. Getting this wrong rebinds real join columns
        // (caught by dataflowTest.testICYIYY / snowflake/1027.sql).
        TResultColumn matchingResultCol = null;
        if (resultColumns != null && resultColumns.size() > 0
                && isClauseExposingSelectListAlias(dbVendor, objName.getLocation(), enclosingSelect.getGsqlparser())) {
            matchingResultCol = resultColumns.findLateralAliasDefinition(dbVendor, objName);
        }
        // SAP IQ and SQL Anywhere: the alias of an enclosing block, by the rule shared with
        // Phase 1. Only the tables as StatementMetadata bound them in their own statement's
        // context prove a column absent (no catalog lookup: its current database is the
        // script's last one by now). Phase 1 decided on the catalog it had then; where the
        // two differ, resolver2's answer is the one kept, getSourceColumn() included.
        if (matchingResultCol == null) {
            matchingResultCol = enclosingSelect.findEnclosingAliasDefinition(objName, null);
            if (matchingResultCol != null && objName.getSourceColumn() != matchingResultCol) {
                objName.setSourceColumn(matchingResultCol);
                matchingResultCol.getTargetColumns().addObjectName(objName);
            }
        }
        if (matchingResultCol == null) {
            dropUnconfirmedEnclosingAlias(objName, resultColumns, dbVendor);
            return;
        }

        // Defensive: never re-point a column that lives inside the alias's own
        // defining expression. findLateralAliasDefinition()'s ordering rule already
        // excludes it; this keeps the guarantee local and explicit.
        if (isColumnWithinResultColumn(objName, matchingResultCol)) return;

        // In SAP IQ and SQL Anywhere the alias occludes a column of the same name, proven
        // or not, so the two checks below, which give a real column precedence, do not apply.
        boolean occluding = isAliasOccludingColumn(dbVendor);

        // MySQL: in HAVING the alias yields to a column of that name only when GROUP BY names
        // the column, proven or not; otherwise the alias wins over a column it collides with.
        if (!occluding && objName.getLocation() == ESqlClause.having
                && TSQLEnv.isHavingAliasOutrankingUngroupedColumn(dbVendor, enclosingSelect.getGsqlparser())) {
            if (objName.getSourceTable() != null && isColumnNamedInGroupBy(enclosingSelect, columnName, identifierServiceFor(dbVendor))) {
                dropAliasLinkOfKeptColumn(objName, matchingResultCol);
                return;
            }
            occluding = true;
        }

        // Do not discard a PROVABLE column link. Only an unproven (MAYBE/inferred)
        // match may be replaced by the alias binding.
        if (!occluding && isCurrentResolutionAuthoritative(objName, columnName)) {
            dropAliasLinkOfKeptColumn(objName, matchingResultCol);
            return;
        }

        // Nor discard a link the SQL itself evidences. In
        //   SELECT CAST(id AS INT) AS id FROM t1 WHERE id > 1
        // the inner `id` cannot be referring to the alias being defined, so it can
        // only be a real column of t1 — which means t1 HAS an `id`, and Snowflake's
        // collision rule gives that real column precedence over the alias. No
        // catalog is needed to know this; the query says it.
        if (!occluding && aliasDefinitionReferencesSameColumn(matchingResultCol, columnName, dbVendor,
                objName)) {
            dropAliasLinkOfKeptColumn(objName, matchingResultCol);
            return;
        }

        // Clear the source table: an alias is not a physical column of any table.
        objName.setSourceTable(null);
        // ...and point getSourceColumn() at the alias, as Phase 1 does when it finds no column: where Phase 1 kept
        // a column it found (a MySQL HAVING name GROUP BY does not name), resolver2's answer is the one kept.
        if (objName.getSourceColumn() != matchingResultCol) {
            objName.setSourceColumnOnly(matchingResultCol);
            matchingResultCol.getTargetColumns().addObjectName(objName);
        }
        // ...and the candidates that came with it, or a consumer reading
        // getCandidateTables() would still publish the tables we just ruled out.
        objName.clearCandidateTables();

        ColumnSource source = new ColumnSource(
            null,  // namespace - not from a table
            columnName,
            matchingResultCol,  // definition node - the TResultColumn with the alias
            1.0,   // high confidence
            "select_list_alias"
        );
        ResolutionResult result = ResolutionResult.exactMatch(source);
        objName.setResolution(result);
        // replace, not register: this CORRECTS an earlier resolution for the same
        // reference, and registerResolution() would leave the superseded table
        // binding in the reverse indexes and double-count the reference.
        resolutionContext.replaceResolution(objName, result);

        logDebug("SELECT list alias: " + columnName + " -> alias from SELECT list");
    }

    /**
     * resolver2 keeps the column link of {@code objName}: drop the link Phase 1 made to the alias
     * {@code aliasColumn} when it found no column, so that {@code getSourceColumn()} does not contradict
     * {@code getSourceTable()} and lineage does not follow it.
     */
    private static void dropAliasLinkOfKeptColumn(TObjectName objName, TResultColumn aliasColumn) {
        if (objName.getSourceColumn() != aliasColumn || objName.getSourceTable() == null) return;
        objName.setSourceColumnOnly(null);
        aliasColumn.getTargetColumns().removeElement(objName);
    }

    /**
     * Does GROUP BY of {@code select} name a column called {@code columnName}, qualified or not? For MySQL's HAVING
     * precedence ({@link TSQLEnv#isHavingAliasOutrankingUngroupedColumn}); the caller has already found a column of
     * that name for the HAVING reference. Decided on the GROUP BY text, not on its binding: GROUP BY references may
     * not be bound yet when HAVING is.
     */
    private static boolean isColumnNamedInGroupBy(TSelectSqlStatement select, String columnName,
                                                  IdentifierService names) {
        TGroupBy groupBy = select.getGroupByClause();
        if (groupBy == null || groupBy.getItems() == null) return false;
        for (int i = 0; i < groupBy.getItems().size(); i++) {
            TGroupByItem item = groupBy.getItems().getGroupByItem(i);
            TExpression expr = (item == null) ? null : item.getExpr();
            if (expr == null || expr.getExpressionType() != EExpressionType.simple_object_name_t) continue;
            TObjectName name = expr.getObjectOperand();
            if (name == null) continue;
            // the column part; a one-token name GROUP BY bound to an alias carries it as its object token
            TSourceToken part = name.getPartToken();
            if (part == null && name.getStartToken() != null && name.getStartToken() == name.getEndToken()) {
                part = name.getStartToken();
            }
            if (part == null) continue;
            if (names.areEqual(part.toString(), columnName, ESQLDataObjectType.dotColumn)) {
                return true;
            }
        }
        return false;
    }

    /**
     * Phase 1 bound {@code objName} to an alias of an enclosing block that resolver2 does not
     * confirm (their catalogs differed): drop that link, so that {@code getSourceColumn()} does not
     * contradict resolver2's resolution and lineage does not follow it (Mantis 4752).
     */
    private static void dropUnconfirmedEnclosingAlias(TObjectName objName, TResultColumnList ownResultColumns,
            EDbVendor vendor) {
        if (!isAliasOccludingColumn(vendor)) return;
        TResultColumn bound = objName.getSourceColumn();
        if (bound == null || bound.getAliasClause() == null || !bound.isMatchedUsingAlias(objName)) return;
        if (ownResultColumns != null) {
            for (int i = 0; i < ownResultColumns.size(); i++) {
                if (ownResultColumns.getResultColumn(i) == bound) return; // an alias of its own block
            }
        }
        objName.setSourceColumnOnly(null);
        bound.getTargetColumns().removeElement(objName);
    }

    /**
     * See {@link TSQLEnv#isAliasOccludingColumn}: in SAP IQ and SQL Anywhere a column
     * link, proven or not, never keeps an unqualified name that matches an alias of
     * its SELECT block (Mantis 4752).
     */
    private static boolean isAliasOccludingColumn(EDbVendor vendor) {
        return TSQLEnv.isAliasOccludingColumn(vendor);
    }

    /**
     * Is a SELECT-list alias of the same block visible in {@code clause}, for
     * {@code vendor}?
     *
     * <p>Deliberately a per-vendor AND per-clause decision, and deliberately NOT
     * derived from {@link TSQLEnv#isAliasReferenceForbidden}. That map is a single
     * whole-dialect flag answering "does this dialect ever allow an alias reference
     * outside ORDER BY", which is a coarser question: Firebird and SAP HANA are
     * {@code false} there, yet neither documents a SELECT-list alias as visible in
     * WHERE. Treating that flag as a per-clause permission would rebind real table
     * columns in those dialects, so per CLAUDE.md's "never apply one vendor's rule to
     * all" this allow-lists only dialects whose documentation states the alias is
     * referenceable:</p>
     * <ul>
     *   <li><b>Snowflake</b> — aliases are referenceable in WHERE / GROUP BY /
     *       HAVING / QUALIFY (a real column of the same name still wins, which the
     *       authority check in {@link #isCurrentResolutionAuthoritative} enforces);</li>
     *   <li><b>Vertica</b> — an alias "can be referenced elsewhere in the SELECT
     *       statement, for example in the query predicate or ORDER BY clause";</li>
     *   <li><b>Teradata</b> — already relied upon by
     *       {@link #handleTeradataNamedAliasResolution}, whose whole purpose is a
     *       WHERE-clause alias reference.</li>
     *   <li><b>SAP IQ, SQL Anywhere</b> and {@code dbvsybase} — see
     *       {@link #isAliasOccludingColumn}: the alias is visible anywhere in its SELECT
     *       block, the select list included, and it wins over a column of the same
     *       name.</li>
     * </ul>
     *
     * <p>The dialects of {@link TSQLEnv#isAliasVisibleInHaving} (MySQL, BigQuery, Hive,
     * Spark SQL, Databricks) expose it in HAVING only.</p>
     *
     * <p>Every other dialect returns false and keeps its previous behavior. Adding
     * one means citing its documentation and pinning it with a test in BOTH
     * directions, as {@code testMantis4659} does.</p>
     *
     * <p>{@code joinCondition}/{@code join} are excluded for ALL vendors: they are
     * part of the FROM clause, evaluated before any alias exists.</p>
     */
    private static boolean isClauseExposingSelectListAlias(EDbVendor vendor, ESqlClause clause, TGSqlParser parser) {
        if (vendor == null || clause == null) return false;
        if (clause == ESqlClause.having && TSQLEnv.isAliasVisibleInHaving(vendor, parser)) return true;
        boolean occluding = isAliasOccludingColumn(vendor);
        switch (vendor) {
            case dbvsnowflake:
            case dbvvertica:
            case dbvteradata:
                break;
            default:
                if (!occluding) return false;
                break;
        }
        switch (clause) {
            case selectList:
                // A later select-list item reusing an alias. The dialects above reach the alias there
                // through ScopeBuilder's lateral column alias path instead.
                return occluding;
            case where:
            case having:
            case groupby:
            case orderby:
            case qualify:
                return true;
            default:
                return false;
        }
    }

    /**
     * Does the namespace this column currently resolves to actually PROVE that it
     * exposes a column of this name?
     *
     * <p>Used by {@link #handleSelectListAliasResolution} to decide whether an
     * existing table link is worth keeping. Returns false whenever the namespace
     * cannot decide — a bare table with no catalog metadata, or a CTE answering
     * MAYBE through its base-table fallback — because such a link was inferred,
     * never proven.</p>
     *
     * <p>An {@code AMBIGUOUS} result is checked candidate by candidate. Its
     * {@code getColumnSource()} is null, so looking only there would read
     * "not authoritative" and let the alias overwrite a genuine ambiguity between
     * two PROVEN real columns.</p>
     */
    private boolean isCurrentResolutionAuthoritative(TObjectName objName, String columnName) {
        ResolutionResult current = objName.getResolution();
        if (current == null) return false;

        if (current.getStatus() == ResolutionStatus.AMBIGUOUS) {
            AmbiguousColumnSource ambiguous = current.getAmbiguousSource();
            if (ambiguous == null || ambiguous.getCandidates() == null) return false;
            for (ColumnSource candidate : ambiguous.getCandidates()) {
                if (isProvenColumnSource(candidate, columnName)) return true;
            }
            return false;
        }

        return isProvenColumnSource(current.getColumnSource(), columnName);
    }

    /**
     * Does the alias's own defining expression reference a real column of the same
     * name?
     *
     * <p>{@code SELECT CAST(id AS INT) AS id FROM t1 WHERE id > 1} — the {@code id}
     * inside the CAST is positioned before the alias token, so it cannot be a
     * reference to the alias being defined; the resolver has bound it to {@code t1}.
     * That is proof, taken from the query text alone, that {@code t1} really has an
     * {@code id} column, so the WHERE reference must stay on the table under the
     * "real column outranks a same-named alias" rule — no catalog required.</p>
     *
     * <p>Containment is decided by token span, the same way
     * {@link #isColumnWithinResultColumn} does it, so no expression tree is walked
     * recursively.</p>
     *
     * <p>The inner reference must be able to read one of the SAME tables the WHERE
     * reference could be reading. Span containment alone is not enough:
     * in {@code SELECT (SELECT max(id) FROM t2) AS id FROM t1 WHERE id > 1} the inner
     * {@code id} sits inside the alias expression but belongs to {@code t2}, and says
     * nothing about whether {@code t1} has an {@code id}. Requiring the same table
     * keeps a nested subquery, CASE arm or OVER clause over a DIFFERENT table from
     * being read as evidence about this one.</p>
     */
    private boolean aliasDefinitionReferencesSameColumn(TResultColumn resultCol,
                                                        String columnName,
                                                        EDbVendor vendor,
                                                        TObjectName reference) {
        if (resultCol == null || columnName == null || reference == null) return false;
        TExpression expr = resultCol.getExpr();
        if (expr == null || expr.getStartToken() == null || expr.getEndToken() == null) return false;

        List<TTable> referenceTables = possibleSourceTables(reference);
        if (referenceTables.isEmpty()) return false;

        long exprStart = expr.getStartToken().posinlist;
        long exprEnd = expr.getEndToken().posinlist;

        for (TObjectName candidate : columnToScopeMap.keySet()) {
            if (candidate == null || candidate.getStartToken() == null) continue;
            long pos = candidate.getStartToken().posinlist;
            if (pos < exprStart || pos > exprEnd) continue;
            if (!sameName(vendor, ESQLDataObjectType.dotColumn,
                    candidate.getColumnNameOnly(), columnName)) {
                continue;
            }
            if (sharesAnyTable(referenceTables, possibleSourceTables(candidate))) return true;
        }
        return false;
    }

    /**
     * The tables a reference could be reading, as an identity list: the resolved one
     * if it has it, otherwise its candidates.
     *
     * <p>An unqualified name over several metadata-free tables resolves AMBIGUOUS with
     * {@code sourceTable == null} and the tables in {@code candidateTables}. Reading
     * only {@code sourceTable} would treat that as "no information" and let the alias
     * overwrite a genuine ambiguity between real columns.</p>
     */
    private static List<TTable> possibleSourceTables(TObjectName reference) {
        List<TTable> tables = new ArrayList<TTable>();
        if (reference == null) return tables;
        if (reference.getSourceTable() != null) {
            tables.add(reference.getSourceTable());
            return tables;
        }
        TTableList candidates = reference.getCandidateTables();
        for (int i = 0; candidates != null && i < candidates.size(); i++) {
            TTable candidate = candidates.getTable(i);
            if (candidate != null) tables.add(candidate);
        }
        return tables;
    }

    /** Identity intersection — {@link TTable} does not override equals. */
    private static boolean sharesAnyTable(List<TTable> left, List<TTable> right) {
        for (TTable l : left) {
            for (TTable r : right) {
                if (l == r) return true;
            }
        }
        return false;
    }

    /**
     * Can this {@link ColumnSource} be taken as PROOF that a real column of that
     * name exists?
     *
     * <p>Two independent ways to qualify, because neither alone covers the field:</p>
     * <ul>
     *   <li>the namespace vouches for its output schema
     *       ({@link BindingMetadataAuthority#lookupOutputSchema}) — catalog metadata,
     *       an explicit CTE column list, a subquery projection;</li>
     *   <li>or the source itself was recorded at full confidence. Namespaces with a
     *       FIXED output schema — {@code ValuesNamespace} ({@code v(c)}),
     *       {@code UnnestNamespace}, {@code PivotNamespace} — know their columns
     *       exactly but never override {@code getMetadataState()}, so the first test
     *       reports METADATA_UNAVAILABLE for them and would strip a real
     *       {@code values_alias_column} binding. Every definite source in the
     *       resolver is built at confidence 1.0 and every inferred one below it
     *       ({@code inferred_from_usage} 0.8, for example), so full confidence is
     *       the codebase's own marker for "proven".</li>
     * </ul>
     */
    private boolean isProvenColumnSource(ColumnSource source, String columnName) {
        if (source == null) return false;
        INamespace namespace = source.getSourceNamespace();
        if (namespace == null) return false;  // already alias-shaped, nothing to protect
        if (BindingMetadataAuthority.lookupOutputSchema(namespace, columnName)
                == ColumnAuthority.AUTHORITATIVE_PRESENT) {
            return true;
        }
        return source.getConfidence() >= 1.0;
    }

    /**
     * Check if a column reference is within a QUALIFY clause.
     *
     * @param objName The column reference to check
     * @return true if the column is within a QUALIFY clause
     */
    private boolean isInQualifyClause(TObjectName objName) {
        if (objName == null) return false;

        // Get the column's scope to find the enclosing SELECT statement
        IScope scope = columnToScopeMap.get(objName);
        if (scope == null) return false;

        TSelectSqlStatement enclosingSelect = findEnclosingSelectFromScope(scope);
        if (enclosingSelect == null) return false;

        // Check if this SELECT has a QUALIFY clause
        TQualifyClause qualifyClause = enclosingSelect.getQualifyClause();
        if (qualifyClause == null) return false;

        // Check if the column's token position is within the QUALIFY clause's range
        if (objName.getStartToken() != null && qualifyClause.getStartToken() != null &&
            qualifyClause.getEndToken() != null) {
            long objPos = objName.getStartToken().posinlist;
            long qualifyStart = qualifyClause.getStartToken().posinlist;
            long qualifyEnd = qualifyClause.getEndToken().posinlist;

            return objPos >= qualifyStart && objPos <= qualifyEnd;
        }

        return false;
    }

    /**
     * Gets or builds the Teradata NAMED alias index for a SELECT statement.
     * Optimization C: Caches the alias map for O(1) lookup instead of O(N) iteration.
     *
     * @param selectStmt The SELECT statement to get/build the index for
     * @return Map from lowercase alias name to TResultColumn, or null if no aliases
     */
    private Map<String, TResultColumn> getTeradataNamedAliasIndex(TSelectSqlStatement selectStmt) {
        if (selectStmt == null) return null;

        // Check cache first
        Map<String, TResultColumn> index = teradataNamedAliasCache.get(selectStmt);
        if (index != null) {
            return index;
        }

        // Build index for this SELECT statement
        TResultColumnList resultColumns = selectStmt.getResultColumnList();
        if (resultColumns == null || resultColumns.size() == 0) {
            // Cache empty map to avoid rebuilding
            index = java.util.Collections.emptyMap();
            teradataNamedAliasCache.put(selectStmt, index);
            return index;
        }

        index = new java.util.LinkedHashMap<>();
        // S2: store keys via the vendor-aware identifier normalizer so the
        // index honors quoted-vs-unquoted distinctions on Teradata.
        //
        // CRITICAL — vendor source-of-truth invariant (codex round-1+2):
        // we read {@code sqlStatements.get(0).dbvendor} here, which is the
        // SAME source the lookup at
        // {@link #handleTeradataNamedAliasResolution} reads (line 916,
        // also gated on {@code dbVendor == EDbVendor.dbvteradata}). Storage
        // and lookup share this single source so the index keys built here
        // and the lookup keys produced there are guaranteed to use
        // identical vendor rules. The fall-through to
        // {@code selectStmt.dbvendor} is purely defensive and only fires if
        // the resolver's statement list is unexpectedly empty.
        EDbVendor indexVendor = (sqlStatements != null && sqlStatements.size() > 0
                && sqlStatements.get(0).dbvendor != null)
                ? sqlStatements.get(0).dbvendor
                : (selectStmt.dbvendor != null ? selectStmt.dbvendor : EDbVendor.dbvgeneric);
        for (int i = 0; i < resultColumns.size(); i++) {
            TResultColumn resultCol = resultColumns.getResultColumn(i);
            if (resultCol == null) continue;

            // Check if this result column has a NAMED alias
            if (resultCol.getAliasClause() != null &&
                resultCol.getAliasClause().getAliasName() != null) {
                String aliasName = resultCol.getAliasClause().getAliasName().toString();
                if (aliasName != null && !aliasName.isEmpty()) {
                    String key = normalizeName(
                            indexVendor,
                            gudusoft.gsqlparser.sqlenv.ESQLDataObjectType.dotColumn,
                            aliasName);
                    index.put(key, resultCol);
                }
            }
        }

        // Cache the index (even if empty, to avoid rebuilding)
        teradataNamedAliasCache.put(selectStmt, index);

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE && !index.isEmpty()) {
            logDebug("Built Teradata NAMED alias index for SELECT with " + index.size() + " aliases");
        }

        return index;
    }

    /**
     * Check if a column reference (TObjectName) is within a result column's expression.
     * This is used to prevent treating source columns in expressions like "CAST(ID AS DECIMAL) AS ID"
     * as references to the alias.
     *
     * @param objName The column reference to check
     * @param resultCol The result column to check against
     * @return true if objName is within resultCol's expression tree
     */
    private boolean isColumnWithinResultColumn(TObjectName objName, TResultColumn resultCol) {
        if (objName == null || resultCol == null) return false;

        // Get the expression of the result column
        TExpression expr = resultCol.getExpr();
        if (expr == null) return false;

        // Check by comparing start/end positions
        // If objName's position is within resultCol's expression, it's part of it
        long objStart = objName.getStartToken() != null ? objName.getStartToken().posinlist : -1;
        long objEnd = objName.getEndToken() != null ? objName.getEndToken().posinlist : -1;
        long exprStart = expr.getStartToken() != null ? expr.getStartToken().posinlist : -1;
        long exprEnd = expr.getEndToken() != null ? expr.getEndToken().posinlist : -1;

        if (objStart >= 0 && exprStart >= 0 && objEnd >= 0 && exprEnd >= 0) {
            return objStart >= exprStart && objEnd <= exprEnd;
        }

        return false;
    }

    /**
     * Handle subquery aliased/calculated column resolution.
     *
     * <p>When a column reference resolves through a subquery (or CTE containing subqueries),
     * and the underlying column is an alias or calculated expression, we should NOT trace
     * it to the base table. This method ensures that such columns have their sourceTable
     * cleared to prevent incorrect attribution.</p>
     *
     * <p>This is essential for queries like:</p>
     * <pre>
     * WITH DataCTE AS (
     *     SELECT t.col, COUNT(*) AS cnt FROM table1 t ...
     * )
     * SELECT * FROM DataCTE
     * </pre>
     * <p>The 'cnt' column should NOT be traced to 'table1' because it's a calculated column.</p>
     *
     * @param objName The column reference to check
     */
    private void handleSubqueryAliasedColumnResolution(TObjectName objName) {
        if (objName == null) return;

        // Check if column has a table qualifier pointing to a subquery/CTE
        // If so, we should KEEP the sourceTable link for lineage tracing
        // The qualifier explicitly tells us which subquery the column belongs to
        String tableQualifier = objName.getTableString();
        if (tableQualifier != null && !tableQualifier.isEmpty()) {
            IScope scope = columnToScopeMap.get(objName);
            if (scope != null) {
                TTable qualifiedTable = findTableByQualifier(scope, tableQualifier);
                if (qualifiedTable != null &&
                    (qualifiedTable.getSubquery() != null || qualifiedTable.getCTE() != null)) {
                    // Column has qualifier pointing to a subquery/CTE
                    // Keep the sourceTable link for lineage tracing (e.g., a.num_emp -> subquery a)
                    // Don't clear sourceTable - this link is correct and needed
                    logDebug("Subquery/CTE qualified column: " + objName.toString() +
                        " - keeping sourceTable link to " + tableQualifier);
                    return;
                }
            }
        }

        // For unqualified columns (or columns qualified with base tables),
        // check if this is a calculated column or alias that should not trace to base tables
        ColumnSource source = objName.getColumnSource();
        if (source != null) {
            if (source.isCalculatedColumn() || source.isColumnAlias()) {
                TTable currentSource = objName.getSourceTable();
                if (currentSource != null) {
                    // Only clear if sourceTable is a base table (not subquery/CTE)
                    // For subquery/CTE references, keep the link for lineage tracing
                    if (currentSource.getSubquery() == null && currentSource.getCTE() == null) {
                        objName.setSourceTable(null);
                        logDebug("Calculated/alias column: " + objName.getColumnNameOnly() +
                            " cleared sourceTable (was " + currentSource.getName() + ") - not linked to base table");
                    }
                }
            }
        }
    }

    /**
     * Gets or builds the FromScopeIndex for a scope (Performance Optimization B).
     *
     * <p>This method implements lazy initialization: the index is built on first access
     * and cached for subsequent lookups within the same resolution pass.</p>
     *
     * @param scope The scope to get the index for (SelectScope, UpdateScope, or FromScope)
     * @return The cached or newly built FromScopeIndex, or null if scope has no FROM clause
     */
    private FromScopeIndex getFromScopeIndex(IScope scope) {
        if (scope == null) {
            return null;
        }

        // Get the actual FromScope to use as cache key
        IScope fromScope = null;
        if (scope instanceof SelectScope) {
            fromScope = ((SelectScope) scope).getFromScope();
        } else if (scope instanceof gudusoft.gsqlparser.resolver2.scope.UpdateScope) {
            fromScope = ((gudusoft.gsqlparser.resolver2.scope.UpdateScope) scope).getFromScope();
        } else if (scope instanceof FromScope) {
            fromScope = scope;
        }

        if (fromScope == null) {
            return null;
        }

        // Check cache first (lazy initialization)
        FromScopeIndex index = fromScopeIndexCache.get(fromScope);
        if (index == null) {
            // S2: thread the GlobalScope's matcher into the index so per-vendor
            // identifier rules govern alias / table-name lookups (BigQuery
            // tables sensitive, Oracle / Postgres quoted sensitive, etc.).
            gudusoft.gsqlparser.resolver2.matcher.INameMatcher matcher =
                globalScope != null ? globalScope.getNameMatcher() : null;
            // Build index and cache it
            index = new FromScopeIndex(fromScope.getChildren(), matcher);
            fromScopeIndexCache.put(fromScope, index);

            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                logDebug("Built FromScopeIndex for scope: " + index);
            }
        }

        return index;
    }

    /**
     * Find a table by its qualifier (alias or name) in the scope.
     * Uses FromScopeIndex for O(1) lookup instead of O(N) linear scan.
     */
    private TTable findTableByQualifier(IScope scope, String qualifier) {
        if (scope == null || qualifier == null) return null;

        // Use indexed lookup (Performance Optimization B)
        FromScopeIndex index = getFromScopeIndex(scope);
        if (index != null) {
            return index.findTableByQualifier(qualifier);
        }

        return null;
    }

    /**
     * Check if a column name is an alias (not a passthrough column) in the subquery.
     */
    private boolean isColumnAnAliasInSubquery(TSelectSqlStatement subquery, String columnName) {
        if (subquery == null || columnName == null) return false;

        TResultColumnList resultCols = subquery.getResultColumnList();
        if (resultCols == null) return false;

        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;

        for (int i = 0; i < resultCols.size(); i++) {
            TResultColumn rc = resultCols.getResultColumn(i);
            if (rc == null) continue;

            // Check if this result column has an alias matching the column name
            if (rc.getAliasClause() != null && rc.getAliasClause().getAliasName() != null) {
                String alias = rc.getAliasClause().getAliasName().toString();
                if (alias != null && sameName(dbVendor, ESQLDataObjectType.dotColumn, alias, columnName)) {
                    // Found matching alias - check if it's a calculated column
                    TExpression expr = rc.getExpr();
                    if (expr != null) {
                        // Not a simple column reference = calculated
                        if (expr.getExpressionType() != EExpressionType.simple_object_name_t) {
                            return true;
                        }
                    }
                }
            }

            // Also check for SQL Server proprietary alias syntax: alias = expr
            // In this case, the alias is the column name itself
            String colName = getResultColumnName(rc);
            if (colName != null && sameName(dbVendor, ESQLDataObjectType.dotColumn, colName, columnName)) {
                TExpression expr = rc.getExpr();
                if (expr != null && expr.getExpressionType() != EExpressionType.simple_object_name_t) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Get the column name from a result column (handles aliases and SQL Server proprietary syntax).
     */
    private String getResultColumnName(TResultColumn rc) {
        if (rc == null) return null;

        // Check for explicit alias
        if (rc.getAliasClause() != null && rc.getAliasClause().getAliasName() != null) {
            return rc.getAliasClause().getAliasName().toString();
        }

        // Check for SQL Server proprietary alias: alias = expr
        // In this case, the expression itself contains the alias
        TExpression expr = rc.getExpr();
        if (expr != null && expr.getExpressionType() == EExpressionType.assignment_t) {
            // The left side is the alias
            if (expr.getLeftOperand() != null && expr.getLeftOperand().getObjectOperand() != null) {
                return expr.getLeftOperand().getObjectOperand().toString();
            }
        }

        return null;
    }

    /**
     * Find the enclosing SELECT statement from a scope.
     * Traverses up the scope hierarchy to find a SelectScope and gets its node.
     *
     * @param scope The scope to start from
     * @return The enclosing SELECT statement, or null if not found
     */
    private TSelectSqlStatement findEnclosingSelectFromScope(IScope scope) {
        if (scope == null) return null;

        IScope currentScope = scope;
        int maxIterations = 100; // Prevent infinite loops
        int iterations = 0;

        while (currentScope != null && iterations < maxIterations) {
            iterations++;

            // Check if current scope is a SelectScope
            if (currentScope instanceof SelectScope) {
                TParseTreeNode node = currentScope.getNode();
                if (node instanceof TSelectSqlStatement) {
                    return (TSelectSqlStatement) node;
                }
            }

            // Move up to parent scope
            currentScope = currentScope.getParent();
        }
        return null;
    }

    /**
     * Collect a column reference for namespace enhancement if it targets a star namespace.
     * This is called during resolution to gather columns that need to be added to namespaces.
     *
     * @param objName The column reference
     * @param scope The scope where the column should be resolved
     */
    private void collectForEnhancementIfNeeded(TObjectName objName, IScope scope) {
        if (objName == null || scope == null) return;

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

        // Get the resolution result to check status
        gudusoft.gsqlparser.resolver2.model.ResolutionResult result = objName.getResolution();

        // Find candidate namespace from scope's FROM clause
        INamespace candidateNamespace = findCandidateNamespace(objName, scope);

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("[TSQLResolver2] collectForEnhancement: column=" + columnName +
                    ", candidateNs=" + (candidateNamespace != null ? candidateNamespace.getDisplayName() : "null") +
                    ", hasStar=" + (candidateNamespace != null ? candidateNamespace.hasStarColumn() : "N/A"));
        }

        if (candidateNamespace != null) {
            // Determine confidence based on context
            double confidence = 0.7; // Default for unqualified reference
            String evidence = "outer_reference";

            // Higher confidence for qualified references (e.g., "a.column")
            if (objName.getTableToken() != null) {
                confidence = 0.9;
                evidence = "qualified_reference";
            }

            // Collect for enhancement
            namespaceEnhancer.collectColumnRef(
                columnName,
                candidateNamespace,
                objName,
                confidence,
                evidence
            );
        }
    }

    /**
     * Find the candidate namespace for a column reference.
     * Looks at the scope's FROM clause to find namespaces with star columns.
     * Uses FromScopeIndex for O(1) lookup instead of O(N) linear scan.
     */
    private INamespace findCandidateNamespace(TObjectName objName, IScope scope) {
        // Use indexed lookup (Performance Optimization B)
        FromScopeIndex index = getFromScopeIndex(scope);
        if (index == null) {
            return null;
        }

        String tablePrefix = objName.getTableToken() != null ?
            objName.getTableToken().toString() : null;

        return index.findCandidateNamespace(tablePrefix);
    }

    /**
     * Build the global scope
     */
    private void buildGlobalScope() {
        logDebug("Building global scope");

        // Get SQLEnv and vendor for qualified name resolution
        TSQLEnv sqlEnv = globalContext != null ? globalContext.getSqlEnv() : null;
        EDbVendor vendor = EDbVendor.dbvoracle; // Default

        // Try to get vendor from statements
        if (sqlStatements != null && sqlStatements.size() > 0) {
            vendor = sqlStatements.get(0).dbvendor;
        }

        // Create global scope with sqlEnv and vendor for proper qualified name resolution
        globalScope = new GlobalScope(globalContext, config.getNameMatcher(), sqlEnv, vendor);

        logDebug("GlobalScope created with defaults: catalog=" +
                 globalScope.getDefaultCatalog() + ", schema=" + globalScope.getDefaultSchema());
    }

    /**
     * Process a single statement
     */
    private void processStatement(Object statement) {
        if (statement instanceof TSelectSqlStatement) {
            processSelectStatement((TSelectSqlStatement) statement);
        }
        // TODO: Add support for INSERT, UPDATE, DELETE, etc.
    }

    /**
     * Process a SELECT statement
     */
    private void processSelectStatement(TSelectSqlStatement select) {
        processSelectStatement(select, globalScope);
    }

    /**
     * Process a SELECT statement with a specific parent scope.
     * This is used for recursive processing of CTE subqueries.
     */
    private void processSelectStatement(TSelectSqlStatement select, IScope givenParentScope) {
        logDebug("Processing SELECT statement");

        // Create SELECT scope (will be child of CTE scope if CTEs exist, otherwise child of given parent scope)
        IScope parentScope = givenParentScope;

        // Process CTEs (WITH clause) if present
        CTEScope cteScope = null;
        if (select.getCteList() != null && select.getCteList().size() > 0) {
            cteScope = processCTEs(select.getCteList(), givenParentScope);
            parentScope = cteScope;  // CTEs become parent of SELECT
        }

        SelectScope selectScope = new SelectScope(parentScope, select);

        // Process FROM clause
        if (select.tables != null && select.tables.size() > 0) {
            FromScope fromScope = processFromClause(select, selectScope);
            selectScope.setFromScope(fromScope);
        }

        // Process column references in SELECT list
        if (select.getResultColumnList() != null) {
            List<TObjectName> selectListColumns = collectObjectNamesFromResultColumns(select.getResultColumnList());
            processColumnReferences(selectListColumns, selectScope);
        }

        // Process WHERE clause
        if (select.getWhereClause() != null &&
            select.getWhereClause().getCondition() != null) {
            List<TObjectName> whereColumns = select.getWhereClause().getCondition().getColumnsInsideExpression();
            processColumnReferences(whereColumns, selectScope);
        }

        // Process GROUP BY clause
        GroupByScope groupByScope = null;
        if (select.getGroupByClause() != null) {
            groupByScope = processGroupBy(select, selectScope);
        }

        // Process HAVING clause
        if (select.getGroupByClause() != null &&
            select.getGroupByClause().getHavingClause() != null) {
            processHaving(select, selectScope, groupByScope);
        }

        // Process ORDER BY clause
        if (select.getOrderbyClause() != null) {
            processOrderBy(select, selectScope);
        }
    }

    /**
     * Process FROM clause and build FROM scope
     */
    private FromScope processFromClause(TSelectSqlStatement select, IScope parentScope) {
        FromScope fromScope = new FromScope(parentScope, select.tables);

        // Process each relation (table or join)
        ArrayList<TTable> relations = select.getRelations();
        if (relations != null) {
            for (TTable table : relations) {
                processTableOrJoin(table, fromScope);
            }
        }

        return fromScope;
    }

    /**
     * Recursively process a table or join expression and add to FROM scope
     */
    private void processTableOrJoin(TTable table, FromScope fromScope) {
        if (table.getTableType() == ETableSource.join) {
            // This is a JOIN - recursively process left and right tables
            TJoinExpr joinExpr = table.getJoinExpr();
            if (joinExpr != null) {
                logDebug("Processing JOIN: " + joinExpr.getJointype());

                // Recursively process left table
                TTable leftTable = joinExpr.getLeftTable();
                if (leftTable != null) {
                    processTableOrJoin(leftTable, fromScope);
                }

                // Recursively process right table
                TTable rightTable = joinExpr.getRightTable();
                if (rightTable != null) {
                    processTableOrJoin(rightTable, fromScope);
                }

                // TODO: Create JoinScope to handle nullable semantics
                // For now, we just add the base tables to FROM scope
            }
        } else {
            // This is a base table (objectname, subquery, etc.)
            INamespace namespace = createNamespaceForTable(table);

            // Validate namespace (load metadata)
            namespace.validate();

            // Determine alias
            String alias = table.getAliasName() != null
                ? table.getAliasName()
                : table.getName();

            // Add to FROM scope
            fromScope.addChild(namespace, alias, false);

            logDebug("Added table to FROM scope: " + alias);
        }
    }

    /**
     * Process CTEs (WITH clause) and build CTE scope
     */
    private CTEScope processCTEs(TCTEList cteList, IScope parentScope) {
        CTEScope cteScope = new CTEScope(parentScope, cteList);
        logDebug("Processing WITH clause with " + cteList.size() + " CTE(s)");

        // Process each CTE in order (later CTEs can reference earlier ones)
        for (int i = 0; i < cteList.size(); i++) {
            TCTE cte = cteList.getCTE(i);

            // Get CTE name
            String cteName = cte.getTableName() != null ? cte.getTableName().toString() : null;
            if (cteName == null) {
                logDebug("Skipping CTE with null name");
                continue;
            }

            // Get CTE subquery
            TSelectSqlStatement cteSubquery = cte.getSubquery();
            if (cteSubquery == null) {
                logDebug("Skipping CTE '" + cteName + "' with null subquery");
                continue;
            }

            // Create CTENamespace
            CTENamespace cteNamespace = new CTENamespace(
                cte,
                cteName,
                cteSubquery,
                config.getNameMatcher()
            );

            // Validate namespace (load column metadata from subquery)
            cteNamespace.validate();

            // Add to CTE scope (makes it visible to later CTEs and main query)
            cteScope.addCTE(cteName, cteNamespace);

            logDebug("Added CTE to scope: " + cteName +
                    " (columns=" + cteNamespace.getExplicitColumns().size() +
                    ", recursive=" + cteNamespace.isRecursive() + ")");

            // Recursively process CTE subquery
            // This ensures that:
            // 1. Columns within the CTE are properly resolved
            // 2. Nested CTEs within this CTE are handled
            // 3. Later CTEs can reference this CTE's columns
            logDebug("Recursively processing CTE subquery: " + cteName);
            processSelectStatement(cteSubquery, cteScope);
        }

        return cteScope;
    }

    /**
     * Process GROUP BY clause and build GROUP BY scope
     */
    private GroupByScope processGroupBy(TSelectSqlStatement select, SelectScope selectScope) {
        GroupByScope groupByScope = new GroupByScope(selectScope, select.getGroupByClause());
        logDebug("Processing GROUP BY clause");

        // Set the FROM scope for column resolution
        if (selectScope.getFromScope() != null) {
            groupByScope.setFromScope(selectScope.getFromScope());
        }

        // Process column references in GROUP BY items
        if (select.getGroupByClause().getItems() != null) {
            for (int i = 0; i < select.getGroupByClause().getItems().size(); i++) {
                gudusoft.gsqlparser.nodes.TGroupByItem item = select.getGroupByClause().getItems().getGroupByItem(i);
                if (item.getExpr() != null) {
                    List<TObjectName> groupByColumns = item.getExpr().getColumnsInsideExpression();
                    processColumnReferences(groupByColumns, groupByScope);
                }
            }
        }

        return groupByScope;
    }

    /**
     * Process HAVING clause and build HAVING scope
     */
    private void processHaving(TSelectSqlStatement select, SelectScope selectScope, GroupByScope groupByScope) {
        logDebug("Processing HAVING clause");

        HavingScope havingScope = new HavingScope(
            selectScope,
            select.getGroupByClause().getHavingClause()
        );

        // Set GROUP BY scope for grouped column resolution
        if (groupByScope != null) {
            havingScope.setGroupByScope(groupByScope);
        }

        // Set SELECT scope for alias resolution
        havingScope.setSelectScope(selectScope);

        // Process column references in HAVING condition
        List<TObjectName> havingColumns = select.getGroupByClause().getHavingClause().getColumnsInsideExpression();
        processColumnReferences(havingColumns, havingScope);
    }

    /**
     * Process ORDER BY clause and build ORDER BY scope
     */
    private void processOrderBy(TSelectSqlStatement select, SelectScope selectScope) {
        logDebug("Processing ORDER BY clause");

        OrderByScope orderByScope = new OrderByScope(selectScope, select.getOrderbyClause());

        // Set SELECT scope for alias resolution
        orderByScope.setSelectScope(selectScope);

        // Set FROM scope for direct column resolution (database-dependent)
        if (selectScope.getFromScope() != null) {
            orderByScope.setFromScope(selectScope.getFromScope());
        }

        // Process column references in ORDER BY items
        if (select.getOrderbyClause().getItems() != null) {
            for (int i = 0; i < select.getOrderbyClause().getItems().size(); i++) {
                gudusoft.gsqlparser.nodes.TOrderByItem item = select.getOrderbyClause().getItems().getOrderByItem(i);
                if (item.getSortKey() != null) {
                    List<TObjectName> orderByColumns = item.getSortKey().getColumnsInsideExpression();
                    processColumnReferences(orderByColumns, orderByScope);
                }
            }
        }
    }

    /**
     * Create appropriate namespace for a table
     */
    private INamespace createNamespaceForTable(TTable table) {
        // Check if it's a subquery
        if (table.getSubquery() != null) {
            SubqueryNamespace namespace = new SubqueryNamespace(
                table.getSubquery(), table.getAliasName(), config.getNameMatcher());
            namespace.setSqlEnv(globalContext != null ? globalContext.getSqlEnv() : null);
            namespace.setStatementMetadata(statementMetadata);
            return namespace;
        }

        // Regular table - pass sqlEnv and vendor for qualified name resolution
        TSQLEnv sqlEnv = globalContext != null ? globalContext.getSqlEnv() : null;
        EDbVendor vendor = table.dbvendor != null ? table.dbvendor : EDbVendor.dbvoracle;
        TableNamespace namespace = new TableNamespace(table, config.getNameMatcher(), sqlEnv, vendor);
        if (statementMetadata != null) statementMetadata.configure(namespace, table);
        return namespace;
    }

    /**
     * Collect all TObjectName from TResultColumnList
     */
    private List<TObjectName> collectObjectNamesFromResultColumns(
            gudusoft.gsqlparser.nodes.TResultColumnList resultColumns) {
        List<TObjectName> objNames = new ArrayList<>();

        for (int i = 0; i < resultColumns.size(); i++) {
            gudusoft.gsqlparser.nodes.TResultColumn rc = resultColumns.getResultColumn(i);
            if (rc.getExpr() != null) {
                // Get all column references from the expression
                List<TObjectName> exprColumns = rc.getExpr().getColumnsInsideExpression();
                if (exprColumns != null) {
                    objNames.addAll(exprColumns);
                }
            }
        }

        return objNames;
    }

    /**
     * Process column references (TObjectName list)
     */
    private void processColumnReferences(List<TObjectName> objectNames, IScope scope) {
        if (objectNames == null) return;

        for (TObjectName objName : objectNames) {
            // Record column-to-scope mapping for iterative resolution (Principle 1)
            columnToScopeMap.put(objName, scope);
            allColumnReferences.add(objName);

            // Resolve the column reference
            nameResolver.resolve(objName, scope);

            // Handle USING column priority for JOIN...USING syntax
            handleUsingColumnResolution(objName);

            // Handle Teradata NAMED alias resolution
            handleTeradataNamedAliasResolution(objName);
            handleQualifyClauseAliasResolution(objName);
            handleSelectListAliasResolution(objName);
        }
    }

    // Detailed legacy sync timing (for profiling)
    private static long globalTimeClearLinked = 0;
    private static long globalTimeFillAttributes = 0;
    private static long globalTimeSyncColumns = 0;
    private static long globalTimePopulateOrphans = 0;
    private static long globalTimeClearHints = 0;

    /**
     * Get detailed legacy sync timing breakdown.
     */
    public static String getLegacySyncTimings() {
        long total = globalTimeClearLinked + globalTimeFillAttributes + globalTimeSyncColumns + globalTimePopulateOrphans + globalTimeClearHints;
        return String.format(
            "LegacySync Breakdown:\n" +
            "  ClearLinkedColumns: %d ms (%.1f%%)\n" +
            "  FillTableAttributes: %d ms (%.1f%%)\n" +
            "  SyncColumnToLegacy: %d ms (%.1f%%)\n" +
            "  PopulateOrphanColumns: %d ms (%.1f%%)\n" +
            "  ClearSyntaxHints: %d ms (%.1f%%)\n" +
            "  Total: %d ms",
            globalTimeClearLinked, total > 0 ? 100.0 * globalTimeClearLinked / total : 0,
            globalTimeFillAttributes, total > 0 ? 100.0 * globalTimeFillAttributes / total : 0,
            globalTimeSyncColumns, total > 0 ? 100.0 * globalTimeSyncColumns / total : 0,
            globalTimePopulateOrphans, total > 0 ? 100.0 * globalTimePopulateOrphans / total : 0,
            globalTimeClearHints, total > 0 ? 100.0 * globalTimeClearHints / total : 0,
            total);
    }

    /**
     * Create cloned columns for star column tracing.
     *
     * <p>This is a CORE part of TSQLResolver2's name resolution. When a column traces
     * through a CTE or subquery with SELECT * to a physical table, we create a cloned
     * TObjectName with sourceTable pointing to the traced physical table.
     *
     * <p>Example:
     * <pre>
     * WITH cte AS (SELECT * FROM physical_table)
     * SELECT a FROM cte
     * </pre>
     *
     * <p>For column 'a' in the outer SELECT:
     * <ul>
     *   <li>Original column: sourceTable = cte (immediate source)</li>
     *   <li>Cloned column: sourceTable = physical_table (traced through star)</li>
     * </ul>
     *
     * <p>Both columns are added to allColumnReferences for complete lineage tracking.
     * This ensures the formatter can output both the immediate source and the traced
     * physical table when needed.
     */
    private void createTracedColumnClones() {
        // Collect clones to add (avoid ConcurrentModificationException)
        java.util.List<TObjectName> clonesToAdd = new java.util.ArrayList<>();

        // Mantis 4651: this method owns the clone bookkeeping consulted by
        // syncColumnToLegacy(), so start from a clean slate on every invocation.
        tracedStarClones.clear();
        tracedStarCloneKeys.clear();

        // Index of existing (sourceTable, column) pairs for O(1) dedup — table by
        // identity, column by canonical identity (Mantis 4651).
        Map<TTable, Set<CanonKey>> existingKeys = new IdentityHashMap<TTable, Set<CanonKey>>();
        for (TObjectName existing : allColumnReferences) {
            if (existing.getSourceTable() != null) {
                String existingColName = existing.getColumnNameOnly();
                if (existingColName != null) {
                    addColumnKey(existingKeys, existing.getSourceTable(), existingColName);
                }
            }
        }

        for (TObjectName column : allColumnReferences) {
            // Skip star columns - they represent all columns from a table and shouldn't be cloned
            String colName = column.getColumnNameOnly();
            if (colName != null && colName.equals("*")) {
                continue;
            }

            // Skip columns without resolution
            gudusoft.gsqlparser.resolver2.model.ResolutionResult resolution = column.getResolution();
            if (resolution == null || !resolution.isExactMatch()) {
                continue;
            }

            // STRUCT tree: a field read through the derived table or CTE, x.customer.address.city, traces to its
            // BASE column customer of the physical table, so the clone names that column; cloning the whole reference
            // invented a column named by the last field, orders_raw.city (Mantis 4830, 4831, 4818)
            TSourceToken fieldBase = column.getFieldPathRead() != null ? column.getFieldPathBaseToken() : null;
            if (fieldBase != null) {
                colName = fieldBase.toString();
            }

            gudusoft.gsqlparser.resolver2.model.ColumnSource source = resolution.getColumnSource();
            if (source == null) {
                continue;
            }
            // x.p over (SELECT payload FROM events) AS x(p): the alias list renames an explicit SELECT-list item, so
            // the reference stays at the derived table; a clone would invent events.p (Mantis 4814 follow-up)
            if (source.isRenamedByAliasList()) {
                continue;
            }

            TTable sourceTable = column.getSourceTable();
            if (sourceTable == null) {
                continue;
            }

            // Only process CTE or subquery columns
            if (!sourceTable.isCTEName() && sourceTable.getTableType() != ETableSource.subquery) {
                continue;
            }

            // Get the traced physical table
            TTable finalTable = source.getFinalTable();
            if (finalTable == null || finalTable == sourceTable) {
                continue;
            }

            // Skip if finalTable is also a CTE or subquery
            if (finalTable.isCTEName() || finalTable.getTableType() == ETableSource.subquery) {
                continue;
            }

            // Skip subquery columns when the column matches an explicit column in the subquery's
            // SELECT list. Cloning is only needed when tracing through star columns.
            // For example, in "SELECT al1.COL1, al1.COL3 FROM (SELECT t1.COL1, t2.* FROM t1, t2) al1":
            // - al1.COL1 matches explicit "t1.COL1" -> don't clone (stays at subquery level)
            // - al1.COL3 doesn't match explicit column, must come from t2.* -> clone to t2
            if (sourceTable.getTableType() == ETableSource.subquery) {
                TSelectSqlStatement subquery = sourceTable.getSubquery();
                if (subquery != null && subqueryHasExplicitColumn(subquery, colName)) {
                    continue;
                }
            }

            // Skip UNION scenarios - syncToLegacyStructures already handles linking to all
            // UNION branch tables via getAllFinalTables(). Creating clones would cause duplicates.
            java.util.List<TTable> allFinalTables = source.getAllFinalTables();
            if (allFinalTables != null && allFinalTables.size() > 1) {
                continue;
            }

            // Skip UNQUALIFIED join condition columns - they should not be traced to the source
            // subquery's underlying table via star column expansion.
            // This is particularly important for MERGE ON clause columns which may
            // belong to the target table rather than the source subquery.
            // QUALIFIED columns (like S.id) should still be traced as they explicitly reference
            // the source subquery.
            // Note: We check location only because ownStmt may be null for unresolved columns.
            if (column.getLocation() == ESqlClause.joinCondition
                    && (column.getTableString() == null || column.getTableString().isEmpty())) {
                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("createTracedColumnClones: Skipping unqualified join condition column " + column.toString() +
                        " - should not be traced to subquery's underlying table");
                }
                continue;
            }

            // O(1) dedup check using the index instead of an O(n) linear scan
            if (addColumnKey(existingKeys, finalTable, colName)) {
                // Clone the column and set sourceTable to the traced physical table
                TObjectName clonedColumn;
                if (fieldBase != null) {
                    // the base column as written, with its own token: position and quote state give lineage the
                    // column's identity and display, not a legacy fold of the text
                    EDbVendor vendor = config != null ? config.getVendor() : column.dbvendor;
                    clonedColumn = TObjectName.createObjectName(vendor, EDbObjectType.column, fieldBase.clone());
                } else {
                    clonedColumn = column.clone();
                }
                clonedColumn.setSourceTable(finalTable);
                clonesToAdd.add(clonedColumn);
                // Mantis 4651: remember the clone so the subquery fallback in
                // syncColumnToLegacy() does not link the original reference to the
                // very same physical column a second time.
                tracedStarClones.add(clonedColumn);
                addColumnKey(tracedStarCloneKeys, finalTable, colName);

                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("createTracedColumnClones: Cloned column " + column.toString() +
                        " with sourceTable traced from " + sourceTable.getTableName() +
                        " to physical table " + finalTable.getTableName());
                }
            }
        }

        // Add all clones to allColumnReferences (local copy in TSQLResolver2)
        allColumnReferences.addAll(clonesToAdd);

        // Also add to scopeBuildResult so consumers using scopeBuildResult.getAllColumnReferences()
        // (like TestGetTableColumn2 for star column expansion tests) can see the clones
        if (scopeBuildResult != null && !clonesToAdd.isEmpty()) {
            scopeBuildResult.addColumnReferences(clonesToAdd);
        }

        // S3: tag every clone as SYNTHETIC_STAR_CLONE so the binding post-pass
        // (S5+) can skip them. Clones bypass NameResolver.resolve() entirely
        // (see comment block above) — without this tag the post-pass would see
        // a TObjectName in getAllColumnReferences() with no recorded
        // ResolutionResult and no skip reason, breaking the coverage invariant.
        if (resolutionContext.isBindingTraceEnabled() && !clonesToAdd.isEmpty()) {
            for (TObjectName clone : clonesToAdd) {
                resolutionContext.recordColumnSkipReason(
                    clone,
                    gudusoft.gsqlparser.resolver2.binding.BindingSkipReason.SYNTHETIC_STAR_CLONE);
            }
        }

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE && !clonesToAdd.isEmpty()) {
            logInfo("createTracedColumnClones: Created " + clonesToAdd.size() + " traced column clones");
        }
    }

    /**
     * Sync results to legacy structures for backward compatibility.
     * This populates:
     * - TTable.linkedColumns: columns resolved to this table
     * - TObjectName.sourceTable: already set in setResolution()
     * - TObjectName.linkedColumnDef: from ColumnSource.definitionNode
     * - TObjectName.sourceColumn: from ColumnSource.definitionNode (if TResultColumn)
     */
    private void syncToLegacyStructures() {
        if (!config.isLegacyCompatibilityEnabled()) {
            logInfo("Legacy compatibility disabled, skipping sync");
            return;
        }

        for (int i = 0; i < sqlStatements.size(); i++) {
        }

        logInfo("Syncing to legacy structures...");

        long phaseStart;

        // Clear existing linkedColumns on all tables
        phaseStart = System.currentTimeMillis();
        clearAllLinkedColumns();

        // Clear existing orphanColumns on all statements
        // These will be repopulated in Phase 4b based on TSQLResolver2 resolution
        for (int i = 0; i < sqlStatements.size(); i++) {
            clearOrphanColumnsRecursive(sqlStatements.get(i));
        }
        linkedColumnIdentityCache.clear();
        globalTimeClearLinked += System.currentTimeMillis() - phaseStart;

        // Phase 1: Fill TTable.getAttributes() for all tables
        // This uses the namespace data already collected during resolution
        phaseStart = System.currentTimeMillis();
        Set<TTable> processedTables = new HashSet<>();
        for (int i = 0; i < sqlStatements.size(); i++) {
            fillTableAttributesRecursive(sqlStatements.get(i), processedTables);
        }
        globalTimeFillAttributes += System.currentTimeMillis() - phaseStart;
        logInfo("Filled attributes for " + processedTables.size() + " tables");

        // Phase 2: Iterate through all column references and sync to legacy structures
        phaseStart = System.currentTimeMillis();
        int syncCount = 0;
        for (TObjectName column : allColumnReferences) {
            if (syncColumnToLegacy(column)) {
                syncCount++;
            }
        }
        globalTimeSyncColumns += System.currentTimeMillis() - phaseStart;

        // Phase 3: Link CTAS target table columns
        // For CREATE TABLE AS SELECT, the SELECT list columns should be linked to the target table
        for (int i = 0; i < sqlStatements.size(); i++) {
            linkCTASTargetTableColumns(sqlStatements.get(i));
        }

        // Phase 4: Sync implicit database/schema from USE DATABASE/USE SCHEMA to AST
        // This enables TObjectName.getAnsiSchemaName() and getAnsiCatalogName() to work correctly
        syncImplicitDbSchemaToAST();

        // Phase 4b: Populate orphan columns
        // Columns with sourceTable=null (unresolved or ambiguous) should be added to
        // their containing statement's orphanColumns list. This enables TGetTableColumn
        // to report them as orphan columns (with linkOrphanColumnToFirstTable option).
        phaseStart = System.currentTimeMillis();
        populateOrphanColumns();
        globalTimePopulateOrphans += System.currentTimeMillis() - phaseStart;

        // Phase 4c: Expand star columns using push-down inferred columns
        // For SELECT * and SELECT table.*, expand to individual columns based on:
        // 1. Inferred columns from the namespace (via push-down algorithm)
        // 2. This enables star column expansion without TSQLEnv metadata
        phaseStart = System.currentTimeMillis();
        expandStarColumnsUsingPushDown();
        long expandTime = System.currentTimeMillis() - phaseStart;
        logInfo("Star column expansion took " + expandTime + "ms");

        // Phase 5: Clear orphan column syntax hints for resolved columns
        // The old resolver adds "sphint" (syntax hint) warnings for columns that can't be resolved.
        // TSQLResolver2 resolves these columns but doesn't clear the syntax hints.
        // This phase cleans up those hints to maintain compatibility with tests expecting no hints.
        phaseStart = System.currentTimeMillis();
        clearOrphanColumnSyntaxHints();
        globalTimeClearHints += System.currentTimeMillis() - phaseStart;

        logInfo("Legacy sync complete: " + syncCount + "/" + allColumnReferences.size() + " columns synced");
    }

    /**
     * Link SELECT list columns to CTAS target table.
     * For CREATE TABLE AS SELECT statements, the output column names (aliases)
     * should be linked to the target table. The source column references
     * remain linked to their source tables.
     *
     * NOTE: For CTAS, the parser (TCreateTableSqlStatement.doParseStatement) already
     * correctly creates and links alias columns to the target table. The source columns
     * that were incorrectly added are filtered out in clearLinkedColumnsRecursive().
     * This method now only handles cases where the parser didn't create alias columns.
     */
    private void linkCTASTargetTableColumns(TCustomSqlStatement stmt) {
        if (stmt == null) return;

        // CTAS columns are already handled by the parser (TCreateTableSqlStatement.doParseStatement)
        // and incorrectly added source columns are filtered in clearLinkedColumnsRecursive().
        // No additional processing needed here for CTAS.

        // Process nested statements (for other statement types that might need CTAS handling)
        for (int i = 0; i < stmt.getStatements().size(); i++) {
            linkCTASTargetTableColumns(stmt.getStatements().get(i));
        }
    }

    /**
     * Populate orphanColumns for unresolved columns.
     * Columns with sourceTable=null should be added to their containing statement's orphanColumns.
     * This enables TGetTableColumn to report these as "missed" columns.
     */
    private void populateOrphanColumns() {
        int addedCount = 0;
        for (TObjectName column : allColumnReferences) {
            if (column == null) continue;

            // Skip non-column types that should not be in orphan columns
            EDbObjectType dbObjectType = column.getDbObjectType();
            if (dbObjectType == EDbObjectType.column_alias    // alias clause column definitions (e.g., AS x (numbers, animals))
                || dbObjectType == EDbObjectType.variable     // stored procedure variables
                || dbObjectType == EDbObjectType.parameter    // stored procedure parameters
                || dbObjectType == EDbObjectType.cursor       // cursors
                || dbObjectType == EDbObjectType.constant     // constants
                || dbObjectType == EDbObjectType.label        // labels
            ) {
                continue;
            }

            // ClickHouse expression-CTE alias (WITH <expr> AS ident): resolved
            // as a scalar symbol in phase 1; it is neither a table column nor
            // a missed column, so it must not be reported as orphan.
            if (column.getExpressionCteRef() != null && column.getSourceTable() == null) {
                continue;
            }

            // Check resolution status directly - ambiguous columns should be added to orphanColumns
            // Note: column.getColumnSource() returns the first candidate for ambiguous columns,
            // which would cause them to be incorrectly skipped. We need to check the resolution status first.
            // IMPORTANT: This check must come BEFORE the sourceTable check because Phase 1 (linkColumnToTable)
            // might have already set sourceTable during parsing, but TSQLResolver2 correctly marked it as ambiguous.
            // NOTE: Skip star columns (*) since they are handled specially via sourceTableList
            ResolutionResult resolution = column.getResolution();
            String columnName = column.getColumnNameOnly();
            boolean isStarColumn = columnName != null && columnName.equals("*");

            if (resolution != null && resolution.getStatus() == ResolutionStatus.AMBIGUOUS && !isStarColumn) {
                // Ambiguous columns should be added to orphanColumns so they appear as "missed"
                // Clear sourceTable if it was set by Phase 1 (linkColumnToTable) so the column
                // doesn't also appear as resolved in the output
                if (column.getSourceTable() != null) {
                    if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                        logInfo("populateOrphanColumns: Clearing sourceTable for AMBIGUOUS column: " + column.toString()
                            + " at (" + column.getLineNo() + "," + column.getColumnNo() + ")"
                            + " was linked to " + column.getSourceTable().getTableName()
                            + " with " + (resolution.getAmbiguousSource() != null ?
                                resolution.getAmbiguousSource().getCandidateCount() : 0) + " candidates");
                    }
                    column.setSourceTable(null);
                }
                // Fall through to add to orphanColumns
            } else {
                // Star columns (*) should NEVER be orphan columns - they represent all columns
                // from all tables and are handled specially via sourceTableList and linked
                // to tables in syncColumnToLegacy() which runs after this phase.
                if (isStarColumn) {
                    continue;
                }

                // For non-ambiguous columns, skip if they have a sourceTable
                if (column.getSourceTable() != null) {
                    continue;
                }

                // Also skip columns that have a ColumnSource with a valid table
                ColumnSource source = column.getColumnSource();
                if (source != null) {
                    TTable finalTable = source.getFinalTable();
                    if (finalTable != null) {
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("populateOrphanColumns: Skipping column with ColumnSource: " + column.toString()
                                + " at (" + column.getLineNo() + "," + column.getColumnNo() + ")"
                                + " -> resolved to " + finalTable.getTableName());
                        }
                        continue;
                    }
                    // Also check overrideTable for derived table columns
                    TTable overrideTable = source.getOverrideTable();
                    if (overrideTable != null) {
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("populateOrphanColumns: Skipping column with ColumnSource (override): " + column.toString()
                                + " at (" + column.getLineNo() + "," + column.getColumnNo() + ")"
                                + " -> resolved to " + overrideTable.getTableName());
                        }
                        continue;
                    }
                }
            }

            // Debug: log columns being added to orphan
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                ColumnSource debugSource = column.getColumnSource();
                logInfo("populateOrphanColumns: Adding orphan column: " + column.toString()
                    + " at (" + column.getLineNo() + "," + column.getColumnNo() + ")"
                    + ", hasColumnSource=" + (debugSource != null)
                    + (debugSource != null ? ", namespace=" + (debugSource.getSourceNamespace() != null ?
                        debugSource.getSourceNamespace().getClass().getSimpleName() : "null") : ""));
            }

            // Find the containing statement for this column
            TCustomSqlStatement containingStmt = findContainingStatement(column);
            if (containingStmt != null) {
                // Set ownStmt so TSQLResolver2ResultFormatter can use getOwnStmt().getFirstPhysicalTable()
                // to link orphan columns to the first physical table (matching TGetTableColumn behavior)
                column.setOwnStmt(containingStmt);

                TObjectNameList orphanColumns = containingStmt.getOrphanColumns();
                if (orphanColumns != null && addColumnByIdentityIfAbsent(orphanColumns, column)) {
                    addedCount++;
                }
            } else {
                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("Could not find containing statement for orphan column: " + column.toString());
                }
            }
        }
        logInfo("Populated " + addedCount + " orphan columns");
    }

    /**
     * Find the statement that contains a column reference.
     * First tries to use the scope information (more reliable), then falls back to AST traversal.
     * For PL/SQL blocks, searches for the innermost DML statement that contains the column.
     */
    private TCustomSqlStatement findContainingStatement(TObjectName column) {
        // First, try to use the scope information from columnToScopeMap
        // The scope's node is typically the containing statement
        IScope scope = columnToScopeMap.get(column);
        if (scope != null) {
            TParseTreeNode scopeNode = scope.getNode();
            if (scopeNode instanceof TCustomSqlStatement) {
                TCustomSqlStatement stmt = (TCustomSqlStatement) scopeNode;
                // If the scope is a PL/SQL block or procedure, search for DML statements within it
                // that actually contain the column (by line number)
                if (isPLSQLBlockStatement(stmt)) {
                    TCustomSqlStatement dmlStmt = findDMLStatementContaining(stmt, column);
                    if (dmlStmt != null) {
                        return dmlStmt;
                    }
                }
                return stmt;
            }
        }

        // Fallback: traverse up the AST to find the nearest TCustomSqlStatement parent
        TParseTreeNode node = column;
        while (node != null) {
            if (node instanceof TCustomSqlStatement) {
                return (TCustomSqlStatement) node;
            }
            node = node.getParentObjectName();
        }

        // Last resort: search all statements for a DML statement containing the column
        TCustomSqlStatement result = null;
        if (sqlStatements.size() > 0) {
            for (int i = 0; i < sqlStatements.size(); i++) {
                TCustomSqlStatement stmt = sqlStatements.get(i);
                TCustomSqlStatement dmlStmt = findDMLStatementContaining(stmt, column);
                if (dmlStmt != null) {
                    result = dmlStmt;
                    break;
                }
            }
            if (result == null) {
                result = sqlStatements.get(0);
            }
        }
        return result;
    }

    /**
     * Check if a statement is a PL/SQL block type statement.
     */
    private boolean isPLSQLBlockStatement(TCustomSqlStatement stmt) {
        if (stmt == null) return false;
        String className = stmt.getClass().getSimpleName();
        return className.startsWith("TPlsql") || className.startsWith("TPLSql") ||
               className.contains("Block") || className.contains("Procedure") ||
               className.contains("Function") || className.contains("Package");
    }

    /**
     * DML Statement Range for efficient line-based lookup.
     * Used by the DML index cache (Performance Optimization A).
     */
    private static class DmlRange implements Comparable<DmlRange> {
        final long startLine;
        final long endLine;
        final TCustomSqlStatement stmt;

        DmlRange(TCustomSqlStatement stmt) {
            this.stmt = stmt;
            this.startLine = stmt.getStartToken() != null ? stmt.getStartToken().lineNo : -1;
            this.endLine = stmt.getEndToken() != null ? stmt.getEndToken().lineNo : -1;
        }

        boolean contains(long line) {
            return startLine >= 0 && startLine <= line && line <= endLine;
        }

        // Sort by startLine for binary search
        @Override
        public int compareTo(DmlRange other) {
            return Long.compare(this.startLine, other.startLine);
        }
    }

    /**
     * Cache for DML statement ranges per parent statement (Performance Optimization A).
     * Built lazily on first access, cleared at start of each resolve() call.
     * Uses IdentityHashMap because we need object identity, not equals().
     */
    private final Map<TCustomSqlStatement, List<DmlRange>> dmlIndexCache = new IdentityHashMap<>();

    /**
     * Build DML index for a parent statement.
     */
    private List<DmlRange> buildDmlIndex(TCustomSqlStatement parent) {
        final List<DmlRange> ranges = new ArrayList<>();
        parent.acceptChildren(new TParseTreeVisitor() {
            @Override
            public void preVisit(TInsertSqlStatement stmt) {
                ranges.add(new DmlRange(stmt));
            }
            @Override
            public void preVisit(TUpdateSqlStatement stmt) {
                ranges.add(new DmlRange(stmt));
            }
            @Override
            public void preVisit(TDeleteSqlStatement stmt) {
                ranges.add(new DmlRange(stmt));
            }
            @Override
            public void preVisit(TSelectSqlStatement stmt) {
                ranges.add(new DmlRange(stmt));
            }
        });
        // Sort by startLine for efficient lookup
        java.util.Collections.sort(ranges);
        return ranges;
    }

    /**
     * Get or build the DML index for a parent statement (Performance Optimization A).
     */
    private List<DmlRange> getDmlIndex(TCustomSqlStatement parent) {
        return dmlIndexCache.computeIfAbsent(parent, this::buildDmlIndex);
    }

    /**
     * Find the innermost DML statement (INSERT/UPDATE/DELETE/SELECT) within a parent statement
     * that contains the given column reference (by line number range).
     * Uses cached DML index for O(log N) lookup instead of O(N) traversal.
     */
    private TCustomSqlStatement findDMLStatementContaining(TCustomSqlStatement parent, TObjectName column) {
        if (parent == null || column == null) return null;

        long columnLine = column.getLineNo();
        TCustomSqlStatement result = null;

        // Use cached DML index (Performance Optimization A)
        List<DmlRange> ranges = getDmlIndex(parent);

        // Find all DML statements that contain the column by line number
        // Need to check all ranges that could contain the column (can't use pure binary search
        // because ranges can overlap and we want the innermost one)
        for (DmlRange range : ranges) {
            // Optimization: if startLine > columnLine, no more ranges can contain it
            if (range.startLine > columnLine) {
                break;
            }
            if (range.contains(columnLine)) {
                // Found a matching DML statement - prefer the innermost one (later startLine)
                if (result == null ||
                    (range.startLine >= result.getStartToken().lineNo)) {
                    result = range.stmt;
                }
            }
        }

        return result;
    }

    /**
     * Sync implicit database/schema from USE DATABASE/USE SCHEMA statements to AST.
     * This enables TObjectName.getAnsiSchemaName() and getAnsiCatalogName() to work correctly
     * for unqualified object names.
     *
     * This is similar to what TDatabaseObjectResolver does in the legacy resolver:
     * it visits all TObjectName nodes and sets implicitDatabaseName/implicitSchemaName
     * based on the current database/schema context.
     */
    private void syncImplicitDbSchemaToAST() {
        if (statementMetadata == null) return;
        for (Map.Entry<TCustomSqlStatement, gudusoft.gsqlparser.resolver2.metadata.StatementMetadata.Context> entry
                : statementMetadata.getContexts().entrySet()) {
            gudusoft.gsqlparser.resolver2.metadata.StatementMetadata.Context context = entry.getValue();
            entry.getKey().acceptChildren(new ImplicitDbSchemaVisitor(context.catalog, context.schema));
        }
    }

    /**
     * Visitor to set implicit database/schema on TObjectName nodes.
     */
    private static class ImplicitDbSchemaVisitor extends TParseTreeVisitor {
        private final String defaultCatalog;
        private final String defaultSchema;

        public ImplicitDbSchemaVisitor(String defaultCatalog, String defaultSchema) {
            this.defaultCatalog = defaultCatalog;
            this.defaultSchema = defaultSchema;
        }

        @Override
        public void preVisit(TObjectName node) {
            if (node == null) return;

            // Skip column objects - they don't need implicit DB/schema
            if (node.getDbObjectType() == EDbObjectType.column) return;

            // Skip objects with a db_link - they refer to remote databases
            // and should not inherit the current session's default schema/catalog
            if (node.getDblink() != null) return;

            // Set default database name if not qualified
            if (node.getDatabaseToken() == null) {
                node.setImplictDatabaseName(defaultCatalog);
            }

            // Set default schema name if not qualified
            if (node.getSchemaToken() == null) {
                node.setImplictSchemaName(defaultSchema);
            }
        }
    }

    /**
     * Selectively clear orphan column syntax hints (sphint) based on TSQLResolver2 resolution.
     *
     * Phase 1 (linkColumnToTable during parsing) adds sphint hints for columns it can't resolve.
     * TSQLResolver2 should:
     * 1. KEEP sphint hints for columns that are in allColumnReferences with NOT_FOUND/AMBIGUOUS status
     *    (these are genuinely orphan/ambiguous columns)
     * 2. CLEAR sphint hints for all other columns:
     *    - Columns successfully resolved (EXACT_MATCH)
     *    - Columns filtered out by ScopeBuilder (package constants, function keywords, etc.)
     *    - Columns in contexts TSQLResolver2 doesn't collect (MERGE VALUES, etc.)
     */
    private void clearOrphanColumnSyntaxHints() {
        // Build a set of positions for columns that should KEEP their sphint hints
        // These are columns in allColumnReferences with NOT_FOUND or AMBIGUOUS status
        Set<String> orphanPositions = new HashSet<>();

        for (TObjectName col : allColumnReferences) {
            if (col == null) continue;
            gudusoft.gsqlparser.resolver2.model.ResolutionResult resolution = col.getResolution();
            if (resolution != null) {
                ResolutionStatus status = resolution.getStatus();
                // Only keep sphint for genuinely AMBIGUOUS columns
                // NOT_FOUND columns might be due to TSQLResolver2 scope issues (e.g., MERGE WHEN clause)
                // so we clear their sphint to match old resolver behavior
                if (status == ResolutionStatus.AMBIGUOUS) {
                    TSourceToken startToken = col.getStartToken();
                    if (startToken != null) {
                        String key = startToken.lineNo + ":" + startToken.columnNo;
                        orphanPositions.add(key);
                    }
                }
            }
        }

        // Clear sphint hints for positions NOT in orphanPositions
        for (int i = 0; i < sqlStatements.size(); i++) {
            TCustomSqlStatement stmt = sqlStatements.get(i);
            if (stmt == null) continue;
            clearNonOrphanSphintHintsRecursive(stmt, orphanPositions);
        }
    }

    /**
     * Recursively clear sphint hints except for genuinely orphan columns.
     */
    private void clearNonOrphanSphintHintsRecursive(TCustomSqlStatement stmt, Set<String> orphanPositions) {
        if (stmt == null) return;

        // Clear sphint hints that are NOT for genuinely orphan columns
        if (stmt.getSyntaxHints() != null && stmt.getSyntaxHints().size() > 0) {
            for (int j = stmt.getSyntaxHints().size() - 1; j >= 0; j--) {
                TSyntaxError syntaxError = stmt.getSyntaxHints().get(j);
                if (syntaxError.errortype == EErrorType.sphint) {
                    String key = syntaxError.lineNo + ":" + syntaxError.columnNo;
                    if (!orphanPositions.contains(key)) {
                        // This sphint is NOT for a genuinely orphan column - clear it
                        stmt.getSyntaxHints().remove(j);
                        logDebug("Cleared sphint at line " + syntaxError.lineNo);
                    }
                    // Keep sphint hints for genuinely orphan columns (in orphanPositions)
                }
            }
        }

        // Note: orphanColumns is populated by populateOrphanColumns() in Phase 4b
        // DO NOT clear it here - TGetTableColumn relies on orphanColumns for
        // linkOrphanColumnToFirstTable functionality

        // Process nested statements
        for (int k = 0; k < stmt.getStatements().size(); k++) {
            clearNonOrphanSphintHintsRecursive(stmt.getStatements().get(k), orphanPositions);
        }
    }



    /**
     * Filter UNNEST table's linkedColumns to keep only legitimate columns.
     * Phase 1 (linkColumnToTable) may incorrectly link external variables to UNNEST
     * when UNNEST is the only table in scope. This method removes such incorrect links.
     *
     * Legitimate columns for UNNEST:
     * - Implicit column: the alias (e.g., "arry_pair" from "UNNEST(...) AS arry_pair")
     * - WITH OFFSET column (e.g., "pos" from "WITH OFFSET AS pos")
     * - Derived struct field columns (from UNNEST of STRUCT arrays)
     */
    private void filterUnnestLinkedColumns(TTable unnestTable) {
        if (unnestTable == null || unnestTable.getTableType() != ETableSource.unnest) {
            return;
        }

        TObjectNameList linkedColumns = unnestTable.getLinkedColumns();
        if (linkedColumns == null || linkedColumns.size() == 0) {
            return;
        }

        // Build set of legitimate column names
        java.util.Set<String> legitimateNames = new java.util.HashSet<>();

        // 1. Implicit column (alias name)
        String aliasName = unnestTable.getAliasName();
        if (aliasName != null && !aliasName.isEmpty()) {
            legitimateNames.add(aliasName.toUpperCase(java.util.Locale.ROOT));
        }

        // 2. WITH OFFSET column
        TUnnestClause unnestClause = unnestTable.getUnnestClause();
        if (unnestClause != null && unnestClause.getWithOffset() != null) {
            if (unnestClause.getWithOffsetAlais() != null &&
                unnestClause.getWithOffsetAlais().getAliasName() != null) {
                legitimateNames.add(unnestClause.getWithOffsetAlais().getAliasName().toString().toUpperCase(java.util.Locale.ROOT));
            } else {
                legitimateNames.add("OFFSET");
            }
        }

        // 3. Derived struct field columns
        if (unnestClause != null && unnestClause.getDerivedColumnList() != null) {
            for (int i = 0; i < unnestClause.getDerivedColumnList().size(); i++) {
                TObjectName derivedCol = unnestClause.getDerivedColumnList().getObjectName(i);
                if (derivedCol != null) {
                    legitimateNames.add(derivedCol.toString().toUpperCase(java.util.Locale.ROOT));
                }
            }
        }

        // 4. Explicit alias columns (Presto/Trino syntax: UNNEST(...) AS t(col1, col2))
        if (unnestTable.getAliasClause() != null &&
            unnestTable.getAliasClause().getColumns() != null) {
            for (int i = 0; i < unnestTable.getAliasClause().getColumns().size(); i++) {
                TObjectName colName = unnestTable.getAliasClause().getColumns().getObjectName(i);
                if (colName != null) {
                    legitimateNames.add(colName.toString().toUpperCase(java.util.Locale.ROOT));
                }
            }
        }

        // Collect columns to keep
        java.util.List<TObjectName> toKeep = new java.util.ArrayList<>();
        for (int i = 0; i < linkedColumns.size(); i++) {
            TObjectName col = linkedColumns.getObjectName(i);
            if (col != null) {
                String colName = col.getColumnNameOnly();
                if (colName != null && legitimateNames.contains(colName.toUpperCase(java.util.Locale.ROOT))) {
                    toKeep.add(col);
                }
            }
        }

        // Clear and re-add only legitimate columns
        linkedColumns.clear();
        for (TObjectName col : toKeep) {
            linkedColumns.addObjectName(col);
        }
    }

    /**
     * Clear linkedColumns on all tables in all statements.
     */
    private void clearAllLinkedColumns() {
        // Use a set to track processed statements and avoid processing duplicates
        // This is important when processing subqueries within tables, as the same
        // subquery might be reachable from multiple paths
        java.util.Set<TCustomSqlStatement> processed = new java.util.HashSet<>();
        for (int i = 0; i < sqlStatements.size(); i++) {
            clearLinkedColumnsRecursive(sqlStatements.get(i), processed);
        }
    }

    /**
     * Recursively clear orphanColumns on statements.
     * These will be repopulated with genuinely unresolved columns in Phase 4b.
     */
    private void clearOrphanColumnsRecursive(TCustomSqlStatement stmt) {
        if (stmt == null) return;

        if (stmt.getOrphanColumns() != null) {
            stmt.getOrphanColumns().clear();
        }

        // Process nested statements
        for (int i = 0; i < stmt.getStatements().size(); i++) {
            clearOrphanColumnsRecursive(stmt.getStatements().get(i));
        }

        // Also handle stored procedure/function body statements
        if (stmt instanceof gudusoft.gsqlparser.stmt.TStoredProcedureSqlStatement) {
            gudusoft.gsqlparser.stmt.TStoredProcedureSqlStatement sp =
                (gudusoft.gsqlparser.stmt.TStoredProcedureSqlStatement) stmt;
            for (int i = 0; i < sp.getBodyStatements().size(); i++) {
                clearOrphanColumnsRecursive(sp.getBodyStatements().get(i));
            }
        }
    }

    private void clearLinkedColumnsRecursive(TCustomSqlStatement stmt, java.util.Set<TCustomSqlStatement> processed) {
        if (stmt == null) return;

        // Skip if already processed to avoid redundant work and potential infinite loops
        if (processed.contains(stmt)) {
            return;
        }
        processed.add(stmt);

        // Skip DAX statements - they populate their own linkedColumns during parsing
        // via TDaxFunction.doParse() which calls psql.linkColumnToTable() directly.
        // TSQLResolver2's ScopeBuilder doesn't traverse DAX expressions, so we must
        // preserve the linkedColumns that DAX parsing already established.
        if (stmt instanceof TDaxStmt) {
            return;
        }

        // Skip ALTER TABLE statements - they populate linkedColumns during parsing
        // via TAlterTableOption.doParse() which directly adds columns to the target table's
        // linkedColumns. TSQLResolver2's ScopeBuilder doesn't traverse these option nodes,
        // so we must preserve the linkedColumns that parsing already established.
        if (stmt instanceof TAlterTableStatement) {
            return;
        }

        // For CREATE TABLE statements, we need special handling:
        // - Regular CREATE TABLE (with column definitions): Preserve constraint columns
        //   populated during TConstraint.doParse()
        // - CTAS (CREATE TABLE AS SELECT): Filter out source columns incorrectly added
        //   to target table, but preserve the correctly created alias columns
        boolean isCreateTable = (stmt instanceof TCreateTableSqlStatement);
        if (isCreateTable) {
            TCreateTableSqlStatement ctas = (TCreateTableSqlStatement) stmt;
            boolean isCTAS = (ctas.getSubQuery() != null);
            // For CTAS, filter out source columns from target table's linkedColumns
            // The old resolver incorrectly adds source columns (from the SELECT) to the target table
            // Keep only columns whose sourceTable is the target table itself
            if (isCTAS && ctas.getTargetTable() != null) {
                TTable targetTable = ctas.getTargetTable();
                TObjectNameList linkedColumns = targetTable.getLinkedColumns();
                if (linkedColumns != null && linkedColumns.size() > 0) {
                    // Collect columns to keep (those belonging to target table)
                    java.util.List<TObjectName> toKeep = new java.util.ArrayList<>();
                    for (int i = 0; i < linkedColumns.size(); i++) {
                        TObjectName col = linkedColumns.getObjectName(i);
                        if (col != null && col.getSourceTable() == targetTable) {
                            toKeep.add(col);
                        }
                    }
                    // Clear and re-add only the columns to keep
                    linkedColumns.clear();
                    for (TObjectName col : toKeep) {
                        linkedColumns.addObjectName(col);
                    }
                }
            }
        }

        if (!isCreateTable && stmt.tables != null) {
            // Check if this statement contains a TD_UNPIVOT table
            // TD_UNPIVOT populates linkedColumns on its inner table during TTDUnpivot.doParse()
            // If we clear linkedColumns here, we lose those column references
            boolean hasTDUnpivot = false;
            for (int i = 0; i < stmt.tables.size(); i++) {
                TTable table = stmt.tables.getTable(i);
                if (table != null && table.getTableType() == ETableSource.td_unpivot) {
                    hasTDUnpivot = true;
                    break;
                }
            }

            for (int i = 0; i < stmt.tables.size(); i++) {
                TTable table = stmt.tables.getTable(i);
                if (table != null && table.getLinkedColumns() != null) {
                    // For UNNEST tables, filter out incorrectly linked columns from Phase 1.
                    // Phase 1 (linkColumnToTable) may have linked external variables to UNNEST
                    // when it's the only table in scope. Keep only legitimate columns:
                    // - Implicit column (the UNNEST alias, e.g., "arry_pair" from "UNNEST(...) AS arry_pair")
                    // - WITH OFFSET column (e.g., "pos" from "WITH OFFSET AS pos")
                    if (table.getTableType() == ETableSource.unnest) {
                        filterUnnestLinkedColumns(table);
                        continue;
                    }
                    // Skip TD_UNPIVOT tables - they don't have their own columns but
                    // TTDUnpivot.doParse() populates columns on the inner table
                    if (table.getTableType() == ETableSource.td_unpivot) {
                        continue;
                    }
                    // If this statement contains TD_UNPIVOT, skip clearing all tables
                    // because TD_UNPIVOT populates linkedColumns on inner tables
                    if (hasTDUnpivot) {
                        continue;
                    }
                    table.getLinkedColumns().clear();
                }
            }
        }

        // Skip recursive processing if this statement contains TD_UNPIVOT
        // TD_UNPIVOT's inner table (in the ON clause) has columns populated during parsing
        // and those columns need to be preserved
        boolean hasTDUnpivot = false;
        if (stmt.tables != null) {
            for (int i = 0; i < stmt.tables.size(); i++) {
                TTable table = stmt.tables.getTable(i);
                if (table != null && table.getTableType() == ETableSource.td_unpivot) {
                    hasTDUnpivot = true;
                    break;
                }
            }
        }

        if (!hasTDUnpivot) {
            for (int i = 0; i < stmt.getStatements().size(); i++) {
                clearLinkedColumnsRecursive(stmt.getStatements().get(i), processed);
            }

            // Also process subqueries within tables - these are NOT in getStatements()
            // but are accessed via table.getSubquery()
            if (stmt.tables != null) {
                for (int i = 0; i < stmt.tables.size(); i++) {
                    TTable table = stmt.tables.getTable(i);
                    if (table != null && table.getSubquery() != null) {
                        clearLinkedColumnsRecursive(table.getSubquery(), processed);
                    }
                }
            }
        }
    }

    /**
     * Recursively fill TTable.getAttributes() for all tables in a statement.
     * Uses namespace data already collected during name resolution.
     *
     * Processing order is important:
     * 1. Process CTEs first
     * 2. Process leaf tables (objectname, function, etc.) - not JOIN or subquery
     * 3. Process subqueries (recursively)
     * 4. Process JOIN tables last (they depend on child tables having attributes)
     */
    private void fillTableAttributesRecursive(TCustomSqlStatement stmt, Set<TTable> processedTables) {
        if (stmt == null) return;

        // Skip DAX statements - they use their own attribute/linkedColumn mechanism
        // established during TDaxFunction.doParse() parsing phase.
        if (stmt instanceof TDaxStmt) {
            return;
        }

        // Skip ALTER TABLE statements - they use their own linkedColumn mechanism
        // established during TAlterTableOption.doParse() parsing phase.
        if (stmt instanceof TAlterTableStatement) {
            return;
        }

        // Skip CREATE TABLE statements - they use their own linkedColumn mechanism
        // established during TConstraint.doParse() parsing phase.
        if (stmt instanceof TCreateTableSqlStatement) {
            return;
        }

        // Phase 1: Process CTE tables first
        if (stmt instanceof TSelectSqlStatement) {
            TSelectSqlStatement selectStmt = (TSelectSqlStatement) stmt;
            TCTEList cteList = selectStmt.getCteList();
            if (cteList != null) {
                for (int i = 0; i < cteList.size(); i++) {
                    TCTE cte = cteList.getCTE(i);
                    if (cte != null && cte.getSubquery() != null) {
                        fillTableAttributesRecursive(cte.getSubquery(), processedTables);
                    }
                }
            }
        }

        // Collect tables by type for proper processing order
        List<TTable> leafTables = new ArrayList<>();
        List<TTable> subqueryTables = new ArrayList<>();
        List<TTable> joinTables = new ArrayList<>();

        // First, collect from stmt.tables
        if (stmt.tables != null) {
            for (int i = 0; i < stmt.tables.size(); i++) {
                TTable table = stmt.tables.getTable(i);
                if (table == null || processedTables.contains(table)) continue;

                switch (table.getTableType()) {
                    case join:
                        joinTables.add(table);
                        // Also collect nested tables within the join
                        collectNestedJoinTables(table, leafTables, subqueryTables, joinTables, processedTables);
                        break;
                    case subquery:
                        subqueryTables.add(table);
                        break;
                    default:
                        leafTables.add(table);
                        break;
                }
            }
        }

        // Also collect from getRelations() - JOIN tables are often stored there
        if (stmt.getRelations() != null) {
            for (int i = 0; i < stmt.getRelations().size(); i++) {
                IRelation rel = stmt.getRelations().get(i);
                if (!(rel instanceof TTable)) continue;
                TTable table = (TTable) rel;
                if (processedTables.contains(table)) continue;

                if (table.getTableType() == ETableSource.join) {
                    if (!joinTables.contains(table)) {
                        joinTables.add(table);
                        // Also collect nested tables within the join
                        collectNestedJoinTables(table, leafTables, subqueryTables, joinTables, processedTables);
                    }
                }
            }
        }

        // Phase 2: Process leaf tables first (objectname, function, xml, etc.)
        for (TTable table : leafTables) {
            if (!processedTables.contains(table)) {
                fillTableAttributes(table, processedTables, stmt);
                processedTables.add(table);
            }
        }

        // Phase 3: Process subqueries (recursively process their contents first)
        for (TTable table : subqueryTables) {
            if (!processedTables.contains(table)) {
                if (table.getSubquery() != null) {
                    fillTableAttributesRecursive(table.getSubquery(), processedTables);
                }
                fillTableAttributes(table, processedTables, stmt);
                processedTables.add(table);
            }
        }

        // Phase 4: Process JOIN tables last (they need child tables to have attributes)
        for (TTable table : joinTables) {
            if (!processedTables.contains(table)) {
                fillTableAttributes(table, processedTables, stmt);
                processedTables.add(table);
            }
        }

        // Process nested statements
        for (int i = 0; i < stmt.getStatements().size(); i++) {
            fillTableAttributesRecursive(stmt.getStatements().get(i), processedTables);
        }
    }

    /**
     * Collect nested tables within a JOIN expression.
     * This ensures all component tables are processed before the JOIN itself.
     */
    private void collectNestedJoinTables(TTable joinTable,
                                         List<TTable> leafTables,
                                         List<TTable> subqueryTables,
                                         List<TTable> joinTables,
                                         Set<TTable> processedTables) {
        if (joinTable == null || joinTable.getJoinExpr() == null) return;

        TJoinExpr joinExpr = joinTable.getJoinExpr();

        // Process left table
        TTable leftTable = joinExpr.getLeftTable();
        if (leftTable != null && !processedTables.contains(leftTable)) {
            switch (leftTable.getTableType()) {
                case join:
                    joinTables.add(leftTable);
                    collectNestedJoinTables(leftTable, leafTables, subqueryTables, joinTables, processedTables);
                    break;
                case subquery:
                    subqueryTables.add(leftTable);
                    break;
                default:
                    leafTables.add(leftTable);
                    break;
            }
        }

        // Process right table
        TTable rightTable = joinExpr.getRightTable();
        if (rightTable != null && !processedTables.contains(rightTable)) {
            switch (rightTable.getTableType()) {
                case join:
                    joinTables.add(rightTable);
                    collectNestedJoinTables(rightTable, leafTables, subqueryTables, joinTables, processedTables);
                    break;
                case subquery:
                    subqueryTables.add(rightTable);
                    break;
                default:
                    leafTables.add(rightTable);
                    break;
            }
        }
    }

    /**
     * Fill TTable.getAttributes() for a single table using namespace data.
     * This converts the namespace's columnSources to TAttributeNode objects.
     *
     * @param table The table to fill attributes for
     * @param processedTables Set of already processed tables to avoid duplicates
     * @param stmt The statement context (used for UNNEST to get the SELECT statement)
     */
    private void fillTableAttributes(TTable table, Set<TTable> processedTables, TCustomSqlStatement stmt) {
        if (table == null) return;

        // Clear existing attributes
        table.getAttributes().clear();

        String displayName = table.getDisplayName(true);
        if (displayName == null || displayName.isEmpty()) {
            displayName = table.getAliasName();
            if (displayName == null || displayName.isEmpty()) {
                displayName = table.getName();
            }
        }

        // First, try to use existing namespace from ScopeBuildResult
        // Skip namespace lookup for UNNEST tables - they need special handling via initAttributesForUnnest
        INamespace existingNamespace = null;
        if (table.getTableType() != ETableSource.unnest) {
            existingNamespace = scopeBuildResult != null
                ? scopeBuildResult.getNamespaceForTable(table)
                : null;
        }

        if (existingNamespace != null) {
            // Use existing namespace's column sources
            // Returns false if namespace has no real metadata (only inferred columns)
            if (fillAttributesFromNamespace(table, existingNamespace, displayName)) {
                return;
            }
            // Fall through to legacy logic if no real metadata
        }

        // Fall back to type-specific handling if no namespace found
        switch (table.getTableType()) {
            case objectname:
                if (table.isCTEName()) {
                    // CTE reference - use initAttributesFromCTE
                    TCTE cte = table.getCTE();
                    if (cte != null) {
                        table.initAttributesFromCTE(cte);
                    }
                } else {
                    // Physical table - create TableNamespace and extract columns
                    fillPhysicalTableAttributes(table, displayName);
                }
                break;

            case subquery:
                // Subquery - use initAttributesFromSubquery
                if (table.getSubquery() != null) {
                    String prefix = "";
                    if (table.getAliasClause() != null) {
                        prefix = table.getAliasClause().toString() + ".";
                    }
                    table.initAttributesFromSubquery(table.getSubquery(), prefix);
                }
                break;

            case join:
                // JOIN - combine attributes from left and right tables
                // First, add USING columns to the left and right tables (if present)
                if (table.getJoinExpr() != null) {
                    addUsingColumnsToTables(table.getJoinExpr());
                    // Then initialize the join expression's attributes (which pulls from left/right tables)
                    table.getJoinExpr().initAttributes(0);
                }
                table.initAttributesForJoin();
                break;

            case function:
                // Table function
                table.initAttributeForTableFunction();
                break;

            case xmltable:
                // XML table
                table.initAttributeForXMLTable();
                break;

            case tableExpr:
                // Table expression
                TAttributeNode.addNodeToList(
                    new TAttributeNode(displayName + ".*", table),
                    table.getAttributes()
                );
                break;

            case rowList:
                // Row list
                table.initAttributeForRowList();
                break;

            case unnest:
                // UNNEST - initialize attributes using the SELECT statement context
                if (stmt instanceof TSelectSqlStatement) {
                    TSelectSqlStatement select = (TSelectSqlStatement) stmt;
                    table.initAttributesForUnnest(getSqlEnv(), select);
                }
                break;

            case pivoted_table:
                // PIVOT table
                table.initAttributesForPivotTable();
                break;
        }
    }

    /**
     * Fill table attributes from an existing namespace's column sources.
     * This uses the namespace data that was collected during ScopeBuilder traversal.
     *
     * @return true if attributes were successfully filled, false if should fall back to legacy logic
     */
    private boolean fillAttributesFromNamespace(TTable table, INamespace namespace, String displayName) {
        // Ensure namespace is validated
        if (!namespace.isValidated()) {
            namespace.validate();
        }

        // For TableNamespace without actual metadata (only inferred columns),
        // return false to fall back to legacy logic which uses wildcards
        if (namespace instanceof TableNamespace) {
            TableNamespace tableNs = (TableNamespace) namespace;
            // Check if the namespace has actual metadata by seeing if there are any columns
            // with high confidence from metadata sources (not inferred)
            Map<String, ColumnSource> columnSources = namespace.getAllColumnSources();
            boolean hasRealMetadata = false;
            for (ColumnSource source : columnSources.values()) {
                if (source.getConfidence() >= 1.0 &&
                    !("inferred_from_usage".equals(source.getEvidence()))) {
                    hasRealMetadata = true;
                    break;
                }
            }
            if (!hasRealMetadata) {
                // No real metadata, fall back to legacy logic with wildcards
                return false;
            }

            // Has metadata - use namespace columns
            for (Map.Entry<String, ColumnSource> entry : columnSources.entrySet()) {
                String colName = entry.getKey();
                ColumnSource source = entry.getValue();
                // Only include columns with real metadata, not inferred ones
                if (source.getConfidence() >= 1.0 &&
                    !("inferred_from_usage".equals(source.getEvidence()))) {
                    TAttributeNode.addNodeToList(
                        new TAttributeNode(displayName + "." + colName, table),
                        table.getAttributes()
                    );
                }
            }

            // If no columns after filtering, add wildcard
            if (table.getAttributes().isEmpty()) {
                TAttributeNode.addNodeToList(
                    new TAttributeNode(displayName + ".*", table),
                    table.getAttributes()
                );
            }
            return true;
        }

        // For other namespace types (SubqueryNamespace, CTENamespace, etc.),
        // use all column sources
        Map<String, ColumnSource> columnSources = namespace.getAllColumnSources();
        if (columnSources != null && !columnSources.isEmpty()) {
            for (Map.Entry<String, ColumnSource> entry : columnSources.entrySet()) {
                String colName = entry.getKey();
                TAttributeNode.addNodeToList(
                    new TAttributeNode(displayName + "." + colName, table),
                    table.getAttributes()
                );
            }
        }

        // If no columns found, add wildcard attribute
        if (table.getAttributes().isEmpty()) {
            TAttributeNode.addNodeToList(
                new TAttributeNode(displayName + ".*", table),
                table.getAttributes()
            );
        }
        return true;
    }

    /**
     * Fill attributes for a physical table using TableNamespace.
     */
    private void fillPhysicalTableAttributes(TTable table, String displayName) {
        // Create namespace for this table with sqlEnv and vendor for qualified name resolution
        TSQLEnv sqlEnv = globalContext != null ? globalContext.getSqlEnv() : null;
        EDbVendor vendor = table.dbvendor != null ? table.dbvendor : EDbVendor.dbvoracle;
        TableNamespace namespace = new TableNamespace(table, config.getNameMatcher(), sqlEnv, vendor);
        if (statementMetadata != null) statementMetadata.configure(namespace, table);

        // Validate to populate columnSources
        namespace.validate();

        // Convert columnSources to TAttributeNode
        Map<String, ColumnSource> columnSources = namespace.getAllColumnSources();
        if (columnSources != null && !columnSources.isEmpty()) {
            for (Map.Entry<String, ColumnSource> entry : columnSources.entrySet()) {
                String colName = entry.getKey();
                TAttributeNode.addNodeToList(
                    new TAttributeNode(displayName + "." + colName, table),
                    table.getAttributes()
                );
            }
        }

        // If no columns found from metadata, add wildcard attribute
        // (this allows any column to potentially match)
        if (table.getAttributes().isEmpty()) {
            // Add columns from linkedColumns if available
            if (table.getLinkedColumns() != null && table.getLinkedColumns().size() > 0) {
                for (TObjectName col : table.getLinkedColumns()) {
                    if (col.getCandidateTables() != null && col.getCandidateTables().size() > 1) {
                        continue;  // Skip ambiguous columns
                    }
                    TAttributeNode.addNodeToList(
                        new TAttributeNode(displayName + "." + col.getColumnNameOnly(), table),
                        table.getAttributes()
                    );
                }
            }
            // Add wildcard attribute
            TAttributeNode.addNodeToList(
                new TAttributeNode(displayName + ".*", table),
                table.getAttributes()
            );
        }
    }

    /**
     * Add USING columns to the left and right tables in a JOIN expression.
     * USING columns should appear in both tables' attribute lists before the wildcard.
     * This method recursively handles nested JOINs.
     */
    private void addUsingColumnsToTables(TJoinExpr joinExpr) {
        if (joinExpr == null) return;

        // Recursively handle nested joins
        TTable leftTable = joinExpr.getLeftTable();
        TTable rightTable = joinExpr.getRightTable();

        if (leftTable != null && leftTable.getTableType() == ETableSource.join && leftTable.getJoinExpr() != null) {
            addUsingColumnsToTables(leftTable.getJoinExpr());
        }
        if (rightTable != null && rightTable.getTableType() == ETableSource.join && rightTable.getJoinExpr() != null) {
            addUsingColumnsToTables(rightTable.getJoinExpr());
        }

        // Handle USING columns in this join
        gudusoft.gsqlparser.nodes.TObjectNameList usingColumns = joinExpr.getUsingColumns();
        if (usingColumns == null || usingColumns.size() == 0) return;

        // Add USING columns to both tables
        for (int i = 0; i < usingColumns.size(); i++) {
            TObjectName usingCol = usingColumns.getObjectName(i);
            if (usingCol == null) continue;
            String colName = usingCol.getColumnNameOnly();

            // Add to left table (insert before wildcard if possible)
            if (leftTable != null && leftTable.getTableType() != ETableSource.join) {
                addColumnAttributeBeforeWildcard(leftTable, colName);
            }

            // Add to right table (insert before wildcard if possible)
            if (rightTable != null && rightTable.getTableType() != ETableSource.join) {
                addColumnAttributeBeforeWildcard(rightTable, colName);
            }
        }
    }

    /**
     * Add a column attribute to a table, inserting before the wildcard (*) if present.
     * This ensures USING columns appear before the wildcard in the attribute list.
     */
    private void addColumnAttributeBeforeWildcard(TTable table, String columnName) {
        if (table == null || columnName == null) return;

        String displayName = table.getDisplayName(true);
        if (displayName == null || displayName.isEmpty()) {
            displayName = table.getAliasName();
            if (displayName == null || displayName.isEmpty()) {
                displayName = table.getName();
            }
        }

        String attrName = displayName + "." + columnName;

        // Check if attribute already exists
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        ArrayList<TAttributeNode> attrs = table.getAttributes();
        for (TAttributeNode attr : attrs) {
            if (sameQualifiedName(dbVendor, ESQLDataObjectType.dotColumn, attr.getName(), attrName)) {
                return;  // Already exists
            }
        }

        // Find the wildcard position
        int wildcardIndex = -1;
        for (int i = 0; i < attrs.size(); i++) {
            if (attrs.get(i).getName().endsWith(".*")) {
                wildcardIndex = i;
                break;
            }
        }

        // Insert before wildcard or add to end
        TAttributeNode newAttr = new TAttributeNode(attrName, table);
        if (wildcardIndex >= 0) {
            attrs.add(wildcardIndex, newAttr);
        } else {
            TAttributeNode.addNodeToList(newAttr, attrs);
        }
    }

    /**
     * Sync a single column to legacy structures.
     * @return true if column was synced (had a sourceTable)
     */
    private boolean syncColumnToLegacy(TObjectName column) {
        if (column == null) return false;

        // Special handling for star columns (SELECT *)
        // Star columns represent ALL tables in the FROM clause and should be synced to ALL tables
        // in their sourceTableList, not just the first one.
        String columnName = column.getColumnNameOnly();
        if (columnName != null && columnName.equals("*")) {
            java.util.ArrayList<TTable> sourceTableList = column.getSourceTableList();
            if (sourceTableList != null && sourceTableList.size() > 0) {
                boolean synced = false;
                for (TTable starTable : sourceTableList) {
                    if (starTable == null) continue;
                    // Skip subquery types - the star should be linked to physical tables
                    if (starTable.getTableType() == ETableSource.subquery) continue;
                    gudusoft.gsqlparser.nodes.TObjectNameList starLinkedColumns = starTable.getLinkedColumns();
                    if (starLinkedColumns != null && addColumnByIdentityIfAbsent(starLinkedColumns, column)) {
                        synced = true;
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("syncColumnToLegacy: Synced star column to sourceTableList table: "
                                + starTable.getTableName());
                        }
                    }
                }
                return synced;
            }
        }

        // Check if column is AMBIGUOUS - don't sync to legacy if it's ambiguous
        // Ambiguous columns should be added to orphanColumns, not linkedColumns
        // NOTE: Skip this check for star columns (*) since they are handled specially
        // via sourceTableList and should be linked to all tables in the FROM clause
        ResolutionResult resolution = column.getResolution();
        if (resolution != null && resolution.getStatus() == ResolutionStatus.AMBIGUOUS) {
            // Don't treat star columns as ambiguous - they're supposed to match all tables
            if (columnName != null && columnName.equals("*")) {
                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("syncColumnToLegacy: Star column has AMBIGUOUS status, proceeding with normal sync");
                }
                // Fall through to normal processing
            } else {
                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("syncColumnToLegacy: Skipping AMBIGUOUS column: " + column.toString()
                        + " with " + (resolution.getAmbiguousSource() != null ?
                            resolution.getAmbiguousSource().getCandidateCount() : 0) + " candidates");
                }
                // Clear sourceTable if it was set by Phase 1 (linkColumnToTable)
                // This ensures the column will be treated as orphan by TGetTableColumn
                if (column.getSourceTable() != null) {
                    column.setSourceTable(null);
                }
                return false;
            }
        }

        TTable sourceTable = column.getSourceTable();
        ColumnSource source = column.getColumnSource();

        // Handle columns resolved through PlsqlVariableNamespace
        // These are stored procedure variables/parameters - mark them as variables
        // so they won't be added to orphan columns
        if (source != null && source.getSourceNamespace() instanceof gudusoft.gsqlparser.resolver2.namespace.PlsqlVariableNamespace) {
            column.setDbObjectTypeDirectly(EDbObjectType.variable);
            // Variables don't need to be linked to tables
            return false;
        }

        // Fix for subquery columns: When a column is EXPLICITLY QUALIFIED with a subquery alias
        // (e.g., mm.material_id), the old resolver Phase 1 may have incorrectly set sourceTable
        // to the physical table inside the subquery. TSQLResolver2 should correct this to point
        // to the subquery TTable itself. This preserves the intermediate layer for data lineage:
        // mm.material_id -> subquery mm -> physical table
        //
        // IMPORTANT: Only apply this correction for QUALIFIED columns. Unqualified columns
        // (like those inferred from star column expansion) should keep their physical table
        // sourceTable for proper data lineage tracing.
        if (source != null && column.isQualified()) {
            INamespace ns = source.getSourceNamespace();
            if (ns instanceof SubqueryNamespace) {
                TTable subqueryTable = ns.getSourceTable();
                // If the subquery's TTable is different from the current sourceTable,
                // use the subquery's TTable to maintain proper semantic layering
                if (subqueryTable != null && subqueryTable != sourceTable) {
                    if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                        logInfo("syncColumnToLegacy: Correcting sourceTable from " +
                            (sourceTable != null ? sourceTable.getTableName() : "null") +
                            " to subquery " + subqueryTable.getTableName() + " for qualified column " + column.toString());
                    }
                    sourceTable = subqueryTable;
                    column.setSourceTable(sourceTable);
                }
            }
        }

        // If sourceTable is null, try to get it from ColumnSource
        // This handles columns resolved to derived tables (subqueries with aliases)
        // where TSQLResolver2 resolved via ColumnSource but didn't set sourceTable on TObjectName
        if (sourceTable == null && source != null) {
            // For alias columns (isColumnAlias) or passthroughs to aliases (getFinalColumnName != null),
            // prefer the immediate source table (subquery/CTE) over the traced physical table.
            // The alias name doesn't exist in the physical table, so linking with alias name is wrong.
            boolean isAliasColumn = source.isColumnAlias() || source.getFinalColumnName() != null;
            if (isAliasColumn) {
                INamespace ns = source.getSourceNamespace();
                if (ns != null) {
                    TTable immediateTable = ns.getSourceTable();
                    if (immediateTable != null) {
                        sourceTable = immediateTable;
                        column.setSourceTable(sourceTable);
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("syncColumnToLegacy: Set sourceTable to immediate source for alias column "
                                + column.toString() + " -> " + immediateTable.getTableName());
                        }
                    }
                }
            }
            if (sourceTable == null) {
                TTable finalTable = source.getFinalTable();
                if (finalTable != null) {
                    sourceTable = finalTable;
                    column.setSourceTable(sourceTable);
                    if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                        logInfo("syncColumnToLegacy: Set sourceTable from ColumnSource.getFinalTable() for "
                            + column.toString() + " -> " + finalTable.getTableName());
                    }
                } else {
                    // Try getAllFinalTables() - this may succeed when getFinalTable() returns null
                    // For example, columns inferred through star push-down may have overrideTable set
                    // which getAllFinalTables() will return as a single-element list
                    java.util.List<TTable> allFinalTables = source.getAllFinalTables();
                    if (allFinalTables != null && !allFinalTables.isEmpty()) {
                        // Use the first non-subquery table from allFinalTables
                        for (TTable candidateTable : allFinalTables) {
                            if (candidateTable != null && candidateTable.getTableType() != ETableSource.subquery) {
                                sourceTable = candidateTable;
                                column.setSourceTable(sourceTable);
                                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                                    logInfo("syncColumnToLegacy: Set sourceTable from ColumnSource.getAllFinalTables() for "
                                        + column.toString() + " -> " + candidateTable.getTableName());
                                }
                                break;
                            }
                        }
                    }

                    // Fallback: try overrideTable for cases like derived tables in JOIN ON clauses
                    if (sourceTable == null) {
                        TTable overrideTable = source.getOverrideTable();
                        if (overrideTable != null) {
                            sourceTable = overrideTable;
                            column.setSourceTable(sourceTable);
                            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                                logInfo("syncColumnToLegacy: Set sourceTable from ColumnSource.getOverrideTable() for "
                                    + column.toString() + " -> " + overrideTable.getTableName());
                            }
                        }
                    }
                }
            }
        }

        if (sourceTable == null) {
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE && source != null) {
                logInfo("syncColumnToLegacy: Column " + column.toString()
                    + " has ColumnSource but no table. Namespace: "
                    + (source.getSourceNamespace() != null ? source.getSourceNamespace().getClass().getSimpleName() : "null")
                    + ", evidence: " + source.getEvidence());
            }
            return false;
        }

        // For struct-field access (e.g., customer.customer_id in BigQuery),
        // create a synthetic column representing the base column (e.g., "customer")
        // instead of using the original column which has the field name (e.g., "customer_id")
        if (source != null && source.isStructFieldAccess()) {
            String baseColumnName = source.getExposedName();
            if (source.getSourceNamespace() instanceof TableNamespace) {
                baseColumnName = ((TableNamespace) source.getSourceNamespace()).getColumnLookupSpelling(baseColumnName);
            }
            if (baseColumnName != null && !baseColumnName.isEmpty()) {
                // Create synthetic TObjectName for the base column
                EDbVendor vendor = config != null ? config.getVendor() : EDbVendor.dbvbigquery;
                TObjectName baseColumn = TObjectName.createObjectName(
                    vendor, EDbObjectType.column, baseColumnName);
                baseColumn.setSourceTable(sourceTable);
                // It reports the position of the base column as written, customer in customer.name: lineage
                // publishes that position for the column it models after the first reference to it.
                TSourceToken written = column.getFieldPathBaseToken();
                if (written == null) {
                    written = column.getStructStarBaseToken();
                }
                TSourceToken synthetic = baseColumn.getPartToken();
                if (written != null && synthetic != null) {
                    synthetic.lineNo = written.lineNo;
                    synthetic.columnNo = written.columnNo;
                    synthetic.offset = written.offset;
                }

                // Add the base column to linkedColumns (avoid duplicates by name)
                gudusoft.gsqlparser.nodes.TObjectNameList linkedColumns = sourceTable.getLinkedColumns();
                if (linkedColumns != null && !containsColumnByName(linkedColumns, baseColumnName)) {
                    addColumnByIdentityIfAbsent(linkedColumns, baseColumn);
                }
                return true;  // Skip adding the original struct-qualified column to linkedColumns.
                // DataFlowAnalyzer uses FieldPath from the original TObjectName to match
                // against the synthetic base column via getStructFieldFullName().
            }
        }

        // 1. Add to TTable.linkedColumns (avoid duplicates)
        gudusoft.gsqlparser.nodes.TObjectNameList linkedColumns = sourceTable.getLinkedColumns();
        if (linkedColumns != null) {
            // Mantis 4651: a traced star clone stands for a physical column that a
            // direct reference elsewhere in the statement may already have linked to
            // this very table; adding it again would report the column twice from
            // TTable.getLinkedColumns().
            //
            // The existing entry must already carry this table as its sourceTable to
            // count as covering the clone. An entry linked here by one of the
            // trace-through paths keeps pointing at the derived table it came from,
            // and consumers filtering linkedColumns by sourceTable — for instance
            // ModelBindingManager.guessTable — reject those, so treating a same-named
            // one as coverage would silently drop the physical link.
            boolean redundantClone = tracedStarClones.contains(column)
                    && containsColumnLinkedFromTable(linkedColumns, column.getColumnNameOnly(),
                            sourceTable);
            if (!redundantClone) {
                addColumnByIdentityIfAbsent(linkedColumns, column);
            }
        }

        // 2. For UNION scenarios, also add to all final tables from UNION branches
        // This is critical for star column push-down tests that expect columns to be
        // linked to ALL tables in a UNION, not just the first one.
        if (source != null) {
            java.util.List<TTable> allFinalTables = source.getAllFinalTables();
            if (allFinalTables != null && allFinalTables.size() > 1) {
                for (TTable unionTable : allFinalTables) {
                    if (unionTable == null || unionTable == sourceTable) continue;
                    // Skip subquery types - only link to physical tables
                    if (unionTable.getTableType() == ETableSource.subquery) continue;
                    gudusoft.gsqlparser.nodes.TObjectNameList unionLinkedColumns = unionTable.getLinkedColumns();
                    if (unionLinkedColumns != null) {
                        addColumnByIdentityIfAbsent(unionLinkedColumns, column);
                    }
                }
            }

            // 2b. For CTE columns, also link to the CTE reference table
            // When a column is resolved through a CTE, it should be linked to both:
            // - The CTE reference table (immediate source)
            // - The underlying physical tables (final source)
            INamespace ns = source.getSourceNamespace();
            if (ns instanceof gudusoft.gsqlparser.resolver2.namespace.CTENamespace) {
                gudusoft.gsqlparser.resolver2.namespace.CTENamespace cteNs =
                        (gudusoft.gsqlparser.resolver2.namespace.CTENamespace) ns;
                TTable cteTable = cteNs.getReferencingTable();
                if (cteTable != null && cteTable != sourceTable) {
                    gudusoft.gsqlparser.nodes.TObjectNameList cteLinkedColumns = cteTable.getLinkedColumns();
                    if (cteLinkedColumns != null) {
                        addColumnByIdentityIfAbsent(cteLinkedColumns, column);
                    }
                }
            }

            // 2c. For subquery columns, also link to the underlying physical tables
            // When sourceTable is a subquery (e.g., qualified column S.id from MERGE USING subquery),
            // TGetTableColumn needs the column to be linked to physical tables for output.
            // Use getFinalTable() to trace through to the ultimate physical table.
            // IMPORTANT: Only link if a column with the same name doesn't already exist -
            // this avoids duplicates when both outer and inner queries reference the same column.
            // EXCEPTION: Skip MERGE ON clause columns - they should not be linked to the source
            // subquery's underlying table because they may belong to the target table instead.
            if (sourceTable.getTableType() == ETableSource.subquery) {
                // Skip UNQUALIFIED join condition columns - they should not be traced to the source
                // subquery's underlying table via star column expansion.
                // This is particularly important for MERGE ON clause columns which may
                // belong to the target table rather than the source subquery.
                // QUALIFIED columns (like S.id) should still be traced as they explicitly reference
                // the source subquery.
                // Note: We check location only because ownStmt may be null for unresolved columns.
                boolean isUnqualifiedJoinConditionColumn = (column.getLocation() == ESqlClause.joinCondition)
                        && (column.getTableString() == null || column.getTableString().isEmpty());
                if (isUnqualifiedJoinConditionColumn && TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("syncColumnToLegacy: Skipping unqualified join condition column " + column.toString() +
                        " - should not be traced to subquery's underlying table");
                }

                // Skip alias columns - the alias name doesn't exist in the physical table,
                // so linking an alias-named column to the physical table produces wrong output
                // (e.g., TestTableEmployee.name instead of TestTableEmployee.ename).
                // getFinalTable() traces through aliases to find the physical table, but the
                // column name is still the alias. Only non-alias columns should be linked.
                boolean isAliasColumnForLinking = source.isColumnAlias() || source.getFinalColumnName() != null;

                if (!isUnqualifiedJoinConditionColumn && !isAliasColumnForLinking) {
                    TTable finalTable = source.getFinalTable();
                    if (finalTable != null && finalTable != sourceTable &&
                        finalTable.getTableType() != ETableSource.subquery) {
                        gudusoft.gsqlparser.nodes.TObjectNameList finalLinkedColumns = finalTable.getLinkedColumns();
                        // Mantis 4651: skip when createTracedColumnClones() already produced a
                        // clone for this (physical table, column name). The clone carries the
                        // same link with sourceTable pointing at the physical table — which is
                        // what consumers that filter linkedColumns by sourceTable expect — so
                        // linking the original reference here as well would only duplicate it.
                        if (finalLinkedColumns != null
                                && !hasColumnKey(tracedStarCloneKeys, finalTable,
                                        column.getColumnNameOnly())
                                && !containsColumnByName(finalLinkedColumns, column.getColumnNameOnly())
                                && addColumnByIdentityIfAbsent(finalLinkedColumns, column)) {
                            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                                logInfo("syncColumnToLegacy: Also linked " + column.toString() +
                                    " to underlying physical table " + finalTable.getTableName());
                            }
                        }
                    }
                }
            }

        }

        // 3. Sync linkedColumnDef and sourceColumn from ColumnSource
        if (source != null) {
            Object defNode = source.getDefinitionNode();

            // Set linkedColumnDef if definition is a TColumnDefinition
            if (defNode instanceof gudusoft.gsqlparser.nodes.TColumnDefinition) {
                column.setLinkedColumnDef((gudusoft.gsqlparser.nodes.TColumnDefinition) defNode);
            }

            // Set sourceColumn if definition is a TResultColumn
            // BUT skip for CTE explicit columns - these reference the CTE column name (e.g., "mgr_dept")
            // not the underlying SELECT column (e.g., "grp"). The CTE column is a TObjectName,
            // not a TResultColumn, so we cannot set it as sourceColumn.
            if (defNode instanceof TResultColumn) {
                String evidence = source.getEvidence();
                boolean isCTEExplicitColumn = evidence != null && evidence.startsWith("cte_explicit_column");
                if (!isCTEExplicitColumn) {
                    column.setSourceColumn((TResultColumn) defNode);
                }
            }
            // Special case: for star-inferred columns, set sourceColumn to the star column
            // The definitionNode is intentionally null to avoid affecting formatter output,
            // but we still need to set sourceColumn for legacy API compatibility.
            // Use setSourceColumnOnly to avoid changing dbObjectType which affects filtering.
            else if (defNode == null && source.getEvidence() != null
                    && source.getEvidence().contains("auto_inferred")) {
                // This is a star-inferred column - get the star column from the namespace
                INamespace namespace = source.getSourceNamespace();
                if (namespace != null) {
                    TResultColumn starColumn = namespace.getStarColumn();
                    if (starColumn != null) {
                        column.setSourceColumnOnly(starColumn);
                    }
                }
            }
        }

        return true;
    }

    /**
     * Check if a column already exists in the list (by identity).
     */
    private boolean addColumnByIdentityIfAbsent(TObjectNameList list, TObjectName column) {
        // Identity-index mirror of {@code list}. Equivalent to the previous
        // linear {@code containsColumn} scan (which tested {@code ==}), but O(1).
        //
        // Correctness (provably identical to the linear scan):
        //   These lists (linkedColumns/orphanColumns) are append-only within a
        //   resolution run, and the index is updated in lockstep on every add
        //   below. The only way the index can drift from the list is an append
        //   made outside this method; such an append changes list.size() without
        //   changing the index size, so the size guard rebuilds the index before
        //   it is consulted. Therefore, at every membership test the index equals
        //   the identity-set of the list's current contents, so the add/skip
        //   decision matches the linear scan exactly.
        Set<TObjectName> columns = linkedColumnIdentityCache.get(list);
        if (columns == null || columns.size() != list.size()) {
            columns = java.util.Collections.newSetFromMap(new IdentityHashMap<TObjectName, Boolean>());
            for (int i = 0; i < list.size(); i++) {
                columns.add(list.getObjectName(i));
            }
            linkedColumnIdentityCache.put(list, columns);
        }
        if (!columns.add(column)) {
            return false;
        }
        list.addObjectName(column);
        return true;
    }

    /**
     * Mantis 4651 — canonical identity of a column-name segment under the statement's
     * vendor, so that key equality is exactly {@link SQLUtil#sameName}.
     *
     * <p>A folded string would not do. {@code toLowerCase()} (what the clone dedup
     * used before) collapses {@code "A"} and {@code "a"}, which on a
     * quoted-case-preserving vendor — PostgreSQL, Oracle — are two different columns;
     * and rendering a {@link CanonKey} to text re-splits them on the collation-based
     * vendors, whose canonical text is diagnostic rather than authoritative.</p>
     */
    private CanonKey canonColumnKey(String columnName) {
        IdentifierService configured = configuredIdentifierService();
        if (configured != null) {
            return configured.canonKeyOf(columnName != null ? columnName : "", ESQLDataObjectType.dotColumn,
                    gudusoft.gsqlparser.sqlenv.IdentifierInputForm.SQL_LEXICAL);
        }
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        return SQLUtil.canonKey(dbVendor, ESQLDataObjectType.dotColumn,
                columnName != null ? columnName : "");
    }

    private Boolean configuresIdentifierRules;

    /**
     * The analysis's identifier service when this resolution runs on rules of its own, else null.
     * The star push-down's own column bookkeeping must then compare under those rules: two
     * columns the configuration keeps apart are two columns of the base table, not one.
     * Under the vendor default the existing comparisons stay exactly as they were.
     */
    private IdentifierService configuredIdentifierService() {
        if (configuresIdentifierRules == null) {
            gudusoft.gsqlparser.sqlenv.IdentifierProfile profile = config.getIdentifierProfile();
            configuresIdentifierRules = profile != null && !profile.hasSameEffectivePolicyAs(
                    gudusoft.gsqlparser.sqlenv.IdentifierProfile.forVendor(profile.getVendor(),
                            gudusoft.gsqlparser.sqlenv.IdentifierProfile.VendorFlags.defaults()));
        }
        return configuresIdentifierRules ? config.identifierService() : null;
    }

    /** Column-name equality for the push-down bookkeeping; see {@link #configuredIdentifierService}. */
    private boolean samePushedColumnName(EDbVendor dbVendor, String a, String b) {
        IdentifierService configured = configuredIdentifierService();
        return configured != null ? configured.areEqual(a, b, ESQLDataObjectType.dotColumn)
                : SQLUtil.sameName(dbVendor, ESQLDataObjectType.dotColumn, a, b);
    }

    /**
     * Single-segment name equality under THIS resolution's rules: the configured identifier
     * profile when there is one ({@link #configuredIdentifierService}), else the vendor default
     * ({@link SQLUtil#sameName}). Every alias and qualifier lookup in this class compares through
     * it, so a configured profile (MySQL {@code lower_case_table_names=0}, a SQL Server
     * {@code _AI} collation) decides the lookup the same way the namespaces' matcher does.
     */
    private boolean sameName(EDbVendor vendor, ESQLDataObjectType type, String a, String b) {
        IdentifierService configured = configuredIdentifierService();
        return configured != null ? configured.areEqual(a, b, type) : SQLUtil.sameName(vendor, type, a, b);
    }

    /** {@link #sameName} for possibly-qualified names ({@link SQLUtil#compareIdentifier}). */
    private boolean sameQualifiedName(EDbVendor vendor, ESQLDataObjectType type, String a, String b) {
        IdentifierService configured = configuredIdentifierService();
        return configured != null ? SQLUtil.compareIdentifier(configured, type, a, b)
                : SQLUtil.compareIdentifier(vendor, type, a, b);
    }

    /** The stored form of a name under this resolution's rules (see {@link #sameName}). */
    private String normalizeName(EDbVendor vendor, ESQLDataObjectType type, String name) {
        IdentifierService configured = configuredIdentifierService();
        if (configured == null) {
            return IdentifierService.normalizeStatic(vendor, type, name);
        }
        return name == null || name.isEmpty() ? name : configured.normalize(name, type);
    }

    /** The identifier service of this resolution: the configured one, else the vendor default. */
    private IdentifierService identifierServiceFor(EDbVendor vendor) {
        IdentifierService configured = configuredIdentifierService();
        return configured != null ? configured : IdentifierService.defaultServiceFor(vendor);
    }

    /**
     * Mantis 4651 — record {@code (table, columnName)} in an identity-keyed index.
     *
     * @return true when the pair was not already present
     */
    private boolean addColumnKey(Map<TTable, Set<CanonKey>> index, TTable table, String columnName) {
        Set<CanonKey> keys = index.get(table);
        if (keys == null) {
            keys = new HashSet<CanonKey>();
            index.put(table, keys);
        }
        return keys.add(canonColumnKey(columnName));
    }

    /** Mantis 4651 — companion lookup for {@link #addColumnKey}. */
    private boolean hasColumnKey(Map<TTable, Set<CanonKey>> index, TTable table, String columnName) {
        Set<CanonKey> keys = index.get(table);
        return keys != null && keys.contains(canonColumnKey(columnName));
    }

    /**
     * Mantis 4651 — like {@link #containsColumnByName}, but only counts entries that
     * are already linked <em>from</em> {@code table}, i.e. whose {@code sourceTable}
     * is that table. An entry that merely traces through to {@code table} while
     * pointing at some derived table is not equivalent to a physical link.
     */
    private boolean containsColumnLinkedFromTable(TObjectNameList list, String columnName,
                                                  TTable table) {
        if (columnName == null) return false;
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        for (int i = 0; i < list.size(); i++) {
            TObjectName col = list.getObjectName(i);
            if (col != null && col.getSourceTable() == table
                    && samePushedColumnName(dbVendor, col.getColumnNameOnly(), columnName)) {
                return true;
            }
        }
        return false;
    }

    /**
     * Check if a column with the given name already exists in the list.
     * Used for struct-field access where we create synthetic columns.
     */
    private boolean containsColumnByName(gudusoft.gsqlparser.nodes.TObjectNameList list, String columnName) {
        if (columnName == null) return false;
        // P0d.3a: canonical equality replaces the hand-rolled strip-quotes-then-
        // equalsIgnoreCase (which collapsed quoted-case distinctions on
        // quoted-case-preserving vendors).
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        for (int i = 0; i < list.size(); i++) {
            TObjectName col = list.getObjectName(i);
            if (col != null) {
                if (samePushedColumnName(dbVendor, col.getColumnNameOnly(), columnName)) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Check if a subquery SELECT statement has an explicit (non-star) column with the given name.
     * This is used to determine whether to create traced column clones:
     * - If the column matches an explicit column in the subquery, don't clone (stays at subquery level)
     * - If the column doesn't match explicit columns (must come from star), clone to physical table
     *
     * @param subquery the SELECT statement to check
     * @param columnName the column name to look for (may have quotes)
     * @return true if the subquery has an explicit column matching the name
     */
    private boolean subqueryHasExplicitColumn(TSelectSqlStatement subquery, String columnName) {
        if (subquery == null || columnName == null) {
            return false;
        }

        // For combined queries (UNION/INTERSECT/EXCEPT), follow left chain iteratively
        TSelectSqlStatement current = subquery;
        while (current.isCombinedQuery()) {
            current = current.getLeftStmt();
            if (current == null) {
                return false;
            }
        }
        subquery = current;

        TResultColumnList resultColumns = subquery.getResultColumnList();
        if (resultColumns == null) {
            return false;
        }

        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;

        for (int i = 0; i < resultColumns.size(); i++) {
            TResultColumn rc = resultColumns.getResultColumn(i);
            if (rc == null) {
                continue;
            }

            String colStr = rc.toString();
            // Skip star columns - they're not explicit columns
            if (colStr != null && (colStr.equals("*") || colStr.endsWith(".*"))) {
                continue;
            }

            // Get the effective column name (alias if present, otherwise the column name)
            String effectiveName = null;
            if (rc.getAliasClause() != null && rc.getAliasClause().getAliasName() != null) {
                effectiveName = rc.getAliasClause().getAliasName().toString();
            } else if (rc.getExpr() != null && rc.getExpr().getObjectOperand() != null) {
                // For simple column references like "t1.COL1", get the column name
                effectiveName = rc.getExpr().getObjectOperand().getColumnNameOnly();
            }

            if (effectiveName != null) {
                if (sameName(dbVendor, ESQLDataObjectType.dotColumn, effectiveName, columnName)) {
                    return true;
                }
            }
        }

        return false;
    }

    /**
     * Expand star columns using push-down inferred columns from namespaces.
     *
     * This is the core of the star column push-down algorithm:
     * 1. Find all star columns in SELECT lists
     * 2. For each star column, find its source namespace(s)
     * 3. Get inferred columns from the namespace (collected during resolution)
     * 4. Expand the star column by populating attributeNodesDerivedFromFromClause
     *
     * This enables star column expansion without TSQLEnv metadata by using
     * columns referenced in outer queries to infer what the star expands to.
     */
    private void expandStarColumnsUsingPushDown() {
        starExpansionSources.clear();
        int expandedCount = 0;
        Set<TCustomSqlStatement> processedStmts = new HashSet<>();

        // Track expanded star columns by their string representation for syncing
        Map<String, ArrayList<TAttributeNode>> expandedStarCols = new HashMap<>();

        // Process all statements recursively
        for (int i = 0; i < sqlStatements.size(); i++) {
            expandedCount += expandStarColumnsInStatement(sqlStatements.get(i), processedStmts, expandedStarCols);
        }

        // Sync expanded attributes to column references in getAllColumnReferences()
        // The result column TObjectNames might be different instances than those collected
        // during scope building, so we need to copy the expanded attrs
        if (scopeBuildResult != null && !expandedStarCols.isEmpty()) {
            for (TObjectName colRef : scopeBuildResult.getAllColumnReferences()) {
                if (colRef == null) continue;
                String colStr = colRef.toString();
                if (colStr == null || !colStr.endsWith("*")) continue;

                // Skip if already has expanded attrs
                ArrayList<TAttributeNode> existingAttrs = colRef.getAttributeNodesDerivedFromFromClause();
                if (existingAttrs != null && !existingAttrs.isEmpty()) continue;

                // Find matching expanded star column
                ArrayList<TAttributeNode> expandedAttrs = expandedStarCols.get(colStr);
                if (expandedAttrs != null && !expandedAttrs.isEmpty()) {
                    // Copy the expanded attrs to this column reference
                    for (TAttributeNode attr : expandedAttrs) {
                        TAttributeNode.addNodeToList(attr, existingAttrs);
                    }
                    if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                        logInfo("Synced " + expandedAttrs.size() + " expanded attrs to column reference: " + colStr);
                    }
                }
            }
        }

        logInfo("Expanded star columns using push-down: " + expandedCount + " columns added");
    }

    /**
     * Recursively expand star columns in a statement and its nested statements.
     * Uses processedStmts to track ALL statements (not just SELECTs) to prevent infinite loops.
     */
    private int expandStarColumnsInStatement(TCustomSqlStatement stmt, Set<TCustomSqlStatement> processedStmts,
                                              Map<String, ArrayList<TAttributeNode>> expandedStarCols) {
        if (stmt == null) return 0;

        // Cycle detection: skip if already processed this statement
        if (processedStmts.contains(stmt)) {
            return 0;
        }
        processedStmts.add(stmt);

        int count = 0;

        // Handle SELECT statements
        if (stmt instanceof TSelectSqlStatement) {
            TSelectSqlStatement select = (TSelectSqlStatement) stmt;
            count += expandStarColumnsInSelect(select, expandedStarCols);

            // Handle UNION/INTERSECT/EXCEPT - iteratively collect all branches
            if (select.isCombinedQuery()) {
                Deque<TSelectSqlStatement> unionStack = new ArrayDeque<>();
                if (select.getLeftStmt() != null) unionStack.push(select.getLeftStmt());
                if (select.getRightStmt() != null) unionStack.push(select.getRightStmt());
                while (!unionStack.isEmpty()) {
                    TSelectSqlStatement branch = unionStack.pop();
                    if (branch == null || processedStmts.contains(branch)) continue;
                    processedStmts.add(branch);
                    count += expandStarColumnsInSelect(branch, expandedStarCols);
                    if (branch.isCombinedQuery()) {
                        if (branch.getLeftStmt() != null) unionStack.push(branch.getLeftStmt());
                        if (branch.getRightStmt() != null) unionStack.push(branch.getRightStmt());
                    } else {
                        // Process tables with subqueries in this branch
                        if (branch.tables != null) {
                            for (int i = 0; i < branch.tables.size(); i++) {
                                TTable table = branch.tables.getTable(i);
                                if (table != null && table.getSubquery() != null) {
                                    count += expandStarColumnsInStatement(table.getSubquery(), processedStmts, expandedStarCols);
                                }
                            }
                        }
                        if (branch.getCteList() != null) {
                            for (int i = 0; i < branch.getCteList().size(); i++) {
                                TCTE cte = branch.getCteList().getCTE(i);
                                if (cte != null && cte.getSubquery() != null) {
                                    count += expandStarColumnsInStatement(cte.getSubquery(), processedStmts, expandedStarCols);
                                }
                            }
                        }
                    }
                }
            }
        }

        // Handle MERGE statements specially - process the USING clause
        if (stmt instanceof gudusoft.gsqlparser.stmt.TMergeSqlStatement) {
            gudusoft.gsqlparser.stmt.TMergeSqlStatement merge = (gudusoft.gsqlparser.stmt.TMergeSqlStatement) stmt;
            TTable usingTable = merge.getUsingTable();
            if (usingTable != null && usingTable.getSubquery() != null) {
                count += expandStarColumnsInStatement(usingTable.getSubquery(), processedStmts, expandedStarCols);
            }
        }

        // Process nested statements
        if (stmt.getStatements() != null) {
            for (int i = 0; i < stmt.getStatements().size(); i++) {
                Object nested = stmt.getStatements().get(i);
                if (nested instanceof TCustomSqlStatement) {
                    count += expandStarColumnsInStatement((TCustomSqlStatement) nested, processedStmts, expandedStarCols);
                }
            }
        }

        // Process tables with subqueries
        if (stmt.tables != null) {
            for (int i = 0; i < stmt.tables.size(); i++) {
                TTable table = stmt.tables.getTable(i);
                if (table != null && table.getSubquery() != null) {
                    count += expandStarColumnsInStatement(table.getSubquery(), processedStmts, expandedStarCols);
                }
            }
        }

        // Process CTEs
        if (stmt.getCteList() != null) {
            for (int i = 0; i < stmt.getCteList().size(); i++) {
                TCTE cte = stmt.getCteList().getCTE(i);
                if (cte != null && cte.getSubquery() != null) {
                    count += expandStarColumnsInStatement(cte.getSubquery(), processedStmts, expandedStarCols);
                }
            }
        }

        return count;
    }

    /**
     * Expand star columns in a SELECT statement's result column list.
     */
    private int expandStarColumnsInSelect(TSelectSqlStatement select, Map<String, ArrayList<TAttributeNode>> expandedStarCols) {
        if (select == null || select.getResultColumnList() == null) return 0;

        int count = 0;
        TResultColumnList resultCols = select.getResultColumnList();

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("expandStarColumnsInSelect: Processing SELECT with " + resultCols.size() + " result columns");
        }

        for (int i = 0; i < resultCols.size(); i++) {
            TResultColumn rc = resultCols.getResultColumn(i);
            if (rc == null || rc.getExpr() == null) continue;

            TObjectName objName = rc.getExpr().getObjectOperand();
            if (objName == null) continue;

            String colStr = objName.toString();
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE && colStr != null) {
                logInfo("expandStarColumnsInSelect: Column " + i + ": " + colStr);
            }
            if (colStr == null || !colStr.endsWith("*")) continue;

            // This is a star column - expand it
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                logInfo("expandStarColumnsInSelect: Found star column: " + colStr);
            }
            count += expandSingleStarColumn(objName, select, colStr, rc);

            // Track the expanded attrs for syncing to column references
            ArrayList<TAttributeNode> attrList = objName.getAttributeNodesDerivedFromFromClause();
            if (attrList != null && !attrList.isEmpty()) {
                expandedStarCols.put(colStr, attrList);
            }
        }

        return count;
    }

    /**
     * Expand a single star column using push-down inferred columns.
     *
     * @param starColumn The star column TObjectName (e.g., "*" or "src.*")
     * @param select The containing SELECT statement
     * @param colStr The string representation of the star column
     * @param resultColumn The TResultColumn containing the star (for EXCEPT column list)
     * @return Number of columns added
     */
    private int expandSingleStarColumn(TObjectName starColumn, TSelectSqlStatement select, String colStr, TResultColumn resultColumn) {
        ArrayList<TAttributeNode> attrList = starColumn.getAttributeNodesDerivedFromFromClause();

        // Skip if already expanded
        if (!attrList.isEmpty()) {
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                logInfo("expandSingleStarColumn: " + colStr + " already expanded with " + attrList.size() + " attrs");
            }
            return 0;
        }

        // Collect EXCEPT column names to exclude from expansion
        // (BigQuery: SELECT * EXCEPT (col1, col2) FROM ...)
        Set<String> exceptColumns = new HashSet<>();
        if (resultColumn != null) {
            TObjectNameList exceptList = resultColumn.getExceptColumnList();
            if (exceptList != null && exceptList.size() > 0) {
                for (int i = 0; i < exceptList.size(); i++) {
                    TObjectName exceptCol = exceptList.getObjectName(i);
                    if (exceptCol != null) {
                        String exceptName = exceptCol.getColumnNameOnly();
                        if (exceptName != null && !exceptName.isEmpty()) {
                            exceptColumns.add(exceptName.toUpperCase(java.util.Locale.ROOT));
                        }
                    }
                }
                if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                    logInfo("expandSingleStarColumn: Found " + exceptColumns.size() +
                            " EXCEPT columns: " + exceptColumns);
                }
            }
        }

        int count = 0;
        boolean isQualified = colStr.contains(".") && !colStr.equals("*");

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("expandSingleStarColumn: " + colStr + " isQualified=" + isQualified);
        }

        if (isQualified) {
            // Qualified star (e.g., "src.*") - find the specific table/namespace
            String tablePrefix = colStr.substring(0, colStr.lastIndexOf('.'));
            count += expandQualifiedStar(starColumn, select, tablePrefix, attrList, exceptColumns);
        } else {
            // Unqualified star (*) - expand from all tables in FROM clause
            count += expandUnqualifiedStar(starColumn, select, attrList, exceptColumns);
        }

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("expandSingleStarColumn: " + colStr + " expanded to " + count + " columns");
        }

        return count;
    }

    /**
     * Expand a qualified star column (e.g., "src.*") using namespace inferred columns.
     *
     * @param starColumn The star column TObjectName
     * @param select The containing SELECT statement
     * @param tablePrefix The table prefix (e.g., "src" from "src.*")
     * @param attrList The list to add expanded attributes to
     * @param exceptColumns Column names to exclude (from EXCEPT clause), uppercase
     */
    private int expandQualifiedStar(TObjectName starColumn, TSelectSqlStatement select,
                                     String tablePrefix, ArrayList<TAttributeNode> attrList,
                                     Set<String> exceptColumns) {
        int count = 0;

        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
            logInfo("expandQualifiedStar: tablePrefix=" + tablePrefix +
                    ", exceptColumns=" + (exceptColumns != null ? exceptColumns : "none"));
        }

        // Find the source table by alias or name
        TTable sourceTable = findTableByPrefixInSelect(select, tablePrefix);
        if (sourceTable == null) {
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                logInfo("expandQualifiedStar: No source table found for " + tablePrefix);
            }
            // Fall back to just adding the qualified star attribute
            TAttributeNode.addNodeToList(
                new TAttributeNode(tablePrefix + ".*", null),
                attrList
            );
            return 0;
        }

        // Collect inferred columns from multiple sources:
        // 1. The table's own namespace (TableNamespace)
        // 2. If the SELECT is a CTE definition, the CTE's namespace
        // 3. If the SELECT is a subquery, the containing scope's namespace
        Set<String> allInferredCols = new HashSet<>();

        // Source 1: Get namespace for this table
        INamespace tableNamespace = scopeBuildResult != null
            ? scopeBuildResult.getNamespaceForTable(sourceTable)
            : null;

        if (tableNamespace != null) {
            Set<String> inferredCols = tableNamespace.getInferredColumns();
            if (inferredCols != null) {
                allInferredCols.addAll(inferredCols);
            }
        }

        // Source 2: Check if this SELECT is part of a CTE definition
        // If so, the CTE namespace may have inferred columns from outer queries
        Set<String> cteInferredCols = getInferredColumnsFromContainingCTE(select);
        if (cteInferredCols != null) {
            if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                logInfo("expandQualifiedStar: Adding " + cteInferredCols.size() +
                        " CTE inferred columns for " + tablePrefix);
            }
            allInferredCols.addAll(cteInferredCols);
        }

        // Source 3: Check the SELECT's output scope for inferred columns
        // IMPORTANT: For qualified star columns (like ta.*), only use scope-level inferred columns
        // if they actually exist in this table's namespace. Otherwise we'd incorrectly add columns
        // from other tables in the FROM clause to this star's expanded attributes.
        IScope selectScope = scopeBuildResult != null
            ? scopeBuildResult.getScopeForStatement(select)
            : null;
        if (selectScope != null) {
            Set<String> scopeInferredCols = getInferredColumnsFromScope(selectScope);
            if (scopeInferredCols != null && tableNamespace != null) {
                // Only add scope-level inferred columns that actually exist in this table's namespace
                // This prevents columns from other tables being incorrectly associated with this star
                Map<String, ColumnSource> columnSources = tableNamespace.getAllColumnSources();
                Set<String> tableInferredCols = tableNamespace.getInferredColumns();
                for (String scopeCol : scopeInferredCols) {
                    // Check if this column can be resolved within this table's namespace
                    boolean hasInNamespace = (columnSources != null && columnSources.containsKey(scopeCol)) ||
                                             (tableInferredCols != null && tableInferredCols.contains(scopeCol));
                    if (hasInNamespace) {
                        allInferredCols.add(scopeCol);
                    }
                }
            } else if (scopeInferredCols != null && tableNamespace == null) {
                // No table namespace - add all scope columns (fallback for edge cases)
                allInferredCols.addAll(scopeInferredCols);
            }
        }

        if (!allInferredCols.isEmpty()) {
            // Expand using inferred columns, filtering out EXCEPT columns
            for (String colName : allInferredCols) {
                // Skip columns in EXCEPT clause
                if (exceptColumns != null && !exceptColumns.isEmpty() &&
                    exceptColumns.contains(colName.toUpperCase(java.util.Locale.ROOT))) {
                    if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                        logInfo("expandQualifiedStar: Skipping EXCEPT column: " + colName);
                    }
                    continue;
                }
                String attrName = tablePrefix + "." + colName;
                TAttributeNode.addNodeToList(
                    new TAttributeNode(attrName, sourceTable),
                    attrList
                );
                count++;
            }
        } else if (tableNamespace != null) {
            // No inferred columns - try to get from namespace's column sources
            Map<String, ColumnSource> columnSources = tableNamespace.getAllColumnSources();
            if (columnSources != null && !columnSources.isEmpty()) {
                for (Map.Entry<String, ColumnSource> sourceEntry : columnSources.entrySet()) {
                    String colName = sourceEntry.getKey();
                    // Skip columns in EXCEPT clause
                    if (exceptColumns != null && !exceptColumns.isEmpty() &&
                        exceptColumns.contains(colName.toUpperCase(java.util.Locale.ROOT))) {
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("expandQualifiedStar: Skipping EXCEPT column from sources: " + colName);
                        }
                        continue;
                    }
                    String attrName = tablePrefix + "." + colName;
                    TAttributeNode expanded = new TAttributeNode(attrName, sourceTable);
                    if (TAttributeNode.addNodeToList(expanded, attrList) && sourceEntry.getValue() != null) {
                        starExpansionSources.put(expanded, sourceEntry.getValue());
                    }
                    count++;
                }
            }
        }

        // If no columns were added, add the star as fallback
        if (count == 0) {
            TAttributeNode.addNodeToList(
                new TAttributeNode(tablePrefix + ".*", sourceTable),
                attrList
            );
        }

        return count;
    }

    /**
     * Get inferred columns from a CTE that contains the given SELECT statement.
     * Used for push-down: when outer queries reference columns from a CTE,
     * those columns are inferred in the CTE's namespace and should be used
     * to expand star columns in the CTE's SELECT.
     */
    private Set<String> getInferredColumnsFromContainingCTE(TSelectSqlStatement select) {
        if (select == null || scopeBuildResult == null || namespaceEnhancer == null) {
            return null;
        }

        // Find the CTE that defines this SELECT
        Set<INamespace> starNamespaces = namespaceEnhancer.getStarNamespaces();
        if (starNamespaces == null) {
            return null;
        }

        for (INamespace ns : starNamespaces) {
            if (ns instanceof CTENamespace) {
                CTENamespace cteNs = (CTENamespace) ns;
                TSelectSqlStatement cteSelect = cteNs.getSelectStatement();
                // Check both by reference and by start token position
                if (cteSelect == select ||
                    (cteSelect != null && select != null &&
                     cteSelect.getStartToken() != null && select.getStartToken() != null &&
                     cteSelect.getStartToken().posinlist == select.getStartToken().posinlist)) {
                    Set<String> inferredCols = cteNs.getInferredColumns();
                    if (inferredCols != null && !inferredCols.isEmpty()) {
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("getInferredColumnsFromContainingCTE: Found CTE " + cteNs.getDisplayName() +
                                    " with " + inferredCols.size() + " inferred columns");
                        }
                        return inferredCols;
                    }
                }
            } else if (ns instanceof SubqueryNamespace) {
                SubqueryNamespace subNs = (SubqueryNamespace) ns;
                TSelectSqlStatement subSelect = subNs.getSelectStatement();
                if (subSelect == select ||
                    (subSelect != null && select != null &&
                     subSelect.getStartToken() != null && select.getStartToken() != null &&
                     subSelect.getStartToken().posinlist == select.getStartToken().posinlist)) {
                    Set<String> inferredCols = subNs.getInferredColumns();
                    if (inferredCols != null && !inferredCols.isEmpty()) {
                        if (TBaseType.DUMP_RESOLVER_LOG_TO_CONSOLE) {
                            logInfo("getInferredColumnsFromContainingCTE: Found Subquery with " +
                                    inferredCols.size() + " inferred columns");
                        }
                        return inferredCols;
                    }
                }
            }
        }

        return null;
    }

    /**
     * Get inferred columns from namespaces in a scope's FROM clause.
     */
    private Set<String> getInferredColumnsFromScope(IScope scope) {
        if (scope == null) {
            return null;
        }

        Set<String> result = new HashSet<>();

        // Check all namespaces in the scope's children
        for (gudusoft.gsqlparser.resolver2.model.ScopeChild child : scope.getChildren()) {
            INamespace ns = child.getNamespace();
            if (ns != null) {
                Set<String> inferredCols = ns.getInferredColumns();
                if (inferredCols != null) {
                    result.addAll(inferredCols);
                }
            }
        }

        return result.isEmpty() ? null : result;
    }

    /**
     * Expand an unqualified star column (*) using all tables in FROM clause.
     *
     * @param starColumn The star column TObjectName
     * @param select The containing SELECT statement
     * @param attrList The list to add expanded attributes to
     * @param exceptColumns Column names to exclude (from EXCEPT clause), uppercase
     */
    private int expandUnqualifiedStar(TObjectName starColumn, TSelectSqlStatement select,
                                       ArrayList<TAttributeNode> attrList, Set<String> exceptColumns) {
        int count = 0;

        if (select.tables == null) return 0;

        for (int i = 0; i < select.tables.size(); i++) {
            TTable table = select.tables.getTable(i);
            if (table == null) continue;

            // Skip certain table types
            if (table.getTableType() == ETableSource.join) continue;

            String tablePrefix = table.getAliasName();
            if (tablePrefix == null || tablePrefix.isEmpty()) {
                tablePrefix = table.getName();
            }
            if (tablePrefix == null) continue;

            // Get namespace for this table
            INamespace namespace = scopeBuildResult != null
                ? scopeBuildResult.getNamespaceForTable(table)
                : null;

            if (namespace != null) {
                Set<String> inferredCols = namespace.getInferredColumns();

                if (inferredCols != null && !inferredCols.isEmpty()) {
                    for (String colName : inferredCols) {
                        // Skip columns in EXCEPT clause
                        if (exceptColumns != null && !exceptColumns.isEmpty() &&
                            exceptColumns.contains(colName.toUpperCase(java.util.Locale.ROOT))) {
                            continue;
                        }
                        String attrName = tablePrefix + "." + colName;
                        TAttributeNode.addNodeToList(
                            new TAttributeNode(attrName, table),
                            attrList
                        );
                        count++;
                    }
                } else {
                    Map<String, ColumnSource> columnSources = namespace.getAllColumnSources();
                    if (columnSources != null && !columnSources.isEmpty()) {
                        for (Map.Entry<String, ColumnSource> sourceEntry : columnSources.entrySet()) {
                            String colName = sourceEntry.getKey();
                            // Skip columns in EXCEPT clause
                            if (exceptColumns != null && !exceptColumns.isEmpty() &&
                                exceptColumns.contains(colName.toUpperCase(java.util.Locale.ROOT))) {
                                continue;
                            }
                            String attrName = tablePrefix + "." + colName;
                            TAttributeNode expanded = new TAttributeNode(attrName, table);
                            if (TAttributeNode.addNodeToList(expanded, attrList) && sourceEntry.getValue() != null) {
                                starExpansionSources.put(expanded, sourceEntry.getValue());
                            }
                            count++;
                        }
                    }
                }
            }

            // If no columns for this table, add the star as fallback
            if (count == 0 || (namespace != null && namespace.getInferredColumns().isEmpty()
                              && namespace.getAllColumnSources().isEmpty())) {
                TAttributeNode.addNodeToList(
                    new TAttributeNode(tablePrefix + ".*", table),
                    attrList
                );
            }
        }

        return count;
    }

    /**
     * Find a table by its prefix (alias or name) in a SELECT statement.
     */
    private TTable findTableByPrefixInSelect(TSelectSqlStatement select, String prefix) {
        if (select == null || select.tables == null || prefix == null) return null;

        // P0d.3a: canonical equality replaces the hand-rolled quote/backtick/bracket
        // stripper — the engine strips the vendor's own quote forms and folds by quote
        // state, so quoted-case distinctions survive where the vendor preserves them.
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;
        String trimmedPrefix = prefix.trim();

        for (int i = 0; i < select.tables.size(); i++) {
            TTable table = select.tables.getTable(i);
            if (table == null) continue;

            // Check alias first
            String alias = table.getAliasName();
            if (alias != null && sameName(dbVendor, ESQLDataObjectType.dotTable, alias.trim(), trimmedPrefix)) {
                return table;
            }

            // Check table name
            String name = table.getName();
            if (name != null && sameName(dbVendor, ESQLDataObjectType.dotTable, name.trim(), trimmedPrefix)) {
                return table;
            }

            // Check full table name (with schema)
            if (table.getTableName() != null) {
                String fullName = table.getTableName().toString();
                if (fullName != null && sameQualifiedName(dbVendor, ESQLDataObjectType.dotTable, fullName.trim(), trimmedPrefix)) {
                    return table;
                }
            }
        }

        return null;
    }

    /**
     * Get resolution statistics
     */
    public ResolutionStatistics getStatistics() {
        return resolutionContext.getStatistics();
    }

    /**
     * Get the resolution context (for advanced queries)
     */
    public ResolutionContext getContext() {
        return resolutionContext;
    }

    /**
     * Get the global scope
     */
    public GlobalScope getGlobalScope() {
        return globalScope;
    }

    /**
     * Get the configuration
     */
    public TSQLResolverConfig getConfig() {
        return config;
    }

    /**
     * Get the pass history (for iterative resolution analysis)
     *
     * @return list of all resolution passes (empty if non-iterative or not yet resolved)
     */
    public List<ResolutionPass> getPassHistory() {
        return new ArrayList<>(passHistory);
    }

    /**
     * Get the convergence detector (for iterative resolution analysis)
     *
     * @return convergence detector (null if iterative resolution is disabled)
     */
    public ConvergenceDetector getConvergenceDetector() {
        return convergenceDetector;
    }

    /**
     * Get the scope build result (for testing and analysis)
     *
     * @return scope build result from ScopeBuilder (null if not yet resolved)
     */
    public ScopeBuildResult getScopeBuildResult() {
        return scopeBuildResult;
    }

    /**
     * The namespace member a star-expanded attribute stands for.
     *
     * <p>Populated while {@code SELECT *} / {@code SELECT t.*} are expanded into
     * {@link TObjectName#getAttributeNodesDerivedFromFromClause()} from a namespace's column
     * sources, and keyed by the attribute's identity. {@code null} for an attribute that was not
     * produced from a namespace member — one built from an inferred column name, a fallback
     * {@code t.*} node, or a node created by the legacy resolver — so a consumer treats
     * {@code null} as "unknown", never as "absent".
     *
     * @param attribute an entry of a star column's expanded attribute list
     * @return the {@link ColumnSource} it was expanded from, or {@code null} if not recorded
     */
    public ColumnSource getStarExpansionSource(TAttributeNode attribute) {
        return attribute == null ? null : starExpansionSources.get(attribute);
    }

    /**
     * Return the diagnostic outcome of the most recent {@link #resolve()} call.
     *
     * @return an immutable non-null outcome
     */
    public RunOutcome getRunOutcome() {
        return runOutcome;
    }

    /**
     * Get the resolution result access interface.
     * This provides a clean, statement-centric API for accessing resolution results.
     *
     * <p>Usage example:</p>
     * <pre>
     * TSQLResolver2 resolver = new TSQLResolver2(null, parser.sqlstatements);
     * resolver.resolve();
     *
     * IResolutionResult result = resolver.getResult();
     *
     * for (TCustomSqlStatement stmt : parser.sqlstatements) {
     *     for (TTable table : result.getTables(stmt)) {
     *         System.out.println("Table: " + table.getFullName());
     *         for (TObjectName col : result.getColumnsForTable(stmt, table)) {
     *             System.out.println("  Column: " + col.getColumnNameOnly());
     *         }
     *     }
     * }
     * </pre>
     *
     * @return resolution result access interface
     * @throws IllegalStateException if resolve() has not been called
     */
    public IResolutionResult getResult() {
        if (scopeBuildResult == null) {
            throw new IllegalStateException(
                "Must call resolve() before getResult()");
        }
        return new ResolutionResultImpl(scopeBuildResult, sqlStatements);
    }

    // ===== Binding Diagnostic API (plan §5.3, S2 stub / S5 wired) =====

    /**
     * Aggregate binding result populated by {@link BindingDiagnosticPostPass}
     * after iterative resolution converges (S5). Always non-null — including
     * when binding flags are off, when {@code resolve()} has not been called,
     * or when {@link gudusoft.gsqlparser.TGSqlParser#parse()} returned a
     * syntax error (plan §5.6, §12).
     *
     * @return the binding result; never null
     */
    public BindingResult getBindingResult() {
        return bindingResult != null ? bindingResult : BindingResult.empty();
    }

    /**
     * Convenience accessor for {@link #getBindingResult()}{@code .getDiagnostics()}.
     *
     * @return the diagnostics list; never null
     */
    public List<BindingDiagnostic> getBindingDiagnostics() {
        return getBindingResult().getDiagnostics();
    }

    /**
     * Convenience accessor for {@link #getBindingResult()}{@code .hasErrors()}.
     *
     * @return whether any ERROR-severity binding diagnostic was emitted
     */
    public boolean hasBindingErrors() {
        return getBindingResult().hasErrors();
    }

    // ===== Star Column Reverse Inference Support (Principle 3) =====

    /**
     * Star Column push-down context for reverse inference.
     * Tracks which columns should be added to which Namespaces based on
     * outer layer references.
     */
    private static class StarPushDownContext {
        /** Namespace -> (ColumnName -> Confidence) */
        private final Map<INamespace, Map<String, Double>> pushDownMap = new HashMap<>();

        /**
         * Record that a column should be added to a namespace.
         * If the same column is pushed multiple times, keep the highest confidence.
         */
        public void pushColumn(INamespace namespace, String columnName, double confidence) {
            Map<String, Double> columns = pushDownMap.computeIfAbsent(namespace, k -> new HashMap<>());
            columns.put(columnName, Math.max(confidence, columns.getOrDefault(columnName, 0.0)));
        }

        /**
         * Get all columns that should be pushed to each namespace.
         */
        public Map<INamespace, java.util.Set<String>> getAllPushDownColumns() {
            Map<INamespace, java.util.Set<String>> result = new HashMap<>();
            for (Map.Entry<INamespace, Map<String, Double>> entry : pushDownMap.entrySet()) {
                result.put(entry.getKey(), entry.getValue().keySet());
            }
            return result;
        }

        /**
         * Get the confidence score for a specific column in a namespace.
         */
        public double getConfidence(INamespace namespace, String columnName) {
            return pushDownMap.getOrDefault(namespace, java.util.Collections.emptyMap())
                              .getOrDefault(columnName, 0.0);
        }

        /**
         * Get the total number of columns to be pushed down across all namespaces.
         */
        public int getTotalPushedColumns() {
            return pushDownMap.values().stream()
                              .mapToInt(Map::size)
                              .sum();
        }
    }

    /**
     * Represents a star column source (CTE or subquery with SELECT *).
     * Used for reverse inference to track which columns are required from the star.
     */
    private static class StarColumnSource {
        private final String name;  // CTE name or subquery alias
        private final INamespace namespace;  // The namespace for this source
        private final INamespace underlyingTableNamespace;  // Namespace of the table behind SELECT *
        private final java.util.Set<String> requiredColumns = new java.util.HashSet<>();

        public StarColumnSource(String name, INamespace namespace, INamespace underlyingTableNamespace) {
            this.name = name;
            this.namespace = namespace;
            this.underlyingTableNamespace = underlyingTableNamespace;
        }

        public String getName() {
            return name;
        }

        public INamespace getNamespace() {
            return namespace;
        }

        public void addRequiredColumn(String columnName) {
            requiredColumns.add(columnName);
        }

        public java.util.Set<String> getRequiredColumns() {
            return requiredColumns;
        }

        public boolean hasUnderlyingTable() {
            return underlyingTableNamespace != null;
        }

        public INamespace getUnderlyingTableNamespace() {
            return underlyingTableNamespace;
        }

        @Override
        public String toString() {
            return String.format("StarColumnSource[%s, required=%d]", name, requiredColumns.size());
        }
    }

    /**
     * Collect all star column sources (CTEs and subqueries with SELECT *).
     * Traverses the scope tree to find CTENamespace and SubqueryNamespace
     * that use SELECT * in their subqueries.
     */
    private List<StarColumnSource> collectAllStarColumnSources() {
        List<StarColumnSource> sources = new ArrayList<>();

        // Traverse global scope tree
        if (globalScope != null) {
            collectStarSourcesFromScope(globalScope, sources);
        }

        // Also traverse UPDATE scopes (for Teradata UPDATE...FROM syntax)
        if (scopeBuilder != null) {
            for (UpdateScope updateScope : scopeBuilder.getUpdateScopeMap().values()) {
                collectStarSourcesFromScope(updateScope, sources);
            }
            for (DeleteScope deleteScope : scopeBuilder.getDeleteScopeMap().values()) {
                collectStarSourcesFromScope(deleteScope, sources);
            }
        }

        logDebug("Collected " + sources.size() + " star column sources");
        return sources;
    }

    /**
     * Recursively collect star column sources from a scope and its children.
     */
    private void collectStarSourcesFromScope(IScope scope, List<StarColumnSource> sources) {
        // Check all child namespaces in this scope
        for (gudusoft.gsqlparser.resolver2.model.ScopeChild child : scope.getChildren()) {
            INamespace namespace = child.getNamespace();

            // Use the new interface method to check for star columns
            if (namespace.hasStarColumn()) {
                TSelectSqlStatement selectStmt = namespace.getSelectStatement();
                INamespace underlyingNs = selectStmt != null ? getFirstTableNamespace(selectStmt) : null;

                StarColumnSource starSource = new StarColumnSource(
                    namespace.getDisplayName(),
                    namespace,
                    underlyingNs
                );
                sources.add(starSource);

                logDebug("Found star source: " + namespace.getDisplayName());
            }
        }

        // Recursively traverse child scopes based on scope type
        if (scope instanceof SelectScope) {
            SelectScope selectScope = (SelectScope) scope;
            if (selectScope.getFromScope() != null) {
                collectStarSourcesFromScope(selectScope.getFromScope(), sources);
            }
        } else if (scope instanceof UpdateScope) {
            UpdateScope updateScope = (UpdateScope) scope;
            if (updateScope.getFromScope() != null) {
                collectStarSourcesFromScope(updateScope.getFromScope(), sources);
            }
        } else if (scope instanceof DeleteScope) {
            DeleteScope deleteScope = (DeleteScope) scope;
            if (deleteScope.getFromScope() != null) {
                collectStarSourcesFromScope(deleteScope.getFromScope(), sources);
            }
        }
    }


    /**
     * Get the first table namespace from a SELECT statement's FROM clause.
     * Returns the DynamicStarSource if available.
     */
    private INamespace getFirstTableNamespace(TSelectSqlStatement select) {
        if (select == null || select.tables == null || select.tables.size() == 0) {
            return null;
        }

        // Get first table
        TTable firstTable = select.tables.getTable(0);
        String tableName = firstTable.getAliasName() != null
            ? firstTable.getAliasName()
            : firstTable.getName();

        // Search for corresponding namespace in all dynamic namespaces
        List<INamespace> dynamicNamespaces = getAllDynamicNamespaces();
        for (INamespace ns : dynamicNamespaces) {
            if (ns.getDisplayName().equals(tableName)) {
                return ns;
            }
        }

        return null;
    }

    /**
     * Collect all outer references to a star column source.
     * Searches through allColumnReferences for columns that reference this star source.
     */
    private List<TObjectName> collectOuterReferencesToSource(StarColumnSource starSource) {
        List<TObjectName> references = new ArrayList<>();

        if (starSource == null || starSource.getName() == null) {
            return references;
        }

        String sourceName = starSource.getName();
        EDbVendor dbVendor = sqlStatements.get(0).dbvendor;

        // Search through all collected column references
        for (TObjectName objName : allColumnReferences) {
            if (objName == null) {
                continue;
            }

            // Check if this column reference is from the star source
            // E.g., for CTE named "my_cte", check if objName is like "my_cte.col1"
            String tableQualifier = getTableQualifier(objName);

            if (tableQualifier != null && sameQualifiedName(dbVendor, ESQLDataObjectType.dotTable, tableQualifier, sourceName)) {
                references.add(objName);
                logDebug("Found outer reference: " + objName + " -> " + sourceName);
            }
        }

        logDebug("Collected " + references.size() + " outer references for: " + sourceName);
        return references;
    }

    /**
     * Get the table qualifier from a TObjectName.
     * E.g., for "schema.table.column", returns "table"
     * E.g., for "table.column", returns "table"
     * E.g., for "column", returns null
     */
    private String getTableQualifier(TObjectName objName) {
        if (objName == null) {
            return null;
        }

        // TObjectName has parts like: [schema, table, column]
        // or [table, column]
        // or [column]

        // If there are 3 or more parts, the second-to-last is the table
        // If there are 2 parts, the first is the table
        // If there is 1 part, there's no table qualifier

        String fullName = objName.toString();
        String[] parts = fullName.split("\\.");

        if (parts.length >= 3) {
            // schema.table.column -> return table
            return parts[parts.length - 2];
        } else if (parts.length == 2) {
            // table.column -> return table
            return parts[0];
        } else {
            // Just column name, no qualifier
            return null;
        }
    }

    /**
     * Get all DynamicStarSource namespaces from the scope tree.
     * This is used to apply inference results to namespaces that need enhancement.
     */
    private List<INamespace> getAllDynamicNamespaces() {
        List<INamespace> result = new ArrayList<>();

        // Collect from global scope tree
        if (globalScope != null) {
            collectDynamicNamespacesFromScope(globalScope, result);
        }

        return result;
    }

    /**
     * Recursively collect DynamicStarSource namespaces from a scope and its children.
     */
    private void collectDynamicNamespacesFromScope(IScope scope, List<INamespace> result) {
        if (scope == null) {
            return;
        }

        // Get all child namespaces from this scope
        for (gudusoft.gsqlparser.resolver2.model.ScopeChild child : scope.getChildren()) {
            INamespace namespace = child.getNamespace();
            if (namespace instanceof gudusoft.gsqlparser.resolver2.namespace.DynamicStarSource) {
                result.add(namespace);
                logDebug("Found DynamicStarSource: " + namespace.getDisplayName());
            }
        }

        // Recursively traverse child scopes based on scope type
        if (scope instanceof SelectScope) {
            SelectScope selectScope = (SelectScope) scope;

            // Traverse FROM scope
            if (selectScope.getFromScope() != null) {
                collectDynamicNamespacesFromScope(selectScope.getFromScope(), result);
            }
        } else if (scope instanceof CTEScope) {
            CTEScope cteScope = (CTEScope) scope;

            // CTEs are already included in the children check above
            // But we need to check their subqueries by traversing nested scopes
            // The CTE namespaces themselves contain references to subquery scopes
        } else if (scope instanceof FromScope) {
            FromScope fromScope = (FromScope) scope;

            // FROM scope children are already checked above
            // No additional child scopes to traverse
        } else if (scope instanceof GroupByScope) {
            GroupByScope groupByScope = (GroupByScope) scope;

            // GroupBy scope typically doesn't have child scopes
        } else if (scope instanceof HavingScope) {
            HavingScope havingScope = (HavingScope) scope;

            // Having scope typically doesn't have child scopes
        } else if (scope instanceof OrderByScope) {
            OrderByScope orderByScope = (OrderByScope) scope;

            // OrderBy scope typically doesn't have child scopes
        }

        // Additionally, traverse parent-child scope relationships
        // by checking if any of the namespaces contain nested SELECT statements
        for (gudusoft.gsqlparser.resolver2.model.ScopeChild child : scope.getChildren()) {
            INamespace namespace = child.getNamespace();

            // If this is a SubqueryNamespace, it contains a SELECT with its own scope tree
            if (namespace instanceof gudusoft.gsqlparser.resolver2.namespace.SubqueryNamespace) {
                // Subquery scopes are processed during scope building
                // and would be in statementScopeCache if we tracked them
            }
        }
    }

    // ===== Logging helpers =====

    private void logInfo(String message) {
        TBaseType.log("[TSQLResolver2] " + message, TLog.INFO);
    }

    private void logDebug(String message) {
        TBaseType.log("[TSQLResolver2] " + message, TLog.DEBUG);
    }

    private void logError(String message) {
        TBaseType.log("[TSQLResolver2] " + message, TLog.ERROR);
    }
}