feat(storage,driver,adapters): add streaming Arrow batches, binary COPY, staged upserts, and index lifecycle

docs/learn/data-dictionary.md

@@ -59,18 +59,18 @@ with spec.provide_session(config) as session:
 
 ## Core Introspection and DDL Methods
 
-| Method | Returned Metadata | Description |
-| --- | --- | --- |
-| `get_tables(driver, schema=None)` | `list[TableMetadata]` | Lists user tables in topological/catalog order (including self-referential root tables). |
-| `get_columns(driver, table=None, schema=None)` | `list[ColumnMetadata]` | Returns column names, types, nullability, defaults, and identity/generated expressions. |
-| `get_indexes(driver, table=None, schema=None)` | `list[IndexMetadata]` | Returns index names, indexed column lists, and uniqueness flags. |
-| `get_foreign_keys(driver, table=None, schema=None)` | `list[ForeignKeyMetadata]` | Returns referencing/referenced schemas, tables, columns, and `on_delete` / `on_update` actions. |
-| `get_ddl(driver, object_name, ...)` | `DDLResult` | Extracts native engine DDL when available or synthesizes dialect-aware AST DDL (`target_dialect` optional). |
-| `get_schema_ddl(driver, schema=None, ...)` | `MetadataResult` | Emits topologically ordered `CREATE` or reverse-topological `DROP` `DDLResult` items across a schema. |
-| `get_table_profile(driver, table, ...)` | `list[ColumnProfileMetadata]` | Samples live rows to compute null counts, cardinality, max string lengths, and narrower SQL types. |
-| `infer_table_ddl_from_data(driver, table, ...)` | `DDLResult` | Generates a typed `CREATE TABLE` statement from profiled column values (`target_table` / `target_dialect`). |
-| `get_version(driver)` | `VersionInfo \| None` | Parses and caches the connected database server's `major`, `minor`, and `patch` version. |
-| `get_optimal_type(driver, type_category)` | `str` | Resolves the best DDL type for categories such as `"json"`, `"uuid"`, `"binary"`, `"text"`, or `"timestamp"`. |
+| Method | Returns | Description |
+| :--- | :--- | :--- |
+| `get_tables()` | `list[TableMetadata]` | Lists user tables in topological or catalog order. |
+| `get_columns()` | `list[ColumnMetadata]` | Returns column names, types, nullability, defaults, and identity. |
+| `get_indexes()` | `list[IndexMetadata]` | Returns index names, indexed column lists, and uniqueness flags. |
+| `get_foreign_keys()` | `list[ForeignKeyMetadata]` | Returns referencing/referenced columns and on-delete/update actions. |
+| `get_ddl()` | `DDLResult` | Extracts native engine DDL or synthesizes dialect-aware AST DDL. |
+| `get_schema_ddl()` | `MetadataResult` | Emits ordered `CREATE` or reverse-ordered `DROP` statements across a schema. |
+| `get_table_profile()` | `list[ColumnProfileMetadata]` | Samples live rows to compute nulls, cardinality, and narrower types. |
+| `infer_table_ddl_from_data()` | `DDLResult` | Generates a typed `CREATE TABLE` statement from profiled column values. |
+| `get_version()` | `VersionInfo | None` | Returns database server major, minor, and patch version. |
+| `get_optimal_type()` | `str` | Resolves optimal DDL types for abstract categories like `json` or `uuid`. |
 
 ### Metadata Record Fields
 
@@ -93,6 +93,9 @@ with spec.provide_session(config) as session:
 
 When migrating loosely typed staging tables (`TEXT` / `VARCHAR` landing zones) or raw JSON/Arrow datasets, `get_table_profile` and `infer_table_ddl_from_data` inspect live sample values to recommend tighter types (`BIGINT`, `NUMERIC(p, s)`, `BOOLEAN`, `DATE`, `TIMESTAMP`, `UUID`) and `NOT NULL` constraints.
 
+!!! note "Cross-Dialect Transpilation"
+    Pass `target_dialect` to `get_ddl` or `get_schema_ddl` to transpile catalog DDL from one database engine to another using SQLGlot AST compilation.
+
 ## Dependency Ordering and Portable DDL Helpers
 
 When exporting schemas, seeding fixtures, or truncating tables in foreign-key order, combine `get_foreign_keys` with `dependency_edges_from_foreign_keys`, `sort_dependencies`, and `sort_ddl_results` from `sqlspec.data_dictionary`:
@@ -111,7 +114,10 @@ sorted_plan = sort_dependencies(edges)
 
 ## Identifier Quoting vs Catalog Lookups
 
-Catalog views store unquoted identifier names (`schema_versions`, not `"schema_versions"`). Pass bare table and schema names (`table="members"`, `schema="public"`) to `data_dictionary` methods; synthesized DDL quotes reserved words and mixed-case catalog names automatically for the target dialect.
+Catalog views store unquoted identifier names (`schema_versions`, not `"schema_versions"`). Pass bare table and schema names (`table="members"`, `schema="public"`) to `data_dictionary` methods. Synthesized DDL quotes reserved words and mixed-case catalog names automatically for the target dialect.
+
+!!! tip "Bare Catalog Identifiers"
+    Always pass bare unquoted identifier names (`table="members"`, `schema="public"`) to `data_dictionary` methods. SQLSpec normalizes catalog queries and quotes reserved words or mixed-case identifiers automatically in generated DDL.
 
 ## Recap
 

docs/reference/driver/common.md

@@ -27,7 +27,7 @@ Shared driver execution results, exception handler protocols and base classes, m
       heading_level: 3
       members: false
 
-::: sqlspec.driver._query_cache.CachedQuery
+::: sqlspec.core.cache.CachedQuery
     options:
       heading_level: 3
       members: false

sqlspec/adapters/adbc/config.py

@@ -1,5 +1,6 @@
 """ADBC database configuration."""
 
+from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, NotRequired, TypedDict, cast
 
@@ -34,7 +35,6 @@
 from sqlspec.utils.config_tools import normalize_connection_config
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
     from types import TracebackType
 
     from sqlspec.observability import ObservabilityConfig
@@ -98,6 +98,8 @@ class AdbcConnectionParams(TypedDict):
     """Database authentication password lifted into ``db_kwargs`` or ``connect()`` as required by the driver."""
     token: NotRequired[str]
     """Bearer, OAuth refresh, or Quack/FlightSQL authentication token."""
+    auth_token_provider: NotRequired[Callable[[], str]]
+    """Callable returning a fresh password or bearer token on each ``create_connection`` call."""
     project_id: NotRequired[str]
     """Google Cloud project ID mapped to ``adbc.bigquery.sql.project_id`` for ``adbc-driver-bigquery``."""
     dataset_id: NotRequired[str]

sqlspec/adapters/adbc/core.py

@@ -729,6 +729,18 @@ def build_connection_config(connection_config: "Mapping[str, Any]") -> "dict[str
     if not uses_driver_manager:
         config.pop("entrypoint", None)
 
+    auth_token_provider = config.pop("auth_token_provider", None)
+    if callable(auth_token_provider):
+        fresh_token = str(auth_token_provider())
+        if driver_kind == "quack" and "username" not in config and "user" not in config:
+            config["token"] = fresh_token
+        elif driver_kind in {"gizmosql", "flightsql"} and "username" not in config and "user" not in config:
+            config["authorization_header"] = (
+                fresh_token if fresh_token.lower().startswith("bearer ") else f"Bearer {fresh_token}"
+            )
+        else:
+            config["password"] = fresh_token
+
     _lift_driver_specific_db_kwargs(config, driver_kind, resolved_connect_path)
 
     config.pop("driver_name", None)

sqlspec/adapters/adbc/driver.py

@@ -55,13 +55,12 @@
 from sqlspec.utils.text import normalize_identifier, quote_identifier, split_qualified_identifier
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Mapping, Sequence
 
     from sqlspec.adapters.adbc._typing import AdbcConnection, AdbcRawCursor
     from sqlspec.builder import QueryBuilder
     from sqlspec.core import ArrowResult, SQLResult, Statement, StatementFilter
-    from sqlspec.driver import ExecutionResult
-    from sqlspec.driver._query_cache import CachedQuery
+    from sqlspec.driver import CachedQuery, ExecutionResult
     from sqlspec.storage import StorageBridgeJob, StorageDestination, StorageFormat, StorageTelemetry
     from sqlspec.typing import ArrowReturnFormat, StatementParameters
 
@@ -715,10 +714,35 @@ def load_from_arrow(
         partitioner: "dict[str, object] | None" = None,
         overwrite: bool = False,
         telemetry: "StorageTelemetry | None" = None,
+        stage_table: "bool | str | None" = None,
+        conflict_keys: "Sequence[str] | None" = None,
+        on_conflict: Literal["update", "nothing"] = "update",
+        exclude_update_columns: "Sequence[str] | None" = None,
+        drop_stage_table: bool = True,
+        drop_indexes: bool = False,
+        concurrent_indexes: bool = False,
+        type_casts: "Mapping[str, str] | None" = None,
     ) -> "StorageBridgeJob":
         """Ingest an Arrow payload directly through the ADBC cursor."""
 
         self._require_capability("arrow_import_enabled")
+        if stage_table or conflict_keys or drop_indexes:
+            return self._execute_staged_load_sync(
+                table,
+                source,
+                raw_loader=self.load_from_arrow,
+                partitioner=partitioner,
+                overwrite=overwrite,
+                telemetry=telemetry,
+                stage_table=stage_table,
+                conflict_keys=conflict_keys,
+                on_conflict=on_conflict,
+                exclude_update_columns=exclude_update_columns,
+                drop_stage_table=drop_stage_table,
+                drop_indexes=drop_indexes,
+                concurrent_indexes=concurrent_indexes,
+                type_casts=type_casts,
+            )
         ensure_pyarrow()
 
         source_data: Any = source.get_data() if hasattr(source, "get_data") else source
@@ -773,11 +797,13 @@ def load_from_storage(
         file_format: "StorageFormat",
         partitioner: "dict[str, object] | None" = None,
         overwrite: bool = False,
+        **kwargs: Any,
     ) -> "StorageBridgeJob":
         """Read an artifact from storage and ingest it via ADBC."""
 
-        arrow_table, inbound = self._read_storage_arrow(source, file_format=file_format)
-        return self.load_from_arrow(table, arrow_table, partitioner=partitioner, overwrite=overwrite, telemetry=inbound)
+        return super().load_from_storage(
+            table, source, file_format=file_format, partitioner=partitioner, overwrite=overwrite, **kwargs
+        )
 
     def collect_rows(self, cursor: "AdbcRawCursor", fetched: "list[Any]") -> "tuple[list[Any], list[str], int]":
         """Collect ADBC rows for the direct execution path."""

sqlspec/adapters/aiomysql/driver.py

@@ -12,7 +12,7 @@
 import tempfile
 from collections.abc import Sized
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Final, cast
+from typing import TYPE_CHECKING, Any, Final, Literal, cast
 
 from sqlspec.adapters.aiomysql._typing import (
     AIOMYSQL_INSERT_VALUES_PATTERN,
@@ -50,7 +50,7 @@
 from sqlspec.utils.type_guards import supports_json_type
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Mapping, Sequence
 
     from sqlspec.adapters.aiomysql._typing import AiomysqlConnection
     from sqlspec.core import SQL, SQLResult, StatementConfig
@@ -346,10 +346,35 @@ async def load_from_arrow(
         partitioner: "dict[str, object] | None" = None,
         overwrite: bool = False,
         telemetry: "StorageTelemetry | None" = None,
+        stage_table: "bool | str | None" = None,
+        conflict_keys: "Sequence[str] | None" = None,
+        on_conflict: Literal["update", "nothing"] = "update",
+        exclude_update_columns: "Sequence[str] | None" = None,
+        drop_stage_table: bool = True,
+        drop_indexes: bool = False,
+        concurrent_indexes: bool = False,
+        type_casts: "Mapping[str, str] | None" = None,
     ) -> "StorageBridgeJob":
         """Load Arrow data into MySQL using batched inserts."""
 
         self._require_capability("arrow_import_enabled")
+        if stage_table or conflict_keys or drop_indexes:
+            return await self._execute_staged_load_async(
+                table,
+                source,
+                raw_loader=self.load_from_arrow,
+                partitioner=partitioner,
+                overwrite=overwrite,
+                telemetry=telemetry,
+                stage_table=stage_table,
+                conflict_keys=conflict_keys,
+                on_conflict=on_conflict,
+                exclude_update_columns=exclude_update_columns,
+                drop_stage_table=drop_stage_table,
+                drop_indexes=drop_indexes,
+                concurrent_indexes=concurrent_indexes,
+                type_casts=type_casts,
+            )
         arrow_table = self._coerce_arrow_table(source)
         if overwrite:
             statement = f"TRUNCATE TABLE {format_identifier(table)}"
@@ -405,12 +430,12 @@ async def load_from_storage(
         file_format: "StorageFormat",
         partitioner: "dict[str, object] | None" = None,
         overwrite: bool = False,
+        **kwargs: Any,
     ) -> "StorageBridgeJob":
         """Load staged artifacts from storage into MySQL."""
 
-        arrow_table, inbound = await self._read_storage_arrow(source, file_format=file_format)
-        return await self.load_from_arrow(
-            table, arrow_table, partitioner=partitioner, overwrite=overwrite, telemetry=inbound
+        return await super().load_from_storage(
+            table, source, file_format=file_format, partitioner=partitioner, overwrite=overwrite, **kwargs
         )
 
     def collect_rows(self, cursor: Any, fetched: "list[Any]") -> "tuple[list[Any], list[str], int]":