feat(dbapi): use inline begin to eliminate BeginTransaction RPC (#1502)

Fixesgoogleapis/google-cloud-python#15871
## Summary
`Connection.transaction_checkout()` currently calls
`Transaction.begin()` explicitly before the first query, which sends a
standalone `BeginTransaction` gRPC RPC. This is unnecessary because the
`Transaction` class already supports **inline begin** — piggybacking
`BeginTransaction` onto the first `ExecuteSql`/`ExecuteBatchDml` request
via `TransactionSelector(begin=...)`.
This PR removes the explicit `begin()` call, letting the existing inline
begin logic in `execute_sql()`, `execute_update()`, and `batch_update()`
handle transaction creation. This saves **one gRPC round-trip per
transaction** (~16ms measured on the emulator).
### What changed
**`google/cloud/spanner_dbapi/connection.py`** —
`transaction_checkout()`:
- Removed `self._transaction.begin()` on
[L413](https://github.com/googleapis/python-spanner/blob/v3.63.0/google/cloud/spanner_dbapi/connection.py#L413)
- The transaction is now returned with `_transaction_id=None`
- Updated docstring to explain the inline begin behavior
**`tests/unit/spanner_dbapi/test_connection.py`**:
- Added `test_transaction_checkout_does_not_call_begin` to assert
`begin()` is not called
**`tests/mockserver_tests/test_dbapi_inline_begin.py`** (new):
- 9 mockserver tests verifying inline begin behavior for read-write
transactions
- Covers: no `BeginTransactionRequest` sent, first `ExecuteSqlRequest`
uses `TransactionSelector(begin=...)`, transaction ID reuse on second
statement, rollback, read-only unaffected, retry after abort
**`tests/mockserver_tests/test_tags.py`**:
- Updated 4 read-write tag tests: removed `BeginTransactionRequest` from
expected RPC sequences, adjusted tag index offsets
**`tests/mockserver_tests/test_dbapi_isolation_level.py`**:
- Updated 4 isolation level tests: verify isolation level on
`ExecuteSqlRequest.transaction.begin.isolation_level` instead of
`BeginTransactionRequest.options.isolation_level`
### Why this is safe
The inline begin code path already exists and is battle-tested —
`Session.run_in_transaction()` creates a `Transaction` without calling
`begin()` and relies on the same inline begin logic ([session.py
L566](https://github.com/googleapis/python-spanner/blob/v3.63.0/google/cloud/spanner_v1/session.py#L566)).
Specific safety analysis:
1. **`transaction_checkout()` callers always execute SQL immediately**:
It's only called from `run_statement()` → `execute_sql()` and
`batch_dml_executor` → `batch_update()`. Both set `_transaction_id` via
inline begin before any commit/rollback path.
2. **`execute_sql`/`execute_update`/`batch_update` handle
`_transaction_id is None`**: They acquire a lock, use
`_make_txn_selector()` which returns `TransactionSelector(begin=...)`,
and store the returned `_transaction_id` ([transaction.py
L612-L623](https://github.com/googleapis/python-spanner/blob/v3.63.0/google/cloud/spanner_v1/transaction.py#L612-L623)).
3. **`rollback()` handles `_transaction_id is None`**: Skips the RPC —
correct when no server-side transaction exists ([transaction.py
L163](https://github.com/googleapis/python-spanner/blob/v3.63.0/google/cloud/spanner_v1/transaction.py#L163)).
4. **`commit()` handles `_transaction_id is None`**: Falls back to
`_begin_mutations_only_transaction()` for mutation-only transactions
([transaction.py
L263-L267](https://github.com/googleapis/python-spanner/blob/v3.63.0/google/cloud/spanner_v1/transaction.py#L263-L267)).
5. **Retry mechanism is compatible**: `_set_connection_for_retry()`
resets `_spanner_transaction_started=False`, so replayed statements go
through `transaction_checkout()` again, create a fresh `Transaction`,
and use inline begin.
### PEP 249 conformance
This change is fully conformant with [PEP 249 (DB-API
2.0)](https://peps.python.org/pep-0249/). The spec does not define a
`begin()` method — transactions are implicit. The PEP author [clarified
on the DB-SIG mailing
list](https://mail.python.org/pipermail/db-sig/2010-September/005645.html)
that *"transactions start implicitly after you connect and after you
call `.commit()` or `.rollback()`"*, and the mechanism by which the
driver starts the server-side transaction is an implementation detail.
Deferring the server-side begin to the first SQL execution (as psycopg2
and other mature DB-API drivers do) is the standard approach. The
observable transactional semantics — atomicity between
`commit()`/`rollback()` calls — are unchanged.
### Performance impact
Before (4 RPCs per read-write transaction):
```
BeginTransaction → ExecuteSql (read) → ExecuteSql (write) → Commit
```
After (3 RPCs per read-write transaction):
```
ExecuteSql (read, with inline begin) → ExecuteSql (write) → Commit
```
Measured ~16ms savings per transaction on the Spanner emulator.
### Context
This optimization was identified while profiling SQLAlchemy/DBAPI
performance against Spanner. The DBAPI was created ~2 years before
inline begin support was added to the Python client library (inline
begin landed in [PR
#740](#740),
Dec 2022). The `run_in_transaction` path was updated to use inline
begin, but `transaction_checkout` was not.
## Test plan
- [x] All 198 DBAPI unit tests pass (`tests/unit/spanner_dbapi/`)
- [x] New unit test verifies `begin()` is not called by
`transaction_checkout()`
- [x] 9 new mockserver tests verify inline begin RPC behavior
(`tests/mockserver_tests/test_dbapi_inline_begin.py`)
- [x] 8 existing mockserver tests updated for inline begin expectations
(`test_tags.py`, `test_dbapi_isolation_level.py`)
- [x] All 53 DBAPI system tests pass against Spanner emulator
(`tests/system/test_dbapi.py`)
- [ ] CI system tests (will run automatically)
---------
Co-authored-by: rahul2393 <irahul@google.com>
Co-authored-by: Knut Olav Løite <koloite@gmail.com>

Author: 3 people

google/cloud/spanner_dbapi/batch_dml_executor.py

@@ -104,6 +104,11 @@ def run_batch_dml(cursor: "Cursor", statements: List[Statement]):
 connection._transaction=None
 raiseAborted(status.message)
 elifstatus.code!=OK:
+ifnottransaction._transaction_id:
+# This should normally not happen,
+# but we safeguard against it just to be sure.
+transaction._reset_and_begin()
+continue
 raiseOperationalError(status.message)
 
 cursor._batch_dml_rows_count=res
@@ -116,6 +121,11 @@ def run_batch_dml(cursor: "Cursor", statements: List[Statement]):
 raise
 else:
 connection._transaction_helper.retry_transaction()
+exceptExceptionasex:
+ifnottransaction._transaction_id:
+transaction._reset_and_begin()
+continue
+raiseex
 
 
 def_do_batch_update_autocommit(transaction, statements):

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()
 
 returnself._transaction
 

google/cloud/spanner_dbapi/cursor.py

@@ -366,6 +366,16 @@ def _execute_in_rw_transaction(self):
 raise
 else:
 self.transaction_helper.retry_transaction()
+exceptExceptionasex:
+# In case of inline-begin failure, the transaction isn't started.
+# We immediately retry with an explicit BeginTransaction.
+transaction=getattr(self.connection, "_transaction", None)
+iftransactionandnottransaction._transaction_id:
+transaction._reset_and_begin()
+
+# Let the existing retry loop handle the retry of the statement
+continue
+raiseex
 else:
 self.connection.database.run_in_transaction(
 self._do_execute_update_in_autocommit,

google/cloud/spanner_v1/database_sessions_manager.py

@@ -155,8 +155,6 @@ def _build_multiplexed_session(self) -> Session:
  )
 session.create()
 
-self._database.logger.info("Created multiplexed session.")
-
 returnsession
 
 def_build_maintenance_thread(self) ->Thread:

google/cloud/spanner_v1/testing/mock_spanner.py

@@ -15,9 +15,11 @@
 importinspect
 importgrpc
 fromconcurrentimportfutures
+fromdataclassesimportdataclass
 
-fromgoogle.protobufimportempty_pb2
 fromgrpc_status.rpc_statusimport_Status
+fromgoogle.rpc.code_pb2importOK
+fromgoogle.protobufimportempty_pb2
 
 fromgoogle.cloud.spanner_v1import (
 TransactionOptions,
@@ -53,10 +55,23 @@ def get_result(self, sql: str) -> result_set.ResultSet:
 returnresult
 
 defadd_error(self, method: str, error: _Status):
+ifnothasattr(self, "_errors_list"):
+self._errors_list= {}
+ifmethodnotinself._errors_list:
+self._errors_list[method] = []
+self._errors_list[method].append(error)
 self.errors[method] =error
 
 defpop_error(self, context):
 name=inspect.currentframe().f_back.f_code.co_name
+ifhasattr(self, "_errors_list") andnameinself._errors_list:
+ifself._errors_list[name]:
+error=self._errors_list[name].pop(0)
+context.abort_with_status(error)
+return
+return# Queue is empty, return normally (no error)
+
+# Fallback to single error
 error: _Status|None=self.errors.pop(name, None)
 iferror:
 context.abort_with_status(error)
@@ -94,6 +109,12 @@ def get_result_as_partial_result_sets(
 returnpartials
 
 
+@dataclass
+classBatchDmlResponseConfig:
+status: _Status
+include_transaction_id: bool=True
+
+
 # An in-memory mock Spanner server that can be used for testing.
 classSpannerServicer(spanner_grpc.SpannerServicer):
 def__init__(self):
@@ -103,6 +124,7 @@ def __init__(self):
 self.transaction_counter=0
 self.transactions= {}
 self._mock_spanner=MockSpanner()
+self._batch_dml_response_configs= []
 
 @property
 defmock_spanner(self):
@@ -115,6 +137,15 @@ def requests(self):
 defclear_requests(self):
 self._requests= []
 
+defadd_batch_dml_response_status(self, status, include_transaction_id=True):
+ifnothasattr(self, "_batch_dml_response_configs"):
+self._batch_dml_response_configs= []
+self._batch_dml_response_configs.append(
+BatchDmlResponseConfig(
+status=status, include_transaction_id=include_transaction_id
+ )
+ )
+
 defCreateSession(self, request, context):
 self._requests.append(request)
 returnself.__create_session(request.database, request.session)
@@ -176,6 +207,14 @@ def ExecuteBatchDml(self, request, context):
 self.mock_spanner.pop_error(context)
 response=spanner.ExecuteBatchDmlResponse()
 started_transaction=self.__maybe_create_transaction(request)
+
+config=None
+if (
+hasattr(self, "_batch_dml_response_configs")
+andself._batch_dml_response_configs
+ ):
+config=self._batch_dml_response_configs.pop(0)
+
 first=True
 forstatementinrequest.statements:
 result=self.mock_spanner.get_result(statement.sql)
@@ -184,8 +223,16 @@ def ExecuteBatchDml(self, request, context):
 self.mock_spanner.get_result(statement.sql)
  )
 result.metadata=result_set.ResultSetMetadata(result.metadata)
-result.metadata.transaction=started_transaction
+ifconfigisNoneorconfig.include_transaction_id:
+result.metadata.transaction=started_transaction
+first=False
 response.result_sets.append(result)
+
+ifconfigisnotNone:
+response.status.CopyFrom(config.status)
+else:
+response.status.code=OK
+
 returnresponse
 
 defRead(self, request, context):

google/cloud/spanner_v1/transaction.py

@@ -214,6 +214,12 @@ def wrapped_method(*args, **kwargs):
 
 self.rolled_back=True
 
+def_reset_and_begin(self):
+"""This function can be used to reset the transaction and execute an explicit BeginTransaction RPC if the first statement in the transaction failed, and that statement included an inlined BeginTransaction option."""
+self._read_request_count=0
+self._execute_sql_request_count=0
+self.begin()
+
 defcommit(
 self, return_commit_stats=False, request_options=None, max_commit_delay=None
  ):

tests/mockserver_tests/mock_server_test_base.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 importlogging
+importos
 importunittest
 
 importgrpc
@@ -65,6 +66,19 @@ def aborted_status() -> _Status:
 returnstatus
 
 
+definvalid_argument_status() ->_Status:
+error=status_pb2.Status(
+code=code_pb2.INVALID_ARGUMENT,
+message="Invalid argument.",
+ )
+status=_Status(
+code=code_to_grpc_status_code(error.code),
+details=error.message,
+trailing_metadata=(("grpc-status-details-bin", error.SerializeToString()),),
+ )
+returnstatus
+
+
 def_make_partial_result_sets(
 fields: list[tuple[str, TypeCode]], results: list[dict]
 ) ->list[result_set.PartialResultSet]:
@@ -174,6 +188,9 @@ class MockServerTestBase(unittest.TestCase):
 
 def__init__(self, *args, **kwargs):
 super(MockServerTestBase, self).__init__(*args, **kwargs)
+# Disable built-in metrics for tests to avoid Unauthenticated errors
+os.environ["SPANNER_DISABLE_BUILTIN_METRICS"] ="true"
+
 self._client=None
 self._instance=None
 self._database=None