feat(dbapi): use inline begin in transaction_checkout to eliminate BeginTransaction RPC

Remove the explicit `self._transaction.begin()` call from
`Connection.transaction_checkout()`. The transaction is now returned
with `_transaction_id=None`, allowing `Transaction.execute_sql()`,
`execute_update()`, and `batch_update()` to use their existing
inline begin logic via `TransactionSelector(begin=...)`.

This piggybacks the BeginTransaction onto the first ExecuteSql RPC,
eliminating one round-trip per transaction (~16ms on the emulator).

The inline begin code path already exists in the Transaction class
and is used by `Session.run_in_transaction()`. This change makes the
DBAPI layer use the same optimization.

Safety:
- `transaction_checkout()` is only called from `run_statement()` and
  `batch_dml_executor`, both of which immediately execute SQL, so
  `_transaction_id` is always set before commit/rollback.
- `Transaction.rollback()` already handles `_transaction_id is None`
  (skips the RPC — correct when no server-side txn exists).
- `Transaction.commit()` handles `_transaction_id is None` for
  mutation-only transactions via `_begin_mutations_only_transaction()`.
- The retry mechanism (`_set_connection_for_retry`) resets
  `_spanner_transaction_started=False`, so replayed statements
  create a fresh Transaction and use inline begin again.

All 198 DBAPI unit tests and 53 system tests (against the Spanner
emulator) pass with this change.

Made-with: Cursor

Author: waiho-gumloop

google/cloud/spanner_dbapi/connection.py

@@ -392,7 +392,14 @@ def transaction_checkout(self):
         this connection yet. Return the started one otherwise.
 
         This method is a no-op if the connection is in autocommit mode and no
-        explicit transaction has been started
+        explicit transaction has been started.
+
+        The transaction is returned without calling ``begin()``. The
+        underlying ``Transaction.execute_sql`` and ``execute_update``
+        methods detect ``_transaction_id is None`` and use *inline begin*
+        — piggybacking a ``BeginTransaction`` on the first RPC via
+        ``TransactionSelector(begin=...)``. This eliminates a separate
+        ``BeginTransaction`` RPC round-trip per transaction.
 
         :rtype: :class:`google.cloud.spanner_v1.transaction.Transaction`
         :returns: A Cloud Spanner transaction object, ready to use.
@@ -410,7 +417,6 @@ def transaction_checkout(self):
                 self.transaction_tag = None
                 self._snapshot = None
                 self._spanner_transaction_started = True
-                self._transaction.begin()
 
             return self._transaction
 

tests/unit/spanner_dbapi/test_connection.py

@@ -211,6 +211,26 @@ def test_transaction_checkout(self):
         connection._autocommit = True
         self.assertIsNone(connection.transaction_checkout())
 
+    def test_transaction_checkout_does_not_call_begin(self):
+        """transaction_checkout must not call Transaction.begin().
+
+        The transaction should be returned with _transaction_id=None so that
+        execute_sql/execute_update can use inline begin via
+        TransactionSelector(begin=...), eliminating a separate
+        BeginTransaction RPC.
+        """
+        connection = Connection(INSTANCE, DATABASE)
+        mock_session = mock.MagicMock()
+        mock_transaction = mock.MagicMock()
+        mock_session.transaction.return_value = mock_transaction
+        connection._session_checkout = mock.MagicMock(return_value=mock_session)
+
+        txn = connection.transaction_checkout()
+
+        self.assertEqual(txn, mock_transaction)
+        self.assertTrue(connection._spanner_transaction_started)
+        mock_transaction.begin.assert_not_called()
+
     def test_snapshot_checkout(self):
         connection = build_connection(read_only=True)
         connection.autocommit = False