Enable configuration of a CDC mutation info Callable for CDC Writes into BigQuery

[prodriguezdefino]

This change enables the configuration of a Callable function to populate the CDC mutation information on a per record basis when using BigQuery as destination with CDC writes.

Given that beam.io.WriteToBigQuery supports input data as Dict or beam.Row, the configured CDC info Callable should conform to one of those 2 types: Callable[[Dict], Dict] or Callable[[beam.Row], beam.Row].

Example configuration for Dict type data:

    ...
    data = [
        # record: (name, value)
        {
            'name': 'cdc_test', 'value': 5
        },
        {
            'name': 'cdc_test', 'value': 3
        }
    ]

    schema = {
        "fields": [{
            "name": "name", "type": "STRING"
        }, {
            "name": "value", "type": "INTEGER"
        }]
    }

    def cdc_info_fn(data: Dict) -> Dict:
      return {
          'mutation_type': 'UPSERT',
          'change_sequence_number': 'AAA/' + str(data["value"])
      }
    
    with beam.Pipeline(argv=self.args) as p:
      _ = (
          p
          | beam.Create(data)
          | beam.io.WriteToBigQuery(
              table=table_id,
              method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API,
              use_cdc_writes=cdc_info_fn,
              schema=schema,
              primary_key=["name"],
              use_at_least_once=True))

# expected data written on BQ table after pipeline finishes
# {
#    'name': 'cdc_test', 'value': 5
# }

Analogously, in case of processing Beam Rows the pipeline would look like:

    ...
    row_data = [
        beam.Row(name="cdc_test", value=5, route=3),
        beam.Row(name="cdc_test", value=3, route=3),
        beam.Row(name="cdc_test", value=2, route=1)
    ]
    def cdc_info_rows(row: beam.Row) -> beam.Row:
      return beam.Row(
          mutation_type="UPSERT",
          change_sequence_number="AAA/" + str(row.value + row.route))

    with beam.Pipeline(argv=self.args) as p:
       _ = (
          p
          | beam.Create(row_data)
          | beam.io.WriteToBigQuery(
              table=table_id,
              method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API,
              use_cdc_writes=cdc_info_rows,
              primary_key=["name"],
              use_at_least_once=True))

# expected data written on BQ table after pipeline finishes
# {
#    'name': 'cdc_test', 'value': 5, 'route': 3
# }

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.