laminar_connectors/mongodb/
write_model.rs

1//! Write mode configuration for the `MongoDB` sink connector.
2//!
3//! Defines [`WriteMode`] which determines how incoming `RecordBatch` rows
4//! are translated into `MongoDB` write operations.
5
6use crate::error::ConnectorError;
7
8/// Write operation mode for the `MongoDB` sink.
9///
10/// Determines how incoming `RecordBatch` rows are translated into
11/// `MongoDB` write operations.
12#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
13#[serde(tag = "mode", rename_all = "snake_case")]
14pub enum WriteMode {
15    /// Append-only inserts using `insertOne` / `insertMany`.
16    #[default]
17    Insert,
18
19    /// Upsert by caller-supplied key fields. Uses `replaceOne` with
20    /// `upsert: true`, keyed by the specified fields.
21    Upsert {
22        /// Fields used to match existing documents for upsert.
23        key_fields: Vec<String>,
24    },
25
26    /// Routes operations based on the incoming event's `operationType`.
27    ///
28    /// Only valid for `LaminarEvent<MongoDbChangeEvent>` (CDC fan-out
29    /// replication). Maps operations as follows:
30    ///
31    /// - `insert` → idempotent `replaceOne(..., upsert: true)` by document key
32    /// - `update` → `updateOne` using `$set`/`$unset` from `updateDescription`
33    /// - `replace` → `replaceOne` with `upsert: true`
34    /// - `delete` → `deleteOne` using `documentKey._id`
35    /// - lifecycle and expanded events → rejected because a fixed destination cannot replay them
36    CdcReplay,
37}
38
39/// Validates that a write mode is compatible with time series collections.
40///
41/// Time series collections only accept `Insert`. Any other mode returns
42/// an error.
43///
44/// # Errors
45///
46/// Returns `ConnectorError::ConfigurationError` if the mode is not `Insert`.
47pub fn validate_timeseries_write_mode(mode: &WriteMode) -> Result<(), ConnectorError> {
48    if matches!(mode, WriteMode::Insert) {
49        Ok(())
50    } else {
51        Err(ConnectorError::ConfigurationError(format!(
52            "time series collections only support Insert write mode, got: {mode:?}"
53        )))
54    }
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60
61    #[test]
62    fn test_write_mode_default() {
63        assert!(matches!(WriteMode::default(), WriteMode::Insert));
64    }
65
66    #[test]
67    fn test_validate_timeseries_insert_ok() {
68        validate_timeseries_write_mode(&WriteMode::Insert).unwrap();
69    }
70
71    #[test]
72    fn test_validate_timeseries_upsert_fails() {
73        let mode = WriteMode::Upsert {
74            key_fields: vec!["id".to_string()],
75        };
76        let err = validate_timeseries_write_mode(&mode).unwrap_err();
77        assert!(err.to_string().contains("time series"));
78    }
79
80    #[test]
81    fn test_validate_timeseries_cdc_replay_fails() {
82        let err = validate_timeseries_write_mode(&WriteMode::CdcReplay).unwrap_err();
83        assert!(err.to_string().contains("time series"));
84    }
85}
laminar_connectors/mongodb/write_model.rs

laminar_connectors/mongodb/
write_model.rs
