feat(api): model BigQuery CDC settings and mappings

crates/clickhouse-cloud-api/README.md

@@ -6,6 +6,17 @@ Requires Rust 1.94 or later.
 
 ## Updated Cloud API surface
 
+BigQuery replication modes now include `Cdc` and `CdcOnly`, alongside `Snapshot`,
+matching the modes documented by the API. BigQuery settings add optional `replication_method`, `sync_interval_seconds`,
+`query_cdc_pull_sync_parallelism`, `sync_delay_seconds`, and
+`pull_window_size_seconds`; table mappings add `query_cdc_watermark_column` and
+`events_function`. These strings remain open as specified by the API. Responses
+preserve absent and null values as `None`, and `TryFrom` converts settings and
+mappings back to request types after checking required fields. Struct-literal
+callers must supply the new fields or use `..Default::default()`. The API requires
+CDC settings and table mapping fields to match the replication method; snapshot
+requests omit them.
+
 `openapi_key_get_list(organization_id, limit, cursor)` returns one page of API keys.
 Migrate existing calls by passing `None, None` for the new arguments. The server
 accepts limits from 1–250 and defaults to 250; continue with the response's

crates/clickhouse-cloud-api/src/convert/clickpipes.rs

@@ -57,3 +57,63 @@ impl TryFrom<ClickPipeKinesisSchemaRegistryResponse> for ClickPipeKinesisSchemaR
         })
     }
 }
+
+impl TryFrom<crate::models::ClickPipeBigQueryPipeSettingsResponse>
+    for crate::models::ClickPipeBigQueryPipeSettings
+{
+    type Error = MissingRequiredFields;
+
+    fn try_from(
+        value: crate::models::ClickPipeBigQueryPipeSettingsResponse,
+    ) -> Result<Self, Self::Error> {
+        Ok(Self {
+            replication_mode: value
+                .replication_mode
+                .ok_or_else(|| MissingRequiredFields::new(vec!["replicationMode"]))?,
+            allow_nullable_columns: value.allow_nullable_columns,
+            initial_load_parallelism: value.initial_load_parallelism,
+            snapshot_num_rows_per_partition: value.snapshot_num_rows_per_partition,
+            snapshot_number_of_parallel_tables: value.snapshot_number_of_parallel_tables,
+            pull_window_size_seconds: value.pull_window_size_seconds,
+            query_cdc_pull_sync_parallelism: value.query_cdc_pull_sync_parallelism,
+            replication_method: value.replication_method,
+            sync_delay_seconds: value.sync_delay_seconds,
+            sync_interval_seconds: value.sync_interval_seconds,
+        })
+    }
+}
+
+impl TryFrom<crate::models::ClickPipeBigQueryPipeTableMappingResponse>
+    for crate::models::ClickPipeBigQueryPipeTableMapping
+{
+    type Error = MissingRequiredFields;
+
+    fn try_from(
+        value: crate::models::ClickPipeBigQueryPipeTableMappingResponse,
+    ) -> Result<Self, Self::Error> {
+        let mut missing = Vec::new();
+        if value.source_dataset_name.is_none() {
+            missing.push("sourceDatasetName");
+        }
+        if value.source_table.is_none() {
+            missing.push("sourceTable");
+        }
+        if value.target_table.is_none() {
+            missing.push("targetTable");
+        }
+        if !missing.is_empty() {
+            return Err(MissingRequiredFields::new(missing));
+        }
+        Ok(Self {
+            source_dataset_name: value.source_dataset_name.expect("checked above"),
+            source_table: value.source_table.expect("checked above"),
+            target_table: value.target_table.expect("checked above"),
+            excluded_columns: value.excluded_columns,
+            sorting_keys: value.sorting_keys,
+            table_engine: value.table_engine,
+            use_custom_sorting_key: value.use_custom_sorting_key,
+            events_function: value.events_function,
+            query_cdc_watermark_column: value.query_cdc_watermark_column,
+        })
+    }
+}

crates/clickhouse-cloud-api/src/models/clickpipes.rs

@@ -53,6 +53,10 @@ pub enum ClickPipeBigQueryPipeSettingsReplicationmode {
     #[serde(rename = "snapshot")]
     #[default]
     Snapshot,
+    #[serde(rename = "cdc")]
+    Cdc,
+    #[serde(rename = "cdc_only")]
+    CdcOnly,
     /// Catch-all for unknown or newly-added values.
     #[serde(untagged)]
     Unknown(String),
@@ -62,14 +66,16 @@ impl std::fmt::Display for ClickPipeBigQueryPipeSettingsReplicationmode {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Snapshot => write!(f, "snapshot"),
+            Self::Cdc => write!(f, "cdc"),
+            Self::CdcOnly => write!(f, "cdc_only"),
             Self::Unknown(s) => write!(f, "{s}"),
         }
     }
 }
 
 impl ClickPipeBigQueryPipeSettingsReplicationmode {
     /// Wire values accepted by the API, excluding the catch-all.
-    pub const VALUES: &'static [&'static str] = &["snapshot"];
+    pub const VALUES: &'static [&'static str] = &["snapshot", "cdc", "cdc_only"];
 }
 
 /// Inline enum for `ClickPipeBigQueryPipeTableMapping.tableEngine`.
@@ -2012,6 +2018,25 @@ pub struct ClickPipeBigQueryPipeSettings {
         skip_serializing_if = "Option::is_none"
     )]
     pub snapshot_number_of_parallel_tables: Option<f64>,
+    #[serde(
+        rename = "pullWindowSizeSeconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub pull_window_size_seconds: Option<i64>,
+    #[serde(
+        rename = "queryCdcPullSyncParallelism",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub query_cdc_pull_sync_parallelism: Option<i64>,
+    #[serde(rename = "replicationMethod", skip_serializing_if = "Option::is_none")]
+    pub replication_method: Option<String>,
+    #[serde(rename = "syncDelaySeconds", skip_serializing_if = "Option::is_none")]
+    pub sync_delay_seconds: Option<i64>,
+    #[serde(
+        rename = "syncIntervalSeconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub sync_interval_seconds: Option<i64>,
 }
 
 /// `ClickPipeBigQueryPipeSettings` from the ClickHouse Cloud API, in response
@@ -2044,6 +2069,25 @@ pub struct ClickPipeBigQueryPipeSettingsResponse {
         skip_serializing_if = "Option::is_none"
     )]
     pub snapshot_number_of_parallel_tables: Option<f64>,
+    #[serde(
+        rename = "pullWindowSizeSeconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub pull_window_size_seconds: Option<i64>,
+    #[serde(
+        rename = "queryCdcPullSyncParallelism",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub query_cdc_pull_sync_parallelism: Option<i64>,
+    #[serde(rename = "replicationMethod", skip_serializing_if = "Option::is_none")]
+    pub replication_method: Option<String>,
+    #[serde(rename = "syncDelaySeconds", skip_serializing_if = "Option::is_none")]
+    pub sync_delay_seconds: Option<i64>,
+    #[serde(
+        rename = "syncIntervalSeconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub sync_interval_seconds: Option<i64>,
 }
 
 /// `ClickPipeBigQueryPipeTableMapping` from the ClickHouse Cloud API.
@@ -2066,6 +2110,13 @@ pub struct ClickPipeBigQueryPipeTableMapping {
         skip_serializing_if = "Option::is_none"
     )]
     pub use_custom_sorting_key: Option<bool>,
+    #[serde(rename = "eventsFunction", skip_serializing_if = "Option::is_none")]
+    pub events_function: Option<String>,
+    #[serde(
+        rename = "queryCdcWatermarkColumn",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub query_cdc_watermark_column: Option<String>,
 }
 
 /// `ClickPipeBigQueryPipeTableMapping` from the ClickHouse Cloud API, in
@@ -2093,6 +2144,13 @@ pub struct ClickPipeBigQueryPipeTableMappingResponse {
         skip_serializing_if = "Option::is_none"
     )]
     pub use_custom_sorting_key: Option<bool>,
+    #[serde(rename = "eventsFunction", skip_serializing_if = "Option::is_none")]
+    pub events_function: Option<String>,
+    #[serde(
+        rename = "queryCdcWatermarkColumn",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub query_cdc_watermark_column: Option<String>,
 }
 
 /// `ClickPipeBigQuerySource` from the ClickHouse Cloud API, selected by `authentication`.

crates/clickhouse-cloud-api/tests/models_test.rs

@@ -8896,3 +8896,125 @@ fn mongodb_structured_ingestion_response_tolerates_absent_and_null_fields() {
         serde_json::json!({})
     );
 }
+
+#[test]
+fn bigquery_cdc_fields_round_trip_and_write_back() {
+    let settings = serde_json::json!({
+        "replicationMode": "cdc", "replicationMethod": "query_based",
+        "syncIntervalSeconds": 60, "queryCdcPullSyncParallelism": 4,
+        "syncDelaySeconds": 0, "pullWindowSizeSeconds": 86400
+    });
+    let response: ClickPipeBigQueryPipeSettingsResponse =
+        serde_json::from_value(settings.clone()).unwrap();
+    let request = ClickPipeBigQueryPipeSettings::try_from(response).unwrap();
+    assert_eq!(request.replication_method.as_deref(), Some("query_based"));
+    assert_eq!(request.sync_interval_seconds, Some(60));
+    assert_eq!(request.query_cdc_pull_sync_parallelism, Some(4));
+    assert_eq!(request.sync_delay_seconds, Some(0));
+    assert_eq!(request.pull_window_size_seconds, Some(86400));
+    assert_eq!(serde_json::to_value(&request).unwrap(), settings);
+    assert_eq!(
+        serde_json::from_value::<ClickPipeBigQueryPipeSettings>(settings).unwrap(),
+        request
+    );
+
+    let mapping = serde_json::json!({
+        "sourceDatasetName": "sales", "sourceTable": "orders", "targetTable": "orders_raw",
+        "queryCdcWatermarkColumn": "updated_at", "eventsFunction": "future_function"
+    });
+    let response: ClickPipeBigQueryPipeTableMappingResponse =
+        serde_json::from_value(mapping.clone()).unwrap();
+    let request = ClickPipeBigQueryPipeTableMapping::try_from(response).unwrap();
+    assert_eq!(
+        request.query_cdc_watermark_column.as_deref(),
+        Some("updated_at")
+    );
+    assert_eq!(request.events_function.as_deref(), Some("future_function"));
+    assert_eq!(serde_json::to_value(request).unwrap(), mapping);
+}
+
+#[test]
+fn bigquery_cdc_response_fields_tolerate_absence_and_null() {
+    for value in [
+        serde_json::json!({}),
+        serde_json::json!({
+            "replicationMode": null, "replicationMethod": null, "syncIntervalSeconds": null,
+            "queryCdcPullSyncParallelism": null, "syncDelaySeconds": null, "pullWindowSizeSeconds": null
+        }),
+    ] {
+        let response: ClickPipeBigQueryPipeSettingsResponse =
+            serde_json::from_value(value).unwrap();
+        assert_eq!(response, ClickPipeBigQueryPipeSettingsResponse::default());
+        assert_eq!(
+            serde_json::to_value(&response).unwrap(),
+            serde_json::json!({})
+        );
+        assert_eq!(
+            ClickPipeBigQueryPipeSettings::try_from(response)
+                .unwrap_err()
+                .fields(),
+            &["replicationMode"]
+        );
+    }
+    for value in [
+        serde_json::json!({}),
+        serde_json::json!({
+            "eventsFunction": null, "queryCdcWatermarkColumn": null
+        }),
+    ] {
+        let response: ClickPipeBigQueryPipeTableMappingResponse =
+            serde_json::from_value(value).unwrap();
+        assert_eq!(
+            response,
+            ClickPipeBigQueryPipeTableMappingResponse::default()
+        );
+        assert_eq!(
+            serde_json::to_value(&response).unwrap(),
+            serde_json::json!({})
+        );
+        assert_eq!(
+            ClickPipeBigQueryPipeTableMapping::try_from(response)
+                .unwrap_err()
+                .fields(),
+            &["sourceDatasetName", "sourceTable", "targetTable"]
+        );
+    }
+    assert!(
+        serde_json::from_value::<ClickPipeBigQueryPipeSettings>(serde_json::json!({})).is_err()
+    );
+    assert!(
+        serde_json::from_value::<ClickPipeBigQueryPipeTableMapping>(serde_json::json!({})).is_err()
+    );
+}
+
+#[test]
+fn bigquery_replication_modes_cover_cdc_and_preserve_future_values() {
+    for (wire, expected) in [
+        (
+            "snapshot",
+            ClickPipeBigQueryPipeSettingsReplicationmode::Snapshot,
+        ),
+        ("cdc", ClickPipeBigQueryPipeSettingsReplicationmode::Cdc),
+        (
+            "cdc_only",
+            ClickPipeBigQueryPipeSettingsReplicationmode::CdcOnly,
+        ),
+        (
+            "future",
+            ClickPipeBigQueryPipeSettingsReplicationmode::Unknown("future".into()),
+        ),
+    ] {
+        let parsed: ClickPipeBigQueryPipeSettingsReplicationmode =
+            serde_json::from_value(serde_json::json!(wire)).unwrap();
+        assert_eq!(parsed, expected);
+        assert_eq!(parsed.to_string(), wire);
+        assert_eq!(
+            serde_json::to_value(parsed).unwrap(),
+            serde_json::json!(wire)
+        );
+    }
+    assert_eq!(
+        ClickPipeBigQueryPipeSettingsReplicationmode::VALUES,
+        &["snapshot", "cdc", "cdc_only"]
+    );
+}

crates/clickhousectl/src/cloud/clickpipes.rs

@@ -80,14 +80,13 @@ 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::{
-    ClickPipeBigQueryPipeSettingsReplicationmode, ClickPipeBigQueryPipeTableMappingTableengine,
-    ClickPipeKafkaOffsetStrategy, ClickPipeMongoDBPipeSettingsReplicationmode,
-    ClickPipeMongoDBPipeTableMappingTableengine, ClickPipeMutateMongoDBSourceReadpreference,
-    ClickPipeMutateMySQLSourceType, ClickPipeMySQLPipeSettingsReplicationmechanism,
-    ClickPipeMySQLPipeSettingsReplicationmode, ClickPipeMySQLPipeTableMappingTableengine,
-    ClickPipePostKafkaSourceAuthentication, ClickPipePostKafkaSourceFormat,
-    ClickPipePostKafkaSourceType, ClickPipePostgresPipeTableMapping,
-    ClickPipePostgresPipeTableMappingTableengine,
+    ClickPipeBigQueryPipeTableMappingTableengine, ClickPipeKafkaOffsetStrategy,
+    ClickPipeMongoDBPipeSettingsReplicationmode, ClickPipeMongoDBPipeTableMappingTableengine,
+    ClickPipeMutateMongoDBSourceReadpreference, ClickPipeMutateMySQLSourceType,
+    ClickPipeMySQLPipeSettingsReplicationmechanism, ClickPipeMySQLPipeSettingsReplicationmode,
+    ClickPipeMySQLPipeTableMappingTableengine, ClickPipePostKafkaSourceAuthentication,
+    ClickPipePostKafkaSourceFormat, ClickPipePostKafkaSourceType,
+    ClickPipePostgresPipeTableMapping, ClickPipePostgresPipeTableMappingTableengine,
 };
 use tabled::{Table, Tabled, settings::Style};
 
@@ -1951,7 +1950,7 @@ pub struct BigQueryCreateArgs {
     #[arg(
         long,
         default_value = "snapshot",
-        value_parser = PossibleValuesParser::new(ClickPipeBigQueryPipeSettingsReplicationmode::VALUES),
+        value_parser = PossibleValuesParser::new(["snapshot"]),
     )]
     pub replication_mode: String,
 
@@ -5985,6 +5984,7 @@ 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()
     };
     let source = match authentication {
         GcpAuthentication::ServiceAccount => {
@@ -10193,9 +10193,7 @@ mod tests {
         for &value in REPLICATION_MODES {
             assert_postgres_value("--replication-mode", value);
         }
-        for &value in ClickPipeBigQueryPipeSettingsReplicationmode::VALUES {
-            assert_bigquery_value("--replication-mode", value);
-        }
+        assert_bigquery_value("--replication-mode", "snapshot");
         for &value in DB_AUTHS {
             assert_mysql_value("--auth", value);
         }

crates/clickhousectl/tests/cli_request_shape_test.rs

@@ -8497,6 +8497,11 @@ async fn bigquery_destination_omits_table_columns_managed_table_definition() {
         "initialLoadParallelism",
         "snapshotNumRowsPerPartition",
         "snapshotNumberOfParallelTables",
+        "replicationMethod",
+        "syncIntervalSeconds",
+        "queryCdcPullSyncParallelism",
+        "syncDelaySeconds",
+        "pullWindowSizeSeconds",
     ] {
         assert!(
             settings.get(field).is_none(),