perf(spanner): eliminate redundant tracing span in run_in_transaction

Eliminate the redundant inner 'CloudSpanner.Session.run_in_transaction' span and duplicate MetricsCapture context, leaving 'CloudSpanner.Database.run_in_transaction' as the single transaction lifecycle span (matching Java and Go client behavior).

- Attach transaction retry, abort, and rollback events directly to the active span.
- Guard span event creation, retry delay computation, and exception stringification behind span.is_recording().
- Refactor the run_in_transaction retry loop by extracting _create_transaction_for_attempt and _handle_aborted helpers.

Author: olavloite

packages/google-cloud-spanner/.cross_sync/generate.py

@@ -12,8 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from __future__ import annotations
-from typing import Sequence
+
 import ast
+from typing import Sequence
+
 """
 Entrypoint for initiating an async -> sync conversion using CrossSync
 
@@ -35,12 +37,13 @@ def extract_header_comments(file_path) -> str:
                 header.append(line)
             else:
                 break
-    header.append("\n# This file is automatically generated by CrossSync. Do not edit manually.\n\n")
+    header.append(
+        "\n# This file is automatically generated by CrossSync. Do not edit manually.\n\n"
+    )
     return "".join(header)
 
 
 class CrossSyncOutputFile:
-
     def __init__(self, output_path: str, ast_tree, header: str | None = None):
         self.output_path = output_path
         self.tree = ast_tree
@@ -56,15 +59,19 @@ def render(self, with_formatter=True, save_to_disk: bool = True) -> str:
         """
         full_str = self.header + ast.unparse(self.tree)
         if with_formatter:
-            import black  # type: ignore
-            import autoflake  # type: ignore
-
-            full_str = black.format_str(
-                autoflake.fix_code(full_str, remove_all_unused_imports=True),
-                mode=black.FileMode(),
-            )
+            try:
+                import autoflake  # type: ignore
+                import black  # type: ignore
+
+                full_str = black.format_str(
+                    autoflake.fix_code(full_str, remove_all_unused_imports=True),
+                    mode=black.FileMode(),
+                )
+            except ImportError:
+                pass
         if save_to_disk:
             import os
+
             os.makedirs(os.path.dirname(self.output_path), exist_ok=True)
             with open(self.output_path, "w") as f:
                 f.write(full_str)
@@ -73,10 +80,15 @@ def render(self, with_formatter=True, save_to_disk: bool = True) -> str:
 
 def convert_files_in_dir(directory: str) -> set[CrossSyncOutputFile]:
     import glob
+    import os
+
     from transformers import CrossSyncFileProcessor
 
-    # find all python files in the directory
-    files = glob.glob(directory + "/**/*.py", recursive=True)
+    # find all python files in the directory or use single file
+    if os.path.isfile(directory):
+        files = [directory]
+    else:
+        files = glob.glob(directory + "/**/*.py", recursive=True)
     # keep track of the output files pointed to by the annotated classes
     artifacts: set[CrossSyncOutputFile] = set()
     file_transformer = CrossSyncFileProcessor()

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/session.py

@@ -518,6 +518,107 @@ def transaction(self, client_context=None) -> Transaction:
 
         return Transaction(self, client_context=client_context)
 
+    def _create_transaction_for_attempt(
+        self,
+        client_context=None,
+        transaction_tag=None,
+        exclude_txn_from_change_streams=None,
+        isolation_level=None,
+        read_lock_mode=None,
+        previous_transaction_id=None,
+    ) -> Transaction:
+        """Create and configure a transaction instance for a single attempt.
+
+        :type client_context: :class:`~google.cloud.spanner_v1.client_context.ClientContext`
+        :param client_context: (Optional) client context to use for the transaction.
+
+        :type transaction_tag: str
+        :param transaction_tag: (Optional) transaction tag.
+
+        :type exclude_txn_from_change_streams: bool
+        :param exclude_txn_from_change_streams: (Optional) whether to exclude from change streams.
+
+        :type isolation_level: int
+        :param isolation_level: (Optional) isolation level.
+
+        :type read_lock_mode: int
+        :param read_lock_mode: (Optional) read lock mode.
+
+        :type previous_transaction_id: bytes
+        :param previous_transaction_id: (Optional) previous transaction id for multiplexed sessions.
+
+        :rtype: :class:`~google.cloud.spanner_v1.transaction.Transaction`
+        :returns: A configured Transaction instance.
+        """
+        transaction = self.transaction(client_context=client_context)
+        transaction.transaction_tag = transaction_tag
+        transaction.exclude_txn_from_change_streams = exclude_txn_from_change_streams
+        transaction.isolation_level = isolation_level
+        transaction.read_lock_mode = read_lock_mode
+
+        if self.is_multiplexed:
+            transaction._multiplexed_session_previous_transaction_id = (
+                previous_transaction_id
+            )
+        return transaction
+
+    @CrossSync.convert
+    async def _handle_aborted(
+        self,
+        exception,
+        span,
+        event_name,
+        attempts,
+        deadline,
+        default_retry_delay,
+        include_cause=False,
+    ):
+        """Handle an Aborted error: record trace event and delay until next retry attempt.
+
+        :type exception: :class:`google.api_core.exceptions.Aborted`
+        :param exception: The aborted exception.
+
+        :type span: :class:`opentelemetry.trace.Span`
+        :param span: The current active span.
+
+        :type event_name: str
+        :param event_name: The span event name to record.
+
+        :type attempts: int
+        :param attempts: The retry attempt number.
+
+        :type deadline: float
+        :param deadline: Timestamp deadline for retrying.
+
+        :type default_retry_delay: float
+        :param default_retry_delay: Default delay between retries.
+
+        :type include_cause: bool
+        :param include_cause: (Optional) Whether to include the exception string as cause.
+        """
+        if span and span.is_recording():
+            cause = (
+                exception.errors[0] if getattr(exception, "errors", None) else exception
+            )
+            delay_seconds = _get_retry_delay(
+                cause,
+                attempts,
+                default_retry_delay=default_retry_delay,
+            )
+            attributes = {
+                "attempt": attempts,
+                "delay_seconds": delay_seconds,
+            }
+            if include_cause:
+                attributes["cause"] = str(exception)
+            add_span_event(span, event_name, attributes)
+        await _delay_until_retry(
+            exception,
+            deadline,
+            attempts,
+            default_retry_delay=default_retry_delay,
+        )
+
     @CrossSync.convert
     async def run_in_transaction(self, func, *args, **kw):
         """Perform a unit of work in a transaction, retrying on abort.
@@ -568,126 +669,94 @@ async def run_in_transaction(self, func, *args, **kw):
         database = self._database
         log_commit_stats = database.log_commit_stats
 
-        extra_attributes = {}
-        if transaction_tag:
-            extra_attributes["transaction.tag"] = transaction_tag
-
-        with (
-            trace_call(
-                "CloudSpanner.Session.run_in_transaction",
-                self,
-                extra_attributes=extra_attributes,
-                observability_options=getattr(database, "observability_options", None),
-            ) as span,
-            MetricsCapture(self._resource_info),
-        ):
-            attempts: int = 0
-
-            # If a transaction using a multiplexed session is retried after an aborted
-            # user operation, it should include the previous transaction ID in the
-            # transaction options used to begin the transaction. This allows the backend
-            # to recognize the transaction and increase the lock order for the new
-            # transaction that is created.
-            # See :attr:`~google.cloud.spanner_v1.types.TransactionOptions.ReadWrite.multiplexed_session_previous_transaction_id`
-            previous_transaction_id: Optional[bytes] = None
-
-            while True:
-                txn = self.transaction(client_context=client_context)
-                txn.transaction_tag = transaction_tag
-                txn.exclude_txn_from_change_streams = exclude_txn_from_change_streams
-                txn.isolation_level = isolation_level
-                txn.read_lock_mode = read_lock_mode
-
-                if self.is_multiplexed:
-                    txn._multiplexed_session_previous_transaction_id = (
-                        previous_transaction_id
-                    )
+        span = get_current_span()
+        attempts: int = 0
+
+        # If a transaction using a multiplexed session is retried after an aborted
+        # user operation, it should include the previous transaction ID in the
+        # transaction options used to begin the transaction. This allows the backend
+        # to recognize the transaction and increase the lock order for the new
+        # transaction that is created.
+        # See :attr:`~google.cloud.spanner_v1.types.TransactionOptions.ReadWrite.multiplexed_session_previous_transaction_id`
+        previous_transaction_id: Optional[bytes] = None
+
+        while True:
+            transaction = self._create_transaction_for_attempt(
+                client_context=client_context,
+                transaction_tag=transaction_tag,
+                exclude_txn_from_change_streams=exclude_txn_from_change_streams,
+                isolation_level=isolation_level,
+                read_lock_mode=read_lock_mode,
+                previous_transaction_id=previous_transaction_id,
+            )
+            attempts += 1
 
-                attempts += 1
-                span_attributes = dict(attempt=attempts)
+            try:
+                return_value = await CrossSync.run_if_async(
+                    func, transaction, *args, **kw
+                )
 
-                try:
-                    return_value = await CrossSync.run_if_async(func, txn, *args, **kw)
-
-                except Aborted as exc:
-                    previous_transaction_id = txn._transaction_id
-                    delay_seconds = _get_retry_delay(
-                        exc.errors[0],
-                        attempts,
-                        default_retry_delay=default_retry_delay,
-                    )
-                    attributes = dict(delay_seconds=delay_seconds, cause=str(exc))
-                    attributes.update(span_attributes)
-                    add_span_event(
-                        span,
-                        "Transaction was aborted in user operation, retrying",
-                        attributes,
-                    )
-                    await _delay_until_retry(
-                        exc,
-                        deadline,
-                        attempts,
-                        default_retry_delay=default_retry_delay,
-                    )
-                    continue
+            except Aborted as exc:
+                previous_transaction_id = transaction._transaction_id
+                await self._handle_aborted(
+                    exc,
+                    span,
+                    "Transaction was aborted in user operation, retrying",
+                    attempts,
+                    deadline,
+                    default_retry_delay,
+                    include_cause=True,
+                )
+                continue
 
-                except GoogleAPICallError:
-                    add_span_event(
-                        span,
-                        "User operation failed due to GoogleAPICallError, not retrying",
-                        span_attributes,
-                    )
-                    raise
+            except GoogleAPICallError:
+                add_span_event(
+                    span,
+                    "User operation failed due to GoogleAPICallError, not retrying",
+                    {"attempt": attempts},
+                )
+                raise
 
-                except Exception:
-                    add_span_event(
-                        span,
-                        "User operation failed. Invoking Transaction.rollback(), not retrying",
-                        span_attributes,
-                    )
-                    await txn.rollback()
-                    raise
+            except Exception:
+                add_span_event(
+                    span,
+                    "User operation failed. Invoking Transaction.rollback(), not retrying",
+                    {"attempt": attempts},
+                )
+                await transaction.rollback()
+                raise
+
+            try:
+                await transaction.commit(
+                    return_commit_stats=log_commit_stats,
+                    request_options=commit_request_options,
+                    max_commit_delay=max_commit_delay,
+                )
 
-                try:
-                    await txn.commit(
-                        return_commit_stats=log_commit_stats,
-                        request_options=commit_request_options,
-                        max_commit_delay=max_commit_delay,
-                    )
+            except Aborted as exc:
+                previous_transaction_id = transaction._transaction_id
+                await self._handle_aborted(
+                    exc,
+                    span,
+                    "Transaction was aborted during commit, retrying",
+                    attempts,
+                    deadline,
+                    default_retry_delay,
+                )
+                continue
 
-                except Aborted as exc:
-                    previous_transaction_id = txn._transaction_id
-                    delay_seconds = _get_retry_delay(
-                        exc.errors[0],
-                        attempts,
-                        default_retry_delay=default_retry_delay,
-                    )
-                    attributes = dict(delay_seconds=delay_seconds)
-                    attributes.update(span_attributes)
-                    add_span_event(
-                        span,
-                        "Transaction was aborted during commit, retrying",
-                        attributes,
-                    )
-                    await _delay_until_retry(
-                        exc,
-                        deadline,
-                        attempts,
-                        default_retry_delay=default_retry_delay,
-                    )
+            except GoogleAPICallError:
+                add_span_event(
+                    span,
+                    "Transaction.commit failed due to GoogleAPICallError, not retrying",
+                    {"attempt": attempts},
+                )
+                raise
 
-                except GoogleAPICallError:
-                    add_span_event(
-                        span,
-                        "Transaction.commit failed due to GoogleAPICallError, not retrying",
-                        span_attributes,
+            else:
+                if log_commit_stats and transaction.commit_stats:
+                    database.logger.info(
+                        "CommitStats: {}".format(transaction.commit_stats),
+                        extra={"commit_stats": transaction.commit_stats},
                     )
-                    raise
-
-                else:
-                    if log_commit_stats and txn.commit_stats:
-                        database.logger.info(
-                            "CommitStats: {}".format(txn.commit_stats),
-                            extra={"commit_stats": txn.commit_stats},
-                        )
-                    return return_value
+                return return_value

packages/google-cloud-spanner/google/cloud/spanner_v1/_helpers.py

@@ -856,7 +856,7 @@ def _delay_until_retry(exc, deadline, attempts, default_retry_delay=None):
     :param attempts: number of call retries
     """
 
-    cause = exc.errors[0]
+    cause = exc.errors[0] if getattr(exc, "errors", None) else exc
     now = time.time()
     if now >= deadline:
         raise