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.

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