# Transforms

> Mutate, filter and convert messages as they pass through the connector runtime, and how each transform is configured.

Rendered page: https://iggy.apache.org/docs/connectors/transforms/

Source: https://github.com/apache/iggy-website/blob/main/content/docs/connectors/transforms.mdx

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:

```toml
[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:

```toml
[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:

| Value                | Result                                    |
|----------------------|-------------------------------------------|
| `date_time`          | Current date and time as an RFC 3339 string |
| `timestamp_nanos`    | Current Unix timestamp in nanoseconds     |
| `timestamp_micros`   | Current Unix timestamp in microseconds    |
| `timestamp_millis`   | Current Unix timestamp in milliseconds    |
| `timestamp_seconds`  | Current Unix timestamp in seconds         |
| `uuid_v4`            | Random UUID (version 4)                   |
| `uuid_v7`            | Time-ordered UUID (version 7)             |

## delete_fields

Removes the listed fields from the message payload:

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

## update_fields

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

```toml
[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:

```toml
[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`:

```toml
[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](/docs/connectors/sdk)** for the schema loading details:

```toml
[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:

```toml
[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`):

```toml
[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.