From 63a1feb6954812a15acc0aadd0c8e9918f232609 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 20:26:38 +0000
Subject: [PATCH 01/12] Add PostgreSQL bulk load via COPY FROM STDIN

Overrides bulk_insert() in PostgreSQLDialect to stream an in-memory CSV
payload to PostgreSQL using the native COPY protocol instead of SQLAlchemy
executemany. Supports both psycopg2 (copy_expert) and psycopg3 (cursor.copy),
falling back to the base-class implementation for other drivers.

Adds integration tests parametrised over both drivers, and updates the
PostgreSQL CI workflow to a matrix that runs once per driver.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 .../workflows/integration-tests-postgres.yml  |  21 +-
 src/xml2db/dialect/postgresql.py              |  86 ++++++
 tests/test_bulk_insert_postgres.py            | 257 ++++++++++++++++++
 3 files changed, 358 insertions(+), 6 deletions(-)
 create mode 100644 tests/test_bulk_insert_postgres.py

diff --git a/.github/workflows/integration-tests-postgres.yml b/.github/workflows/integration-tests-postgres.yml
index 5598f96..702fe16 100644
--- a/.github/workflows/integration-tests-postgres.yml
+++ b/.github/workflows/integration-tests-postgres.yml
@@ -10,6 +10,15 @@ on:
 jobs:
   integration-tests:
     runs-on: ubuntu-24.04
+    strategy:
+      matrix:
+        include:
+          - driver: psycopg2
+            package: psycopg2
+            db_string: postgresql+psycopg2://postgres:postgres@localhost:5432/postgres
+          - driver: psycopg
+            package: "psycopg[binary]"
+            db_string: postgresql+psycopg://postgres:postgres@localhost:5432/postgres
     services:
       postgres:
         image: postgres
@@ -25,23 +34,23 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
-          
+
     steps:
       - name: Check out repository code
         uses: actions/checkout@v4
-      
+
       - name: Set up Python 3.12
         uses: actions/setup-python@v5
         with:
           python-version: 3.12
-          
+
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          python -m pip install flake8 .[tests] psycopg2
-    
+          python -m pip install flake8 .[tests] "${{ matrix.package }}"
+
       - name: Test with pytest
         run: |
           pytest tests -x
         env:
-          DB_STRING: postgresql+psycopg2://postgres:postgres@localhost:5432/postgres
+          DB_STRING: ${{ matrix.db_string }}
diff --git a/src/xml2db/dialect/postgresql.py b/src/xml2db/dialect/postgresql.py
index 23a9cfe..d909e30 100644
--- a/src/xml2db/dialect/postgresql.py
+++ b/src/xml2db/dialect/postgresql.py
@@ -1,3 +1,7 @@
+import csv
+import io
+from typing import Any
+
 from .base import DatabaseDialect
 
 
@@ -11,3 +15,85 @@ class PostgreSQLDialect(DatabaseDialect):
     """
 
     MAX_IDENTIFIER_LENGTH: int = 63
+
+    def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
+        """Bulk-insert records via PostgreSQL's ``COPY FROM STDIN``.
+
+        Builds an in-memory CSV payload and streams it to the server using
+        the driver's native COPY protocol.  Supported drivers:
+
+        - **psycopg2** — uses ``cursor.copy_expert()``.
+        - **psycopg** (psycopg3) — uses ``cursor.copy()``.
+
+        Falls back to the base-class parameterised executemany for any other
+        driver.
+
+        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.
+        """
+        if not records:
+            return
+
+        driver = conn.dialect.driver
+        if driver not in ("psycopg2", "psycopg"):
+            super().bulk_insert(conn, table, records)
+            return
+
+        col_by_key = {col.key: col for col in table.columns}
+        col_keys = [k for k in records[0] if k in col_by_key]
+
+        # Python-side scalar defaults absent from records (e.g. default=False).
+        # executemany applies these automatically; our COPY path must do it manually.
+        extra_defaults: dict = {}
+        for col in table.columns:
+            if col.key not in records[0] and col.key in col_by_key:
+                d = col.default
+                if d is not None and d.is_scalar:
+                    extra_defaults[col.key] = d.arg
+
+        all_col_keys = col_keys + list(extra_defaults.keys())
+
+        buf = io.StringIO()
+        writer = csv.writer(buf)
+        writer.writerow(all_col_keys)
+        for record in records:
+            row = []
+            for key in all_col_keys:
+                v = record.get(key) if key in col_keys else extra_defaults[key]
+                if v is None:
+                    row.append("")
+                elif isinstance(v, bytes):
+                    # PostgreSQL bytea hex format: \xDEADBEEF
+                    row.append("\\x" + v.hex())
+                elif isinstance(v, bool):
+                    # bool must precede the general str() path: bool subclasses int,
+                    # so str(True) → '1' which PostgreSQL COPY rejects for boolean.
+                    row.append("true" if v else "false")
+                else:
+                    # str() on datetime → "YYYY-MM-DD HH:MM:SS[.f][±HH:MM]",
+                    # which PostgreSQL's text input parser accepts.
+                    row.append(str(v))
+            writer.writerow(row)
+        buf.seek(0)
+
+        col_names = ", ".join(f'"{col_by_key[k].name}"' for k in all_col_keys)
+        full_name = (
+            f'"{table.schema}"."{table.name}"'
+            if table.schema
+            else f'"{table.name}"'
+        )
+        copy_sql = (
+            f"COPY {full_name} ({col_names}) FROM STDIN "
+            f"WITH (FORMAT CSV, HEADER, NULL '')"
+        )
+
+        raw_conn = conn.connection.dbapi_connection
+        if driver == "psycopg2":
+            cur = raw_conn.cursor()
+            cur.copy_expert(copy_sql, buf)
+        else:  # psycopg3
+            cur = raw_conn.cursor()
+            with cur.copy(copy_sql) as copy:
+                copy.write(buf.read().encode("utf-8"))
diff --git a/tests/test_bulk_insert_postgres.py b/tests/test_bulk_insert_postgres.py
new file mode 100644
index 0000000..60c534a
--- /dev/null
+++ b/tests/test_bulk_insert_postgres.py
@@ -0,0 +1,257 @@
+"""Integration tests for PostgreSQLDialect.bulk_insert with psycopg2 and psycopg3."""
+import datetime
+import os
+
+import pytest
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    Column,
+    DateTime,
+    Double,
+    Integer,
+    LargeBinary,
+    MetaData,
+    SmallInteger,
+    String,
+    Table,
+    create_engine,
+    select,
+)
+from sqlalchemy.engine import make_url
+
+from xml2db.dialect.postgresql import PostgreSQLDialect
+
+
+def _make_pg_engine(driver: str):
+    """Return a PostgreSQL engine using the given driver, or skip."""
+    db_string = os.getenv("DB_STRING", "")
+    if not db_string:
+        pytest.skip("DB_STRING not set")
+    url = make_url(db_string)
+    if url.get_dialect().name != "postgresql":
+        pytest.skip("DB_STRING is not a PostgreSQL connection")
+    if driver == "psycopg2":
+        pytest.importorskip("psycopg2")
+    else:
+        pytest.importorskip("psycopg")
+    url = url.set(drivername=f"postgresql+{driver}")
+    return create_engine(url)
+
+
+@pytest.fixture(params=["psycopg2", "psycopg"])
+def pg_engine(request):
+    engine = _make_pg_engine(request.param)
+    yield engine
+    engine.dispose()
+
+
+def _make_table(engine, name, *extra_cols):
+    meta = MetaData()
+    table = Table(
+        name,
+        meta,
+        Column("id", Integer, key="id"),
+        Column("label", String(200), key="label"),
+        *extra_cols,
+    )
+    meta.create_all(engine)
+    return table, meta
+
+
+def _roundtrip(engine, table, records):
+    dialect = PostgreSQLDialect()
+    with engine.begin() as conn:
+        dialect.bulk_insert(conn, table, records)
+    with engine.connect() as conn:
+        return conn.execute(select(table).order_by(table.c.id)).mappings().all()
+
+
+def _drop(meta, engine):
+    meta.drop_all(engine)
+
+
+# ---------------------------------------------------------------------------
+# Basic types
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_basic(pg_engine):
+    table, meta = _make_table(pg_engine, "pg_bi_basic")
+    try:
+        records = [{"id": 1, "label": "hello"}, {"id": 2, "label": None}]
+        rows = _roundtrip(pg_engine, table, records)
+        assert len(rows) == 2
+        assert rows[0]["id"] == 1
+        assert rows[0]["label"] == "hello"
+        assert rows[1]["label"] is None
+    finally:
+        _drop(meta, pg_engine)
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_empty(pg_engine):
+    table, meta = _make_table(pg_engine, "pg_bi_empty")
+    try:
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, [])
+        with pg_engine.connect() as conn:
+            count = conn.execute(select(table)).rowcount
+        assert count == 0
+    finally:
+        _drop(meta, pg_engine)
+
+
+# ---------------------------------------------------------------------------
+# Numeric types
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_numeric_types(pg_engine):
+    meta = MetaData()
+    table = Table(
+        "pg_bi_numeric",
+        meta,
+        Column("i", Integer, key="i"),
+        Column("bi", BigInteger, key="bi"),
+        Column("si", SmallInteger, key="si"),
+        Column("d", Double, key="d"),
+    )
+    meta.create_all(pg_engine)
+    try:
+        records = [{"i": 1, "bi": 10**15, "si": 32767, "d": 3.14}]
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with pg_engine.connect() as conn:
+            rows = conn.execute(select(table)).mappings().all()
+        assert rows[0]["i"] == 1
+        assert rows[0]["bi"] == 10**15
+        assert rows[0]["si"] == 32767
+        assert abs(rows[0]["d"] - 3.14) < 1e-9
+    finally:
+        meta.drop_all(pg_engine)
+
+
+# ---------------------------------------------------------------------------
+# Boolean
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_boolean(pg_engine):
+    meta = MetaData()
+    table = Table(
+        "pg_bi_bool",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, key="flag"),
+    )
+    meta.create_all(pg_engine)
+    try:
+        records = [
+            {"id": 1, "flag": True},
+            {"id": 2, "flag": False},
+            {"id": 3, "flag": None},
+        ]
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with pg_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert rows[0]["flag"] is True
+        assert rows[1]["flag"] is False
+        assert rows[2]["flag"] is None
+    finally:
+        meta.drop_all(pg_engine)
+
+
+# ---------------------------------------------------------------------------
+# DateTime
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_datetime(pg_engine):
+    meta = MetaData()
+    table = Table(
+        "pg_bi_dt",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("ts", DateTime(timezone=True), key="ts"),
+    )
+    meta.create_all(pg_engine)
+    try:
+        dt = datetime.datetime(2023, 9, 27, 14, 35, 54, 274602,
+                               tzinfo=datetime.timezone.utc)
+        records = [{"id": 1, "ts": dt}, {"id": 2, "ts": None}]
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with pg_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert rows[0]["ts"] is not None
+        assert rows[1]["ts"] is None
+    finally:
+        meta.drop_all(pg_engine)
+
+
+# ---------------------------------------------------------------------------
+# Binary (bytea)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_binary(pg_engine):
+    meta = MetaData()
+    table = Table(
+        "pg_bi_binary",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("data", LargeBinary, key="data"),
+    )
+    meta.create_all(pg_engine)
+    try:
+        payload = b"\xde\xad\xbe\xef" * 8
+        records = [{"id": 1, "data": payload}, {"id": 2, "data": None}]
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with pg_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert bytes(rows[0]["data"]) == payload
+        assert rows[1]["data"] is None
+    finally:
+        meta.drop_all(pg_engine)
+
+
+# ---------------------------------------------------------------------------
+# Python-side scalar column defaults
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_pg_bulk_insert_scalar_column_default(pg_engine):
+    meta = MetaData()
+    table = Table(
+        "pg_bi_default",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, default=False, key="flag"),
+    )
+    meta.create_all(pg_engine)
+    try:
+        # Records do NOT contain 'flag'; the default must be applied.
+        records = [{"id": 1}, {"id": 2}]
+        dialect = PostgreSQLDialect()
+        with pg_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with pg_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert rows[0]["flag"] is False
+        assert rows[1]["flag"] is False
+    finally:
+        meta.drop_all(pg_engine)

From 2c67c932a43201c4871a75f3260833e5da5ad6a8 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 21:25:03 +0000
Subject: [PATCH 02/12] Add MSSQL bulk load via BCP for large batches

Adds a BCP-based bulk_insert() to MSSQLDialect. When the bcp binary is
available on PATH and the connection uses SQL authentication, batches of
1000+ rows are loaded via a subprocess call to bcp with a temp file
(tab-separated, UTF-8), falling back to fast_executemany for smaller
batches or when bcp is unavailable. Pass use_bcp=False to DataModel to
disable BCP entirely.

Also threads use_bcp through DataModel.__init__() -> get_dialect() ->
dialect constructor, adding **kwargs forwarding to DatabaseDialect so
unknown options are silently ignored by other dialects.

Updates the MSSQL CI workflow to install mssql-tools18 (provides bcp)
and adds it to PATH for the test step.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 .github/workflows/integration-tests-mssql.yml |   3 +-
 src/xml2db/dialect/__init__.py                |   7 +-
 src/xml2db/dialect/base.py                    |   3 +
 src/xml2db/dialect/mssql.py                   | 112 +++++++
 src/xml2db/model.py                           |   3 +-
 tests/test_bulk_insert_mssql.py               | 292 ++++++++++++++++++
 6 files changed, 416 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_bulk_insert_mssql.py

diff --git a/.github/workflows/integration-tests-mssql.yml b/.github/workflows/integration-tests-mssql.yml
index 6780a82..1546179 100644
--- a/.github/workflows/integration-tests-mssql.yml
+++ b/.github/workflows/integration-tests-mssql.yml
@@ -35,7 +35,7 @@ jobs:
           sudo dpkg -i packages-microsoft-prod.deb
           rm packages-microsoft-prod.deb
           sudo apt-get update
-          sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18
+          sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18 mssql-tools18
           python -m pip install --upgrade pip
           python -m pip install flake8 .[tests] pyodbc
     
@@ -44,3 +44,4 @@ jobs:
           pytest tests -x
         env:
           DB_STRING: mssql+pyodbc://sa:MyTestPassword1@localhost:1433/master?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes
+          PATH: /opt/mssql-tools18/bin:${{ env.PATH }}
diff --git a/src/xml2db/dialect/__init__.py b/src/xml2db/dialect/__init__.py
index c783f3a..b881586 100644
--- a/src/xml2db/dialect/__init__.py
+++ b/src/xml2db/dialect/__init__.py
@@ -49,7 +49,7 @@
 }
 
 
-def get_dialect(db_type: str | None) -> DatabaseDialect:
+def get_dialect(db_type: str | None, **kwargs) -> DatabaseDialect:
     """Return a :class:`DatabaseDialect` instance for the given backend name.
 
     Args:
@@ -57,9 +57,12 @@ def get_dialect(db_type: str | None) -> DatabaseDialect:
             ``"mssql"``, ``"mysql"``, ``"duckdb"``. ``None`` or any
             unrecognised string falls back to the base
             :class:`DatabaseDialect`, which uses safe generic defaults.
+        **kwargs: Extra keyword arguments forwarded to the dialect constructor.
+            Unknown kwargs are silently ignored by subclasses that do not
+            declare them.
 
     Returns:
         An instantiated :class:`DatabaseDialect` (or subclass) ready for use.
     """
     cls = DIALECT_REGISTRY.get(db_type, DatabaseDialect)
-    return cls()
+    return cls(**kwargs)
diff --git a/src/xml2db/dialect/base.py b/src/xml2db/dialect/base.py
index 3ff9e42..b9c566d 100644
--- a/src/xml2db/dialect/base.py
+++ b/src/xml2db/dialect/base.py
@@ -39,6 +39,9 @@ class DatabaseDialect:
 
     MAX_IDENTIFIER_LENGTH: int = 63  # conservative default; matches PostgreSQL
 
+    def __init__(self, **kwargs):
+        pass
+
     # ------------------------------------------------------------------
     # Identifier handling
     # ------------------------------------------------------------------
diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index d5657c2..8ec7732 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -1,3 +1,7 @@
+import os
+import shutil
+import subprocess
+import tempfile
 from typing import Any, List, TYPE_CHECKING
 
 from sqlalchemy import Index
@@ -9,16 +13,31 @@
     from ..table.column import DataModelColumn
 
 
+# Records below this count go through fast_executemany; at or above, BCP is used.
+_BCP_THRESHOLD = 1000
+
+
 class MSSQLDialect(DatabaseDialect):
     """Dialect for Microsoft SQL Server.
 
     MSSQL supports identifiers up to 128 characters, so no truncation is
     needed. Columnstore index support and MSSQL-specific type mappings are
     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.
     """
 
     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
@@ -73,3 +92,96 @@ def relation_extra_indexes(
                 mssql_clustered=True,
             ),
         )
+
+    @staticmethod
+    def _format_bcp_value(v: Any) -> str:
+        """Format a Python value as a BCP character-mode field (tab-separated)."""
+        if v is None:
+            return ""
+        if isinstance(v, bool):
+            # Must precede int: bool is a subclass of int.
+            return "1" if v else "0"
+        if isinstance(v, bytes):
+            return "0x" + v.hex()
+        s = str(v)
+        # 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:
+        """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 lacks SQL auth credentials, are
+        handled by the base-class ``fast_executemany`` path.
+
+        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.
+        """
+        if not records:
+            return
+
+        url = conn.engine.url
+        if (
+            self.bcp_path is None
+            or len(records) < _BCP_THRESHOLD
+            or not url.username
+            or not url.password
+        ):
+            super().bulk_insert(conn, table, records)
+            return
+
+        col_by_key = {col.key: col for col in table.columns}
+        col_keys = [k for k in records[0] if k in col_by_key]
+
+        # Python-side scalar defaults absent from records (same as other dialects).
+        extra_defaults: dict = {}
+        for col in table.columns:
+            if col.key not in records[0] and col.key in col_by_key:
+                d = col.default
+                if d is not None and d.is_scalar:
+                    extra_defaults[col.key] = d.arg
+
+        all_col_keys = col_keys + list(extra_defaults.keys())
+
+        full_name = (
+            f"[{table.schema}].[{table.name}]"
+            if table.schema
+            else f"[{table.name}]"
+        )
+
+        fd, data_path = tempfile.mkstemp(suffix=".bcp")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
+                for record in records:
+                    row = []
+                    for key in all_col_keys:
+                        v = record.get(key) if key in col_keys else extra_defaults[key]
+                        row.append(self._format_bcp_value(v))
+                    f.write("\t".join(row) + "\n")
+
+            cmd = [
+                self.bcp_path, full_name, "in", data_path,
+                "-S", f"{url.host},{url.port or 1433}",
+                "-d", url.database,
+                "-U", url.username,
+                "-P", url.password,
+                "-c",       # character (text) mode
+                "-t", "\t", # tab field separator
+                "-r", "\n", # newline row terminator
+                "-k",       # empty field → NULL (not column default)
+                "-b", "10000",
+                "-C", "65001",  # UTF-8
+            ]
+            result = subprocess.run(cmd, capture_output=True, text=True)
+            if result.returncode != 0:
+                raise RuntimeError(
+                    f"BCP failed (exit {result.returncode}):\n"
+                    f"{result.stdout}\n{result.stderr}"
+                )
+        finally:
+            if os.path.exists(data_path):
+                os.unlink(data_path)
diff --git a/src/xml2db/model.py b/src/xml2db/model.py
index de0edd7..0e9f6f3 100644
--- a/src/xml2db/model.py
+++ b/src/xml2db/model.py
@@ -80,6 +80,7 @@ def __init__(
         db_type: str = None,
         db_schema: str = None,
         temp_prefix: str = None,
+        use_bcp: bool = True,
     ):
         self.model_config = self._validate_config(model_config)
         self.tables_config = model_config.get("tables", {}) if model_config else {}
@@ -118,7 +119,7 @@ def __init__(
                 )
             self.db_type = self.engine.dialect.name
 
-        self.dialect = get_dialect(self.db_type)
+        self.dialect = get_dialect(self.db_type, use_bcp=use_bcp)
         self.model_config = self.dialect.validate_model_config(self.model_config)
         self.db_schema = db_schema
         self.temp_prefix = str(uuid4())[:8] if temp_prefix is None else temp_prefix
diff --git a/tests/test_bulk_insert_mssql.py b/tests/test_bulk_insert_mssql.py
new file mode 100644
index 0000000..45bbc93
--- /dev/null
+++ b/tests/test_bulk_insert_mssql.py
@@ -0,0 +1,292 @@
+"""Integration tests for MSSQLDialect.bulk_insert (fast_executemany and BCP)."""
+import os
+
+import pytest
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    Column,
+    Integer,
+    LargeBinary,
+    MetaData,
+    SmallInteger,
+    String,
+    Table,
+    create_engine,
+    select,
+)
+from sqlalchemy.engine import make_url
+
+from xml2db.dialect.mssql import MSSQLDialect, _BCP_THRESHOLD
+
+
+def _make_mssql_engine():
+    db_string = os.getenv("DB_STRING", "")
+    if not db_string:
+        pytest.skip("DB_STRING not set")
+    url = make_url(db_string)
+    if url.get_dialect().name != "mssql":
+        pytest.skip("DB_STRING is not an MSSQL connection")
+    pytest.importorskip("pyodbc")
+    return create_engine(url, fast_executemany=True)
+
+
+@pytest.fixture(scope="module")
+def mssql_engine():
+    engine = _make_mssql_engine()
+    yield engine
+    engine.dispose()
+
+
+def _make_table(engine, name, *extra_cols):
+    meta = MetaData()
+    table = Table(
+        name,
+        meta,
+        Column("id", Integer, key="id"),
+        Column("label", String(200), key="label"),
+        *extra_cols,
+    )
+    meta.create_all(engine)
+    return table, meta
+
+
+def _insert_and_read(engine, table, records, use_bcp=False):
+    dialect = MSSQLDialect(use_bcp=use_bcp)
+    with engine.begin() as conn:
+        dialect.bulk_insert(conn, table, records)
+    with engine.connect() as conn:
+        return conn.execute(select(table).order_by(table.c.id)).mappings().all()
+
+
+def _drop(meta, engine):
+    meta.drop_all(engine)
+
+
+# ---------------------------------------------------------------------------
+# fast_executemany path (< _BCP_THRESHOLD rows)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_mssql_fast_executemany_basic(mssql_engine):
+    table, meta = _make_table(mssql_engine, "mssql_bi_basic_fem")
+    try:
+        records = [{"id": 1, "label": "hello"}, {"id": 2, "label": None}]
+        rows = _insert_and_read(mssql_engine, table, records)
+        assert len(rows) == 2
+        assert rows[0]["label"] == "hello"
+        assert rows[1]["label"] is None
+    finally:
+        _drop(meta, mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_fast_executemany_numeric(mssql_engine):
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_numeric_fem",
+        meta,
+        Column("i", Integer, key="i"),
+        Column("bi", BigInteger, key="bi"),
+        Column("si", SmallInteger, key="si"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        records = [{"i": 1, "bi": 10**15, "si": 32767}]
+        rows = _insert_and_read(mssql_engine, table, records)
+        assert rows[0]["i"] == 1
+        assert rows[0]["bi"] == 10**15
+        assert rows[0]["si"] == 32767
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_fast_executemany_boolean(mssql_engine):
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_bool_fem",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, key="flag"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        records = [{"id": 1, "flag": True}, {"id": 2, "flag": False}, {"id": 3, "flag": None}]
+        rows = _insert_and_read(mssql_engine, table, records)
+        assert rows[0]["flag"] is True
+        assert rows[1]["flag"] is False
+        assert rows[2]["flag"] is None
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_fast_executemany_scalar_default(mssql_engine):
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_default_fem",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, default=False, key="flag"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        records = [{"id": 1}, {"id": 2}]
+        rows = _insert_and_read(mssql_engine, table, records)
+        assert rows[0]["flag"] is False
+        assert rows[1]["flag"] is False
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+# ---------------------------------------------------------------------------
+# BCP path (>= _BCP_THRESHOLD rows)
+# ---------------------------------------------------------------------------
+
+
+def _require_bcp():
+    import shutil
+    if shutil.which("bcp") is None:
+        pytest.skip("bcp not found on PATH")
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_basic(mssql_engine):
+    _require_bcp()
+    table, meta = _make_table(mssql_engine, "mssql_bi_basic_bcp")
+    try:
+        records = [{"id": i, "label": f"row{i}"} for i in range(_BCP_THRESHOLD)]
+        records.append({"id": _BCP_THRESHOLD, "label": None})
+        rows = _insert_and_read(mssql_engine, table, records, use_bcp=True)
+        assert len(rows) == _BCP_THRESHOLD + 1
+        assert rows[0]["label"] == "row0"
+        assert rows[-1]["label"] is None
+    finally:
+        _drop(meta, mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_numeric(mssql_engine):
+    _require_bcp()
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_numeric_bcp",
+        meta,
+        Column("i", Integer, key="i"),
+        Column("bi", BigInteger, key="bi"),
+        Column("si", SmallInteger, key="si"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        base = [{"i": j, "bi": 10**15 + j, "si": j % 32767} for j in range(_BCP_THRESHOLD)]
+        dialect = MSSQLDialect(use_bcp=True)
+        with mssql_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, base)
+        with mssql_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.i)).mappings().all()
+        assert rows[0]["bi"] == 10**15
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_boolean(mssql_engine):
+    _require_bcp()
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_bool_bcp",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, key="flag"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        records = (
+            [{"id": 0, "flag": True}, {"id": 1, "flag": False}, {"id": 2, "flag": None}]
+            + [{"id": i + 3, "flag": i % 2 == 0} for i in range(_BCP_THRESHOLD)]
+        )
+        dialect = MSSQLDialect(use_bcp=True)
+        with mssql_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with mssql_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert rows[0]["flag"] is True
+        assert rows[1]["flag"] is False
+        assert rows[2]["flag"] is None
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_binary(mssql_engine):
+    _require_bcp()
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_binary_bcp",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("data", LargeBinary, key="data"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        payload = b"\xde\xad\xbe\xef" * 8
+        records = (
+            [{"id": 0, "data": payload}, {"id": 1, "data": None}]
+            + [{"id": i + 2, "data": payload} for i in range(_BCP_THRESHOLD)]
+        )
+        dialect = MSSQLDialect(use_bcp=True)
+        with mssql_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with mssql_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert bytes(rows[0]["data"]) == payload
+        assert rows[1]["data"] is None
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_scalar_default(mssql_engine):
+    _require_bcp()
+    meta = MetaData()
+    table = Table(
+        "mssql_bi_default_bcp",
+        meta,
+        Column("id", Integer, key="id"),
+        Column("flag", Boolean, default=False, key="flag"),
+    )
+    meta.create_all(mssql_engine)
+    try:
+        records = [{"id": i} for i in range(_BCP_THRESHOLD)]
+        dialect = MSSQLDialect(use_bcp=True)
+        with mssql_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with mssql_engine.connect() as conn:
+            rows = conn.execute(select(table).order_by(table.c.id)).mappings().all()
+        assert all(r["flag"] is False for r in rows)
+    finally:
+        meta.drop_all(mssql_engine)
+
+
+# ---------------------------------------------------------------------------
+# BCP fallback when bcp binary is not found
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.dbtest
+def test_mssql_bcp_fallback_when_unavailable(mssql_engine):
+    """With bcp_path=None, bulk_insert must still work via fast_executemany."""
+    table, meta = _make_table(mssql_engine, "mssql_bi_fallback")
+    try:
+        records = [{"id": i, "label": f"r{i}"} for i in range(_BCP_THRESHOLD)]
+        dialect = MSSQLDialect(use_bcp=True)
+        dialect.bcp_path = None  # simulate bcp not on PATH
+        with mssql_engine.begin() as conn:
+            dialect.bulk_insert(conn, table, records)
+        with mssql_engine.connect() as conn:
+            count = conn.execute(select(table)).rowcount
+        assert count == _BCP_THRESHOLD
+    finally:
+        _drop(meta, mssql_engine)

From 0332942f056e53c22328440c2d4ab2efa4c57562 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 21:34:31 +0000
Subject: [PATCH 03/12] Temporarily disable non-MSSQL CI workflows during
 iteration

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 .github/workflows/integration-tests-duckdb.yml   | 4 ----
 .github/workflows/integration-tests-mysql.yml    | 4 ----
 .github/workflows/integration-tests-postgres.yml | 4 ----
 .github/workflows/python-package.yml             | 5 +----
 4 files changed, 1 insertion(+), 16 deletions(-)

diff --git a/.github/workflows/integration-tests-duckdb.yml b/.github/workflows/integration-tests-duckdb.yml
index 7c63c5b..6d2816a 100644
--- a/.github/workflows/integration-tests-duckdb.yml
+++ b/.github/workflows/integration-tests-duckdb.yml
@@ -1,10 +1,6 @@
 name: Duckdb integration tests
 
 on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/integration-tests-mysql.yml b/.github/workflows/integration-tests-mysql.yml
index f1a8d89..f701bd1 100644
--- a/.github/workflows/integration-tests-mysql.yml
+++ b/.github/workflows/integration-tests-mysql.yml
@@ -1,10 +1,6 @@
 name: MySQL integration tests
 
 on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/integration-tests-postgres.yml b/.github/workflows/integration-tests-postgres.yml
index 702fe16..9344f27 100644
--- a/.github/workflows/integration-tests-postgres.yml
+++ b/.github/workflows/integration-tests-postgres.yml
@@ -1,10 +1,6 @@
 name: PostgreSQL integration tests
 
 on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index f4328ce..094dd8e 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -4,10 +4,7 @@
 name: Python package
 
 on:
-  push:
-    branches: [ "main" ]
-  pull_request:
-    branches: [ "main" ]
+  workflow_dispatch:
   workflow_dispatch:
 
 jobs:

From 892963e621adfd708e2deb85400ae4a7c4ea125b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 21:40:00 +0000
Subject: [PATCH 04/12] Fix test_mssql_fast_executemany_numeric: add missing id
 column

The _insert_and_read helper orders by table.c.id; the numeric test table
was missing that column.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 tests/test_bulk_insert_mssql.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/test_bulk_insert_mssql.py b/tests/test_bulk_insert_mssql.py
index 45bbc93..2e66ad7 100644
--- a/tests/test_bulk_insert_mssql.py
+++ b/tests/test_bulk_insert_mssql.py
@@ -87,13 +87,14 @@ def test_mssql_fast_executemany_numeric(mssql_engine):
     table = Table(
         "mssql_bi_numeric_fem",
         meta,
+        Column("id", Integer, key="id"),
         Column("i", Integer, key="i"),
         Column("bi", BigInteger, key="bi"),
         Column("si", SmallInteger, key="si"),
     )
     meta.create_all(mssql_engine)
     try:
-        records = [{"i": 1, "bi": 10**15, "si": 32767}]
+        records = [{"id": 1, "i": 1, "bi": 10**15, "si": 32767}]
         rows = _insert_and_read(mssql_engine, table, records)
         assert rows[0]["i"] == 1
         assert rows[0]["bi"] == 10**15

From 608124b2301ee77083cbc7c6f13c5f20c0e6a1fd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 21:54:55 +0000
Subject: [PATCH 05/12] Fix BCP SSL certificate verification for mssql-tools18

Add -No flag to BCP command when TrustServerCertificate=yes is present
in the connection URL query, allowing BCP to connect to servers with
self-signed certificates (same trust level as pyodbc).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 src/xml2db/dialect/mssql.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index 8ec7732..2f8945d 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -176,6 +176,8 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 "-b", "10000",
                 "-C", "65001",  # UTF-8
             ]
+            if str(url.query.get("TrustServerCertificate", "")).lower() == "yes":
+                cmd.append("-No")  # optional encryption, skip cert verification
             result = subprocess.run(cmd, capture_output=True, text=True)
             if result.returncode != 0:
                 raise RuntimeError(

From 4605317c884009b2ae95421723d91490386036c3 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 21:59:01 +0000
Subject: [PATCH 06/12] Fix BCP trust-cert flag and duplicate workflow_dispatch

Use -u flag (trust server certificate) instead of -No which is not
supported in mssql-tools18 BCP. Also remove duplicate workflow_dispatch
trigger in python-package.yml.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 .github/workflows/python-package.yml | 1 -
 src/xml2db/dialect/mssql.py          | 2 +-
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 094dd8e..03156a4 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -5,7 +5,6 @@ name: Python package
 
 on:
   workflow_dispatch:
-  workflow_dispatch:
 
 jobs:
   build:
diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index 2f8945d..d20d3a3 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -177,7 +177,7 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 "-C", "65001",  # UTF-8
             ]
             if str(url.query.get("TrustServerCertificate", "")).lower() == "yes":
-                cmd.append("-No")  # optional encryption, skip cert verification
+                cmd.append("-u")  # trust server certificate (mssql-tools18)
             result = subprocess.run(cmd, capture_output=True, text=True)
             if result.returncode != 0:
                 raise RuntimeError(

From 56a4bb29954e1567d8325e5694839d2cb5457614 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 22:04:41 +0000
Subject: [PATCH 07/12] Fall back to fast_executemany for binary columns in BCP
 path

BCP character mode cannot load hex-encoded data into varbinary columns
(produces "Text column data incomplete" error). Detect LargeBinary
columns and skip BCP for those tables.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 src/xml2db/dialect/mssql.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index d20d3a3..3292f4c 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -4,7 +4,7 @@
 import tempfile
 from typing import Any, List, TYPE_CHECKING
 
-from sqlalchemy import Index
+from sqlalchemy import Index, LargeBinary
 from sqlalchemy.dialects import mssql as mssql_dialect
 
 from .base import DatabaseDialect
@@ -147,6 +147,11 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
 
         all_col_keys = col_keys + list(extra_defaults.keys())
 
+        # BCP character mode cannot load binary columns; fall back for those.
+        if any(isinstance(col_by_key[k].type, LargeBinary) for k in all_col_keys if k in col_by_key):
+            super().bulk_insert(conn, table, records)
+            return
+
         full_name = (
             f"[{table.schema}].[{table.name}]"
             if table.schema

From fea1fb18eda4bfa518146da822d14d2a67c3d08d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 19 Jun 2026 22:14:51 +0000
Subject: [PATCH 08/12] Use BCP format file with SQLVARBINARY for binary
 columns

Instead of falling back to fast_executemany for binary columns, generate
a non-XML BCP format file that specifies SQLVARBINARY with a 4-byte
length prefix for LargeBinary columns and SQLCHAR for everything else.
The data file is written in binary mode: raw bytes with prefix for binary
fields, UTF-8 text for character fields. NULL binary is encoded as
prefix value -1 (0xFFFFFFFF).

This allows BCP to handle the record hash column (VARBINARY) present in
almost all xml2db tables while keeping the fast character-mode path for
all other column types.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 src/xml2db/dialect/mssql.py | 83 ++++++++++++++++++++++++++++---------
 1 file changed, 64 insertions(+), 19 deletions(-)

diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index 3292f4c..29727af 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -1,5 +1,6 @@
 import os
 import shutil
+import struct
 import subprocess
 import tempfile
 from typing import Any, List, TYPE_CHECKING
@@ -95,14 +96,12 @@ def relation_extra_indexes(
 
     @staticmethod
     def _format_bcp_value(v: Any) -> str:
-        """Format a Python value as a BCP character-mode field (tab-separated)."""
+        """Format a Python value as a BCP character field (non-binary columns only)."""
         if v is None:
             return ""
         if isinstance(v, bool):
             # Must precede int: bool is a subclass of int.
             return "1" if v else "0"
-        if isinstance(v, bytes):
-            return "0x" + v.hex()
         s = str(v)
         # Tab is the field delimiter; replace any occurrence in string values.
         return s.replace("\t", " ")
@@ -114,7 +113,10 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
         BCP is unavailable or the connection lacks SQL auth credentials, are
         handled by the base-class ``fast_executemany`` path.
 
-        BCP does not participate in the caller's SQLAlchemy transaction.
+        BCP uses a non-XML format file so that binary columns (LargeBinary /
+        VARBINARY) are sent as raw bytes with a 4-byte length prefix, while
+        all other columns are sent as UTF-8 character data.  BCP does not
+        participate in the caller's SQLAlchemy transaction.
 
         Args:
             conn: A SQLAlchemy ``Connection`` already within a transaction.
@@ -146,11 +148,13 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                     extra_defaults[col.key] = d.arg
 
         all_col_keys = col_keys + list(extra_defaults.keys())
+        n_cols = len(all_col_keys)
 
-        # BCP character mode cannot load binary columns; fall back for those.
-        if any(isinstance(col_by_key[k].type, LargeBinary) for k in all_col_keys if k in col_by_key):
-            super().bulk_insert(conn, table, records)
-            return
+        # Per-column flag: True if the column holds binary data.
+        col_is_binary = [
+            isinstance(col_by_key[k].type, LargeBinary)
+            for k in all_col_keys
+        ]
 
         full_name = (
             f"[{table.schema}].[{table.name}]"
@@ -158,15 +162,56 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
             else f"[{table.name}]"
         )
 
-        fd, data_path = tempfile.mkstemp(suffix=".bcp")
+        # Map column key → 1-based position in the SQL table (for format file).
+        col_pos = {col.key: idx + 1 for idx, col in enumerate(table.columns)}
+
+        data_fd, data_path = tempfile.mkstemp(suffix=".dat")
+        fmt_fd, fmt_path = tempfile.mkstemp(suffix=".fmt")
         try:
-            with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
+            # Write the data file in binary mode so we can embed raw bytes for
+            # binary columns while still writing UTF-8 text for other columns.
+            with os.fdopen(data_fd, "wb") as f:
                 for record in records:
-                    row = []
-                    for key in all_col_keys:
+                    for i, key in enumerate(all_col_keys):
                         v = record.get(key) if key in col_keys else extra_defaults[key]
-                        row.append(self._format_bcp_value(v))
-                    f.write("\t".join(row) + "\n")
+                        is_last = (i == n_cols - 1)
+                        terminator = b"\n" if is_last else b"\t"
+
+                        if col_is_binary[i]:
+                            # 4-byte little-endian length prefix; -1 (0xFFFFFFFF) = NULL.
+                            if v is None:
+                                f.write(struct.pack("<i", -1))
+                            else:
+                                raw = v if isinstance(v, (bytes, bytearray)) else bytes(v)
+                                f.write(struct.pack("<i", len(raw)))
+                                f.write(raw)
+                            f.write(terminator)
+                        else:
+                            f.write(self._format_bcp_value(v).encode("utf-8"))
+                            f.write(terminator)
+
+            # Write the non-XML BCP format file.
+            #   SQLCHAR      → character field delimited by tab / newline
+            #   SQLVARBINARY → binary field with 4-byte length prefix + terminator
+            fmt_lines = ["14.0", str(n_cols)]
+            for i, key in enumerate(all_col_keys):
+                is_last = (i == n_cols - 1)
+                term = r"\n" if is_last else r"\t"
+                col = col_by_key.get(key)
+                col_name = col.name if col else key
+                server_col = col_pos.get(key, i + 1)
+
+                if col_is_binary[i]:
+                    fmt_lines.append(
+                        f"{i+1}\tSQLVARBINARY\t4\t0\t\"{term}\"\t{server_col}\t{col_name}\t\"\""
+                    )
+                else:
+                    fmt_lines.append(
+                        f"{i+1}\tSQLCHAR\t0\t8000\t\"{term}\"\t{server_col}\t{col_name}\t\"\""
+                    )
+
+            with os.fdopen(fmt_fd, "w") as f:
+                f.write("\n".join(fmt_lines) + "\n")
 
             cmd = [
                 self.bcp_path, full_name, "in", data_path,
@@ -174,12 +219,10 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 "-d", url.database,
                 "-U", url.username,
                 "-P", url.password,
-                "-c",       # character (text) mode
-                "-t", "\t", # tab field separator
-                "-r", "\n", # newline row terminator
-                "-k",       # empty field → NULL (not column default)
+                "-f", fmt_path,
+                "-k",           # empty char field → NULL (not column default)
                 "-b", "10000",
-                "-C", "65001",  # UTF-8
+                "-C", "65001",  # UTF-8 for character fields
             ]
             if str(url.query.get("TrustServerCertificate", "")).lower() == "yes":
                 cmd.append("-u")  # trust server certificate (mssql-tools18)
@@ -192,3 +235,5 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
         finally:
             if os.path.exists(data_path):
                 os.unlink(data_path)
+            if os.path.exists(fmt_path):
+                os.unlink(fmt_path)

From 9f42f2beaf7860fb0d16a123a666971608546e4d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 20 Jun 2026 06:17:33 +0000
Subject: [PATCH 09/12] Fix BCP binary encoding: hex without 0x prefix

BCP character mode accepts hex-encoded binary without the 0x prefix.
Simplify _format_bcp_value to emit v.hex() for bytes values and revert
to plain character mode (no format file needed).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 src/xml2db/dialect/mssql.py | 83 ++++++++-----------------------------
 1 file changed, 17 insertions(+), 66 deletions(-)

diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index 29727af..7fe7739 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -1,11 +1,10 @@
 import os
 import shutil
-import struct
 import subprocess
 import tempfile
 from typing import Any, List, TYPE_CHECKING
 
-from sqlalchemy import Index, LargeBinary
+from sqlalchemy import Index
 from sqlalchemy.dialects import mssql as mssql_dialect
 
 from .base import DatabaseDialect
@@ -96,12 +95,15 @@ def relation_extra_indexes(
 
     @staticmethod
     def _format_bcp_value(v: Any) -> str:
-        """Format a Python value as a BCP character field (non-binary columns only)."""
+        """Format a Python value as a BCP character-mode field (tab-separated)."""
         if v is None:
             return ""
         if isinstance(v, bool):
             # Must precede int: bool is a subclass of int.
             return "1" if v else "0"
+        if isinstance(v, (bytes, bytearray)):
+            # BCP character mode accepts hex without 0x prefix for binary columns.
+            return v.hex()
         s = str(v)
         # Tab is the field delimiter; replace any occurrence in string values.
         return s.replace("\t", " ")
@@ -113,10 +115,7 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
         BCP is unavailable or the connection lacks SQL auth credentials, are
         handled by the base-class ``fast_executemany`` path.
 
-        BCP uses a non-XML format file so that binary columns (LargeBinary /
-        VARBINARY) are sent as raw bytes with a 4-byte length prefix, while
-        all other columns are sent as UTF-8 character data.  BCP does not
-        participate in the caller's SQLAlchemy transaction.
+        BCP does not participate in the caller's SQLAlchemy transaction.
 
         Args:
             conn: A SQLAlchemy ``Connection`` already within a transaction.
@@ -148,13 +147,6 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                     extra_defaults[col.key] = d.arg
 
         all_col_keys = col_keys + list(extra_defaults.keys())
-        n_cols = len(all_col_keys)
-
-        # Per-column flag: True if the column holds binary data.
-        col_is_binary = [
-            isinstance(col_by_key[k].type, LargeBinary)
-            for k in all_col_keys
-        ]
 
         full_name = (
             f"[{table.schema}].[{table.name}]"
@@ -162,56 +154,15 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
             else f"[{table.name}]"
         )
 
-        # Map column key → 1-based position in the SQL table (for format file).
-        col_pos = {col.key: idx + 1 for idx, col in enumerate(table.columns)}
-
-        data_fd, data_path = tempfile.mkstemp(suffix=".dat")
-        fmt_fd, fmt_path = tempfile.mkstemp(suffix=".fmt")
+        fd, data_path = tempfile.mkstemp(suffix=".bcp")
         try:
-            # Write the data file in binary mode so we can embed raw bytes for
-            # binary columns while still writing UTF-8 text for other columns.
-            with os.fdopen(data_fd, "wb") as f:
+            with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
                 for record in records:
-                    for i, key in enumerate(all_col_keys):
+                    row = []
+                    for key in all_col_keys:
                         v = record.get(key) if key in col_keys else extra_defaults[key]
-                        is_last = (i == n_cols - 1)
-                        terminator = b"\n" if is_last else b"\t"
-
-                        if col_is_binary[i]:
-                            # 4-byte little-endian length prefix; -1 (0xFFFFFFFF) = NULL.
-                            if v is None:
-                                f.write(struct.pack("<i", -1))
-                            else:
-                                raw = v if isinstance(v, (bytes, bytearray)) else bytes(v)
-                                f.write(struct.pack("<i", len(raw)))
-                                f.write(raw)
-                            f.write(terminator)
-                        else:
-                            f.write(self._format_bcp_value(v).encode("utf-8"))
-                            f.write(terminator)
-
-            # Write the non-XML BCP format file.
-            #   SQLCHAR      → character field delimited by tab / newline
-            #   SQLVARBINARY → binary field with 4-byte length prefix + terminator
-            fmt_lines = ["14.0", str(n_cols)]
-            for i, key in enumerate(all_col_keys):
-                is_last = (i == n_cols - 1)
-                term = r"\n" if is_last else r"\t"
-                col = col_by_key.get(key)
-                col_name = col.name if col else key
-                server_col = col_pos.get(key, i + 1)
-
-                if col_is_binary[i]:
-                    fmt_lines.append(
-                        f"{i+1}\tSQLVARBINARY\t4\t0\t\"{term}\"\t{server_col}\t{col_name}\t\"\""
-                    )
-                else:
-                    fmt_lines.append(
-                        f"{i+1}\tSQLCHAR\t0\t8000\t\"{term}\"\t{server_col}\t{col_name}\t\"\""
-                    )
-
-            with os.fdopen(fmt_fd, "w") as f:
-                f.write("\n".join(fmt_lines) + "\n")
+                        row.append(self._format_bcp_value(v))
+                    f.write("\t".join(row) + "\n")
 
             cmd = [
                 self.bcp_path, full_name, "in", data_path,
@@ -219,10 +170,12 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 "-d", url.database,
                 "-U", url.username,
                 "-P", url.password,
-                "-f", fmt_path,
-                "-k",           # empty char field → NULL (not column default)
+                "-c",       # character (text) mode
+                "-t", "\t", # tab field separator
+                "-r", "\n", # newline row terminator
+                "-k",       # empty field → NULL (not column default)
                 "-b", "10000",
-                "-C", "65001",  # UTF-8 for character fields
+                "-C", "65001",  # UTF-8
             ]
             if str(url.query.get("TrustServerCertificate", "")).lower() == "yes":
                 cmd.append("-u")  # trust server certificate (mssql-tools18)
@@ -235,5 +188,3 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
         finally:
             if os.path.exists(data_path):
                 os.unlink(data_path)
-            if os.path.exists(fmt_path):
-                os.unlink(fmt_path)

From aa886bf57150b30fcd2b65f1d5ec0542bfe1b4df Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 20 Jun 2026 06:20:35 +0000
Subject: [PATCH 10/12] Fix rowcount assertion in fallback test

SELECT rowcount is always -1 in DBAPI; use fetchall() instead.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 tests/test_bulk_insert_mssql.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_bulk_insert_mssql.py b/tests/test_bulk_insert_mssql.py
index 2e66ad7..4c85bdb 100644
--- a/tests/test_bulk_insert_mssql.py
+++ b/tests/test_bulk_insert_mssql.py
@@ -287,7 +287,7 @@ def test_mssql_bcp_fallback_when_unavailable(mssql_engine):
         with mssql_engine.begin() as conn:
             dialect.bulk_insert(conn, table, records)
         with mssql_engine.connect() as conn:
-            count = conn.execute(select(table)).rowcount
+            count = len(conn.execute(select(table)).fetchall())
         assert count == _BCP_THRESHOLD
     finally:
         _drop(meta, mssql_engine)

From 041a495cd9e4ad01b7f3a000d78620ee2c472777 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 20 Jun 2026 06:28:45 +0000
Subject: [PATCH 11/12] Re-enable push/PR triggers on all CI workflows

Restores push+pull_request triggers on main for python-package,
postgres, mysql, and duckdb workflows.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 .github/workflows/integration-tests-duckdb.yml   | 4 ++++
 .github/workflows/integration-tests-mysql.yml    | 4 ++++
 .github/workflows/integration-tests-postgres.yml | 4 ++++
 .github/workflows/python-package.yml             | 4 ++++
 4 files changed, 16 insertions(+)

diff --git a/.github/workflows/integration-tests-duckdb.yml b/.github/workflows/integration-tests-duckdb.yml
index 6d2816a..7c63c5b 100644
--- a/.github/workflows/integration-tests-duckdb.yml
+++ b/.github/workflows/integration-tests-duckdb.yml
@@ -1,6 +1,10 @@
 name: Duckdb integration tests
 
 on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/integration-tests-mysql.yml b/.github/workflows/integration-tests-mysql.yml
index f701bd1..f1a8d89 100644
--- a/.github/workflows/integration-tests-mysql.yml
+++ b/.github/workflows/integration-tests-mysql.yml
@@ -1,6 +1,10 @@
 name: MySQL integration tests
 
 on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/integration-tests-postgres.yml b/.github/workflows/integration-tests-postgres.yml
index 9344f27..702fe16 100644
--- a/.github/workflows/integration-tests-postgres.yml
+++ b/.github/workflows/integration-tests-postgres.yml
@@ -1,6 +1,10 @@
 name: PostgreSQL integration tests
 
 on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 03156a4..f4328ce 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -4,6 +4,10 @@
 name: Python package
 
 on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
   workflow_dispatch:
 
 jobs:

From b99487ec25bc25a62014d32e9eb1eb665b3c9c80 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 20 Jun 2026 09:31:06 +0000
Subject: [PATCH 12/12] Support BCP with Kerberos/Windows trusted connections

When Trusted_Connection=yes is in the URL query, BCP is invoked with
-T (trusted connection) instead of -U/-P, mirroring how the ODBC driver
handles Kerberos auth on Linux. Falls back to fast_executemany only when
both SQL credentials and Trusted_Connection are absent.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QvgSwLLGgKaDbZaEYYp7BJ
---
 src/xml2db/dialect/mssql.py | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/src/xml2db/dialect/mssql.py b/src/xml2db/dialect/mssql.py
index 7fe7739..936d481 100644
--- a/src/xml2db/dialect/mssql.py
+++ b/src/xml2db/dialect/mssql.py
@@ -112,8 +112,9 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> 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 lacks SQL auth credentials, are
-        handled by the base-class ``fast_executemany`` path.
+        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``.
 
         BCP does not participate in the caller's SQLAlchemy transaction.
 
@@ -126,11 +127,12 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
             return
 
         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 url.username
-            or not url.password
+            or (not has_sql_auth and not trusted)
         ):
             super().bulk_insert(conn, table, records)
             return
@@ -168,8 +170,6 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 self.bcp_path, full_name, "in", data_path,
                 "-S", f"{url.host},{url.port or 1433}",
                 "-d", url.database,
-                "-U", url.username,
-                "-P", url.password,
                 "-c",       # character (text) mode
                 "-t", "\t", # tab field separator
                 "-r", "\n", # newline row terminator
@@ -177,6 +177,10 @@ def bulk_insert(self, conn: Any, table: Any, records: list) -> None:
                 "-b", "10000",
                 "-C", "65001",  # UTF-8
             ]
+            if has_sql_auth:
+                cmd += ["-U", url.username, "-P", url.password]
+            else:
+                cmd.append("-T")  # Kerberos / Windows trusted connection
             if str(url.query.get("TrustServerCertificate", "")).lower() == "yes":
                 cmd.append("-u")  # trust server certificate (mssql-tools18)
             result = subprocess.run(cmd, capture_output=True, text=True)