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

@@ -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

@@ -516,6 +516,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 +663,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 +747,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