laminar_connectors/mongodb/write_model/
mod.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    /// Applies `MongoDB` CDC history records (`output.mode=history`) in order, mirroring the
27    /// source collection with its exact BSON types:
28    ///
29    /// - `insert`, `replace`, `snapshot` → `replaceOne(..., upsert: true)` by document key
30    /// - `update` → `replaceOne` with the post-image when present, otherwise `updateOne`
31    ///   from `updateDescription`
32    /// - `delete` → `deleteOne` by document key
33    /// - collection metadata events → no write
34    /// - `drop`, `rename`, `dropDatabase`, `invalidate`, unknown → rejected; a fixed destination
35    ///   never receives destructive DDL
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;
laminar_connectors/mongodb/write_model/mod.rs

laminar_connectors/mongodb/write_model/
mod.rs
