feat(bigquery): implement watermark column query based replication for bigquery sources by dtunikov · Pull Request #4739 · PeerDB-io/peerdb

dtunikov · 2026-08-27T08:29:30Z

This PR implement the third BigQuery replication mode - query based replication based on user-specified watermark/cursor column (for example created_at).

Added a new field to TableMapping called watermark_column. It's a required field for BigQuery sources with query-based replication mode.
Implemented new pullTableQuery function in biguqery/cdc.go. It fetches data from the table using a simple SQL query like: SELECT ... FROM ... WHERE col > lower AND col <= upper.
Adjusted initial snapshot checkpoint calculation logic. For query-based CDC we use SELECT max(watermark_column) FROM table as a snapshot boundary and store it as an initial CDC checkpoint. So, in this mode we don't use FOR SYSTEM_TIME AS OF TIMESTAMP, instead we query: SELECT ... FROM ... WHERE col < max(watermark_col).

Resolves DBI-1056.

dtunikov · 2026-08-27T08:33:17Z

  BigqueryCdcEventsFunction bigquery_cdc_events_function = 10;
+  // the column to use as a cursor for query-based CDC replication
+  // required if replication_mode is BIGQUERY_REPLICATION_MODE_QUERY
+  string watermark_column = 11;


The plan is to have an option on the UI to configure watermark_column per table + to be able to type in a global watermark_column name (since in most cases the same column name is used for all tables, like created_at).
I think it can be handled completely on the UI and here we just get the column for every configured table. (e.g. we won't need to store global_watermark_column on the flow config lvl)

another thing is that I decided not to add bigquery_ prefix to its name, because it might be used for other query-based CDC connectors in the future.

claude · 2026-08-27T08:48:08Z

Code review

Two issues found. Checked for bugs and CLAUDE.md compliance.

1. `tm` can be `nil` and is dereferenced unconditionally — `flow/connectors/bigquery/cdc.go` L144-L171

tm stays nil when no entry in cfg.TableMappings matches req.SourceTableIdentifier, and every branch below dereferences it with raw field access — tm.WatermarkColumn (L164), tm.BigqueryCdcEventsFunction (L165, L167), and even the "unreachable" error string at L171. Raw field access on a nil protobuf pointer panics; the generated GetX() getters would not.

This is a regression: the pre-PR code read tableMapping.GetBigqueryCdcEventsFunction() into a plain enum value and fell through to the default: (APPENDS) arm when nothing matched.

It is reachable because cfg is re-fetched here from the catalog (internal.FetchConfigFromDB, L139) while req.SourceTableIdentifier comes from the workflow's SyncFlowOptions.TableMappings (flow/activities/flowable_isolated_cdc.go:240). Those two diverge when tables are added mid-mirror — cdc_flow.go appends AdditionalTables to SyncFlowOptions and only refreshes the catalog afterwards, via a best-effort activity that merely logs a warning on failure.

Suggested fix — add an explicit error after the lookup loop:

	if tm == nil {
		return model.PullTableRecordsResult{}, fmt.Errorf("no table mapping found for source table %s", req.SourceTableIdentifier)
	}

2. `TIMESTAMP(<TIMESTAMP column>)` has no matching GoogleSQL signature — `flow/connectors/bigquery/cdc.go` L553-L557

BigQuery's TIMESTAMP() is only defined for TIMESTAMP(string_expression[, tz]), TIMESTAMP(date_expression[, tz]), and TIMESTAMP(datetime_expression[, tz]). There is no identity overload, and GoogleSQL has no implicit coercion from TIMESTAMP to DATETIME/DATE/STRING, so TIMESTAMP(ts_col) fails at analysis time with No matching signature for function TIMESTAMP for argument types: TIMESTAMP.

Meanwhile source.go L86-L91 rejects any watermark column whose type is not bigquery.TimestampFieldType — so the only column type QUERY mode admits is exactly the one these queries cannot wrap. The two halves of the feature disagree; one of them has to change.

The same wrap appears in two more places:

qrep_object_pull.go L417 — SELECT MAX(TIMESTAMP(%s)) AS wm FROM %s
qrep_object_pull.go L511 — WHERE TIMESTAMP(%s) <= TIMESTAMP('%s UTC') (the literal side there is the valid STRING overload; only the column side is the problem)

Corroborating signals that the wrap is unintentional: the ORDER BY on cdc.go L556 uses the raw col while the WHERE wraps it; the pre-existing export code at qrep_object_pull.go L501 uses CAST(x AS TIMESTAMP) rather than TIMESTAMP(x); and the case civil.Date: branch in pullTableQuery is dead code under a TIMESTAMP-only validator.

Suggested fix: drop the TIMESTAMP() wrapper in all three places, since the column is already validated as TIMESTAMP — or, if DATE/DATETIME watermark columns are meant to be supported, relax source.go:88 and apply the conversion conditionally on the field type. Note that qrep_object_pull_test.go:88 currently pins the invalid shape in its expected string and would need updating alongside.

Separately, even with a valid overload, wrapping the watermark column in a function makes the predicate non-sargable and defeats partition pruning on what is typically the partitioning column — a real cost on every CDC poll and on the snapshot export.

github-actions · 2026-08-30T19:36:50Z

⚠️ Deprecated destination connector

This issue or pull request relates to a deprecated destination connector (Snowflake, BigQuery, ElasticSearch, Kafka, Redpanda, Confluent, Azure Event Hubs, Google Pub/Sub, or S3).

These destinations are no longer actively maintained, but remain functional. We are unlikely to prioritize new work here.

Note: BigQuery is deprecated only as a destination — it remains a supported source.

If you depend on one of these connectors, we recommend:

Pin to a known-good PeerDB version so behavior stays stable.
Fork the repository if you need to carry your own changes.

See the deprecated connectors documentation for details and migration guidance.

dtunikov · 2026-08-31T15:57:52Z

+) (map[string]time.Time, error) {
+	checkpointByTable := make(map[string]time.Time, len(cfg.TableMappings))
+	for _, tableMapping := range cfg.TableMappings {
+		watermark, err := c.maxWatermarkValue(ctx, tableMapping.SourceTableIdentifier, tableMapping.GetWatermarkColumn())


We could also use BQ current timestamp as an initial watermark value.
I don't have a strong preference here, both ways should work fine.

dtunikov · 2026-09-01T11:38:13Z

 enum BigqueryCdcEventsFunction {
-  BIGQUERY_CDC_EVENTS_FUNCTION_APPENDS = 0;
-  BIGQUERY_CDC_EVENTS_FUNCTION_CHANGES = 1;
+  BIGQUERY_CDC_EVENTS_FUNCTION_UNSPECIFIED = 0;


since it's not used in production yet - it's safe to update these enums
would make validation more predictable in peerdb and clickpipes

Jeremyyang920 · 2026-09-02T02:07:42Z

+	buildQueryModePullQuery := func(dsTable string, watermarkColumn string, exclude map[string]struct{}) string {
+		col := quotedIdentifier(watermarkColumn)
+		return fmt.Sprintf("SELECT *%s FROM %s WHERE TIMESTAMP(%s) > @start AND TIMESTAMP(%s) <= @end ORDER BY %s",
+			exceptClause(exclude), dsTable, col, col, col)
+	}


nit: Could we maybe extra this to a package level function, then above for runPullQuery we can use a named type instead of the closure definition. Would also ensure that if we ever change the signature, we don't need to update it in multiple places.

will do that in the follow-up PR where I replaced EXCEPT with explicit column selection set (since this code will change a bit there)

github-actions · 2026-09-02T10:57:19Z

❌ Test Failure

Analysis: Not flaky — the PR's newly added replication-mode validation rejects BIGQUERY_REPLICATION_MODE_UNSPECIFIED, deterministically breaking three Test_BigQuery_Source_CDC_Validation subtests (which pass a nil SourceConnectorConfig) identically across all three matrix jobs.
Confidence: 0.96

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

View workflow run

github-actions · 2026-09-03T11:55:58Z

🔄 Flaky Test Detected

Analysis: TestPeerFlowE2ETestSuitePG_CH/Test_CTID_Inherited_Table hit the 60s "UNEXPECTED STATUS TIMEOUT STATUS_SNAPSHOT" poll cap in only the pg16 matrix job while passing on pg17 and pg18 with the same code, indicating a slow initial snapshot under CI load rather than a functional failure.
Confidence: 0.8

✅ Automatically retrying the workflow

View workflow run

github-actions · 2026-09-03T12:20:47Z

🔄 Flaky Test Detected

Analysis: Three TestApiPg subtests on a single matrix leg all hit the same 60s "UNEXPECTED STATUS TIMEOUT STATUS_SNAPSHOT" wait within a 5-second window while the identical wait helper passed in many other tests in the same run and on the other two matrix legs, indicating a transient snapshot/worker stall rather than a code defect.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

- use native bigquery.FieldType in pkg ColumnInfo

Tables without a primary key constraint have TableConstraints == nil, which panicked on tableMeta.TableConstraints.PrimaryKey.Columns.

Adding BIGQUERY_REPLICATION_MODE_UNSPECIFIED = 0 (and BIGQUERY_CDC_EVENTS_FUNCTION_UNSPECIFIED = 0) moved EVENTS and APPENDS off the proto zero value, so Test_BigQuery_Source_CDC_Validation's base config, which relied on those defaults, started tripping the new "invalid replication mode" guard in ValidateMirrorSource. Set the replication mode and CDC events function explicitly on the base config, restore it (rather than nil) after the QUERY subtest, and cover the unspecified-mode rejection directly. Also reject an unset replication mode in ValidateSourceCDC. The flow connector already guards it, but the pkg is meant to be callable from outside the flow module, and there it silently skipped every CDC check. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The regenerated TableMapping interface has a required watermarkColumn field, and TableMapRow derives from it, so both table-mapping object literals in the mirror create handlers stopped type checking. Pass the row's value through in reformattedTableMapping and default it to empty when building rows from the source schema; there is no UI control for it yet. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

github-actions · 2026-09-04T10:14:54Z

🔄 Flaky Test Detected

Analysis: Two MariaDB→ClickHouse e2e tests hit the 60s "UNEXPECTED STATUS TIMEOUT" wait in SetupCDCFlowStatusQuery (mirrors still in STATUS_SETUP/STATUS_SNAPSHOT) within ~22s of each other on one matrix leg, while the other two matrix legs passed on the same commit and the PR only touches BigQuery code — a transient CI resource stall rather than a real bug.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

dtunikov · 2026-09-04T11:35:00Z

  BigqueryCdcEventsFunction bigquery_cdc_events_function = 10;
+  // the column to use as a cursor for query-based CDC replication
+  // required if replication_mode is BIGQUERY_REPLICATION_MODE_QUERY
+  string query_cdc_watermark_column = 11;


no bigquery prefix since it will be used by other query-cdc connectors in the future

dtunikov · 2026-09-04T11:36:00Z

+}
+
+// GetTables returns information about the specified tables
+func GetTables(


if we encounter issues with datasets that have a lot of tables - we can consider using plain SELECT over bigquery system tables.

dtunikov · 2026-09-04T11:36:45Z

 }
+
+// ColumnInfo describes a single column of a BigQuery table.
+type ColumnInfo struct {


tried to move everything related to validation here so that it'd be re-usable from the ClickPipes

for my own understanding: we'll also call this directly from discovery and that's why we moved this here?

dtunikov · 2026-09-04T11:37:55Z

 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )

 func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {


logic was moved to pkg so that it'd be easy to share with the ClickPipes
ValidateMirrorSource just builds bqvalidate.SourceConfig and passed it to pkg functions

itsbilal

Mostly looks good to me, just some clarifying questions. I'll look at the tests next!

itsbilal · 2026-09-04T14:52:06Z

+	from := fmt.Sprintf("%s FOR SYSTEM_TIME AS OF TIMESTAMP('%s UTC')", dsTable.stringQuoted(), boundLiteral)
+	if watermarkColumn != "" {
+		from = fmt.Sprintf("%s WHERE TIMESTAMP(%s) <= TIMESTAMP('%s UTC')",
+			dsTable.stringQuoted(), quotedIdentifier(watermarkColumn), boundLiteral)


watermarkColumn should never be pre-quoted right?

itsbilal · 2026-09-04T15:49:20Z

 }
+
+// ColumnInfo describes a single column of a BigQuery table.
+type ColumnInfo struct {


for my own understanding: we'll also call this directly from discovery and that's why we moved this here?

itsbilal · 2026-09-04T15:58:28Z

-	_, err = it.Next()
-	if err != nil && !errors.Is(err, iterator.Done) {
-		return fmt.Errorf("failed to access staging bucket: %w", exceptions.NewBigQueryError(err))
+	tablesByKey, err := bqvalidate.ValidateSourceTables(ctx, sourceConfig)


nit: maybe we can have a function in bqvalidate that calls both this and ValidateSourceCDC in the CDC case? Feels a little awkward to thread state from one call into another (tablesByKey).

itsbilal · 2026-09-04T19:06:28Z

+// SELECT ... WHERE watermark_column > lower AND watermark_column <= upper scan,
+// rather than APPENDS()/CHANGES(). The initial snapshot is bounded by the
+// watermark column's max value at setup time instead of FOR SYSTEM_TIME AS OF.
+func (s BigQueryClickhouseSuite) Test_BigQuery_CDC_Query_Mode() {


nit: I see this is a convention in this file but usually we don't use underscore in test function names right? It'd just be TestBigQueryCDCQueryMode

dtunikov requested review from a team as code owners August 27, 2026 08:29

dtunikov commented Aug 27, 2026

View reviewed changes

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

dtunikov force-pushed the bq/watermark-column-replication branch from d7a13e0 to 4aa256b Compare August 30, 2026 19:39

dtunikov removed the deprecated label Aug 31, 2026

dtunikov commented Aug 31, 2026

View reviewed changes

dtunikov commented Sep 1, 2026

View reviewed changes

Jeremyyang920 reviewed Sep 2, 2026

View reviewed changes

Comment thread flow/connectors/bigquery/cdc.go Outdated

dtunikov force-pushed the bq/watermark-column-replication branch from 4955cc0 to c0f6c21 Compare September 2, 2026 10:27

dtunikov force-pushed the bq/watermark-column-replication branch 2 times, most recently from 47774c4 to 08a8139 Compare September 3, 2026 11:13

Base automatically changed from bq/isolate-tables-flow to main September 4, 2026 09:36

dtunikov added 11 commits September 4, 2026 11:36

implement watermark column query based replication for bigquery sources

b51ad22

count cdc bytes for watermark column pull mode

fb6a813

move source validation to pkg for reusability

d9e6497

- don't require staging bucket for cdc only BQ mirror

1eed31b

- use native bigquery.FieldType in pkg ColumnInfo

check that watermark column is not excluded

998ac8b

extend source validation

312d51d

fix(bigquery): guard nil TableConstraints in GetTables

254170d

Tables without a primary key constraint have TableConstraints == nil, which panicked on tableMeta.TableConstraints.PrimaryKey.Columns.

introduce unspecified enum value

25bc4f1

add unspecified enum to proto

bd2e571

use %q for identifiers

56d0840

use slices.IndexFunc instead of custom helper func

41f0965

dtunikov and others added 6 commits September 4, 2026 11:36

implement watermark column query based replication for bigquery sources

24340dd

rm tmp file

7cb323a

rename table mapping watermark column to query cdc watermark column

ebb62e9

rename ui refs to watermark col

183fbdc

dtunikov force-pushed the bq/watermark-column-replication branch from 9955a78 to 183fbdc Compare September 4, 2026 09:36

fallback to window start when watermark column is not there

1bd1415

dtunikov commented Sep 4, 2026

View reviewed changes

itsbilal reviewed Sep 4, 2026

View reviewed changes

itsbilal approved these changes Sep 4, 2026

View reviewed changes

Uh oh!

Conversation

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

Uh oh!

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!

claude Bot commented Aug 27, 2026

Code review

1. tm can be nil and is dereferenced unconditionally — flow/connectors/bigquery/cdc.go L144-L171

2. TIMESTAMP(<TIMESTAMP column>) has no matching GoogleSQL signature — flow/connectors/bigquery/cdc.go L553-L557

Uh oh!

github-actions Bot commented Aug 30, 2026

⚠️ Deprecated destination connector

Uh oh!

dtunikov Aug 31, 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!

dtunikov Sep 1, 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!

Uh oh!

github-actions Bot commented Sep 2, 2026

❌ Test Failure

Uh oh!

github-actions Bot commented Sep 3, 2026

🔄 Flaky Test Detected

Uh oh!

github-actions Bot commented Sep 3, 2026

🔄 Flaky Test Detected

Uh oh!

github-actions Bot commented Sep 4, 2026

🔄 Flaky Test Detected

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!

itsbilal left a comment

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dtunikov commented Aug 27, 2026 •

edited

Loading

dtunikov Aug 27, 2026 •

edited

Loading

1. `tm` can be `nil` and is dereferenced unconditionally — `flow/connectors/bigquery/cdc.go` L144-L171

2. `TIMESTAMP(<TIMESTAMP column>)` has no matching GoogleSQL signature — `flow/connectors/bigquery/cdc.go` L553-L557

dtunikov Aug 31, 2026 •

edited

Loading

dtunikov Sep 1, 2026 •

edited

Loading