Is it possible to do CDC stream UPSERT with java library?

[ismailsimsek]

is it possible to do CDC stream inserts? is this feature available in java library?

getting following error

    JSONArray jsonArr = new JSONArray();
    JSONObject record = new JSONObject().put("c_id", 2).put("c_string", "record-1").put("_CHANGE_TYPE", "UPSERT");
    jsonArr.put(record);
    AppendRowsResponse response = streamWriter.append(jsonArr).get();

Exception in thread "main" com.google.cloud.bigquery.storage.v1.Exceptions$AppendSerializationError: INVALID_ARGUMENT: Append serialization failed for writer: projects/myproject-dev/datasets/stage/tables/test_CDC_stream_data_loading/_default
	at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.append(SchemaAwareStreamWriter.java:207)
	at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.append(SchemaAwareStreamWriter.java:109)
	at com.google.cloud.bigquery.storage.v1.JsonStreamWriter.append(JsonStreamWriter.java:62)
	at experiments.TestCDCStreamLoading.main(TestCDCStreamLoading.java:72)

[faisalhasnain]

I am getting same error

[faisalhasnain]

this CDC feature works in python but not in java, any ideas when it will get fixed?

[ismailsimsek]

i believe the error is related to missing field _CHANGE_TYPE in the table schema.

it works as append mode(without failing), when unknown fields set to be ignored .setIgnoreUnknownFields(TRUE)

[Neenu1995]

To use the upsert functionality, it needs to specify the table schema with the _change_type field. You can do this by changing the constructor you use inside the DataWriter.initialize method by using link.

In order to padding the _change_type into the current table schema, the code can be:

final String CHANGE_TYPE_PSEUDO_COLUMN = "_change_type";

    tableSchema.toBuilder()
        .addFields(
            TableFieldSchema.newBuilder()
                .setName(CHANGE_TYPE_PSEUDO_COLUMN)
                .setType(TableFieldSchema.Type.STRING)
                .setMode(Mode.NULLABLE)
                .build())
        .build();
  }

[faisalhasnain]

yeah that's what i did to get it working, thanks for sharing :)

[PhongChuong]

We recently added CDC upsert sample which can be found here:
samples/snippets/src/main/java/com/example/bigquerystorage/JsonWriterStreamCdc.java

[augi]

@Neenu1995 @PhongChuong We are publishing a JSON to PubSub, and the subscription uses the Use Table Schema settings. We have the _CHANGE_TYPE field included in the JSON, but it is still unrecognized.

Does this mean that we should alter the BigQuery table to have the _change_type column?

EDIT: This is not possible.

[augi]

Just for the record, the issue was that the BigQuery table didn´t have PK specified. The component responsible for writing then probably doesn´t expect the _CHANGE_TYPE field.

[AndyCorlin]

Thanks for the tips about primary key and to add the _CHANGE_TYPE to the spec!

I found this code in com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter how to fetch the schema from BigQuery and then used @Neenu1995 's code to add the pseudo column, before creating the JsonStreamWriter.

            String streamName = tableName + "/_default";
            GetWriteStreamRequest writeStreamRequest = GetWriteStreamRequest.newBuilder().setName(streamName).setView(WriteStreamView.FULL).build();
            WriteStream writeStream = this.client.getWriteStream(writeStreamRequest);
            TableSchema tableSchema = writeStream
                    .getTableSchema()
                    .toBuilder()
                    .addFields(
                            TableFieldSchema.newBuilder()
                                    .setName(CHANGE_TYPE_PSEUDO_COLUMN)
                                    .setType(TableFieldSchema.Type.STRING)
                                    .setMode(TableFieldSchema.Mode.NULLABLE)
                                    .build())
                    .build();

            bigquerystorage/latest/com/google/cloud/bigquery/storage/v1/JsonStreamWriter.html
            return JsonStreamWriter.newBuilder(tableName, tableSchema, client)