[BEAM-14447] BigQueryWriteIntegrationTests.test_big_query_write_insert_errors_reporting failing in Python PostCommit

• Type: Bug
• Status: Resolved
• Priority: P1
• Resolution: Fixed
• Fix Version/s: Not applicable
• Component/s: io-py-gcp, test-failures
• Assignee: Pablo Estrada
• Reporter: Brian Hulette
• Created: 09/May/22 16:58
• Updated: 12/May/22 20:29
• Resolved: 12/May/22 20:29

Description

A new test added in https://github.com/apache/beam/pull/17517 is failing at HEAD.

 

https://ci-beam.apache.org/job/beam_PostCommit_Python37/5198/testReport/junit/apache_beam.io.gcp.bigquery_write_it_test/BigQueryWriteIntegrationTests/test_big_query_write_insert_errors_reporting/

 

        pipeline_verifiers = [
            BigqueryFullResultMatcher(
                project=self.project,
                query="SELECT number, str FROM %s" % table_id,
                data=[(1, 'some_string')]),
            BigqueryFullResultMatcher(
                project=self.project,
                query="SELECT table, reason, row_json FROM %s" % errors_table_id,
                data=
                [(
                    table_id,
                    '[{"reason": "invalid", "location": "", "debugInfo": "", \
    "message": "Missing required field: Msg_0_CLOUD_QUERY_TABLE.str."}]',
                    '{"number": 2}'),
                 (
                     table_id,
                     '[{"reason": "invalid", "location": "additional_field_str", \
    "debugInfo": "", "message": "no such field: additional_field_str."}]',
                     '{"number": 3, "str": "some_string", "additional_field_str": \
    "some_string"}')])
        ]
    
        args = self.test_pipeline.get_full_options_as_args(
            on_success_matcher=hc.all_of(*pipeline_verifiers))
    
        with beam.Pipeline(argv=args) as p:
          # pylint: disable=expression-not-assigned
          errors = (
              p | 'create' >> beam.Create(input_data)
              | 'write' >> beam.io.WriteToBigQuery(
                  table_id,
                  schema=table_schema,
                  create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                  write_disposition=beam.io.BigQueryDisposition.WRITE_EMPTY))
          (
              errors["FailedRows"]
              | 'ParseErrors' >> beam.Map(
                  lambda err: {
                      "table": err[0],
                      "reason": json.dumps(err[2]),
                      "row_json": json.dumps(err[1])
                  })
>             | 'WriteErrors' >> beam.io.WriteToBigQuery(
                  errors_table_id,
                  schema=errors_table_schema,
                  create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                  write_disposition=beam.io.BigQueryDisposition.WRITE_EMPTY))
E                 KeyError: 'FailedRows'

apache_beam/io/gcp/bigquery_write_it_test.py:461: KeyError 

Issue Links

• relates to: BEAM-14383 Improve "FailedRows" errors returned by beam.io.WriteToBigQuery
• links to: GitHub Pull Request #17601