FEAT: Add table-valued parameter support

CHANGELOG.md

@@ -7,6 +7,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ## [Unreleased]
 
 ### Added
+- **AB#48177:** Added explicit `TableValuedParameter(type_name, columns, rows,
+  schema=...)` input values. Sync `execute()` streams lists, tuples and iterators
+  through native ODBC data-at-execution using one row of additional storage and
+  one procedure invocation, without discovering table metadata.
 - New feature: Support for macOS and Linux.
 - Documentation: Added API documentation in the Wiki.
 - New `token_provider=` parameter on `connect()` / `Connection` for Microsoft

README.md

@@ -208,6 +208,58 @@ connection.close()
 
 ```
 
+### Table-valued parameters
+
+Declare a table type and an input `READONLY` parameter on SQL Server:
+
+```sql
+CREATE TYPE dbo.ItemBatch AS TABLE (id int, label nvarchar(100));
+GO
+CREATE PROCEDURE dbo.ProcessItems @items dbo.ItemBatch READONLY
+AS SELECT id, label FROM @items;
+```
+
+Pass one explicit value to synchronous `execute()`; ordinary nested lists/tuples are **not**
+implicitly treated as TVPs:
+
+```python
+from mssql_python import TableValuedParameter, SQL_INTEGER, SQL_WVARCHAR
+
+items = TableValuedParameter(
+    type_name="dbo.ItemBatch",
+    columns=[("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 100)],
+    rows=[(1, "Keyboard"), (2, None)],
+)
+cursor.execute("EXEC dbo.ProcessItems ?", items)
+```
+
+Column tuples are `(name, sql_type[, size_or_precision[, scale]])`. Use the
+driver's integer `SQL_*` constants. Columns bind **positionally**; names are
+diagnostic labels, not a schema lookup. Character and binary columns require a
+size; `0` means MAX for variable-length types. Decimal columns require both
+precision and scale, e.g. `("amount", SQL_DECIMAL, 20, 6)`. Precision is 1–38;
+scale is 0–precision. Temporal scale defaults to 7 (0–7 supported).
+Sizes for Unicode columns are UTF-16 code units; binary sizes are bytes.
+Rows must contain one cell per column. Supported cells include integers, floats,
+bools, `Decimal`, strings, bytes/bytearray, date/time/datetime, UUID and `None`.
+
+Use an unquoted `schema.Type` or `type_name="ItemBatch", schema="dbo"` (default
+schema: `dbo`); SQL quoting/bracket syntax and database-qualified names are not
+parsed. The constructor does not consume rows. `rows=None` is invalid: use `[]`
+for an empty table; individual cells may be `None`.
+
+With the default **msodbcsql18** sync provider, `rows` can also be a synchronous
+iterator/generator. It is pulled during `execute()`, with one row of additional
+driver-side binding storage regardless of total input length. A large cell still
+occupies memory for that cell. This is one SQL execution, **not** `executemany()`;
+multiple TVPs, scalar parameters and named parameters can be combined. The
+native row-count descriptor permits up to 2,147,483,647 rows. Materialized sources
+can be reused; iterators are one-shot and may be partially consumed on failure.
+Iterator exceptions, including `KeyboardInterrupt`, cancel/reset the statement
+before releasing its buffers. `executemany()` and `setinputsizes()` overrides
+for statements containing TVPs are not supported. The `mssql-odbc` provider does
+not currently support this ODBC TVP path.
+
 ## Still have questions?
 
 Check out our [FAQ](https://github.com/microsoft/mssql-python/wiki/Frequently-Asked-Questions). Still not answered? Create an [issue](https://github.com/microsoft/mssql-python/issues/new/choose) to ask a question.

mssql_python/__init__.py

@@ -66,6 +66,7 @@
 
 # Row Objects
 from .row import Row, RowMapping
+from .table_valued_parameter import TableValuedParameter
 
 # Logging Configuration (Simplified single-level DEBUG system)
 from .logging import logger, setup_logging, driver_logger
@@ -324,6 +325,7 @@ def _cleanup_connections():
 )
 
 __all__ = [
+    "TableValuedParameter",
     # Exception classes
     "Warning",
     "Error",

mssql_python/cursor.py

@@ -29,8 +29,10 @@
     DatabaseError,
 )
 from mssql_python.row import Row
+from mssql_python.table_valued_parameter import TableValuedParameter
 from mssql_python.perf_timer import perf_phase
 from mssql_python import get_settings
+from mssql_python.odbc_provider import ProviderManager, PROVIDER_MSSQL_ODBC
 from mssql_python.parameter_helper import (
     detect_and_convert_parameters,
     parse_pyformat_params,
@@ -389,6 +391,7 @@ def __init__(self, connection: "Connection", timeout: int = 0) -> None:
 
         self._connection: "Connection" = connection  # Store as private attribute
         self._timeout: int = timeout
+        self._executing = False
         self._inputsizes: Optional[List[Tuple[int, int, int, int]]] = None
         # self.connection.autocommit = False
         self._initialize_cursor()
@@ -1104,6 +1107,8 @@ def close(self) -> None:
         if getattr(self, "closed", True):
             # Do nothing - not calling _check_closed() here since we want this to be idempotent
             return
+        if getattr(self, "_executing", False):
+            raise ProgrammingError("Cannot close a cursor while it is executing", "")
 
         # Clear messages per DBAPI
         self.messages = []
@@ -1129,6 +1134,8 @@ def _check_closed(self) -> None:
         Raises:
             ProgrammingError: If the cursor is closed.
         """
+        if getattr(self, "_executing", False):
+            raise ProgrammingError("Cannot reuse a cursor while it is executing", "")
         if self.closed:
             raise ProgrammingError(
                 driver_error="Operation cannot be performed: The cursor is closed.",
@@ -1799,8 +1806,8 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
         #
         # Important: If you pass a tuple/list/dict as the ONLY argument,
         # it will be unwrapped for parameter binding. This means you cannot
-        # pass a tuple as a single parameter value (but SQL Server doesn't
-        # support tuple types as parameter values anyway).
+        # pass a tuple as a single scalar parameter value. A TVP is explicitly
+        # wrapped in TableValuedParameter and remains one parameter.
         with perf_phase("py::execute::param_prep"):
             if parameters:
                 # Check if single parameter is a nested container that should be unwrapped
@@ -1838,6 +1845,11 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
 
         # Validate that inputsizes matches parameter count if both are present
         if parameters and self._inputsizes:
+            if any(isinstance(parameter, TableValuedParameter) for parameter in parameters):
+                raise NotSupportedError(
+                    "setinputsizes does not support table-valued parameters",
+                    "Remove setinputsizes for statements that pass TVP rows",
+                )
             if len(self._inputsizes) != len(parameters):
 
                 warnings.warn(
@@ -1855,28 +1867,41 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
             self.is_stmt_prepared = [False]
         effective_use_prepare = use_prepare and not same_sql
 
-        with perf_phase("py::execute::cpp_call"):
-            if parameters:
-                ret = ddbc_bindings.DDBCSQLExecute(
-                    self.hstmt,
-                    operation,
-                    parameters,
-                    self._inputsizes,
-                    self.is_stmt_prepared,
-                    effective_use_prepare,
-                    encoding_settings,
-                )
-            else:
-                ret = ddbc_bindings.DDBCSQLExecDirect(self.hstmt, operation)
-        # Check return code
+        self._executing = True
         try:
-
+            with perf_phase("py::execute::cpp_call"):
+                if parameters:
+                    if any(isinstance(parameter, TableValuedParameter) for parameter in parameters):
+                        if ProviderManager.effective() == PROVIDER_MSSQL_ODBC:
+                            raise NotSupportedError(
+                                "Table-valued parameters are not supported by "
+                                "the mssql-odbc provider",
+                                "Use the default msodbcsql18 provider for TVP statements",
+                            )
+                    ret = ddbc_bindings.DDBCSQLExecute(
+                        self.hstmt,
+                        operation,
+                        parameters,
+                        self._inputsizes,
+                        self.is_stmt_prepared,
+                        effective_use_prepare,
+                        encoding_settings,
+                    )
+                else:
+                    ret = ddbc_bindings.DDBCSQLExecDirect(self.hstmt, operation)
             # Check for errors but don't raise exceptions for info/warning messages
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
-        except Exception as e:  # pylint: disable=broad-exception-caught
+        except BaseException as e:  # Reset even when a TVP iterator raises KeyboardInterrupt.
             logger.warning("Execute failed, resetting cursor: %s", e)
-            self._reset_cursor()
+            try:
+                self._reset_cursor()
+            except Exception as cleanup_error:
+                # A callback may have closed the connection. Preserve its
+                # original error instead of replacing it with an allocation error.
+                logger.warning("Cursor recovery failed: %s", cleanup_error)
             raise
+        finally:
+            self._executing = False
 
         # Capture any diagnostic messages (SQL_SUCCESS_WITH_INFO, etc.)
         with perf_phase("py::execute::diag_records"):
@@ -2554,6 +2579,11 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
                     if hasattr(seq_of_parameters, "__getitem__")
                     else []
                 )
+                if any(isinstance(value, TableValuedParameter) for value in column):
+                    raise NotSupportedError(
+                        "executemany does not support table-valued parameters",
+                        "Use execute for each TVP operation",
+                    )
                 sample_value, min_val, max_val, _ = self._compute_column_type(column)
 
                 if self._inputsizes and col_index < len(self._inputsizes):

mssql_python/mssql_python.pyi

@@ -21,6 +21,34 @@ import datetime
 import logging
 import pyarrow
 
+class TableValuedParameter:
+    def __init__(
+        self,
+        type_name: str,
+        columns: Sequence[Union[Tuple[str, int], Tuple[str, int, int], Tuple[str, int, int, int]]],
+        rows: Union[
+            List[Sequence[Any]],
+            Tuple[Sequence[Any], ...],
+            Iterator[Sequence[Any]],
+        ],
+        *,
+        schema: Optional[str] = None,
+    ) -> None: ...
+    @property
+    def type_name(self) -> str: ...
+    @property
+    def schema(self) -> str: ...
+    @property
+    def columns(self) -> Tuple[Tuple[str, int, int, int], ...]: ...
+    @property
+    def rows(
+        self,
+    ) -> Union[
+        List[Sequence[Any]],
+        Tuple[Sequence[Any], ...],
+        Iterator[Sequence[Any]],
+    ]: ...
+
 # GLOBALS - DB-API 2.0 Required Module Globals
 # https://www.python.org/dev/peps/pep-0249/#module-interface
 apilevel: str  # "2.0"

mssql_python/pybind/connection/connection.cpp

@@ -173,6 +173,10 @@ void Connection::disconnect(bool rollbackBeforeDisconnect) {
                 for (const auto& handle : childHandles) {
                     handle->markImplicitlyFreed();
                 }
+                for (const auto& buffers : _cleanupState->failedParameterBuffers) {
+                    buffers->release();
+                }
+                _cleanupState->failedParameterBuffers.clear();
                 _childStatementHandles.clear();
                 _allocationsSinceCompaction = 0;
             }