[Feature] Add MongoDB CDC source - #2
Merged
Merged
Conversation
Snapshot checkpoints were persisted as untagged JSON, and ResumeToken deserializes from any JSON object, so a restart after a completed snapshot fed the server a bogus resume token and never re-snapshotted. Offsets are now an externally tagged MongoOffset enum (snapshot_in_progress / snapshot_complete / token); a completed snapshot records its cluster time so a restart starts the change stream there, an interrupted one re-runs, and unreadable offsets fail loudly. Update and delete events dropped the primary key when a collection had no pre-images enabled (the MongoDB default): the Postgres sink errored with "no old row" and Iceberg equality deletes got a null _id. `old` now falls back to the change event's documentKey. CRUD events whose document could not be looked up are skipped instead of emitting a row-less change. Also: - pass the shutdown token into the snapshot so a long scan is cancellable, and treat a channel closed during shutdown as a clean stop - fail instead of exiting successfully when the server closes the change stream - commit one resume token per drained server batch (or 1000 events) rather than one offset write and sink flush per document - merge conflicting BSON types during schema inference (nulls ignored, numerics widened, otherwise text) in both the source and schema discovery - fix pre-1970 timestamps, prefer wallTime over clusterTime, drop a dead enum variant, and sort the discovery sample for stable schemas Examples and docs: fix a self-referencing `const db` that made the MongoDB init script throw, correct the namespace type in mongo-to-iceberg.json, align mongo-to-pg.json credentials, and document source requirements and pre-images. Tests: unit coverage for change event conversion, offset round-trips and BSON type merging, a guard that every example config deserializes, and integration tests for deletes without pre-images, snapshot-phase offset resume, collection filtering, shutdown mid-snapshot, and Mongo to Postgres replication.
Schema discovery sampled only the oldest 100 documents of a collection, and callers treat that sample as the authoritative source schema. A field added later in the collection's life was therefore invisible: it was typed as text, skipped by Iceberg schema evolution, or — worst — confirmed as "dropped" and removed from the replication target along with its data. Sampling now takes half the budget from each end of the collection, and drop confirmation is disabled entirely for schemaless sources, where a field's absence from a sample proves nothing. Change stream positions could also become unresumable. The resume token of an `invalidate` event was stored even though the server refuses it in `resumeAfter`, and an idle or fully filtered stream never advanced the offset at all, so it eventually aged out of the oplog. Invalidate tokens are no longer stored, a drained batch falls back to the stream's own resume token, and a start position that has fallen out of the oplog now triggers a fresh snapshot instead of failing identically on every restart. Also: - map MongoDB `int` to a 64-bit column: every BSON integer arrives as i64, so a 32-bit target truncated in Iceberg and failed the insert in Postgres - cache mongodb::Client per connection string, since discovery runs per write batch in replication mode and each call re-ran topology discovery - fail fast on a configured collection that does not exist, instead of running indefinitely and replicating nothing Tests: unit coverage for drop-confirmation gating and lost-resume-point detection; integration tests for unknown collections, offset advancement while the watched collection is idle, invalidate tokens staying out of the offset store, and a sparse MongoDB field not dropping its target column.
Postgres sink: - Escape embedded double quotes in identifiers. MongoDB field names may contain `"`, which terminated the quoted identifier and let the rest of the name run as SQL — a crafted field name could execute arbitrary DDL. - Prefix bytes parameters with `\x`. Params are sent as TEXT and cast server-side, so bare hex was read as the escape format and stored literally, double-encoding every bytea value. - Surface the wrapped DbError: tokio-postgres renders server failures as the bare string "db error", hiding the actual cause. Iceberg sink: - Keep only the last event per primary key in a replication batch. Equality deletes never apply to data files added by the same commit, so two versions of a key in one batch both survived as duplicate rows. - Reject values that cannot be converted to the target column type instead of writing NULL. A schemaless source changing a field's type silently dropped the value. Also replaces a truncating `as i32` cast and a boolean parse that treated any unrecognised string as false. MongoDB source: - Explain BSONObjectTooLarge failures. An update to a large document with pre-images enabled produces an event over the 16 MB limit that replays on every restart; the driver's text gives no hint at the way out.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
MongoDB via official
mongodb3.5 driver and change streams (replica set required). Snapshot at recorded cluster time, thendb.watch()withupdateLookup+whenAvailablepre-images. Wired into all four sinks in both CDC and replication mode.Highlights:
MongoOffsetenum (snapshot-in-progress / snapshot-complete / resume token), committed once per drained server batch. Auto re-snapshots when the stored position ages out of the oplog._idas PK, top-level fields flattened (nested docs and arrays become JSON text), types inferred by sampling 100 docs. Column drops are never inferred, since a field missing from a sample proves nothing."could execute arbitrary SQL),byteadouble-encoding, unreadabledb errormessages; Iceberg duplicate rows when one key changed twice in a batch, plus silent NULLing of values that don't fit the column type.Known limits:
_idcollisions: replication keys on_idalone, but MongoDB only guarantees_iduniqueness per shard unless_idis the shard key or its prefix. UsingdocumentKeyas the PK would fix this.dropandrenameare skipped, so the target keeps orphan rows and renamed collections silently stop flowing.