> ## Documentation Index
> Fetch the complete documentation index at: https://docs.startree.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Connect MongoDB

> Choose and configure a path for getting MongoDB data into StarTree Cloud.

Getting MongoDB data into StarTree means picking one of three ingestion paths and writing the table or job configuration yourself.

This page helps you pick, then walks through each one.

## Pick a path

| Path                                                | Use when                                                                             | Freshness | Effort                                                          |
| --------------------------------------------------- | ------------------------------------------------------------------------------------ | --------- | --------------------------------------------------------------- |
| [Direct Change Streams](#direct-change-streams-cdc) | You want a live, current-state table and you do not already run Kafka for this data. | Seconds   | High — you own a manifest of physical MongoDB collections       |
| [Debezium into Kafka](#debezium-into-kafka)         | You already publish MongoDB CDC to Kafka.                                            | Seconds   | Medium — reuse your existing pipeline, add ingestion transforms |
| [`mongodump` BSON files](#mongodump-bson-files)     | You need a one-time load, a periodic refresh, or a backfill.                         | Batch     | Low                                                             |

<Info>
  A frequent wrong turn: the [Debezium message decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/debezium) does **not** parse MongoDB envelopes. Its `dbz.source` accepts only `mysql` and `postgres`. Ingest Debezium's MongoDB output with the plain JSON decoder instead — see [Debezium into Kafka](#debezium-into-kafka).
</Info>

## Direct Change Streams (CDC)

The StarTree MongoDB CDC connector opens MongoDB Change Streams straight from Pinot servers. No Kafka, Kinesis, or Pulsar sits in between.

### What you need first

* **MongoDB Atlas, a replica set, or a sharded cluster.** Standalone `mongod` processes do not serve Change Streams.
* **MongoDB Server 6.0 or later** if you want a current-state upsert table, because that mode requires Change Stream post-images.
* **Network reachability and credentials on both Pinot controllers and servers.** The identity needs `find` and `changeStream` on every source namespace.
* **An oplog retention window longer than your worst-case Pinot outage.** The oplog is the only replay buffer. If it ages out, recovery means rebuilding the table from a snapshot, not resuming.
* **A partition plan.** This is the part teams underestimate — read the next section before anything else.

### The partition plan is the real design work

MongoDB Change Streams have no Kafka-style durable partitions, so the connector cannot discover them. You supply a manifest of physical, disjoint collections, and each one becomes one Pinot partition group:

```text theme={null}
app.orders_00  ->  Pinot partition group 0
app.orders_01  ->  Pinot partition group 1
app.orders_02  ->  Pinot partition group 2
```

Two consequences drive most of the design:

**Every key must stay in its partition forever.** Every insert, update, replace, and delete for one `_id` has to land in the same collection for that key's lifetime. A common way to guarantee that is to encode the partition into the immutable `_id` at creation time:

```text theme={null}
_id = "02:ord_123"  ->  app.orders_02
_id = "07:ord_456"  ->  app.orders_07
```

Do not route with `hash(_id) % currentPartitionCount`. Growing the count moves existing keys and breaks upsert partition affinity.

**Cursor load multiplies.** Each Pinot replica opens its own cursor per partition, so plan for `source partitions x Pinot replicas`. On a sharded deployment, a cursor opened through `mongos` fans out to every shard regardless of shard key, so plan closer to `source partitions x Pinot replicas x MongoDB shards`.

### Minimum stream config

```json theme={null}
{
  "streamType": "mongodb",
  "stream.mongodb.topic.name": "mongo-orders",
  "stream.mongodb.consumer.factory.class.name": "ai.startree.pinot.plugin.stream.mongodb.MongoCdcConsumerFactory",
  "stream.mongodb.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
  "stream.mongodb.consumer.prop.auto.offset.reset": "largest",
  "sourcePartitions": "[{\"partitionId\":0,\"epoch\":1,\"sourceId\":\"atlas-prod-a/app/orders-00\",\"connectionStringEnvironmentVariable\":\"MONGODB_ATLAS_URI\",\"database\":\"<DATABASE>\",\"collection\":\"<COLLECTION>\"}]",
  "outputMode": "full_document",
  "fullDocument": "required",
  "maxAwaitTimeMs": "500"
}
```

`stream.mongodb.topic.name` is only a logical stream name — it does not identify a MongoDB namespace. Prefer `connectionStringEnvironmentVariable` or a `connectionRef` over an inline `connectionString`, because table configs can be logged or exposed.

<Warning>
  This config is for an **empty or brand-new** source collection. With `auto.offset.reset=largest` the controller captures a Change Stream boundary at table creation and consumes only changes after it — documents already sitting in the collection are never emitted, and any that are never updated again stay permanently absent from the table.

  For a collection that already holds data, do not use this config as-is. Capture an exact boundary per namespace first, snapshot/backfill up to it, then start CDC from that same boundary supplied as each descriptor's `initialOffset`. See [Capturing Initial Offsets](/corecapabilities/ingestdata/adv-concepts/realtime/mongodb-cdc#capturing-initial-offsets).
</Warning>

Pick an output mode:

* `change_stream` (default) keeps the lossless MongoDB event envelope with `fullDocument` nested inside it.
* `full_document` lifts document fields to the top level and adds reserved `__mongodb_cdc_*` metadata columns. Use this for a current-state upsert table. It needs `fullDocument=required` and post-images enabled on every source collection:

  ```javascript theme={null}
  db.runCommand({
    collMod: "orders_00",
    changeStreamPreAndPostImages: {enabled: true}
  })
  ```

For the full option set — registry-based partition discovery, capturing exact initial offsets, the upsert schema, and adding capacity to a live table — see [MongoDB CDC Ingestion](/corecapabilities/ingestdata/adv-concepts/realtime/mongodb-cdc).

## Debezium into Kafka

If Debezium already streams your MongoDB collections into Kafka, ingest that topic as a normal Kafka stream. The one thing to get right is the decoder.

The StarTree [Debezium message decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/debezium) handles MySQL and Postgres only. For MongoDB, use `JSONMessageDecoder` and unpack the envelope with ingestion transforms.

Debezium's MongoDB connector puts the changed document in `after` as a **JSON string**, not a nested object, so read fields out of it with `jsonPathString`:

```json theme={null}
{
  "streamConfigs": {
    "streamType": "kafka",
    "stream.kafka.topic.name": "<TOPIC>",
    "stream.kafka.broker.list": "<BROKER_LIST>",
    "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
    "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
    "stream.kafka.consumer.prop.auto.offset.reset": "smallest"
  },
  "ingestionConfig": {
    "transformConfigs": [
      {"columnName": "orderId", "transformFunction": "jsonPathString(after, '$._id')"},
      {"columnName": "status", "transformFunction": "jsonPathString(after, '$.status')"},
      {"columnName": "amount", "transformFunction": "jsonPathDouble(after, '$.amount')"},
      {"columnName": "opType", "transformFunction": "op"},
      {"columnName": "eventTimeMs", "transformFunction": "ts_ms"}
    ]
  }
}
```

<Warning>
  The transforms above alone cannot back an upsert table. On a delete, Debezium sets `after` to null and carries the deleted document's `_id` only in the **Kafka record key**. `JSONMessageDecoder` decodes the record value, so every `jsonPathString(after, ...)` transform yields null and the row has no primary key — Pinot cannot match the existing row to remove it.
</Warning>

For an upsert table, unwrap the envelope in Debezium instead of in Pinot. The MongoDB `ExtractNewDocumentState` SMT lifts the document fields to the top level of the value, but it only preserves the deleted document's `_id` if you ask it to — both of the `delete.tombstone.handling.mode` settings below are required:

```properties theme={null}
transforms=unwrap
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.delete.tombstone.handling.mode=rewrite
transforms.unwrap.delete.tombstone.handling.mode.rewrite-with-id=true
transforms.unwrap.add.fields=op,ts_ms
```

`rewrite` keeps deletes as real records carrying a `__deleted` boolean instead of emitting null-valued tombstones, and `rewrite-with-id` copies `id` from the record key into the payload as `_id`. Without both, a delete still reaches Pinot with no primary key and cannot remove the existing row.

Records then arrive as flat documents, so `JSONMessageDecoder` needs no `jsonPath` transforms at all: `_id` becomes a normal column you can name in `primaryKeyColumns`, `__deleted` is the `deleteRecordColumn`, and `__ts_ms` serves as the upsert comparison column. `add.fields` prefixes what it adds with `__` by default, configurable via `add.fields.prefix`.

Also set the connector's `capture.mode` to a full-document mode such as `change_streams_update_full`, so updates carry the whole document rather than a partial update description. Without it, updates arrive as patches that Pinot cannot merge.

## `mongodump` BSON files

Apache Pinot reads BSON natively, so `mongodump` output loads without conversion. Use this for backfills, periodic snapshots, or any table that does not need to be live.

Dump a collection and stage the files where your ingestion job can read them:

```bash theme={null}
mongodump --uri "$MONGODB_URI" --db app --collection orders --out ./dump
aws s3 cp ./dump/app/orders.bson s3://<BUCKET_NAME>/mongo/orders/orders.bson
```

Then set the input format in the ingestion job spec:

```yaml theme={null}
executionFrameworkSpec:
  name: 'standalone'
  segmentGenerationJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner'
  segmentTarPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner'

jobType: SegmentCreationAndTarPush
inputDirURI: 's3://<BUCKET_NAME>/mongo/orders/'
includeFileNamePattern: 'glob:**/*.bson'
outputDirURI: 's3://<BUCKET_NAME>/segments/orders/'
overwriteOutput: true

pinotFSSpecs:
  - scheme: s3
    className: org.apache.pinot.plugin.filesystem.S3PinotFS
    configs:
      region: '<AWS_REGION>'

recordReaderSpec:
  dataFormat: 'bson'
  className: 'org.apache.pinot.plugin.inputformat.bson.BSONRecordReader'

tableSpec:
  tableName: 'orders'
  schemaURI: '<CONTROLLER_URI>/tables/orders/schema'
  tableConfigURI: '<CONTROLLER_URI>/tables/orders'

pinotClusterSpecs:
  - controllerURI: '<CONTROLLER_URI>'
```

`className` is optional — Pinot resolves `dataFormat: 'bson'` to `BSONRecordReader` on its own — but naming it explicitly makes the spec self-documenting.

### How BSON types land in Pinot

| BSON type                                       | Pinot-side value                                                                                                    |
| ----------------------------------------------- | ------------------------------------------------------------------------------------------------------------------- |
| `Double`, `Int32`, `Int64`, `Boolean`, `String` | Same type, passed through                                                                                           |
| `ObjectId`                                      | `STRING` — the 24-character hex form                                                                                |
| `DateTime`                                      | Timestamp, millisecond precision                                                                                    |
| `Timestamp` (internal replication type)         | Timestamp at **second** granularity; the within-second ordinal is dropped                                           |
| `Decimal128`                                    | `BigDecimal`. `NaN` and `Infinity` become null                                                                      |
| `Binary` (including UUID subtypes)              | Raw `byte[]`, not a `java.util.UUID`                                                                                |
| Embedded document                               | Map — flatten it with [complex type handling](/corecapabilities/ingestdata/dataportal/data-modeling/unnesting-json) |
| Array                                           | Multi-value column                                                                                                  |
| Everything else                                 | The driver's `toString()` rendering                                                                                 |

Because `Timestamp` loses sub-second ordering, do not use an oplog `ts` field as an upsert comparison column.

## Troubleshooting

| Symptom                           | Cause                                                  | Fix                                                                                |
| --------------------------------- | ------------------------------------------------------ | ---------------------------------------------------------------------------------- |
| `dbz.source` rejects `mongodb`    | The Debezium decoder supports MySQL and Postgres only. | Use `JSONMessageDecoder` with ingestion transforms.                                |
| Change Streams fail to open       | Source is a standalone `mongod`.                       | Change Streams need a replica set, sharded cluster, or Atlas.                      |
| CDC resume fails after an outage  | Oplog or post-image history expired.                   | Rebuild the table from a snapshot plus matching Change Stream boundaries.          |
| Deletes never reach the table     | No delete column is wired up.                          | Use `__mongodb_cdc_is_deleted` (CDC connector) or derive one from `op` (Debezium). |
| MongoDB load higher than expected | Cursors scale with partitions x replicas x shards.     | Reduce partitions or replicas, or isolate sources on separate deployments.         |
| Updates arrive as partial patches | Debezium is not capturing post-images.                 | Set the connector's capture mode to a full-document mode.                          |