apache/beam · error · ValueError

Invalid tag.

Error message

Invalid tag.

What it means

MergingTriggerState.get_state falls through to ValueError('Invalid tag.', tag) when the tag is none of the recognized mergeable types (_CombiningValueStateTag, _ListStateTag, _SetStateTag, _WatermarkHoldStateTag, and not an RMW tag caught earlier). It guards against unknown or corrupted tag objects reaching the merge path.

Source

Thrown at sdks/python/apache_beam/transforms/trigger.py:1164

      original_tag, tag = tag, tag.without_extraction()
    values = [
        self.raw_state.get_state(window_id, tag)
        for window_id in self._get_ids(window)
    ]
    if isinstance(tag, _ReadModifyWriteStateTag):
      raise ValueError(
          'Merging requested for non-mergeable state tag: %r.' % tag)
    elif isinstance(tag, _CombiningValueStateTag):
      return original_tag.combine_fn.extract_output(
          original_tag.combine_fn.merge_accumulators(values))
    elif isinstance(tag, _ListStateTag):
      return [v for vs in values for v in vs]
    elif isinstance(tag, _SetStateTag):
      return {v for vs in values for v in vs}
    elif isinstance(tag, _WatermarkHoldStateTag):
      return tag.timestamp_combiner_impl.combine_all(values)
    else:
      raise ValueError('Invalid tag.', tag)

  def clear_state(self, window, tag):
    for window_id in self._get_ids(window):
      self.raw_state.clear_state(window_id, tag)
    if tag is None:
      del self.window_ids[window]
      self._persist_window_ids()

  def merge(self, to_be_merged, merge_result):
    for window in to_be_merged:
      if window != merge_result:
        if window in self.window_ids:
          if merge_result in self.window_ids:
            merge_window_ids = self.window_ids[merge_result]
          else:
            merge_window_ids = self.window_ids[merge_result] = []
          merge_window_ids.extend(self.window_ids.pop(window))
          self._persist_window_ids()

View on GitHub (pinned to 12126d8942)

Solutions

1. Use one of the supported tag types: _CombiningValueStateTag, _ListStateTag, _SetStateTag, or _WatermarkHoldStateTag.
2. Inspect the tag passed in (it appears in the exception args) and fix its construction site.
3. Check for Beam version drift: a tag class that was supported may no longer be; update the tag to the current API.

Example fix

// before
class MyTag: pass
state.get_state(window, MyTag())  # ValueError: Invalid tag
// after
tag = _ListStateTag('items')
state.get_state(window, tag)

from apache_beam.transforms import trigger as t
if not isinstance(tag, (t._CombiningValueStateTag, t._ListStateTag, t._SetStateTag, t._WatermarkHoldStateTag)):
    raise TypeError(f'Unsupported state tag type: {type(tag)}')

def is_supported_merge_tag(tag):
    from apache_beam.transforms import trigger as t
    return isinstance(tag, (t._CombiningValueStateTag, t._ListStateTag, t._SetStateTag, t._WatermarkHoldStateTag))

try:
    value = state.get_state(window, tag)
except ValueError as e:
    logger.error('Bad tag %r: %s', tag, e)
    raise

When it happens

Trigger: Calling merging get_state with a custom/foreign state-tag object that subclasses none of the supported tag types, or a tag instance whose class hierarchy was changed between Beam versions.

Common situations: Custom state tags defined by user trigger extensions; code upgraded across Beam versions where tag classes were renamed or restructured; deserialized/pickled tag objects of the wrong type.

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

• combine_fn must be specified.
• count (%d) must be a positive integer.
• Merging requested for non-mergeable state tag: %r.
• No window for %s
• {module} is not registered for pickle by value

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