Add MySQL bulk load and dialect-controlled engine creation

.github/workflows/integration-tests-mysql.yml

@@ -13,6 +13,7 @@ jobs:
     services:
       mysql:
         image: mysql:8.0
+        command: --local-infile=1
         env:
           MYSQL_DATABASE: test_db
           MYSQL_USER: user
@@ -21,13 +22,18 @@ jobs:
           TZ: Europe/Paris
         ports:
           - 3306:3306
-        options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=3
+        options: >-
+          --health-cmd="mysqladmin ping"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=3
     steps:
       - name: Verify MySQL connection
         run: |
           mysql --version
           sudo apt-get install -y mysql-client
           mysql --host 127.0.0.1 --port 3306 -uroot -psecretroot -e "SHOW DATABASES"
+          mysql --host 127.0.0.1 --port 3306 -uroot -psecretroot -e "SHOW VARIABLES LIKE 'local_infile'"
 
       - name: Check out repository code
         uses: actions/checkout@v4

README.md

@@ -37,8 +37,8 @@ SQL views or stored procedures aimed at extracting, correcting and formatting th
 
 This package uses `sqlalchemy` to interact with the database, so it should work with different database backends. 
 Automated integration tests run against PostgreSQL, MySQL, MS SQL Server and DuckDB. You may have to install additional 
-packages to connect to your database (e.g. `psycopg2` for PostgreSQL, `pymysql` for MySQL, `pyodbc` for MS SQL Server or
-`duckdb_engine` for DuckDB).
+packages to connect to your database (e.g. `psycopg2` or `psycopg` for PostgreSQL, `pymysql` or `mysqlclient` for
+MySQL, `pyodbc` for MS SQL Server, or `duckdb-engine` for DuckDB).
 
 **Please read the [package documentation website](https://cre-dev.github.io/xml2db) for all the details!**
 

docs/getting_started.md

@@ -33,7 +33,9 @@ data_model = DataModel(
 
 At this stage, it is not required to provide a connection string and have an actual database set up, but it will be 
 necessary if you want to use this [`DataModel`](api/data_model.md) to actually import data. You may need to install the
-Python package of the connector you use in your `sqlalchemy` connection string (`psycopg2` in the example above).
+Python package of the connector you use in your `sqlalchemy` connection string (e.g. `psycopg2` or `psycopg` for
+PostgreSQL, `pymysql` or `mysqlclient` for MySQL, `pyodbc` for SQL Server, `duckdb-engine` for DuckDB). See
+[How it works](how_it_works.md#bulk-loading) for which drivers enable native bulk loading.
 
 You can provide an optional model configuration, which will allow forcing or preventing some schema simplification, set
 columns types manually, etc. By default, some simplifications will be applied when possible, in order to limit the 

docs/how_it_works.md

@@ -184,6 +184,34 @@ We keep the primary keys from the flat data model created at the previous stage,
     random one, which can be useful if you want to decompose the process of loading data and merging it with the target
     tables later, for instance to gain a finer control over concurrency.
 
+### Bulk loading
+
+For each supported backend, `xml2db` uses a native bulk-loading mechanism to fill the temporary tables, which is
+significantly faster than row-by-row inserts for large datasets. When the native path is unavailable (wrong driver,
+missing tool, or server configuration), it falls back silently to SQLAlchemy's standard `executemany`.
+
+| Backend | Mechanism | Required driver / tool | Default threshold |
+|---|---|---|---|
+| PostgreSQL | `COPY FROM STDIN` | `psycopg2` or `psycopg` | 0 (always used) |
+| MySQL / MariaDB | `LOAD DATA LOCAL INFILE` | `pymysql` or `mysqlclient`; server `local_infile=ON` | 100 rows |
+| MS SQL Server | `bcp` utility | `bcp` on PATH; SQL or Windows/Kerberos auth | 100 rows |
+| DuckDB | `read_csv()` | built-in | 100 rows |
+
+The threshold column means that batches smaller than that number always use `executemany` (avoiding temp-file overhead
+for small inserts). PostgreSQL's `COPY` is in-protocol and has no file overhead, so there is no threshold.
+
+For MySQL, when a connection string is passed to `DataModel`, `local_infile=True` is injected automatically into the
+connection arguments. The MySQL server must also have `local_infile=ON` (e.g. launched with `--local-infile=1`);
+if not, bulk loading is silently skipped.
+
+You can control this behaviour via the `bulk_load` and `bulk_load_threshold` arguments of
+[`Document.insert_into_target_tables`](api/document.md#xml2db.document.Document.insert_into_target_tables):
+
+- `bulk_load=None` (default): use the native path when available, fall back to `executemany` silently.
+- `bulk_load=False`: always use `executemany`, regardless of what is available.
+- `bulk_load=True`: require the native path; raise a `RuntimeError` with an actionable message if the required
+  driver, tool, or server setting is missing.
+
 ### Merging the data
 
 The last step is to merge the temporary tables data into the target tables, while enforcing deduplication, keeping 

docs/index.md

@@ -43,8 +43,9 @@ SQL views or stored procedures aimed at extracting, correcting and formatting th
 
 This package uses `sqlalchemy` to interact with the database, so it should work with different database backends. 
 Automated integration tests run against PostgreSQL, MySQL, MS SQL Server and DuckDB. You may have to install additional 
-packages to connect to your database (e.g. `psycopg2` for PostgreSQL, `pymysql` for MySQL, `pyodbc` for MS SQL Server or
-`duckdb_engine` for DuckDB).
+packages to connect to your database (e.g. `psycopg2` or `psycopg` for PostgreSQL, `pymysql` or `mysqlclient` for
+MySQL, `pyodbc` for MS SQL Server, or `duckdb-engine` for DuckDB). See [How it works](how_it_works.md#bulk-loading)
+for which drivers enable native bulk loading.
 
 ## How to visualize your data model 
 

src/xml2db/dialect/base.py

@@ -14,6 +14,7 @@
     SmallInteger,
     BigInteger,
     LargeBinary,
+    create_engine as _sa_create_engine,
 )
 from sqlalchemy import inspect as sqlalchemy_inspect
 import sqlalchemy.schema
@@ -317,11 +318,45 @@ def validate_model_config(self, config: dict) -> dict:
             )
         return config
 
+    # ------------------------------------------------------------------
+    # Engine creation
+    # ------------------------------------------------------------------
+
+    def create_engine(self, connection_string: str, **kwargs: Any) -> Any:
+        """Create a SQLAlchemy engine for this backend.
+
+        The base implementation is a plain pass-through to
+        :func:`sqlalchemy.create_engine`. Subclasses override this to inject
+        backend-specific engine options (e.g. ``fast_executemany`` for MSSQL,
+        ``local_infile`` for MySQL).
+
+        This method is only called when ``DataModel`` is given a
+        ``connection_string``; callers that supply a pre-built ``db_engine``
+        bypass it entirely.
+
+        Args:
+            connection_string: A SQLAlchemy database URL.
+            **kwargs: Additional keyword arguments forwarded to
+                :func:`sqlalchemy.create_engine`.
+
+        Returns:
+            A :class:`sqlalchemy.Engine` instance.
+        """
+        return _sa_create_engine(connection_string, **kwargs)
+
     # ------------------------------------------------------------------
     # Data loading
     # ------------------------------------------------------------------
 
-    def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
+    def bulk_insert(
+        self,
+        conn: Any,
+        table: Any,
+        records: list,
+        *,
+        bulk_load: bool | None = None,
+        bulk_load_threshold: int | None = None,
+    ) -> None:
         """Insert records into a staging table.
 
         The base implementation uses SQLAlchemy's parameterised executemany,
@@ -332,6 +367,11 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
             conn: A SQLAlchemy ``Connection`` already within a transaction.
             table: The SQLAlchemy ``Table`` object to insert into.
             records: A list of dicts mapping column keys to Python values.
+            bulk_load: ``True`` to require bulk loading (raise if unavailable),
+                ``False`` to always use executemany, or ``None`` (default) to
+                use bulk loading when available and fall back silently otherwise.
+            bulk_load_threshold: Minimum number of records to trigger bulk
+                loading.  ``None`` delegates the choice to the subclass.
         """
         if records:
             conn.execute(table.insert(), records)

src/xml2db/dialect/duckdb.py

@@ -20,6 +20,9 @@
 
 from .base import DatabaseDialect
 
+# Records below this count skip read_csv (temp-file overhead).
+_READ_CSV_THRESHOLD = 100
+
 
 class DuckDBDialect(DatabaseDialect):
     """Dialect for DuckDB.
@@ -87,21 +90,42 @@ def _select_expr(self, key: str, col: Any) -> str:
                 return f'CAST("{key}" AS {duckdb_type})'
         return f'"{key}"'  # String / unknown: keep as VARCHAR
 
-    def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
+    def bulk_insert(
+        self,
+        conn: Any,
+        table: Any,
+        records: list,
+        *,
+        bulk_load: bool | None = None,
+        bulk_load_threshold: int | None = None,
+    ) -> None:
         """Bulk-insert records via a temporary CSV file and DuckDB's ``read_csv``.
 
         All CSV columns are read as VARCHAR (``all_varchar=true``) and then
         explicitly cast to their target types in the ``SELECT`` clause.
         Binary columns are hex-encoded in the CSV and decoded with ``unhex()``.
 
+        Falls back to the base-class parameterised executemany when
+        ``bulk_load=False`` or the batch is below the effective threshold.
+
         Args:
             conn: A SQLAlchemy ``Connection`` already within a transaction.
             table: The SQLAlchemy ``Table`` object to insert into.
             records: A list of dicts mapping column keys to Python values.
+            bulk_load: ``True`` or ``None`` (default) to use read_csv for
+                batches at or above the threshold; ``False`` to always use
+                executemany.
+            bulk_load_threshold: Override the minimum batch size.  Defaults to
+                :data:`_READ_CSV_THRESHOLD` (100).
         """
         if not records:
             return
 
+        threshold = bulk_load_threshold if bulk_load_threshold is not None else _READ_CSV_THRESHOLD
+        if bulk_load is False or len(records) < threshold:
+            super().bulk_insert(conn, table, records)
+            return
+
         # Map column key -> SQLAlchemy Column object
         col_by_key = {col.key: col for col in table.columns}
 

src/xml2db/dialect/mssql.py

@@ -14,7 +14,7 @@
 
 
 # Records below this count go through fast_executemany; at or above, BCP is used.
-_BCP_THRESHOLD = 1000
+_BCP_THRESHOLD = 100
 
 
 class MSSQLDialect(DatabaseDialect):
@@ -25,19 +25,14 @@ class MSSQLDialect(DatabaseDialect):
     handled in this class.
 
     When the ``bcp`` utility is available on PATH and the connection uses SQL
-    authentication, :meth:`bulk_insert` switches to BCP for batches of
-    :data:`_BCP_THRESHOLD` rows or more. Smaller batches always use
-    ``fast_executemany`` (enabled at engine level) to avoid BCP's subprocess
-    overhead. Pass ``use_bcp=False`` to :class:`~xml2db.DataModel` to disable
-    BCP entirely.
+    authentication or a trusted connection, :meth:`bulk_insert` switches to BCP
+    for batches of :data:`_BCP_THRESHOLD` rows or more. Smaller batches always
+    use ``fast_executemany`` (enabled at engine level) to avoid BCP's subprocess
+    overhead.
     """
 
     MAX_IDENTIFIER_LENGTH: int = 128
 
-    def __init__(self, use_bcp: bool = True, **kwargs):
-        super().__init__(**kwargs)
-        self.bcp_path = shutil.which("bcp") if use_bcp else None
-
     def validate_table_config(self, config: dict) -> dict:
         """Allow ``as_columnstore`` through unchanged for MSSQL."""
         return config
@@ -93,6 +88,12 @@ def relation_extra_indexes(
             ),
         )
 
+    def create_engine(self, connection_string: str, **kwargs: Any) -> Any:
+        """Create a MSSQL engine with ``fast_executemany`` and ``SERIALIZABLE`` isolation."""
+        kwargs.setdefault("fast_executemany", True)
+        kwargs.setdefault("isolation_level", "SERIALIZABLE")
+        return super().create_engine(connection_string, **kwargs)
+
     @staticmethod
     def _format_bcp_value(v: Any) -> str:
         """Format a Python value as a BCP character-mode field (tab-separated)."""
@@ -108,32 +109,71 @@ def _format_bcp_value(v: Any) -> str:
         # Tab is the field delimiter; replace any occurrence in string values.
         return s.replace("\t", " ")
 
-    def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
+    def bulk_insert(
+        self,
+        conn: Any,
+        table: Any,
+        records: list,
+        *,
+        bulk_load: bool | None = None,
+        bulk_load_threshold: int | None = None,
+    ) -> None:
         """Bulk-insert records, using BCP for large batches when available.
 
-        Batches smaller than :data:`_BCP_THRESHOLD` rows, or any batch when
-        BCP is unavailable or the connection has neither SQL credentials nor
-        ``Trusted_Connection=yes``, fall back to ``fast_executemany``.
-        SQL auth uses ``-U``/``-P``; Kerberos/Windows auth uses ``-T``.
+        Batches smaller than the effective threshold, or any batch when
+        ``bulk_load=False``, use ``fast_executemany`` unconditionally.
 
+        When ``bulk_load=True`` and the batch meets the threshold but BCP
+        prerequisites are not satisfied (binary not on PATH, unsupported auth),
+        a :class:`RuntimeError` is raised with an actionable message.
+
+        SQL auth uses ``-U``/``-P``; Kerberos/Windows auth uses ``-T``.
         BCP does not participate in the caller's SQLAlchemy transaction.
 
         Args:
             conn: A SQLAlchemy ``Connection`` already within a transaction.
             table: The SQLAlchemy ``Table`` object to insert into.
             records: A list of dicts mapping column keys to Python values.
+            bulk_load: ``True`` to require BCP (raise if unavailable for this
+                batch), ``False`` to always use fast_executemany, or ``None``
+                (default) to use BCP when available and fall back silently.
+            bulk_load_threshold: Override the minimum batch size for BCP.
+                Defaults to :data:`_BCP_THRESHOLD` (100).
         """
         if not records:
             return
 
+        threshold = bulk_load_threshold if bulk_load_threshold is not None else _BCP_THRESHOLD
+
+        # bulk_load=False or batch too small → always use fast_executemany.
+        if bulk_load is False or len(records) < threshold:
+            super().bulk_insert(conn, table, records)
+            return
+
+        # Check BCP prerequisites.
         url = conn.engine.url
         trusted = str(url.query.get("Trusted_Connection", "")).lower() == "yes"
         has_sql_auth = bool(url.username and url.password)
-        if (
-            self.bcp_path is None
-            or len(records) < _BCP_THRESHOLD
-            or (not has_sql_auth and not trusted)
-        ):
+        bcp_path = shutil.which("bcp")
+
+        if bcp_path is None:
+            if bulk_load is True:
+                raise RuntimeError(
+                    "bulk_load=True requires the bcp utility on PATH. "
+                    "Install mssql-tools (Linux/macOS) or SQL Server Command Line Utilities "
+                    "(Windows), or set bulk_load=False to use fast_executemany instead."
+                )
+            super().bulk_insert(conn, table, records)
+            return
+
+        if not has_sql_auth and not trusted:
+            if bulk_load is True:
+                raise RuntimeError(
+                    "bulk_load=True requires SQL Server authentication (username and password "
+                    "in the connection string) or a Windows/Kerberos trusted connection "
+                    "(Trusted_Connection=yes in the connection string query parameters). "
+                    "Set bulk_load=False to use fast_executemany instead."
+                )
             super().bulk_insert(conn, table, records)
             return
 
@@ -167,7 +207,7 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                     f.write("\t".join(row) + "\n")
 
             cmd = [
-                self.bcp_path, full_name, "in", data_path,
+                bcp_path, full_name, "in", data_path,
                 "-S", f"{url.host},{url.port or 1433}",
                 "-d", url.database,
                 "-c",       # character (text) mode