[BEAM-11659] Add new schema types to Pub/Sub SQL

[dpcollins-google]

Add pubsub handling of byte array payload schemas and ARRAY<ROW<VARCHAR key, VARCHAR value>> attributes schemas.

These enable pure-sql lossless copies to and from Pub/Sub Lite and kafka.

---

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.

[dpcollins-google]

R: @amaliujia

[amaliujia]

cc @robinyqiu @TheNeuralBit

[TheNeuralBit]

LGTM overall, thank you. A couple more minor requests:

• Add test(s) in PubsubTableProviderIT for the new configurations
• Update https://beam.apache.org/documentation/dsls/sql/extensions/create-external-table/#pubsub for the new attribute/payload options (this could be done in a separate PR if you prefer)

I'll run some internal tests to double-check everything

[TheNeuralBit]

Run SQL PostCommit

[dpcollins-google]

Add test(s) in PubsubTableProviderIT for the new configurations

Done

[dpcollins-google]

Update https://beam.apache.org/documentation/dsls/sql/extensions/create-external-table/#pubsub for the new

I'm going to bulk update this once the other two PRs are merged.

[dpcollins-google]

Run SQL PostCommit

[TheNeuralBit]

Thanks! It looks like the new tests aren't passing though. I think one issue is using BYTES instead of VARBINARY, the other is less clear. Maybe because value isn't allowed to be an identifier? You might try wrapping it in backticks or changing the name.

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[TheNeuralBit]

Run SQL PostCommit

[dpcollins-google]

Run SQL PostCommit

[dpcollins-google]

Run Java PreCommit

[TheNeuralBit]

Run Java PreCommit

[TheNeuralBit]

LGTM, thank you! Ran internal tests and they look good.

I can merge when CI is green