Transforms

Transforms mutate, filter, or convert messages as they flow through the connector runtime. They run inside the runtime process, not inside the plugins. For source connectors they're applied after the plugin's messages are decoded and before producing to Iggy. For sink connectors, after consuming from Iggy and before handing the messages to the plugin.

A transform receives each decoded message and returns either the (possibly modified) message, or nothing - in which case the message is dropped from the batch.

Configuration

Transforms are configured per connector in the transforms section of the connector configuration file. Each transform type is a table keyed by its name, so a connector can have at most one instance of each type:

[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "message"
value.static = "hello"

[transforms.delete_fields]
enabled = true
fields = ["internal_id"]

Two rules apply to every transform:

• The enabled key is required. A transform section without it fails to load, and a transform with enabled = false is skipped.
• The order of the sections in the file doesn't determine the execution order, so don't rely on one transform seeing the output of another.

The field-level transforms (add_fields, delete_fields, filter_fields, update_fields, unwrap_envelope) operate on JSON payloads only. Messages with any other payload format pass through them unchanged.

add_fields

Adds new fields to the message payload. Each field has a key and a value, which is either static or computed at runtime:

[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "source"
value.static = "orders-service"

[[transforms.add_fields.fields]]
key = "ingested_at"
value.computed = "timestamp_millis"

The available computed values are:

delete_fields

Removes the listed fields from the message payload:

[transforms.delete_fields]
enabled = true
fields = ["debug_info", "internal_id"]

update_fields

Sets fields to static or computed values, optionally guarded by a condition:

[transforms.update_fields]
enabled = true

[[transforms.update_fields.fields]]
key = "updated_at"
value.computed = "date_time"

[[transforms.update_fields.fields]]
key = "id"
value.computed = "uuid_v7"
condition = "key_not_exists"

The condition values are always (the default when omitted - sets the field whether or not it exists), key_exists (update only when the field is already present), and key_not_exists (add only when the field is absent).

filter_fields

Keeps or drops fields of the payload based on key and value patterns:

[transforms.filter_fields]
enabled = true
keep_fields = ["id"]
include_matching = true

[[transforms.filter_fields.patterns]]
key_pattern = { starts_with = "user_" }

[[transforms.filter_fields.patterns]]
key_pattern = { regex = "^order_[0-9]+$" }
value_pattern = "is_not_null"

• keep_fields: fields that are always kept, regardless of the patterns.
• patterns: a list of patterns. A field matches when any pattern matches. Within one pattern, key_pattern and value_pattern must both match when both are set (an omitted one matches everything).
• include_matching: true (default) keeps the matching fields and drops the rest. false drops the matching fields and keeps the rest.

key_pattern variants: exact, starts_with, ends_with, contains, regex - each takes a string, e.g. key_pattern = { contains = "name" }.

value_pattern variants: equals (any JSON value), contains (substring), regex, greater_than, less_than, between (e.g. { between = [1.0, 10.0] }), and the parameterless checks is_null, is_not_null, is_string, is_number, is_boolean, is_object, is_array (written as plain strings, e.g. value_pattern = "is_number").

unwrap_envelope

Promotes a nested field of a JSON envelope to the top-level payload. Given { "table": "...", "operation": "...", "data": { ... } }, setting field = "data" replaces the whole payload with the contents of data:

[transforms.unwrap_envelope]
enabled = true
field = "data"

field must not be empty. Messages that aren't JSON objects, or that don't contain the field, pass through unchanged.

proto_convert

Converts messages between Protocol Buffers and other formats. See the SDK documentation for the schema loading details:

[transforms.proto_convert]
enabled = true
source_format = "proto"
target_format = "json"
schema_path = "schemas/message.proto"
message_type = "com.example.Message"
include_paths = ["."]
preserve_unknown_fields = false

[transforms.proto_convert.conversion_options]
validate_messages = true
pretty_json = false
include_metadata = false
type_url_prefix = "type.googleapis.com"
strict_mode = false

Optional keys: schema_path, message_type, field_mappings (rename fields during conversion, e.g. field_mappings = { "old" = "new" }), and descriptor_set (pre-compiled descriptor bytes instead of a .proto file).

flat_buffer_convert

Converts messages between FlatBuffers and other formats:

[transforms.flat_buffer_convert]
enabled = true
source_format = "flat_buffer"
target_format = "json"
schema_path = "schemas/message.fbs"
root_table_name = "Message"
include_paths = ["."]
preserve_unknown_fields = false

[transforms.flat_buffer_convert.conversion_options]
verify_buffers = true
pretty_json = false
include_metadata = false
buffer_size_hint = 1024
strict_mode = false

Optional keys: schema_path, root_table_name, field_mappings.

avro_convert

Converts messages between Avro and other formats. The schema can be given inline (schema_json) or as a file (schema_path):

[transforms.avro_convert]
enabled = true
source_format = "avro"
target_format = "json"
schema_path = "schemas/message.avsc"

[transforms.avro_convert.conversion_options]
pretty_json = false
include_metadata = false
strict_mode = false

Optional keys: schema_path, schema_json, field_mappings.

Unlike the field-level transforms, the three format-conversion transforms define no per-key defaults: every key shown in their examples (except the ones listed as optional) must be present, or the connector configuration fails to load. The source_format and target_format keys accept any schema value: json, raw, text, proto, flat_buffer, avro. The FlatBuffers transform additionally restricts the pair: only JSON to/from FlatBuffers, FlatBuffers to Text or Raw, and identity conversions are accepted. Any other combination fails the configuration load.

Adding a new transform

The extension point is the SDK (core/connectors/sdk/src/transforms/):

1. Add a new module implementing the Transform trait.
2. Add a variant to the TransformType enum.
3. Add a match arm to the from_config function deserializing your config.

The runtime loads transforms generically by TransformType, so no runtime changes are needed.