Currently, I am using the MongoDB Kafka Source Connector to stream the event in my local database to another MongoDB instance in the cloud.  We want to replace the remote DB by a Postgresql instance and to do it, we want to use the Debezium Mongodb source connector with the Debezium sink connector for Postgresql.

My source connector has an aggregation pipeline to modify the payload to another schema before putting it into Kafka. The example below shows how to achieve such behavior:

```
{
  "name": "metrics",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "connection.uri": "mongodb://172.0.0.2:27017",
    "database": "metrics",
    "collection": "metrics",
    "copy.existing": "true",
    "pipeline": "[{"$match": { "$and": [

, {"fullDocument.eventId": 1404 }] } }]"
  }
}
```

It would be great to Debezium source connector for MongoDB support this kind of aggregation pipeline to transform each streamed document. In another words, The feature consists in a SMT using mongoDB aggregation stage.

MongodDB Documentation Example

Thanks