faucet_sink_redshift/
config.rs

1//! Amazon Redshift sink configuration.
2
3use faucet_common_redshift::RedshiftConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// How the sink loads rows into Redshift.
9#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
10#[serde(rename_all = "snake_case")]
11pub enum RedshiftWriteStrategy {
12    /// Stage each page to S3 and bulk-load it with `COPY … FROM 's3://…'` — the
13    /// default and by far the fastest path for Redshift (the recommended way to
14    /// load data). Requires `staging_bucket` and `iam_role`.
15    #[default]
16    Copy,
17    /// Multi-row `INSERT INTO … VALUES (…), (…)`. Portable and needs no S3, but
18    /// much slower than `COPY` for anything beyond small batches. Redshift does
19    /// not recommend row-by-row inserts for bulk data.
20    Insert,
21}
22
23impl RedshiftWriteStrategy {
24    /// Lower-case wire name, for error messages.
25    pub fn as_str(&self) -> &'static str {
26        match self {
27            Self::Copy => "copy",
28            Self::Insert => "insert",
29        }
30    }
31}
32
33/// Format of the staged file that `COPY` reads.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
35#[serde(rename_all = "snake_case")]
36pub enum RedshiftCopyFormat {
37    /// Newline-delimited JSON objects, loaded with `FORMAT AS JSON 'auto'`.
38    /// Maps by column **name** (order-independent) and handles NULLs and typed
39    /// columns cleanly — the default.
40    #[default]
41    Jsonl,
42    /// RFC-4180 CSV, loaded with `FORMAT AS CSV`. Column order is taken from the
43    /// destination table's schema and passed explicitly in the `COPY` column
44    /// list.
45    Csv,
46}
47
48fn default_batch_size() -> usize {
49    DEFAULT_BATCH_SIZE
50}
51
52fn default_max_connections() -> u32 {
53    5
54}
55
56/// Configuration for the Amazon Redshift sink.
57#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
58pub struct RedshiftSinkConfig {
59    /// Connection block (host / port / database / user / credentials / tls),
60    /// flattened to the config top level.
61    #[serde(flatten)]
62    pub connection: RedshiftConnection,
63    /// Target table name.
64    pub table_name: String,
65    /// Optional schema (namespace) qualifying [`table_name`](Self::table_name).
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub schema: Option<String>,
68    /// How rows are loaded. Defaults to [`RedshiftWriteStrategy::Copy`].
69    #[serde(default)]
70    pub write_strategy: RedshiftWriteStrategy,
71    /// Staged-file format for the `COPY` path. Defaults to
72    /// [`RedshiftCopyFormat::Jsonl`]. Ignored by the `insert` strategy.
73    #[serde(default)]
74    pub copy_format: RedshiftCopyFormat,
75    /// S3 bucket used to stage `COPY` files. **Required** when
76    /// `write_strategy: copy`.
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub staging_bucket: Option<String>,
79    /// Key prefix for staged objects (e.g. `redshift-staging/`). Defaults to
80    /// empty.
81    #[serde(default)]
82    pub staging_prefix: String,
83    /// IAM role ARN Redshift assumes to read the staged file
84    /// (`COPY … IAM_ROLE '<arn>'`). **Required** when `write_strategy: copy`.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub iam_role: Option<String>,
87    /// AWS region of the staging bucket (used for both the S3 client and the
88    /// `COPY … REGION '<region>'` clause). `None` uses the SDK default.
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub region: Option<String>,
91    /// Custom endpoint URL for S3-compatible services (e.g. MinIO) — testing
92    /// aid; production loads use real S3.
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub endpoint_url: Option<String>,
95    /// Rows per load unit. For `insert`, the per-statement multi-row chunk size;
96    /// for `copy`, the number of rows per staged S3 object. Defaults to
97    /// [`DEFAULT_BATCH_SIZE`]. `0` = one unit for the whole page.
98    #[serde(default = "default_batch_size")]
99    pub batch_size: usize,
100    /// Maximum number of connections in the pool. Defaults to 5.
101    #[serde(default = "default_max_connections")]
102    pub max_connections: u32,
103}
104
105impl RedshiftSinkConfig {
106    /// Validate the config. Enforces that the `copy` strategy has a staging
107    /// bucket and IAM role.
108    pub fn validate(&self) -> Result<(), FaucetError> {
109        if self.table_name.trim().is_empty() {
110            return Err(FaucetError::Config(
111                "redshift sink: `table_name` must not be empty".into(),
112            ));
113        }
114        if self.write_strategy == RedshiftWriteStrategy::Copy {
115            let bucket_ok = self
116                .staging_bucket
117                .as_ref()
118                .is_some_and(|b| !b.trim().is_empty());
119            if !bucket_ok {
120                return Err(FaucetError::Config(
121                    "redshift sink: write_strategy: copy requires a non-empty `staging_bucket`"
122                        .into(),
123                ));
124            }
125            let role_ok = self.iam_role.as_ref().is_some_and(|r| !r.trim().is_empty());
126            if !role_ok {
127                return Err(FaucetError::Config(
128                    "redshift sink: write_strategy: copy requires a non-empty `iam_role`".into(),
129                ));
130            }
131        }
132        faucet_core::validate_batch_size(self.batch_size)?;
133        Ok(())
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140    use faucet_common_redshift::RedshiftConnection;
141
142    fn base() -> RedshiftSinkConfig {
143        RedshiftSinkConfig {
144            connection: RedshiftConnection::new("host", "db", "user", "pw"),
145            table_name: "events".into(),
146            schema: None,
147            write_strategy: RedshiftWriteStrategy::Copy,
148            copy_format: RedshiftCopyFormat::Jsonl,
149            staging_bucket: Some("stage".into()),
150            staging_prefix: String::new(),
151            iam_role: Some("arn:aws:iam::123:role/redshift".into()),
152            region: None,
153            endpoint_url: None,
154            batch_size: DEFAULT_BATCH_SIZE,
155            max_connections: default_max_connections(),
156        }
157    }
158
159    #[test]
160    fn valid_copy_config_passes() {
161        base().validate().unwrap();
162    }
163
164    #[test]
165    fn valid_insert_config_needs_no_bucket() {
166        let mut c = base();
167        c.write_strategy = RedshiftWriteStrategy::Insert;
168        c.staging_bucket = None;
169        c.iam_role = None;
170        c.validate().unwrap();
171    }
172
173    #[test]
174    fn copy_requires_bucket() {
175        let mut c = base();
176        c.staging_bucket = None;
177        match c.validate() {
178            Err(FaucetError::Config(m)) => assert!(m.contains("staging_bucket"), "got: {m}"),
179            other => panic!("expected Config error, got {other:?}"),
180        }
181    }
182
183    #[test]
184    fn copy_requires_iam_role() {
185        let mut c = base();
186        c.iam_role = Some("  ".into());
187        match c.validate() {
188            Err(FaucetError::Config(m)) => assert!(m.contains("iam_role"), "got: {m}"),
189            other => panic!("expected Config error, got {other:?}"),
190        }
191    }
192
193    #[test]
194    fn rejects_empty_table_name() {
195        let mut c = base();
196        c.table_name = " ".into();
197        assert!(c.validate().is_err());
198    }
199
200    #[test]
201    fn rejects_oversized_batch() {
202        let mut c = base();
203        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
204        assert!(c.validate().is_err());
205    }
206
207    #[test]
208    fn defaults_copy_and_jsonl() {
209        let json = r#"{
210            "host": "h", "database": "db", "user": "u",
211            "credentials": {"type": "password", "config": {"password": "pw"}},
212            "table_name": "t",
213            "staging_bucket": "b",
214            "iam_role": "arn:x"
215        }"#;
216        let c: RedshiftSinkConfig = serde_json::from_str(json).unwrap();
217        assert_eq!(c.write_strategy, RedshiftWriteStrategy::Copy);
218        assert_eq!(c.copy_format, RedshiftCopyFormat::Jsonl);
219        assert_eq!(c.max_connections, 5);
220        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
221        c.validate().unwrap();
222    }
223
224    #[test]
225    fn write_strategy_round_trips() {
226        assert_eq!(RedshiftWriteStrategy::Copy.as_str(), "copy");
227        assert_eq!(RedshiftWriteStrategy::Insert.as_str(), "insert");
228    }
229}
faucet_sink_redshift/config.rs

faucet_sink_redshift/
config.rs
