laminar_connectors/mongodb/source/
history.rs

1//! Versioned, immutable change-history records.
2//!
3//! Operation names are data. No column uses the engine's `_op`, `_ts_ms`, or `__weight`
4//! changelog names, so append and mutable sinks cannot interpret a history record as a delete.
5
6use std::sync::Arc;
7
8use arrow_array::builder::{
9    Int32Builder, Int64Builder, StringBuilder, TimestampMicrosecondBuilder,
10};
11use arrow_array::{ArrayRef, RecordBatch};
12use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
13use mongodb::bson::{RawBsonRef, RawDocumentBuf, Timestamp};
14use sha2::{Digest, Sha256};
15use uuid::Uuid;
16
17use super::super::change_event::canonical_document_extjson;
18use super::buffering::{BufferedMongoPayload, ChangeRecord, SnapshotRecord};
19use super::checkpoint::encode_timestamp;
20use super::decoding::{decode_change, optional_extjson};
21use super::ConnectorError;
22
23/// Version of the history record layout; incremented on any incompatible change.
24pub const MONGODB_HISTORY_VERSION: i32 = 1;
25/// `operation` value of a document copied by the initial snapshot rather than observed.
26pub const SNAPSHOT_OPERATION: &str = "snapshot";
27
28/// Returns the Arrow schema of `MongoDB` change-history records.
29///
30/// | Column | Type | Nullable | Content |
31/// |---|---|---|---|
32/// | `event_id` | Utf8 | no | SHA-256 of deployment, collection UUID and resume token (or snapshot id and key) |
33/// | `event_version` | Int32 | no | [`MONGODB_HISTORY_VERSION`] |
34/// | `operation` | Utf8 | no | Server `operationType`, or `snapshot` for copied documents |
35/// | `database` | Utf8 | no | Namespace database |
36/// | `collection` | Utf8 | yes | Namespace collection (absent for `dropDatabase`/`invalidate`) |
37/// | `collection_uuid` | Utf8 | no | Bound collection UUID |
38/// | `document_key` | Utf8 | yes | Canonical Extended JSON document key |
39/// | `full_document` | Utf8 | yes | Canonical Extended JSON document image |
40/// | `update_description` | Utf8 | yes | Canonical Extended JSON update delta |
41/// | `event_details` | Utf8 | yes | Canonical Extended JSON of every other event field |
42/// | `resume_token` | Utf8 | yes | Opaque resume token JSON (absent for snapshot rows) |
43/// | `cluster_time_seconds` | Int64 | no | Event cluster time, or the snapshot read time |
44/// | `cluster_time_increment` | Int64 | no | Cluster time increment |
45/// | `wall_time` | Timestamp(us) | yes | Server wall time (millisecond precision) |
46/// | `txn_number` | Int64 | yes | Transaction number when the change belongs to a transaction |
47/// | `lsid` | Utf8 | yes | Canonical Extended JSON transaction session id |
48/// | `snapshot_id` | Utf8 | yes | Snapshot read time `<seconds>.<increment>` for snapshot rows |
49#[must_use]
50pub fn mongodb_history_schema() -> SchemaRef {
51    let utf8 = |name: &str, nullable| Field::new(name, DataType::Utf8, nullable);
52    Arc::new(Schema::new(vec![
53        utf8("event_id", false),
54        Field::new("event_version", DataType::Int32, false),
55        utf8("operation", false),
56        utf8("database", false),
57        utf8("collection", true),
58        utf8("collection_uuid", false),
59        utf8("document_key", true),
60        utf8("full_document", true),
61        utf8("update_description", true),
62        utf8("event_details", true),
63        utf8("resume_token", true),
64        Field::new("cluster_time_seconds", DataType::Int64, false),
65        Field::new("cluster_time_increment", DataType::Int64, false),
66        Field::new(
67            "wall_time",
68            // Microseconds: lakehouse formats such as Iceberg have no millisecond timestamp.
69            DataType::Timestamp(TimeUnit::Microsecond, None),
70            true,
71        ),
72        Field::new("txn_number", DataType::Int64, true),
73        utf8("lsid", true),
74        utf8("snapshot_id", true),
75    ]))
76}
77
78/// Physical identity bound before the first record is emitted.
79pub(super) struct HistoryIdentity<'a> {
80    pub(super) deployment: &'a str,
81    pub(super) collection_uuid: Uuid,
82    pub(super) database: &'a str,
83    pub(super) collection: &'a str,
84}
85
86impl HistoryIdentity<'_> {
87    fn digest(&self, kind: &[u8], parts: &[&[u8]]) -> String {
88        let mut digest = Sha256::new();
89        digest.update(b"laminardb-mongodb-event-v1\0");
90        digest.update(kind);
91        digest.update(self.deployment.as_bytes());
92        digest.update(self.collection_uuid.as_bytes());
93        for part in parts {
94            digest.update(u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
95            digest.update(part);
96        }
97        format!("{:x}", digest.finalize())
98    }
99}
100
101struct HistoryBuilders {
102    event_id: StringBuilder,
103    version: Int32Builder,
104    operation: StringBuilder,
105    database: StringBuilder,
106    collection: StringBuilder,
107    collection_uuid: StringBuilder,
108    document_key: StringBuilder,
109    full_document: StringBuilder,
110    update_description: StringBuilder,
111    details: StringBuilder,
112    resume_token: StringBuilder,
113    seconds: Int64Builder,
114    increment: Int64Builder,
115    wall_time: TimestampMicrosecondBuilder,
116    txn_number: Int64Builder,
117    lsid: StringBuilder,
118    snapshot_id: StringBuilder,
119}
120
121impl HistoryBuilders {
122    fn with_capacity(rows: usize) -> Self {
123        let text = || StringBuilder::with_capacity(rows, rows * 64);
124        Self {
125            event_id: StringBuilder::with_capacity(rows, rows * 64),
126            version: Int32Builder::with_capacity(rows),
127            operation: text(),
128            database: text(),
129            collection: text(),
130            collection_uuid: StringBuilder::with_capacity(rows, rows * 36),
131            document_key: text(),
132            full_document: StringBuilder::with_capacity(rows, rows * 256),
133            update_description: text(),
134            details: text(),
135            resume_token: text(),
136            seconds: Int64Builder::with_capacity(rows),
137            increment: Int64Builder::with_capacity(rows),
138            wall_time: TimestampMicrosecondBuilder::with_capacity(rows),
139            txn_number: Int64Builder::with_capacity(rows),
140            lsid: text(),
141            snapshot_id: text(),
142        }
143    }
144
145    fn append_time(&mut self, at: Timestamp) {
146        self.seconds.append_value(i64::from(at.time));
147        self.increment.append_value(i64::from(at.increment));
148    }
149
150    fn append_change(
151        &mut self,
152        record: &ChangeRecord,
153        identity: &HistoryIdentity<'_>,
154    ) -> Result<(), ConnectorError> {
155        let change = decode_change(&record.raw)?;
156        if change
157            .collection_uuid
158            .is_some_and(|uuid| uuid != identity.collection_uuid)
159        {
160            return Err(ConnectorError::ReadError(format!(
161                "MongoDB change event for {}.{} carries a different collection UUID than the \
162                 bound collection {}",
163                identity.database, identity.collection, identity.collection_uuid
164            )));
165        }
166        let cluster_time = change.cluster_time.ok_or_else(|| {
167            ConnectorError::ReadError("MongoDB change event omitted clusterTime".into())
168        })?;
169        self.event_id
170            .append_value(identity.digest(b"change\0", &[record.token.as_bytes()]));
171        self.version.append_value(MONGODB_HISTORY_VERSION);
172        self.operation.append_value(change.operation);
173        self.database
174            .append_value(change.database.unwrap_or(identity.database));
175        self.collection.append_option(change.collection);
176        self.collection_uuid
177            .append_value(identity.collection_uuid.hyphenated().to_string());
178        self.document_key
179            .append_option(optional_extjson(change.document_key)?);
180        self.full_document
181            .append_option(optional_extjson(change.full_document)?);
182        self.update_description
183            .append_option(optional_extjson(change.update_description)?);
184        self.details
185            .append_option(optional_extjson(change.details.as_deref())?);
186        self.resume_token.append_value(&record.token);
187        self.append_time(cluster_time);
188        self.wall_time.append_option(change.wall_time_us);
189        self.txn_number.append_option(change.txn_number);
190        self.lsid.append_option(optional_extjson(change.lsid)?);
191        self.snapshot_id.append_null();
192        Ok(())
193    }
194
195    fn append_snapshot(
196        &mut self,
197        record: &SnapshotRecord,
198        snapshot_at: Timestamp,
199        identity: &HistoryIdentity<'_>,
200    ) -> Result<(), ConnectorError> {
201        let snapshot_id = encode_timestamp(snapshot_at);
202        let id = record
203            .raw
204            .get("_id")
205            .map_err(|error| ConnectorError::ReadError(format!("snapshot document _id: {error}")))?
206            .ok_or_else(|| ConnectorError::ReadError("snapshot document has no _id".into()))?;
207        let key = id_key(id);
208        self.event_id.append_value(identity.digest(
209            b"snapshot\0",
210            &[snapshot_id.as_bytes(), record.key.as_bytes()],
211        ));
212        self.version.append_value(MONGODB_HISTORY_VERSION);
213        self.operation.append_value(SNAPSHOT_OPERATION);
214        self.database.append_value(identity.database);
215        self.collection.append_value(identity.collection);
216        self.collection_uuid
217            .append_value(identity.collection_uuid.hyphenated().to_string());
218        self.document_key
219            .append_value(canonical_document_extjson(&key)?);
220        self.full_document
221            .append_value(canonical_document_extjson(&record.raw)?);
222        self.update_description.append_null();
223        self.details.append_null();
224        self.resume_token.append_null();
225        self.append_time(snapshot_at);
226        self.wall_time.append_null();
227        self.txn_number.append_null();
228        self.lsid.append_null();
229        self.snapshot_id.append_value(snapshot_id);
230        Ok(())
231    }
232
233    fn finish(mut self, schema: &SchemaRef) -> Result<RecordBatch, ConnectorError> {
234        let columns: Vec<ArrayRef> = vec![
235            Arc::new(self.event_id.finish()),
236            Arc::new(self.version.finish()),
237            Arc::new(self.operation.finish()),
238            Arc::new(self.database.finish()),
239            Arc::new(self.collection.finish()),
240            Arc::new(self.collection_uuid.finish()),
241            Arc::new(self.document_key.finish()),
242            Arc::new(self.full_document.finish()),
243            Arc::new(self.update_description.finish()),
244            Arc::new(self.details.finish()),
245            Arc::new(self.resume_token.finish()),
246            Arc::new(self.seconds.finish()),
247            Arc::new(self.increment.finish()),
248            Arc::new(self.wall_time.finish()),
249            Arc::new(self.txn_number.finish()),
250            Arc::new(self.lsid.finish()),
251            Arc::new(self.snapshot_id.finish()),
252        ];
253        RecordBatch::try_new(Arc::clone(schema), columns)
254            .map_err(|error| ConnectorError::Internal(format!("MongoDB history batch: {error}")))
255    }
256}
257
258/// Build one history batch from record payloads in their emitted order.
259pub(super) fn history_batch<'a>(
260    records: impl ExactSizeIterator<Item = &'a BufferedMongoPayload>,
261    schema: &SchemaRef,
262    identity: &HistoryIdentity<'_>,
263    snapshot_at: Option<Timestamp>,
264) -> Result<RecordBatch, ConnectorError> {
265    let mut builders = HistoryBuilders::with_capacity(records.len());
266    for payload in records {
267        match payload {
268            BufferedMongoPayload::Change(record) => builders.append_change(record, identity)?,
269            BufferedMongoPayload::Snapshot(record) => {
270                let at = snapshot_at.ok_or_else(|| {
271                    ConnectorError::Internal("snapshot row without a snapshot time".into())
272                })?;
273                builders.append_snapshot(record, at, identity)?;
274            }
275            BufferedMongoPayload::HighWatermark { .. } | BufferedMongoPayload::SnapshotComplete => {
276                return Err(ConnectorError::Internal(
277                    "MongoDB history batch received a progress marker as a record".into(),
278                ));
279            }
280        }
281    }
282    builders.finish(schema)
283}
284
285/// `{ "_id": <value> }`, the document key of a replica-set collection.
286pub(super) fn id_key(id: RawBsonRef<'_>) -> RawDocumentBuf {
287    let mut key = RawDocumentBuf::new();
288    key.append("_id", id.to_raw_bson());
289    key
290}
laminar_connectors/mongodb/source/history.rs

laminar_connectors/mongodb/source/
history.rs
