feat(bigquery): implement watermark column query based replication for bigquery sources

[dtunikov]

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]

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.

[dtunikov]

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]

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.