feat: Add BigQuery append-only history mode

[1neoneo3]

What

Adds a BigQuery append-only history write mode for the BigQuery destination.

This PR depends on #874 for durable destination write stream state. It is currently a draft follow-up PR for #873.

User-facing behavior

This adds a BigQuery write mode that preserves source changes as append-only history rows instead of maintaining only the latest row state.

• current_state remains the default BigQuery write mode.
• append_only creates append-only BigQuery tables.
• Initial copy rows are written as history rows.
• CDC insert/update/delete events are appended as new rows.
• Existing BigQuery current-state behavior is kept unchanged.
• Updating an existing destination from one write_mode to another is rejected, because the physical table schema and write behavior are different.

BigQuery table shape

Append-only tables include the source columns plus flat Supabase ETL metadata columns:

• _etl_uuid
• _etl_source_timestamp
• _etl_change_sequence_number
• _etl_change_type
• _etl_sort_keys

The metadata is intentionally flat rather than a nested datastream_metadata struct so it can be written through the Storage Write API path used by this implementation.

Change event mapping

The append-only mode maps source events to _etl_change_type values as follows:

Source event	_etl_change_type
initial copy	INSERT
insert	INSERT
update old image	UPDATE-DELETE
update new image	UPDATE-INSERT
delete	DELETE

For updates, the destination emits both before and after rows when the old image is available. Key-only old images are expanded into sparse before rows so primary-key changes can still be represented in history.

Write path and retry behavior

Append-only writes use BigQuery Storage Write API committed streams.

The implementation uses the durable stream state added in #874 to remember:

• the BigQuery committed stream name
• the next append offset
• the associated destination table id
• the latest durable append-only sequence watermark

That lets retries and process restarts resume from the stored offset instead of generating a new stream position and duplicating rows.

The stored stream state is monotonic by committed-stream offset. Stale writers cannot move the offset backwards, and same-offset updates cannot move the sequence watermark backwards. Initial-copy sequence values are stable row identities rather than globally increasing CDC sequence values, so offset advancement is allowed even when the next copied row's sequence sorts lower than the previous copied row. Schema changes rotate to a fresh committed stream while preserving the sequence watermark. This keeps replay filtering intact if the process crashes around a schema change.

For BigQuery ALREADY_EXISTS responses, the committed stream path only treats the response as success for a retried append request. A first-attempt overlap is still returned as an error because batch boundaries may not match a previous append.

Ordering and initial copy

Initial copy rows are emitted as INSERT rows.

For initial copy rows, the implementation builds a stable row identity from the source row and stores it in _etl_uuid / _etl_change_sequence_number using a reserved copy sequence prefix plus the row hash. This makes the metadata deterministic for the same copied row, which is important if the copy has to be retried.

Initial copy rows also get a dedicated low sort key prefix so they sort before CDC rows for the same source table, even when the initial copy timestamp is newer than an already captured CDC timestamp.

For example, a copied row and a later CDC row would sort like this:

Row source	_etl_change_type	_etl_change_sequence_number example	_etl_sort_keys example
initial copy	INSERT	0000000000000000/0000000000000000/<row-hash>	["0000000000000000", "0000000000000000/0000000000000000/<row-hash>", "copy/<sequence>/<row-hash>"]
CDC insert/update/delete	INSERT / UPDATE-* / DELETE	0000000000000001/0000000000000002/0000000000000003	["800000000000002a", "0000000000000001/0000000000000002", "0000000000000003"]

The CDC sequence number is derived from the event sequence key and an internal ordinal. That gives consumers a deterministic replay order: first sort by _etl_sort_keys, then interpret _etl_change_type to reconstruct history.

Schema changes

Append-only tables can evolve with source schema changes.

• New source columns are added to the append-only table.
• Renamed / removed source columns are preserved as historical columns where needed.
• Append-only tables do not require a source primary key in the same way current-state tables do, because rows are appended rather than merged into a current-state table.

PR split

This PR is split with pattern 1: bottom-up.

• feat: Persist destination write stream state #874: durable destination write stream state foundation
• feat: Add BigQuery append-only history mode #875: BigQuery append-only mode built on that foundation

Because this is a fork-to-upstream PR, GitHub cannot use the fork branch from #874 as the upstream base branch. Until #874 is merged, the normal GitHub diff for this PR still includes the #874 foundation commit.

For a focused review of only the append-only slice, use this compare view:

1neoneo3/etl@feature/bigquery-append-only-history...feature/bigquery-write-stream-state

After #874 is merged, this PR can be rebased onto the updated main, and the normal GitHub diff will shrink to the append-only implementation only.

Out of scope / follow-ups

• Per-table append-only selection is not implemented in this PR.
• Pattern-based table write mode selection is not implemented in this PR.
• Switching an existing BigQuery table in place from current_state to append_only is intentionally not supported.
• This PR does not try to exactly reproduce Datastream's nested metadata layout; it preserves the append-only semantics using Supabase ETL _etl_* metadata columns.

Tests

• cargo check -p etl --features test-utils
• cargo check -p etl-destinations --features test-utils,bigquery
• cargo check -p etl-replicator
• cargo test -p etl --features test-utils memory_store_destination_write_stream_state_is_monotonic --test main -- --nocapture
• cargo test -p etl --features test-utils notifying_store_destination_write_stream_state_is_monotonic --test main -- --nocapture
• cargo test -p etl --features test-utils state_store_destination_write_stream_state_persists_and_clears --test main -- --nocapture
• cargo test -p etl-destinations --features test-utils,bigquery append_only_rows_after_sequence -- --nocapture
• cargo test -p etl-destinations --features test-utils,bigquery committed_stream_already_exists_only_succeeds_on_retry_attempts -- --nocapture
• cargo test -p etl-destinations --features test-utils,bigquery committed_stream_max_inflight_uses_connection_pool_size_as_proactive_limit -- --nocapture
• cargo test -p etl-destinations --features bigquery,test-utils append_only -- --nocapture
• cargo test -p etl-destinations --features bigquery,test-utils committed_stream -- --nocapture
• cargo test -p etl-destinations --features bigquery committed_stream_ --lib
• cargo test -p etl write_stream_state_ --lib
• cargo test -p etl-destinations --features bigquery,test-utils append_only_copy_rows_after_sequence --lib
• git diff --no-ext-diff --check

Local BigQuery smoke validation was also run without adding the smoke test source to this PR, to keep the review diff focused. This used a real BigQuery project and local test Postgres:

• env TESTS_BIGQUERY_PROJECT_ID=oripaone TESTS_BIGQUERY_SA_KEY_PATH=/home/mk/.config/gcp/automation-bigquery.json REQUIRE_BIGQUERY_CREDENTIALS=true cargo test -p etl-destinations --features bigquery,test-utils local_append_only_smoke --test main -- --ignored --nocapture

The latest local smoke run passed 11 ignored tests. Under those conditions, the following behavior has been verified:

• Initial copy rows are written as append-only INSERT rows.
• CDC insert/update/delete rows are written with deterministic append-only metadata.
• Replaying already durable initial-copy and CDC batches does not duplicate PK-table rows.
• A destination restart can resume from durable committed stream state and continue appending later CDC rows.
• A table added to an existing publication is initial-copied after pipeline restart, and later CDC rows for that table append successfully.
• A source schema ADD COLUMN can be followed by a committed stream append.
• A source schema DROP COLUMN preserves the historical BigQuery column and future appends still succeed.
• A source schema RENAME COLUMN preserves the old historical BigQuery column, adds the new column, and future appends still succeed.
• JSONB, timestamptz, numeric, arrays, and NaN float values append successfully through the append-only Storage Write path.
• Initial copy rows populate _etl_source_timestamp.
• No-primary-key duplicate initial-copy rows are preserved as distinct history rows.
• Multi-table append-only initial copy, CDC append, and replay deduplication work across separate tables.
• WAL retention smoke also passed outside the committed test source:
  • 1,000,000 initial-copy rows plus 100,000 CDC insert rows generated while table copy was running; BigQuery row count reached 1,100,000 in 176.53s, max observed retained WAL was 15,420,960 bytes, max observed unconfirmed WAL was 15,420,904 bytes, and no replication slot reached lost status.
  • 1,000,000 initial-copy rows plus 1,000,000 CDC insert rows generated while table copy was running; BigQuery row count reached 2,000,000 in 228.63s, max observed retained WAL was 150,197,736 bytes, max observed unconfirmed WAL was 150,197,680 bytes, and no replication slot reached lost status.
  • 100,000,000 initial-copy rows plus 1,000,000 CDC insert rows generated while table copy was running; the first large runs exposed missing committed-stream retry classification for CANCELLED and UNKNOWN, which this PR now treats as retryable transport uncertainty together with DEADLINE_EXCEEDED; after that fix, BigQuery row count reached 101,000,000 in 13,804.76s, max observed retained WAL was 6,508,233,144 bytes, max observed unconfirmed WAL was 6,508,233,088 bytes, and no replication slot reached lost status.
• Larger append-only table copies also passed outside the committed test source, using the same local smoke harness with row-count-only BigQuery verification:
  • 10,000 initial-copy rows + 1,000 CDC insert rows
  • 100,000 initial-copy rows + 1,000 CDC insert rows
  • 1,000,000 initial-copy rows + 1,000 CDC insert rows
  • 3,000,000 initial-copy rows + 1,000 CDC insert rows
  • 10,000,000 initial-copy rows + 1,000 CDC insert rows; initial copy count verified in 1411.83s and CDC count verified in 3.65s

The following existing BigQuery current-state regression tests were also run to check that this PR does not break the default mode:

• env TESTS_BIGQUERY_PROJECT_ID=oripaone TESTS_BIGQUERY_SA_KEY_PATH=/home/mk/.config/gcp/automation-bigquery.json REQUIRE_BIGQUERY_CREDENTIALS=true cargo test -p etl-destinations --features bigquery,test-utils table_insert_update_delete --test main -- --nocapture
• env TESTS_BIGQUERY_PROJECT_ID=oripaone TESTS_BIGQUERY_SA_KEY_PATH=/home/mk/.config/gcp/automation-bigquery.json REQUIRE_BIGQUERY_CREDENTIALS=true cargo test -p etl-destinations --features bigquery,test-utils table_truncate_with_batching --test main -- --nocapture
• env TESTS_BIGQUERY_PROJECT_ID=oripaone TESTS_BIGQUERY_SA_KEY_PATH=/home/mk/.config/gcp/automation-bigquery.json REQUIRE_BIGQUERY_CREDENTIALS=true cargo test -p etl-destinations --features bigquery,test-utils schema_change_add_column_defaults --test main -- --nocapture
• env TESTS_BIGQUERY_PROJECT_ID=oripaone TESTS_BIGQUERY_SA_KEY_PATH=/home/mk/.config/gcp/automation-bigquery.json REQUIRE_BIGQUERY_CREDENTIALS=true cargo test -p etl-destinations --features bigquery,test-utils table_schema_change --test main -- --nocapture

Refs #873

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 494b203d33

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 22c5ad991a

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[iambriccardo]

Thanks for the PR! This is a welcome addition but given the complexity, we are not looking to add this at this point in time.

[1neoneo3]

Thanks for the review and the feedback! I understand the reasoning. I'll close this PR. Thanks again for taking the time to review it.