WIP insert multiple batches at once from s3/gcs to clickhouse - #4732
Draft
dtunikov wants to merge 79 commits into
Draft
WIP insert multiple batches at once from s3/gcs to clickhouse#4732dtunikov wants to merge 79 commits into
dtunikov wants to merge 79 commits into
Conversation
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
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.
- 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.
…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.
- move cdc mode function on table mappings level - fix e2e tests
Fixes TS build error after BigqueryCdcEventsFunction field was added to TableMapping.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
## Summary Add ClickHouse error code **517 (`CANNOT_ASSIGN_ALTER`)** to `retryableExceptions` in `flow/pkg/clickhouse/query_retry.go`, so schema-delta `ALTER TABLE ... ADD COLUMN` statements retry with the existing backoff instead of failing the whole sync batch. ## The bug When the destination table is a `Replicated*MergeTree` (self-hosted CH with Keeper), `ReplayTableSchemaDeltas` issues one `ALTER TABLE ... ADD COLUMN IF NOT EXISTS ...` per added column, sequentially on one connection, with ~1 ms between statements. On the ClickHouse side each ALTER commits a metadata-version bump to Keeper, but the replica applies it **asynchronously** (typically tens of ms). The next ALTER's assignment check then finds the replica behind and fails: ``` code: 517, message: Metadata on replica is not up to date with common metadata in Zookeeper. It means that this replica still not applied some of previous alters. Probably too many alters executing concurrently (highly not recommended). You can retry this error. ``` Because 517 is not in `retryableExceptions`, the error propagates out of `ReplayTableSchemaDeltas` and fails the entire sync batch. Worse, the failure is **self-sustaining**: when Temporal retries the batch, the first `ADD COLUMN IF NOT EXISTS` is now a no-op (its column already exists) — but a no-op ALTER on a Replicated table *still bumps the shared metadata version* (observed on ClickHouse 26.7.3: the znode version incremented once per retry cycle), so the second column's ALTER hits the same race on **every** replay. We observed a production mirror (MySQL/MariaDB → self-hosted CH, 23 tables) livelock this way for 46 hours — 280 consecutive identical failures at the Temporal retry cadence, CDC fully stopped — after a 4-column source migration. Nothing self-heals; the replica is fully caught up between cycles, and the race re-arms each time. ## The fix One line: `chproto.ErrCannotAssignAlter` added to `retryableExceptions`. The existing `Exec` backoff (1 s on first retry) is orders of magnitude longer than the replica's metadata apply latency, so the retried ALTER succeeds. We ran this patch on the affected deployment: the wedged batch completed on its first retry and a ~970k-row backlog drained immediately. ClickHouse's own error text explicitly marks the condition retryable. ## Related observation (not changed here) `flow/connectors/clickhouse/clickhouse.go` sets `"alter_sync": uint64(3)` with the comment "synchronous wait for ALTER commands; skip replicas that are down". In ClickHouse OSS, `alter_sync` valid values are 0/1/2 (26.7.3 treats other values as "do not wait"), which is what leaves the ~1 ms race window in the first place. If 3 is intentional for ClickHouse Cloud, a version/dialect-aware value might be worth considering — happy to file separately. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01Mr6Ur67uzqbJSgLyME5ZXL
For tables where the primary key is generated always by identity, non-default values we get in logical replication cannot be inserted. The user needs to change it to generated by default or serial etc. This PR wires the notification to the user for this scenario
…nto bq-cdc/4-pull-records
…records in the window
- add ApplySchemaDeltas to isolatedTablePullSyncLoop
dtunikov
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
INSERT INTO ... SELECT ... FROM s3(['url1', 'url2'], 'access_key', 'secret_key', 'Avro')