It looks like you're trying to achieve CDC writes and you would need to set: use_cdc_writes to true and you would need to provide row_mutation_info on a per record basis.

For more info, please refer to https://beam.apache.org/releases/pydoc/2.61.0/apache_beam.io.gcp.bigquery.html