This is an automated email from the ASF dual-hosted git repository.
tvalentyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 9f5f83e03d7 [Python] Keep BigQuery load jobs distinct for table
aliases (#40001)
9f5f83e03d7 is described below
commit 9f5f83e03d71ee2ec067bd5e3b0186a91c51036b
Author: Bruno Volpato <[email protected]>
AuthorDate: Thu Oct 22 12:54:08 2026 -0400
[Python] Keep BigQuery load jobs distinct for table aliases (#40001)
* [Python] Keep BigQuery load jobs distinct for table aliases
* Document BigQuery destination-alias fix
* [Python] Disable gRPC fork support before tox test startup
---
CHANGES.md | 1 +
.../apache_beam/io/gcp/bigquery_file_loads.py | 11 ++-
.../apache_beam/io/gcp/bigquery_file_loads_test.py | 80 ++++++++++++++++++++++
sdks/python/tox.ini | 3 +
4 files changed, 88 insertions(+), 7 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 4e2fffb8938..1f57af6cdb7 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -105,6 +105,7 @@
## Bugfixes
+* (Python) Fixed BigQuery `FILE_LOADS` silently omitting files when dynamic
destinations use different spellings of the same table
([#40001](https://github.com/apache/beam/pull/40001)).
* (Python) DaskRunner now preserves outputs emitted by `DoFn.finish_bundle()`
instead of silently dropping them
([#40519](https://github.com/apache/beam/issues/40519)).
* (Python) BigQuery streaming inserts now reset the retry budget and backoff
for each batch instead of sharing them across a bundle
([#39995](https://github.com/apache/beam/pull/39995)).
* (Python) State write failures reported by portable runners now fail the
bundle instead of silently losing state updates
([#39992](https://github.com/apache/beam/pull/39992)).
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
b/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
index 79940d109f0..a812467ef18 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
@@ -758,14 +758,11 @@ class TriggerLoadJobs(beam.DoFn):
if table_reference.projectId is None:
table_reference.projectId = vp.RuntimeValueProvider.get_value(
'project', str, '') or self.project
- # Load jobs for a single destination are always triggered from the same
- # worker. This means that we can generate a deterministic numbered job id,
- # and not need to worry.
+ # Use the same destination identity as file grouping. Equivalent table
+ # references can form separate groups with different files but identical
+ # partition numbers, so normalizing the reference would collide job IDs.
destination_hash = _bq_uuid(
- '%s:%s.%s' % (
- table_reference.projectId,
- table_reference.datasetId,
- table_reference.tableId))
+ bigquery_tools.get_hashable_destination(destination))
job_name = '%s_%s_pane%s_partition%s' % (
load_job_name_prefix, destination_hash, pane_info.index, partition_key)
_LOGGER.info('Load job has %s files. Job name is %s.', len(files),
job_name)
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
index 7ccda7ecad8..91ecb0d700d 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
@@ -408,6 +408,86 @@ class TestPartitionFiles(unittest.TestCase):
class TestBigQueryFileLoads(_TestCaseWithTempDirCleanUp):
+ @parameterized.expand([
+ ('append', BigQueryDisposition.WRITE_APPEND),
+ ('truncate', BigQueryDisposition.WRITE_TRUNCATE),
+ ])
+ def test_equivalent_destinations_keep_distinct_load_jobs(
+ self, unused_name, write_disposition):
+ destinations = [
+ 'project:dataset.table', 'dataset.table', 'project.dataset.table'
+ ]
+ files = ['gs://bucket/file%d' % i for i in range(len(destinations))]
+ jobs = {}
+
+ def insert_job(request, upload=None):
+ job = request.job
+ job_id = job.jobReference.jobId
+ if job_id in jobs:
+ raise HttpError({'status': '409'},
+ ('Already Exists: Job project:US.' + job_id).encode(),
+ '')
+ job.status = bigquery_api.JobStatus(state='DONE')
+ jobs[job_id] = job
+ return job
+
+ client = mock.Mock()
+ client.jobs.Insert.side_effect = insert_job
+ client.jobs.Get.side_effect = lambda request: jobs[request.jobId]
+ schema = mock.Mock(return_value=_ELEMENTS_SCHEMA)
+ parameters = mock.Mock(return_value={'timePartitioning': {'type': 'DAY'}})
+ loader = bqfl.TriggerLoadJobs(
+ project='project',
+ schema=schema,
+ temporary_tables=True,
+ additional_bq_parameters=parameters,
+ test_client=client)
+ loader.start_bundle()
+
+ # File grouping keeps these equivalent destination strings separate, so
+ # each group has a partition zero containing different source files.
+ partitions = [(destination, (0, [path]))
+ for destination, path in zip(destinations, files)]
+ for partition in partitions:
+ list(loader.process(partition, 'load', pane_info=mock.Mock(index=0)))
+ loaded = [result.value for result in loader.finish_bundle()]
+ first_ids = [job.jobId for _, job in loaded]
+ self.assertEqual(
+ 'load_29b51c1b5b03c4804123fd20083029da_pane0_partition0', first_ids[0])
+
+ # Replaying the same partitions must resolve to the same jobs, without
+ # treating a different destination group's files as an already-run job.
+ loader.start_bundle()
+ for partition in partitions:
+ list(loader.process(partition, 'load', pane_info=mock.Mock(index=0)))
+ replayed = [result.value for result in loader.finish_bundle()]
+ self.assertEqual(first_ids, [job.jobId for _, job in replayed])
+ self.assertEqual([call(dest) for dest in destinations] * 2,
+ schema.call_args_list)
+ self.assertEqual([call(dest) for dest in destinations] * 2,
+ parameters.call_args_list)
+
+ copier = bqfl.TriggerCopyJobs(
+ project='project',
+ write_disposition=write_disposition,
+ test_client=client)
+ copier.setup()
+ copier.start_bundle()
+ copier.process(replayed, 'copy')
+ list(copier.finish_bundle())
+
+ copy_jobs = [job for job in jobs.values() if job.configuration.copy]
+ copied_files = [
+ path for job in copy_jobs
+ for path in jobs[job.configuration.copy.sourceTable.tableId].
+ configuration.load.sourceUris
+ ]
+ self.assertCountEqual(files, copied_files)
+ self.assertEqual(3, len(set(first_ids)))
+ self.assertEqual(
+ [write_disposition, 'WRITE_APPEND', 'WRITE_APPEND'],
+ [job.configuration.copy.writeDisposition for job in copy_jobs])
+
def test_trigger_load_jobs_with_empty_files(self):
destination = "project:dataset.table"
empty_files = []
diff --git a/sdks/python/tox.ini b/sdks/python/tox.ini
index 7a822cab759..b579f91fd2e 100644
--- a/sdks/python/tox.ini
+++ b/sdks/python/tox.ini
@@ -46,6 +46,9 @@ allowlist_externals =
chmod
setenv =
RUN_SKIPPED_PY3_TESTS=0
+ # Set before importing gRPC; test-class setup is too late.
+ # TODO(https://github.com/grpc/grpc/issues/37710): Remove once fixed.
+ GRPC_ENABLE_FORK_SUPPORT=0
# Use an isolated tmp dir for tests that get slowed down by scanning /tmp.
TMPDIR={envtmpdir}
# Silence warning about ignoring PYTHONPATH.
