src/databao_context_engine/plugins/databases/duckdb_introspector.py [152:295]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        return r"""
            WITH d AS (
                SELECT 
                    *
                FROM 
                    duckdb_constraints()
                WHERE 
                    schema_name = {SCHEMA}
                    AND constraint_type = 'UNIQUE'
            ),
            cols AS (
                SELECT
                    d.table_name,
                    d.constraint_name,
                    r.pos AS position,
                    d.constraint_column_names[r.pos] AS column_name
                FROM 
                    d,
                    range(1, length(d.constraint_column_names) + 1) AS r(pos)
            )
            SELECT
                table_name,
                constraint_name,
                position,
                column_name
            FROM 
                cols
            ORDER BY
                table_name, 
                constraint_name, 
                position;
        """

    def _sql_checks(self) -> str:
        return r"""
            SELECT
                d.table_name,
                d.constraint_name,
                d.expression        AS expression,
                TRUE                AS validated
            FROM 
                duckdb_constraints() AS d
            WHERE 
                d.schema_name = {SCHEMA}
                AND d.constraint_type = 'CHECK'
            ORDER BY 
                d.table_name, 
                d.constraint_name; 
           """

    def _sql_foreign_keys(self) -> str:
        return r"""
            WITH d AS (
                SELECT 
                    *
                FROM 
                    duckdb_constraints()
                WHERE 
                    schema_name = {SCHEMA}
                    AND constraint_type = 'FOREIGN KEY'
            ),
            cols AS (
                SELECT
                    d.schema_name,
                    d.table_name,
                    d.constraint_name,
                    r.pos AS position,
                    d.constraint_column_names[r.pos]  AS from_column,
                    d.referenced_column_names[r.pos]  AS to_column
                FROM 
                    d,
                    range(1, length(d.constraint_column_names) + 1) AS r(pos)
            ),
            ref AS (
            SELECT
                rc.constraint_schema AS schema_name,
                rc.constraint_name,
                tc.table_schema AS ref_schema,
                tc.table_name   AS ref_table
            FROM 
                information_schema.referential_constraints rc
                JOIN information_schema.table_constraints tc ON 
                    tc.constraint_schema = rc.unique_constraint_schema 
                    AND tc.constraint_name = rc.unique_constraint_name
            ),
            rules AS (
                SELECT
                    constraint_schema AS schema_name,
                    constraint_name,
                    lower(update_rule) AS on_update,
                    lower(delete_rule) AS on_delete
                FROM 
                    information_schema.referential_constraints
            )
            SELECT
                c.table_name,
                c.constraint_name,
                c.position,
                c.from_column,
                r.ref_schema,
                r.ref_table,
                c.to_column,
                coalesce(u.on_update, 'no action') AS on_update,
                coalesce(u.on_delete, 'no action') AS on_delete,
                TRUE AS enforced,
                TRUE AS validated
            FROM 
                cols c JOIN ref r ON r.schema_name = c.schema_name AND r.constraint_name = c.constraint_name
            LEFT JOIN rules u ON u.schema_name = c.schema_name AND u.constraint_name = c.constraint_name
            ORDER BY 
                c.table_name, 
                c.constraint_name, 
                c.position;
        """

    def _sql_indexes(self) -> str:
        return r"""
            WITH idx AS (
                SELECT
                    table_name,
                    index_name,
                    is_unique,
                    string_split(trim(BOTH '[]' FROM expressions), ',') AS expr_list
                FROM 
                    duckdb_indexes()
                WHERE 
                    schema_name = {SCHEMA}
            )
            SELECT
                table_name,
                index_name,
                pos AS position,
                trim(expr_list[pos]) AS expr,
                is_unique
            FROM 
                idx,
                range(1, length(expr_list) + 1) AS r(pos)
            ORDER BY
                table_name,
                index_name,
                position;
         """

    def _sql_sample_rows(self, catalog: str, schema: str, table: str, limit: int) -> SQLQuery:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



src/databao_context_engine/plugins/databases/mysql_introspector.py [182:266]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        return r"""
            SELECT
                tc.TABLE_NAME         AS table_name,
                tc.CONSTRAINT_NAME    AS constraint_name,
                kcu.COLUMN_NAME       AS column_name,
                kcu.ORDINAL_POSITION  AS position
            FROM 
                INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc
                JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu ON kcu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME AND kcu.TABLE_SCHEMA = tc.TABLE_SCHEMA AND kcu.TABLE_NAME = tc.TABLE_NAME
            WHERE 
                tc.TABLE_SCHEMA = {SCHEMA}
                AND tc.CONSTRAINT_TYPE = 'UNIQUE'
            ORDER BY 
                tc.TABLE_NAME, 
                tc.CONSTRAINT_NAME, 
                kcu.ORDINAL_POSITION
        """

    def _sql_checks(self) -> str:
        return r"""
            SELECT
                tc.TABLE_NAME        AS table_name,
                tc.CONSTRAINT_NAME   AS constraint_name,
                cc.CHECK_CLAUSE      AS expression,
                TRUE                 AS validated
            FROM 
                INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc
                JOIN INFORMATION_SCHEMA.CHECK_CONSTRAINTS cc ON cc.CONSTRAINT_SCHEMA = tc.TABLE_SCHEMA AND cc.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
            WHERE 
                tc.TABLE_SCHEMA = {SCHEMA}
                AND tc.CONSTRAINT_TYPE = 'CHECK'
            ORDER BY 
                tc.TABLE_NAME, 
                tc.CONSTRAINT_NAME
        """

    def _sql_foreign_keys(self) -> str:
        return r"""
            SELECT
                kcu.TABLE_NAME                 AS table_name,
                kcu.CONSTRAINT_NAME            AS constraint_name,
                kcu.ORDINAL_POSITION           AS position,
                kcu.COLUMN_NAME                AS from_column,
                kcu.REFERENCED_TABLE_SCHEMA    AS ref_schema,
                kcu.REFERENCED_TABLE_NAME      AS ref_table,
                kcu.REFERENCED_COLUMN_NAME     AS to_column,
                LOWER(rc.UPDATE_RULE)          AS on_update,
                LOWER(rc.DELETE_RULE)          AS on_delete,
                TRUE                           AS enforced,
                TRUE                           AS validated
            FROM 
                INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu
                JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA AND tc.TABLE_NAME = kcu.TABLE_NAME
                JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc ON rc.CONSTRAINT_SCHEMA = kcu.TABLE_SCHEMA AND rc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
            WHERE 
                kcu.TABLE_SCHEMA = {SCHEMA}
                AND tc.CONSTRAINT_TYPE = 'FOREIGN KEY'
            ORDER BY 
                kcu.TABLE_NAME, 
                kcu.CONSTRAINT_NAME, 
                kcu.ORDINAL_POSITION
        """

    def _sql_indexes(self) -> str:
        return r"""
            SELECT
                s.TABLE_NAME                                    AS table_name,
                s.INDEX_NAME                                    AS index_name,
                s.SEQ_IN_INDEX                                  AS position,
                COALESCE(s.EXPRESSION, s.COLUMN_NAME)           AS expr,
                (s.NON_UNIQUE = 0)                              AS is_unique,
                s.INDEX_TYPE                                    AS method,
                NULL                                            AS predicate
            FROM 
                INFORMATION_SCHEMA.STATISTICS s
            WHERE 
                s.TABLE_SCHEMA = {SCHEMA}
                AND s.INDEX_NAME <> 'PRIMARY'
            ORDER BY 
                s.TABLE_NAME, 
                s.INDEX_NAME, 
                s.SEQ_IN_INDEX
        """

    def _sql_sample_rows(self, catalog: str, schema: str, table: str, limit: int) -> SQLQuery:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -