apache/beam · error · ValueError

Read operation in the constructor only works with the root…

Error message

Read operation in the constructor only works with the root of the pipeline.

What it means

SpannerIO's Read ReadFromSpanner transform supports pre-built ReadOperation objects only when the transform is the root of the pipeline (input is PBegin). If the transform receives a real PCollection as input while read operations were also supplied in the constructor, the library raises this ValueError because both sources of read configuration conflict.

Solutions

Do not pass read_operations (or sql/table) to ReadFromSpanner when piping from an existing PCollection; use SpannerIO.Read with an input PCollection of ReadOperation objects instead.
Apply the transform to the pipeline root: p | ReadFromSpanner(project=..., instance=..., database=..., read_operations=[...]).
If you need per-element reads, build ReadOperation objects upstream and pipe them into SpannerRead with an explicit input PCollection.

Example fix

# before
pcoll | ReadFromSpanner(project='p', instance='i', database='d', read_operations=[op])
# after
p | ReadFromSpanner(project='p', instance='i', database='d', read_operations=[op])

Defensive patterns

Strategy: validation

Validate before calling

def use_read_ops_at_root(read_operations, has_input_pcoll):
    if read_operations and has_input_pcoll:
        raise ValueError('read_operations only allowed when ReadFromSpanner is applied to pipeline root (PBegin)')
    return True

Type guard

def is_pipeline_root(pbegin_or_coll):
    from apache_beam import PBegin
    return isinstance(pbegin_or_coll, PBegin)

Prevention

Never pass read_operations/sql/table to ReadFromSpanner unless applied directly to the Pipeline object
Use SpannerRead with a PCollection of ReadOperation for non-root reads
Pin a check in code review when refactoring reads out of the root

When it happens

Trigger: Calling beam.Pipeline | ReadFromSpanner(..., read_operations=[...]) applied to an existing PCollection instead of pbegin, e.g. piping ReadFromSpanner after another transform while also passing read_operations/sql/table to the constructor.

Common situations: Developers chain ReadFromSpanner into an existing pipeline after another step (e.g. reading side inputs) but forget that constructor-supplied read_operations only work at pipeline root; often after refactoring a standalone read into a larger pipeline.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

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

      if table is not None:
        if columns is None:
          raise ValueError("Columns are required with the table name.")
        self._read_operations = [
            ReadOperation.table(
                table=table, columns=columns, index=index, keyset=keyset)
        ]
      elif sql is not None:
        self._read_operations = [
            ReadOperation.query(
                sql=sql, params=params, param_types=param_types)
        ]

  def expand(self, pbegin):
    if self._read_operations is not None and isinstance(pbegin, PBegin):
      pcoll = pbegin.pipeline | Create(self._read_operations)
    elif not isinstance(pbegin, PBegin):
      if self._read_operations is not None:
        raise ValueError(
            "Read operation in the constructor only works with "
            "the root of the pipeline.")
      pcoll = pbegin
    else:
      raise ValueError(
          "Spanner required read operation, sql or table "
          "with columns.")

    if self._transaction is None:
      # reading as batch read using the spanner partitioning query to create
      # batches.
      p = (
          pcoll
          | 'Generate Partitions' >> ParDo(
              _CreateReadPartitions(spanner_configuration=self._configuration))
          | 'Reshuffle' >> Reshuffle()
          | 'Read From Partitions' >> ParDo(
              _ReadFromPartitionFn(spanner_configuration=self._configuration)))

View on GitHub (pinned to 12126d8942)