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

laminar_connectors/mongodb/
mod.rs
