feat(bigquery): implement appends/changes event based cdc modes by dtunikov · Pull Request #4707 · PeerDB-io/peerdb

dtunikov · 2026-08-19T13:30:07Z

What

This PR makes BigQuery a valid source connector.
Mainly it implements PullRecords method and BigQuery CDC mirror validation. (check if ordering key is set for cdc mirrors, etc)
It supports two change capture modes:

APPENDS - available for all BQ tables with time-travel enabled (7d window by default). Captures only INSERTs.
CHANGES - available only when enable_change_history option is set for the table. Captures INSERTs/UPDATEs/DELETEs.

The new replication mode setting was added to BigQuery mirrors:

enum BigQueryReplicationMode {
  // Default: table-valued function (APPENDS()/CHANGES()) based CDC polling.
  BIGQUERY_REPLICATION_MODE_EVENTS = 0;
  // query based: standard "SELECT ... FROM table WHERE col > ? AND col <= ?"
  BIGQUERY_REPLICATION_MODE_QUERY = 1;
}

and the new BQ specific setting was added to the table mapping struct (only relevant for BIGQUERY_REPLICATION_MODE_EVENTS replication mode):

enum BigqueryCdcEventsFunction {
  BIGQUERY_CDC_EVENTS_FUNCTION_APPENDS = 0;
  BIGQUERY_CDC_EVENTS_FUNCTION_CHANGES = 1;
}

Important files for review

flow/connectors/bigquery/cdc.go - core BQ cdc logic
flow/connectors/bigquery/source.go - changes to mirror validation
flow/connectors/bigquery/qrep_object_pull.go - move checkpoint by 1ms forward to ensure that we don't have an overlap with the snapshot
flow/e2e/bigquery_cdc_test.go - e2e test cases

What is not included in this PR

changes to the PeerDB UI to enable BigQuery CDC feature yet.
error classification, need to explore it a bit more and don't want to overload this PR
table isolation (if tableA facing errors - we should continue cdc for other pipe's tables) + parallel table processing

Resolves: DBI-1038

Capture T (BigQuery's own CURRENT_TIMESTAMP(), not local wall-clock) once per ExportTxSnapshot call and append FOR SYSTEM_TIME AS OF TIMESTAMP('<T> UTC') to every table's EXPORT DATA statement, so all tables in a snapshot read a consistent point in time. Persist T as the initial CDC checkpoint via SetLastOffset, but from ExportTxSnapshot itself rather than SetupReplication as the plan originally sketched: SnapshotFlowWorkflow only calls ExportTxSnapshot when InitialSnapshotOnly is true, and only calls SetupReplication when it's false (cloneTablesWithSlot, or the no-snapshot CDC-only branch) - the two are mutually exclusive per run, so T is never in scope inside SetupReplication. Gating the write on !InitialSnapshotOnly here is therefore a forward-looking no-op until a later chunk changes how BigQuery mirrors that continue into CDC get their initial load wired up; documented in code at the capture site. Refactored the SQL-building half of bigQueryExportQueryStatement into a pure buildBigQueryExportSQL taking an already-resolved schema, so it's unit testable without a live BigQuery client (mirrors the existing bigQuerySchemaToQRecordSchema/datasetTable test patterns in this package).

…t code ExportTxSnapshot only ever runs on SnapshotFlowWorkflow's pure snapshot-only branch (InitialSnapshotOnly && DoInitialSnapshot) - the continue-to-CDC and CDC-only branches call SetupReplication instead and never touch ExportTxSnapshot. So the checkpoint-persist code added there, gated on !InitialSnapshotOnly, could never actually run; removed. SetupReplication now does what MySQL's does: capture a starting position (BigQuery's current timestamp T, no replication slot to open) and persist it as the initial CDC checkpoint. Unlike MySQL, BigQuery's initial load reads pre-exported Parquet from GCS rather than querying the live table, so when the mirror wants an initial load (req.DoInitialSnapshot), this also runs that export as of T - the per-table export-job loop is factored out of ExportTxSnapshot into a shared exportTablesAsOf helper so both callers use it. Returns a zero-value SetupReplicationResult (no slot/snapshot name), same as MySQL: cloneTablesWithSlot falls back to the mirror's configured SnapshotStagingPath when no override is given, which is exactly where this export writes.

…bject_pull_test.go

…(tab-width 4)

Newline/tab padding made the query harder to read; collapse to single-line format string.

Introduces a connector-specific mirror-config extension point on FlowConnectionConfigs/FlowConnectionConfigsCore, mirroring how Peer already does this for peer-level config. BigqueryCdcConfig (with a cdc_mode of APPENDS or CHANGES) is the first variant. No converter work is needed: flow/proto_conversions copies between the two messages by field number, so a oneof - whose wrapper types are scoped per parent message - is carried over without special-casing. No behavior change yet - BigqueryCdcConfig is not read anywhere. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

start_peer_flow_job builds FlowConnectionConfigs as a fully exhaustive struct literal (no ..Default::default()), so it needs every field set explicitly. The new source_connector_config oneof broke this build; nexus doesn't set any connector-specific mirror config today, so None is correct here.

Copying by field number handles the oneof without special-casing, but the wrapper types are scoped per parent message, so assert the variant is rebuilt against the destination's own type in both directions - and that an unset oneof stays unset rather than spuriously populating a variant. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…ecting Snapshot-only mirrors skip these checks (mirrors MySQL's ValidateMirrorSource pattern). CHANGES mode needs a real PK constraint plus enable_change_history on the source table; APPENDS mode needs an explicit MergeTree destination engine when there's no PK, since ORDER BY tuple() on a keyless ReplacingMergeTree collapses the table on writes.

…GE_TREE CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE is the same collapsing dedup engine as the plain variant, just wrapped for replication (see how normalize.go's engine switch groups the two under one case) - a keyless table hits the same ORDER BY tuple() collapse either way, so the APPENDS-mode keyless-engine check needs to catch both.

Gives BigQueryConnector real CDCPullConnectorCore bodies (SetupReplConn, UpdateReplStateLastOffset, PullFlowCleanup, EnsurePullability) and a real PullRecords: self-paced polling of APPENDS() per mapped table over (checkpoint, upper], converting rows to InsertRecords via a new BigQuery value -> QValue converter, advancing the checkpoint text once the window closes. No delete/insert pairing yet -- that's CHANGES mode, next chunk.

CHANGES() reports an UPDATE as a delete+insert pair sharing the same PK and _CHANGE_TIMESTAMP, so PullRecords needs to pair those back into one UpdateRecord instead of emitting two records -- otherwise every update on a CHANGES-mode mirror would be replicated as a delete followed by an unrelated insert. Reads cdc_mode once per PullRecords call (it's a per-mirror setting) to pick APPENDS or CHANGES per the "Decisions locked in" plan.

…e source

- cleanup comments

…CHANGES bq functions

…st does not exist" error

Rewrites the now-stale Test_BigQuery_Source_CDC_Not_Supported, which asserted CDC gets rejected - chunk 3 replaced that with mode-specific validation. Adds e2e coverage for snapshot->CDC handoff, APPENDS insert-only polling, CHANGES insert/update/delete pairing, and pause/resume from the persisted checkpoint; none of this has been run against a live BigQuery instance yet.

- cleanup comments

bigquery.go holds the shared BigQueryConnector struct used by both source and destination code, so source-only changes there were wrongly picked up by the deprecated-connector labeler.

jgao54 · 2026-08-22T21:32:00Z

+			return nil, err
+		}
+
+		lastCheckpoint := recordBatchPull.GetLastCheckpoint()


we also periodically persist/advance on an idle stream for other connectors, but logic is implemented inside PullRecord since empty batch stays open until at least one record is received, so for consistency this could be moved to inside PullRecord.

…records in the window

ilidemi · 2026-08-24T19:15:34Z

+	c.logger.Info("[bigquery] PullRecords polled window",
+		slog.Time("start", checkpoint), slog.Time("end", upper),
+		slog.Int("records", recordCount), slog.Int64("bytes", bytesProcessed))


Nit: please add channelLen (to have telemetry for bottlenecks) and log end-start as elapsedMinutes float. slog.Time logs nanoseconds so it's less intuitive to skim when reading logs

Haven't looked at the further PRs yet, but would be cool to also report records/bytes/channelLen in ongoing manner like in other connectors. This helps investigate tickets about a single batch taking long and OOMs

oh yeah, I'll make sure we have all necessary logs in the follow-up PR
this PullRecords function is removed anyways in the follow-up in favor of PullTableRecords for isolated tables CDC

ilidemi · 2026-08-24T19:47:47Z

+	// droppedExcludeColumns remembers, per source table, excluded columns that
+	// BigQuery has reported as no longer existing, so later polls stop asking BigQuery
+	// to EXCEPT them.
+	droppedExcludeColumns map[string]map[string]struct{}


Keep in mind the columns can come back too, so client-side filtering would be needed as well.

Also, maybe more of a product note: excluding columns in a way that we don't even see the data was an often requested feature in other source types, so EXCEPT here works great, but seems if we advertise it we'd need to caveat that it only works as long as the column is already present at the first batch and doesn't disappear-reappear. If it becomes an important first-party scenario, maybe could have some option to always query the known columns and not support column adds. Couldn't think of a way to have SELECT * EXCEPT fully work here, as even if we always do the elimination loop, an excluded column could get re-added in between retries and we'd receive the bytes.

ilidemi · 2026-08-24T20:07:15Z

+	bigquery.TimestampFieldType:  {types.QValueKindTimestamp, types.QValueKindArrayTimestamp},
+	bigquery.DateTimeFieldType:   {types.QValueKindTimestamp, types.QValueKindArrayTimestamp},
+	bigquery.DateFieldType:       {types.QValueKindDate, types.QValueKindArrayDate},
+	bigquery.TimeFieldType:       {types.QValueKindTime, types.QValueKindArrayString},


Seems to be a miss that we don't have an QValueKindArrayTime. Mind adding it for BQ and creating a ticket for other connectors?

2f93adc

created a ticket for other connectors

ilidemi · 2026-08-24T20:15:09Z

+	if fieldSchema.Type == bigquery.RecordFieldType {
+		// Preserve field names and values as JSON text in STRING or ARRAY<STRING> values.


Did we clear it with Product that JSON strings are desired here? BQ records are typed, CH has Nested, so seems like our internal bottleneck

Tuple would be the best typed one, you're right

changed to native JSON for Struct and ArrayString for repeated Struct (since we don't support Array(JSON) yet in peerdb)

ilidemi · 2026-08-24T20:27:06Z

+	return fmt.Sprintf("[%s, %s)", bigQueryRangeBoundString(value.Start), bigQueryRangeBoundString(value.End))
+}
+
+func bigQueryRangeBoundString(value bigquery.Value) string {


Please add a subissue to DBI-295 or augment DBI-296 so we pretty it up at the same time

ilidemi · 2026-08-24T20:50:02Z

 		TargetForSetting: protos.DynconfTarget_BIGQUERY,
 	},
+	{
+		Name: "PEERDB_BIGQUERY_CDC_SAFETY_LAG_SECONDS",


Seems there was a 10 min restriction for CHANGES that got removed with GA: https://www.linkedin.com/feed/update/urn:li:activity:7488597379827445760/

For appends though, the outcome of underestimating this would be silently missing rows, right? Would it make sense to make it user-configurable somewhere in advanced settings? Also wonder if there's any telemetry we could insert to track this, even at a higher cost during private preview

Edit: reading further, seems CHANGES also doesn't document any guarantees wrt end_timestamp coverage, just that you can pass null and not get an error

jgao54 · 2026-08-24T21:09:10Z

ah, didn't see the new review from @ilidemi coming in, just approving what I have reviewed so far. Please make sure to address the rest.

dtunikov · 2026-08-26T15:28:07Z

@ilidemi regarding this
Yep, I've left a comment in the next PR about the same thing:
https://github.com/PeerDB-io/peerdb/pull/4727/changes#r3862432408
I agree that it'd make sense to make safety_lag/max_query_window (ofc with some reasonable default value)
As I mentioned in the comment above, we could even have this logic available for all future query-based sources. I think this might pop up for other DWHs as well.

…-records

dtunikov · 2026-08-27T14:18:11Z

@ilidemi regarding EXCEPT.
I put a PR to select columns explicitly instead of relying on * with EXCEPT.
#4740

github-actions · 2026-08-27T15:51:56Z

🔄 Flaky Test Detected

Analysis: Flaky infrastructure failure: the pure-unit test TestRunPipeline_FilterStripsLines failed because its printf child process was SIGKILLed externally (consistent with OOM/resource pressure from go test -p 32 alongside the full Tilt docker stack), not from any assertion mismatch — it passed on the other two matrix legs and is unrelated to the PR's BigQuery-CDC changes.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

- align qrep qField conversion with cdc path

github-actions · 2026-08-28T10:11:26Z

❌ Test Failure

Analysis: Not flaky: all three matrix legs fail identically because PR #4707 changes the BigQuery TIMESTAMP mapping from QValueKindTimestamp to QValueKindTimestampTZ without updating the trips1kExpectedQValueColumns expectations in flow/e2e/bigquery_source_test.go.
Confidence: 0.95

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

dtunikov and others added 30 commits August 17, 2026 11:21

cleanup comments

3083f69

make unit test readable

4489f07

fix(bigquery): satisfy gci import order and lll line length in qrep_o…

d30fff6

…bject_pull_test.go

fix(bigquery): split remaining long expected literals to satisfy lll …

0462a5a

…(tab-width 4)

fix(bigquery): drop whitespace fillers from EXPORT DATA SQL

a0eef66

Newline/tab padding made the query harder to read; collapse to single-line format string.

move validation to pkg

5d8abe2

bring back InitialSnapshotOnly validation to big query connector

4e96624

update last checkpoint even if we didn't receieve any records from th…

298d69e

…e source

allow to create cdc big query mirrors

56dfa0d

- make query window and safety lag configurable via dynamic envs

631bcfc

- cleanup comments

start_timestamp is inclusive, end_timestamp is exclusive for APPENDS/…

80c55ac

…CHANGES bq functions

count fetched bytes for billing using len(json.Marshal(values))

c7b8c6a

cleanup comments

fb84cfa

skip DELETE events that predate upcoming UPDATE

4f8f5a2

exclude columns using bigquery EXCEPT keyword

bd00ce4

cache dropped excluded columns when BQ returns "in SELECT * EXCEPT li…

6884cd1

…st does not exist" error

fix missing column extraction logic for removed excluded columns

40d46a5

- fix numeric values without scale

cd7e61d

- cleanup comments

extend type system tests

8e87427

fix(labeler): stop auto-labeling bigquery.go as deprecated

9232fb0

bigquery.go holds the shared BigQueryConnector struct used by both source and destination code, so source-only changes there were wrongly picked up by the deprecated-connector labeler.

jgao54 reviewed Aug 22, 2026

View reviewed changes

dtunikov added 3 commits August 24, 2026 09:08

fix TargetForSetting, bq -> ch

c2ed196

simplify BigQueryTypeToQValueKind code

f1715fc

move checkpoint forward from the PullRecords itself when there is no …

c61e837

…records in the window

ilidemi reviewed Aug 24, 2026

View reviewed changes

jgao54 approved these changes Aug 24, 2026

View reviewed changes

dtunikov added 2 commits August 27, 2026 11:22

Merge branch 'main' of github.com:PeerDB-io/peerdb into bq-cdc/4-pull…

9e5b729

…-records

use window.start as commitTimeNano when changeTimestamp is missing

cb94b63

dtunikov added 2 commits August 27, 2026 16:44

handle bigQueryChangeTypeDelete explicitly

16cd32a

add QValueArrayTime type for timestamps without a date (e.g. 15:00:00)

2f93adc

github-actions Bot added the deprecated label Aug 27, 2026

remove destination validation from big query source validation

51807ce

dtunikov removed the deprecated label Aug 27, 2026

use native JSON for BigQuery Struct type

1a196e2

github-actions Bot added the deprecated label Aug 27, 2026

dtunikov removed the deprecated label Aug 27, 2026

ilidemi approved these changes Aug 28, 2026

View reviewed changes

- map timestamp to timestamptz

6a38543

- align qrep qField conversion with cdc path

github-actions Bot added the deprecated label Aug 28, 2026

cleanup comments

f5b6366

dtunikov removed the deprecated label Aug 28, 2026

fix tests for timetamptz

e82ff46

github-actions Bot added the deprecated label Aug 28, 2026

dtunikov merged commit 4d7b5c0 into main Aug 28, 2026
30 of 31 checks passed

dtunikov deleted the bq-cdc/4-pull-records branch August 28, 2026 11:15

		if fieldSchema.Type == bigquery.RecordFieldType {
		// Preserve field names and values as JSON text in STRING or ARRAY<STRING> values.

Uh oh!

Conversation

dtunikov commented Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Important files for review

What is not included in this PR

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dtunikov Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ilidemi Aug 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jgao54 commented Aug 24, 2026

Uh oh!

dtunikov commented Aug 26, 2026

Uh oh!

dtunikov commented Aug 27, 2026

Uh oh!

github-actions Bot commented Aug 27, 2026

🔄 Flaky Test Detected

Uh oh!

github-actions Bot commented Aug 28, 2026

❌ Test Failure

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

dtunikov commented Aug 19, 2026 •

edited

Loading

dtunikov Aug 27, 2026 •

edited

Loading

ilidemi Aug 24, 2026 •

edited

Loading