faucet_sink_duckdb/
config.rs

1//! DuckDB 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 DuckdbColumnMapping {
11    /// Insert each record as a single JSON text column.
12    Json { column: String },
13    /// Map top-level JSON keys directly to table columns. Only keys that match
14    /// existing columns are inserted; extra keys are ignored.
15    AutoMap,
16}
17
18impl Default for DuckdbColumnMapping {
19    fn default() -> Self {
20        Self::Json {
21            column: "data".into(),
22        }
23    }
24}
25
26/// Configuration for the DuckDB sink.
27#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
28pub struct DuckdbSinkConfig {
29    /// Path to the DuckDB database file, or `:memory:`. A `duckdb://` /
30    /// `duckdb:` scheme prefix is accepted and stripped. The target table must
31    /// already exist.
32    pub database: String,
33    /// Target table name.
34    pub table_name: String,
35    /// How to map JSON records to columns.
36    #[serde(default)]
37    pub column_mapping: DuckdbColumnMapping,
38    /// Maximum number of rows per multi-row INSERT. Defaults to
39    /// [`DEFAULT_BATCH_SIZE`] (1000). `batch_size = 0` writes the entire
40    /// upstream slice as a single multi-row INSERT (no re-chunking).
41    #[serde(default = "default_batch_size")]
42    pub batch_size: usize,
43}
44
45fn default_batch_size() -> usize {
46    DEFAULT_BATCH_SIZE
47}
48
49impl DuckdbSinkConfig {
50    /// Create a new config with required fields and sensible defaults.
51    pub fn new(database: impl Into<String>, table_name: impl Into<String>) -> Self {
52        Self {
53            database: database.into(),
54            table_name: table_name.into(),
55            column_mapping: DuckdbColumnMapping::default(),
56            batch_size: DEFAULT_BATCH_SIZE,
57        }
58    }
59
60    /// Set the column mapping strategy.
61    pub fn column_mapping(mut self, mapping: DuckdbColumnMapping) -> Self {
62        self.column_mapping = mapping;
63        self
64    }
65
66    /// Set the maximum number of rows per multi-row INSERT. `0` disables
67    /// re-chunking.
68    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
69        self.batch_size = batch_size;
70        self
71    }
72
73    /// The filesystem path with any `duckdb://` / `duckdb:` scheme stripped.
74    pub(crate) fn resolved_path(&self) -> &str {
75        self.database
76            .trim_start_matches("duckdb://")
77            .trim_start_matches("duckdb:")
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84
85    #[test]
86    fn default_config() {
87        let config = DuckdbSinkConfig::new(":memory:", "events");
88        assert_eq!(config.table_name, "events");
89        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
90        assert!(matches!(
91            config.column_mapping,
92            DuckdbColumnMapping::Json { ref column } if column == "data"
93        ));
94    }
95
96    #[test]
97    fn builder_methods() {
98        let config = DuckdbSinkConfig::new(":memory:", "events")
99            .column_mapping(DuckdbColumnMapping::AutoMap)
100            .with_batch_size(100);
101        assert_eq!(config.batch_size, 100);
102        assert!(matches!(
103            config.column_mapping,
104            DuckdbColumnMapping::AutoMap
105        ));
106    }
107
108    #[test]
109    fn batch_size_deserializes_and_defaults() {
110        let json = r#"{ "database": ":memory:", "table_name": "events" }"#;
111        let config: DuckdbSinkConfig = serde_json::from_str(json).unwrap();
112        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
113        let json2 = r#"{ "database": ":memory:", "table_name": "e", "batch_size": 250 }"#;
114        let config2: DuckdbSinkConfig = serde_json::from_str(json2).unwrap();
115        assert_eq!(config2.batch_size, 250);
116    }
117}
faucet_sink_duckdb/config.rs

faucet_sink_duckdb/
config.rs
