fix(dbapi): accumulate rows across executemany iterations (ISSUE-57)

For DML statements with a RETURNING clause (or any result-producing
iteration), executemany previously overwrote _rows / _description
on every param-set iteration — so fetchall() after executemany only
returned rows from the LAST parameter set, silently dropping the
rest. This broke SQLAlchemy's insert_executemany_returning
optimization for bulk inserts.

Accumulate rows and preserve description across iterations. Add an
integration test that verifies executemany+RETURNING returns every
row from every parameter set in insert order, for both sync and
async cursors.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -140,18 +140,32 @@ async def executemany(
         An empty ``seq_of_parameters`` must not leave stale SELECT
         state around: reset description / rows so callers can't
         confuse an empty executemany with a preceding SELECT.
+
+        For statements with a RETURNING clause, rows produced by each
+        iteration are accumulated into ``_rows`` so a subsequent
+        ``fetchall`` yields every returned row across parameter sets.
         """
         self._check_closed()
 
         self._description = None
         self._rows = []
         self._row_index = 0
         total_affected = 0
+        accumulated_rows: list[tuple[Any, ...]] = []
+        accumulated_desc: list[tuple[str, int | None, None, None, None, None, None]] | None = None
         for params in seq_of_parameters:
             await self.execute(operation, params)
             if self._rowcount >= 0:
                 total_affected += self._rowcount
+            if self._description is not None:
+                if accumulated_desc is None:
+                    accumulated_desc = self._description
+                accumulated_rows.extend(self._rows)
         self._rowcount = total_affected
+        if accumulated_desc is not None:
+            self._description = accumulated_desc
+            self._rows = accumulated_rows
+            self._row_index = 0
         return self
 
     def _check_result_set(self) -> None:

src/dqlitedbapi/cursor.py

@@ -265,16 +265,33 @@ async def _executemany_async(
         state around: reset description / rows to None / empty so
         callers can't confuse an empty executemany with a preceding
         SELECT.
+
+        For statements with a RETURNING clause (or any result-producing
+        DML), each iteration's rows are accumulated so ``fetchall`` at
+        the end yields every returned row across all parameter sets.
+        Without the accumulation, ``_execute_async`` would overwrite
+        ``_rows`` on each iteration and only the rows from the last
+        parameter set would survive.
         """
         self._description = None
         self._rows = []
         self._row_index = 0
         total_affected = 0
+        accumulated_rows: list[tuple[Any, ...]] = []
+        accumulated_desc: list[tuple[str, int | None, None, None, None, None, None]] | None = None
         for params in seq_of_parameters:
             await self._execute_async(operation, params)
             if self._rowcount >= 0:
                 total_affected += self._rowcount
+            if self._description is not None:
+                if accumulated_desc is None:
+                    accumulated_desc = self._description
+                accumulated_rows.extend(self._rows)
         self._rowcount = total_affected
+        if accumulated_desc is not None:
+            self._description = accumulated_desc
+            self._rows = accumulated_rows
+            self._row_index = 0
 
     def _check_result_set(self) -> None:
         if self._description is None:

tests/integration/test_executemany_returning.py

@@ -0,0 +1,44 @@
+"""Integration test for ISSUE-57: executemany must accumulate rows from
+every parameter set when the statement has a RETURNING clause."""
+
+import pytest
+
+import dqlitedbapi
+from dqlitedbapi.aio import aconnect
+
+
+@pytest.mark.integration
+def test_sync_executemany_returning_accumulates_rows(cluster_address: str) -> None:
+    with dqlitedbapi.connect(cluster_address) as conn:
+        cur = conn.cursor()
+        cur.execute("DROP TABLE IF EXISTS emany_ret")
+        cur.execute("CREATE TABLE emany_ret (id INTEGER PRIMARY KEY, x TEXT)")
+        params = [("a",), ("b",), ("c",)]
+        cur.executemany("INSERT INTO emany_ret (x) VALUES (?) RETURNING id", params)
+        assert cur.rowcount == 3
+        rows = cur.fetchall()
+        assert len(rows) == 3
+        # ids must be 1, 2, 3 in insert order.
+        assert [r[0] for r in rows] == [1, 2, 3]
+        conn.rollback()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_async_executemany_returning_accumulates_rows(
+    cluster_address: str,
+) -> None:
+    conn = await aconnect(cluster_address)
+    try:
+        cur = conn.cursor()
+        await cur.execute("DROP TABLE IF EXISTS emany_ret_async")
+        await cur.execute("CREATE TABLE emany_ret_async (id INTEGER PRIMARY KEY, x TEXT)")
+        params = [("alpha",), ("beta",), ("gamma",), ("delta",)]
+        await cur.executemany("INSERT INTO emany_ret_async (x) VALUES (?) RETURNING id, x", params)
+        assert cur.rowcount == 4
+        rows = await cur.fetchall()
+        assert len(rows) == 4
+        assert [r[1] for r in rows] == ["alpha", "beta", "gamma", "delta"]
+        await conn.rollback()
+    finally:
+        await conn.close()