fix: preserve adapter transactions and simplify MySQL bulk loading

docs/changelog.rst

@@ -14,6 +14,11 @@ Unreleased
 
 **Fixed:**
 
+* MySQL adapters accept either ``local_infile=True`` or
+  ``allow_local_infile=True`` to enable eligible native bulk loads. A separate
+  bulk-load opt-in is no longer required; set
+  ``enable_local_infile_bulk_load=False`` to retain ``executemany``.
+
 * ``sqlspec.extensions.litestar.LitestarConfig`` now exposes the complete plugin
   configuration, including ``session_table=True``, through the same type as
   ``sqlspec.config.LitestarConfig``.

docs/usage/bulk_ingest.rst

@@ -77,8 +77,8 @@ Capability matrix
    * - MySQL family (pymysql, asyncmy, aiomysql, mysql-connector)
      - ``executemany`` (default); ``LOAD DATA LOCAL INFILE`` (opt-in)
      - Server-managed
-     - ``enable_local_infile_bulk_load`` + connection ``local_infile`` /
-       ``allow_local_infile``
+     - Connection ``local_infile=True`` or ``allow_local_infile=True``;
+       ``enable_local_infile_bulk_load=False`` forces fallback
    * - bigquery
      - Parquet load job (default); Arrow Storage Write API (opt-in)
      - All-or-nothing load job / PENDING write stream
@@ -101,14 +101,29 @@ Security and opt-in paths
 
 Some fast paths are opt-in because they read local files or change semantics:
 
-- **MySQL ``LOAD DATA LOCAL INFILE``** requires both the adapter feature
-  ``enable_local_infile_bulk_load`` and the connection's local-infile setting
-  (``local_infile=True`` for pymysql/aiomysql/asyncmy, ``allow_local_infile=True``
-  for mysql-connector). Enabling the feature without the connection gate raises
-  :class:`~sqlspec.exceptions.ImproperConfigurationError` at config construction.
+- **MySQL ``LOAD DATA LOCAL INFILE``** is enabled by either
+  ``local_infile=True`` or ``allow_local_infile=True`` in the connection
+  configuration. Both names work for each MySQL adapter. If both are set,
+  either true value enables loading. If neither is true, loading stays off.
+  SQLSpec sends only the driver's native flag.
+
+  This one flag also turns on bulk loads for supported data. Set
+  ``driver_features={"enable_local_infile_bulk_load": False}`` to keep using
+  ``executemany``. Setting that feature to true with no connection opt-in raises
+  :class:`~sqlspec.exceptions.ImproperConfigurationError` when you create the config.
   The MySQL server must also have ``local_infile`` enabled. mysql-connector
   additionally honors ``allow_local_infile_in_path`` -- the staged temp file must
   live under that directory when it is set.
+  Connection opt-in trusts the configured MySQL server to request client files.
+
+  For asyncmy bulk loads, the requested filename must match that operation's
+  payload; this is not a connection-wide file restriction for other queries.
+  SQLSpec uses asyncmy's native sender
+  (version 0.2.13 or newer), removes its private UTF-8 payload after each attempt,
+  and closes the connection if native loading fails or is cancelled.
+  When bulk loading is disabled, and for nested, binary or duration values,
+  asyncmy uses ``executemany``. As with the other MySQL adapters, ``overwrite=True`` first
+  truncates the table; a later load failure does not restore those rows.
 - **Oracle direct path load** is the default bulk-ingest transport in Thin mode.
   Set ``enable_direct_path_load=False`` to force ``executemany``. Connections
   that do not expose the Direct Path Load API, including Thick-mode connections,
@@ -134,11 +149,25 @@ MySQL ``LOAD DATA LOCAL INFILE``:
 
     config = PyMysqlConfig(
         connection_config={"host": "localhost", "local_infile": True},
-        driver_features={"enable_local_infile_bulk_load": True},
     )
     with config.provide_session() as driver:
         driver.load_from_arrow("orders", arrow_table)
 
+Asyncmy with explicit LOCAL INFILE consent:
+
+.. code-block:: python
+
+    from sqlspec.adapters.asyncmy import AsyncmyConfig
+
+    config = AsyncmyConfig(
+        connection_config={
+            "host": "localhost",
+            "allow_local_infile": True,
+        },
+    )
+    async with config.provide_session() as driver:
+        await driver.load_from_arrow("orders", arrow_table)
+
 Oracle per-call batch error and array-DML row-count reporting:
 
 .. code-block:: python

pyproject.toml

@@ -38,7 +38,7 @@ aiomysql = ["aiomysql"]
 aiosqlite = ["aiosqlite"]
 alloydb = ["google-cloud-alloydb-connector"]
 arrow-odbc = ["arrow-odbc>=10.4", "pyarrow"]
-asyncmy = ["asyncmy"]
+asyncmy = ["asyncmy>=0.2.13"]
 asyncpg = ["asyncpg"]
 attrs = ["attrs", "cattrs"]
 bigquery = ["google-cloud-bigquery", "google-cloud-bigquery-storage", "google-cloud-storage"]

sqlspec/adapters/aiomysql/config.py

@@ -35,7 +35,6 @@
 __all__ = ("AiomysqlConfig", "AiomysqlConnectionParams", "AiomysqlDriverFeatures", "AiomysqlPoolParams")
 
 _POOL_ONLY_CONFIG_KEYS = frozenset({"maxsize", "minsize", "pool_recycle"})
-_AIOMYSQL_LOCAL_INFILE_GATE = "allow_local_infile"
 aiomysql: "AiomysqlModule" = cast("AiomysqlModule", AiomysqlModule)
 
 
@@ -84,28 +83,19 @@ class AiomysqlPoolParams(AiomysqlConnectionParams):
     pool_recycle: NotRequired[int]
 
 
-def _normalize_local_infile(connection_config: "Mapping[str, Any]", *, strip_consent_gate: bool) -> "dict[str, Any]":
-    """Normalize aiomysql local-infile settings and SQLSpec's consent gate."""
+def _normalize_local_infile(connection_config: "Mapping[str, Any]") -> "dict[str, Any]":
+    """Normalize aiomysql local-infile aliases to the native connection flag."""
     config = dict(connection_config)
 
     config.pop("enable_local_infile", None)
-    allow_local_infile = bool(config.get(_AIOMYSQL_LOCAL_INFILE_GATE, False))
-    local_infile = bool(config.get("local_infile", False))
-    if local_infile and not allow_local_infile:
-        msg = (
-            "Aiomysql local_infile=True requires allow_local_infile=True because "
-            "LOAD DATA LOCAL INFILE can read client files."
-        )
-        raise ImproperConfigurationError(msg)
-    config["local_infile"] = bool(local_infile and allow_local_infile)
-    if strip_consent_gate:
-        config.pop(_AIOMYSQL_LOCAL_INFILE_GATE, None)
+    allow_local_infile = bool(config.pop("allow_local_infile", False))
+    config["local_infile"] = bool(config.get("local_infile", False) or allow_local_infile)
     return config
 
 
 def _normalize_connection_kwargs(connection_config: "Mapping[str, Any]") -> "dict[str, Any]":
     """Build aiomysql.connect-compatible kwargs from SQLSpec connection config."""
-    config = _normalize_local_infile(connection_config, strip_consent_gate=True)
+    config = _normalize_local_infile(connection_config)
 
     for key in _POOL_ONLY_CONFIG_KEYS:
         config.pop(key, None)
@@ -253,9 +243,7 @@ def __init__(
             observability_config: Adapter-level observability overrides for lifecycle hooks and observers
             **kwargs: Additional keyword arguments
         """
-        connection_config = _normalize_local_infile(
-            normalize_connection_config(connection_config), strip_consent_gate=False
-        )
+        connection_config = _normalize_local_infile(normalize_connection_config(connection_config))
 
         connection_config.setdefault("host", "localhost")
         connection_config.setdefault("port", 3306)
@@ -271,11 +259,9 @@ def __init__(
         # Track initialized connections to ensure callback runs exactly once per physical connection
         self._initialized_connections: WeakSet[Any] = WeakSet()
 
+        features_dict.setdefault("enable_local_infile_bulk_load", connection_config["local_infile"])
         if features_dict.get("enable_local_infile_bulk_load") and not connection_config.get("local_infile"):
-            msg = (
-                "enable_local_infile_bulk_load requires local_infile=True and "
-                "allow_local_infile=True in connection_config."
-            )
+            msg = "enable_local_infile_bulk_load requires local_infile=True or allow_local_infile=True in connection_config."
             raise ImproperConfigurationError(msg)
 
         super().__init__(

sqlspec/adapters/asyncmy/_typing.py

@@ -5,18 +5,24 @@
 """
 
 import contextlib
+import os
 from typing import TYPE_CHECKING, Any
 
 import asyncmy as _asyncmy  # pyright: ignore
 from asyncmy import Connection  # pyright: ignore
 from asyncmy import errors as _asyncmy_errors  # pyright: ignore
+from asyncmy.connection import LoadLocalFile as _LoadLocalFile  # pyright: ignore
+from asyncmy.connection import MySQLResult as _AsyncmyResult  # pyright: ignore
 from asyncmy.constants import FIELD_TYPE as _ASYNCMY_FIELD_TYPE  # pyright: ignore
 from asyncmy.cursors import Cursor as _AsyncmyCursor  # pyright: ignore
 from asyncmy.cursors import DictCursor as _AsyncmyDictCursor  # pyright: ignore
 from asyncmy.pool import Pool as _AsyncmyPool  # pyright: ignore
+from asyncmy.protocol import LoadLocalPacketWrapper as _LoadLocalPacketWrapper  # pyright: ignore
+
+from sqlspec.exceptions import SQLSpecError
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable
+    from collections.abc import Awaitable, Callable, Iterator
     from types import TracebackType
     from typing import Protocol, TypeAlias
 
@@ -30,7 +36,7 @@ async def commit(self) -> object: ...
 
         async def rollback(self) -> object: ...
 
-        async def close(self) -> object: ...
+        def close(self) -> None: ...
 
     class AsyncmyModuleProtocol(Protocol):
         def connect(self, *args: Any, **kwargs: Any) -> "AsyncmyConnection": ...
@@ -70,6 +76,7 @@ class AsyncmyFieldTypeProtocol(Protocol):
     "AsyncmyPool",
     "AsyncmyRawCursor",
     "AsyncmySessionContext",
+    "asyncmy_local_infile",
 )
 
 
@@ -148,3 +155,72 @@ async def __aexit__(
             await self._release_connection(self._connection)
             self._connection = None
         return None
+
+
+class _AsyncmyLocalInfileResult(_AsyncmyResult):
+    """Normalize the upstream filename handoff while retaining its native sender."""
+
+    __slots__ = ("_filename",)
+
+    def __init__(self, connection: Any, filename: str) -> None:
+        super().__init__(connection)
+        self._filename = filename
+
+    async def _read_load_local_packet(self, first_packet: Any) -> None:
+        request = _LoadLocalPacketWrapper(first_packet).filename
+        if not self.connection._local_infile or os.fsdecode(request) != self._filename:
+            msg = "MySQL requested an unexpected LOCAL INFILE payload."
+            raise SQLSpecError(msg)
+        await _LoadLocalFile(self._filename, self.connection).send_data()  # type: ignore[no-untyped-call]
+        packet = await self.connection.read_packet()
+        if not packet.is_ok_packet():
+            msg = "MySQL did not acknowledge the LOCAL INFILE payload."
+            raise SQLSpecError(msg)
+        self._read_ok_packet(packet)  # type: ignore[attr-defined]
+
+
+@contextlib.contextmanager
+def asyncmy_local_infile(connection: "AsyncmyConnection", filename: str) -> "Iterator[None]":
+    """Scope the asyncmy 0.2.13/0.2.14 filename handoff fix to one native load.
+
+    Args:
+        connection: Physical connection exclusively held by this operation.
+        filename: Owned payload path expected in the server's file request.
+
+    Yields:
+        Control while native result reading uses the filename adapter.
+    """
+    raw: Any = connection
+    missing = object()
+    previous = raw.__dict__.get("_read_query_result", missing)
+
+    async def read_result(unbuffered: bool = False) -> None:
+        raw._result = None
+        result = _AsyncmyLocalInfileResult(raw, filename)
+        if unbuffered:
+            try:
+                await result.init_unbuffered_query()  # type: ignore[no-untyped-call]
+            except BaseException:
+                result.unbuffered_active = False
+                result.connection = None
+                raise
+        else:
+            await result.read()  # type: ignore[no-untyped-call]
+        raw._result = result
+        raw._affected_rows = result.affected_rows
+        if result.server_status:
+            raw.server_status = result.server_status
+
+    raw._read_query_result = read_result
+    try:
+        yield
+    except BaseException:
+        with contextlib.suppress(Exception):
+            raw.close()
+        raw._connected = False
+        raise
+    finally:
+        if previous is missing:
+            del raw._read_query_result
+        else:
+            raw._read_query_result = previous

sqlspec/adapters/asyncmy/config.py

@@ -38,7 +38,6 @@
 
 _ASYNCMY_POOL_ONLY_KEYS = frozenset(("minsize", "maxsize", "pool_recycle"))
 _ASYNCMY_POOL_KEYS = _ASYNCMY_POOL_ONLY_KEYS | {"echo"}
-_ASYNCMY_LOCAL_INFILE_GATE = "allow_local_infile"
 asyncmy: "AsyncmyModule" = cast("AsyncmyModule", AsyncmyModule)
 
 
@@ -120,12 +119,8 @@ def _normalize_connection_config(connection_config: "Mapping[str, Any] | None")
             raise ImproperConfigurationError(msg)
         config["cursor_cls"] = cursor_class
 
-    allow_local_infile = bool(config.pop(_ASYNCMY_LOCAL_INFILE_GATE, False))
-    local_infile = bool(config.get("local_infile", False))
-    if local_infile and not allow_local_infile:
-        msg = "Asyncmy local_infile=True requires allow_local_infile=True because LOAD DATA LOCAL INFILE can read client files."
-        raise ImproperConfigurationError(msg)
-    config["local_infile"] = bool(local_infile and allow_local_infile)
+    allow_local_infile = bool(config.pop("allow_local_infile", False))
+    config["local_infile"] = bool(config.get("local_infile", False) or allow_local_infile)
 
     return config
 
@@ -159,6 +154,9 @@ class AsyncmyDriverFeatures(TypedDict):
     MySQL/MariaDB handle JSON natively, but custom serializers can be provided
     for specialized use cases.
 
+    enable_local_infile_bulk_load: Use native LOCAL INFILE for eligible Arrow rows.
+     Defaults to the connection's local_infile or allow_local_infile opt-in.
+     Set False to force executemany on an opted-in connection.
     json_serializer: Custom JSON serializer function.
      Defaults to sqlspec.utils.serializers.to_json.
      Use for performance (orjson) or custom encoding.
@@ -178,6 +176,7 @@ class AsyncmyDriverFeatures(TypedDict):
      Defaults to "poll_queue".
     """
 
+    enable_local_infile_bulk_load: NotRequired[bool]
     json_serializer: NotRequired["Callable[[Any], str]"]
     json_deserializer: NotRequired["Callable[[str], Any]"]
     on_connection_create: "NotRequired[Callable[[AsyncmyConnection], Awaitable[None]]]"
@@ -296,12 +295,9 @@ def __init__(
         # Track initialized connections to ensure callback runs exactly once per physical connection
         self._initialized_connections: WeakSet[Any] = WeakSet()
 
-        if features_dict.get("enable_local_infile_bulk_load"):
-            msg = (
-                "asyncmy does not currently support SQLSpec's LOAD DATA LOCAL INFILE bulk path reliably. "
-                "Use aiomysql, mysql-connector, or pymysql for LOCAL INFILE bulk loads, or omit "
-                "enable_local_infile_bulk_load to use asyncmy batched executemany."
-            )
+        features_dict.setdefault("enable_local_infile_bulk_load", connection_config["local_infile"])
+        if features_dict.get("enable_local_infile_bulk_load") and not connection_config.get("local_infile"):
+            msg = "enable_local_infile_bulk_load requires local_infile=True or allow_local_infile=True in connection_config."
             raise ImproperConfigurationError(msg)
 
         super().__init__(