apache/beam · error · ValueError

Cannot insert different protos %r and %r with the same ID %r

Error message

Cannot insert different protos %r and %r with the same ID %r

What it means

Raised by PipelineContext.put_proto when two different proto objects are registered under the same ID in the pipeline context while ignore_duplicates=True. The context maps component IDs (PCollection, Transform, Coder proto IDs) to their proto messages; silently overwriting a differing proto would corrupt the pipeline model, so the mismatch is rejected. It is an internal consistency guard for the Beam pipeline proto builder.

Solutions

Ensure every proto registered gets a unique ID (use context' id-generating helpers such as ptransform.label uniqueness or context.put_proto with fresh IDs)
Check for reuse of PipelineContext objects or proto messages across pipelines; create a fresh context per pipeline
If a duplicate proto is truly identical, allow it: pass ignore_duplicates=True only where protos are equal, or reuse the existing ID from get_by_proto
Update Beam version — older versions had ID-generation collisions; regenerate your pipeline with a current SDK

Example fix

// before
ctx = PipelineContext(default_environment=None)
id = ctx.put_proto(same_id, proto_a, ignore_duplicates=True)
id = ctx.put_proto(same_id, proto_b, ignore_duplicates=True)  # raises
// after
id_a = ctx.put_proto(proto_a)
id_b = ctx.put_proto(proto_b)  # let the context mint distinct IDs
assert id_a != id_b

Defensive patterns

Strategy: try-catch

Validate before calling

existing = ctx._id_to_proto.get(id)
if existing is not None and existing != proto:
    raise ValueError(f'ID {id!r} already bound to a different proto')

Type guard

def is_id_free(ctx, id: str) -> bool:
    return id not in ctx._id_to_proto

Try / catch

try:
    ctx.put_proto(proto, ignore_duplicates=True)
except ValueError as e:
    if 'Cannot insert different protos' in str(e):
        existing = ctx.get_by_id(id)
        id = ctx.put_proto(proto)  # mint a fresh unique ID
    else:
        raise

Prevention

Never assign component ID strings manually; let PipelineContext generate them
Use get_by_proto to reuse the existing ID for an identical proto instead of re-inserting
Create a fresh PipelineContext per pipeline; never share one across pipelines
Hash-compare protos before reuse to confirm they are truly equal

When it happens

Trigger: Calling get_by_proto (or any code path that calls put_proto with ignore_duplicates=True) when an ID string is reused for a semantically different proto message — e.g. two distinct pcollections/transforms generated the same ID string, or a proto object's id() key hashing collided/reused after object reuse.

Common situations: Custom pipeline construction or runner code that manually assigns component IDs; reusing a PipelineContext across pipelines; frameworks generating PCollection IDs from non-unique labels; errors where a None/uniqueness check was skipped before generating IDs.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/pipeline_context.py:146

            obj=obj, obj_type=self._obj_type, label=label),
        maybe_new_proto)

  def get_id_to_proto_map(self) -> dict[str, message.Message]:
    return self._id_to_proto

  def get_proto_from_id(self, id: str) -> message.Message:
    return self.get_id_to_proto_map()[id]

  def put_proto(
      self,
      id: str,
      proto: message.Message,
      ignore_duplicates: bool = False) -> str:
    if not ignore_duplicates and id in self._id_to_proto:
      raise ValueError("Id '%s' is already taken." % id)
    elif (ignore_duplicates and id in self._id_to_proto and
          self._id_to_proto[id] != proto):
      raise ValueError(
          'Cannot insert different protos %r and %r with the same ID %r',
          self._id_to_proto[id],
          proto,
          id)
    self._id_to_proto[id] = proto
    return id

  def __getitem__(self, id: str) -> Any:
    return self.get_by_id(id)

  def __contains__(self, id: str) -> bool:
    return id in self._id_to_proto


class PipelineContext(object):
  """For internal use only; no backwards-compatibility guarantees.

  Used for accessing and constructing the referenced objects of a Pipeline.

View on GitHub (pinned to 12126d8942)