perf(mssql): preserve native bulk loading and streamline execution

docs/changelog.rst

@@ -17,6 +17,10 @@ Unreleased
 * BigQuery supports native query resource controls,
   explicit STRUCT parameters, typed empty arrays, and configurable Storage Write
   stream modes while retaining the atomic PENDING default.
+* The mssql-python adapter can load Arrow streams with native BulkCopy options. Columns
+  map by name by default, and overwrite still uses DELETE.
+* Pymssql connection types include native encryption settings.
+
 * SQLite and aiosqlite can register custom window functions on Python 3.11
   and later when the SQLite runtime supports them. Choose a transaction lock
   mode or set the batch size for Arrow imports. Defaults stay the same.
@@ -48,6 +52,16 @@ Unreleased
 
 **Fixed:**
 
+* SQL Server event queue DDL guards use the configured table and index names.
+  Arrow ODBC index checks no longer include column text in the table name.
+
+* Arrow read failures in the mssql-python adapter use SQLSpec error types.
+* ADBC ADK stores reuse cached PostgreSQL placeholder conversion and preserve
+  question marks in quoted identifiers, literals, and comments.
+
+* Arrow ODBC pagination reuses compiled placeholder positions instead of
+  parsing SQL again. ADBC keeps bound values in its ADK store queries.
+  DuckDB Arrow loads keep sparse dictionary fields and quote table names.
 * SQLite pools replace lost in-memory connections. Arrow imports roll back
   writes on failure or cancellation when the adapter owns the transaction.
 
@@ -89,13 +103,6 @@ Unreleased
   ``IF NOT EXISTS`` guards. Cached row converters refresh when the configured
   JSON deserializer changes.
 
-* ADBC ADK stores reuse cached PostgreSQL placeholder conversion and preserve
-  question marks in quoted identifiers, literals, and comments.
-
-* Arrow ODBC pagination reuses compiled placeholder positions instead of
-  parsing SQL again. ADBC keeps bound values in its ADK store queries.
-  DuckDB Arrow loads keep sparse dictionary fields and quote table names.
-
 * Psycopg reads COPY files in chunks, not all at once. ADK stores use
   RETURNING to cut round trips. Psqlpy closes a connection if setup fails.
 

docs/reference/adapters/mssql_python.rst

@@ -94,3 +94,15 @@ Use these types inside ``extension_config["adk"]``.
 .. autoclass:: sqlspec.adapters.mssql_python.adk.MssqlPythonADKConfig
    :members:
    :show-inheritance:
+
+Native Arrow loading
+--------------------
+
+``load_from_arrow()`` accepts tables, record batches, record batch readers, and
+Arrow C stream sources. It forwards ``batch_size``, ``timeout``, ``table_lock``,
+``check_constraints``, ``fire_triggers``, ``keep_identity``, ``keep_nulls``,
+``use_internal_transaction``, and ``column_mappings`` to native BulkCopy.
+Field names supply default mappings; sources without schema metadata require
+explicit mappings. Native internal transactions apply per batch, not to the
+caller connection transaction. ``overwrite=True`` retains DELETE semantics;
+stream consumption failures can leave a partial load.

sqlspec/adapters/arrow_odbc/events/store.py

@@ -1,6 +1,5 @@
 """arrow-odbc event queue store with T-SQL and Db2 DDL."""
 
-import re
 from typing import Final
 
 from sqlspec.adapters.arrow_odbc.config import ArrowOdbcConfig
@@ -68,26 +67,15 @@ def _wrap_create_statement(self, statement: str, object_type: str) -> str:
         if self._db2:
             return statement
         if object_type == "table":
-            match = re.search(r"CREATE TABLE\s+(\S+)", statement, re.IGNORECASE)
-            if match:
-                table_name = match.group(1)
-                return f"IF OBJECT_ID(N'{_object_name(table_name)}', N'U') IS NULL BEGIN {statement}; END"
+            return f"IF OBJECT_ID(N'{_object_name(self.table_name)}', N'U') IS NULL BEGIN {statement}; END"
         if object_type == "index":
-            match = re.search(r"CREATE INDEX\s+(\S+)\s+ON\s+([^\s(]+)", statement, re.IGNORECASE)
-            if match:
-                index_name = match.group(1).strip("[]")
-                table_name = match.group(2)
-                return f"IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{index_name}' AND object_id = OBJECT_ID(N'{_object_name(table_name)}')) BEGIN {statement}; END"
+            return f"IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{self._index_name()}' AND object_id = OBJECT_ID(N'{_object_name(self.table_name)}')) BEGIN {statement}; END"
         return statement
 
     def _wrap_drop_statement(self, statement: str) -> str:
         if self._db2:
             return statement
-        match = re.search(r"DROP TABLE\s+(\S+)", statement, re.IGNORECASE)
-        if match:
-            table_name = match.group(1)
-            return f"IF OBJECT_ID(N'{_object_name(table_name)}', N'U') IS NOT NULL DROP TABLE {table_name};"
-        return statement
+        return f"IF OBJECT_ID(N'{_object_name(self.table_name)}', N'U') IS NOT NULL {statement};"
 
 
 def _split_table_name(table_name: str) -> tuple[str, str]:

sqlspec/adapters/mssql_python/adk/store.py

@@ -1,12 +1,12 @@
 """mssql-python ADK stores for Google Agent Development Kit session storage."""
 
-import re
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, cast
 
 from typing_extensions import NotRequired
 
 from sqlspec.adapters.mssql_python._typing import MssqlPythonCursor, MssqlPythonError
+from sqlspec.adapters.mssql_python.core import extract_error_number
 from sqlspec.config import ADKConfig
 from sqlspec.extensions.adk import BaseSyncADKStore, StoredEvent, StoredSession, normalize_session_list_options
 from sqlspec.extensions.adk.memory.store import BaseSyncADKMemoryStore
@@ -26,7 +26,6 @@
 MSSQL_DUPLICATE_OBJECT_ERROR: Final[int] = 2714
 MSSQL_DUPLICATE_INDEX_ERROR: Final[int] = 1913
 MSSQL_SCHEMA: Final[str] = "dbo"
-MSSQL_ERROR_NUMBER_PATTERN: Final[re.Pattern[str]] = re.compile(r"\(([-]?\d+)\)")
 JSON_FALLBACK_COLUMN_TYPE: Final[str] = "NVARCHAR(MAX)"
 JSON_NATIVE_COLUMN_TYPE: Final[str] = "JSON"
 
@@ -848,17 +847,7 @@ def _cursor_rowcount(cursor: Any) -> int:
 
 def _is_mssql_table_missing(exc: BaseException) -> bool:
     text = str(exc).lower()
-    return "invalid object name" in text or _mssql_error_number(exc) == MSSQL_TABLE_NOT_FOUND_ERROR
-
-
-def _mssql_error_number(exc: BaseException) -> "int | None":
-    matches = MSSQL_ERROR_NUMBER_PATTERN.findall(str(exc))
-    if not matches:
-        return None
-    try:
-        return int(matches[-1])
-    except ValueError:
-        return None
+    return "invalid object name" in text or extract_error_number(exc) == MSSQL_TABLE_NOT_FOUND_ERROR
 
 
 def _quote_identifier(identifier: str) -> str:

sqlspec/adapters/mssql_python/core.py

@@ -40,10 +40,11 @@
     "create_mapped_exception",
     "default_statement_config",
     "driver_profile",
+    "extract_error_number",
     "materialize_tuple_rows",
 )
 
-_ERROR_NUMBER_PATTERN: Final[re.Pattern[str]] = re.compile(r"\(([-]?\d+)(?:,|\))")
+_ERROR_NUMBER_PATTERN: Final[re.Pattern[str]] = re.compile(r"(?:\(([-]?\d+)(?:,|\))|\bMsg\s+([-]?\d+)\b)")
 _MSSQL_CONSTRAINT_547: Final[int] = 547
 _VERSION_PATTERN: Final[re.Pattern[str]] = re.compile(r"(\d+)")
 _VERSION_PART_COUNT: Final[int] = 3
@@ -98,9 +99,54 @@
 }
 
 
+def extract_error_number(exc: BaseException | None) -> int | None:
+    """Extract numeric SQL Server error code using fast attribute/string parsing before regex fallback."""
+    if exc is None:
+        return None
+    for attr in ("number", "error_code", "errno"):
+        val = getattr(exc, attr, None)
+        if isinstance(val, int) and not isinstance(val, bool):
+            return val
+    ddbc_err = getattr(exc, "ddbc_error", None)
+    if isinstance(ddbc_err, str) and ddbc_err.startswith("("):
+        end_idx = ddbc_err.find(",")
+        if end_idx == -1:
+            end_idx = ddbc_err.find(")")
+        if end_idx != -1:
+            num_str = ddbc_err[1:end_idx].strip()
+            try:
+                return int(num_str)
+            except ValueError:
+                pass
+
+    if exc.args:
+        first_arg = exc.args[0]
+        if isinstance(first_arg, int) and not isinstance(first_arg, bool):
+            return first_arg
+        if isinstance(first_arg, str):
+            msg = first_arg
+            start_idx = msg.rfind("(")
+            if start_idx != -1:
+                end_idx = msg.find(",", start_idx)
+                if end_idx == -1:
+                    end_idx = msg.find(")", start_idx)
+                if end_idx != -1:
+                    num_str = msg[start_idx + 1 : end_idx].strip()
+                    try:
+                        return int(num_str)
+                    except ValueError:
+                        pass
+
+    matches = _ERROR_NUMBER_PATTERN.findall(str(exc))
+    if not matches:
+        return None
+    last_match = matches[-1]
+    return int(last_match[0] or last_match[1])
+
+
 def create_mapped_exception(error: Exception, *, logger: "Logger | None" = None) -> SQLSpecError:
     """Map a mssql-python exception to SQLSpec's exception hierarchy."""
-    error_number = _extract_error_number(error)
+    error_number = extract_error_number(error)
     if error_number == _MSSQL_CONSTRAINT_547:
         message = str(error)
         if "check constraint" in message.lower():
@@ -131,15 +177,17 @@ def create_mapped_exception(error: Exception, *, logger: "Logger | None" = None)
 
 
 def materialize_tuple_rows(fetched: "Sequence[Any] | None") -> "list[tuple[Any, ...]]":
-    """Materialize mssql-python ``Row`` objects into plain tuples.
+    """Materialize mssql-python Row objects into plain tuples.
 
-    ``mssql-python`` returns ``mssql_python.Row`` objects that are iterable and
-    indexable but are not ``tuple`` subclasses. The driver reports
-    ``row_format="tuple"``, so fetched rows are converted to real tuples to keep
-    that contract accurate when results are materialized.
+    Uses native tuple storage when available, avoiding a tuple copy per row.
     """
     if not fetched:
         return []
+    first = fetched[0]
+    if isinstance(first, tuple):
+        return list(fetched) if not isinstance(fetched, list) else fetched
+    if hasattr(first, "_values"):
+        return [tuple(row._values) if not isinstance(row._values, tuple) else row._values for row in fetched]
     return [tuple(row) for row in fetched]
 
 
@@ -330,16 +378,6 @@ def _append_port(server: str, port: Any) -> str:
     return f"{server},{port}"
 
 
-def _extract_error_number(exc: Exception) -> "int | None":
-    matches = _ERROR_NUMBER_PATTERN.findall(str(exc))
-    if not matches:
-        return None
-    try:
-        return int(matches[-1])
-    except ValueError:
-        return None
-
-
 MSSQL_PYTHON_VERSION: Final[tuple[int, int, int]] = _parse_version()
 driver_profile = build_profile()
 default_statement_config = build_statement_config()

sqlspec/adapters/mssql_python/data_dictionary.py

@@ -14,23 +14,21 @@
     SystemMetadataRequest,
     SystemMetadataResult,
     TableMetadata,
-    VersionInfo,
     ensure_system_metadata_request,
     get_data_dictionary_loader,
     get_dialect_config,
     system_metadata_gated_result,
 )
 from sqlspec.data_dictionary.dialects.mssql import (
+    MssqlVersionInfo,
     build_mssql_metadata_capability_profile,
     build_mssql_system_metadata_capability,
     build_mssql_system_metadata_result,
     build_mssql_table_ddl_result,
     extract_mssql_version_value,
     get_mssql_data_dictionary_options,
-    is_mssql_azure_sql,
     list_mssql_available_features,
     merge_mssql_table_lists,
-    mssql_supports_native_json,
     parse_mssql_engine_edition,
     parse_mssql_version_components,
     resolve_mssql_feature_flag,
@@ -51,44 +49,6 @@
 logger = get_logger("sqlspec.adapters.mssql_python.data_dictionary")
 
 
-class MssqlVersionInfo(VersionInfo):
-    """MSSQL database version info with build, revision, and Azure SQL detection."""
-
-    def __init__(
-        self,
-        major: int,
-        minor: int = 0,
-        build: int = 0,
-        revision: int = 0,
-        edition: str | None = None,
-        engine_edition: int | None = None,
-    ) -> None:
-        super().__init__(major, minor, 0)
-        self.build = build
-        self.revision = revision
-        self.edition = edition
-        self.engine_edition = engine_edition
-        self.is_azure_sql = is_mssql_azure_sql(engine_edition)
-
-    def supports_native_json(self) -> bool:
-        """Return whether this server supports the native JSON type."""
-        return mssql_supports_native_json(self.major, is_azure_sql=self.is_azure_sql)
-
-    @property
-    def version_tuple(self) -> "tuple[int, int, int]":
-        """Get version tuple using the MSSQL build number as the third component."""
-        return (self.major, self.minor, self.build)
-
-    def __str__(self) -> str:
-        """String representation of version info."""
-        version_str = f"{self.major}.{self.minor}.{self.build}.{self.revision}"
-        if self.edition:
-            version_str += f" ({self.edition})"
-        if self.is_azure_sql:
-            version_str += " [Azure]"
-        return version_str
-
-
 class _MssqlDataDictionaryMixin:
     """Shared helpers for MSSQL data dictionaries."""