package gudusoft.gsqlparser.sqlenv;

import gudusoft.gsqlparser.EDataType;
import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.TBaseType;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.nodes.TObjectName;
import gudusoft.gsqlparser.nodes.TTypeName;
import gudusoft.gsqlparser.sqlenv.catalog.CatalogStoreProvider;
import gudusoft.gsqlparser.sqlenv.catalog.ICatalogProvider;
import gudusoft.gsqlparser.util.Logger;
import gudusoft.gsqlparser.util.LoggerFactory;
import gudusoft.gsqlparser.util.SQLUtil;

import java.util.ArrayList;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import static gudusoft.gsqlparser.sqlenv.ESQLDataObjectType.*;

/**
 * SQL environment includes the metadata of a list of databases. The typical scenario is there is one database includes
 * some schemas, and each schema includes some tables, views, procedures and etc.
 * <br>
 * <br>Database known as catalog in ANSI SQL.
 * <br>Each catalog including a list of schemas.
 * <br>Each schema including a list of schema objects such as table, procedure, function, trigger and more.
 * <br>
 * Implement your own concrete class derived from this class to get the metadata from a real database.
 * Usually, this is done by querying the INFORMATION_SCHEMA in the {@link #initSQLEnv()} method which should be override
 * in your own class.
 * <br><br>
 *
 */
public abstract class TSQLEnv {

    private static final Logger logger = LoggerFactory.getLogger(TSQLEnv.class);

    public static final Map<EDbVendor, Boolean> columnCollationCaseSensitive;
    
    static {
        columnCollationCaseSensitive = new EnumMap<>(EDbVendor.class);
        columnCollationCaseSensitive.put(EDbVendor.dbvaccess, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvansi, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvathena, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvazuresql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvbigquery, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvclickhouse, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvcouchbase, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvdax, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvdb2, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvexasol, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvfirebird, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvgeneric, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvgreenplum, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvhana, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvhive, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvimpala, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvinformix, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvmdx, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvmysql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvmssql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvnetezza, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvodbc, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvopenedge, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvoracle, true);
        columnCollationCaseSensitive.put(EDbVendor.dbvdameng, true);
        columnCollationCaseSensitive.put(EDbVendor.dbvoceanbase, false);  // mirrors MySQL — default tenant mode is MYSQL
        columnCollationCaseSensitive.put(EDbVendor.dbvpostgresql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvduckdb, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvpresto, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvflink, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvredshift, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvsnowflake, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvsoql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvsparksql, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvsybase, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvteradata, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvtrino, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvvertica, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvdatabricks, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvgaussdb, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvedb, false);
        columnCollationCaseSensitive.put(EDbVendor.dbvdoris, false);  // Doris is MySQL-compatible, case-insensitive
        columnCollationCaseSensitive.put(EDbVendor.dbvstarrocks, false);  // StarRocks is MySQL-compatible, case-insensitive
        columnCollationCaseSensitive.put(EDbVendor.dbvsqlite, false);  // SQLite is case-insensitive for identifiers
        columnCollationCaseSensitive.put(EDbVendor.dbvpowerquery, true);  // M identifiers are case-sensitive

        // Compile-time check to ensure all enum values are covered
        if (columnCollationCaseSensitive.size() != EDbVendor.values().length) {
            throw new IllegalStateException("columnCollationCaseSensitive map must contain all EDbVendor values");
        }
    }

    public static final Map<EDbVendor, Boolean> functionCollationCaseSensitive;
    
    static {
        functionCollationCaseSensitive = new EnumMap<>(EDbVendor.class);
        functionCollationCaseSensitive.put(EDbVendor.dbvaccess, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvansi, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvathena, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvazuresql, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvbigquery, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvclickhouse, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvcouchbase, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvdax, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvdb2, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvexasol, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvfirebird, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvgeneric, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvgreenplum, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvhana, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvhive, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvimpala, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvinformix, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvmdx, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvmysql, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvmssql, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvnetezza, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvodbc, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvopenedge, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvoracle, true);
        functionCollationCaseSensitive.put(EDbVendor.dbvdameng, true);
        functionCollationCaseSensitive.put(EDbVendor.dbvoceanbase, false);  // mirrors MySQL — default tenant mode is MYSQL
        functionCollationCaseSensitive.put(EDbVendor.dbvpostgresql, true);
        functionCollationCaseSensitive.put(EDbVendor.dbvduckdb, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvpresto, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvflink, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvredshift, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvsnowflake, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvsoql, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvsparksql, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvsybase, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvteradata, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvtrino, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvvertica, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvdatabricks, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvgaussdb, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvedb, false);
        functionCollationCaseSensitive.put(EDbVendor.dbvdoris, false);  // Doris is MySQL-compatible, case-insensitive
        functionCollationCaseSensitive.put(EDbVendor.dbvstarrocks, false);  // StarRocks is MySQL-compatible, case-insensitive
        functionCollationCaseSensitive.put(EDbVendor.dbvsqlite, false);  // SQLite is case-insensitive for identifiers
        functionCollationCaseSensitive.put(EDbVendor.dbvpowerquery, true);  // M is case-sensitive

        // Compile-time check to ensure all enum values are covered
        if (functionCollationCaseSensitive.size() != EDbVendor.values().length) {
            throw new IllegalStateException("functionCollationCaseSensitive map must contain all EDbVendor values");
        }
    }
    public static final Map<EDbVendor, Boolean> tableCollationCaseSensitive;
    
    static {
        tableCollationCaseSensitive = new EnumMap<>(EDbVendor.class);
        tableCollationCaseSensitive.put(EDbVendor.dbvaccess, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvansi, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvathena, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvazuresql, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvbigquery, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvclickhouse, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvcouchbase, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvdax, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvdb2, true);
        tableCollationCaseSensitive.put(EDbVendor.dbvexasol, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvfirebird, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvgeneric, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvgreenplum, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvhana, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvhive, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvimpala, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvinformix, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvmdx, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvmysql, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvmssql, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvnetezza, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvodbc, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvopenedge, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvoracle, true);
        tableCollationCaseSensitive.put(EDbVendor.dbvdameng, true);
        tableCollationCaseSensitive.put(EDbVendor.dbvoceanbase, false);  // mirrors MySQL — default tenant mode is MYSQL
        tableCollationCaseSensitive.put(EDbVendor.dbvpostgresql, true);
        tableCollationCaseSensitive.put(EDbVendor.dbvduckdb, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvpresto, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvflink, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvredshift, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvsnowflake, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvsoql, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvsparksql, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvsybase, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvteradata, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvtrino, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvvertica, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvdatabricks, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvgaussdb, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvedb, false);
        tableCollationCaseSensitive.put(EDbVendor.dbvdoris, false);  // Doris is MySQL-compatible, case-insensitive
        tableCollationCaseSensitive.put(EDbVendor.dbvstarrocks, false);  // StarRocks is MySQL-compatible, case-insensitive
        tableCollationCaseSensitive.put(EDbVendor.dbvsqlite, false);  // SQLite is case-insensitive for identifiers
        tableCollationCaseSensitive.put(EDbVendor.dbvpowerquery, true);  // M is case-sensitive

        // Compile-time check to ensure all enum values are covered
        if (tableCollationCaseSensitive.size() != EDbVendor.values().length) {
            throw new IllegalStateException("tableCollationCaseSensitive map must contain all EDbVendor values");
        }
    }

    public static final Map<EDbVendor, Boolean> catalogCollationCaseSensitive;
    
    static {
        catalogCollationCaseSensitive = new EnumMap<>(EDbVendor.class);
        catalogCollationCaseSensitive.put(EDbVendor.dbvaccess, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvansi, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvathena, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvazuresql, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvbigquery, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvclickhouse, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvcouchbase, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvdax, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvdb2, true);
        catalogCollationCaseSensitive.put(EDbVendor.dbvexasol, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvfirebird, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvgeneric, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvgreenplum, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvhana, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvhive, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvimpala, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvinformix, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvmdx, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvmysql, true);
        catalogCollationCaseSensitive.put(EDbVendor.dbvmssql, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvnetezza, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvodbc, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvopenedge, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvoracle, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvdameng, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvoceanbase, true);  // mirrors MySQL — catalog (database) is case-sensitive
        catalogCollationCaseSensitive.put(EDbVendor.dbvpostgresql, true);
        catalogCollationCaseSensitive.put(EDbVendor.dbvduckdb, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvpresto, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvflink, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvredshift, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvsnowflake, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvsoql, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvsparksql, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvsybase, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvteradata, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvtrino, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvvertica, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvdatabricks, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvgaussdb, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvedb, false);
        catalogCollationCaseSensitive.put(EDbVendor.dbvdoris, true);  // Doris is MySQL-compatible, catalog (database) is case-sensitive
        catalogCollationCaseSensitive.put(EDbVendor.dbvstarrocks, true);  // StarRocks is MySQL-compatible, catalog (database) is case-sensitive
        catalogCollationCaseSensitive.put(EDbVendor.dbvsqlite, true);  // SQLite database names (file paths) are case-sensitive on most OS
        catalogCollationCaseSensitive.put(EDbVendor.dbvpowerquery, true);  // M catalog references are case-sensitive

        // Compile-time check to ensure all enum values are covered
        if (catalogCollationCaseSensitive.size() != EDbVendor.values().length) {
            throw new IllegalStateException("catalogCollationCaseSensitive map must contain all EDbVendor values");
        }
    }

    public static final Map<EDbVendor, Boolean> defaultCollationCaseSensitive;
    
    static {
        defaultCollationCaseSensitive = new EnumMap<>(EDbVendor.class);
        defaultCollationCaseSensitive.put(EDbVendor.dbvaccess, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvansi, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvathena, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvazuresql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvbigquery, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvclickhouse, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvcouchbase, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvdax, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvdb2, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvexasol, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvfirebird, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvgeneric, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvgreenplum, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvhana, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvhive, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvimpala, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvinformix, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvmdx, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvmysql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvmssql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvnetezza, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvodbc, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvopenedge, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvoracle, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvdameng, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvoceanbase, false);  // mirrors MySQL — default tenant mode is MYSQL
        defaultCollationCaseSensitive.put(EDbVendor.dbvpostgresql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvduckdb, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvpresto, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvflink, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvredshift, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvsnowflake, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvsoql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvsparksql, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvsybase, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvteradata, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvtrino, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvvertica, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvdatabricks, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvgaussdb, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvedb, false);
        defaultCollationCaseSensitive.put(EDbVendor.dbvdoris, false);  // Doris is MySQL-compatible, case-insensitive
        defaultCollationCaseSensitive.put(EDbVendor.dbvstarrocks, false);  // StarRocks is MySQL-compatible, case-insensitive
        defaultCollationCaseSensitive.put(EDbVendor.dbvsqlite, false);  // SQLite is case-insensitive by default
        defaultCollationCaseSensitive.put(EDbVendor.dbvpowerquery, true);  // M is case-sensitive by default

        // Compile-time check to ensure all enum values are covered
        if (defaultCollationCaseSensitive.size() != EDbVendor.values().length) {
            throw new IllegalStateException("defaultCollationCaseSensitive map must contain all EDbVendor values");
        }
    }



    public static final Map<EDbVendor, Boolean> isAliasReferenceForbidden;
    
    static {
        isAliasReferenceForbidden = new EnumMap<>(EDbVendor.class);
        isAliasReferenceForbidden.put(EDbVendor.dbvaccess, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvansi, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvathena, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvazuresql, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvbigquery, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvclickhouse, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvcouchbase, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvdax, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvdb2, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvexasol, false);  // Exasol disallows forward references
        isAliasReferenceForbidden.put(EDbVendor.dbvfirebird, false);  // Firebird follows standard SQL behavior
        isAliasReferenceForbidden.put(EDbVendor.dbvgeneric, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvgreenplum, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvhana, false);  // SAP HANA follows standard SQL behavior
        isAliasReferenceForbidden.put(EDbVendor.dbvhive, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvimpala, false);  // Impala follows Hive's behavior
        isAliasReferenceForbidden.put(EDbVendor.dbvinformix, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvmdx, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvmysql, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvmssql, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvnetezza, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvodbc, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvopenedge, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvoracle, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvdameng, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvoceanbase, true);  // mirrors MySQL — alias forward reference forbidden
        isAliasReferenceForbidden.put(EDbVendor.dbvpostgresql, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvduckdb, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvpresto, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvflink, true);  // Flink follows standard SQL behavior like SparkSQL
        isAliasReferenceForbidden.put(EDbVendor.dbvredshift, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvsnowflake, false);  // Snowflake disallows column alias references
        isAliasReferenceForbidden.put(EDbVendor.dbvsoql, false);
        isAliasReferenceForbidden.put(EDbVendor.dbvsparksql, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvsybase, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvteradata, false);  // Teradata follows standard SQL behavior
        isAliasReferenceForbidden.put(EDbVendor.dbvtrino, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvvertica, false);  // Vertica disallows forward references
        isAliasReferenceForbidden.put(EDbVendor.dbvdatabricks, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvgaussdb, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvedb, true);
        isAliasReferenceForbidden.put(EDbVendor.dbvdoris, true);  // Doris is MySQL-compatible, alias reference forbidden
        isAliasReferenceForbidden.put(EDbVendor.dbvstarrocks, true);  // StarRocks is MySQL-compatible, alias reference forbidden
        isAliasReferenceForbidden.put(EDbVendor.dbvsqlite, true);  // SQLite follows PostgreSQL behavior, alias reference forbidden in same clause
        isAliasReferenceForbidden.put(EDbVendor.dbvpowerquery, true);  // M step-to-step references are forbidden before binding

        // Compile-time check to ensure all enum values are covered
        if (isAliasReferenceForbidden.size() != EDbVendor.values().length) {
            throw new IllegalStateException("isAliasReferenceForbidden map must contain all EDbVendor values");
        }
    }

    public static String getStmtSeparatorChar(EDbVendor dbVendor){
        String ret = ";";
        switch (dbVendor){
            case dbvoracle:
            case dbvteradata:
            case dbvpostgresql:
            case dbvduckdb:
            case dbvredshift:
            case dbvgreenplum:
                ret = "/";
                break;
            case dbvdameng:
                ret = "/";
                break;
            case dbvdb2:
                ret = "@";
                break;
            case dbvmysql:
                ret = "$";
                break;
            default:
                break;
        }
        return ret;
    }


    /**
     * Whether this database support schema or not?
     *
     * @param dbVendor
     * @return
     */
    public static boolean supportSchema(EDbVendor dbVendor){
        return (!((dbVendor == EDbVendor.dbvmysql)
                ||(dbVendor == EDbVendor.dbvteradata)||(dbVendor == EDbVendor.dbvhive)||(dbVendor == EDbVendor.dbvimpala)
                ));
    }

    /**
     * Whether this database support catalog or not?
     *
     * @param dbVendor
     * @return
     */
    public static boolean supportCatalog(EDbVendor dbVendor) {
//        return (!((dbVendor == EDbVendor.dbvoracle)
//        ));
        return true;
    }

    /**
     * used to delimit a database identifier, such as [ used in SQL Server, ` used in MySQL
     *
     * @param dbVendor
     * @return
     */
    public static String delimitedChar(EDbVendor dbVendor){
        String ret = "\"";
        switch (dbVendor){
            case dbvmssql:
            case dbvazuresql:
                ret = "[";
                break;
            case dbvathena:
            case dbvmysql:
            case dbvbigquery:
            case dbvcouchbase:
            case dbvhive:
            case dbvimpala:
            case dbvdatabricks:
                ret = "`";
                break;
            case dbvdax:
                ret = "'";
                break;
            default:
                break;
        }
        return ret;
    }

    public static boolean isDelimitedIdentifier(EDbVendor dbVendor, String identifier){
        boolean ret = false;
        switch (dbVendor){
            case dbvmssql:
            case dbvazuresql:
                ret = identifier.startsWith("[")||identifier.startsWith("\"")||identifier.startsWith("'");
                break;
            case dbvmysql:
            case dbvbigquery:
            case dbvcouchbase:
            case dbvhive:
            case dbvimpala:
                ret = identifier.startsWith("`");
                break;
            case dbvdax:
                ret = identifier.startsWith("'");
                break;
            default:
                ret = identifier.startsWith("\"");
                break;
        }
        return ret;
    }
    
    public static boolean endsWithDelimitedIdentifier(EDbVendor dbVendor, String identifier){
        boolean ret = false;
        switch (dbVendor){
            case dbvmssql:
            case dbvazuresql:
                ret = identifier.endsWith("]")||identifier.endsWith("\"");
                break;
            case dbvmysql:
            case dbvbigquery:
            case dbvcouchbase:
            case dbvhive:
            case dbvimpala:
                ret = identifier.endsWith("`");
                break;
            case dbvdax:
                ret = identifier.endsWith("'");
                break;
            default:
                ret = identifier.endsWith("\"");
                break;
        }
        return ret;
    }

    public boolean isDelimitedIdentifier(String identifier){
        return TSQLEnv.isDelimitedIdentifier(this.getDBVendor(),identifier);
    }


    public boolean compareColumn(String ident1, String ident2){
        // Route through NameService when flag is enabled
        if (TBaseType.USE_EXT_NAME_MATCHER) {
            return nameService.equals(ESQLDataObjectType.dotColumn, ident1, ident2);
        }
        return compareIdentifier(this.getDBVendor(),ESQLDataObjectType.dotColumn,ident1,ident2);
    }

    public boolean compareTable(String ident1, String ident2){
        // Route through NameService when flag is enabled
        if (TBaseType.USE_EXT_NAME_MATCHER) {
            return nameService.equals(ESQLDataObjectType.dotTable, ident1, ident2);
        }
        return compareIdentifier(this.getDBVendor(),ESQLDataObjectType.dotTable,ident1,ident2);
    }

    public boolean compareIdentifier(ESQLDataObjectType objectType, String ident1, String ident2){
        // Route through NameService when flag is enabled
        if (TBaseType.USE_EXT_NAME_MATCHER) {
            return nameService.equals(objectType, ident1, ident2);
        }

        // Legacy path: use existing logic
        switch (objectType){
            case dotTable:
                return compareTable(ident1,ident2);
            case dotColumn:
                return compareColumn(ident1,ident2);
            default:
                return compareIdentifier(this.getDBVendor(),objectType,ident1,ident2);
        }
    }

    public static boolean compareColumn( EDbVendor dbVendor, TObjectName sourceColumn, TObjectName targetColumn){
        return compareQualifiedColumn(dbVendor,sourceColumn.getColumnNameOnly(),targetColumn.getColumnNameOnly(),
                sourceColumn.getTableString(),targetColumn.getTableString(),
                sourceColumn.getSchemaString(),targetColumn.getSchemaString(),
                sourceColumn.getDatabaseString(),targetColumn.getDatabaseString());
    }

    public static boolean compareTable( EDbVendor dbVendor, TObjectName sourceTable, TObjectName targetTable){
        return compareQualifiedTable(dbVendor,
                sourceTable.getTableString(),targetTable.getTableString(),
                sourceTable.getSchemaString(),targetTable.getSchemaString(),
                sourceTable.getDatabaseString(),targetTable.getDatabaseString()
        );
    }

    public static boolean compareQualifiedTable( EDbVendor dbVendor, String sourceTable, String targetTable, String sourceSchema, String targetSchema, String sourceDatabase, String targetDatabase){
        boolean ret = compareIdentifier(dbVendor,dotTable,sourceTable,targetTable);
        if (!ret) return ret;

        // compare schema
        ret = compareIdentifier(dbVendor,dotSchema,sourceSchema,targetSchema);
        if (!ret) return ret;

        // compare database
        return compareIdentifier(dbVendor,dotCatalog,sourceDatabase,targetDatabase);
    }

    public static boolean compareQualifiedColumn( EDbVendor dbVendor, String sourceColumn, String targetColumn, String sourceTable, String targetTable, String sourceSchema, String targetSchema, String sourceDatabase, String targetDatabase){

        boolean ret = compareIdentifier(dbVendor,dotColumn,sourceColumn,targetColumn);
        if (!ret) return ret;

        return compareQualifiedTable(dbVendor,sourceTable,targetTable,sourceSchema,targetSchema,sourceDatabase,targetDatabase);
    }

    public static boolean compareIdentifier( EDbVendor dbVendor, ESQLDataObjectType objectType, TObjectName source, TObjectName target){
        return compareIdentifier(dbVendor,objectType,source.toString(),target.toString());
    }

    public static boolean compareIdentifier( EDbVendor dbVendor, ESQLDataObjectType objectType, String ident1, String ident2){

//        ident1 = IdentifierService.normalizeStatic(dbVendor,objectType,ident1);
//        ident2 = IdentifierService.normalizeStatic(dbVendor,objectType,ident1);
//
//        return IdentifierService.areEqualStatic(dbVendor,objectType,ident1,ident2);

        ident1 = normalizeIdentifier(dbVendor,objectType,ident1); // IdentifierService.normalizeStatic(dbVendor,objectType,ident1);
        ident2 = normalizeIdentifier(dbVendor,objectType,ident2);  // IdentifierService.normalizeStatic(dbVendor,objectType,ident1);

        boolean collationSensitive = false;
        switch (objectType){
            case dotCatalog:
            case dotSchema:
                collationSensitive = catalogCollationCaseSensitive.get(dbVendor);
                break;
            case dotOraclePackage:
                        case dotProcedure:
                        case dotTrigger:
            case dotTable:
                collationSensitive = tableCollationCaseSensitive.get(dbVendor);
                break;
            case dotFunction:
                collationSensitive = functionCollationCaseSensitive.get(dbVendor);
                break;
            case dotColumn:
                collationSensitive = columnCollationCaseSensitive.get(dbVendor);
                break;
            case dotDblink: // oracle dblink
                collationSensitive = false;
                break;
            default:
                collationSensitive = defaultCollationCaseSensitive.get(dbVendor);
                break;
        }

                if (collationSensitive) {
                        return ident1.equals(ident2);
                } else {
                        return ident1.equalsIgnoreCase(ident2);
                }

    }

    public String normalizeIdentifier(ESQLDataObjectType sqlDataObjectType, String identifier) {
        return this.getIdentifierService().normalize(identifier,sqlDataObjectType);
    }

        /**
         * 1. remove delimited char if it's delimited/quoted identifier
         * 2. change the case of the name in the same way as it saved to the information_schema
         *
         * @param dbVendor
         * @param sqlDataObjectType
         * @param identifier
         * @return
         */
    public static String normalizeIdentifier(EDbVendor dbVendor, ESQLDataObjectType sqlDataObjectType, String identifier){
        String ret;

        if (identifier == null) return identifier;
        if (identifier.length() == 0) return identifier;

        if (TSQLEnv.isDelimitedIdentifier(dbVendor,identifier)){
            ret = TBaseType.getTextWithoutQuoted(identifier);
        }else {
            switch (dbVendor){
                case dbvdb2:
                case dbvoracle:
                    ret =  identifier.toUpperCase();
                    break;
                case dbvdameng:
                    ret =  identifier.toUpperCase();
                    break;
                case dbvpostgresql:
                case dbvduckdb:
                case dbvgreenplum:
                case dbvredshift:
                    ret =  identifier.toLowerCase();
                    break;
                default:
                    ret =  identifier;
                    break;
            }
        }
        return ret;
    }

    /**
     * 比较 一个数据库对象名是否等于或者属于另一个对象
     * 等于 就是完全相等（根据不同数据库的比较规则）
     * 属于 表示如下情况：
     * 1. column1 -&gt; 属于 -&gt; table1.column1
     * 2. table1 -&gt; 属于 -&gt; db1.schema1.table1
     * 3. `schema1.table1` -&gt; 属于 -&gt; `db1`.`schema1`.`table1`
     * 4. `schema1.table1` -&gt; 不属于 -&gt; `db1`.`schema2`.`table1`
     *
     * @param sub
     * @param whole
     * @return
     */
    public static boolean matchSubObjectNameToWhole(EDbVendor dbVendor, ESQLDataObjectType sqlDataObjectType,String sub, String whole){
        List<String> subParts = SQLUtil.parseNames(sub);
        // `data.RETAIL_PROD_EXCEPTIONS_SOURCE` 没有被分开，需要去掉 `` 后再次拆分
        if ((subParts.size() == 1)&&(sub.indexOf(".") != -1)){
            subParts = SQLUtil.parseNames(IdentifierService.normalizeStatic(dbVendor,sqlDataObjectType,sub));
        }

        List<String> wholeParts = SQLUtil.parseNames(whole);
        if ((wholeParts.size() == 1)&&(whole.indexOf(".") != -1)){
            wholeParts = SQLUtil.parseNames(IdentifierService.normalizeStatic(dbVendor,sqlDataObjectType,whole));
        }


        if(subParts.size() >wholeParts.size()) return false;
        int k=0;
        for(String s:subParts){
            subParts.set(k,TBaseType.removePrefixOrSuffixQuoteChar(normalizeIdentifier(dbVendor,sqlDataObjectType,s)));
            k++;
        }
        k=0;
        for(String s:wholeParts){
            wholeParts.set(k, TBaseType.removePrefixOrSuffixQuoteChar(normalizeIdentifier(dbVendor,sqlDataObjectType,s)));
            k++;
        }

        boolean ret = false;
        int i= subParts.size() -1;
        int j = wholeParts.size() -1;
        while (i >= 0){
            if ( !subParts.get(i).toUpperCase().equals(wholeParts.get(j).toUpperCase())) break;
            if (i == 0) ret = true;
            i--;
            j--;
        }

        return ret;
    }

    private boolean enableGetMetadataFromDDL = true;

    public void setEnableGetMetadataFromDDL(boolean enableGetMetadataFromDDL) {
        this.enableGetMetadataFromDDL = enableGetMetadataFromDDL;
    }

    /**
     * If this option is enabled, SQLEnv will collect table/view/function/procedure metadata
     * from the create table/create view/create function/create procedure statement during the parse of the SQL script.
     * <br>A TSQLEnv object instance must be passed to {@link TGSqlParser} before parsing the SQL script. And this TSQLEnv
     * instance can be passed to another {@link TGSqlParser} object with the collected database metadata.
     * <br>Default value is true.
     *
     * @return
     */
    public boolean isEnableGetMetadataFromDDL() {
        return enableGetMetadataFromDDL;
    }

    /**
     * create a SQL environment.
     *
     * @param dbVendor the database vendor
     */
    public TSQLEnv(EDbVendor dbVendor){
        this.dbVendor = dbVendor;
        this.nameService = new NameService(dbVendor);

        // ===== Phase 0: Initialize IdentifierService first (required by CatalogStore) =====
        // Create IdentifierProfile and IdentifierService
        this.identifierProfile = IdentifierProfile.forVendor(
            dbVendor,
            IdentifierProfile.VendorFlags.defaults()  // Default configuration
        );
        this.collatorProvider = createCollatorProvider(dbVendor);  // SQL Server needs this
        this.identifierService = new IdentifierService(identifierProfile, collatorProvider);

        // Phase 3.5: Create CatalogStoreProvider (which internally creates CatalogStore)
        this.catalogProvider = new CatalogStoreProvider(dbVendor, IdentifierProfile.VendorFlags.defaults());
    }

    private EDbVendor dbVendor;

    // Name matching service (Phase 2 integration)
    private final NameService nameService;

    // ===== Phase 3.5: Final catalog implementation =====
    private IdentifierProfile identifierProfile;
    private IdentifierService identifierService;
    private CollatorProvider collatorProvider;  // SQL Server specific
    private ICatalogProvider catalogProvider;   // Uses CatalogStoreProvider

    /**
     * the database vendor where this SQL environment is generated from.
     *
     * @return the database vendor
     */
    public EDbVendor getDBVendor() {
        return dbVendor;
    }

    /**
     * Returns the catalog store (Phase 3.5 integration).
     *
     * <p>Note: This method delegates to the underlying CatalogStoreProvider.
     * Direct access to CatalogStore is provided for backward compatibility
     * with existing test code.</p>
     *
     * @return the catalog store
     */
    public CatalogStore getCatalogStore() {
        if (catalogProvider instanceof CatalogStoreProvider) {
            return ((CatalogStoreProvider) catalogProvider).getCatalogStore();
        }
        return null;
    }

    // ===== Phase 0: Helper methods and getters =====

    /**
     * Create CollatorProvider for SQL Server
     *
     * @param vendor database vendor
     * @return CollatorProvider instance, null for non-SQL Server databases
     */
    private CollatorProvider createCollatorProvider(EDbVendor vendor) {
        if (vendor == EDbVendor.dbvmssql || vendor == EDbVendor.dbvazuresql) {
            return new CollatorProvider();  // SQL Server collation
        }
        return null;
    }

    /**
     * Get IdentifierService (Phase 0)
     *
     * @return IdentifierService instance
     */
    public IdentifierService getIdentifierService() {
        return identifierService;
    }

    /**
     * Get IdentifierProfile (Phase 0)
     *
     * @return IdentifierProfile instance
     */
    public IdentifierProfile getIdentifierProfile() {
        return identifierProfile;
    }

    /**
     * Get ICatalogProvider (Phase 3)
     *
     * @return ICatalogProvider instance
     */
    public ICatalogProvider getCatalogProvider() {
        return catalogProvider;
    }

    /**
     * Catalog-input compatibility hook (Phase 1D).
     *
     * <p>Lets a {@code TSQLEnv} subclass in {@code gudusoft.gsqlparser.sqlenv.compat}
     * install an alternative {@link ICatalogProvider} (typically one backed by a
     * {@code CatalogRuntime}) without reflection or constructor re-entry. See spike
     * memo {@code docs/designs/catalog-input-interface-spikes/tsqlenv-hook.md}.</p>
     *
     * <p>{@code protected final} keeps the surface tight — only subclasses may invoke
     * it, and they cannot override it. The intended caller is
     * {@code CatalogRuntimeToSQLEnvBridge}. End users should continue to use the
     * legacy {@code TSQLEnv} mutation API; this hook is part of the catalog-input
     * compatibility bridge and is not a general extensibility point.</p>
     *
     * @param provider the replacement catalog provider; must not be {@code null}.
     */
    protected final void setCatalogProviderForCompatibility(ICatalogProvider provider) {
        if (provider == null) {
            throw new IllegalArgumentException(
                "TSQLEnv.setCatalogProviderForCompatibility: provider must not be null");
        }
        this.catalogProvider = provider;
    }

    /**
     * This method must be override in the subclass to build a SQL environment with real metadata.
     * <br>this usually done by querying the INFORMATION_SCHEMA
     */
    public abstract void initSQLEnv();

    /**
     * a list of catalog/database in this SQL environment.
     *
     * @return a list of catalog/database
     */
    public List<TSQLCatalog> getCatalogList() {
        return catalogList;
    }

    private String serverName = null; // SQL Server

    private List<TSQLCatalog> catalogList = new CopyOnWriteArrayList<>();

    /**
     * add a catalog to the SQL environment, called internally.
     *
     * @param sqlCatalog catalog
     * @return return false if a catalog with the same name already exists.
     */
    protected boolean doAddCatalog(TSQLCatalog sqlCatalog){
        // same monitor as getSQLCatalog so the check-then-add is atomic under
        // concurrent registration; readers iterate the COW list lock-free
        synchronized (catalogList) {
            boolean isFound = false;
            for(TSQLCatalog c : catalogList){
                if ( sqlCatalog.getName().compareTo(c.getName()) == 0){
                    isFound = true;
                    break;
                }
            }
            if (!isFound){
                catalogList.add(sqlCatalog);
            }
            return !isFound;
        }
    }

    private Map<String, TSQLSchemaObject> schemaObjectList = new ConcurrentHashMap<>();

    // Fast canonical index for fully-qualified lookups (additive, keeps existing map intact)
    private final Map<NameKey, TSQLSchemaObject> objectIndex = new ConcurrentHashMap<NameKey, TSQLSchemaObject>();
    // Reverse index for tables by object name only (speeds up ..table lookups).
    // COW lists: registration appends while other threads' name resolution reads
    private final Map<String, java.util.concurrent.CopyOnWriteArrayList<TSQLTable>> tablesByName =
            new ConcurrentHashMap<String, java.util.concurrent.CopyOnWriteArrayList<TSQLTable>>();

    /**
     * put a schema object into the hashmap which can be used to find a schema object in a more efficient way.
     *
     * @param schemaObjectName  name of schema object
     * @param schemaObject instance of a schema object
     * @return always return true.
     */
    protected boolean putSchemaObject(String schemaObjectName, TSQLSchemaObject schemaObject){
        String newSchemaName = schemaObjectName;
        if (schemaObject.getDataObjectType() == dotTable){
            if (!tableCollationCaseSensitive.get(this.getDBVendor())){
                newSchemaName = newSchemaName.toUpperCase();
            }
        }else {
            if (!defaultCollationCaseSensitive.get(this.getDBVendor())){
                newSchemaName = newSchemaName.toUpperCase();
            }
        }

        if (schemaObject.getDataObjectType() == dotFunction){
            newSchemaName = newSchemaName+"$function";
        }else if (schemaObject.getDataObjectType() == dotProcedure){
            newSchemaName = newSchemaName+"$procedure";
        }

        schemaObjectList.put(newSchemaName,schemaObject);

        // ===== Phase 3.5: Write to catalog provider (no redundancy) =====
        try {
            catalogProvider.addObject(schemaObject);
        } catch (Throwable e) {
            // Log error but maintain backward compatibility
            logger.error("[CatalogProvider] Error adding object: " + schemaObject + ", error=" + e.getMessage(), e);
        }

        // Maintain fast index
        try {
            List<String> parts = SQLUtil.parseNames(schemaObjectName);
            if (parts.size() >= 3) {
                String catalog = this.normalizeIdentifier(ESQLDataObjectType.dotCatalog, parts.get(0));
                String schema = this.normalizeIdentifier(ESQLDataObjectType.dotSchema, parts.get(1));
                String object = this.normalizeIdentifier(
                        (schemaObject.getDataObjectType() == ESQLDataObjectType.dotColumn) ? ESQLDataObjectType.dotTable : schemaObject.getDataObjectType(),
                        SQLUtil.mergeSegments(parts, 2));
                String server = (getDefaultServerName() == null) ? DEFAULT_SERVER_NAME : getDefaultServerName();
                NameKey key = new NameKey(schemaObject.getDataObjectType(), server, catalog, schema, object);
                objectIndex.put(key, schemaObject);

                if (schemaObject instanceof TSQLTable) {
                    String objOnly = this.normalizeIdentifier(ESQLDataObjectType.dotTable, getObjectName(schemaObjectName));
                    // COW list + atomic get-or-create: concurrent registration must not
                    // clobber the list or corrupt it under a concurrent reader
                    tablesByName.computeIfAbsent(objOnly,
                            k -> new java.util.concurrent.CopyOnWriteArrayList<TSQLTable>())
                            .addIfAbsent((TSQLTable) schemaObject);
                }
            }
        } catch (Throwable ignore) {
            // keep backward compatibility even if fast index building fails
        }
        return true;
    }

    /**
     * add a table
     *
     * @param qualifiedTableName, must be in syntax like: catalog.schema.table
     * @param sqlTable, table instance
     */
//    public void addSQLTable(String qualifiedTableName, TSQLTable sqlTable){
//        schemaObjectList.put(qualifiedTableName,sqlTable);
//    }
//
//    public void addRoutine(String qualifiedRoutineName, TSQLRoutine sqlRoutine){
//        schemaObjectList.put(qualifiedRoutineName,sqlRoutine);
//    }

    public static final String DEFAULT_SERVER_NAME = "DEFAULT_SERVER";
    public static final String DEFAULT_DB_NAME = "DEFAULT";
    public static final String DEFAULT_SCHEMA_NAME = "DEFAULT";

    private String defaultCatalogName = null;
    private String defaultSchemaName = null;
    private String defaultServerName = null;

    protected TSQLSchemaObject doSearchSchemaObject(String catalog, String schema, String table, ESQLDataObjectType objectType){
        TSQLSchemaObject result = null;

        String normalizedCurrentCatalogName = this.normalizeIdentifier(ESQLDataObjectType.dotCatalog, defaultCatalogName);
        String normalizedCurrentSchemaName = this.normalizeIdentifier(ESQLDataObjectType.dotSchema, defaultSchemaName);

        if ((catalog.length()>0)&&(schema.length()>0)){ // catalog.schema.table
            result = doSearchSchemaObject(catalog+"."+schema+"."+table,objectType);
        }else if (schema.length()>0){ //.schema.table
            if (defaultCatalogName != null){
                result = doSearchSchemaObject(normalizedCurrentCatalogName+"."+schema+"."+table,objectType);
            }else{
                for(TSQLCatalog c : catalogList){
                    result = doSearchSchemaObject(c.name+"."+schema+"."+table,objectType);
                    if (result != null) break;
                }
            }
        }else if (catalog.length()>0){ // catalog..table
            if (defaultSchemaName != null){
                result = doSearchSchemaObject(catalog+"."+normalizedCurrentSchemaName+"."+table,objectType);
            }else{
                for(TSQLCatalog c : catalogList){
                    if ( c.compareTo(catalog) != 0) continue;
                    for(TSQLSchema s: c.getSchemaList()){
                        result = doSearchSchemaObject(s.getQualifiedName()+"."+table,objectType);
                        if (result != null) break;
                    }
                    if (result != null) break;
                }
            }
        }else{ // ..table, search under the current default database and schema.
            // The current default database and schema can be set by the user.
            // if current default database and schema is not set, then search in all databases and schemas
            if ((objectType == dotTable) && (defaultCatalogName == null) && (defaultSchemaName == null)){
                String canonicalTable = this.normalizeIdentifier(ESQLDataObjectType.dotTable, table);
                List<TSQLTable> candidates = tablesByName.get(canonicalTable);
                if (candidates != null && !candidates.isEmpty()){
                    return candidates.get(0);
                }
            }
            for(TSQLCatalog c : catalogList){
                if ((defaultCatalogName != null) && ( c.compareTo(defaultCatalogName) != 0)) continue;
                for(TSQLSchema s: c.getSchemaList()){
                    if ((defaultSchemaName != null)&&(s.compareTo(defaultSchemaName) != 0)) continue;
                    result = doSearchSchemaObject(s.getQualifiedName()+"."+table,objectType);
                    if (result != null) break;
                }
                if (result != null) break;
            }
        }

        return result;
    }

    protected TSQLSchemaObject searchSchemaObject(TObjectName qualifiedName, ESQLDataObjectType objectType){

        String catalog = this.normalizeIdentifier(ESQLDataObjectType.dotCatalog, qualifiedName.getDatabaseString());
        String schema = this.normalizeIdentifier(ESQLDataObjectType.dotSchema, qualifiedName.getSchemaString());
        String table = this.normalizeIdentifier(ESQLDataObjectType.dotTable, qualifiedName.getTableString());

        schema = (schema == null)?getDefaultSchemaName():schema;
        schema = (schema == null)?"":schema;

        catalog = (catalog == null)?getDefaultCatalogName():catalog;
        catalog = (catalog == null)?"":catalog;

        return doSearchSchemaObject(catalog,schema,table,objectType);
    }

    /**
     * Phase 1: 使用 IdentifierService 的新方法处理多段名
     */
    public TSQLSchemaObject searchSchemaObject(String qualifiedName, ESQLDataObjectType objectType){
        // Phase 1: 使用 IdentifierService 的新方法
        IdentifierService svc = getIdentifierService();

        // 1. 解析多段名
        List<String> parts = svc.parseQualifiedName(qualifiedName);

        // 2. 展开厂商特定语法（MSSQL ".."）
        // Pass the default schema name to avoid relying on global state
        parts = svc.expandVendorSpecific(parts, dbVendor, getDefaultSchemaName());

                if (parts.size() < 3) {
                        return null;
                }

                boolean supportCatalog = TSQLEnv.supportCatalog(dbVendor);
                boolean supportSchema = TSQLEnv.supportSchema(dbVendor);

        // 3. 使用 normalizeSegment 规范化各段（Phase 1: 单段处理）
        String catalog = svc.normalizeSegment(parts.get(0), ESQLDataObjectType.dotCatalog);
        String schema = svc.normalizeSegment(parts.get(1), ESQLDataObjectType.dotSchema);
        String table = svc.normalizeSegment(SQLUtil.mergeSegments(parts, 2), ESQLDataObjectType.dotTable);

        TSQLSchemaObject object = doSearchSchemaObject(catalog,schema,table,objectType);

        //如果不是同时支持database和schema，且没有找到object，交换database和schema顺序查找
                if (!(supportCatalog && supportSchema) && object == null) {
                        catalog = svc.normalizeSegment(parts.get(1), ESQLDataObjectType.dotCatalog);
                        schema = svc.normalizeSegment(parts.get(0), ESQLDataObjectType.dotSchema);
                        object = doSearchSchemaObject(catalog, schema, table, objectType);
                }

                return object;

//        String normalizedCurrentCatalogName = TSQLObject.normalizeIdentifier(this,ESQLDataObjectType.dotCatalog, defaultCatalogName);
//        String normalizedCurrentSchemaName = TSQLObject.normalizeIdentifier(this,ESQLDataObjectType.dotSchema, defaultSchemaName);
//
//        if ((catalog.length()>0)&&(schema.length()>0)){ // catalog.schema.table
//            result = doSearchSchemaObject(catalog+"."+schema+"."+table,objectType);
//        }else if (schema.length()>0){ //.schema.table
//            if (defaultCatalogName != null){
//                result = doSearchSchemaObject(normalizedCurrentCatalogName+"."+schema+"."+table,objectType);
//            }else{
//                for(TSQLCatalog c : catalogList){
//                    result = doSearchSchemaObject(c.name+"."+schema+"."+table,objectType);
//                    if (result != null) break;
//                }
//            }
//        }else if (catalog.length()>0){ // catalog..table
//            if (defaultSchemaName != null){
//                result = doSearchSchemaObject(catalog+"."+normalizedCurrentSchemaName+"."+table,objectType);
//            }else{
//                for(TSQLCatalog c : catalogList){
//                    if ( c.compareTo(catalog) != 0) continue;
//                    for(TSQLSchema s: c.getSchemaList()){
//                        result = doSearchSchemaObject(s.getQualifiedName()+"."+table,objectType);
//                        if (result != null) break;
//                    }
//                    if (result != null) break;
//                }
//            }
//        }else{ // ..table
//            for(TSQLCatalog c : catalogList){
//                if ((defaultCatalogName != null) && ( c.compareTo(defaultCatalogName) != 0)) continue;
//                for(TSQLSchema s: c.getSchemaList()){
//                    if ((defaultSchemaName != null)&&(s.compareTo(defaultSchemaName) != 0)) continue;
//                    result = doSearchSchemaObject(s.getQualifiedName()+"."+table,objectType);
//                    if (result != null) break;
//                }
//                if (result != null) break;
//            }
//        }
//
//        return result;

    }
    /**
     * find a table in the SQL environment by using a qualified table name: catalogName.schemaName.tableName
     *
     * @param qualifiedTablename, can be catalog.schema.table,
     *                            or .schema.table, use currentCatalogName or iterate all catalogs
     *                            or catalog..table, use current schema name or iterate all schema under catalog
     *                            or ..table, use currentCatalogName or iterate all catalogs, use current schema
     *                                          or iterate all schema
     * @return a table
     */
    public TSQLTable searchTable(String qualifiedTablename){
        TSQLSchemaObject result = searchSchemaObject(qualifiedTablename,dotTable);

        if (result instanceof TSQLTable){
            return (TSQLTable)result;
        }

        // The name may be a synonym; dereference it to its base table/view so a
        // synonym reference follows through to its base for column lineage.
        TSQLSchemaObject synonym = searchSchemaObject(qualifiedTablename, ESQLDataObjectType.dotSynonyms);
        if (synonym instanceof TSQLSynonyms){
            return ((TSQLSynonyms) synonym).resolveBaseTable();
        }
        return null;
    }

    public TSQLTable searchTable(TObjectName tableName){
        if ((tableName.getSchemaToken() == null) && (tableName.getDatabaseToken() == null)) return searchTable(".."+tableName.getTableString());
        if (tableName.getDatabaseToken() == null) return searchTable("."+tableName.getSchemaString()+"."+tableName.getTableString());
        if (tableName.getSchemaToken() == null) return searchTable(tableName.getDatabaseString()+".."+tableName.getTableString());
        return searchTable(tableName.toString());
    }
    /**
     * called by {@link #searchTable(String)} method internally.
     *
     * @param qualifiedName table name
     * @return return a table instance if found, otherwise, return null
     */
//    TSQLTable doSearchTable(String qualifiedTablename){
//        TSQLTable result = null;
//        TSQLSchemaObject schemaObject = schemaObjectList.get(qualifiedTablename);
//
//        if (schemaObject instanceof TSQLTable){
//            result = (TSQLTable)schemaObject;
//        }
//        return result;
//    }

    /**
     * Phase 1: 使用 IdentifierService 新方法 + 双写日志
     */
    private TSQLSchemaObject doSearchSchemaObject( String qualifiedName, ESQLDataObjectType objectType){
        // Phase 1: 使用 IdentifierService
        IdentifierService svc = getIdentifierService();

        // ===== Phase 3: Try catalog provider first =====
        try {
            List<String> parts = svc.parseQualifiedName(qualifiedName);
            parts = svc.expandVendorSpecific(parts, dbVendor);

            if (parts.size() >= 3) {
                // Phase 1: 使用 normalizeSegment 规范化单段名
                String catalog = svc.normalizeSegment(parts.get(0), ESQLDataObjectType.dotCatalog);
                String schema = svc.normalizeSegment(parts.get(1), ESQLDataObjectType.dotSchema);
                String object = svc.normalizeSegment(SQLUtil.mergeSegments(parts, 2), objectType);

                TSQLSchemaObject result = catalogProvider.findObject(catalog, schema, object, objectType);
                if (result != null) {
                    return result;
                }
            }
        } catch (Throwable e) {
            // Log error but continue to fallback paths
            // System.err.println("[CatalogProvider] Search error: " + e.getMessage());
        }

        // Fast-path using canonical index when fully qualified
        try {
            List<String> parts = svc.parseQualifiedName(qualifiedName);
            parts = svc.expandVendorSpecific(parts, dbVendor);

            if (parts.size() >= 3) {
                String server = (getDefaultServerName() == null) ? DEFAULT_SERVER_NAME : getDefaultServerName();

                // Phase 1: 使用新方法规范化
                String catalogNew = svc.normalizeSegment(parts.get(0), ESQLDataObjectType.dotCatalog);
                String schemaNew = svc.normalizeSegment(parts.get(1), ESQLDataObjectType.dotSchema);
                String objectNew = svc.normalizeSegment(SQLUtil.mergeSegments(parts, 2),
                        (objectType == ESQLDataObjectType.dotColumn) ? ESQLDataObjectType.dotTable : objectType);

                // Phase 1: 双写日志 - 比较新旧键
                if (TBaseType.LOG_KEY_COMPARISON) {
                    String catalogOld = this.normalizeIdentifier(ESQLDataObjectType.dotCatalog, parts.get(0));
                    String schemaOld = this.normalizeIdentifier(ESQLDataObjectType.dotSchema, parts.get(1));
                    String objectOld = this.normalizeIdentifier(
                            (objectType == ESQLDataObjectType.dotColumn) ? ESQLDataObjectType.dotTable : objectType,
                            SQLUtil.mergeSegments(parts, 2));

                    if (!catalogNew.equals(catalogOld) || !schemaNew.equals(schemaOld) || !objectNew.equals(objectOld)) {
                        System.err.println("[Phase1] Key mismatch in " + qualifiedName + " (" + objectType + ")");
                        System.err.println("  catalog: OLD=" + catalogOld + " NEW=" + catalogNew);
                        System.err.println("  schema:  OLD=" + schemaOld + " NEW=" + schemaNew);
                        System.err.println("  object:  OLD=" + objectOld + " NEW=" + objectNew);
                    }
                }

                NameKey key = new NameKey(objectType, server, catalogNew, schemaNew, objectNew);
                TSQLSchemaObject hit = objectIndex.get(key);
                if (hit != null) {
                    return hit;
                }
            }
        } catch (Throwable ignore) {
            // fallback to legacy map below
        }
        String newSchemaName = qualifiedName;
        if (objectType == dotTable){
            if (!tableCollationCaseSensitive.get(this.getDBVendor())){
                newSchemaName = newSchemaName.toUpperCase();
            }
        }else {
            if (!defaultCollationCaseSensitive.get(this.getDBVendor())){
                newSchemaName = newSchemaName.toUpperCase();
            }
        }

        if (objectType == dotFunction){
            newSchemaName = newSchemaName+"$function";
        }else if (objectType == dotProcedure){
            newSchemaName = newSchemaName+"$procedure";
        }
        return  schemaObjectList.get(newSchemaName);
    }

    /**
     * the current active catalog/database in the SQL environment.
     * If search a table in syntax like this:  .schemaName.tableName and this current catalog name is not null,
     * then, search the table in this syntax: currentCatalogName.schemaName.tableName.
     * If the current catalog name is null, search all catalogs in the catalog list.
     *
     * @return the name of the current active catalog/database
     */
    public String getDefaultCatalogName() {
        return defaultCatalogName;
    }

    /**
     * the current schema name in the SQL environment.
     * <br>
     * If search a table in syntax like this:  catalogname..tableName and this current schema name is not null,
     * then, search the table in this syntax: catalogName.currentSchemaName.tableName.
     * If the current schema name is null, search all schemas under catalogName
     *
     * @return the default schema name
     */
    public String getDefaultSchemaName() {
//        if ((dbVendor == EDbVendor.dbvmssql)&&(defaultSchemaName == null)){
//            return "dbo";
//        }
        return defaultSchemaName;
    }

    public void setDefaultCatalogName(String defaultCatalogName) {
        this.defaultCatalogName = defaultCatalogName;
    }

    public void setDefaultSchemaName(String defaultSchemaName) {
        this.defaultSchemaName = defaultSchemaName;
    }

    public String getDefaultServerName() {
        return defaultServerName;
    }

    public void setDefaultServerName(String defaultServerName) {
        this.defaultServerName = defaultServerName;
    }



    /**
     * create a new catalog/database in the SQL environment.
     *
     * @param catalogName catalog name
     * @return instance of the created catalog
     */
    public TSQLCatalog createSQLCatalog(String catalogName){
        return getSQLCatalog(catalogName,true);
    }

    /**
     * get a catalog from the SQL environment if already exists, otherwise, create a new catalog.
     *
     * @param catalogName catalog name
     * @param createIfNotExist if this value is true, then create a new catalog if it's not exists
     * @return a catalog instance
     */
    public TSQLCatalog getSQLCatalog(String catalogName, boolean createIfNotExist){
        TSQLCatalog result = searchCatalog(catalogName);
        if ((result == null)&&(createIfNotExist)){
            // synchronized so concurrent get-or-create of the same catalog can't
            // produce duplicate twins (searchCatalog stays lock-free on the COW list)
            synchronized (catalogList) {
                result = searchCatalog(catalogName);
                if (result == null){
                    result = new TSQLCatalog(this,catalogName);
                }
            }
        }
        return result;
    }

    /**
     * search catalog in the catalog list, return null if not found.
     *
     * @param catalogName catalog name
     * @return null if not found.
     */
    public TSQLCatalog searchCatalog(String catalogName){
        TSQLCatalog result = null;
        for(TSQLCatalog c : catalogList){
            if (c.compareTo(catalogName)==0){
                result = c;
                break;
            }
        }
        return result;
    }

    /**
     * create a new schema and add to the catalog
     *
     * @param qualifiedSchemaName must be a qualified name like: catalog.schema. Otherwise, a null exception will be raised.
     * @return a new schema instance
     */
    public TSQLSchema createSQLSchema(String qualifiedSchemaName){
        return  getSQLSchema(qualifiedSchemaName,true);
    }

    /**
     * get a schema from the specified catalog, if not exists, create a new schema in the catalog.
     *
     * @param qualifiedSchemaName must be a qualified name like: catalog.schema. Otherwise, a null exception will be raised.
     * @param createIfNotExist  if this value is true, then create a new schema if it's not exists
     * @return a schema instance
     */
    public TSQLSchema getSQLSchema(String qualifiedSchemaName, boolean createIfNotExist){
        TSQLSchema result = null;
        String[] parts = SQLUtil.parseNames(qualifiedSchemaName).toArray(new String[0]);
        String catalogName = parts[0];
        String schemaName = parts[1];
        TSQLCatalog catalog = getSQLCatalog(catalogName,createIfNotExist);
        return  catalog.getSchema(schemaName,createIfNotExist);
    }


    /**
     *
     * @param qualifiedTablename 需要完整的tablename，例如 catalog.schema.table, 如果仅传入 table name，
     *                           需要利用 default catalog, default schema 拼接完整的名称： catalog.schema.table
     * @param columnNameOnly
     * @return
     */
    public  ArrayList<String> getColumnsInTable(String qualifiedTablename,boolean columnNameOnly){

        //TSQLTable tsqlTable = searchTable(getAFullQualifiedSchemaObjectName(qualifiedTablename));
        TSQLTable tsqlTable = searchTable(qualifiedTablename);
        if (tsqlTable == null) return null;
        //return tsqlTable.searchColumn(columnName);
        return  tsqlTable.getColumns(columnNameOnly);
    }

    public boolean columnInTable(String qualifiedTablename, String columnName){
        TSQLTable tsqlTable = searchTable(qualifiedTablename);
        if (tsqlTable == null) return false;
        return tsqlTable.searchColumn(columnName);
    }

    public TSQLColumn getColumnInTable(String qualifiedTablename, String columnName){
        StringBuilder builder = new StringBuilder();
        String db = (getDatabaseName(qualifiedTablename) == null)?getDefaultCatalogName():getDatabaseName(qualifiedTablename);
        if((db == null) || (db.length() == 0)) {
                builder.append(DEFAULT_DB_NAME).append(".");
        }
        
        String schema = (getSchemaName(qualifiedTablename) == null)?getDefaultSchemaName():getSchemaName(qualifiedTablename);
        if ((schema == null)||(schema.length() == 0)) {
                builder.append(DEFAULT_SCHEMA_NAME).append(".");
        }
        
        builder.append(qualifiedTablename);
        
        TSQLTable tsqlTable = searchTable(builder.toString());
        if (tsqlTable == null) return null;
        return tsqlTable.getColumn(columnName);
    }

//    String[] getCatalogSchemaNameWithDefault(String qualifiedObjectName){
//        String[] catalogSchemaName = new String[2];
//
//        if (qualifiedObjectName.startsWith("`") && qualifiedObjectName.endsWith("`")){
//            qualifiedObjectName = qualifiedObjectName.substring(1,qualifiedObjectName.length()-1);
//        }
//
//        String db = (getDatabaseName(qualifiedObjectName) == null)?getDefaultCatalogName():getDatabaseName(qualifiedObjectName);
//        if((db == null) || (db.length() == 0)) db = DEFAULT_DB_NAME;
//        String schema = (getSchemaName(qualifiedObjectName) == null)?getDefaultSchemaName():getSchemaName(qualifiedObjectName);
//        if ((schema == null)||(schema.length() == 0)) schema = DEFAULT_SCHEMA_NAME;
//
//        catalogSchemaName[0] = db;
//        catalogSchemaName[1] = schema;
//
//        return  catalogSchemaName;
//    }
    public TSQLSchemaObject doAddSchemaObject(String qualifiedObjectName,ESQLDataObjectType objectType){

        if (qualifiedObjectName.startsWith("`") && qualifiedObjectName.endsWith("`")){
            qualifiedObjectName = SQLUtil.trimColumnStringQuote(qualifiedObjectName);
        }

        String db = (getDatabaseName(qualifiedObjectName) == null)?getDefaultCatalogName():getDatabaseName(qualifiedObjectName);
        if((db == null) || (db.length() == 0)) db = DEFAULT_DB_NAME;
        String schema = (getSchemaName(qualifiedObjectName) == null)?getDefaultSchemaName():getSchemaName(qualifiedObjectName);
        if ((schema == null)||(schema.length() == 0)) schema = DEFAULT_SCHEMA_NAME;

        // getCatalogSchemaNameWithDefault 这个函数有问题，为导致测试用例失败
        //        String[] catalogSchemaName = getCatalogSchemaNameWithDefault(qualifiedObjectName);
//        String db = catalogSchemaName[0];
//        String schema = catalogSchemaName[1];

        TSQLCatalog sqlCatalog = createSQLCatalog(db);
        TSQLSchema sqlSchema = sqlCatalog.createSchema(schema);
        TSQLSchemaObject created = sqlSchema.createSchemaObject(getObjectName(qualifiedObjectName),objectType);
        // Ensure fast index contains the new object as well (already handled via putSchemaObject called by schema)
        return created;
    }

    public TSQLSchemaObject doAddSchemaObject(TObjectName qualifiedObjectName,ESQLDataObjectType objectType){
        String db = (getDatabaseName(qualifiedObjectName) == null)?getDefaultCatalogName():getDatabaseName(qualifiedObjectName);
        if((db == null) || (db.length() == 0)) db = DEFAULT_DB_NAME;
        String schema = (getSchemaName(qualifiedObjectName) == null)?getDefaultSchemaName():getSchemaName(qualifiedObjectName);
        if ((schema == null)||(schema.length() == 0)) schema = DEFAULT_SCHEMA_NAME;
        TSQLCatalog sqlCatalog = createSQLCatalog(db);
        TSQLSchema sqlSchema = sqlCatalog.createSchema(schema);
        return sqlSchema.createSchemaObject(getObjectName(qualifiedObjectName),objectType);
    }

    /**
     * Add a function to SQLEnv, if a function with the same already exists, just return the existing one.
     *
     * @param qualifiedFunctionName
     * @return function
     */
    public TSQLFunction addFunction(String qualifiedFunctionName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLFunction)doAddSchemaObject(qualifiedFunctionName,ESQLDataObjectType.dotFunction);
    }

    public TSQLFunction addFunction(TObjectName qualifiedFunctionName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLFunction)doAddSchemaObject(qualifiedFunctionName,ESQLDataObjectType.dotFunction);
    }

    public TSQLProcedure addOraclePackage(String qualifiedProcedureName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLProcedure)doAddSchemaObject(qualifiedProcedureName, ESQLDataObjectType.dotOraclePackage);
    }

    public TSQLProcedure addProcedure(String qualifiedProcedureName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLProcedure)doAddSchemaObject(qualifiedProcedureName,ESQLDataObjectType.dotProcedure);
    }

    public TSQLRoutine addSQLRoutine(String qualifiedProcedureName, boolean fromDDL, ESQLDataObjectType type){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLRoutine)doAddSchemaObject(qualifiedProcedureName,type);
    }

    public TSQLTrigger addTrigger(String qualifiedTriggerName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLTrigger)doAddSchemaObject(qualifiedTriggerName,ESQLDataObjectType.dotTrigger);
    }

    /**
     * Add a new table to the SQLEnv. If the table with the same name already exists in the SQLEnv,
     * the existing table will be returned.
     * <br>dbName.schemaName.tablename. If dbName is not specified, {@link #getDefaultCatalogName()} will be used to put this table in.
     * If schemaName is not specified, {@link #getDefaultSchemaName()} will be used to put this table in.
     * <br>If the specified database and schema are not already exists, a new database/schema will be generated automatically.
     *
     * @param qualifiedTableName qualified table name
     * @param fromDDL is this table generated from a create table statement
     *
     * @return SQL Table
     */
    public TSQLTable addTable(String qualifiedTableName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        return (TSQLTable)doAddSchemaObject(qualifiedTableName,ESQLDataObjectType.dotTable);
    }

    public TSQLTable addView(String qualifiedViewName, boolean fromDDL){
        if ((fromDDL)&&(!isEnableGetMetadataFromDDL())) return  null;
        TSQLTable tsqlTable = (TSQLTable)doAddSchemaObject(qualifiedViewName,ESQLDataObjectType.dotTable);
        // doAddSchemaObject returns null on an object type/name conflict
        if (tsqlTable != null) {
            tsqlTable.setView(true);
        }
        return tsqlTable;
    }

    public static String getObjectName(TObjectName schemaObjectName){
        if (schemaObjectName.getObjectString().length() == 0) return null;
        else
          return schemaObjectName.getObjectString();
    }

    public static String getObjectName(String schemaObjectName){
        String[] names = SQLUtil.parseNames(schemaObjectName).toArray(new String[0]);
        return names[names.length-1];
    }

    public static String getDatabaseName(TObjectName schemaObjectName){
        if (schemaObjectName.getDatabaseString().length() == 0)
            return null;
        else
           return schemaObjectName.getDatabaseString();
    }

    public static String getDatabaseName(String schemaObjectName){
        String[] names = SQLUtil.parseNames(schemaObjectName).toArray(new String[0]);
        if (names.length == 3){
            return names[0];
        }else if (names.length == 4){
            return names[1];
        }else{
            return null;
        }
    }

    public static String getSchemaName(TObjectName schemaObjectName){
        if (schemaObjectName.getSchemaString().length() == 0) return null;
        else
          return schemaObjectName.getSchemaString();
    }

    public static String getSchemaName(String schemaObjectName){
        String[] names = SQLUtil.parseNames(schemaObjectName).toArray(new String[0]);
        if (names.length == 2){
            return names[0];
        }else if (names.length == 3){
            return names[1];
        }else if (names.length == 4){
            return names[2];
        }else{
            return null;
        }
    }

    public TSQLFunction searchFunction(TObjectName qualifiedTablename) {
        TSQLSchemaObject result = searchSchemaObject(qualifiedTablename,dotFunction);

        if (result instanceof TSQLFunction) {
            return (TSQLFunction) result;
        } else return null;
    }

    public TSQLFunction searchFunction(String qualifiedTablename) {
        TSQLSchemaObject result = searchSchemaObject( getAFullQualifiedSchemaObjectName(qualifiedTablename),dotFunction);

        if (result instanceof TSQLFunction) {
            return (TSQLFunction) result;
        } else return null;
    }

    /**
     * If the input name is fully qualified with db and schema, just return it without any modification.
     * If the schema name is missed, use {@link #getDefaultSchemaName()} instead
     * If the database name is missed, use {@link #getDefaultCatalogName()} instead
     *
     * some sample result:
     * <br>
     * db.schema.table
     * .schema.table,  {@link #getDefaultCatalogName()} is null
     * ..table, both {@link #getDefaultCatalogName()} and {@link #getDefaultSchemaName()} are null.
     *
     * @param schemaObjectName
     * @return
     */
    public String getAFullQualifiedSchemaObjectName(String schemaObjectName){
        if (SQLUtil.parseNames(schemaObjectName).size() == 3) return schemaObjectName;
        String schema = (getDefaultSchemaName() == null)?"":getDefaultSchemaName();
        String db = (getDefaultCatalogName() == null)?"":getDefaultCatalogName();

        String[] names = SQLUtil.parseNames(schemaObjectName).toArray(new String[0]);
        if (names.length == 1){ // no prefixed schema, db
            return db+"."+schema+"."+schemaObjectName;
        }else if (names.length == 2){ // prefix with schema, no database
            return db+"."+names[0]+"."+names[1];
        }else {
            return schemaObjectName;
        }
    }



    public int getNumberOfTables(){
        int numOfTables = 0;
        for(int i=0;i<getCatalogList().size();i++){
            TSQLCatalog tsqlCatalog = getCatalogList().get(i);
            for(TSQLSchema tsqlSchema: tsqlCatalog.getSchemaList()){
                for(TSQLSchemaObject schemaObject: tsqlSchema.getSchemaObjectList()){
                    switch (schemaObject.getDataObjectType()){
                        case dotTable:
                            numOfTables++;
                            break;
                        case dotProcedure:
                            break;
                        case dotFunction:
                            break;
                        case dotOraclePackage:
                            break;
                    }
                } //schema object list
            } // schema list
        } //catalog list
        return numOfTables;
    }

    public String toString(){
        String lcResult = "";
        StringBuilder tables = new StringBuilder();
        StringBuilder views = new StringBuilder();
        StringBuilder functions = new StringBuilder();
        StringBuilder oraclePackages = new StringBuilder();
        StringBuilder procedures = new StringBuilder();
        for(int i=0;i<getCatalogList().size();i++){
            TSQLCatalog tsqlCatalog = getCatalogList().get(i);
            lcResult = lcResult+"\ndatabase:"+tsqlCatalog.getName();
            for(TSQLSchema tsqlSchema: tsqlCatalog.getSchemaList()){
                lcResult = lcResult+"\n\t"+"schema:"+tsqlSchema.getName();
                for(TSQLSchemaObject schemaObject: tsqlSchema.getSchemaObjectList()){
                    switch (schemaObject.getDataObjectType()){
                        case dotTable:
                            TSQLTable tsqlTable = (TSQLTable)schemaObject;
                            if (tsqlTable.isView()){
                                views.append("\n\t\t\t"+schemaObject.getName());
                                for(TSQLColumn column: tsqlTable.getColumnList()){
                                    views.append("\n\t\t\t\t"+column.getName());

                                }
                            }else{
                                tables.append("\n\t\t\t"+schemaObject.getName());
                                for(TSQLColumn column: tsqlTable.getColumnList()){
                                    tables.append("\n\t\t\t\t"+column.getName());
                                    if (column.getColumnDataType() != null){
                                        tables.append(",\t"+column.getColumnDataType().toString());
                                        TTypeName subType;
                                        switch (column.getColumnDataType().getDataType()){
                                            case array_t:
                                                subType = column.getColumnDataType().getTypeOfList();

                                                if (subType.getDataType() == EDataType.struct_t){

                                                    for(int k=0;k<subType.getColumnDefList().size();k++){
                                                        //System.out.println(subType.getColumnDefList().getColumn(k).getColumnName());
                                                        //System.out.println(subType.getColumnDefList().getColumn(k).getDatatype());
                                                        tables.append("\n\t\t\t\t\t\t"+ subType.getColumnDefList().getColumn(k).getColumnName().toString()
                                                                +",\t"+ subType.getColumnDefList().getColumn(k).getDatatype().toString());
                                                    }
                                                }
                                                break;
                                            case struct_t:
                                                subType = column.getColumnDataType();
                                                for(int k=0;k<subType.getColumnDefList().size();k++){
                                                    tables.append("\n\t\t\t\t\t\t"+ subType.getColumnDefList().getColumn(k).getColumnName().toString()
                                                            +",\t"+ subType.getColumnDefList().getColumn(k).getDatatype().toString());
                                                }
                                                break;
                                        }
                                    }
                                }
                            }

                            break;
                        case dotProcedure:
                            procedures.append("\n\t\t\t"+schemaObject.getName());
                            break;
                        case dotFunction:
                            functions.append("\n\t\t\t"+schemaObject.toString());
                            //TSQLFunction f = (TSQLFunction) schemaObject;

                            break;
                        case dotOraclePackage:
                            oraclePackages.append("\n\t\t\t"+schemaObject.getName());
                            break;
                    }
                } //schema object list
                lcResult = lcResult+"\n\t\t"+"tables:"+tables.toString();
                lcResult = lcResult+"\n\t\t"+"views:"+views.toString();
                if(oraclePackages.length()>0) {
                    lcResult = lcResult + "\n\t\t" + "oracle package:" + oraclePackages.toString();
                }
                lcResult = lcResult+"\n\t\t"+"procedure:"+procedures.toString();
                lcResult = lcResult+"\n\t\t"+"function:"+functions.toString();
                tables.setLength(0);
                views.setLength(0);
                procedures.setLength(0);
                functions.setLength(0);
            } // schema list
        } //catalog list
        return lcResult;
    }

    public String getProcedureParameterValue(String procedureName, int paramPos){
        return "";
    }
    public void getVariableValue(String variableName, String variableValue){}

    private static boolean usedBySqlflow = false;

    public static boolean isUsedBySqlflow() {
        return usedBySqlflow;
    }
}