apache/beam · error · ValueError

Unsupported atomic type

Error message

Unsupported atomic type: {0}

What it means

`typing_from_runner_api` maps schema_pb2 atomic_type enum values to Python primitives via ATOMIC_TYPE_TO_PRIMITIVE. When the proto contains an atomic_type with no entry in that table, the KeyError is converted into this ValueError, indicating an unknown/unsupported atomic type (often from a newer protocol version).

Solutions

1. Upgrade apache-beam so the enum value is known and mapped
2. Check which atomic_type value is present and change the schema to use a supported primitive
3. Align the producing SDK version with the Python SDK

if fieldtype_proto.WhichOneof('type_info') == 'atomic_type' and fieldtype_proto.atomic_type not in schema_pb2.AtomicType.items():
  raise ValueError(f'unsupported atomic_type {fieldtype_proto.atomic_type}')

try:
  pytype = converter.typing_from_runner_api(fieldtype)
except ValueError as e:
  log.error('upgrade apache-beam to decode this schema'); raise

When it happens

Trigger: Decoding a FieldType proto whose atomic_type is not in ATOMIC_TYPE_TO_PRIMITIVE, e.g. proto produced by a newer Beam version or cross-language pipeline with extended atomic types.

Common situations: Version mismatch between runner and Python SDK; cross-language schemas using types Python doesn't map.

Related errors

• Converting BigQuery type
• Encountered an unsupported type
• Encountered option with unsupported type. Only atomic_type…
• Failed to decode schema due to an issue with Field proto
• Unrecognized atomic_type

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

Appendix: source

Thrown at sdks/python/apache_beam/typehints/schemas.py:566

      self, fieldtype_proto: schema_pb2.FieldType) -> type:
    if fieldtype_proto.nullable:
      # In order to determine the inner type, create a copy of fieldtype_proto
      # with nullable=False and pass back to typing_from_runner_api
      base_type = schema_pb2.FieldType()
      base_type.CopyFrom(fieldtype_proto)
      base_type.nullable = False
      base = self.typing_from_runner_api(base_type)
      if base == Any:
        return base
      else:
        return Optional[base]

    type_info = fieldtype_proto.WhichOneof("type_info")
    if type_info == "atomic_type":
      try:
        return ATOMIC_TYPE_TO_PRIMITIVE[fieldtype_proto.atomic_type]
      except KeyError:
        raise ValueError(
            "Unsupported atomic type: {0}".format(fieldtype_proto.atomic_type))
    elif type_info == "array_type":
      return Sequence[self.typing_from_runner_api(
          fieldtype_proto.array_type.element_type)]
    elif type_info == "map_type":
      return Mapping[
          self.typing_from_runner_api(fieldtype_proto.map_type.key_type),
          self.typing_from_runner_api(fieldtype_proto.map_type.value_type)]
    elif type_info == "row_type":
      schema = fieldtype_proto.row_type.schema
      schema_options = [
          self.option_from_runner_api(option_proto)
          for option_proto in schema.options
      ]
      field_options = {
          field.name: [
              self.option_from_runner_api(option_proto)
              for option_proto in field.options

View on GitHub (pinned to 12126d8942)