feat(migrations): support specifying schema for migrations (#471)

## Summary

Add `MigrationConfig.default_schema` and `version_table_schema` so a
migration run can target a specific schema/dataset, plus the per-adapter
driver hooks that apply, validate, and log that selection. Includes
related cleanup found during review: shared identifier-quoting helpers,
Oracle literal-case contract, and a comprehensive docs matrix.

## Feature: migration default schema

- `MigrationConfig.default_schema` — sets the session schema for the
migration run (per-dialect: `SET search_path` / `ALTER SESSION SET
CURRENT_SCHEMA` / `SET schema` / default_dataset).
- `MigrationConfig.version_table_schema` — controls where the tracker
lives; defaults to `default_schema` when present.
- Fail-fast validation via `driver.has_schema()` before any DDL runs;
misconfigured runs raise `MigrationError` with a clear message.
- Structured logging: every transition emits a `migration.schema.*`
event (`applied`, `noop`, `reset`) through `log_with_context`, with the
resolved driver class in the payload.

### Adapter support matrix

| Adapter | `supports_migration_schemas` | Hook |
| --- | --- | --- |
| asyncpg, psycopg, psqlpy | yes | `SET search_path TO "<schema>"` |
| cockroach_asyncpg, cockroach_psycopg | yes | inherits PG behavior |
| adbc (postgres dialect) | yes (derived from dialect) | `SET
search_path TO "<schema>"` |
| adbc (sqlite/duckdb/snowflake/...) | no | structured noop |
| oracledb | yes | `ALTER SESSION SET CURRENT_SCHEMA = "<schema>"`
(literal case) |
| duckdb | yes | `SET search_path = "<schema>"` (identifier form) |
| bigquery | yes | `default_dataset` on the client |
| spanner | yes | per-session default schema |
| sqlite, aiosqlite, asyncmy, aiomysql, mysqlconnector, pymysql,
mssql_python, arrow_odbc | no | structured noop |

See `docs/usage/migrations.rst` for the per-adapter contract
(literal-case requirements, opt-in flags, expected SQL surface).

## Refactor: identifier-quoting helpers

Consolidated seven adapter-local duplicates into
`sqlspec/utils/text.py`:

- `quote_identifier(value)` — ANSI form `"..."` with `""`-escaping. Used
by sqlite, aiosqlite, postgres family (asyncpg, psycopg, psqlpy, adbc),
duckdb, oracle.
- `quote_backtick_identifier(value)` — MySQL family form `` `...` ``
with `` `` ``-escaping. Used by asyncmy, aiomysql, mysqlconnector,
pymysql.

Removed duplicates: `_quote_sqlite_identifier`, `_quote_identifier`
(psqlpy), `_quote_mysql_identifier` (×4), `_quote_duckdb_search_path`.
Both helpers added to the mypyc include list.

## Bug fix: Oracle case-folding

Oracle migration hooks previously called `.upper()` and bound
`UPPER(:schema_name)`, which broke mixed-case Oracle users (`CREATE USER
"MixedCase"` was unreachable). Now matches the PG/DuckDB contract: the
schema value is interpolated literally and bound verbatim. Callers pass
the stored identifier (typically uppercase for unquoted creates, exact
case for quoted creates) — documented in `migrations.rst`.

## Other adjustments

- BigQuery: shared `SyncMigrationTracker` instead of an adapter-specific
tracker; SQLGlot DDL rendering patched so tracker SQL emits `PRIMARY KEY
NOT ENFORCED` and orders `DEFAULT` before `NOT NULL`. Integration tests
for the full migration workflow are gated behind native BigQuery
(emulator rejects column `DEFAULT` values in tracker DDL).
- ADBC: `supports_migration_schemas` is now a `@property` deriving from
dialect, so a single config class lights up only when targeting a
supported dialect.
- Tracker: handles dotted-form `version_table_name` correctly (strips
embedded `schema.table` prefixes so the qualified version_table cannot
be double-prefixed).
- Removed dead `_get_existing_columns_sql` from `oracledb/migrations.py`
(replaced by `data_dictionary.get_columns(..., schema=...)` earlier in
this PR).

## Docs

- `docs/usage/migrations.rst` — comprehensive adapter matrix,
per-dialect contract, literal-case guidance.
- `docs/examples/patterns/migrations_with_schema.py` — runnable example.

## Tests

- Unit: per-adapter migration-schema tests (postgres family, oracle,
duckdb, adbc), noop assertions for unsupported adapters, tracker
idempotency, command/runner validation, `quote_identifier` escape
behavior.
- Integration: full migration workflow against asyncpg, psycopg, psqlpy,
oracledb, duckdb, adbc; data-dictionary coverage tests for
psycopg/psqlpy/asyncpg; Spanner session-defaults test.

## Follow-up

Identifier handling for non-migration sites (data-dictionary metadata
queries, per-dialect case-folding normalizers) is intentionally out of
scope here and tracked separately in #477, #478, #479.

Author: cofin

docs/examples/patterns/migrations_with_schema.py

@@ -0,0 +1,72 @@
+from pathlib import Path
+
+__all__ = ("test_migrations_with_schema",)
+
+
+def test_migrations_with_schema(tmp_path: Path) -> None:
+    # start-example
+    from sqlspec.adapters.duckdb import DuckDBConfig
+    from sqlspec.migrations.commands import SyncMigrationCommands
+
+    migration_dir = tmp_path / "migrations"
+    db_path = tmp_path / "app.duckdb"
+
+    config = DuckDBConfig(
+        connection_config={"database": str(db_path)},
+        migration_config={
+            "script_location": str(migration_dir),
+            "version_table_name": "schema_versions",
+            "default_schema": "app_schema",
+            "version_table_schema": "admin_schema",
+        },
+    )
+
+    try:
+        with config.provide_session() as session:
+            session.execute("CREATE SCHEMA app_schema")
+            session.execute("CREATE SCHEMA admin_schema")
+
+        commands = SyncMigrationCommands(config)
+        commands.init(str(migration_dir), package=True)
+
+        (migration_dir / "0001_create_users.py").write_text(
+            '''"""Create users."""
+
+
+def up():
+    """Create an unqualified table in app_schema."""
+    return ["CREATE TABLE users (id INTEGER PRIMARY KEY, name VARCHAR NOT NULL)"]
+
+
+def down():
+    """Drop the unqualified table from app_schema."""
+    return ["DROP TABLE IF EXISTS users"]
+'''
+        )
+
+        commands.upgrade()
+
+        with config.provide_session() as session:
+            users_table = session.select_value(
+                """
+                SELECT table_name
+                FROM information_schema.tables
+                WHERE table_schema = ? AND table_name = ?
+                """,
+                ("app_schema", "users"),
+            )
+            tracker_table = session.select_value(
+                """
+                SELECT table_name
+                FROM information_schema.tables
+                WHERE table_schema = ? AND table_name = ?
+                """,
+                ("admin_schema", "schema_versions"),
+            )
+
+        assert users_table == "users"
+        assert tracker_table == "schema_versions"
+    finally:
+        if config.connection_instance:
+            config.close_pool()
+    # end-example

docs/usage/migrations.rst

@@ -74,6 +74,119 @@ specific extension, ``migration_config["include_extensions"]`` to opt in
 explicitly by extension name, or ``migration_config["enabled"] = False`` to
 disable migrations entirely for a database config.
 
+Configuring a Default Schema
+----------------------------
+
+Use ``migration_config["default_schema"]`` when migration SQL should run
+against a pre-existing schema without qualifying every table in each migration
+file. SQLSpec validates the schema before creating the tracker table or applying
+DDL, then configures the migration session before each migration is executed.
+
+Use ``migration_config["version_table_schema"]`` when the migration tracker
+table should live somewhere different from the objects managed by migrations.
+If ``version_table_schema`` is not set, the tracker schema resolves to
+``default_schema``. If neither field is set, the tracker table is unqualified and
+uses the adapter's normal default namespace.
+
+.. code-block:: python
+
+    from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+    config = AsyncpgConfig(
+        connection_config={"dsn": "postgresql://localhost/app"},
+        migration_config={
+            "script_location": "migrations/postgres",
+            "version_table_name": "schema_versions",
+            "default_schema": "app_schema",
+            "version_table_schema": "admin_schema",
+        },
+    )
+
+The operator must create the target schema before running migrations. The
+migration role also needs the database-specific privileges to create objects
+there. For PostgreSQL, that usually means ``USAGE`` and
+``CREATE`` on the target schema, plus permission to create or update the
+tracker table.
+
+Adapter support is opt-in via the ``supports_migration_schemas`` class flag on
+each config. Configuring ``default_schema`` against an adapter that does not
+opt in raises ``MigrationError`` before any DDL is issued.
+
+.. list-table::
+   :header-rows: 1
+   :widths: 28 18 54
+
+   * - Adapter
+     - Default schema
+     - Mechanism
+   * - ``asyncpg``
+     - Supported
+     - ``SET LOCAL search_path`` (transactional) / ``SET search_path`` + ``RESET`` (non-transactional);
+       validates ``information_schema.schemata``.
+   * - ``psycopg`` (sync and async)
+     - Supported
+     - Same as ``asyncpg``.
+   * - ``psqlpy``
+     - Supported
+     - Same as ``asyncpg``.
+   * - ``cockroach_asyncpg``
+     - Supported
+     - Inherits ``asyncpg`` behavior. CockroachDB exposes the PostgreSQL
+       wire protocol and accepts ``SET search_path``.
+   * - ``cockroach_psycopg`` (sync and async)
+     - Supported
+     - Inherits ``psycopg`` behavior.
+   * - ``adbc`` (PostgreSQL dialect)
+     - Supported
+     - Same as ``asyncpg``. Detection is dialect-based on the configured
+       ADBC URI; ``supports_migration_schemas`` becomes ``True`` only when
+       the resolved dialect is PostgreSQL-compatible.
+   * - ``oracledb`` (sync and async)
+     - Supported
+     - ``ALTER SESSION SET CURRENT_SCHEMA``; validates ``ALL_USERS``. The
+       schema is matched verbatim, so callers must pass the literal stored
+       identifier — typically uppercase for users created unquoted, the
+       exact case for users created with quoted identifiers.
+   * - ``duckdb``
+     - Supported
+     - ``SET search_path``; validates ``information_schema.schemata``.
+   * - ``sqlite``, ``aiosqlite``
+     - Not supported
+     - SQLite has no schema namespace; use ``ATTACH DATABASE`` to layer
+       additional databases instead.
+   * - ``asyncmy``, ``aiomysql``, ``mysqlconnector``, ``pymysql``
+     - Not supported
+     - MySQL conflates schema and database. Select the target database in
+       the connection URL or via ``USE`` inside the migration.
+   * - ``adbc`` (non-PostgreSQL dialects, including SQL Server)
+     - Not supported
+     - ADBC does not expose a portable per-session schema setter for these
+       dialects. Configure the default schema at the user or login level in
+       the underlying database.
+   * - ``mssql_python``
+     - Not supported
+     - SQL Server resolves the default schema from the login. Set it with
+       ``ALTER USER ... WITH DEFAULT_SCHEMA = ...`` in your database.
+   * - ``bigquery``
+     - Not supported
+     - BigQuery requires fully qualified ``project.dataset.table`` references
+       for cross-dataset DDL; there is no session-scoped default dataset.
+   * - ``spanner``
+     - Not supported
+     - Cloud Spanner ties objects to a single schema per database; there is
+       no session-scoped switch.
+   * - ``arrow_odbc``
+     - Not supported
+     - ODBC connection-string semantics vary per driver. Configure the
+       default schema through the underlying DSN.
+
+Example with unqualified DDL:
+
+.. literalinclude:: /examples/patterns/migrations_with_schema.py
+   :language: python
+   :start-after: # start-example
+   :end-before: # end-example
+
 Logging and Echo Controls
 -------------------------
 

sqlspec/adapters/adbc/config.py

@@ -224,8 +224,10 @@ def __init__(
         self._pgvector_available: bool | None = None
         self._paradedb_available: bool | None = None
 
+        dialect = resolve_dialect_from_config(self.connection_config)
+
         if statement_config is None:
-            statement_config = get_statement_config(resolve_dialect_from_config(self.connection_config))
+            statement_config = get_statement_config(dialect)
 
         statement_config, driver_features = apply_driver_features(statement_config, driver_features)
 
@@ -241,6 +243,11 @@ def __init__(
             **kwargs,
         )
 
+    @property
+    def supports_migration_schemas(self) -> bool:  # type: ignore[override]
+        """Migration schema support is only available for PostgreSQL-backed ADBC drivers."""
+        return is_postgres_dialect(resolve_dialect_from_config(self.connection_config))
+
     def create_connection(self) -> AdbcConnection:
         """Create and return a new connection using the specified driver.
 

sqlspec/adapters/adbc/driver.py

@@ -31,6 +31,7 @@
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.module_loader import ensure_pyarrow
 from sqlspec.utils.serializers import to_json
+from sqlspec.utils.text import quote_identifier
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -292,6 +293,35 @@ def rollback(self) -> None:
             msg = f"Failed to rollback transaction: {e}"
             raise SQLSpecError(msg) from e
 
+    def set_migration_session_schema(self, schema: str) -> None:
+        """Set the PostgreSQL search path for migration SQL when using ADBC PostgreSQL."""
+        if not self._is_postgres:
+            super().set_migration_session_schema(schema)
+            return
+        quoted_schema = quote_identifier(schema)
+        with self.with_cursor(self.connection) as cursor:
+            cursor.execute(f'SET search_path TO {quoted_schema}, "$user", public')
+
+    def set_migration_non_transactional_schema(self, schema: str) -> None:
+        """Set the PostgreSQL search path for non-transactional migration SQL."""
+        self.set_migration_session_schema(schema)
+
+    def reset_migration_session_schema(self) -> None:
+        """Reset PostgreSQL search path after non-transactional migration SQL."""
+        if not self._is_postgres:
+            super().reset_migration_session_schema()
+            return
+        with self.with_cursor(self.connection) as cursor:
+            cursor.execute("RESET search_path")
+
+    def has_schema(self, schema: str) -> bool:
+        """Return whether a PostgreSQL schema exists when using ADBC PostgreSQL."""
+        if not self._is_postgres:
+            return super().has_schema(schema)
+        with self.with_cursor(self.connection) as cursor:
+            cursor.execute("SELECT 1 FROM information_schema.schemata WHERE schema_name = $1", parameters=[schema])
+            return cursor.fetchone() is not None
+
     def with_cursor(self, connection: "AdbcConnection") -> "AdbcCursor":
         """Create context manager for cursor.
 

sqlspec/adapters/aiomysql/core.py

@@ -21,6 +21,7 @@
     UniqueViolationError,
 )
 from sqlspec.utils.serializers import from_json, to_json
+from sqlspec.utils.text import quote_backtick_identifier
 from sqlspec.utils.type_converters import build_uuid_coercions
 from sqlspec.utils.type_guards import has_cursor_metadata, has_lastrowid, has_rowcount
 
@@ -117,23 +118,18 @@ def _bool_to_int(value: bool) -> int:
     return int(value)
 
 
-def _quote_mysql_identifier(identifier: str) -> str:
-    normalized = identifier.replace("`", "``")
-    return f"`{normalized}`"
-
-
 def format_identifier(identifier: str) -> str:
     cleaned = identifier.strip()
     if not cleaned:
         msg = "Table name must not be empty"
         raise SQLSpecError(msg)
     parts = [part for part in cleaned.split(".") if part]
-    formatted = ".".join(_quote_mysql_identifier(part) for part in parts)
-    return formatted or _quote_mysql_identifier(cleaned)
+    formatted = ".".join(quote_backtick_identifier(part) for part in parts)
+    return formatted or quote_backtick_identifier(cleaned)
 
 
 def build_insert_statement(table: str, columns: "list[str]") -> str:
-    column_clause = ", ".join(_quote_mysql_identifier(column) for column in columns)
+    column_clause = ", ".join(quote_backtick_identifier(column) for column in columns)
     placeholders = ", ".join("%s" for _ in columns)
     return f"INSERT INTO {format_identifier(table)} ({column_clause}) VALUES ({placeholders})"
 

sqlspec/adapters/aiosqlite/core.py

@@ -18,6 +18,7 @@
     UniqueViolationError,
 )
 from sqlspec.utils.serializers import from_json, to_json
+from sqlspec.utils.text import quote_identifier
 from sqlspec.utils.type_converters import build_decimal_converter, build_time_iso_converter, build_uuid_coercions
 from sqlspec.utils.type_guards import has_rowcount, has_sqlite_error
 
@@ -58,23 +59,18 @@ def _bool_to_int(value: bool) -> int:
     return int(value)
 
 
-def _quote_sqlite_identifier(identifier: str) -> str:
-    normalized = identifier.replace('"', '""')
-    return f'"{normalized}"'
-
-
 def format_identifier(identifier: str) -> str:
     cleaned = identifier.strip()
     if not cleaned:
         msg = "Table name must not be empty"
         raise SQLSpecError(msg)
     parts = [part for part in cleaned.split(".") if part]
-    formatted = ".".join(_quote_sqlite_identifier(part) for part in parts)
-    return formatted or _quote_sqlite_identifier(cleaned)
+    formatted = ".".join(quote_identifier(part) for part in parts)
+    return formatted or quote_identifier(cleaned)
 
 
 def build_insert_statement(table: str, columns: "list[str]") -> str:
-    column_clause = ", ".join(_quote_sqlite_identifier(column) for column in columns)
+    column_clause = ", ".join(quote_identifier(column) for column in columns)
     placeholders = ", ".join("?" for _ in columns)
     return f"INSERT INTO {format_identifier(table)} ({column_clause}) VALUES ({placeholders})"
 

sqlspec/adapters/asyncmy/core.py

@@ -21,6 +21,7 @@
     UniqueViolationError,
 )
 from sqlspec.utils.serializers import from_json, to_json
+from sqlspec.utils.text import quote_backtick_identifier
 from sqlspec.utils.type_converters import build_uuid_coercions
 from sqlspec.utils.type_guards import has_cursor_metadata, has_lastrowid, has_rowcount
 
@@ -117,23 +118,18 @@ def _bool_to_int(value: bool) -> int:
     return int(value)
 
 
-def _quote_mysql_identifier(identifier: str) -> str:
-    normalized = identifier.replace("`", "``")
-    return f"`{normalized}`"
-
-
 def format_identifier(identifier: str) -> str:
     cleaned = identifier.strip()
     if not cleaned:
         msg = "Table name must not be empty"
         raise SQLSpecError(msg)
     parts = [part for part in cleaned.split(".") if part]
-    formatted = ".".join(_quote_mysql_identifier(part) for part in parts)
-    return formatted or _quote_mysql_identifier(cleaned)
+    formatted = ".".join(quote_backtick_identifier(part) for part in parts)
+    return formatted or quote_backtick_identifier(cleaned)
 
 
 def build_insert_statement(table: str, columns: "list[str]") -> str:
-    column_clause = ", ".join(_quote_mysql_identifier(column) for column in columns)
+    column_clause = ", ".join(quote_backtick_identifier(column) for column in columns)
     placeholders = ", ".join("%s" for _ in columns)
     return f"INSERT INTO {format_identifier(table)} ({column_clause}) VALUES ({placeholders})"
 

sqlspec/adapters/asyncpg/config.py

@@ -260,6 +260,7 @@ class AsyncpgConfig(AsyncDatabaseConfig[AsyncpgConnection, "Pool[Record]", Async
     driver_type: "ClassVar[type[AsyncpgDriver]]" = AsyncpgDriver
     connection_type: "ClassVar[type[AsyncpgConnection]]" = type(AsyncpgConnection)  # type: ignore[assignment]
     supports_transactional_ddl: "ClassVar[bool]" = True
+    supports_migration_schemas: "ClassVar[bool]" = True
     supports_native_arrow_export: "ClassVar[bool]" = True
     supports_native_arrow_import: "ClassVar[bool]" = True
     supports_native_parquet_export: "ClassVar[bool]" = True