Add subcursor() for interleaved streaming + DML within a transaction

Adds subcursor() method that creates a cursor sharing the same
underlying Connection and transaction as the parent. This enables
interleaved streaming SELECT + UPDATE/INSERT/DELETE within an
explicit transaction without destroying the streaming result.

Requires the companion duckdb core PR that adds the
enable_suspended_queries setting and suspend/resume mechanics.

Key changes:
- ConnectionGuard: unique_ptr<Connection> to shared_ptr<Connection>
- Add ShareConnection/GetSharedConnection accessors
- Add Subcursor() method that shares the parent's Connection
- Add is_subcursor flag with transaction management guard
- Subcursors cannot call begin/commit/rollback
- Closing a subcursor cleans up any suspended query state

Author: rustyconover

src/duckdb_py/include/duckdb_python/pyconnection/pyconnection.hpp

@@ -131,6 +131,12 @@ struct ConnectionGuard {
 	void SetConnection(unique_ptr<Connection> con) {
 		connection = std::move(con);
 	}
+	void ShareConnection(shared_ptr<Connection> con) {
+		connection = std::move(con);
+	}
+	shared_ptr<Connection> GetSharedConnection() {
+		return connection;
+	}
 	void SetResult(unique_ptr<DuckDBPyRelation> res) {
 		result = std::move(res);
 	}
@@ -142,7 +148,7 @@ struct ConnectionGuard {
 
 private:
 	shared_ptr<DuckDB> database;
-	unique_ptr<Connection> connection;
+	shared_ptr<Connection> connection;
 	unique_ptr<DuckDBPyRelation> result;
 };
 
@@ -165,7 +171,11 @@ struct DuckDBPyConnection : public enable_shared_from_this<DuckDBPyConnection> {
 public:
 	ConnectionGuard con;
 	Cursors cursors;
-	std::mutex py_connection_lock;
+	std::mutex owned_py_connection_lock;
+	//! Points to owned_py_connection_lock by default, or to parent's lock for subcursors
+	std::mutex *py_connection_lock = &owned_py_connection_lock;
+	//! Whether this is a subcursor (shares connection with parent)
+	bool is_subcursor = false;
 	//! MemoryFileSystem used to temporarily store file-like objects for reading
 	shared_ptr<ModifiedMemoryFileSystem> internal_object_filesystem;
 	case_insensitive_map_t<unique_ptr<ExternalDependency>> registered_functions;
@@ -303,6 +313,10 @@ struct DuckDBPyConnection : public enable_shared_from_this<DuckDBPyConnection> {
 	// cursor() is stupid
 	shared_ptr<DuckDBPyConnection> Cursor();
 
+	//! Create a subcursor that shares the same connection and transaction.
+	//! This enables interleaved streaming + DML within a single transaction.
+	shared_ptr<DuckDBPyConnection> Subcursor();
+
 	Optional<py::list> GetDescription();
 
 	int GetRowcount();

src/duckdb_py/pyconnection.cpp

@@ -483,6 +483,8 @@ void DuckDBPyConnection::Initialize(py::handle &m) {
 	connection_module.def("__del__", &DuckDBPyConnection::Close);
 
 	InitializeConnectionMethods(connection_module);
+	connection_module.def("subcursor", &DuckDBPyConnection::Subcursor,
+	                      "Create a cursor sharing the same connection and transaction");
 	connection_module.def_property_readonly("description", &DuckDBPyConnection::GetDescription,
 	                                        "Get result set attributes, mainly column names");
 	connection_module.def_property_readonly("rowcount", &DuckDBPyConnection::GetRowcount, "Get result set row count");
@@ -623,7 +625,7 @@ unique_ptr<PreparedStatement> DuckDBPyConnection::PrepareQuery(unique_ptr<SQLSta
 	{
 		D_ASSERT(py::gil_check());
 		py::gil_scoped_release release;
-		unique_lock<mutex> lock(py_connection_lock);
+		unique_lock<mutex> lock(*py_connection_lock);
 
 		prep = connection.Prepare(std::move(statement));
 		if (prep->HasError()) {
@@ -644,7 +646,7 @@ unique_ptr<QueryResult> DuckDBPyConnection::ExecuteInternal(PreparedStatement &p
 	{
 		D_ASSERT(py::gil_check());
 		py::gil_scoped_release release;
-		unique_lock<std::mutex> lock(py_connection_lock);
+		unique_lock<std::mutex> lock(*py_connection_lock);
 
 		auto pending_query = prep.PendingQuery(named_values);
 		if (pending_query->HasError()) {
@@ -671,7 +673,7 @@ unique_ptr<QueryResult> DuckDBPyConnection::PrepareAndExecuteInternal(unique_ptr
 	{
 		D_ASSERT(py::gil_check());
 		py::gil_scoped_release release;
-		unique_lock<std::mutex> lock(py_connection_lock);
+		unique_lock<std::mutex> lock(*py_connection_lock);
 
 		auto pending_query = con.GetConnection().PendingQuery(std::move(statement), named_values, true);
 
@@ -1855,11 +1857,17 @@ shared_ptr<DuckDBPyConnection> DuckDBPyConnection::UnregisterPythonObject(const
 }
 
 shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Begin() {
+	if (is_subcursor) {
+		throw InvalidInputException("Cannot manage transactions from a subcursor — use the parent connection");
+	}
 	ExecuteFromString("BEGIN TRANSACTION");
 	return shared_from_this();
 }
 
 shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Commit() {
+	if (is_subcursor) {
+		throw InvalidInputException("Cannot manage transactions from a subcursor — use the parent connection");
+	}
 	auto &connection = con.GetConnection();
 	if (connection.context->transaction.IsAutoCommit()) {
 		return shared_from_this();
@@ -1869,6 +1877,9 @@ shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Commit() {
 }
 
 shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Rollback() {
+	if (is_subcursor) {
+		throw InvalidInputException("Cannot manage transactions from a subcursor — use the parent connection");
+	}
 	ExecuteFromString("ROLLBACK");
 	return shared_from_this();
 }
@@ -2024,6 +2035,17 @@ shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Cursor() {
 	return res;
 }
 
+shared_ptr<DuckDBPyConnection> DuckDBPyConnection::Subcursor() {
+	auto res = make_shared_ptr<DuckDBPyConnection>();
+	res->con.SetDatabase(con);
+	res->con.ShareConnection(con.GetSharedConnection());
+	res->is_subcursor = true;
+	// Share the parent's py_connection_lock so concurrent subcursor access is serialized
+	res->py_connection_lock = py_connection_lock;
+	cursors.AddCursor(res);
+	return res;
+}
+
 // these should be functions on the result but well
 Optional<py::tuple> DuckDBPyConnection::FetchOne() {
 	if (!con.HasResult()) {
@@ -2185,7 +2207,7 @@ static shared_ptr<DuckDBPyConnection> FetchOrCreateInstance(const string &databa
 	{
 		D_ASSERT(py::gil_check());
 		py::gil_scoped_release release;
-		unique_lock<mutex> lock(res->py_connection_lock);
+		unique_lock<mutex> lock(*res->py_connection_lock);
 		auto database =
 		    instance_cache.GetOrCreateInstance(database_path, config, cache_instance, InstantiateNewInstance);
 		res->con.SetDatabase(std::move(database));

tests/fast/api/test_subcursor.py

@@ -0,0 +1,248 @@
+"""Tests for subcursor() — a cursor that shares the same connection and transaction."""
+
+import pytest
+import duckdb
+
+
+class TestSubcursor:
+    def test_subcursor_basic_interleaved(self):
+        """Single scan + update pattern with subcursor."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT i AS id, false AS processed FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        scanner = con.subcursor()
+        updater = con.subcursor()
+
+        result = scanner.execute("SELECT id FROM t WHERE NOT processed ORDER BY id")
+        batch = result.fetchmany(1000)
+        assert len(batch) == 1000
+        assert batch[0][0] == 0
+
+        # Update some rows via the subcursor
+        ids = ",".join(str(r[0]) for r in batch[:10])
+        updater.execute(f"UPDATE t SET processed = true WHERE id IN ({ids})")
+
+        # Resume fetching from the scanner — should still work
+        batch2 = result.fetchmany(1000)
+        assert len(batch2) > 0
+
+        con.execute("COMMIT")
+
+        # Verify updates persisted
+        res = con.execute("SELECT COUNT(*) FROM t WHERE processed").fetchone()
+        assert res[0] == 10
+
+    def test_subcursor_multi_table(self):
+        """Multi-table scan pattern with subcursors."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t1 AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("CREATE TABLE t2 AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        scan_a = con.subcursor()
+        scan_b = con.subcursor()
+        updater = con.subcursor()
+
+        result_a = scan_a.execute("SELECT id FROM t1 ORDER BY id")
+        result_b = scan_b.execute("SELECT id FROM t2 ORDER BY id")
+
+        # Interleave fetches from both tables
+        batch_a = result_a.fetchmany(100)
+        assert len(batch_a) == 100
+
+        batch_b = result_b.fetchmany(100)
+        assert len(batch_b) == 100
+
+        # Update via subcursor
+        updater.execute("UPDATE t1 SET id = id + 1000000 WHERE id < 5")
+
+        # Resume both scans
+        batch_a2 = result_a.fetchmany(100)
+        assert len(batch_a2) > 0
+
+        batch_b2 = result_b.fetchmany(100)
+        assert len(batch_b2) > 0
+
+        con.execute("COMMIT")
+
+    def test_subcursor_transaction_guard(self):
+        """Subcursor cannot manage transactions."""
+        con = duckdb.connect(":memory:")
+        sub = con.subcursor()
+
+        with pytest.raises(duckdb.InvalidInputException, match="subcursor"):
+            sub.begin()
+
+        with pytest.raises(duckdb.InvalidInputException, match="subcursor"):
+            sub.commit()
+
+        with pytest.raises(duckdb.InvalidInputException, match="subcursor"):
+            sub.rollback()
+
+    def test_subcursor_shares_transaction(self):
+        """Subcursor sees data from the same transaction."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t (id INTEGER)")
+        con.execute("BEGIN TRANSACTION")
+        con.execute("INSERT INTO t VALUES (1), (2), (3)")
+
+        sub = con.subcursor()
+        # Subcursor should see uncommitted data from the parent's transaction
+        res = sub.execute("SELECT COUNT(*) FROM t").fetchone()
+        assert res[0] == 3
+
+        con.execute("COMMIT")
+
+    def test_subcursor_close(self):
+        """Subcursor close doesn't destroy parent connection."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT 1 AS id")
+
+        sub = con.subcursor()
+        sub.execute("SELECT * FROM t")
+        sub.close()
+
+        # Parent connection should still work
+        res = con.execute("SELECT * FROM t").fetchone()
+        assert res[0] == 1
+
+    def test_subcursor_full_consumption(self):
+        """Verify stream can be fully consumed after suspend/resume."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        scanner = con.subcursor()
+        updater = con.subcursor()
+
+        result = scanner.execute("SELECT id FROM t ORDER BY id")
+        total = 0
+
+        while True:
+            batch = result.fetchmany(2048)
+            if not batch:
+                break
+            total += len(batch)
+            # Interleave an update every few batches
+            if total % 8192 == 0:
+                updater.execute(f"UPDATE t SET id = id WHERE id = {total}")
+
+        assert total == 100000
+        con.execute("COMMIT")
+
+    def test_subcursor_requires_explicit_transaction(self):
+        """Subcursor interleaving requires an explicit transaction (BEGIN)."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT i AS id FROM range(100000) tbl(i)")
+
+        # With explicit transaction and setting enabled, interleaving works
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+        scanner = con.subcursor()
+        updater = con.subcursor()
+
+        result = scanner.execute("SELECT id FROM t ORDER BY id")
+        batch = result.fetchmany(1000)
+        assert len(batch) == 1000
+
+        updater.execute("UPDATE t SET id = id WHERE id < 5")
+
+        # Stream survives the update in explicit transaction mode
+        batch2 = result.fetchmany(1000)
+        assert len(batch2) > 0
+
+        con.execute("COMMIT")
+
+    def test_subcursor_close_cancels_suspended_stream(self):
+        """Closing a subcursor cancels its suspended stream and frees resources."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        scanner = con.subcursor()
+        updater = con.subcursor()
+
+        result = scanner.execute("SELECT id FROM t ORDER BY id")
+        batch = result.fetchmany(1000)
+        assert len(batch) == 1000
+
+        # Suspend the stream by executing on the updater
+        updater.execute("UPDATE t SET id = id WHERE id < 5")
+
+        # Close the scanner — should cancel the suspended stream
+        scanner.close()
+
+        # The connection should still work normally
+        updater.execute("SELECT COUNT(*) FROM t")
+        count = updater.fetchone()[0]
+        assert count == 100000
+
+        con.execute("COMMIT")
+
+    def test_subcursor_close_one_of_multiple(self):
+        """Closing one subcursor doesn't affect other suspended streams."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t1 AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("CREATE TABLE t2 AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        scan1 = con.subcursor()
+        scan2 = con.subcursor()
+
+        result1 = scan1.execute("SELECT id FROM t1 ORDER BY id")
+        result2 = scan2.execute("SELECT id FROM t2 ORDER BY id")
+
+        # Fetch from both
+        batch1 = result1.fetchmany(100)
+        assert len(batch1) == 100
+        batch2 = result2.fetchmany(100)
+        assert len(batch2) == 100
+
+        # Close scan1 — scan2 should still work
+        scan1.close()
+
+        # scan2 should still be able to fetch
+        total2 = len(batch2)
+        while True:
+            batch = result2.fetchmany(2048)
+            if not batch:
+                break
+            total2 += len(batch)
+        assert total2 == 100000
+
+        # scan1 should be unusable
+        with pytest.raises(duckdb.ConnectionException):
+            scan1.execute("SELECT 1")
+
+        con.execute("COMMIT")
+
+    def test_subcursor_abandoned_without_close(self):
+        """Subcursor that goes out of scope should clean up its suspended stream."""
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE TABLE t AS SELECT i AS id FROM range(100000) tbl(i)")
+        con.execute("SET enable_suspended_queries = true")
+        con.execute("BEGIN TRANSACTION")
+
+        def start_and_abandon():
+            scanner = con.subcursor()
+            result = scanner.execute("SELECT id FROM t ORDER BY id")
+            result.fetchmany(100)
+            # Suspend by running something else
+            con.execute("SELECT 1")
+            # scanner goes out of scope here — destructor should clean up
+
+        start_and_abandon()
+        import gc
+        gc.collect()
+
+        # Connection should still work — no leaked suspended state blocking it
+        result = con.execute("SELECT COUNT(*) FROM t").fetchone()
+        assert result[0] == 100000
+
+        con.execute("COMMIT")