apache/beam · error · ValueError

combine_fn must be specified.

Error message

combine_fn must be specified.

What it means

_CombiningValueStateTag.__init__ requires a combine_fn and raises ValueError when it is missing, None, or otherwise falsy. The tag is a trigger-state holder keyed by a CombineFn, so without one the state cannot accumulate or merge values.

Source

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

  """StateTag pointing to an element."""
  def __repr__(self):
    return 'SetStateTag({tag})'.format(tag=self.tag)

  def with_prefix(self, prefix):
    return _SetStateTag(prefix + self.tag)


class _CombiningValueStateTag(_StateTag):
  """StateTag pointing to an element, accumulated with a combiner.

  The given tag must be unique for this step. The given CombineFn will be
  applied (possibly incrementally and eagerly) when adding elements."""

  # TODO(robertwb): Also store the coder (perhaps extracted from the combine_fn)
  def __init__(self, tag, combine_fn):
    super().__init__(tag)
    if not combine_fn:
      raise ValueError('combine_fn must be specified.')
    if not isinstance(combine_fn, core.CombineFn):
      combine_fn = core.CombineFn.from_callable(combine_fn)
    self.combine_fn = combine_fn

  def __repr__(self):
    return 'CombiningValueStateTag(%s, %s)' % (self.tag, self.combine_fn)

  def with_prefix(self, prefix):
    return _CombiningValueStateTag(prefix + self.tag, self.combine_fn)

  def without_extraction(self):
    class NoExtractionCombineFn(core.CombineFn):
      setup = self.combine_fn.setup
      create_accumulator = self.combine_fn.create_accumulator
      add_input = self.combine_fn.add_input
      merge_accumulators = self.combine_fn.merge_accumulators
      compact = self.combine_fn.compact
      extract_output = staticmethod(lambda x: x)

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a CombineFn instance, e.g. combiners.CountCombineFn() or SumCombineFn(), as the second argument.
2. A plain callable is accepted too: it is wrapped via CombineFn.from_callable, so pass the function rather than None.
3. Check that the variable holding the combiner is actually defined at the call site (not shadowed or set to None).

Example fix

// before
tag = _CombiningValueStateTag('sum')
// after
tag = _CombiningValueStateTag('sum', combiners.SumCombineFn())

assert combine_fn is not None, 'combine_fn is required for _CombiningValueStateTag'

def has_combine_fn(fn):
    return bool(fn) and (isinstance(fn, core.CombineFn) or callable(fn))

try:
    tag = _CombiningValueStateTag(name, combine_fn)
except ValueError:
    tag = _CombiningValueStateTag(name, core.CombineFn.from_callable(sum))

When it happens

Trigger: Constructing _CombiningValueStateTag('mytag', None) or with an omitted second argument; also triggered by passing a falsy object (e.g. an empty container) as combine_fn.

Common situations: Custom trigger/state code inside Beam's trigger driver internals; tests that build state tags manually and forget the combiner; refactors that renamed a combine_fn variable leaving it undefined/None.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• count (%d) must be a positive integer.
• Invalid tag.
• document_field cannot be empty
• Expected text content in {type(item).__name__} {item.id}, go
• Expected image content in {type(item).__name__} {item.id}, g

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