Enable BigQuery CDC configuration for Python BigQuery sink

[prodriguezdefino]

Enables the configuration of CDC writes into a BigQuery table by setting the primary key columns to be used in the row mutations. This change adds the configuration of CDC usage for the BigQueryStorageWriteApiSchemaTransformProvider.

By setting configuration.setUseCDCWritesWithPrimaryKey(List.of("col1", "col2")) on the provider's config, the transform creation will create a BigQuery.Write<Row> transform that will configure the right row mutation information by checking for a Row schema like:

{
  cdc_info:{
    mutation_type: "..."              // can be UPSERT or DELETE
    change_sequence_number: "..."     // used for custom insert ordering
  }
  record:{...}                        // actual data to be inserted
}

The implementation also enables the possibility of setting a dynamic destination alongside with CDC usage by using this Row schema:

{
  cdc_info:{
    mutation_type: "..."              // can be UPSERT or DELETE
    change_sequence_number: "..."     // used for custom insert ordering
  }
  destination: "..."                  // destination table for the data
  record:{...}                        // actual data to be inserted
}

Note: In case of using dynamic destination and CDC is only supported when all the destination share the same primary key columns.

---

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

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• 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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[prodriguezdefino]

fixes #32527

[prodriguezdefino]

I have a question @ahmedabu98, how to make sure the xlang tests for BQ have run?

This branch test output for the test looks pretty similar to those that ran in master the last time, but in both cases it seems that the BQ StorageWrite transform is not being registered.

Am I looking at the right place or are the IT for xlang run in some other different task?

[ahmedabu98]

The test you are referencing is to generate wrappers for external transforms (unrelated here since this wrapper is handwritten).

The workflows testing GCP xlang IOs are in beam_PostCommit_Python_Xlang_Gcp_Dataflow and beam_PostCommit_Python_Xlang_Gcp_Direct. These are post commits so to trigger them you just need to modify their respective target files and push it to this branch:

• Direct runner trigger file
• Dataflow runner trigger file

[ahmedabu98]

Left some more comments

[ahmedabu98]

nit:

Suggested change

	"In the case of using CDC writes and setting CREATE_IF_NEEDED mode for the tables"
	+ " a primary key is required.")
	@Nullable
	public abstract List<String> getCdcWritesPrimaryKey();
	"If CREATE_IF_NEEDED disposition is set, BigQuery table(s) will be created with this primary key. "
	+ "Required when CDC writes are enabled with CREATE_IF_NEEDED.")
	@Nullable
	public abstract List<String> getPrimaryKey();

[ahmedabu98]

Let's lean on the existing checks -- I don't think we need to create new ones. The first check is already covered in BigQueryIO, and the second check is covered above in this file

[ahmedabu98]

Can we just instantiate new RowDynamicDestinations(<row schema>, <primary key>) here? And avoid instantiating it again in validateAndIncludeCDCInformation?

[ahmedabu98]

Can we make this a separate if block, outside of this if/else chain? We should be able to apply this method to both dynamic destination and single table cases. The only factor should be whether or not useCdcWrites is true

[ahmedabu98]

can we remove this duplicated code? (see previous comment)

[ahmedabu98]

I wonder why this is possible without explicitly setting "at-least-once" mode

[ahmedabu98]

"If True, your input elements are expected to have a 'record' field representing the record to write, and a 'cdc_info: {mutation_type: , change_sequence_number: }' field representing the mutation information."

[ahmedabu98]

Suggested change

	STORAGE_WRITE_API.
	STORAGE_WRITE_API and at_least_once mode.

[ahmedabu98]

Note that when the user is writing Python dicts, they should be able to supply a function that works on those dicts (ie. they shouldn't have to know what a Beam Row). Most Python users are not aware that Beam Rows and the Java IO are being used under the hood.

[ahmedabu98]

I'm realizing that this callable CDC writes option is more complex than I thought. For it to be complete and not confusing for users, we will have to provide this logic for both Python dict inputs and Beam Row inputs

I suggest we keep the CDC option to a boolean and make this improvement in a future PR

[prodriguezdefino]

Let me try to have it implemented for both in this change, if you still see it incomplete I will remove it after the next review.

[prodriguezdefino]

removed Callable argument for now, will work on it on a separated PR.