feat(bigquery): implement a separate SyncFlow branch for query-based CDC

[dtunikov]

Problem

We need to support CDC for DWH sources such as BigQuery/Snowflake. These sources don't have global wal/binlog/change-stream concept. There is no channel we can listen to for events (inserts, updates, deletes).
For such sources we are going to implement query-based replication. Which, essentially, just queries the tables every SyncInterval seconds, e.g. runs similar query every N secs:
SELECT ... FROM table WHERE col > lower AND col <= upper

We could keep using existing SyncFlow PullRecords function to query the source. But it means that all mirror's tables would be coupled:

• one CDC records stream for all tables
• one table's failure blocks other tables replication
• one table can slow down replication of other tables (if the source is slow or a downstream MV is slow)

Solution

This PR introduces a new SyncFlow branch for query-based CDC sources (for now just BigQuery).

It implements the following requirements:

• Back-pressure must be applied per table
• Tables must be queried in parallel (a single slow table can't affect others)
• Errors must be reported via LogFlowError, but shouldn't block other tables
• No raw table hop. Batches are ingested directly from s3/gcs into the final destination table

Currently, only BigQuery source connector supports new interfaces. In the future this workflow can be used for other query-based CDC connectors as well (e.g. other data warehouses or even OTLP databases in cases when a customer doesn't want to rely on wal/binlog replication).

Tests

Added e2e tests that cover the following scenarios:

• single table's failure doesn't block other tables
• back-pressure kicks in only for the affected table, so it doesn't block sibling tables
• run cdc, pause pipe, remove table from the mirror, unpause -> replication still works

Workflow diagram

flowchart TD
    SF["SyncFlow activity<br/>(flowable.go)"]
    SF -->|isIsolatedTableCDCPath?<br/>BQ source + ClickHouse dest| DEC{branch}
    DEC -->|false| OLD["pullAndSync / pullAndSyncPg<br/>+ shared CDCStream<br/>+ normalizeLoop (unchanged path)"]
    DEC -->|true| SFI["syncFlowIsolatedTables<br/>(flowable_isolated_cdc.go)"]

    SFI --> SETUP["setup: idleTimeout, channelBufferSize,<br/>pullSem(parallelism), normBufferSize,<br/>PruneTableReplicationState, batchIDCounter"]
    SETUP --> SPAWN["for each TableMapping:<br/>spawn 1 sync loop + 1 normalize loop<br/>paired via per-table LastChan(normRequests, normResponses)"]

    subgraph PERTABLE["per source table (independent goroutines)"]
        direction TB

        subgraph SYNCLOOP["isolatedTablePullSyncLoop"]
            direction TB
            S1["GetTableReplicationState"] --> S2{"synced - normalized<br/>>= normBufferSize?"}
            S2 -->|yes: backpressured| S2W["wait on normResponses<br/>(this table only)"] --> S1
            S2 -->|no| S3{"poll due?<br/>(idleTimeout)"}
            S3 -->|not yet| S3W["waitOrDone"] --> S1
            S3 -->|due| S4["acquire pullSem"]
            S4 --> S5["RecordTableReplicationAttempt"]
            S5 --> S6["errgroup:<br/>PullTableRecords -> TableCDCStream<br/>SyncTableCDC drains stream"]
            S6 --> S7["release pullSem"]
            S7 --> S8{"rows > 0?"}
            S8 -->|no| S9["RecordTableReplicationSync<br/>(cursor only)"] --> S1
            S8 -->|yes| S10["RecordTableReplicationSync<br/>(cursor + synced_batch_id++)"]
            S10 --> S11["recordIsolatedTableBatch<br/>(cdc_batches/cdc_flows monitoring)"]
            S11 --> S12["normRequests.Update(batchID)"]
            S12 --> S1
        end

        subgraph NORMLOOP["isolatedTableNormalizeLoop"]
            direction TB
            N1["resume: normResponses.Update(normalizedBatchID);<br/>if synced>normalized, normRequests.Update(synced)"] --> N2{"normRequests > normResponses?"}
            N2 -->|no| N2W["wait on normRequests"] --> N2
            N2 -->|yes| N3["NormalizeTableCDC:<br/>for batch in (lastNorm, reqBatch]:<br/>GetTableAvroStage -> INSERT...SELECT<br/>into final table -> DeleteTableAvroStage"]
            N3 --> N4{"error?"}
            N4 -->|yes| N4B["alert once + backoff"] --> N2
            N4 -->|no| N5["RecordTableReplicationNormalize"]
            N5 --> N6["normResponses.Update(batchID)<br/>(releases sync loop's backpressure wait)"]
            N6 --> N2
        end

        S6 -. "PullTableRecords" .-> BQ[("BigQuery<br/>connector")]
        S6 -. "SyncTableCDC:<br/>typed Avro -> S3/GCS<br/>+ SetTableAvroStage" .-> CH1[("ClickHouse<br/>connector")]
        N3 -. "NormalizeTableCDC" .-> CH2[("ClickHouse<br/>final table")]

        S12 -. signal .-> N2
        N6 -. signal .-> S2
    end

    SPAWN --> SYNCLOOP
    SPAWN --> NORMLOOP

    CAT1[("catalog:<br/>cdc_table_replication_state<br/>(cursor_text, last_attempt_at,<br/>synced_batch_id, normalized_batch_id)")]
    CAT2[("catalog:<br/>cdc_table_avro_stage<br/>(flow, table, batch_id -> avro_file)")]
    S1 -.-> CAT1
    S10 -.-> CAT1
    N1 -.-> CAT1
    N5 -.-> CAT1
    S6 -.-> CAT2
    N3 -.-> CAT2

Loading

Resolves: DBI-1045

[github-actions]

🔄 Flaky Test Detected

Analysis: The flow/e2e package was killed by the global 20m -timeout 1200s cap (no test hung — all in-flight tests were only 9–47s old, and there were zero assertion failures), while the other two matrix legs passed on the same commit at 933s/998s, showing this slowest leg simply lacks headroom under the timeout.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

[jgao54]

general style nit: there's a mixed terminology usage for "isolated cdc" and "query cdc" (as well as both "cdc table" and "table cdc") across variable name, method name, table definitions, etc. would be good to introduce a single terminology/concept for the query-based cdc

[jgao54]

Looks great overall.

[jgao54]

the catalog migration error is expected, i forgot to mention that they just needed to be added to goose https://github.com/PeerDB-io/peerdb/blob/e9d5da54c91c0a5dca73205b120a101129908124/flow/db/README.md

i plan to remove the refinery migration scripts after the next release (or at least freeze them in some way) so going forward we'd only need to modify goose-side; but this is dependent on a release

[github-actions]

❌ Test Failure

Analysis: Not flaky — PR #4727 adds migration 56 to flow/db/migrations without the matching nexus/catalog/migrations file, deterministically failing TestMigrationVersions and TestGooseBootstrapFromRefinery on all three matrix legs, plus its own new BigQuery backpressure e2e test times out on the assertion for the behavior it introduces.
Confidence: 0.95

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

View workflow run