BigQuery Native Storage Write API CDC Integration for Kafka Connect

# [Feature Proposal]: BigQuery Native Storage Write API CDC Integration for Kafka Connect (kcbq)

## Summary & Objective
Integrate BigQuery's native **Change Data Capture (CDC)** capabilities into the Kafka Connect BigQuery sink connector (`kcbq`). This enables direct, low-latency replication of operational data (e.g., from PostgreSQL via Debezium) into BigQuery using the **Storage Write API**, eliminating the need for expensive, recurring SQL `MERGE` queries on staging changelog tables.

---

## Background & Motivation
Replicating transactional database change streams (inserts, updates, deletes) to BigQuery currently creates an append-only "changelog table". Reconstructing the current state of a table requires executing periodic, complex SQL `MERGE` queries. These query-based merges:
* Consume significant Google Cloud compute resources (BigQuery slot hours).
* Introduce analytical latency (often hours of lag between source DB and analytics).

BigQuery's **Storage Write API** natively supports streaming CDC upserts and deletes directly to target tables with sub-second latency and automatic background merging. This feature brings first-class CDC support to the BigQuery Kafka Connect ecosystem.

---

## Proposed Architecture: Decoupled Transformation Architecture

Instead of coupling `kcbq` directly to proprietary upstream CDC schemas (e.g., Debezium wrapper schemas), we adopt a **Decoupled Transformation Architecture** utilizing standard Kafka Connect Single Message Transforms (SMT):

1. **SMT Flattening**: The Debezium `ExtractNewRecordState` (Unwrap) SMT runs on the sink connector to flatten database events:
* **Inserts & Updates**: Unwrapped into flat payloads matching target BigQuery schemas.
* **Deletes**: Rewritten into payloads containing the previous state of the row with the boolean metadata flag `__deleted = true`.
2. **Sink Connector CDC Routing**:
* Inspects record values and extracts primary key columns from record keys.
* Maps events to `_CHANGE_TYPE` (`UPSERT` or `DELETE`).
* Strips transient SMT metadata (e.g. `__deleted`) before sending rows to the Storage Write API stream.
* Generates deterministic sequence numbers for `_CHANGE_SEQUENCE_NUMBER`.

```
+--------------------+ +-----------------+ +-----------------------------+ +--------------------------+
| Debezium Postgres | ---> | Apache Kafka | ---> | ExtractNewRecordState SMT | ---> | BigQuery Sink Connector | ---> BigQuery Storage
| Source Connector | | (Topic Stream) | | (Flattens & Rewrites) | | (kcbq CDC Writer) | Write API (CDC)
+--------------------+ +-----------------+ +-----------------------------+ +--------------------------+
```

---

## Detailed Design & Key Components

### 1. Dynamic In-Memory Client Schema Injection
* **Challenge**: BigQuery Storage Write API CDC streams require `_CHANGE_TYPE` and `_CHANGE_SEQUENCE_NUMBER` in the stream descriptor (`TableSchema`), but these pseudo-columns **must not** physically exist in the destination BigQuery table.
* **Implementation**: `StorageWriteApiBase` and `StorageWriteApiDefaultStream` dynamically construct an in-memory client `TableSchema` containing `_CHANGE_TYPE` and `_CHANGE_SEQUENCE_NUMBER` when `cdcEnabled = true` without modifying the physical table schema in BigQuery.

### 2. Deletions and Tombstones Handling
* **Tombstone Detection**: If the Kafka record value is `null` (raw tombstone), `_CHANGE_TYPE` is mapped to `DELETE`. Key fields are extracted from the Kafka Connect Record Key, leaving non-key fields null.
* **Rewritten Delete Detection**: If the value payload contains `__deleted: true`, `_CHANGE_TYPE` is mapped to `DELETE` and non-key fields are preserved (allowing extraction of custom sequence values).
* **Upsert Detection**: If neither applies, `_CHANGE_TYPE` is set to `UPSERT`.
* **Metadata Stripping**: The transient `__deleted` field is stripped prior to append, avoiding schema errors.

### 3. Deterministic Sequencing & At-Least-Once Delivery
To ensure idempotent delivery during consumer rebalances or task retries, the connector populates `_CHANGE_SEQUENCE_NUMBER`:
* **Offset Sequencing (Default)**: Uses the 64-bit Kafka partition offset formatted as a 16-character hexadecimal string (`String.format("%016x", record.kafkaOffset())`).
* **Custom Field Sequencing**: Configured via `cdcChangeSequenceNumberField` (e.g., `version`, `updated_at`).
* **Delete Sequence Collision Resolution**: In SQL, `DELETE` statements do not increment sequence columns (the delete event inherits the previous update's version). To prevent collisions:
* For 32-bit values, the sequence number is bit-shifted and combined with the 32-bit Kafka offset: `(seqLong << 32) | (record.kafkaOffset() & 0xFFFFFFFFL)`.
* Alternatively, values can be formatted side-by-side into a 128-bit hex string (`String.format("%016x%016x", customSeq, offset)`).

### 4. Primary Key Management & Schema Relaxation
* **Primary Key Auto-Creation**: When `autoCreateTables = true`, `SchemaManager` extracts primary keys from the Kafka Connect key schema and declares `PRIMARY KEY (...) NOT ENFORCED` via `TableConstraints`.
* **Startup Validation**: If CDC is enabled on an existing table without a primary key, the connector fails fast with a clear actionable message directing the user to run `ALTER TABLE ... ADD PRIMARY KEY (...) NOT ENFORCED`.
* **Schema Relaxation for Deletes**: `SchemaManager` automatically relaxes non-primary-key required fields to `NULLABLE` so deletion records (which omit non-key columns) are ingested without failing `REQUIRED` field constraints.
* **Field Deduplication**: Deduplicates primary key columns when both key and value schemas contain identical field names.

### 5. Table Staleness & DDL Management
* Configurable `tableMaxStaleness` parameter (e.g. `15` seconds) to trigger background merge operations using `ALTER TABLE SET OPTIONS (max_staleness = INTERVAL SECOND)`.
* Employs optimistic caching and concurrent schema update mediation (`mediateConcurrentSchemaUpdates`) to avoid 409 conflict errors across multi-task distributed deployments.

---

## Configuration Walkthrough

### Example Sink Connector Configuration
```json
{
"connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
"tasks.max": "3",
"topics": "cdc_postgres_public_customers",
"datasets": ".*=my_bigquery_dataset",
"autoCreateTables": "true",
"sanitizeFieldNames": "true",

"useStorageWriteApi": "true",
"cdcEnabled": "true",
"cdcChangeSequenceNumberField": "version1",
"tableMaxStaleness": 15,
"mediateConcurrentSchemaUpdates": "true",

"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.drop.tombstones": "false"
}

```
## Summary of Code Changes

| Class | Key Changes |
|---|---|
| `write/storage/StorageWriteApiBase.java` | In-memory `TableSchema` builder for `_CHANGE_TYPE` and `_CHANGE_SEQUENCE_NUMBER`. |
| `write/storage/StorageWriteApiDefaultStream.java` | Supplies augmented `TableSchema` to `JsonStreamWriter`. |
| `SinkRecordConverter.java` | Evaluates `__deleted`/tombstones for `_CHANGE_TYPE`, parses compound hex sequence numbers. |
| `BigQuerySchemaConverter.java` | Filters out `__deleted` column from BigQuery table schema definitions. |
| `SchemaManager.java` | Creates `PRIMARY KEY (...) NOT ENFORCED`, relaxes non-key fields, deduplicates key fields, applies `max_staleness`. |
| `BigQuerySinkConfig.java` | Adds `cdcEnabled`, `cdcChangeSequenceNumberField`, `tableMaxStaleness`, and concurrent schema update configs. |
| `StorageWriteApiValidator.java` & `UpsertDeleteValidator.java` | Allows upsert and delete configs when `useStorageWriteApi = true`. |

## Verification & Testing
* **End-to-End Integration Testing**: Streamed live PostgreSQL mutations via Debezium, confirming real-time replication into BigQuery with zero merge lag.
* **32-Operation Test Suite**: Executed complex sequences including `INSERT`, `UPDATE`, `DELETE`, `RE-INSERT`, and duplicate version delete collisions, verifying exact state matching between PostgreSQL and BigQuery.
* **High-Throughput Load Testing**: Verified high-volume CDC streams with concurrency and duplicate version values with 0 discrepancies.