feat: default enable multiplex session for all operations unless explicitly set to false

.github/workflows/integration-tests-against-emulator-with-multiplexed-session.yaml → .github/workflows/integration-tests-against-emulator-with-regular-session.yaml

@@ -3,7 +3,7 @@ on:
     branches:
       - main
   pull_request:
-name: Run Spanner integration tests against emulator with multiplexed sessions
+name: Run Spanner integration tests against emulator with regular sessions
 jobs:
   system-tests:
     runs-on: ubuntu-latest
@@ -21,7 +21,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
-          python-version: 3.8
+          python-version: 3.12
       - name: Install nox
         run: python -m pip install nox
       - name: Run system tests
@@ -30,5 +30,6 @@ jobs:
           SPANNER_EMULATOR_HOST: localhost:9010
           GOOGLE_CLOUD_PROJECT: emulator-test-project
           GOOGLE_CLOUD_TESTS_CREATE_SPANNER_INSTANCE: true
-          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: true
-          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS: true
+          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: false
+          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS: false
+          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_FOR_RW: false

.github/workflows/integration-tests-against-emulator.yaml

@@ -10,7 +10,7 @@ jobs:
 
     services:
       emulator:
-        image: gcr.io/cloud-spanner-emulator/emulator:latest
+        image: gcr.io/cloud-spanner-emulator/emulator:1.5.37
         ports:
           - 9010:9010
           - 9020:9020
@@ -21,7 +21,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
-          python-version: 3.8
+          python-version: 3.12
       - name: Install nox
         run: python -m pip install nox
       - name: Run system tests

.kokoro/presubmit/integration-multiplexed-sessions-enabled.cfg → .kokoro/presubmit/integration-regular-sessions-enabled.cfg

@@ -8,10 +8,15 @@ env_vars: {
 
 env_vars: {
     key: "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"
-    value: "true"
+    value: "false"
 }
 
 env_vars: {
     key: "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS"
-    value: "true"
+    value: "false"
+}
+
+env_vars: {
+    key: "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_FOR_RW"
+    value: "false"
 }

google/cloud/spanner_v1/database.py

@@ -848,7 +848,14 @@ def session(self, labels=None, database_role=None):
         # If role is specified in param, then that role is used
         # instead.
         role = database_role or self._database_role
-        return Session(self, labels=labels, database_role=role)
+        is_multiplexed = False
+        if self.sessions_manager._use_multiplexed(
+            transaction_type=TransactionType.READ_ONLY
+        ):
+            is_multiplexed = True
+        return Session(
+            self, labels=labels, database_role=role, is_multiplexed=is_multiplexed
+        )
 
     def snapshot(self, **kw):
         """Return an object which wraps a snapshot.

google/cloud/spanner_v1/database_sessions_manager.py

@@ -230,15 +230,13 @@ def _use_multiplexed(cls, transaction_type: TransactionType) -> bool:
         """Returns whether to use multiplexed sessions for the given transaction type.
 
         Multiplexed sessions are enabled for read-only transactions if:
-            * _ENV_VAR_MULTIPLEXED is set to true.
+            * _ENV_VAR_MULTIPLEXED != 'false'.
 
         Multiplexed sessions are enabled for partitioned transactions if:
-            * _ENV_VAR_MULTIPLEXED is set to true; and
-            * _ENV_VAR_MULTIPLEXED_PARTITIONED is set to true.
+            * _ENV_VAR_MULTIPLEXED_PARTITIONED != 'false'.
 
         Multiplexed sessions are enabled for read/write transactions if:
-            * _ENV_VAR_MULTIPLEXED is set to true; and
-            * _ENV_VAR_MULTIPLEXED_READ_WRITE is set to true.
+            * _ENV_VAR_MULTIPLEXED_READ_WRITE != 'false'.
 
         :type transaction_type: :class:`TransactionType`
         :param transaction_type: the type of transaction
@@ -254,30 +252,26 @@ def _use_multiplexed(cls, transaction_type: TransactionType) -> bool:
             return cls._getenv(cls._ENV_VAR_MULTIPLEXED)
 
         elif transaction_type is TransactionType.PARTITIONED:
-            return cls._getenv(cls._ENV_VAR_MULTIPLEXED) and cls._getenv(
-                cls._ENV_VAR_MULTIPLEXED_PARTITIONED
-            )
+            return cls._getenv(cls._ENV_VAR_MULTIPLEXED_PARTITIONED)
 
         elif transaction_type is TransactionType.READ_WRITE:
-            return cls._getenv(cls._ENV_VAR_MULTIPLEXED) and cls._getenv(
-                cls._ENV_VAR_MULTIPLEXED_READ_WRITE
-            )
+            return cls._getenv(cls._ENV_VAR_MULTIPLEXED_READ_WRITE)
 
         raise ValueError(f"Transaction type {transaction_type} is not supported.")
 
     @classmethod
     def _getenv(cls, env_var_name: str) -> bool:
         """Returns the value of the given environment variable as a boolean.
 
-        True values are '1' and 'true' (case-insensitive).
-        All other values are considered false.
+        True unless explicitly 'false' (case-insensitive).
+        All other values (including unset) are considered true.
 
         :type env_var_name: str
         :param env_var_name: the name of the boolean environment variable
 
         :rtype: bool
-        :returns: True if the environment variable is set to a true value, False otherwise.
+        :returns: True unless the environment variable is set to 'false', False otherwise.
         """
 
-        env_var_value = getenv(env_var_name, "").lower().strip()
-        return env_var_value in ["1", "true"]
+        env_var_value = getenv(env_var_name, "true").lower().strip()
+        return env_var_value != "false"

google/cloud/spanner_v1/session.py

@@ -275,7 +275,13 @@ def delete(self):
                 current_span, "Deleting Session failed due to unset session_id"
             )
             raise ValueError("Session ID not set by back-end")
-
+        if self._is_multiplexed:
+            add_span_event(
+                current_span,
+                "Skipped deleting Multiplexed Session",
+                {"session.id": self._session_id},
+            )
+            return
         add_span_event(
             current_span, "Deleting Session", {"session.id": self._session_id}
         )

google/cloud/spanner_v1/snapshot.py

@@ -133,6 +133,8 @@ def _restart_on_unavailable(
                 # Update the transaction from the response.
                 if transaction is not None:
                     transaction._update_for_result_set_pb(item)
+                if item.precommit_token is not None and transaction is not None:
+                    transaction._update_for_precommit_token_pb(item.precommit_token)
 
                 if item.resume_token:
                     resume_token = item.resume_token
@@ -1013,9 +1015,6 @@ def _update_for_result_set_pb(
         if result_set_pb.metadata and result_set_pb.metadata.transaction:
             self._update_for_transaction_pb(result_set_pb.metadata.transaction)
 
-        if result_set_pb.precommit_token:
-            self._update_for_precommit_token_pb(result_set_pb.precommit_token)
-
     def _update_for_transaction_pb(self, transaction_pb: Transaction) -> None:
         """Updates the snapshot for the given transaction.
 
@@ -1031,7 +1030,7 @@ def _update_for_transaction_pb(self, transaction_pb: Transaction) -> None:
             self._transaction_id = transaction_pb.id
 
         if transaction_pb.precommit_token:
-            self._update_for_precommit_token_pb(transaction_pb.precommit_token)
+            self._update_for_precommit_token_pb_unsafe(transaction_pb.precommit_token)
 
     def _update_for_precommit_token_pb(
         self, precommit_token_pb: MultiplexedSessionPrecommitToken
@@ -1044,10 +1043,22 @@ def _update_for_precommit_token_pb(
         # Because multiple threads can be used to perform operations within a
         # transaction, we need to use a lock when updating the precommit token.
         with self._lock:
-            if self._precommit_token is None or (
-                precommit_token_pb.seq_num > self._precommit_token.seq_num
-            ):
-                self._precommit_token = precommit_token_pb
+            self._update_for_precommit_token_pb_unsafe(precommit_token_pb)
+
+    def _update_for_precommit_token_pb_unsafe(
+        self, precommit_token_pb: MultiplexedSessionPrecommitToken
+    ) -> None:
+        """Updates the snapshot for the given multiplexed session precommit token.
+        This method is unsafe because it does not acquire a lock before updating
+        the precommit token. It should only be used when the caller has already
+        acquired the lock.
+        :type precommit_token_pb: :class:`~google.cloud.spanner_v1.MultiplexedSessionPrecommitToken`
+        :param precommit_token_pb: The multiplexed session precommit token to update the snapshot with.
+        """
+        if self._precommit_token is None or (
+            precommit_token_pb.seq_num > self._precommit_token.seq_num
+        ):
+            self._precommit_token = precommit_token_pb
 
 
 class Snapshot(_SnapshotBase):

google/cloud/spanner_v1/transaction.py

@@ -285,13 +285,18 @@ def commit(
 
             def wrapped_method(*args, **kwargs):
                 attempt.increment()
+                commit_request_args = {
+                    "mutations": mutations,
+                    **common_commit_request_args,
+                }
+                # Check if session is multiplexed (safely handle mock sessions)
+                is_multiplexed = getattr(self._session, "is_multiplexed", False)
+                if is_multiplexed and self._precommit_token is not None:
+                    commit_request_args["precommit_token"] = self._precommit_token
+
                 commit_method = functools.partial(
                     api.commit,
-                    request=CommitRequest(
-                        mutations=mutations,
-                        precommit_token=self._precommit_token,
-                        **common_commit_request_args,
-                    ),
+                    request=CommitRequest(**commit_request_args),
                     metadata=database.metadata_with_request_id(
                         nth_request,
                         attempt.value,
@@ -516,6 +521,9 @@ def wrapped_method(*args, **kwargs):
         if is_inline_begin:
             self._lock.release()
 
+        if result_set_pb.precommit_token is not None:
+            self._update_for_precommit_token_pb(result_set_pb.precommit_token)
+
         return result_set_pb.stats.row_count_exact
 
     def batch_update(
@@ -660,6 +668,14 @@ def wrapped_method(*args, **kwargs):
         if is_inline_begin:
             self._lock.release()
 
+        if (
+            len(response_pb.result_sets) > 0
+            and response_pb.result_sets[0].precommit_token
+        ):
+            self._update_for_precommit_token_pb(
+                response_pb.result_sets[0].precommit_token
+            )
+
         row_counts = [
             result_set.stats.row_count_exact for result_set in response_pb.result_sets
         ]
@@ -736,9 +752,6 @@ def _update_for_execute_batch_dml_response_pb(
         :type response_pb: :class:`~google.cloud.spanner_v1.types.ExecuteBatchDmlResponse`
         :param response_pb: The execute batch DML response to update the transaction with.
         """
-        if response_pb.precommit_token:
-            self._update_for_precommit_token_pb(response_pb.precommit_token)
-
         # Only the first result set contains the result set metadata.
         if len(response_pb.result_sets) > 0:
             self._update_for_result_set_pb(response_pb.result_sets[0])

tests/_helpers.py

@@ -43,7 +43,7 @@ def is_multiplexed_enabled(transaction_type: TransactionType) -> bool:
     env_var_read_write = "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_FOR_RW"
 
     def _getenv(val: str) -> bool:
-        return getenv(val, "false").lower() == "true"
+        return getenv(val, "true").lower().strip() != "false"
 
     if transaction_type is TransactionType.READ_ONLY:
         return _getenv(env_var)

tests/mockserver_tests/mock_server_test_base.py

@@ -41,6 +41,7 @@
     SpannerServicer,
     start_mock_server,
 )
+from tests._helpers import is_multiplexed_enabled
 
 
 # Creates an aborted status with the smallest possible retry delay.
@@ -228,3 +229,109 @@ def database(self) -> Database:
                 enable_interceptors_in_tests=True,
             )
         return self._database
+
+    def assert_requests_sequence(
+        self,
+        requests,
+        expected_types,
+        transaction_type,
+        allow_multiple_batch_create=True,
+    ):
+        """Assert that the requests sequence matches the expected types, accounting for multiplexed sessions and retries.
+
+        Args:
+            requests: List of requests from spanner_service.requests
+            expected_types: List of expected request types (excluding session creation requests)
+            transaction_type: TransactionType enum value to check multiplexed session status
+            allow_multiple_batch_create: If True, skip all leading BatchCreateSessionsRequest and one optional CreateSessionRequest
+        """
+        from google.cloud.spanner_v1 import (
+            BatchCreateSessionsRequest,
+            CreateSessionRequest,
+        )
+
+        mux_enabled = is_multiplexed_enabled(transaction_type)
+        idx = 0
+        # Skip all leading BatchCreateSessionsRequest (for retries)
+        if allow_multiple_batch_create:
+            while idx < len(requests) and isinstance(
+                requests[idx], BatchCreateSessionsRequest
+            ):
+                idx += 1
+            # For multiplexed, optionally skip a CreateSessionRequest
+            if (
+                mux_enabled
+                and idx < len(requests)
+                and isinstance(requests[idx], CreateSessionRequest)
+            ):
+                idx += 1
+        else:
+            if mux_enabled:
+                self.assertTrue(
+                    isinstance(requests[idx], BatchCreateSessionsRequest),
+                    f"Expected BatchCreateSessionsRequest at index {idx}, got {type(requests[idx])}",
+                )
+                idx += 1
+                self.assertTrue(
+                    isinstance(requests[idx], CreateSessionRequest),
+                    f"Expected CreateSessionRequest at index {idx}, got {type(requests[idx])}",
+                )
+                idx += 1
+            else:
+                self.assertTrue(
+                    isinstance(requests[idx], BatchCreateSessionsRequest),
+                    f"Expected BatchCreateSessionsRequest at index {idx}, got {type(requests[idx])}",
+                )
+                idx += 1
+        # Check the rest of the expected request types
+        for expected_type in expected_types:
+            self.assertTrue(
+                isinstance(requests[idx], expected_type),
+                f"Expected {expected_type} at index {idx}, got {type(requests[idx])}",
+            )
+            idx += 1
+        self.assertEqual(
+            idx, len(requests), f"Expected {idx} requests, got {len(requests)}"
+        )
+
+    def adjust_request_id_sequence(self, expected_segments, requests, transaction_type):
+        """Adjust expected request ID sequence numbers based on actual session creation requests.
+
+        Args:
+            expected_segments: List of expected (method, (sequence_numbers)) tuples
+            requests: List of actual requests from spanner_service.requests
+            transaction_type: TransactionType enum value to check multiplexed session status
+
+        Returns:
+            List of adjusted expected segments with corrected sequence numbers
+        """
+        from google.cloud.spanner_v1 import (
+            BatchCreateSessionsRequest,
+            CreateSessionRequest,
+            ExecuteSqlRequest,
+            BeginTransactionRequest,
+        )
+
+        # Count session creation requests that come before the first non-session request
+        session_requests_before = 0
+        for req in requests:
+            if isinstance(req, (BatchCreateSessionsRequest, CreateSessionRequest)):
+                session_requests_before += 1
+            elif isinstance(req, (ExecuteSqlRequest, BeginTransactionRequest)):
+                break
+
+        # For multiplexed sessions, we expect 2 session requests (BatchCreateSessions + CreateSession)
+        # For non-multiplexed, we expect 1 session request (BatchCreateSessions)
+        mux_enabled = is_multiplexed_enabled(transaction_type)
+        expected_session_requests = 2 if mux_enabled else 1
+        extra_session_requests = session_requests_before - expected_session_requests
+
+        # Adjust sequence numbers based on extra session requests
+        adjusted_segments = []
+        for method, seq_nums in expected_segments:
+            # Adjust the sequence number (5th element in the tuple)
+            adjusted_seq_nums = list(seq_nums)
+            adjusted_seq_nums[4] += extra_session_requests
+            adjusted_segments.append((method, tuple(adjusted_seq_nums)))
+
+        return adjusted_segments