faucet_sink_sqlite/
config.rs

1//! SQLite sink configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// How to map JSON records to table columns.
8#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
9#[serde(rename_all = "snake_case")]
10pub enum SqliteColumnMapping {
11    /// Insert each record as a single JSON text column. The column name
12    /// defaults to `"data"` but can be overridden.
13    Json { column: String },
14    /// Map top-level JSON keys directly to table columns.
15    /// Only keys that match existing columns are inserted; extra keys are ignored.
16    AutoMap,
17}
18
19impl Default for SqliteColumnMapping {
20    fn default() -> Self {
21        Self::Json {
22            column: "data".into(),
23        }
24    }
25}
26
27/// Configuration for the SQLite sink.
28#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
29pub struct SqliteSinkConfig {
30    /// SQLite database URL (file path or `:memory:`).
31    pub database_url: String,
32    /// Target table name.
33    pub table_name: String,
34    /// How to map JSON records to columns.
35    pub column_mapping: SqliteColumnMapping,
36    /// Maximum number of rows per multi-row INSERT. Defaults to
37    /// [`DEFAULT_BATCH_SIZE`] (1000); the recommended value for SQLite.
38    ///
39    /// When the upstream `StreamPage` carries more records than `batch_size`,
40    /// the sink slices the page into `batch_size`-row chunks and issues one
41    /// multi-row INSERT per chunk, each wrapped in its own `BEGIN`/`COMMIT`
42    /// transaction. When `batch_size = 0`, the entire slice is written as a
43    /// single multi-row INSERT inside one transaction — useful when the
44    /// source already emits pages sized for SQLite (e.g. a Postgres source
45    /// configured with `batch_size: 1000`).
46    ///
47    /// `batch_size = 0` is the "no batching" sentinel: upstream framing is
48    /// preserved exactly, no internal re-chunking is performed.
49    #[serde(default = "default_batch_size")]
50    pub batch_size: usize,
51    /// Maximum number of connections in the pool. Defaults to `1`.
52    ///
53    /// SQLite serializes writers at the file level, so a single SQLite file
54    /// can only ever have one writer at a time. A multi-connection pool against
55    /// one file therefore races for the write lock and risks `SQLITE_BUSY`
56    /// errors under concurrency; one writer is the safe default. Raise this only
57    /// if your workload is read-heavy against a WAL database (the sink opens the
58    /// pool in WAL mode with a `busy_timeout`, so extra connections can still
59    /// read concurrently with the single writer).
60    #[serde(default = "default_max_connections")]
61    pub max_connections: u32,
62    /// Write mode, key columns, and optional delete marker. `write_mode`
63    /// defaults to `append`. Upsert/delete require `column_mapping: auto_map`
64    /// and a UNIQUE/PRIMARY KEY constraint on `key`.
65    #[serde(flatten)]
66    pub write: faucet_core::WriteSpec,
67}
68
69fn default_batch_size() -> usize {
70    DEFAULT_BATCH_SIZE
71}
72
73fn default_max_connections() -> u32 {
74    1
75}
76
77impl SqliteSinkConfig {
78    /// Create a new config with required fields and sensible defaults.
79    pub fn new(database_url: impl Into<String>, table_name: impl Into<String>) -> Self {
80        Self {
81            database_url: database_url.into(),
82            table_name: table_name.into(),
83            column_mapping: SqliteColumnMapping::default(),
84            batch_size: DEFAULT_BATCH_SIZE,
85            max_connections: default_max_connections(),
86            write: faucet_core::WriteSpec::default(),
87        }
88    }
89
90    /// Set the column mapping strategy.
91    pub fn column_mapping(mut self, mapping: SqliteColumnMapping) -> Self {
92        self.column_mapping = mapping;
93        self
94    }
95
96    /// Set the maximum number of rows per multi-row INSERT.
97    ///
98    /// Pass `0` to opt out of re-chunking — the entire records slice handed
99    /// to `write_batch` is written in a single multi-row INSERT inside one
100    /// transaction, preserving upstream `StreamPage` framing.
101    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
102        self.batch_size = batch_size;
103        self
104    }
105
106    /// Set the maximum number of connections in the pool.
107    pub fn max_connections(mut self, n: u32) -> Self {
108        self.max_connections = n;
109        self
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116
117    #[test]
118    fn default_config() {
119        let config = SqliteSinkConfig::new("sqlite::memory:", "events");
120        assert_eq!(config.table_name, "events");
121        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
122        assert!(matches!(
123            config.column_mapping,
124            SqliteColumnMapping::Json { ref column } if column == "data"
125        ));
126    }
127
128    #[test]
129    fn default_max_connections_is_one() {
130        // SQLite serializes writers at the file level, so a single-writer pool
131        // is the safe default — a multi-connection pool against one SQLite file
132        // races for the write lock and risks SQLITE_BUSY (audit #146 LOW).
133        let config = SqliteSinkConfig::new("sqlite::memory:", "events");
134        assert_eq!(config.max_connections, 1);
135    }
136
137    #[test]
138    fn default_max_connections_deserializes_to_one_when_absent() {
139        let json = r#"{
140            "database_url": "sqlite::memory:",
141            "table_name": "events",
142            "column_mapping": {"json": {"column": "data"}}
143        }"#;
144        let config: SqliteSinkConfig = serde_json::from_str(json).unwrap();
145        assert_eq!(config.max_connections, 1);
146    }
147
148    #[test]
149    fn builder_methods() {
150        let config = SqliteSinkConfig::new("sqlite::memory:", "events")
151            .column_mapping(SqliteColumnMapping::AutoMap)
152            .with_batch_size(100);
153        assert_eq!(config.batch_size, 100);
154        assert!(matches!(
155            config.column_mapping,
156            SqliteColumnMapping::AutoMap
157        ));
158    }
159
160    #[test]
161    fn json_custom_column() {
162        let config = SqliteSinkConfig::new("sqlite::memory:", "events").column_mapping(
163            SqliteColumnMapping::Json {
164                column: "payload".into(),
165            },
166        );
167        assert!(matches!(
168            config.column_mapping,
169            SqliteColumnMapping::Json { ref column } if column == "payload"
170        ));
171    }
172
173    #[test]
174    fn config_with_file_path() {
175        let config = SqliteSinkConfig::new("/tmp/test.db", "events");
176        assert_eq!(config.database_url, "/tmp/test.db");
177    }
178
179    #[test]
180    fn batch_size_defaults_to_default_batch_size() {
181        let config = SqliteSinkConfig::new("sqlite::memory:", "events");
182        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
183    }
184
185    #[test]
186    fn with_batch_size_overrides_default() {
187        let config = SqliteSinkConfig::new("sqlite::memory:", "events").with_batch_size(250);
188        assert_eq!(config.batch_size, 250);
189    }
190
191    #[test]
192    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
193        let config = SqliteSinkConfig::new("sqlite::memory:", "events").with_batch_size(0);
194        assert_eq!(config.batch_size, 0);
195        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
196    }
197
198    #[test]
199    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
200        let config = SqliteSinkConfig::new("sqlite::memory:", "events")
201            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
202        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
203    }
204
205    #[test]
206    fn batch_size_deserializes_from_json() {
207        let json = r#"{
208            "database_url": "sqlite::memory:",
209            "table_name": "events",
210            "column_mapping": {"json": {"column": "data"}},
211            "batch_size": 250,
212            "max_connections": 5
213        }"#;
214        let config: SqliteSinkConfig = serde_json::from_str(json).unwrap();
215        assert_eq!(config.batch_size, 250);
216    }
217
218    #[test]
219    fn batch_size_defaults_when_absent_in_json() {
220        let json = r#"{
221            "database_url": "sqlite::memory:",
222            "table_name": "events",
223            "column_mapping": {"json": {"column": "data"}},
224            "max_connections": 5
225        }"#;
226        let config: SqliteSinkConfig = serde_json::from_str(json).unwrap();
227        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
228    }
229}
faucet_sink_sqlite/config.rs

faucet_sink_sqlite/
config.rs
