From 65addfae0cb59bb91bbf879cf5a79039717009ac Mon Sep 17 00:00:00 2001 From: sdairs Date: Tue, 20 Oct 2026 12:28:18 +0100 Subject: [PATCH] feat(clickpipes): expose BigQuery CDC settings and mappings (refs #1088) --- crates/clickhousectl/src/cloud/clickpipes.rs | 313 +++++++++++++++++- .../tests/cli_request_shape_test.rs | 120 +++++++ docs/reference/clickpipe-configuration.md | 26 +- 3 files changed, 445 insertions(+), 14 deletions(-) diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index 6b44ca83..663984a3 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/crates/clickhousectl/src/cloud/clickpipes.rs @@ -80,13 +80,14 @@ use crate::cloud::shared::{parse_datetime, parse_serde_enum, resolve_org_id}; use clap::builder::{PossibleValue, PossibleValuesParser, TypedValueParser}; use clap::{ArgGroup, Args, Subcommand}; use clickhouse_cloud_api::models::{ - ClickPipeBigQueryPipeTableMappingTableengine, ClickPipeKafkaOffsetStrategy, - ClickPipeMongoDBPipeSettingsReplicationmode, ClickPipeMongoDBPipeTableMappingTableengine, - ClickPipeMutateMongoDBSourceReadpreference, ClickPipeMutateMySQLSourceType, - ClickPipeMySQLPipeSettingsReplicationmechanism, ClickPipeMySQLPipeSettingsReplicationmode, - ClickPipeMySQLPipeTableMappingTableengine, ClickPipePostKafkaSourceAuthentication, - ClickPipePostKafkaSourceFormat, ClickPipePostKafkaSourceType, - ClickPipePostgresPipeTableMapping, ClickPipePostgresPipeTableMappingTableengine, + ClickPipeBigQueryPipeSettingsReplicationmode, ClickPipeBigQueryPipeTableMappingTableengine, + ClickPipeKafkaOffsetStrategy, ClickPipeMongoDBPipeSettingsReplicationmode, + ClickPipeMongoDBPipeTableMappingTableengine, ClickPipeMutateMongoDBSourceReadpreference, + ClickPipeMutateMySQLSourceType, ClickPipeMySQLPipeSettingsReplicationmechanism, + ClickPipeMySQLPipeSettingsReplicationmode, ClickPipeMySQLPipeTableMappingTableengine, + ClickPipePostKafkaSourceAuthentication, ClickPipePostKafkaSourceFormat, + ClickPipePostKafkaSourceType, ClickPipePostgresPipeTableMapping, + ClickPipePostgresPipeTableMappingTableengine, }; use tabled::{Table, Tabled, settings::Style}; @@ -1940,9 +1941,8 @@ pub struct BigQueryCreateArgs { /// Full table mapping as a JSON object (repeatable) /// - /// Supports excludedColumns, sortingKeys, useCustomSortingKey and - /// tableEngine. Combinable with --table-mapping; unknown fields are - /// rejected. + /// Supports sorting, exclusions, engines and CDC watermark/event fields. + /// Combinable with --table-mapping; unknown fields are rejected. #[arg(long = "table-mapping-json", value_name = "JSON")] pub table_mappings_json: Vec, @@ -1950,10 +1950,33 @@ pub struct BigQueryCreateArgs { #[arg( long, default_value = "snapshot", - value_parser = PossibleValuesParser::new(["snapshot"]), + value_parser = PossibleValuesParser::new(ClickPipeBigQueryPipeSettingsReplicationmode::VALUES), )] pub replication_mode: String, + /// CDC method (required with --replication-mode cdc or cdc_only) + /// + /// Each table needs queryCdcWatermarkColumn for query_based or + /// eventsFunction for events_based in --table-mapping-json. + #[arg(long, value_parser = ["query_based", "events_based"])] + pub replication_method: Option, + + /// CDC sync interval in seconds (not with snapshot replication) + #[arg(long, value_parser = clap::value_parser!(i64).range(1..))] + pub sync_interval_seconds: Option, + + /// Parallel table pulls per CDC sync (not with snapshot replication) + #[arg(long, value_parser = clap::value_parser!(i64).range(1..))] + pub query_cdc_pull_sync_parallelism: Option, + + /// CDC visibility delay in seconds (not with snapshot replication) + #[arg(long, value_parser = clap::value_parser!(i64).range(0..))] + pub sync_delay_seconds: Option, + + /// Maximum CDC pull window in seconds (not with snapshot replication) + #[arg(long, value_parser = clap::value_parser!(i64).range(1..))] + pub pull_window_size_seconds: Option, + /// Preserve BigQuery nullability in destination columns #[arg(long, value_name = "true|false")] pub allow_nullable_columns: Option, @@ -5102,6 +5125,8 @@ const BIGQUERY_TABLE_MAPPING_JSON_FIELDS: &[&str] = &[ "sortingKeys", "useCustomSortingKey", "tableEngine", + "queryCdcWatermarkColumn", + "eventsFunction", ]; fn parse_bigquery_table_mapping_json( @@ -5128,6 +5153,22 @@ fn parse_bigquery_table_mapping_json( &mapping.table_engine, ClickPipeBigQueryPipeTableMappingTableengine::VALUES, )?; + if let Some(column) = mapping.query_cdc_watermark_column { + mapping.query_cdc_watermark_column = Some(required_table_mapping_name( + &flag, + "queryCdcWatermarkColumn", + column, + )?); + } + if let Some(function) = mapping.events_function { + let function = required_table_mapping_name(&flag, "eventsFunction", function)?; + if !["appends", "changes"].contains(&function.as_str()) { + return Err(CloudError::usage(format!( + "{flag}: eventsFunction must be appends or changes" + ))); + } + mapping.events_function = Some(function); + } Ok(mapping) } @@ -5906,6 +5947,53 @@ fn validate_bigquery_create_args(args: &BigQueryCreateArgs) -> CloudResult<()> { } } + for (flag, value, minimum) in [ + ("--sync-interval-seconds", args.sync_interval_seconds, 1), + ( + "--query-cdc-pull-sync-parallelism", + args.query_cdc_pull_sync_parallelism, + 1, + ), + ("--sync-delay-seconds", args.sync_delay_seconds, 0), + ( + "--pull-window-size-seconds", + args.pull_window_size_seconds, + 1, + ), + ] { + if value.is_some_and(|value| value < minimum) { + return Err(CloudError::usage(format!( + "{flag} must be at least {minimum}" + ))); + } + if args.replication_mode == "snapshot" && value.is_some() { + return Err(CloudError::usage(format!( + "{flag} cannot be used with --replication-mode snapshot" + ))); + } + } + match ( + args.replication_mode.as_str(), + args.replication_method.as_deref(), + ) { + ("snapshot", Some(_)) => { + return Err(CloudError::usage( + "--replication-method cannot be used with --replication-mode snapshot", + )); + } + ("cdc" | "cdc_only", None) => { + return Err(CloudError::usage( + "--replication-method is required with --replication-mode cdc or cdc_only", + )); + } + (_, Some(method)) if !["query_based", "events_based"].contains(&method) => { + return Err(CloudError::usage( + "--replication-method must be query_based or events_based", + )); + } + _ => {} + } + let authentication = parse_bigquery_authentication(&args.auth)?; match ( authentication, @@ -5960,6 +6048,21 @@ fn resolve_bigquery_table_mappings( for (position, mapping) in args.table_mappings_json.iter().enumerate() { mappings.push(parse_bigquery_table_mapping_json(position, mapping)?); } + for (position, mapping) in mappings.iter().enumerate() { + let watermark = mapping.query_cdc_watermark_column.is_some(); + let events = mapping.events_function.is_some(); + let valid = match args.replication_method.as_deref() { + Some("query_based") => watermark && !events, + Some("events_based") => events && !watermark, + _ => !watermark && !events, + }; + if !valid { + return Err(CloudError::usage(format!( + "table mapping #{}: query_based requires queryCdcWatermarkColumn only; events_based requires eventsFunction only; snapshot allows neither", + position + 1, + ))); + } + } Ok(mappings) } @@ -5984,7 +6087,11 @@ fn build_bigquery_request( initial_load_parallelism: args.initial_load_parallelism, snapshot_num_rows_per_partition: args.snapshot_rows_per_partition, snapshot_number_of_parallel_tables: args.snapshot_parallel_tables, - ..Default::default() + replication_method: args.replication_method.clone(), + sync_interval_seconds: args.sync_interval_seconds, + query_cdc_pull_sync_parallelism: args.query_cdc_pull_sync_parallelism, + sync_delay_seconds: args.sync_delay_seconds, + pull_window_size_seconds: args.pull_window_size_seconds, }; let source = match authentication { GcpAuthentication::ServiceAccount => { @@ -10193,7 +10300,9 @@ mod tests { for &value in REPLICATION_MODES { assert_postgres_value("--replication-mode", value); } - assert_bigquery_value("--replication-mode", "snapshot"); + for &value in ClickPipeBigQueryPipeSettingsReplicationmode::VALUES { + assert_bigquery_value("--replication-mode", value); + } for &value in DB_AUTHS { assert_mysql_value("--auth", value); } @@ -12929,6 +13038,11 @@ mod tests { table_mappings: vec!["source.events:events".into()], table_mappings_json: vec![], replication_mode: "snapshot".into(), + replication_method: None, + sync_interval_seconds: None, + query_cdc_pull_sync_parallelism: None, + sync_delay_seconds: None, + pull_window_size_seconds: None, allow_nullable_columns: None, initial_load_parallelism: None, snapshot_rows_per_partition: None, @@ -15008,4 +15122,177 @@ mod tests { .contains("requires --service-account-file") ); } + #[test] + fn parses_bigquery_cdc_controls_and_rejects_invalid_values() { + let base = [ + "create", + "bigquery", + "svc-1", + "--name", + "cdc", + "--service-account-file", + "/tmp/account.json", + "--staging-path", + "gs://bucket/staging", + "--table-mapping", + "dataset.orders:orders", + ]; + let mut command = base.to_vec(); + command.extend([ + "--replication-mode", + "cdc_only", + "--replication-method", + "query_based", + "--sync-interval-seconds", + "60", + "--query-cdc-pull-sync-parallelism", + "4", + "--sync-delay-seconds", + "0", + "--pull-window-size-seconds", + "86400", + ]); + let ClickPipeCommands::Create { + command: ClickPipeCreateCommands::BigQuery(args), + } = parse_clickpipe(&command) + else { + panic!("expected BigQuery create") + }; + assert_eq!(args.replication_method.as_deref(), Some("query_based")); + assert_eq!(args.sync_interval_seconds, Some(60)); + assert_eq!(args.query_cdc_pull_sync_parallelism, Some(4)); + assert_eq!(args.sync_delay_seconds, Some(0)); + assert_eq!(args.pull_window_size_seconds, Some(86400)); + let ClickPipeCommands::Create { + command: ClickPipeCreateCommands::BigQuery(defaults), + } = parse_clickpipe(&base) + else { + panic!("expected BigQuery create") + }; + assert_eq!(defaults.replication_method, None); + assert_eq!(defaults.sync_interval_seconds, None); + assert_eq!(defaults.query_cdc_pull_sync_parallelism, None); + assert_eq!(defaults.sync_delay_seconds, None); + assert_eq!(defaults.pull_window_size_seconds, None); + for (flag, value) in [ + ("--replication-method", "invalid"), + ("--sync-interval-seconds", "0"), + ("--query-cdc-pull-sync-parallelism", "0"), + ("--sync-delay-seconds", "-1"), + ("--pull-window-size-seconds", "0"), + ] { + let mut command = base.to_vec(); + command.extend([flag, value]); + assert_rejected(&command); + } + let mut repeated = base.to_vec(); + repeated.extend(["--sync-delay-seconds", "0", "--sync-delay-seconds", "1"]); + assert_rejected(&repeated); + } + + #[test] + fn builds_bigquery_cdc_settings_and_method_specific_mappings() { + for (method, field, value) in [ + ("query_based", "queryCdcWatermarkColumn", "updated_at"), + ("events_based", "eventsFunction", "changes"), + ] { + let mut args = bigquery_builder_args("/unused/key.json".into()); + args.auth = "SERVICE_ACCOUNT_WORKLOAD_IDENTITY".into(); + args.service_account_file = None; + args.project_id = Some("source-project".into()); + args.replication_mode = "cdc_only".into(); + args.replication_method = Some(method.into()); + args.sync_interval_seconds = Some(60); + args.query_cdc_pull_sync_parallelism = Some(4); + args.sync_delay_seconds = Some(0); + args.pull_window_size_seconds = Some(86400); + args.table_mappings.clear(); + args.table_mappings_json = vec![serde_json::json!({ + "sourceDatasetName":"sales", "sourceTable":"orders", "targetTable":"orders_raw", field: value + }).to_string()]; + let request = build_bigquery_request(&args).unwrap(); + let source = serde_json::to_value(request).unwrap()["source"]["bigquery"].clone(); + assert_eq!( + source["settings"], + serde_json::json!({"replicationMode":"cdc_only", + "replicationMethod":method, "syncIntervalSeconds":60, "queryCdcPullSyncParallelism":4, + "syncDelaySeconds":0, "pullWindowSizeSeconds":86400}) + ); + assert_eq!(source["tableMappings"][0][field], value); + assert!(source.get("credentials").is_none()); + } + } + + #[test] + fn rejects_bigquery_cdc_mismatches_before_reading_key_file() { + for (mode, method, mapping) in [ + ("snapshot", Some("query_based"), "{}"), + ("cdc", None, "{}"), + ("cdc_only", Some("invalid"), "{}"), + ( + "cdc", + Some("query_based"), + r#"{"eventsFunction":"appends"}"#, + ), + ( + "cdc_only", + Some("events_based"), + r#"{"queryCdcWatermarkColumn":"updated_at"}"#, + ), + ( + "snapshot", + None, + r#"{"queryCdcWatermarkColumn":"updated_at"}"#, + ), + ( + "cdc", + Some("query_based"), + r#"{"queryCdcWatermarkColumn":" "}"#, + ), + ( + "cdc", + Some("events_based"), + r#"{"eventsFunction":"future"}"#, + ), + ] { + let mut args = bigquery_builder_args("/missing/key.json".into()); + args.replication_mode = mode.into(); + args.replication_method = method.map(str::to_owned); + args.table_mappings.clear(); + let mut mapping: serde_json::Value = serde_json::from_str(mapping).unwrap(); + mapping["sourceDatasetName"] = "sales".into(); + mapping["sourceTable"] = "orders".into(); + mapping["targetTable"] = "orders_raw".into(); + args.table_mappings_json = vec![mapping.to_string()]; + let error = build_bigquery_request(&args).unwrap_err(); + assert!( + !error.message.contains("/missing/key.json"), + "{}", + error.message + ); + } + for (flag, minimum) in [("sync", 1), ("parallelism", 1), ("delay", 0), ("window", 1)] { + let mut args = bigquery_builder_args("/missing/key.json".into()); + match flag { + "sync" => args.sync_interval_seconds = Some(minimum - 1), + "parallelism" => args.query_cdc_pull_sync_parallelism = Some(minimum - 1), + "delay" => args.sync_delay_seconds = Some(minimum - 1), + _ => args.pull_window_size_seconds = Some(minimum - 1), + } + assert!( + build_bigquery_request(&args) + .unwrap_err() + .message + .contains("must be at least") + ); + } + let mut args = bigquery_builder_args("/missing/key.json".into()); + args.sync_delay_seconds = Some(0); + assert!( + build_bigquery_request(&args) + .unwrap_err() + .message + .contains("snapshot") + ); + } } diff --git a/crates/clickhousectl/tests/cli_request_shape_test.rs b/crates/clickhousectl/tests/cli_request_shape_test.rs index 66046e53..68c8dddf 100644 --- a/crates/clickhousectl/tests/cli_request_shape_test.rs +++ b/crates/clickhousectl/tests/cli_request_shape_test.rs @@ -32710,3 +32710,123 @@ async fn snapshot_list_reads_every_page_in_json_and_human_output() { "{stdout}" ); } + +#[tokio::test] +async fn bigquery_cdc_controls_and_mappings_are_sent_for_both_auth_methods() { + for (auth, replication_method, mapping_field, mapping_value) in [ + ( + "SERVICE_ACCOUNT", + "query_based", + "queryCdcWatermarkColumn", + "updated_at", + ), + ( + "SERVICE_ACCOUNT_WORKLOAD_IDENTITY", + "events_based", + "eventsFunction", + "appends", + ), + ] { + let mock = start_mock_clickpipes_api().await; + let directory = tempfile::tempdir().unwrap(); + let key = directory.path().join("service-account.json"); + std::fs::write(&key, "{}").unwrap(); + let mapping = serde_json::json!({"sourceDatasetName":"sales", "sourceTable":"orders", + "targetTable":"orders_raw", mapping_field: mapping_value}) + .to_string(); + let mut args = vec![ + "clickpipe", + "create", + "bigquery", + "svc-id", + "--name", + "cdc-pipe", + "--auth", + auth, + "--project-id", + "source-project", + "--staging-path", + "gs://bucket/staging", + "--replication-mode", + "cdc", + "--replication-method", + replication_method, + "--sync-interval-seconds", + "60", + "--query-cdc-pull-sync-parallelism", + "4", + "--sync-delay-seconds", + "0", + "--pull-window-size-seconds", + "86400", + "--table-mapping-json", + &mapping, + "--org-id", + "org", + ]; + if auth == "SERVICE_ACCOUNT" { + args.extend(["--service-account-file", key.to_str().unwrap()]); + } + let body = invoke_cli_capture_body(&mock, &args).await; + let source = &body["source"]["bigquery"]; + assert_eq!( + source["settings"], + serde_json::json!({"replicationMode":"cdc", + "replicationMethod":replication_method, "syncIntervalSeconds":60, + "queryCdcPullSyncParallelism":4, "syncDelaySeconds":0, "pullWindowSizeSeconds":86400}) + ); + assert_eq!(source["tableMappings"][0][mapping_field], mapping_value); + assert_eq!( + source.get("credentials").is_some(), + auth == "SERVICE_ACCOUNT" + ); + } +} + +#[tokio::test] +async fn bigquery_invalid_cdc_settings_fail_before_key_file_or_http() { + let mock = MockServer::start().await; + for extra in [ + vec!["--replication-mode", "cdc"], + vec!["--sync-delay-seconds", "0"], + vec![ + "--replication-mode", + "cdc", + "--replication-method", + "query_based", + ], + vec![ + "--replication-mode", + "cdc_only", + "--replication-method", + "events_based", + ], + ] { + let mut args = vec![ + "clickpipe", + "create", + "bigquery", + "svc-id", + "--name", + "invalid-cdc", + "--service-account-file", + "/missing/bigquery-key.json", + "--staging-path", + "gs://bucket/staging", + "--table-mapping", + "sales.orders:orders", + "--org-id", + "org", + ]; + args.extend(extra); + let output = invoke_cli_with_cloud_credentials(&mock, &args); + assert_eq!( + output.status.code(), + Some(2), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(!String::from_utf8_lossy(&output.stderr).contains("/missing/bigquery-key.json")); + } + assert!(mock.received_requests().await.unwrap().is_empty()); +} diff --git a/docs/reference/clickpipe-configuration.md b/docs/reference/clickpipe-configuration.md index f2f29fd7..864db98c 100644 --- a/docs/reference/clickpipe-configuration.md +++ b/docs/reference/clickpipe-configuration.md @@ -13,7 +13,7 @@ Use `cloud clickpipe create --help` for flag names, allowed values, def | Object storage | Grant read access through IAM, access keys, an Azure connection string, or a GCP service account. Schema discovery needs a running destination service. | | Postgres / MySQL | Basic auth requires both username and password; `IAM_ROLE` on supported RDS/Aurora sources requires a role ARN and rejects username/password. | | MongoDB | CDC requires MongoDB 5.1+ with a replica set or sharded cluster and Change Streams/oplog access. | -| BigQuery | Snapshot loads require a pre-created GCS staging bucket and permissions for reads, export jobs, and staging objects. Only snapshot replication is supported. | +| BigQuery | Snapshot loads require a pre-created GCS staging bucket and permissions for reads, export jobs, and staging objects. CDC additionally requires a replication method and per-table CDC fields. | | Pub/Sub | Service-account key or enabled workload identity; this source is limited preview. Choose `earliest`, `latest`, or `timestamp` deliberately. | For MySQL CDC, use `ROW` binlogs with `FULL` row images and at least 72 hours of retention. Default `GTID` requires GTID replication; choose `FILE_POS` only for a matching setup. Grant source `SELECT`, `REPLICATION CLIENT`, and `REPLICATION SLAVE`. See [MySQL setup](https://clickhouse.com/docs/integrations/clickpipes/mysql). @@ -80,6 +80,30 @@ Kafka/Kinesis Protobuf schema inputs accept `.proto` or serialized `FileDescript Object storage `--skip-initial-load` and `--start-after` cannot be combined. Schema discovery probes the source without creating a pipe and requires an API key, including for workload identity. `--validate-samples` is optional on creates; the API documents it as having no effect for Postgres/MySQL. +## Configure BigQuery CDC + +BigQuery defaults to a one-time `snapshot`. For `--replication-mode cdc` or +`cdc_only`, supply `--replication-method query_based` or `events_based` and use +`--table-mapping-json` for every table. Query-based CDC requires a TIMESTAMP +`queryCdcWatermarkColumn`; event-based CDC requires `eventsFunction` (`appends` +or `changes`). Each method rejects the other method's table field. + +```bash +chctl cloud clickpipe create bigquery "$SERVICE_ID" --name orders \ + --service-account-file ./service-account.json --staging-path gs://bucket/staging \ + --replication-mode cdc --replication-method query_based \ + --sync-interval-seconds 60 --query-cdc-pull-sync-parallelism 4 \ + --sync-delay-seconds 60 --pull-window-size-seconds 86400 \ + --table-mapping-json '{"sourceDatasetName":"sales","sourceTable":"orders","targetTable":"orders","queryCdcWatermarkColumn":"updated_at"}' +``` + +The sync interval, table parallelism, and pull window must be at least 1; the +visibility delay can be 0. A delay that is too short can skip rows that become +visible late; increase it for batch jobs or Dataflow. The pull window bounds each +catch-up query when a pipe falls behind. Omitted settings use the service defaults. +Snapshot mode rejects these CDC settings and both CDC table fields. The same +controls work with workload identity; BigQuery source updates remain unsupported. + ## Update a pipe Use `clickpipe update --file patch.json` or `--file -`. For CDC, [stop and wait for `Paused`](../guides/clickpipes.md#pause-update-and-resume-cdc) first. GET responses contain fields that cannot be patched; build a request from the writable fields below.