Transform Functions

Transform functions are optional scripts that allow users to embed custom processing logic into an ingest pipeline.

Conceptual Model

A transform function operates on each incoming record produced by the source of an ingest pipeline. Today, supported sources include Kafka, Iceberg, and object stores such as S3.

Each record is processed independently. The transform function does not maintain shared state across records.

Language and Execution Model

Transform functions currently support JavaScript only.

A transform function is executed as a JavaScript script and is invoked with the following global variables:

• arg: The incoming data record.
• meta: Metadata associated with the record.
• Built-in functions: A limited set of built-in helpers is available. See Built-in JavaScript Helpers.

There is no module system or external dependency support.

Built-in JavaScript Helpers

Transform functions may use the following built-in helper functions.

decodeUtf8(value)

Decodes a Uint8Array as UTF-8 and returns a JavaScript string.

This is useful when a source value or metadata field is represented as bytes and needs to be converted before indexing.

arg["decoded_value"] = decodeUtf8(arg["bytes_value"]);

arg

decodeUtf8 expects exactly one argument, and that argument must be a Uint8Array.

log(value)

Writes a value to the transform execution logs and returns undefined.

Use log for temporary debugging while developing a transform. The value is converted to a string before being logged.

log(`processing record ${arg["id"]}`);

arg

log expects exactly one argument.

flattenDocument(arg, options)

Flattens a nested JSON document field into top-level string fields and builds a search-value array.

This is useful for raw log or semi-structured JSON ingestion where nested fields should be indexed under a stable prefix.

flattenDocument(arg, {
  source: "document",
  outputPrefix: "doc@",
  nestedSeparator: "@",
  searchField: "search_field_internal"
});

arg

flattenDocument options

Flattened field output

Given this input record:

{
  "document": {
    "customer": {
      "name": "Alice",
      "email": "alice@example.com"
    },
    "tags": ["VIP", "beta"]
  }
}

The default options produce fields like:

{
  "doc@customer@name": "Alice",
  "doc@customer@email": "alice@example.com",
  "doc@tags[*]": ["VIP", "beta"],
  "search_field_internal": ["alice", "alice@example.com", "vip", "beta"]
}

Primitive values are converted to strings. Empty strings, null, and undefined values are ignored. Repeated flattened values for the same field are deduplicated.

Limit field creation independently from search depth

Use maxFieldDepth to avoid creating flattened fields for deeply nested paths while still collecting their primitive values in the search field:

flattenDocument(arg, {
  source: "document",
  outputPrefix: "doc@",
  nestedSeparator: "@",
  searchField: "search_field_internal",
  maxDepth: 32,
  maxFieldDepth: 1
});

arg

Given this source document:

{
  "eventType": "system.operation.rate_limit.violation",
  "securityContext": {
    "asNumber": "qwe",
    "asOrg": 123
  },
  "severity": "WARN"
}

the transform creates fields only for top-level primitive values, but includes the nested values in the search field:

{
  "doc@eventType": "system.operation.rate_limit.violation",
  "doc@severity": "WARN",
  "search_field_internal": [
    "system.operation.rate_limit.violation",
    "qwe",
    "123",
    "warn"
  ]
}

No doc@securityContext@asNumber or doc@securityContext@asOrg fields are created. If maxFieldDepth is omitted, it defaults to maxDepth, preserving the existing behavior.

maxFieldDepth counts the key segments in the flattened field name, not the levels of the source document. An array contributes a [*] marker rather than a key segment, so it does not advance the depth:

Note that maxDepth does count arrays, because it bounds traversal work rather than the field namespace. A top-level array therefore needs maxDepth of at least 2 for its values to be reached at all.

Preserve dropped subtrees as JSON

maxFieldDepth alone discards a subtree that is too deep to become fields, leaving nothing to filter on or extract from at query time. Set containerJsonSuffix to store such a subtree as a single JSON string instead:

flattenDocument(arg, {
  source: "document",
  maxDepth: 32,
  maxFieldDepth: 1,
  containerJsonSuffix: "@json"
});

arg

Given this source document:

{
  "id": "qwe",
  "role_type": ["a", "b"],
  "_links": {
    "self": { "href": "https://example.test/roles/qwe" }
  }
}

the transform produces:

{
  "doc@id": "qwe",
  "doc@role_type[*]": ["a", "b"],
  "doc@_links@json": "{\"self\":{\"href\":\"https://example.test/roles/qwe\"}}"
}

At maxFieldDepth: 1 every top-level key yields exactly one field: scalars and arrays of scalars directly, anything else as a JSON string. Which subtrees qualify:

• Objects at the limit always qualify, since every key beneath them lands one segment deeper.
• Arrays qualify only when they hold objects or arrays. An array of scalars already becomes a real multi-valued field, which supports exact matching and aggregation; a JSON string would not.
• Only the outermost dropped subtree is preserved, so the same data is not emitted at several field names.

Because the rule is tied to maxFieldDepth, it adjusts automatically: raising the limit to 2 turns _links.self into real fields and moves the JSON string one level deeper.

Preserved values also remain in the search field, exactly as they would without this option.

Subtrees whose JSON exceeds maxContainerJsonChars are dropped rather than truncated. maxStringChars does not apply. A JSON string cut mid-token still contains the tokens a search prefilter matches, so the record would pass a cheap filter and then fail to parse; an absent field fails the filter honestly.

Field names carry the suffix, so a dynamic template can type them differently from ordinary flattened fields. Order the more specific template first:

{
  "dynamic_templates": [
    {
      "container_json": {
        "path_match": "doc@*@json",
        "mapping": { "type": "keyword", "index": false, "doc_values": true }
      }
    },
    {
      "flattened": {
        "path_match": "doc@*",
        "mapping": { "type": "keyword" }
      }
    }
  ]
}

doc_values is required to extract from the field with JSON path functions. Indexing is not: nobody matches a whole JSON blob as a single term, and field-existence checks are served by _field_names rather than the field’s own index.

Filter source keys

Documents whose keys are not stable field names — request parameters, generated identifiers, hashes — mint a new column for every distinct key they carry. maxFields does not bound this, because it applies per document rather than per index.

keyPattern, keyIgnorePattern and maxKeyChars restrict which source keys may produce fields:

flattenDocument(arg, {
  source: "document",
  maxFieldDepth: 1,
  keyPattern: "[A-Za-z0-9_.]{1,40}",
  keyIgnorePattern: "[0-9a-f]{16,}"
});

arg

Patterns are matched against the raw source key, before keyReplacements is applied, so they describe the document rather than the rewritten field name. Both must match the whole key: an unanchored pattern such as [a-z]+ is treated as ^(?:[a-z]+)$ rather than accepting any key that merely contains a letter. An invalid pattern raises an error rather than silently disabling the filter.

Two patterns are provided because a character allowlist cannot exclude values that are already within its character set — a 48-character hexadecimal key satisfies [A-Za-z0-9_.] and needs either keyIgnorePattern or maxKeyChars to reject it.

A filtered key suppresses the field only. Its values are still collected into the search field, exactly as values below maxFieldDepth are, so the content remains findable without acquiring a column named after it.

Key replacement behavior

By default, flattenDocument replaces dots and the configured nested separator with underscores:

keyReplacements: {
  ".": "_",
  "@": "_"
}

This preserves legacy behavior for the default separator.

Dots are not preserved by default because dots are interpreted as object-path syntax by the indexing and query layers. For example, a field named foo.bar may be treated as nested path foo -> bar, not as one literal field name.

To preserve key identity more clearly, provide custom replacements:

flattenDocument(arg, {
  source: "document",
  outputPrefix: "doc@",
  nestedSeparator: "@",
  keyReplacements: {
    ".": "__dot__",
    "@": "__at__"
  }
});

arg

Input:

{
  "document": {
    "recon.request_length": 123,
    "actor@id": "u1"
  }
}

Output fields:

{
  "doc@recon__dot__request_length": "123",
  "doc@actor__at__id": "u1"
}

Percent-style encoding

Replacements are applied in a single pass over the original key. Replacement output is not processed again.

flattenDocument(arg, {
  keyReplacements: {
    "%": "%25",
    ".": "%2E",
    "@": "%40"
  }
});

arg

A key such as:

raw%.@key

becomes:

raw%25%2E%40key

not:

raw%2525%252E%2540key

Compatibility note

Default behavior remains compatible with existing flattenDocument usage that uses the default nestedSeparator: "@".

If you use a custom nestedSeparator, default sanitization also replaces that configured separator in source key segments. This avoids ambiguous flattened field names, but may change field names for existing custom-separator configurations.

If you opt into custom keyReplacements, flattened field names will change. Existing indexed data will not automatically be query-compatible with the new names. During migration, either reingest data or query both old and new field names.

Input Variables

arg

Represents the incoming data payload. Its structure depends on the source:

• Kafka: The message payload (value)
• Iceberg: A single table row
• Object stores (e.g. S3): The parsed output produced by the configured parser (for example, one JSON value per line for NDJSON)

meta

Carries source-specific metadata. Examples include:

• Kafka: key, offset, partition, and operation type
• Object stores: file name, line number, and parser-related metadata

The contents of meta vary by source and should be treated as source-dependent.

Return Values and Pipeline Actions

The value returned by a transform function determines how the ingest pipeline processes the record:

• Single object: Indexes a single document.
• Array of objects: Indexes multiple documents. Internally, the array is unnested and applies the same indexing semantics to each object as if they were returned individually.
• Empty array ([]): Indicates that the record should be ignored and no indexing action should be performed.
• Kafka tombstones: Special delete semantics apply when ingesting Kafka tombstone records. These semantics are described in the Kafka Tombstone Support in Mach5 section.

Key _id Semantics

Ingest pipelines apply special handling to the _id field:

• If an object includes an _id field, the pipeline performs an upsert using that _id (or a delete, if the Kafka operation type is delete).
• If an object does not include an _id field, the record is appended and the pipeline automatically generates an _id.

Examples

Update Field Value

This transformation converts database-style timestamp objects into ISO-8601 strings so they can be efficiently indexed and queried.

Input record example:

{
  id: "event-123",
  extra: {
    inserted_date: {
      year: 2025,
      month: 3,
      day: 4,
      hour: 9,
      minute: 7,
      second: 5,
      microsecond: 123
    },
    detected_date: {
      year: 2024,
      month: 12,
      day: 31,
      hour: 23,
      minute: 59,
      second: 59,
      microsecond: 999999
    },
    last_update: {
      year: 2025,
      month: 1,
      day: 1,
      hour: 0,
      minute: 0,
      second: 0,
      microsecond: 1
    }
  }
}

Transformed output record:

{
  id: "event-123",
  extra: {
    inserted_date: "2025-03-04T09:07:05.000123",
    detected_date: "2024-12-31T23:59:59.999999",
    last_update: "2025-01-01T00:00:00.000001"
  }
}

JS transform - for fields inserted_date, detected_date and last_update, update the value to ISO date format:

function toIsoString(obj) {
  // safety check
  if (!obj || typeof obj !== "object") return undefined;

  let year = obj.year.toString().padStart(4, "0");
  let month = obj.month.toString().padStart(2, "0");
  let day = obj.day.toString().padStart(2, "0");
  let hour = obj.hour.toString().padStart(2, "0");
  let minute = obj.minute.toString().padStart(2, "0");
  let second = obj.second.toString().padStart(2, "0");

  // ensure 6 digits
  let microsecond = obj.microsecond.toString().padStart(6, "0");

  return `${year}-${month}-${day}T${hour}:${minute}:${second}.${microsecond}`;
}

["inserted_date", "detected_date", "last_update"].forEach(key => {
  if (arg["extra"][key]) {
    arg["extra"][key] = toIsoString(arg["extra"][key]);
  }
});

arg

Drop/Ignore Record

This transform drops records that are to be skipped.

Input record example (csv input records):

version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status
2,44010291029,0a4fa7sda8d7ad,10.0.0.1,10.0.0.3,40658,10009,17,1,303,1755475131000,1755475203000,ACCEPT,OK
2,44010291030,-,-,-,-,-,-,-,-,1755475350,1755475381,-,NODATA
2,44010291031,-,-,-,-,-,-,-,-,1755475350,1755475381,-,SKIPDATA

JS transform to ignore NODATA and SKIPDATA records.

if (arg["log-status"] === "SKIPDATA" || arg["log-status"] === "NODATA") {
  //empty array to ignore the record
  arg = [];
}

arg

Insert _id Field for Upsert Operation

Upsert operation requires the ingest record to contain the _id field. If the record does not have the _id field then one can be inserted using the JS transform.

Input record example:

{
  "source_key_hash": "8f3c1a2b9d4e7f6a0c5b2e1d9a8f7c6b",
  "ingest_date": "2025-03-04T09:07:05.000123",
  "indextype": "emailhash"
}

Transformed output record:

{
  "source_key_hash": "8f3c1a2b9d4e7f6a0c5b2e1d9a8f7c6b",
  "ingest_date": "2025-03-04T09:07:05.000123",
  "indextype": "emailhash",
  "_id": "emailhash_8f3c1a2b9d4e7f6a0c5b2e1d9a8f7c6b"
}

JS transform to add _id field

if (arg["indextype"] && arg["source_key_hash"]) {
    // Generate deterministic _id
    arg["_id"] = arg["indextype"] + "_" + arg["source_key_hash"];
}

arg