faucet_source_redshift/
config.rs

1//! Amazon Redshift source configuration.
2
3use faucet_common_redshift::RedshiftConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9/// How the Redshift source replicates rows across runs.
10///
11/// Serializes as `{ type: full }` or
12/// `{ type: incremental, column: "...", initial_value: ... }`.
13#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq)]
14#[serde(tag = "type", rename_all = "snake_case")]
15pub enum RedshiftReplication {
16    /// Every run re-fetches the full result set (default).
17    #[default]
18    Full,
19    /// Only rows not yet read, in `column` order, resuming from the stored
20    /// bookmark (or `initial_value` on the first run).
21    ///
22    /// The source wraps the query as
23    /// `SELECT * FROM (<query>) AS q WHERE q."column" >= CAST($n AS <type>) ORDER BY q."column"`
24    /// (the column's type as the server describes it) and drops the rows it
25    /// already emitted at the bookmark value, so rows that share a cursor
26    /// value (batch loads stamped with one `updated_at`, COPY jobs landing in
27    /// several commits) are never skipped. Every page carries a bookmark, so
28    /// a crash replays at most one page. Write `${bookmark}` in the query
29    /// (with `>=`) to apply the cursor further inside it; a strict `>` is
30    /// refused. Rows whose `column` is NULL are never read.
31    Incremental {
32        /// Column whose value is the replication cursor (e.g. `updated_at`).
33        column: String,
34        /// Lower bound (inclusive) used on the first run, before any bookmark
35        /// is stored.
36        initial_value: Value,
37    },
38}
39
40fn default_max_connections() -> u32 {
41    10
42}
43
44fn default_batch_size() -> usize {
45    DEFAULT_BATCH_SIZE
46}
47
48/// Configuration for the Amazon Redshift query source.
49#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
50pub struct RedshiftSourceConfig {
51    /// Connection block (host / port / database / user / credentials / tls),
52    /// flattened to the config top level.
53    #[serde(flatten)]
54    pub connection: RedshiftConnection,
55    /// SQL query to execute. May contain `${field.path}` parent-context tokens
56    /// (resolved to positional binds at runtime) and, for incremental
57    /// replication, a `${bookmark}` token.
58    #[schemars(extend("x-faucet-sql" = true))]
59    pub query: String,
60    /// Positional bind parameters for the query, applied in `$1, $2, …` order
61    /// before any context- or bookmark-derived values. Defaults to empty.
62    #[serde(default)]
63    pub params: Vec<Value>,
64    /// Maximum number of connections in the pool. Defaults to 10.
65    #[serde(default = "default_max_connections")]
66    pub max_connections: u32,
67    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). Rows are
68    /// drained from the `sqlx` cursor and yielded whenever the buffer reaches
69    /// this size. Defaults to [`DEFAULT_BATCH_SIZE`].
70    ///
71    /// `batch_size = 0` is the "no batching" sentinel: the cursor is fully
72    /// drained and the entire result set is emitted in a single page.
73    #[serde(default = "default_batch_size")]
74    pub batch_size: usize,
75    /// Replication mode. Defaults to [`RedshiftReplication::Full`].
76    #[serde(default)]
77    pub replication: RedshiftReplication,
78    /// Explicit state-store key for the incremental bookmark. When unset, a key
79    /// is derived from the host / database and a query fingerprint.
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub state_key: Option<String>,
82}
83
84impl RedshiftSourceConfig {
85    /// Validate the config; returns a human-readable error if invalid.
86    pub fn validate(&self) -> Result<(), FaucetError> {
87        if self.query.trim().is_empty() {
88            return Err(FaucetError::Config(
89                "redshift: `query` must not be empty".into(),
90            ));
91        }
92        faucet_core::validate_batch_size(self.batch_size)?;
93        let RedshiftReplication::Incremental {
94            column,
95            initial_value,
96        } = &self.replication
97        else {
98            return Ok(());
99        };
100        if column.trim().is_empty() {
101            return Err(FaucetError::Config(
102                "redshift: incremental replication `column` must not be empty".into(),
103            ));
104        }
105        if initial_value.is_null() {
106            return Err(FaucetError::Config(
107                "redshift: incremental replication requires a non-null `initial_value`".into(),
108            ));
109        }
110        if faucet_core::replication::cursor::strict_comparison(&self.query, BOOKMARK_TOKEN) {
111            return Err(FaucetError::Config(format!(
112                "redshift: compare `{BOOKMARK_TOKEN}` with `>=`, not `>`: the source re-reads \
113                 the bookmark value and drops the rows it already emitted, and a strict \
114                 comparison skips rows that share it"
115            )));
116        }
117        Ok(())
118    }
119}
120
121/// The token a query uses to read the incremental bookmark
122/// (`WHERE updated_at >= ${bookmark}`).
123pub const BOOKMARK_TOKEN: &str = "${bookmark}";
124
125#[cfg(test)]
126mod tests {
127    use super::*;
128    use faucet_common_redshift::RedshiftConnection;
129    use serde_json::json;
130
131    fn base() -> RedshiftSourceConfig {
132        RedshiftSourceConfig {
133            connection: RedshiftConnection::new("host", "db", "user", "pw"),
134            query: "SELECT * FROM events".into(),
135            params: Vec::new(),
136            max_connections: default_max_connections(),
137            batch_size: DEFAULT_BATCH_SIZE,
138            replication: RedshiftReplication::Full,
139            state_key: None,
140        }
141    }
142
143    #[test]
144    fn valid_config_passes() {
145        base().validate().unwrap();
146    }
147
148    #[test]
149    fn rejects_empty_query() {
150        let mut c = base();
151        c.query = "  ".into();
152        assert!(c.validate().is_err());
153    }
154
155    #[test]
156    fn rejects_empty_incremental_column() {
157        let mut c = base();
158        c.replication = RedshiftReplication::Incremental {
159            column: "".into(),
160            initial_value: json!(0),
161        };
162        assert!(c.validate().is_err());
163    }
164
165    #[test]
166    fn rejects_bad_incremental_settings() {
167        for (query, initial, needle) in [
168            ("SELECT * FROM t", json!(null), "initial_value"),
169            ("SELECT * FROM t WHERE ts > ${bookmark}", json!(0), ">="),
170        ] {
171            let mut c = base();
172            c.query = query.into();
173            c.replication = RedshiftReplication::Incremental {
174                column: "ts".into(),
175                initial_value: initial,
176            };
177            let err = c.validate().unwrap_err().to_string();
178            assert!(err.contains(needle), "{err}");
179        }
180    }
181
182    #[test]
183    fn rejects_oversized_batch() {
184        let mut c = base();
185        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
186        assert!(c.validate().is_err());
187    }
188
189    #[test]
190    fn deserializes_flattened_connection_and_defaults() {
191        let json = r#"{
192            "host": "h",
193            "database": "db",
194            "user": "u",
195            "credentials": {"type": "password", "config": {"password": "pw"}},
196            "query": "SELECT 1"
197        }"#;
198        let c: RedshiftSourceConfig = serde_json::from_str(json).unwrap();
199        assert_eq!(c.connection.host, "h");
200        assert_eq!(c.connection.port, faucet_common_redshift::DEFAULT_PORT);
201        assert_eq!(c.max_connections, 10);
202        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
203        assert!(matches!(c.replication, RedshiftReplication::Full));
204        c.validate().unwrap();
205    }
206
207    #[test]
208    fn deserializes_incremental_replication() {
209        let json = r#"{
210            "host": "h",
211            "database": "db",
212            "user": "u",
213            "credentials": {"type": "password", "config": {"password": "pw"}},
214            "query": "SELECT * FROM t WHERE ts >= ${bookmark}",
215            "replication": {"type": "incremental", "column": "ts", "initial_value": "2026-01-01"},
216            "state_key": "my-key"
217        }"#;
218        let c: RedshiftSourceConfig = serde_json::from_str(json).unwrap();
219        match &c.replication {
220            RedshiftReplication::Incremental {
221                column,
222                initial_value,
223            } => {
224                assert_eq!(column, "ts");
225                assert_eq!(initial_value, &json!("2026-01-01"));
226            }
227            _ => panic!("expected incremental"),
228        }
229        assert_eq!(c.state_key.as_deref(), Some("my-key"));
230    }
231}
faucet_source_redshift/config.rs

faucet_source_redshift/
config.rs
