faucet_sink_spanner/
config.rs

1//! Cloud Spanner sink configuration.
2
3use faucet_common_spanner::SpannerConnection;
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Configuration for the Cloud Spanner sink.
9#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
10pub struct SpannerSinkConfig {
11    /// Connection block (`project_id` / `instance` / `database` / `auth` /
12    /// `max_sessions` / `emulator_host`), flattened to the config top level.
13    #[serde(flatten)]
14    pub connection: SpannerConnection,
15    /// Target table name.
16    pub table_name: String,
17    /// Maximum rows per Spanner commit. Defaults to [`DEFAULT_BATCH_SIZE`].
18    ///
19    /// Each chunk is applied as one atomic commit. Spanner additionally caps
20    /// a commit at ~80,000 mutated *cells* (rows × columns, plus secondary
21    /// index amplification), so the sink also splits chunks to stay under a
22    /// conservative 60,000-cell budget regardless of `batch_size`.
23    ///
24    /// `batch_size = 0` is the "no batching" sentinel: the entire upstream
25    /// page is written in as few commits as the cell budget allows.
26    #[serde(default = "default_batch_size")]
27    pub batch_size: usize,
28    /// Upper bound (seconds) on waiting for a schema-change operation
29    /// (`ALTER TABLE …` long-running operation) to complete. Defaults to 300.
30    #[serde(default = "default_ddl_timeout_secs")]
31    pub ddl_timeout_secs: u64,
32    /// Write mode, key columns, and optional delete marker. `write_mode`
33    /// defaults to `append`. Upsert/delete require `key` to equal the target
34    /// table's PRIMARY KEY columns (Spanner mutations always key on the PK).
35    #[serde(flatten)]
36    pub write: faucet_core::WriteSpec,
37}
38
39fn default_batch_size() -> usize {
40    DEFAULT_BATCH_SIZE
41}
42
43fn default_ddl_timeout_secs() -> u64 {
44    300
45}
46
47impl SpannerSinkConfig {
48    /// Create a new config with required fields and sensible defaults.
49    pub fn new(
50        project_id: impl Into<String>,
51        instance: impl Into<String>,
52        database: impl Into<String>,
53        table_name: impl Into<String>,
54    ) -> Self {
55        Self {
56            connection: SpannerConnection {
57                project_id: project_id.into(),
58                instance: instance.into(),
59                database: database.into(),
60                auth: Default::default(),
61                max_sessions: 100,
62                emulator_host: None,
63            },
64            table_name: table_name.into(),
65            batch_size: DEFAULT_BATCH_SIZE,
66            ddl_timeout_secs: default_ddl_timeout_secs(),
67            write: faucet_core::WriteSpec::default(),
68        }
69    }
70
71    /// Set the per-commit row count. Pass `0` to opt out of row-count
72    /// re-chunking (the ~60,000-cell budget still applies).
73    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
74        self.batch_size = batch_size;
75        self
76    }
77
78    /// Set the emulator endpoint (`host:port`), for tests and local runs.
79    pub fn with_emulator_host(mut self, host: impl Into<String>) -> Self {
80        self.connection.emulator_host = Some(host.into());
81        self
82    }
83
84    /// Set the bound on schema-change LRO waits.
85    pub fn with_ddl_timeout_secs(mut self, secs: u64) -> Self {
86        self.ddl_timeout_secs = secs;
87        self
88    }
89
90    /// Validate the config at load time: connection block, batch size, and
91    /// write spec (`key` non-empty for upsert/delete).
92    pub fn validate(&self) -> Result<(), faucet_core::FaucetError> {
93        self.connection.validate()?;
94        faucet_core::validate_batch_size(self.batch_size)?;
95        if self.table_name.trim().is_empty() {
96            return Err(faucet_core::FaucetError::Config(
97                "spanner sink: `table_name` must not be empty".into(),
98            ));
99        }
100        self.write.validate()?;
101        Ok(())
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108    use faucet_core::{WriteMode, WriteSpec};
109    use serde_json::json;
110
111    fn base() -> SpannerSinkConfig {
112        SpannerSinkConfig::new("p", "i", "d", "events")
113    }
114
115    #[test]
116    fn default_config() {
117        let config = base();
118        assert_eq!(config.table_name, "events");
119        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
120        assert_eq!(config.ddl_timeout_secs, 300);
121        assert_eq!(config.connection.max_sessions, 100);
122        assert!(matches!(config.write.write_mode, WriteMode::Append));
123        assert!(config.validate().is_ok());
124    }
125
126    #[test]
127    fn builder_methods() {
128        let config = base()
129            .with_batch_size(50)
130            .with_emulator_host("localhost:9010")
131            .with_ddl_timeout_secs(10);
132        assert_eq!(config.batch_size, 50);
133        assert_eq!(
134            config.connection.emulator_host.as_deref(),
135            Some("localhost:9010")
136        );
137        assert_eq!(config.ddl_timeout_secs, 10);
138    }
139
140    #[test]
141    fn deserializes_flattened_connection_and_write_spec() {
142        let config: SpannerSinkConfig = serde_json::from_value(json!({
143            "project_id": "p",
144            "instance": "i",
145            "database": "d",
146            "table_name": "t",
147            "write_mode": "upsert",
148            "key": ["id"]
149        }))
150        .unwrap();
151        assert_eq!(config.connection.project_id, "p");
152        assert!(matches!(config.write.write_mode, WriteMode::Upsert));
153        assert_eq!(config.write.key, vec!["id".to_string()]);
154        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
155        assert!(config.validate().is_ok());
156    }
157
158    #[test]
159    fn validate_rejects_empty_table_name() {
160        let mut config = base();
161        config.table_name = "  ".into();
162        assert!(config.validate().is_err());
163    }
164
165    #[test]
166    fn validate_rejects_upsert_without_key() {
167        let mut config = base();
168        config.write = WriteSpec {
169            write_mode: WriteMode::Upsert,
170            key: vec![],
171            delete_marker: None,
172        };
173        assert!(config.validate().is_err());
174    }
175
176    #[test]
177    fn validate_rejects_oversized_batch() {
178        let config = base().with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
179        assert!(config.validate().is_err());
180    }
181
182    #[test]
183    fn batch_size_zero_is_accepted_as_sentinel() {
184        let config = base().with_batch_size(0);
185        assert!(config.validate().is_ok());
186    }
187
188    #[test]
189    fn debug_does_not_leak_inline_key() {
190        let mut config = base();
191        config.connection.auth = faucet_common_spanner::SpannerCredentials::ServiceAccountKey {
192            json: "TOPSECRET".into(),
193        };
194        let rendered = format!("{config:?}");
195        assert!(!rendered.contains("TOPSECRET"));
196    }
197}
faucet_sink_spanner/config.rs

faucet_sink_spanner/
config.rs
