apache/beam · error · RuntimeError

Invalid tag %r

Error message

Invalid tag %r

What it means

Side input tags follow SIDE_INPUT_REGEX (typically 'ref_PCollection_side_N'). get_sideinput_index extracts the integer index N; a tag that does not match means the runner_api side-input reference is malformed. It indicates corrupted or hand-constructed pipeline proto data.

Solutions

1. Inspect the pipeline proto/JSON and correct the side input tag to the expected 'ref_PCollection_side_<N>' form.
2. Regenerate the pipeline instead of manually editing serialized pipeline artifacts.
3. Check Beam SDK/runner version consistency if the malformed tag comes from a portable pipeline crossing SDK/container versions.

Example fix

// before
tag = 'ref_PCollection_3'  # not a side-input tag
idx = get_sideinput_index(tag)
// after
tag = 'ref_PCollection_side_3'
idx = get_sideinput_index(tag)

import re
SIDE_INPUT_REGEX = re.compile(r'ref_PCollection_side_(\d+)', re.DOTALL)
assert SIDE_INPUT_REGEX.match(tag), f"tag {tag!r} is not a side-input reference"

def is_side_input_tag(tag: str) -> bool:
    import re
    return isinstance(tag, str) and re.match(r'ref_PCollection_side_\d+$', tag, re.DOTALL) is not None

try:
    idx = get_sideinput_index(tag)
except RuntimeError as e:
    log.error('Malformed side-input tag %r; regenerate the pipeline proto', tag)
    raise

When it happens

Trigger: from_runner_api/deserialize receives a side-input tag string that fails SIDE_INPUT_REGEX (e.g. missing '_side_' portion, arbitrary key from user code); called during pipeline construction from from_runner_api_parameter.

Common situations: Hand-editing or transforming a serialized pipeline JSON/proto; passing runner-internal tag keys where an index-bearing tag is expected; runner/portable layers producing nonstandard tags after version mismatch.

Understand the failure class

Background: "invalid id" errors: invalid identifier format — why libraries reject IDs before lookup, and how to fix them — this error's family across 37 libraries.

Related errors

• No producer for
• Only one of context or default_environment may be specified.
• Sessions is not allowed in side inputs
• Side inputs must have defaults for FlatMapTuple.
• Side inputs must have defaults for MapTuple.

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/sideinputs.py:74

  if isinstance(target_window_fn, window.Sessions):
    raise RuntimeError("Sessions is not allowed in side inputs")

  def map_via_end(source_window: window.BoundedWindow) -> window.BoundedWindow:
    return list(
        target_window_fn.assign(
            window.WindowFn.AssignContext(
                source_window.max_timestamp(), window=source_window)))[-1]

  return map_via_end


def get_sideinput_index(tag: str) -> int:
  match = re.match(SIDE_INPUT_REGEX, tag, re.DOTALL)
  if match:
    return int(match.group(1))
  else:
    raise RuntimeError("Invalid tag %r" % tag)


class SideInputMap(object):
  """Represents a mapping of windows to side input values."""
  def __init__(self, view_class: 'pvalue.AsSideInput', view_options, iterable):
    self._window_mapping_fn = view_options.get(
        'window_mapping_fn', _global_window_mapping_fn)
    self._view_class = view_class
    self._view_options = view_options
    self._iterable = iterable
    self._cache: dict[window.BoundedWindow, Any] = {}

  def __getitem__(self, window: window.BoundedWindow) -> Any:
    if window not in self._cache:
      target_window = self._window_mapping_fn(window)
      self._cache[window] = self._view_class._from_runtime_iterable(
          _FilteringIterable(self._iterable, target_window), self._view_options)
    return self._cache[window]

View on GitHub (pinned to 12126d8942)