apache/beam · error · KeyError

Make sure the values passed in `fields` are the keys in the…

Error message

Make sure the values passed in `fields` are the keys in the input `beam.Row`.

What it means

When building format values for the query, __call__ reads request_dict[field] for each field in self.fields. A KeyError means a field listed in fields is not present in the input beam.Row; it is re-raised with this guidance message.

Solutions

1. Align self.fields with the actual keys of the input beam.Row (print row._asdict() keys)
2. Fix typos/casing in the fields list
3. Add the missing column upstream or use condition_value_fn to derive values defensively

Example fix

// before
BigQueryEnrichmentHandler(..., fields=['userID'])
// after (matches row key)
BigQueryEnrichmentHandler(..., fields=['user_id'])

def validate_fields_in_row(fields, row):
    missing = [f for f in fields if f not in row._asdict()]
    if missing:
        raise KeyError(f'fields missing from input Row: {missing}; have {list(row._asdict())}')

def row_has_fields(fields, row) -> bool:
    d = row._asdict()
    return all(f in d for f in fields)

try:
    out = handler(row)
except KeyError as e:
    logger.error('Enrichment key mismatch: %s', e)
    raise

When it happens

Trigger: Input beam.Row lacks a key named in self.fields (typo, renamed column, upstream transform changed schema).

Common situations: Field name mismatch between handler config and upstream PCollection schema; case-sensitivity mistakes; nullable/absent columns removed by earlier transformations.

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

• Both a query and an output type of 'BEAM_ROW' were…
• Converting BigQuery type
• Field is not nullable.
• invalid schema type
• RECORD/STRUCT are not primitive types

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/bigquery.py:216

    if isinstance(request, list):
      values = []
      responses = []
      requests_map: dict[Any, list[beam.Row]] = defaultdict(list)
      batch_size = len(request)
      raw_query = self.query_template
      if batch_size > 1:
        batched_condition_template = ' or '.join(
            [fr'({self.row_restriction_template})'] * batch_size)
        raw_query = self.query_template.replace(
            self.row_restriction_template, batched_condition_template)
      for req in request:
        request_dict = req._asdict()
        try:
          current_values = (
              self.condition_value_fn(req) if self.condition_value_fn else
              [request_dict[field] for field in self.fields])
        except KeyError as e:
          raise KeyError(
              "Make sure the values passed in `fields` are the "
              "keys in the input `beam.Row`." + str(e))
        values.extend(current_values)
        requests_map[self.create_row_key(req)].append(req)
      query = raw_query.format(*values)

      responses_dict = self._execute_query(query)
      unmatched_requests = {
          key: list(reqs)
          for key, reqs in requests_map.items()
      }
      if responses_dict:
        for response in responses_dict:
          response_row = beam.Row(**response)
          response_key = self.create_row_key(response_row)
          if response_key in unmatched_requests:
            for req in unmatched_requests.pop(response_key):
              responses.append((req, response_row))

View on GitHub (pinned to 12126d8942)