apache/beam · error · ValueError

Can only append fields on a schema'd input.

Error message

Can only append fields on a schema'd input.

What it means

Raised by `normalize_fields` when the `append` option is set but the input PCollection has no element schema. Appending fields builds the new output schema from the existing input schema, so an untyped input cannot be appended to; the transform raises rather than guessing types.

Solutions

1. Ensure the input has a schema, e.g. insert a Cast or use `beam.Row(...)` with type annotations upstream.
2. Remove `append: true` if you intend to create a fresh set of fields instead of extending a schema.
3. Use MapToFields on the untyped input to produce a schema first, then append.

Example fix

// before
- type: AddFields
  input: raw_json
  config:
    append: true
    fields: {score: {type: double}}
// after
- type: Cast
  input: raw_json
  config:
    fields: {id: int64, name: string}
- type: AddFields
  input: cast_output
  config:
    append: true
    fields: {score: {type: double}}

try:
    named_fields_from_element_type(pcoll.element_type)
except (TypeError, ValueError):
    raise ValueError('append requires a schema\'d input; add a Cast or beam.Row step upstream.')

try:
    out = normalize_fields(pcoll, fields={'score': 0.0}, append=True)
except ValueError as e:
    if "Can only append fields" in str(e):
        out = normalize_fields(cast_to_schema(pcoll), fields={'score': 0.0}, append=True)
    else:
        raise

When it happens

Trigger: Using a YAML transform (AddFields / MapToFields path) with `append: true` where `named_fields_from_element_type(pcoll.element_type)` fails because the input is not a schema'd row (e.g. plain dict from a generic map).

Common situations: Chaining AddFields after an untyped Map or JSON source; forgetting that Beam YAML transforms require schema'd PCollections; a DoFn without output type hints feeding the transform.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• Can only drop fields on a schema'd input.
• Cannot convert element of type
• Dropping unknown field
• Exploding unknown field
• f"Mapping destinations

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_mapping.py:669

    error_handling: Whether and where to output records that throw errors when
      the above expressions are evaluated.
  """  # pylint: disable=line-too-long
  keep_fn = _as_callable_for_pcoll(pcoll, keep, "keep", language or 'generic')
  return pcoll | beam.Filter(keep_fn)


def is_expr(v):
  return isinstance(v, str) or (isinstance(v, dict) and 'expression' in v)


def normalize_fields(pcoll, fields, drop=(), append=False, language='generic'):
  try:
    input_schema = dict(named_fields_from_element_type(pcoll.element_type))
  except (TypeError, ValueError) as exn:
    if drop:
      raise ValueError("Can only drop fields on a schema'd input.") from exn
    if append:
      raise ValueError("Can only append fields on a schema'd input.") from exn
    elif any(is_expr(x) for x in fields.values()):
      raise ValueError("Can only use expressions on a schema'd input.") from exn
    input_schema = {}

  if drop and not append:
    raise ValueError("Can only drop fields if append is true.")
  for name in drop:
    if name not in input_schema:
      raise ValueError(f'Dropping unknown field "{name}"')
  if append:
    for name in fields:
      if name in input_schema and name not in drop:
        raise ValueError(
            f'Redefinition of field "{name}". '
            'Cannot append a field that already exists in original input.')

  if append:
    return input_schema, {

View on GitHub (pinned to 12126d8942)