apache/beam · error · ValueError

Input schema already has a field named

Error message

Input schema already has a field named {field}.

What it means

When augmenting a PCollection with windowing info, `_ExtractWindowingInfo` refuses to overwrite existing schema fields. yaml_mapping.py:960 raises ValueError if a target field name in `fields` already exists on the input element's schema.

Solutions

1. Rename the output fields via a mapping, e.g. fields: {event_ts: timestamp} to avoid collision.
2. Drop or rename the conflicting column in a prior transform before extracting windowing info.
3. Remove the colliding field from the `fields` list if it's already present.

Example fix

# before
fields: [timestamp, window_start]
# after
fields: {window_ts: timestamp, window_start: window_start}

existing = set(pc.element_type._fields) if hasattr(pc.element_type, '_fields') else set()
clashes = set(fields.keys() if isinstance(fields, Mapping) else fields) & existing
assert not clashes, f'rename these output fields to avoid clashes: {clashes}'

def has_no_field_clash(fields, existing_fields):
    return not (set(fields) & set(existing_fields))

try:
    pc = extract_windowing_info(pc, fields=fields)
except ValueError as e:
    if 'already has a field' in str(e):
        fields = {f'window_{k}': v for k, v in fields.items()}
        pc = extract_windowing_info(pc, fields=fields)
    else:
        raise

When it happens

Trigger: The input PCollection schema already contains a column named e.g. 'timestamp' or 'window_start' and the transform is asked to add windowing info into that same name.

Common situations: Source data already has a 'timestamp' column and the default fields list includes 'timestamp'; pipeline renames upstream fields into windowing-reserved names.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• At most one of --create_test and --fix_tests may be…
• Cannot convert element of type
• "Cannot specify 'callable' with 'path' and 'name' for…
• Chain at missing transforms property.
• Dependencies must be a list of strings, got

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

Appendix: source

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

  """
  if fields is None:
    fields = ['timestamp', 'window_start', 'window_end']
  if not isinstance(fields, Mapping):
    if isinstance(fields, Iterable) and not isinstance(fields, str):
      fields = {fld: fld for fld in fields}
    else:
      raise TypeError(
          'Fields must be a mapping or iterable of strings, got {fields}')

  existing_fields = named_fields_from_element_type(pcoll.element_type)
  new_fields = []
  for field, value in fields.items():
    if value not in _WINDOWING_INFO_TYPES:
      raise ValueError(
          f'{value} is not a valid windowing parameter; '
          f'must be one of {list(_WINDOWING_INFO_TYPES.keys())}')
    elif field in existing_fields:
      raise ValueError(f'Input schema already has a field named {field}.')
    else:
      new_fields.append((field, _WINDOWING_INFO_TYPES[value]))

  def augment_row(
      row,
      timestamp=beam.DoFn.TimestampParam,
      window=beam.DoFn.WindowParam,
      pane_info=beam.DoFn.PaneInfoParam):
    as_dict = row._asdict()
    for field, value in fields.items():
      as_dict[field] = _WINDOWING_INFO_EXTRACTORS[value](locals())
    return beam.Row(**as_dict)

  return pcoll | beam.Map(augment_row).with_output_types(
      row_type.RowTypeConstraint.from_fields(
          existing_fields + new_fields))  # type: ignore[operator]

View on GitHub (pinned to 12126d8942)