faucet_source_duckdb/
config.rs

1//! DuckDB source configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// Configuration for the DuckDB query source.
8#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
9pub struct DuckdbSourceConfig {
10    /// Path to the DuckDB database file, or `:memory:` for an in-memory
11    /// database. A `duckdb://` / `duckdb:` scheme prefix is accepted and
12    /// stripped.
13    pub database: String,
14    /// SQL query to execute.
15    pub query: String,
16    /// Open the database read-only. Defaults to `false` (read-write).
17    ///
18    /// Set `true` to attach to a database file another process holds open —
19    /// DuckDB permits multiple read-only connections to one file but only a
20    /// single read-write connection.
21    #[serde(default)]
22    pub read_only: bool,
23    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). Rows are
24    /// drained from the DuckDB result and yielded whenever the buffer reaches
25    /// this size. Defaults to [`DEFAULT_BATCH_SIZE`].
26    ///
27    /// `batch_size = 0` is the "no batching" sentinel: the entire result set is
28    /// emitted in a single page. Useful for small lookup tables.
29    #[serde(default = "default_batch_size")]
30    pub batch_size: usize,
31}
32
33fn default_batch_size() -> usize {
34    DEFAULT_BATCH_SIZE
35}
36
37impl DuckdbSourceConfig {
38    /// Create a new config with the required database path and query.
39    pub fn new(database: impl Into<String>, query: impl Into<String>) -> Self {
40        Self {
41            database: database.into(),
42            query: query.into(),
43            read_only: false,
44            batch_size: DEFAULT_BATCH_SIZE,
45        }
46    }
47
48    /// Set the per-page row count for
49    /// [`Source::stream_pages`](faucet_core::Source::stream_pages).
50    ///
51    /// Pass `0` to opt out of batching — the entire result set is emitted in a
52    /// single [`StreamPage`](faucet_core::StreamPage).
53    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
54        self.batch_size = batch_size;
55        self
56    }
57
58    /// Open the database read-only.
59    pub fn read_only(mut self, read_only: bool) -> Self {
60        self.read_only = read_only;
61        self
62    }
63
64    /// The filesystem path with any `duckdb://` / `duckdb:` scheme stripped.
65    pub(crate) fn resolved_path(&self) -> &str {
66        self.database
67            .trim_start_matches("duckdb://")
68            .trim_start_matches("duckdb:")
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use super::*;
75
76    #[test]
77    fn default_config() {
78        let config = DuckdbSourceConfig::new("data.duckdb", "SELECT * FROM events");
79        assert_eq!(config.database, "data.duckdb");
80        assert_eq!(config.query, "SELECT * FROM events");
81        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
82        assert!(!config.read_only);
83    }
84
85    #[test]
86    fn resolved_path_strips_scheme() {
87        assert_eq!(
88            DuckdbSourceConfig::new("duckdb:///tmp/a.duckdb", "SELECT 1").resolved_path(),
89            "/tmp/a.duckdb"
90        );
91        assert_eq!(
92            DuckdbSourceConfig::new("duckdb::memory:", "SELECT 1").resolved_path(),
93            ":memory:"
94        );
95        assert_eq!(
96            DuckdbSourceConfig::new(":memory:", "SELECT 1").resolved_path(),
97            ":memory:"
98        );
99    }
100
101    #[test]
102    fn with_batch_size_overrides_default() {
103        let config = DuckdbSourceConfig::new(":memory:", "SELECT 1").with_batch_size(500);
104        assert_eq!(config.batch_size, 500);
105    }
106
107    #[test]
108    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
109        let config = DuckdbSourceConfig::new(":memory:", "SELECT 1").with_batch_size(0);
110        assert_eq!(config.batch_size, 0);
111        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
112    }
113
114    #[test]
115    fn batch_size_deserializes_from_json() {
116        let json = r#"{
117            "database": ":memory:",
118            "query": "SELECT 1",
119            "batch_size": 250
120        }"#;
121        let config: DuckdbSourceConfig = serde_json::from_str(json).unwrap();
122        assert_eq!(config.batch_size, 250);
123        assert!(!config.read_only);
124    }
125}
faucet_source_duckdb/config.rs

faucet_source_duckdb/
config.rs
