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 input table-valued parameter support for stored procedure
+  calls. Pass a materialized list or tuple of rows as one `execute()` parameter,
+  including empty tables and positional or named parameters. The Rust ODBC
+  provider does not yet support TVPs.
 - New feature: Support for macOS and Linux.
 - Documentation: Added API documentation in the Wiki.
 - New `token_provider=` parameter on `connect()` / `Connection` for Microsoft

mssql_python/cursor.py

@@ -31,6 +31,7 @@
 from mssql_python.row import Row
 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,
@@ -365,6 +366,7 @@ def __init__(self, connection: "Connection", timeout: int = 0) -> None:
         # an HSTMT exists.
         self.closed: bool = False
         self.hstmt: Optional[Any] = None
+        self._tvp_metadata_hstmt: Optional[Any] = None
 
         self._connection: "Connection" = connection  # Store as private attribute
         self._timeout: int = timeout
@@ -996,7 +998,7 @@ def _allocate_statement_handle(self) -> None:
         """
         self.hstmt = self._connection._conn.alloc_statement_handle()
 
-    def _set_timeout(self) -> None:
+    def _set_timeout(self, statement_handle=None) -> None:
         """
         Set the query timeout attribute on the statement handle.
         This is called once when the cursor is created and after any handle reallocation.
@@ -1006,12 +1008,13 @@ def _set_timeout(self) -> None:
             logger.debug("_set_timeout: Setting query timeout=%d seconds", self._timeout)
             try:
                 timeout_value = int(self._timeout)
+                target_handle = statement_handle or self.hstmt
                 ret = ddbc_bindings.DDBCSQLSetStmtAttr(
-                    self.hstmt,
+                    target_handle,
                     ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value,
                     timeout_value,
                 )
-                check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
+                check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, target_handle, ret)
                 logger.debug("Query timeout set to %d seconds", timeout_value)
             except Exception as e:  # pylint: disable=broad-exception-caught
                 logger.warning("Failed to set query timeout: %s", str(e))
@@ -1081,6 +1084,9 @@ def close(self) -> None:
             self.hstmt.free()
             self.hstmt = None
             logger.debug("SQLFreeHandle succeeded")
+        if self._tvp_metadata_hstmt:
+            self._tvp_metadata_hstmt.free()
+            self._tvp_metadata_hstmt = None
         self._clear_rownumber()
         self.closed = True
 
@@ -1730,8 +1736,9 @@ 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 remains one
+        # parameter by wrapping its rows in the ordinary outer parameter tuple:
+        # execute("EXEC dbo.proc ?", (rows,))
         with perf_phase("py::execute::param_prep"):
             if parameters:
                 # Check if single parameter is a nested container that should be unwrapped
@@ -1769,6 +1776,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, (list, tuple)) 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(
@@ -1786,22 +1798,33 @@ 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
         try:
-
+            with perf_phase("py::execute::cpp_call"):
+                if parameters:
+                    if any(isinstance(parameter, (list, tuple)) 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",
+                            )
+                        if self._tvp_metadata_hstmt is None:
+                            self._tvp_metadata_hstmt = (
+                                self._connection._conn.alloc_statement_handle()
+                            )
+                            self._set_timeout(self._tvp_metadata_hstmt)
+                    ret = ddbc_bindings.DDBCSQLExecute(
+                        self.hstmt,
+                        self._tvp_metadata_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
@@ -2415,7 +2438,6 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
             if hasattr(seq_of_parameters, "__getitem__")
             else next(iter(seq_of_parameters))
         )
-
         if isinstance(first_row, dict):
             # pyformat style - convert all rows
             # Parse parameter names from SQL (determines order for all rows)
@@ -2480,11 +2502,15 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
         # Prepare parameter type information
         with perf_phase("py::executemany::param_type_detection"):
             for col_index in range(param_count):
-                column = (
-                    [row[col_index] for row in seq_of_parameters]
-                    if hasattr(seq_of_parameters, "__getitem__")
-                    else []
-                )
+                column = []
+                for row in seq_of_parameters:
+                    value = row[col_index]
+                    if isinstance(value, (list, tuple)):
+                        raise NotSupportedError(
+                            "executemany does not support table-valued parameters",
+                            "Use execute for each TVP operation",
+                        )
+                    column.append(value)
                 sample_value, min_val, max_val, _ = self._compute_column_type(column)
 
                 if self._inputsizes and col_index < len(self._inputsizes):