MongoDB New Document State Extraction

Configuration

The configuration is a part of sink task connector and is expressed in a set of properties:

  1. transforms=unwrap,...
  2. transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
  3. transforms.unwrap.drop.tombstones=false
  4. transforms.unwrap.delete.handling.mode=drop
  5. transforms.unwrap.operation.header=true

Array encoding

The SMT converts MongoDB arrays into arrays as defined by Apache Connect (or Apache Avro) schema. The problem is that such arrays must contains elements of the same time. MongoDB allows the user to store elements of heterogeneous types into the same array. To bypass this impedance mismatch it is possible to encode the array in two different ways using array.encoding configuration option.

  1. transforms=unwrap,...
  2. transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
  3. transforms.unwrap.array.encoding=<array|document>

Value array (the default) will encode arrays as the array datatype. It is user’s responsibility to ensure that all elements for a given array instance are of the same time. This option is a restricting one but offers easy processing of arrays by downstream clients.

Value document will convert the array into a struct of structs in the similar way as done by BSON serialization. The main struct contains fields named _0, _1, _2 etc. where the name represents the index of the element in the array. Every element is then passed as the value for the give field.

Let’s suppose an example source MongoDB document with array with heterogeneous types

  1. {
  2. "_id": 1,
  3. "a1": [
  4. {
  5. "a": 1,
  6. "b": "none"
  7. },
  8. {
  9. "a": "c",
  10. "d": "something"
  11. }
  12. ]
  13. }

This document will be encoded as

  1. {
  2. "_id": 1,
  3. "a1": {
  4. "_0": {
  5. "a": 1,
  6. "b": "none"
  7. },
  8. "_1": {
  9. "a": "c",
  10. "d": "something"
  11. }
  12. }
  13. }

This option allows you to process arbitrary arrays but the consumer need to know how to properly handle them.

Note: The underscore in index names is present because Avro encoding requires field names not to start with digit.

Nested structure flattening

When a MongoDB document contains a nested document (structure) it is faithfully encoded as a nested structure field. If the sink connector does support only flat structure it is possible to flatten the internal structure into a flat one with a consistent field naming. To enable this feature the option flatten.struct must be set to true.

  1. transforms=unwrap,...
  2. transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
  3. transforms.unwrap.flatten.struct=<true|false>
  4. transforms.unwrap.flatten.struct.delimiter=<string>

The resulting flat document will consist of fields whose names are created by joining the name of the parent field and the name of the fields in the nested document. Those elements are separated with string defined by an option struct.delimiter by default set to the underscore.

Let’s suppose an example source MongoDB document with a field with a nested document

  1. {
  2. "_id": 1,
  3. "a": {
  4. "b": 1,
  5. "c": "none"
  6. },
  7. "d": 100
  8. }

Such document will be encoded as

  1. {
  2. "_id": 1,
  3. "a_c": 1,
  4. "a_d": "none",
  5. "d": 100
  6. }

This option allows you to convert a hierarchical document into a flat structure suitable for a table-like storage.

MongoDB $unset handling

MongoDB allows $unset operations that remove a certain field from a document. Because the collections are schemaless, it becomes hard to inform consumers/sinkers about the field that is now missing. The approach that Debezium uses is to set the field being removed to a null value.

Given the operation

  1. {
  2. "after":null,
  3. "patch":"{\"$unset\" : {\"a\" : true}}"
  4. }

The final encoding will look like

  1. {
  2. "id": 1,
  3. "a": null
  4. }

Note that other MongoDB operations might cause an $unset internally, $rename is one example.

Determine original operation

When a message is flattened the final result does not show whether it was an insert, update or first read. (Deletions can be detected via tombstones or rewrites, see Configuration options.)

To solve this problem Debezium offers an option to propagate the original operation via a header added to the message. To enable this feature the option operation.header must be set to true.

  1. transforms=unwrap,...
  2. transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
  3. transforms.unwrap.operation.header=true

The possible values are the ones from the op field of MongoDB connector change events.

Adding source metadata fields

The SMT can optionally add metadata fields from the original change event’s source structure to the final flattened record (prefixed with “__“). This functionality can be used to add things like the collection from the change event, or connector-specific fields like the replica set name. For more information on what’s available in the source structure see the documentation for the MongoDB connector.

For example, the configuration

  1. transforms=unwrap,...
  2. transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
  3. transforms.unwrap.add.source.fields=rs,collection

will add

  1. { "__rs" : "rs0", "__collection" : "my-collection", ... }

to the final flattened record.

For DELETE events, this option is only supported when the delete.handling.mode option is set to “rewrite”.

Configuration options

PropertyDefaultDescription

array

The SMT converts MongoDB arrays into arrays as defined by Apache Connect (or Apache Avro) schema.

false

The SMT flattens structs by concatenating the fields into plain properties, using a configurable delimiter.

Delimiter to concat between field names from the input record when generating field names for the output record. Only applies when flatten.struct is set to true

false

The SMT adds the event operation as a message header.
This is deprecated as of Debezium 1.2, please use add.headers and add.fields instead.

true

The SMT removes the tombstone generated by Debezium from the stream.

drop

The SMT can drop, rewrite or pass delete records (none). The rewrite mode will add a deleted field set to true or false depending on the represented operation.

Fields from the change event’s source structure to add as metadata (prefixed with ““) to the flattened record.
This is deprecated as of Debezium 1.2, please use add.headers and add.fields instead.

Specify a list of metadata fields to add to header of the flattened message. In case of duplicate field names (e.g. “tsms” exists twice), the struct should be specified to get the correct field (e.g. “source.tsms”). The fields will be prefixed with ““ or “<struct>_“, depending on the specification of the struct. Please use a comma separated list without spaces.

Specify a list of metadata fields to add to the flattened message. In case of duplicate field names (e.g. “ts_ms” exists twice), the struct should be specified to get the correct field (e.g. “source.ts_ms”). The fields will be prefixed with ““ or “<struct>“, depending on the specification of the struct. Please use a comma separated list without spaces.

false

Whether field names will be sanitized to adhere to Avro naming requirements. See Avro naming for more details.

Known limitations

  • Feeding data changes from a schemaless store such as MongoDB to strictly schema-based datastores such as a relational database can by definition work within certain limits only. Specifically, all fields of documents within one collection with the same name must be of the same type. Otherwise, no consistent column definition can be derived in the target database.

  • Arrays will be restored in the emitted Kafka Connect record correctly, but they are not supported by sink connector just expecting a “flat” message structure.