apache/beam · error · ValueError

Invalid transaction object

Error message

Invalid transaction object: %s. It should be instance of SPANNER_TRANSACTION object created by spannerio.create_transaction transform.

What it means

Beam's SpannerIO write transform expects the transaction side-input to be the internal _SPANNER_TRANSACTION wrapper produced by the spannerio.create_transaction transform. process() raises ValueError when the second element is anything else, because transaction_info is extracted from that wrapper's .transaction field.

Solutions

Add spannerio.create_transaction(database_id) to the pipeline and use its PCollection output as the transaction side input to the write transform.
If you have a raw Transaction, you cannot pass it directly - route writes through create_transaction so Beam manages transaction lifecycle.
Check the write pipeline ordering: write_bundles/process receives (mutation, transaction) pairs, so zip your mutations with the create_transaction output.

Example fix

// before
rows | 'write' >> spannerio.Write().with_transaction(raw_transaction)
// after
transactions = beam.pvalue.AsSingleton(
    pcoll | 'txn' >> spannerio.create_transaction(database_id))
rows | 'write' >> spannerio.Write(database_id).with_transaction(transactions)

Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.io.gcp.experimental.spannerio import _SPANNER_TRANSACTION
assert isinstance(txn, _SPANNER_TRANSACTION), 'use spannerio.create_transaction output'

Type guard

def is_spanner_transaction(obj):
    from apache_beam.io.gcp.experimental.spannerio import _SPANNER_TRANSACTION
    return isinstance(obj, _SPANNER_TRANSACTION)

Try / catch

try:
    _ | write_transform
except ValueError as e:
    if 'Invalid transaction object' in str(e):
        raise RuntimeError('Wire the CreateTransaction transform output as side input') from e
    raise

Prevention

Always insert spannerio.create_transaction into the write pipeline
Never pass client-SDK Transaction objects into Beam transforms
Zip mutations with the create_transaction PCollection as singleton side input

When it happens

Trigger: Passing a raw google.cloud.spanner.Transaction, a spanner Database object, or None as the side input to the write (or spannerio.Write / write mutations) PTransform instead of the output of create_transaction.

Common situations: Wiring a PCollection to the write transform without the CreateTransaction step in the pipeline; refactoring code and substituting a client-created transaction for the Beam wrapper.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/39cd11804966fe0a. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/io/gcp/experimental/spannerio.py:402

    return self._session

  def _close_session(self):
    if self._session is not None:
      self._session.delete()

  def setup(self):
    # setting up client to connect with cloud spanner
    spanner_client = Client(self._spanner_configuration.project)
    instance = spanner_client.instance(self._spanner_configuration.instance)
    self._database = instance.database(
        self._spanner_configuration.database,
        pool=self._spanner_configuration.pool)

  def process(self, element, spanner_transaction):
    # `spanner_transaction` should be the instance of the _SPANNER_TRANSACTION
    # object.
    if not isinstance(spanner_transaction, _SPANNER_TRANSACTION):
      raise ValueError(
          "Invalid transaction object: %s. It should be instance "
          "of SPANNER_TRANSACTION object created by "
          "spannerio.create_transaction transform." % type(spanner_transaction))

    transaction_info = spanner_transaction.transaction

    # We used batch snapshot to reuse the same transaction passed through the
    # side input
    self._snapshot = BatchSnapshot.from_dict(self._database, transaction_info)

    # getting the transaction from the snapshot's session to run read operation.
    # with self._snapshot.session().transaction() as transaction:
    with self._get_session().transaction() as transaction:
      table_id = self._spanner_configuration.table
      query_name = self._spanner_configuration.query_name or ''

      if element.is_sql is True:
        transaction_read = transaction.execute_sql

View on GitHub (pinned to 12126d8942)