[BEAM-11719 ] Use deterministic coders for grouping keys.

[robertwb]

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang				---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[codecov]

Codecov Report

Merging #13839 (1318fbd) into master (c137e88) will increase coverage by 0.01%.
The diff coverage is n/a.

@@ Coverage Diff @@## master #13839 +/- ##
==========================================
+ Coverage 82.97% 82.99% +0.01% 
==========================================
Files 469 469 Lines 58300 58325 +25 ==========================================
+ Hits 48377 48407 +30 + Misses 9923 9918 -5 

Impacted Files	Coverage Δ
.../py38/build/srcs/sdks/python/apache_beam/pvalue.py
...ks/python/apache_beam/internal/metrics/__init__.py
...eam/runners/portability/fn_api_runner/execution.py
...hon/apache_beam/examples/cookbook/mergecontacts.py
...s/python/apache_beam/io/gcp/bigquery_file_loads.py
...testing/benchmarks/nexmark/queries/winning_bids.py
...dks/python/apache_beam/metrics/monitoring_infos.py
...ild/srcs/sdks/python/apache_beam/io/gcp/dicomio.py
.../apache_beam/runners/dataflow/internal/__init__.py
...python/apache_beam/runners/portability/__init__.py
... and 928 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f647b05...1318fbd. Read the comment docs.

[robertwb]

R: @udim
CC: @katsiapis

[udim]

From what I can tell, parse_table_reference is used to normalize and populate TableReference instances. They are essentially a "(project, dataset, table) tuple."
It should be possible to beam.Map(lambda x: (x.tableId, x.datasetId, x.projectId)) | beam.Distinct() | beam.Map(lambda x: parse_table_reference(*x))

cc: @pabloem

[robertwb]

Assuming exactly these three fields would not be as future proof.

[udim]

Ack. The official BigQuery client (which this code doesn't use) has a hashable wrapper for this.
https://googleapis.dev/python/bigquery/latest/generated/google.cloud.bigquery.table.TableReference.html

[robertwb]

Thanks, PTAL.

[robertwb]

Assuming exactly these three fields would not be as future proof.

[robertwb]

Run Python_PVR_Flink PreCommit

[chunyang]

I seem to be bumping into an error caused by deterministic coder enforcement. Do you mind taking a look at ##14112?