apache/beam · error · ValueError

Unexpected phase

Error message

Unexpected phase: %s

What it means

ValueError raised when constructing a phased combiner wrapper (a helper whose `apply` is selected per phase) with a phase string other than 'add', 'merge', 'extract', or 'convert'. The phase dispatch has no branch for the given value, so Beam treats it as an invalid phase.

Solutions

1. Use one of the supported phase strings: 'add', 'merge', 'extract', or 'convert'.
2. If the phase comes from config, validate it against the allowed set before constructing.
3. Check spelling/case of the phase value.

Example fix

// before
PhasedCombineFn(..., phase='combine')  # invalid
// after
PhasedCombineFn(..., phase='add')  # one of add|merge|extract|convert

VALID_PHASES = {'add', 'merge', 'extract', 'convert'}
if phase not in VALID_PHASES:
    raise ValueError(f'phase must be one of {sorted(VALID_PHASES)}, got {phase!r}')

from typing import Literal
def is_valid_phase(p: str) -> bool:
    return p in ('add', 'merge', 'extract', 'convert')

When it happens

Trigger: Constructing the phase-selecting combiner object with `phase='sum'`, `'all'`, a typo like `'adds'`, or an empty/None phase string.

Common situations: Typos in phase names; iterating a config list containing an unsupported phase; confusion over which phases the wrapper supports (add/merge/extract/convert only).

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Expected non-negative n, received
• Unsupported cache format
• Unsupported TimeDomain: %r.
• is not a valid windowing parameter; must be one of
• A BigQuery table or a query must be specified

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/combiners.py:962

class PhasedCombineFnExecutor(object):
  """Executor for phases of combine operations."""
  def __init__(self, phase, fn, args, kwargs):

    self.combine_fn = curry_combine_fn(fn, args, kwargs)

    if phase == 'all':
      self.apply = self.full_combine
    elif phase == 'add':
      self.apply = self.add_only
    elif phase == 'merge':
      self.apply = self.merge_only
    elif phase == 'extract':
      self.apply = self.extract_only
    elif phase == 'convert':
      self.apply = self.convert_to_accumulator
    else:
      raise ValueError('Unexpected phase: %s' % phase)

  def full_combine(self, elements):
    return self.combine_fn.apply(elements)

  def add_only(self, elements):
    return self.combine_fn.add_inputs(
        self.combine_fn.create_accumulator(), elements)

  def merge_only(self, accumulators):
    return self.combine_fn.merge_accumulators(accumulators)

  def extract_only(self, accumulator):
    return self.combine_fn.extract_output(accumulator)

  def convert_to_accumulator(self, element):
    return self.combine_fn.add_input(
        self.combine_fn.create_accumulator(), element)

View on GitHub (pinned to 12126d8942)