[Feature Request]: Support table schema field addition with Storage Write API autoSchemaUpdate

[ahmedabu98]

What would you like to happen?

Auto-schema updates allow Storage API writes to adapt to BigQuery schema changes (e.g. required field relaxes to nullable). However, currently it doesn't support schema field additions (ie. a new column being added). Best to explain with an example:

Say we are writing to a table with schema 1: {field1, field2}
and while we are writing, the table updates to schema 2: {field1, field2, field3}

If we start writing TableRows that match schema 2, they will fail and go to the PCollection's DLQ. This fails not at the data insertion step, but when we convert the TableRows to proto prior to that using StorageApiDynamicDestinationsTableRow. This dynamic destinations object contains a MessageConverter that is instantiated once for each destination. The destination table's schema is fetched in the constructor of this MessageConverter object and is never updated later:

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsTableRow.java

Lines 93 to 123 in 843e7fd

	TableRowConverter(DestinationT destination, DatasetService datasetService) throws Exception {
	TableSchema localTableSchema = getSchema(destination);
	TableReference tableReference = getTable(destination).getTableReference();
	if (localTableSchema == null) {
	// If the table already exists, then try and fetch the schema from the existing
	// table.
	localTableSchema = SCHEMA_CACHE.getSchema(tableReference, datasetService);
	if (localTableSchema == null) {
	if (createDisposition == CreateDisposition.CREATE_NEVER) {
	throw new RuntimeException(
	"BigQuery table "
	+ tableReference
	+ " not found. If you wanted to "
	+ "automatically create the table, set the create disposition to CREATE_IF_NEEDED and specify a "
	+ "schema.");
	} else {
	throw new RuntimeException(
	"Schema must be set for table "
	+ tableReference
	+ " when writing TableRows using Storage API and "
	+ "using a create disposition of CREATE_IF_NEEDED.");
	}
	}
	} else {
	// Make sure we register this schema with the cache, unless there's already a more
	// up-to-date schema.
	localTableSchema =
	MoreObjects.firstNonNull(
	SCHEMA_CACHE.putSchemaIfAbsent(tableReference, localTableSchema), localTableSchema);
	}
	this.tableSchema = localTableSchema;

There should be a way to update the schema when appropriate.

Issue Priority

Priority: 3 (nice-to-have improvement)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[ahmedabu98]

This feature is already supported, closing.

[lucasnoetzold]

There should be a way to update the schema when appropriate.

This feature is already supported, closing.

I ran into this issue recently, is there any way to update the schema without restarting the pipeline?

[SumLare]

@ahmedabu98 having same issue as @lucasnoetzold, since ignoreUnknownValues is required when using withAutoSchemaUpdate, adding new column doesn't refresh the schema and value is always null