laminar_connectors/mongodb/
mod.rs

1//! `MongoDB` CDC source and sink connectors.
2
3pub mod change_event;
4pub mod config;
5pub mod lookup;
6pub mod metrics;
7pub mod sink;
8pub mod source;
9pub mod timeseries;
10pub mod write_model;
11
12// Re-export primary types at module level.
13pub use config::{FullDocumentMode, MongoDbSinkConfig, MongoDbSourceConfig};
14pub use sink::MongoDbSink;
15pub use source::{mongodb_cdc_envelope_schema, MongoDbCdcSource};
16pub use timeseries::{CollectionKind, TimeSeriesConfig, TimeSeriesGranularity};
17pub use write_model::WriteMode;
18
19const MONGODB_LOOKUP_PROPERTIES: &[&str] = &[
20    "connection.uri",
21    "database",
22    "collection",
23    "laminar.source.name",
24    "_arrow_schema",
25    "_primary_key_columns",
26];
27
28use std::sync::Arc;
29
30use crate::config::{ConfigKeySpec, ConnectorInfo};
31use crate::registry::ConnectorRegistry;
32
33/// Registers the `MongoDB` CDC source connector with the given registry.
34///
35/// # Errors
36///
37/// Returns an error if the connector name is already registered or the registry is frozen.
38pub fn register_mongodb_cdc_source(
39    registry: &ConnectorRegistry,
40) -> Result<(), crate::error::ConnectorError> {
41    let info = ConnectorInfo {
42        name: "mongodb-cdc".to_string(),
43        display_name: "MongoDB CDC Source".to_string(),
44        version: env!("CARGO_PKG_VERSION").to_string(),
45        is_source: true,
46        is_sink: false,
47        config_keys: mongodb_cdc_config_keys(),
48    };
49
50    registry.register_source(
51        "mongodb-cdc",
52        info,
53        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
54            Ok(Box::new(MongoDbCdcSource::new(
55                MongoDbSourceConfig::default(),
56                registry.map(Arc::as_ref),
57            )))
58        }),
59    )?;
60
61    // On-demand (partial cache mode) lookup source: find({ pk: { $in: [...] } }).
62    registry.register_lookup_source(
63        "mongodb",
64        ConnectorInfo {
65            name: "mongodb".to_string(),
66            display_name: "MongoDB Lookup Source".to_string(),
67            version: env!("CARGO_PKG_VERSION").to_string(),
68            is_source: true,
69            is_sink: false,
70            config_keys: mongodb_lookup_config_keys(),
71        },
72        Arc::new(MongoLookupFactory),
73    )
74}
75
76struct MongoLookupFactory;
77
78#[async_trait::async_trait]
79impl crate::registry::LookupSourceFactory for MongoLookupFactory {
80    async fn build(
81        &self,
82        config: crate::config::ConnectorConfig,
83        declared_schema: Option<arrow_schema::SchemaRef>,
84    ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, crate::error::ConnectorError>
85    {
86        use crate::mongodb::lookup::{MongoLookupSource, MongoLookupSourceConfig};
87
88        let schema = declared_schema.ok_or_else(|| {
89            crate::error::ConnectorError::ConfigurationError(
90                "mongodb lookup source requires a declared table schema".into(),
91            )
92        })?;
93
94        let pk_columns: Vec<String> = config
95            .get("_primary_key_columns")
96            .unwrap_or("")
97            .split(',')
98            .map(|s| s.trim().to_string())
99            .filter(|s| !s.is_empty())
100            .collect();
101        if pk_columns.is_empty() {
102            return Err(crate::error::ConnectorError::ConfigurationError(
103                "mongodb lookup source requires primary key columns".into(),
104            ));
105        }
106
107        config.reject_unknown_properties(MONGODB_LOOKUP_PROPERTIES, "MongoDB lookup")?;
108        let lookup_config = MongoLookupSourceConfig {
109            connection_uri: config.require("connection.uri")?.to_string(),
110            database: config.require("database")?.to_string(),
111            collection: config.require("collection")?.to_string(),
112            primary_key_columns: pk_columns,
113            schema,
114        };
115
116        let source = MongoLookupSource::open(lookup_config).await?;
117        Ok(Arc::new(source) as Arc<dyn laminar_core::lookup::source::LookupSourceDyn>)
118    }
119}
120
121/// Registers the `MongoDB` sink connector with the given registry.
122///
123/// # Errors
124///
125/// Returns an error if the connector name is already registered or the registry is frozen.
126pub fn register_mongodb_sink(
127    registry: &ConnectorRegistry,
128) -> Result<(), crate::error::ConnectorError> {
129    let info = ConnectorInfo {
130        name: "mongodb-sink".to_string(),
131        display_name: "MongoDB Sink".to_string(),
132        version: env!("CARGO_PKG_VERSION").to_string(),
133        is_source: false,
134        is_sink: true,
135        config_keys: mongodb_sink_config_keys(),
136    };
137
138    registry.register_sink(
139        "mongodb-sink",
140        info,
141        Arc::new(|config, registry: Option<&Arc<prometheus::Registry>>| {
142            MongoDbSink::from_connector_config(config, registry.map(Arc::as_ref))
143                .map(|sink| Box::new(sink) as Box<dyn crate::connector::SinkConnector>)
144        }),
145    )
146}
147
148fn mongodb_cdc_config_keys() -> Vec<ConfigKeySpec> {
149    vec![
150        ConfigKeySpec::required("connection.uri", "MongoDB connection URI"),
151        ConfigKeySpec::required("database", "Database name"),
152        ConfigKeySpec::required("collection", "Fixed collection name"),
153        ConfigKeySpec::optional(
154            "full.document.mode",
155            "Deterministic full document mode (delta or required post-image)",
156            "delta",
157        ),
158        ConfigKeySpec::optional(
159            "pipeline",
160            "JSON array of up to 64 $match stages (maximum 256 KiB)",
161            "[]",
162        ),
163        ConfigKeySpec::optional(
164            "max.buffered.bytes",
165            "Max retained decoded bytes before backpressure (1 MiB to 4 GiB)",
166            config::DEFAULT_MAX_BUFFERED_BYTES.to_string(),
167        ),
168    ]
169}
170
171fn mongodb_sink_config_keys() -> Vec<ConfigKeySpec> {
172    vec![
173        ConfigKeySpec::required("connection.uri", "MongoDB connection URI"),
174        ConfigKeySpec::required("database", "Target database name"),
175        ConfigKeySpec::required("collection", "Target collection name"),
176        ConfigKeySpec::optional("flush.interval.ms", "Max time between flushes (ms)", "250"),
177        ConfigKeySpec::optional(
178            "write.mode",
179            "Write operation mode (insert, upsert, cdc_replay)",
180            "insert",
181        ),
182        ConfigKeySpec::optional(
183            "write.mode.key_fields",
184            "Comma-separated key fields to match documents in upsert mode",
185            "",
186        ),
187        ConfigKeySpec::optional(
188            "timeseries.time_field",
189            "The field in each document containing the date",
190            "",
191        ),
192        ConfigKeySpec::optional(
193            "timeseries.meta_field",
194            "An optional field labeling the data source",
195            "",
196        ),
197        ConfigKeySpec::optional(
198            "timeseries.granularity",
199            "Bucketing granularity (seconds, minutes, hours, custom)",
200            "seconds",
201        ),
202        ConfigKeySpec::optional(
203            "timeseries.bucket_max_span_seconds",
204            "Max span of a single bucket in seconds (custom granularity)",
205            "",
206        ),
207        ConfigKeySpec::optional(
208            "timeseries.bucket_rounding_seconds",
209            "Rounding boundary in seconds (custom granularity)",
210            "",
211        ),
212        ConfigKeySpec::optional(
213            "timeseries.expire_after_seconds",
214            "TTL in seconds (automatically delete documents after this span)",
215            "",
216        ),
217        ConfigKeySpec::optional(
218            "sink.write.timeout.ms",
219            "Complete MongoDB sink write deadline in milliseconds",
220            "30000",
221        ),
222    ]
223}
224
225fn mongodb_lookup_config_keys() -> Vec<ConfigKeySpec> {
226    vec![
227        ConfigKeySpec::required("connection.uri", "MongoDB connection URI"),
228        ConfigKeySpec::required("database", "Database name"),
229        ConfigKeySpec::required("collection", "Collection name"),
230    ]
231}
232
233#[cfg(test)]
234mod tests;
laminar_connectors/mongodb/mod.rs

laminar_connectors/mongodb/
mod.rs
