faucet_source_clickhouse/
config.rs

1//! Configuration for the ClickHouse query source.
2
3use faucet_common_clickhouse::ClickHouseConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9fn default_batch_size() -> usize {
10    DEFAULT_BATCH_SIZE
11}
12
13/// How the source replicates rows across runs.
14///
15/// Serializes as `{ type: full }` or
16/// `{ type: incremental, column: "...", initial_value: ... }`.
17#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq)]
18#[serde(tag = "type", rename_all = "snake_case")]
19pub enum ClickHouseReplication {
20    /// Every run fetches the full result set (default).
21    #[default]
22    Full,
23    /// Only rows whose `column` is strictly greater than the stored bookmark
24    /// (or `initial_value` on the first run) are emitted.
25    ///
26    /// The bookmark is applied two ways: if the query contains the literal
27    /// token `@bookmark`, it is substituted as an injection-safe SQL literal so
28    /// the server filters (efficient pushdown); the source *also* filters
29    /// client-side as a correctness backstop. The new maximum of `column` is
30    /// persisted on the final page.
31    Incremental {
32        /// Column whose value is the replication cursor (e.g. `updated_at`).
33        column: String,
34        /// Lower bound used on the first run, before any bookmark is stored.
35        initial_value: Value,
36    },
37}
38
39/// Configuration for [`ClickHouseSource`](crate::ClickHouseSource).
40#[derive(Clone, Serialize, Deserialize, JsonSchema)]
41pub struct ClickHouseSourceConfig {
42    /// Connection settings (`url` or `host`, `database`, credentials).
43    #[serde(flatten)]
44    pub connection: ClickHouseConnection,
45    /// SQL `SELECT` query to run. The output format is set to `JSONEachRow`
46    /// via the request settings, so **do not** append a `FORMAT` clause. Use
47    /// the literal `@bookmark` token to push the incremental cursor down into
48    /// the `WHERE` clause; use `{key}` tokens to inject parent-context values in
49    /// a matrix child.
50    pub query: String,
51    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). `0` emits
52    /// the whole result set as a single page. Defaults to
53    /// [`DEFAULT_BATCH_SIZE`].
54    #[serde(default = "default_batch_size")]
55    pub batch_size: usize,
56    /// Replication mode. Defaults to [`ClickHouseReplication::Full`].
57    #[serde(default)]
58    pub replication: ClickHouseReplication,
59    /// Explicit state-store key for the bookmark. When unset, a key is derived
60    /// from the connection host and a query fingerprint.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub state_key: Option<String>,
63}
64
65impl std::fmt::Debug for ClickHouseSourceConfig {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        f.debug_struct("ClickHouseSourceConfig")
68            .field("connection", &self.connection)
69            .field("query", &self.query)
70            .field("batch_size", &self.batch_size)
71            .field("replication", &self.replication)
72            .field("state_key", &self.state_key)
73            .finish()
74    }
75}
76
77impl ClickHouseSourceConfig {
78    /// Build a config from a base URL and query, with defaults elsewhere.
79    pub fn new(url: impl Into<String>, query: impl Into<String>) -> Self {
80        Self {
81            connection: ClickHouseConnection::from_url(url),
82            query: query.into(),
83            batch_size: default_batch_size(),
84            replication: ClickHouseReplication::Full,
85            state_key: None,
86        }
87    }
88
89    /// Set the per-page record count.
90    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
91        self.batch_size = batch_size;
92        self
93    }
94
95    /// Configure incremental replication on `column`, starting at `initial`.
96    pub fn incremental(mut self, column: impl Into<String>, initial: Value) -> Self {
97        self.replication = ClickHouseReplication::Incremental {
98            column: column.into(),
99            initial_value: initial,
100        };
101        self
102    }
103
104    /// Validate connection, batch size, and replication settings.
105    pub fn validate(&self) -> Result<(), FaucetError> {
106        self.connection.validate()?;
107        validate_batch_size(self.batch_size)?;
108        if let ClickHouseReplication::Incremental { column, .. } = &self.replication
109            && column.trim().is_empty()
110        {
111            return Err(FaucetError::Config(
112                "ClickHouse incremental replication requires a non-empty `column`".into(),
113            ));
114        }
115        if self.incremental_without_bookmark_pushdown() {
116            tracing::warn!(
117                "ClickHouse incremental replication query has no `@bookmark` token: the \
118                 cursor is applied client-side only, so the server returns the ENTIRE \
119                 result set on every run (correctness is preserved, but it is a full \
120                 re-scan). Add `@bookmark` to the WHERE clause to push the cursor down, \
121                 e.g. `... WHERE {column} > @bookmark`",
122                column = match &self.replication {
123                    ClickHouseReplication::Incremental { column, .. } => column.as_str(),
124                    _ => "<column>",
125                }
126            );
127        }
128        Ok(())
129    }
130
131    /// `true` when replication is `Incremental` but the query omits the
132    /// `@bookmark` token, so the cursor cannot be pushed down and every run
133    /// re-scans the whole result set. Pure predicate so the load-time warning's
134    /// condition is unit-testable.
135    pub(crate) fn incremental_without_bookmark_pushdown(&self) -> bool {
136        matches!(self.replication, ClickHouseReplication::Incremental { .. })
137            && !self.query.contains("@bookmark")
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use serde_json::json;
145
146    #[test]
147    fn config_flattens_connection_fields() {
148        let cfg: ClickHouseSourceConfig = serde_json::from_value(json!({
149            "url": "http://localhost:8123",
150            "database": "analytics",
151            "query": "SELECT 1",
152        }))
153        .unwrap();
154        assert_eq!(cfg.connection.url.as_deref(), Some("http://localhost:8123"));
155        assert_eq!(cfg.connection.database, "analytics");
156        assert_eq!(cfg.batch_size, DEFAULT_BATCH_SIZE);
157    }
158
159    #[test]
160    fn replication_full_is_default() {
161        let cfg = ClickHouseSourceConfig::new("http://h:8123", "SELECT 1");
162        assert_eq!(cfg.replication, ClickHouseReplication::Full);
163    }
164
165    #[test]
166    fn replication_incremental_parses() {
167        let r: ClickHouseReplication = serde_json::from_value(json!({
168            "type": "incremental",
169            "column": "updated_at",
170            "initial_value": "1970-01-01",
171        }))
172        .unwrap();
173        assert_eq!(
174            r,
175            ClickHouseReplication::Incremental {
176                column: "updated_at".into(),
177                initial_value: json!("1970-01-01"),
178            }
179        );
180    }
181
182    #[test]
183    fn validate_rejects_incremental_without_column() {
184        let cfg = ClickHouseSourceConfig {
185            replication: ClickHouseReplication::Incremental {
186                column: "  ".into(),
187                initial_value: json!(0),
188            },
189            ..ClickHouseSourceConfig::new("http://h:8123", "SELECT 1")
190        };
191        assert!(cfg.validate().is_err());
192    }
193
194    #[test]
195    fn validate_rejects_bad_batch_size() {
196        let cfg = ClickHouseSourceConfig::new("http://h:8123", "SELECT 1")
197            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
198        assert!(cfg.validate().is_err());
199    }
200
201    #[test]
202    fn validate_rejects_missing_endpoint() {
203        let mut cfg = ClickHouseSourceConfig::new("http://h:8123", "SELECT 1");
204        cfg.connection.url = None;
205        assert!(cfg.validate().is_err());
206    }
207
208    #[test]
209    fn incremental_without_bookmark_pushdown_flags_missing_token() {
210        let missing = ClickHouseSourceConfig::new("http://h:8123", "SELECT * FROM t")
211            .incremental("updated_at", json!("1970-01-01"));
212        assert!(missing.incremental_without_bookmark_pushdown());
213        assert!(missing.validate().is_ok(), "warn, not hard error");
214
215        let with_token = ClickHouseSourceConfig::new(
216            "http://h:8123",
217            "SELECT * FROM t WHERE updated_at > @bookmark",
218        )
219        .incremental("updated_at", json!("1970-01-01"));
220        assert!(!with_token.incremental_without_bookmark_pushdown());
221
222        let full = ClickHouseSourceConfig::new("http://h:8123", "SELECT * FROM t");
223        assert!(!full.incremental_without_bookmark_pushdown());
224    }
225
226    #[test]
227    fn debug_masks_password() {
228        let mut cfg = ClickHouseSourceConfig::new("http://h:8123", "SELECT 1");
229        cfg.connection.password = Some("s3cret".into());
230        let dbg = format!("{cfg:?}");
231        assert!(dbg.contains("***"));
232        assert!(!dbg.contains("s3cret"));
233    }
234}
faucet_source_clickhouse/config.rs

faucet_source_clickhouse/
config.rs
