Connect MongoDB

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	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	You already publish MongoDB CDC to Kafka.	Seconds	Medium — reuse your existing pipeline, add ingestion transforms
`mongodump` BSON files	You need a one-time load, a periodic refresh, or a backfill.	Batch	Low

A frequent wrong turn: the Debezium message decoder 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.

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:

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:

_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

{
  "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.

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.

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:
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.

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 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:

{
  "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"}
    ]
  }
}

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.

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:

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:

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:

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
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.

Get Started

Ingestion

External Tables

Query Data

Manage Data

Observability

Visualize Data

Cluster Operations

Manage Security

Release Notes

Reference

Pick a path

Direct Change Streams (CDC)

What you need first

The partition plan is the real design work

Minimum stream config

Debezium into Kafka

`mongodump` BSON files

How BSON types land in Pinot

Troubleshooting

​Pick a path

​Direct Change Streams (CDC)

​What you need first

​The partition plan is the real design work

​Minimum stream config

​Debezium into Kafka

​mongodump BSON files

​How BSON types land in Pinot

​Troubleshooting

Pick a path

Direct Change Streams (CDC)

What you need first

The partition plan is the real design work

Minimum stream config

Debezium into Kafka

`mongodump` BSON files

How BSON types land in Pinot

Troubleshooting