From fba6c67ac174b7db96d3fcba7d0994b1bac63794 Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Wed, 30 Sep 2026 15:10:19 +0000 Subject: [PATCH 1/2] Fix partial insertError metric status bug --- .../apache_beam/io/gcp/bigquery_tools.py | 6 +++- .../apache_beam/io/gcp/bigquery_tools_test.py | 30 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools.py b/sdks/python/apache_beam/io/gcp/bigquery_tools.py index 0d62ec5233c1..bec7685d4993 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_tools.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_tools.py @@ -777,7 +777,11 @@ def _insert_all_rows( service_call_metric.call('ok') else: for insert_error in errors: - service_call_metric.call(insert_error['errors'][0]) + # Record the BigQuery reason (e.g. 'invalid') of the first error for + # each failed row. ServiceCallMetric cannot interpret the raw error + # dict and would otherwise record the status 'None'. + row_errors = insert_error.get('errors') or [{}] + service_call_metric.call(row_errors[0].get('reason') or 'unknown') except (ClientError, GoogleAPICallError) as e: # e.code contains the numeric http status code. service_call_metric.call(e.code) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools_test.py b/sdks/python/apache_beam/io/gcp/bigquery_tools_test.py index 078c42160941..c2a5c73f6755 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_tools_test.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_tools_test.py @@ -575,6 +575,36 @@ def test_insert_rows_sets_metric_on_failure(self): self.verify_write_call_metric( "my_project", "my_dataset", "my_table", "ok", 1) + @unittest.skipIf(ClientError is None, 'GCP dependencies are not installed') + def test_insert_rows_sets_metric_on_row_errors(self): + MetricsEnvironment.process_wide_container().reset() + client = mock.Mock() + + def row_error(index, *reasons): + return { + 'index': index, + 'errors': [{ + 'reason': r, 'message': 'msg' + } for r in reasons], + } + + client.insert_rows_json.return_value = [ + row_error(0, 'invalid', 'stopped'), + row_error(1, 'invalid'), + row_error(2, 'stopped'), + ] + wrapper = beam.io.gcp.bigquery_tools.BigQueryWrapper(client) + success, errors = wrapper.insert_rows( + "my_project", "my_dataset", "my_table", [{'a': 1}] * 3) + + self.assertFalse(success) + self.assertEqual(client.insert_rows_json.return_value, errors) + # One metric per failed row, labelled with the reason of its first error. + self.verify_write_call_metric( + "my_project", "my_dataset", "my_table", "invalid", 2) + self.verify_write_call_metric( + "my_project", "my_dataset", "my_table", "stopped", 1) + @unittest.skipIf(ClientError is None, 'GCP dependencies are not installed') def test_start_query_job_priority_configuration(self): client = mock.Mock() From ce7d0621f45f753510d4f0b8b55ef0ac518e54c1 Mon Sep 17 00:00:00 2001 From: Jack McCluskey <34928439+jrmccluskey@users.noreply.github.com> Date: Wed, 30 Sep 2026 12:21:37 -0400 Subject: [PATCH 2/2] Update bigquery_tools.py streamline comment --- sdks/python/apache_beam/io/gcp/bigquery_tools.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools.py b/sdks/python/apache_beam/io/gcp/bigquery_tools.py index bec7685d4993..097b44caa80b 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_tools.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_tools.py @@ -778,8 +778,7 @@ def _insert_all_rows( else: for insert_error in errors: # Record the BigQuery reason (e.g. 'invalid') of the first error for - # each failed row. ServiceCallMetric cannot interpret the raw error - # dict and would otherwise record the status 'None'. + # each failed row. row_errors = insert_error.get('errors') or [{}] service_call_metric.call(row_errors[0].get('reason') or 'unknown') except (ClientError, GoogleAPICallError) as e: