MongoDB CDC

Ingest changes from MongoDB into Firebolt through change streams, with every document queryable as JSON.

Nightly Feature

This feature is currently in private preview. Contact [e-mail] to request early access.

Firebolt connects to your MongoDB deployment and keeps a CDC table continuously in sync with a collection: every insert, update, replace, and delete arrives within seconds, with nothing to deploy in between. Under the hood, the connection is a native MongoDB change stream.

Documents do not need a fixed schema to become queryable: the table holds each document whole, as JSON, keyed by its _id.

Prepare the source deployment

On the MongoDB side you need:

1. A replica set or sharded cluster. Change streams read the oplog; a standalone mongod has none. A sharded collection must be sharded on _id, the only key MongoDB keeps unique across shards.

2. MongoDB 6.0 or later, with document pre- and post-images enabled on the collection. Firebolt reads exact point-in-time images rather than re-fetching documents, so updates are never observed out of order or half-applied:

   db.runCommand({
     collMod: "orders",
     changeStreamPreAndPostImages: { enabled: true }
   })
   

3. A role with find and changeStream privileges on the collection, authenticated with SCRAM.

4. Oplog and pre-image retention sized so the feed’s history outlives any planned ingestion pause. If the stored position ages out of the oplog, ingestion cannot resume from it and the table must be re-bootstrapped.

5. Network reachability from the Firebolt engine to the cluster (VPC peering, PrivateLink, or an Atlas access list).

Create the location

CREATE LOCATION mongo_prod WITH (
    SOURCE   = 'MONGODB',
    URL      = 'mongodb+srv://cluster0.example.mongodb.net/',
    USER     = 'firebolt_cdc',
    PASSWORD = '********'
);

Credentials go in USER and PASSWORD, never in the URL, which is shown as written in information_schema.locations. TLS is on for every host other than localhost unless the URL sets tls=false. See CREATE LOCATION (MongoDB).

Create the stream

A stream binds to one collection, named as <database>.<collection>:

CREATE STREAM orders_changes
LOCATION = 'mongo_prod'
COLLECTION = 'app.orders';

CREATE STREAM connects to the deployment, checks the prerequisites above, and records the current end of the change stream as the stream’s starting position. A stream takes no column list: every MongoDB stream has the same columns, described below.

Read raw change events

SELECT "$op", "$wall_time", _id, doc
FROM READ_STREAM(STREAM orders_changes, consume => false);

Every change row has two columns and a set of pseudo-columns (reference those double-quoted):

Column	Type	Description
_id	TEXT	The document’s _id, rendered as Extended JSON. Never NULL.
doc	JSON	The whole document after the change; NULL for deletes.
$op	TEXT	Operation: I insert, U update or replace, D delete. Snapshot rows are inserts. After a READ_SNAPSHOT, an insert the snapshot may already hold reads as U.
$cluster_time	BIGINT	Cluster timestamp, packed into a 64-bit integer; the ordering authority for the feed. NULL for snapshot rows.
$wall_time	TIMESTAMPTZ	Wall-clock time of the operation on the source. NULL for snapshot rows.
$stream_row_number	BIGINT	1-based position of the row within this statement’s read, in delivery order.
$stream_read_seq	BIGINT	Number of the consuming read that returned the row: the same on every row of one read, and increasing with every consuming read. Selecting it with consume => false raises an error.

The first consuming read starts at the position CREATE STREAM recorded; it does not replay older history. A consuming read has to run inside a writing statement, such as INSERT INTO <table> SELECT ... FROM READ_STREAM(STREAM orders_changes, consume => true).

To load the collection as it is now, use READ_SNAPSHOT:

INSERT INTO orders_load SELECT * FROM READ_SNAPSHOT(STREAM orders_changes);

It returns every document once, with the stream’s _id and doc columns, and moves the stream’s position to where the change feed was just before the scan. The scan is not a point-in-time read, but applying the change feed after it converges anyway, because every change carries the whole document.

Consuming reads advance the stream’s stored position in the same Firebolt transaction that writes the data, so ingestion is exactly-once with no deduplication step.

Create the CDC table and start ingestion

A CDC table over a MongoDB stream has the stream’s two columns, keyed by _id. Attach the stream and resume ingestion:

CREATE CDC TABLE orders (
    _id TEXT,
    doc JSON,
    PRIMARY KEY (_id) NOT ENFORCED
);

ALTER CDC TABLE orders ATTACH STREAM orders_changes;
ALTER CDC TABLE orders RESUME;

The worker loads the collection with READ_SNAPSHOT, then applies the change feed continuously in the order the source committed it. A multi-document transaction becomes visible all at once. The load is the exception: it is not a point-in-time copy, because the collection scan sees each document at some moment while it runs. Until the feed has caught up past the end of the scan, a document changed during the load can show an earlier version, and a transaction that committed during the scan can show some of its documents before it and some after it. Once the feed has passed that point, the table matches the source again. Semantics and lifecycle commands are covered in CDC tables; progress is visible in information_schema.cdc_ingests.

Query document fields

Read any field, including nested ones, through doc. A field is JSON; JSON_VALUE turns a scalar into text, which you can cast:

SELECT _id, doc.status, doc.customer.city
FROM orders
WHERE JSON_VALUE(doc.amount)::DOUBLE PRECISION > 100;

Documents hold relaxed Extended JSON: an int or double is a JSON number, while a type JSON has no number for, such as decimal128 or date, is an object like {"$numberDecimal": "1.5"}.

Fields that appear in documents after the stream was created need no DDL: the table always holds the whole document.

Operations

• Delivery and recovery. The resume token commits atomically with the data it produced. After a crash or Engine restart, run RESUME again; the new worker continues from the last committed token.
• Feed expiry. If ingestion is paused long enough for the stored token to age out of the oplog (or the needed post-images to expire), the worker’s reads fail with a history-lost error, shown in last_error of information_schema.cdc_ingests, until the retry budget runs out and the attachment stops in state = 'error'. Nothing on the source is harmed; recreate the stream and the CDC table, which loads the collection again. A stream no CDC table is attached to can instead be reloaded with READ_SNAPSHOT, which moves its position to just before the scan.
• Collection lifecycle events. A collection drop, rename, or stream invalidation ends the feed. Like any persistent failure, the attachment stops in state = 'error' once the retry budget runs out. Recreate the stream and table against the new collection.
• DROP STREAM removes only Firebolt metadata. MongoDB change streams hold no server-side resource comparable to a replication slot, so there is nothing to clean up on the source.

Limitations

• One collection per stream.
• A stream’s columns are fixed: _id and doc.
• A sharded collection is supported only when it is sharded on _id. CREATE STREAM checks this; resharding the collection later is not detected.
• A document the engine cannot store stops the table’s ingestion at that change: information_schema.cdc_ingests reports it in last_error; once the retry budget runs out, the attachment stops in state = 'error'. To recover, detach the stream and attach a new one; the next RESUME reloads the collection.
• ALTER STREAM is not supported; schema changes are drop-and-recreate.
• The table’s columns are the stream’s: _id and doc. Fields are read through doc.