{"record":{"id":"6c2813a2ee525f30","repo":"apache/beam","slug":"row-missing-r-column-row-keys-s","errorCode":null,"errorMessage":"Row missing %r column. Row keys: %s","messagePattern":"Row missing %r column\\. Row keys: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/bigquery_change_history.py","lineNumber":886,"sourceCode":"        break\n\n      stream_name = stream_names[i]\n      _LOGGER.info(\n          '[Read] try_claim(%d) succeeded: reading stream %s', i, stream_name)\n\n      stream_rows = 0\n      if self._emit_raw_batches:\n        stream_batches = 0\n        for raw_batch in self._read_stream_raw(stream_name):\n          yield TimestampedValue(raw_batch, element.range_start)\n          stream_batches += 1\n        Metrics.counter('BigQueryChangeHistory',\n                        'batches_emitted').inc(stream_batches)\n      else:\n        for row in self._read_stream(stream_name):\n          ts = row.get(self._change_timestamp_column)\n          if ts is None:\n            raise ValueError(\n                'Row missing %r column. Row keys: %s' %\n                (self._change_timestamp_column, list(row.keys())))\n          if isinstance(ts, datetime.datetime):\n            ts = Timestamp.from_utc_datetime(ts)\n\n          yield TimestampedValue(row, ts)\n          stream_rows += 1\n        Metrics.counter('BigQueryChangeHistory',\n                        'rows_emitted').inc(stream_rows)\n\n      streams_read += 1\n      _LOGGER.info(\n          '[Read] Finished reading stream %d for %s: %d rows',\n          i,\n          table_key,\n          stream_rows)\n      Metrics.counter('BigQueryChangeHistory', 'streams_read').inc()\n","sourceCodeStart":868,"sourceCodeEnd":904,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/bigquery_change_history.py#L868-L904","documentation":"While streaming rows read from a BigQuery change-history stream, each row must contain the configured change timestamp column. If row.get(change_timestamp_column) returns None — the column is absent or its value is null — ValueError is raised, reporting the column name and the row's actual keys to help diagnose the mismatch.","triggerScenarios":"The configured _change_timestamp_column (e.g. '_CHANGE_TIMESTAMP' or a custom metadata column) is missing from rows returned by _read_stream, typically because the query used for the stream did not select that metadata column, or a row legitimately has NULL in that column.","commonSituations":"Custom row filters or a restricted column projection excluding the change timestamp column; using the wrong column name for the table (custom change-history columns differ per table); rows written with NULL timestamps by the producing pipeline.","solutions":["Check the row keys in the error message and correct the change_timestamp_column / query to include the actual timestamp column name.","Ensure the underlying change-history query selects the required metadata columns (_CHANGE_TIMESTAMP for CHANGES, _COMMIT_TIMESTAMP-related metadata for APPENDS).","Filter or exclude rows with NULL timestamp values upstream before this DoFn if nulls are expected in your table."],"exampleFix":"// before\nReadFromBigQueryChangeHistory(query=\"SELECT data FROM `proj.ds.table`\", ...)\n// after\nReadFromBigQueryChangeHistory(query=\"SELECT data, _CHANGE_TIMESTAMP FROM `proj.ds.table`\", ...)","handlingStrategy":"validation","validationCode":"if change_timestamp_column not in (row.keys() if isinstance(row, dict) else row_schema_fields):\n    raise KeyError(f\"query must select {change_timestamp_column}\")","typeGuard":null,"tryCatchPattern":"try:\n    _ = row[change_timestamp_column]\nexcept (KeyError, ValueError) as e:\n    log.error(\"row missing change timestamp column: %s\", e)","preventionTips":["Always SELECT the change-history metadata column (e.g. _CHANGE_TIMESTAMP) in the query","Confirm the configured column name against the actual table schema","Filter out NULL timestamp rows upstream if your table can contain them"],"tags":["python","apache-beam","bigquery","cdc","missing-column"],"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"}