<commit_before>= MongoDB CDC Event Flattening
:awestruct-layout: doc
:linkattrs:
:icons: font
:source-highlighter: highlight.js

[NOTE]
====
This single message transformation (SMT) is available since Debezium version 0.7.2.
It is under active development right now, so the emitted message structure or other details may still change as development progresses.
Please see below for a descriptions of known limitations of this transformation.
====

[NOTE]
====
This SMT is supported only for the MongoDB connector.
====

[CAUTION]
====
There is a https://issues.jboss.org/browse/DBZ-624[known issue] which prevents processing of update messages when MongoDB 3.6 is used.
====

The Debezium MongoDB connector generates the data in a form of a complex message structure.
The message consists of two parts:

* operation and metadata
* for inserts, the row data after the insert has been executed; for updates a patch element describing the altered fields

The `after` and `patch` elements are Strings containing JSON representations of the inserted/altered data.
E.g. the general message structure for a insert event looks like this:

[source,json,indent=0]
----
{
  "op": "r",
  "after": "{\"field1\":newvalue1,\"field2\":\"newvalue1\"}",
  "source": { ... }
}
----

More details about the message structure are provided in link:/docs/connectors/mongodb/[the documentation] of the MongoDB connector.

While this structure is a good fit to represent changes to MongoDB's schemaless collections,
it's not understood by existing sink connectors as for instance the Confluent JDBC sink connector.

Therefore Debezium provides a https://kafka.apache.org/documentation/#connect_transforms[a single message transformation] (SMT)
which converts the `after`/`patch` information from the MongoDB CDC events into a structure suitable for consumption by existing sink connectors.
To do so, the SMT parses the JSON strings and reconstructs properly typed Kafka Connect
(comprising the correct message payload and schema) records from that,
which then can be consumed by connectors such as the JDBC sink connector.

Using JSON as visualization of the emitted record structure, the event from above would like this:

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

The SMT should be applied on a sink connector.

== Configuration

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

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope
----

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

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope
transforms.unwrap.array.encoding=<array|struct>
----

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 `struct` will convert the array into a *struct* of *structs* in the similar way as done by http://bsonspec.org/[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
[source,json,indent=0]
----
{
    "_id": 1,<commit_msg>DBZ-925 Change `array.encoding` to correct options<commit_after>= MongoDB CDC Event Flattening
:awestruct-layout: doc
:linkattrs:
:icons: font
:source-highlighter: highlight.js

[NOTE]
====
This single message transformation (SMT) is available since Debezium version 0.7.2.
It is under active development right now, so the emitted message structure or other details may still change as development progresses.
Please see below for a descriptions of known limitations of this transformation.
====

[NOTE]
====
This SMT is supported only for the MongoDB connector.
====

[CAUTION]
====
There is a https://issues.jboss.org/browse/DBZ-624[known issue] which prevents processing of update messages when MongoDB 3.6 is used.
====

The Debezium MongoDB connector generates the data in a form of a complex message structure.
The message consists of two parts:

* operation and metadata
* for inserts, the row data after the insert has been executed; for updates a patch element describing the altered fields

The `after` and `patch` elements are Strings containing JSON representations of the inserted/altered data.
E.g. the general message structure for a insert event looks like this:

[source,json,indent=0]
----
{
  "op": "r",
  "after": "{\"field1\":newvalue1,\"field2\":\"newvalue1\"}",
  "source": { ... }
}
----

More details about the message structure are provided in link:/docs/connectors/mongodb/[the documentation] of the MongoDB connector.

While this structure is a good fit to represent changes to MongoDB's schemaless collections,
it's not understood by existing sink connectors as for instance the Confluent JDBC sink connector.

Therefore Debezium provides a https://kafka.apache.org/documentation/#connect_transforms[a single message transformation] (SMT)
which converts the `after`/`patch` information from the MongoDB CDC events into a structure suitable for consumption by existing sink connectors.
To do so, the SMT parses the JSON strings and reconstructs properly typed Kafka Connect
(comprising the correct message payload and schema) records from that,
which then can be consumed by connectors such as the JDBC sink connector.

Using JSON as visualization of the emitted record structure, the event from above would like this:

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

The SMT should be applied on a sink connector.

== Configuration

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

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope
----

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

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope
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 http://bsonspec.org/[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
[source,json,indent=0]
----
{
    "_id": 1,
{'fraction_non_alphanumeric': 0.08067184120117063, 'fraction_numerical': 0.006234889935106248, 'mean_word_length': 4.662824207492795, 'pattern_counts': {'":': 12, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1883735', 'n_tokens_mistral': 2161, 'n_tokens_neox': 2054, 'n_words': 1014}