diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index c125ad22..fe93a6ab 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/crates/clickhousectl/src/cloud/clickpipes.rs @@ -1965,6 +1965,13 @@ pub struct BigQueryCreateArgs { #[arg(long, value_parser = clap::value_parser!(i64).range(1..))] pub sync_interval_seconds: Option, + /// Cron schedule for CDC syncs, such as "0 * * * *" or "@hourly" + /// + /// UTC unless prefixed with CRON_TZ=; at most once a minute. + /// Not with snapshot replication or --sync-interval-seconds. + #[arg(long, value_name = "CRON", conflicts_with = "sync_interval_seconds")] + pub sync_cron: 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, @@ -5972,6 +5979,16 @@ fn validate_bigquery_create_args(args: &BigQueryCreateArgs) -> CloudResult<()> { ))); } } + if let Some(cron) = &args.sync_cron { + if args.replication_mode == "snapshot" { + return Err(CloudError::usage( + "--sync-cron cannot be used with --replication-mode snapshot", + )); + } + if cron.trim().is_empty() { + return Err(CloudError::usage("--sync-cron cannot be empty")); + } + } match ( args.replication_mode.as_str(), args.replication_method.as_deref(), @@ -6088,7 +6105,7 @@ fn build_bigquery_request( snapshot_num_rows_per_partition: args.snapshot_rows_per_partition, snapshot_number_of_parallel_tables: args.snapshot_parallel_tables, replication_method: args.replication_method.clone(), - sync_cron: None, + sync_cron: args.sync_cron.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, @@ -10098,9 +10115,91 @@ mod tests { assert_eq!(args.initial_load_parallelism, None); assert_eq!(args.snapshot_rows_per_partition, None); assert_eq!(args.snapshot_parallel_tables, None); + assert_eq!(args.sync_cron, None); assert_eq!(args.destination.destination_database, "default"); } + #[test] + fn parses_bigquery_sync_cron_and_rejects_it_with_an_interval() { + let base = [ + "create", + "bigquery", + "svc-1", + "--name", + "pipe-1", + "--service-account-file", + "/tmp/account.json", + "--staging-path", + "gs://bucket/staging", + "--table-mapping", + "dataset.one:one", + "--replication-mode", + "cdc_only", + "--replication-method", + "query_based", + ]; + let mut args: Vec<&str> = base.to_vec(); + args.extend(["--sync-cron", "CRON_TZ=Europe/Berlin 0 * * * *"]); + let ClickPipeCommands::Create { + command: ClickPipeCreateCommands::BigQuery(parsed), + } = parse_clickpipe(&args) + else { + panic!("expected bigquery create"); + }; + assert_eq!( + parsed.sync_cron.as_deref(), + Some("CRON_TZ=Europe/Berlin 0 * * * *") + ); + + let mut argv = vec!["clickhousectl", "cloud", "clickpipe"]; + argv.extend(base); + argv.extend(["--sync-cron", "@hourly", "--sync-interval-seconds", "60"]); + assert_eq!( + crate::cli::Cli::try_parse_from(argv).err().unwrap().kind(), + clap::error::ErrorKind::ArgumentConflict + ); + } + + #[test] + fn bigquery_sync_cron_is_sent_for_cdc_and_rejected_for_snapshots() { + 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".into(); + args.replication_method = Some("events_based".into()); + args.sync_cron = Some("@hourly".into()); + args.table_mappings.clear(); + args.table_mappings_json = vec![ + serde_json::json!({ + "sourceDatasetName": "sales", "sourceTable": "orders", "targetTable": "orders", + "eventsFunction": "changes" + }) + .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", "replicationMethod": "events_based", + "syncCron": "@hourly"}) + ); + + args.sync_cron = Some(" ".into()); + let error = validate_bigquery_create_args(&args).unwrap_err(); + assert!( + error.message.contains("--sync-cron cannot be empty"), + "{error}" + ); + + args.replication_mode = "snapshot".into(); + args.replication_method = None; + args.sync_cron = Some("@hourly".into()); + let error = validate_bigquery_create_args(&args).unwrap_err(); + assert_eq!(error.kind, crate::cloud::client::CloudErrorKind::Usage); + assert!(error.message.contains("--sync-cron"), "{error}"); + } + #[test] fn non_postgres_database_sources_require_one_mapping_form() { let cases = [ @@ -13041,6 +13140,7 @@ mod tests { replication_mode: "snapshot".into(), replication_method: None, sync_interval_seconds: None, + sync_cron: None, query_cdc_pull_sync_parallelism: None, sync_delay_seconds: None, pull_window_size_seconds: None, diff --git a/docs/reference/clickpipe-configuration.md b/docs/reference/clickpipe-configuration.md index 864db98c..7343c46f 100644 --- a/docs/reference/clickpipe-configuration.md +++ b/docs/reference/clickpipe-configuration.md @@ -101,6 +101,10 @@ 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. +To sync on a schedule instead of an interval, pass `--sync-cron` with a 5-field cron +expression or a descriptor such as `@hourly`, in place of `--sync-interval-seconds`. +It runs in UTC unless you prefix it with `CRON_TZ=` (for example, +`--sync-cron "CRON_TZ=Europe/Berlin 0 * * * *"`), and must not fire more than once a minute. Snapshot mode rejects these CDC settings and both CDC table fields. The same controls work with workload identity; BigQuery source updates remain unsupported.