FEAT: streaming support in executemany (#251)

### Work Item / Issue Reference  
<!-- 
IMPORTANT: Please follow the PR template guidelines below.
For mssql-python maintainers: Insert your ADO Work Item ID below (e.g.
AB#37452)
For external contributors: Insert Github Issue number below (e.g. #149)
Only one reference is required - either GitHub issue OR ADO Work Item.
-->

<!-- mssql-python maintainers: ADO Work Item -->
>
[AB#38645](https://sqlclientdrivers.visualstudio.com/c6d89619-62de-46a0-8b46-70b92a84d85e/_workitems/edit/38645)

<!-- External contributors: GitHub Issue -->
> GitHub Issue: #<ISSUE_NUMBER>

-------------------------------------------------------------------
### Summary   
<!-- Insert your summary of changes below. Minimum 10 characters
required. -->
This pull request introduces robust support for streaming large values
(such as NVARCHAR(MAX), VARCHAR(MAX), VARBINARY(MAX)) with the
`executemany` method, ensuring correct handling of parameters that
require Data-At-Execution (DAE) or streaming. It also adds comprehensive
tests for these scenarios and improves logging and code clarity
throughout the codebase.

**Streaming and DAE parameter handling:**

* The `executemany` method in `mssql_python/cursor.py` now automatically
detects DAE/streaming parameters and falls back to row-by-row execution
when necessary, ensuring correct behavior for large values and streaming
inserts.
[[1]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280L1567-L1579)
[[2]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R1597)
[[3]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R1701-R1709)
* In the C++ binding (`mssql_python/pybind/ddbc_bindings.cpp`), the
`SQLExecuteMany_wrap` function checks for DAE parameters and either
performs fast array execution or falls back to row-by-row execution with
streaming, supporting both text and binary data.

**Testing improvements:**

* Added new tests in `tests/test_004_cursor.py` to verify streaming
inserts and fetches for NVARCHAR(MAX), VARCHAR(MAX), and VARBINARY(MAX)
columns using `executemany` and all fetch modes.

**Sample usage and demonstration:**

* Updated `main.py` to include a sample script that demonstrates
streaming inserts and fetches of large NVARCHAR(MAX) values using
`executemany` and different fetch modes.

**Logging and code clarity:**

* Improved debug logging for parameter handling and DAE detection, and
clarified log messages for LOB (Large Object) column handling in fetch
operations.
[[1]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280L1001-R1013)
[[2]](diffhunk://#diff-dde2297345718ec449a14e7dff91b7bb2342b008ecc071f562233646d71144a1L2216-R2276)
[[3]](diffhunk://#diff-dde2297345718ec449a14e7dff91b7bb2342b008ecc071f562233646d71144a1L3273-R3333)
[[4]](diffhunk://#diff-dde2297345718ec449a14e7dff91b7bb2342b008ecc071f562233646d71144a1L3395-R3455)

<!-- 
### PR Title Guide

> For feature requests
FEAT: (short-description)

> For non-feature requests like test case updates, config updates ,
dependency updates etc
CHORE: (short-description) 

> For Fix requests
FIX: (short-description)

> For doc update requests 
DOC: (short-description)

> For Formatting, indentation, or styling update
STYLE: (short-description)

> For Refactor, without any feature changes
REFACTOR: (short-description)

> For release related changes, without any feature changes
RELEASE: #<RELEASE_VERSION> (short-description) 

### Contribution Guidelines

External contributors:
- Create a GitHub issue first:
https://github.com/microsoft/mssql-python/issues/new
- Link the GitHub issue in the "GitHub Issue" section above
- Follow the PR title format and provide a meaningful summary

mssql-python maintainers:
- Create an ADO Work Item following internal processes
- Link the ADO Work Item in the "ADO Work Item" section above  
- Follow the PR title format and provide a meaningful summary
-->

---------

Co-authored-by: Gaurav Sharma <[email protected]>

Author: gargsaumya

mssql_python/cursor.py

@@ -1529,22 +1529,19 @@ def _compute_column_type(self, column):
                 sample_value = v
 
         return sample_value, None, None
-
+    
     def executemany(self, operation: str, seq_of_parameters: list) -> None:
         """
         Prepare a database operation and execute it against all parameter sequences.
         This version uses column-wise parameter binding and a single batched SQLExecute().
         Args:
             operation: SQL query or command.
             seq_of_parameters: Sequence of sequences or mappings of parameters.
-
         Raises:
             Error: If the operation fails.
         """
         self._check_closed()
         self._reset_cursor()
-
-        # Clear any previous messages
         self.messages = []
 
         if not seq_of_parameters:
@@ -1570,6 +1567,7 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
         param_count = len(sample_row)
         param_info = ddbc_bindings.ParamInfo
         parameters_type = []
+        any_dae = False
 
         # Check if we have explicit input sizes set
         if self._inputsizes:
@@ -1673,6 +1671,14 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
                     paraminfo.columnSize = max(max_binary_size, 1)
 
                 parameters_type.append(paraminfo)
+                if paraminfo.isDAE:
+                    any_dae = True
+        
+        if any_dae:
+            log('debug', "DAE parameters detected. Falling back to row-by-row execution with streaming.")
+            for row in seq_of_parameters:
+                self.execute(operation, row)
+            return
 
         # Process parameters into column-wise format with possible type conversions
         # First, convert any Decimal types as needed for NUMERIC/DECIMAL columns
@@ -1705,8 +1711,7 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
         log('debug', "Executing batch query with %d parameter sets:\n%s",
             len(seq_of_parameters), "\n".join(f"  {i+1}: {tuple(p) if isinstance(p, (list, tuple)) else p}" for i, p in enumerate(seq_of_parameters[:5]))  # Limit to first 5 rows for large batches
         )
-        
-        # Execute batched statement
+
         ret = ddbc_bindings.SQLExecuteMany(
             self.hstmt,
             operation,

mssql_python/pybind/ddbc_bindings.cpp

@@ -2000,6 +2000,7 @@ SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle,
                               size_t paramSetSize) {
     SQLHANDLE hStmt = statementHandle->get();
     SQLWCHAR* queryPtr;
+
 #if defined(__APPLE__) || defined(__linux__)
     std::vector<SQLWCHAR> queryBuffer = WStringToSQLWCHAR(query);
     queryPtr = queryBuffer.data();
@@ -2008,15 +2009,63 @@ SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle,
 #endif
     RETCODE rc = SQLPrepare_ptr(hStmt, queryPtr, SQL_NTS);
     if (!SQL_SUCCEEDED(rc)) return rc;
-    std::vector<std::shared_ptr<void>> paramBuffers;
-    rc = BindParameterArray(hStmt, columnwise_params, paramInfos, paramSetSize, paramBuffers);
-    if (!SQL_SUCCEEDED(rc)) return rc;
-    rc = SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)paramSetSize, 0);
-    if (!SQL_SUCCEEDED(rc)) return rc;
-    rc = SQLExecute_ptr(hStmt);
-    return rc;
+
+    bool hasDAE = false;
+    for (const auto& p : paramInfos) {
+        if (p.isDAE) {
+            hasDAE = true;
+            break;
+        }
+    }
+    if (!hasDAE) {
+        std::vector<std::shared_ptr<void>> paramBuffers;
+        rc = BindParameterArray(hStmt, columnwise_params, paramInfos, paramSetSize, paramBuffers);
+        if (!SQL_SUCCEEDED(rc)) return rc;
+
+        rc = SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)paramSetSize, 0);
+        if (!SQL_SUCCEEDED(rc)) return rc;
+
+        rc = SQLExecute_ptr(hStmt);
+        return rc;
+    } else {
+        size_t rowCount = columnwise_params.size();
+        for (size_t rowIndex = 0; rowIndex < rowCount; ++rowIndex) {
+            py::list rowParams = columnwise_params[rowIndex];
+
+            std::vector<std::shared_ptr<void>> paramBuffers;
+            rc = BindParameters(hStmt, rowParams, const_cast<std::vector<ParamInfo>&>(paramInfos), paramBuffers);
+            if (!SQL_SUCCEEDED(rc)) return rc;
+
+            rc = SQLExecute_ptr(hStmt);
+            while (rc == SQL_NEED_DATA) {
+                SQLPOINTER token;
+                rc = SQLParamData_ptr(hStmt, &token);
+                if (!SQL_SUCCEEDED(rc) && rc != SQL_NEED_DATA) return rc;
+
+                py::object* py_obj_ptr = reinterpret_cast<py::object*>(token);
+                if (!py_obj_ptr) return SQL_ERROR;
+
+                if (py::isinstance<py::str>(*py_obj_ptr)) {
+                    std::string data = py_obj_ptr->cast<std::string>();
+                    SQLLEN data_len = static_cast<SQLLEN>(data.size());
+                    rc = SQLPutData_ptr(hStmt, (SQLPOINTER)data.c_str(), data_len);
+                } else if (py::isinstance<py::bytes>(*py_obj_ptr) || py::isinstance<py::bytearray>(*py_obj_ptr)) {
+                    std::string data = py_obj_ptr->cast<std::string>();
+                    SQLLEN data_len = static_cast<SQLLEN>(data.size());
+                    rc = SQLPutData_ptr(hStmt, (SQLPOINTER)data.c_str(), data_len);
+                } else {
+                    LOG("Unsupported DAE parameter type in row {}", rowIndex);
+                    return SQL_ERROR;
+                }
+            }
+
+            if (!SQL_SUCCEEDED(rc)) return rc;
+        }
+        return SQL_SUCCESS;
+    }
 }
 
+
 // Wrap SQLNumResultCols
 SQLSMALLINT SQLNumResultCols_wrap(SqlHandlePtr statementHandle) {
     LOG("Get number of columns in result set");
@@ -2213,7 +2262,7 @@ static py::object FetchLobColumnData(SQLHSTMT hStmt,
             LOG("Loop {}: Appended {} bytes", loopCount, bytesRead);
         }
         if (ret == SQL_SUCCESS) {
-            LOG("Loop {}: SQL_SUCCESS → no more data", loopCount);
+            LOG("Loop {}: SQL_SUCCESS, no more data", loopCount);
             break;
         }
     }
@@ -3270,7 +3319,7 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch
 
     // If we have LOBs → fall back to row-by-row fetch + SQLGetData_wrap
     if (!lobColumns.empty()) {
-        LOG("LOB columns detected → using per-row SQLGetData path");
+        LOG("LOB columns detected, using per-row SQLGetData path");
         while (true) {
             ret = SQLFetch_ptr(hStmt);
             if (ret == SQL_NO_DATA) break;
@@ -3392,7 +3441,7 @@ SQLRETURN FetchAll_wrap(SqlHandlePtr StatementHandle, py::list& rows) {
 
     // If we have LOBs → fall back to row-by-row fetch + SQLGetData_wrap
     if (!lobColumns.empty()) {
-        LOG("LOB columns detected → using per-row SQLGetData path");
+        LOG("LOB columns detected, using per-row SQLGetData path");
         while (true) {
             ret = SQLFetch_ptr(hStmt);
             if (ret == SQL_NO_DATA) break;

tests/test_004_cursor.py

@@ -10661,6 +10661,99 @@ def test_decimal_separator_calculations(cursor, db_connection):
         # Cleanup
         cursor.execute("DROP TABLE IF EXISTS #pytest_decimal_calc_test")
 
+def test_nvarcharmax_executemany_streaming(cursor, db_connection):
+    """Streaming insert + fetch > 4k NVARCHAR(MAX) using executemany with all fetch modes."""
+    try:
+        values = ["Ω" * 4100, "漢" * 5000]
+        cursor.execute("CREATE TABLE #pytest_nvarcharmax (col NVARCHAR(MAX))")
+        db_connection.commit()
+
+        # --- executemany insert ---
+        cursor.executemany("INSERT INTO #pytest_nvarcharmax VALUES (?)", [(v,) for v in values])
+        db_connection.commit()
+
+        # --- fetchall ---
+        cursor.execute("SELECT col FROM #pytest_nvarcharmax ORDER BY LEN(col)")
+        rows = [r[0] for r in cursor.fetchall()]
+        assert rows == sorted(values, key=len)
+
+        # --- fetchone ---
+        cursor.execute("SELECT col FROM #pytest_nvarcharmax ORDER BY LEN(col)")
+        r1 = cursor.fetchone()[0]
+        r2 = cursor.fetchone()[0]
+        assert {r1, r2} == set(values)
+        assert cursor.fetchone() is None
+
+        # --- fetchmany ---
+        cursor.execute("SELECT col FROM #pytest_nvarcharmax ORDER BY LEN(col)")
+        batch = [r[0] for r in cursor.fetchmany(1)]
+        assert batch[0] in values
+    finally:
+        cursor.execute("DROP TABLE #pytest_nvarcharmax")
+        db_connection.commit()
+
+def test_varcharmax_executemany_streaming(cursor, db_connection):
+    """Streaming insert + fetch > 4k VARCHAR(MAX) using executemany with all fetch modes."""
+    try:
+        values = ["A" * 4100, "B" * 5000]
+        cursor.execute("CREATE TABLE #pytest_varcharmax (col VARCHAR(MAX))")
+        db_connection.commit()
+
+        # --- executemany insert ---
+        cursor.executemany("INSERT INTO #pytest_varcharmax VALUES (?)", [(v,) for v in values])
+        db_connection.commit()
+
+        # --- fetchall ---
+        cursor.execute("SELECT col FROM #pytest_varcharmax ORDER BY LEN(col)")
+        rows = [r[0] for r in cursor.fetchall()]
+        assert rows == sorted(values, key=len)
+
+        # --- fetchone ---
+        cursor.execute("SELECT col FROM #pytest_varcharmax ORDER BY LEN(col)")
+        r1 = cursor.fetchone()[0]
+        r2 = cursor.fetchone()[0]
+        assert {r1, r2} == set(values)
+        assert cursor.fetchone() is None
+
+        # --- fetchmany ---
+        cursor.execute("SELECT col FROM #pytest_varcharmax ORDER BY LEN(col)")
+        batch = [r[0] for r in cursor.fetchmany(1)]
+        assert batch[0] in values
+    finally:
+        cursor.execute("DROP TABLE #pytest_varcharmax")
+        db_connection.commit()
+
+def test_varbinarymax_executemany_streaming(cursor, db_connection):
+    """Streaming insert + fetch > 4k VARBINARY(MAX) using executemany with all fetch modes."""
+    try:
+        values = [b"\x01" * 4100, b"\x02" * 5000]
+        cursor.execute("CREATE TABLE #pytest_varbinarymax (col VARBINARY(MAX))")
+        db_connection.commit()
+
+        # --- executemany insert ---
+        cursor.executemany("INSERT INTO #pytest_varbinarymax VALUES (?)", [(v,) for v in values])
+        db_connection.commit()
+
+        # --- fetchall ---
+        cursor.execute("SELECT col FROM #pytest_varbinarymax ORDER BY DATALENGTH(col)")
+        rows = [r[0] for r in cursor.fetchall()]
+        assert rows == sorted(values, key=len)
+
+        # --- fetchone ---
+        cursor.execute("SELECT col FROM #pytest_varbinarymax ORDER BY DATALENGTH(col)")
+        r1 = cursor.fetchone()[0]
+        r2 = cursor.fetchone()[0]
+        assert {r1, r2} == set(values)
+        assert cursor.fetchone() is None
+
+        # --- fetchmany ---
+        cursor.execute("SELECT col FROM #pytest_varbinarymax ORDER BY DATALENGTH(col)")
+        batch = [r[0] for r in cursor.fetchmany(1)]
+        assert batch[0] in values
+    finally:
+        cursor.execute("DROP TABLE #pytest_varbinarymax")
+        db_connection.commit()
+
 def test_date_string_parameter_binding(cursor, db_connection):
     """Verify that date-like strings are treated as strings in parameter binding"""
     table_name = "#pytest_date_string"