apache/beam · error · ValueError

Merging requested for non-mergeable state tag: %r.

Error message

Merging requested for non-mergeable state tag: %r.

What it means

MergingTriggerState.add_state refuses to add a value under a _ReadModifyWriteStateTag, because RMW state cannot be merged across windows - only combining, list, set, and watermark-hold state tags support merging. The library raises ValueError to prevent silently corrupting state during window merge.

Source

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

    self.counter = None

  def set_timer(
      self, window, name, time_domain, timestamp, dynamic_timer_tag=''):
    self.raw_state.set_timer(
        self._get_id(window),
        name,
        time_domain,
        timestamp,
        dynamic_timer_tag=dynamic_timer_tag)

  def clear_timer(self, window, name, time_domain, dynamic_timer_tag=''):
    for window_id in self._get_ids(window):
      self.raw_state.clear_timer(
          window_id, name, time_domain, dynamic_timer_tag=dynamic_timer_tag)

  def add_state(self, window, tag, value):
    if isinstance(tag, _ReadModifyWriteStateTag):
      raise ValueError(
          'Merging requested for non-mergeable state tag: %r.' % tag)
    elif isinstance(tag, _CombiningValueStateTag):
      tag = tag.without_extraction()
    self.raw_state.add_state(self._get_id(window), tag, value)

  def get_state(self, window, tag):
    if isinstance(tag, _CombiningValueStateTag):
      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))

View on GitHub (pinned to 12126d8942)

Solutions

1. Use a _CombiningValueStateTag (with a CombineFn) instead of a _ReadModifyWriteStateTag when state must merge across windows.
2. Restructure so the RMW value is per-window and never participates in merging.
3. If the tag is a combining tag, ensure merging code goes through the _CombiningValueStateTag branch (without_extraction) as the driver expects.

Example fix

// before
tag = _ReadModifyWriteStateTag('count')
state.add_state(window, tag, 1)  # merging state -> ValueError
// after
tag = _CombiningValueStateTag('count', combiners.CountCombineFn())
state.add_state(window, tag, 1)

from apache_beam.transforms.trigger import _ReadModifyWriteStateTag
if isinstance(tag, _ReadModifyWriteStateTag):
    raise TypeError('Use a mergeable (combining/list/set/watermark-hold) tag in merging state')

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

try:
    state.add_state(window, tag, value)
except ValueError as e:
    logger.error('Non-mergeable tag in merging state: %s', e)
    raise

When it happens

Trigger: Calling merging-state add_state(window, tag, value) with a _ReadModifyWriteStateTag (the tag type used by simple bag/RMW state) while the trigger driver is merging window state.

Common situations: Custom trigger or state code that mixes ReadModifyWriteStateTag with a merging trigger driver; code written against the non-merging TriggerDriver then reused in grouped/merging pipelines.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Invalid tag.
• No window for %s
• {module} is not registered for pickle by value
• PCollection not available, please run the pipeline.
• input is already set.

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