diff --git a/crates/clickhouse-cloud-api/README.md b/crates/clickhouse-cloud-api/README.md index 08909e4f..c8c2a656 100644 --- a/crates/clickhouse-cloud-api/README.md +++ b/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 diff --git a/crates/clickhouse-cloud-api/src/convert/clickpipes.rs b/crates/clickhouse-cloud-api/src/convert/clickpipes.rs index 7153622c..0296c582 100644 --- a/crates/clickhouse-cloud-api/src/convert/clickpipes.rs +++ b/crates/clickhouse-cloud-api/src/convert/clickpipes.rs @@ -57,3 +57,63 @@ impl TryFrom for ClickPipeKinesisSchemaR }) } } + +impl TryFrom + for crate::models::ClickPipeBigQueryPipeSettings +{ + type Error = MissingRequiredFields; + + fn try_from( + value: crate::models::ClickPipeBigQueryPipeSettingsResponse, + ) -> Result { + 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 + for crate::models::ClickPipeBigQueryPipeTableMapping +{ + type Error = MissingRequiredFields; + + fn try_from( + value: crate::models::ClickPipeBigQueryPipeTableMappingResponse, + ) -> Result { + 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, + }) + } +} diff --git a/crates/clickhouse-cloud-api/src/models/clickpipes.rs b/crates/clickhouse-cloud-api/src/models/clickpipes.rs index d75eddbd..470a21cd 100644 --- a/crates/clickhouse-cloud-api/src/models/clickpipes.rs +++ b/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,6 +66,8 @@ 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}"), } } @@ -69,7 +75,7 @@ impl std::fmt::Display for ClickPipeBigQueryPipeSettingsReplicationmode { 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, + #[serde( + rename = "pullWindowSizeSeconds", + skip_serializing_if = "Option::is_none" + )] + pub pull_window_size_seconds: Option, + #[serde( + rename = "queryCdcPullSyncParallelism", + skip_serializing_if = "Option::is_none" + )] + pub query_cdc_pull_sync_parallelism: Option, + #[serde(rename = "replicationMethod", skip_serializing_if = "Option::is_none")] + pub replication_method: Option, + #[serde(rename = "syncDelaySeconds", skip_serializing_if = "Option::is_none")] + pub sync_delay_seconds: Option, + #[serde( + rename = "syncIntervalSeconds", + skip_serializing_if = "Option::is_none" + )] + pub sync_interval_seconds: Option, } /// `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, + #[serde( + rename = "pullWindowSizeSeconds", + skip_serializing_if = "Option::is_none" + )] + pub pull_window_size_seconds: Option, + #[serde( + rename = "queryCdcPullSyncParallelism", + skip_serializing_if = "Option::is_none" + )] + pub query_cdc_pull_sync_parallelism: Option, + #[serde(rename = "replicationMethod", skip_serializing_if = "Option::is_none")] + pub replication_method: Option, + #[serde(rename = "syncDelaySeconds", skip_serializing_if = "Option::is_none")] + pub sync_delay_seconds: Option, + #[serde( + rename = "syncIntervalSeconds", + skip_serializing_if = "Option::is_none" + )] + pub sync_interval_seconds: Option, } /// `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, + #[serde(rename = "eventsFunction", skip_serializing_if = "Option::is_none")] + pub events_function: Option, + #[serde( + rename = "queryCdcWatermarkColumn", + skip_serializing_if = "Option::is_none" + )] + pub query_cdc_watermark_column: Option, } /// `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, + #[serde(rename = "eventsFunction", skip_serializing_if = "Option::is_none")] + pub events_function: Option, + #[serde( + rename = "queryCdcWatermarkColumn", + skip_serializing_if = "Option::is_none" + )] + pub query_cdc_watermark_column: Option, } /// `ClickPipeBigQuerySource` from the ClickHouse Cloud API, selected by `authentication`. diff --git a/crates/clickhouse-cloud-api/tests/models_test.rs b/crates/clickhouse-cloud-api/tests/models_test.rs index ae90aa3f..ba4a9eb5 100644 --- a/crates/clickhouse-cloud-api/tests/models_test.rs +++ b/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::(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::(serde_json::json!({})).is_err() + ); + assert!( + serde_json::from_value::(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"] + ); +} diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index 2a0bb316..6b44ca83 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/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); } diff --git a/crates/clickhousectl/tests/cli_request_shape_test.rs b/crates/clickhousectl/tests/cli_request_shape_test.rs index 6189c3ed..66046e53 100644 --- a/crates/clickhousectl/tests/cli_request_shape_test.rs +++ b/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(),