{"record":{"id":"50e71e02e27799f8","repo":"apache/beam","slug":"input-schema-already-has-a-field-named-field","errorCode":null,"errorMessage":"Input schema already has a field named {field}.","messagePattern":"Input schema already has a field named (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_mapping.py","lineNumber":960,"sourceCode":"  \"\"\"\n  if fields is None:\n    fields = ['timestamp', 'window_start', 'window_end']\n  if not isinstance(fields, Mapping):\n    if isinstance(fields, Iterable) and not isinstance(fields, str):\n      fields = {fld: fld for fld in fields}\n    else:\n      raise TypeError(\n          'Fields must be a mapping or iterable of strings, got {fields}')\n\n  existing_fields = named_fields_from_element_type(pcoll.element_type)\n  new_fields = []\n  for field, value in fields.items():\n    if value not in _WINDOWING_INFO_TYPES:\n      raise ValueError(\n          f'{value} is not a valid windowing parameter; '\n          f'must be one of {list(_WINDOWING_INFO_TYPES.keys())}')\n    elif field in existing_fields:\n      raise ValueError(f'Input schema already has a field named {field}.')\n    else:\n      new_fields.append((field, _WINDOWING_INFO_TYPES[value]))\n\n  def augment_row(\n      row,\n      timestamp=beam.DoFn.TimestampParam,\n      window=beam.DoFn.WindowParam,\n      pane_info=beam.DoFn.PaneInfoParam):\n    as_dict = row._asdict()\n    for field, value in fields.items():\n      as_dict[field] = _WINDOWING_INFO_EXTRACTORS[value](locals())\n    return beam.Row(**as_dict)\n\n  return pcoll | beam.Map(augment_row).with_output_types(\n      row_type.RowTypeConstraint.from_fields(\n          existing_fields + new_fields))  # type: ignore[operator]\n\n","sourceCodeStart":942,"sourceCodeEnd":978,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_mapping.py#L942-L978","documentation":"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.","triggerScenarios":"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.","commonSituations":"Source data already has a 'timestamp' column and the default fields list includes 'timestamp'; pipeline renames upstream fields into windowing-reserved names.","solutions":["Rename the output fields via a mapping, e.g. fields: {event_ts: timestamp} to avoid collision.","Drop or rename the conflicting column in a prior transform before extracting windowing info.","Remove the colliding field from the `fields` list if it's already present."],"exampleFix":"# before\nfields: [timestamp, window_start]\n# after\nfields: {window_ts: timestamp, window_start: window_start}","handlingStrategy":"validation","validationCode":"existing = set(pc.element_type._fields) if hasattr(pc.element_type, '_fields') else set()\nclashes = set(fields.keys() if isinstance(fields, Mapping) else fields) & existing\nassert not clashes, f'rename these output fields to avoid clashes: {clashes}'","typeGuard":"def has_no_field_clash(fields, existing_fields):\n    return not (set(fields) & set(existing_fields))","tryCatchPattern":"try:\n    pc = extract_windowing_info(pc, fields=fields)\nexcept ValueError as e:\n    if 'already has a field' in str(e):\n        fields = {f'window_{k}': v for k, v in fields.items()}\n        pc = extract_windowing_info(pc, fields=fields)\n    else:\n        raise","preventionTips":["Inspect the input schema before adding windowing info.","Prefix windowing output field names (e.g. window_ts) to avoid collisions.","Rename conflicting upstream columns in an earlier transform."],"tags":["python","apache-beam","yaml","schema-conflict"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}