Transform at ingestion — mapping hooks¶

The defrost, concretely: legacy exports arrive with mojibake (BÃ©nÃ©dict), DD/MM/YYYY dates, numbers in parentheses for negatives, HTML entities, zero-width characters. The mapping hooks repair all of that while the data flows in — declared in the connector manifest's spec.mapping block, no code.

spec:
  mapping:
    dataset:                    # tier 1 — the whole chunk, once
      - skip-empty-rows: {}
    row:                        # tier 2 — one record; may emit 0..N rows
      - map-columns: { rename: { CUST_NM: name } }
    cell:                       # tier 3 — one value, mutated in place
      - fix-mojibake: { columns: [name] }

Execution order is fixed: dataset → row → cell. Each entry is a single-key object (- hook-name: { params }); an unknown hook name fails the run (unknown dataset hook: <name>).

Where spec.mapping runs — and where it doesn't

csv-file connector runs only. The csv-file connector type parses spec.mapping and applies its hooks before writing. The jdbc and kafka connectors write a 1:1 dump — their specs carry no hooks. The REST/GraphQL/SOAP engine has its own, different mechanism (see REST: the transform: enum).
Connector runs only — not one-shot lm csv ingest. The one-shot path (POST /api/platform/sources/csv/ingest) writes directly without the mapping engine; its spec.mapping (if any) is ignored. To transform, apply a csv-file connector manifest and run it.

Cell hooks¶

Operate on one value at a time. Two parameter conventions:

Multi-column hooks take columns: [a, b]; omitting columns (or columns: ["*"]) applies the hook to every column of the row.
Single-column hooks take column: <name> and refuse to guess.

Parse hooks fail soft: a value that cannot be parsed becomes null and increments the run's cell-error counter — the row is kept, the run continues.

Hook	Params (default)	Behavior
`clean`	`columns`; `strip` (false); `remove-control-chars` (false); `normalize-unicode` (off; `NFC`/`NFD`/`NFKC`/`NFKD`)	always strips a leading BOM; then optional trim, control-char removal, Unicode normalization
`fix-mojibake`	`columns`	recovers UTF-8 read as Latin-1 (`BÃ©nÃ©dict` → `Bénédict`); applies only when the mojibake pattern is detected, keeps the original if recovery would produce `�`
`decode-html-entities`	`columns`	`Coffee & cream` → `Coffee & cream` (HTML4 entity set)
`replace-smart-quotes`	`columns`	curly quotes/guillemets → ASCII quotes, en/em dash → `-`, ellipsis → `...`, NBSP → space
`strip-invisible`	`columns`	removes zero-width and format characters (ZWSP, ZWJ, BOM, soft hyphen — all Unicode `Cf`)
`remove-diacritics`	`columns`	`Café` → `Cafe` (NFD + combining-mark strip)
`upper` / `lower`	`columns`	case conversion
`titlecase`	`columns`	first letter of each whitespace-separated word up, rest down
`parse-date`	`column`; `formats` (list of Java `DateTimeFormatter` patterns)	tries each format in order; a pattern containing `H/h/m/s` parses to a timestamp, otherwise to a date; no match → `null` + cell error; already-temporal values pass through
`parse-number`	`column`; `decimal-separator` (`auto`; or `","`/`"."`); `negatives-in-parens` (false)	strips currency symbols (`$€£¥`) and spaces, resolves thousand vs decimal separators, `(123)` → `-123` when enabled; result is a decimal; failure → `null` + cell error
`parse-email`	`column`; `lower` (false)	trims, optionally lowercases, validates; invalid → `null` + cell error
`parse-phone`	`column`; `default-region` (`US`)	libphonenumber parse → E.164 (`+33612345678`); failure → `null` + cell error
`default`	`column`; `value`	sets `value` when the cell is null or blank; never overwrites
`lookup`	`column`; `table` (inline map); `on-missing` (`keep`; or `null` / `error` / any literal)	replaces the value via the map; on a miss: keep the original, null it (+ cell error), fail the run, or substitute the literal
`mask`	`column`; `keep-first` (0); `keep-last` (0); `mask-char` (`*`)	masks the middle: `4556123498761234` with `keep-last: 4` → `************1234`; values not longer than `keep-first + keep-last` are left untouched
`truncate`	`column`; `length` (unlimited)	hard cut at `length` characters
`split`	`column`; `by` (`,`); `into` (list of new column names) or `into-array` (false)	splits within the row: parts (trimmed) land in named columns, or replace the cell as an array — distinct from `split-row`, which duplicates rows
`regex-extract`	`column`; `pattern`; `groups` (map group-index → target column)	tries a full match, then a find; writes captured groups into the named columns
`regex-replace`	`column`; `pattern`; `replacement` (`""`)	`replaceAll` over the cell
`to-uuid`	`column`; `generate` (false)	valid UUID passes through; invalid → freshly generated UUID or `null`; a null cell is filled only when `generate: true`
`kotlin`	`script` (required, `classpath:` only); `context` (map)	the escape hatch — see below

mapping:
  cell:
    - fix-mojibake: { columns: [name, city] }
    - parse-date:
        column: signup_date
        formats: ["dd/MM/yyyy", "yyyy-MM-dd"]
    - parse-number:
        column: balance
        decimal-separator: ","
        negatives-in-parens: true
    - mask: { column: card_number, keep-last: 4 }

Row hooks¶

One record in, 0..N records out — a row hook can drop, duplicate, or reshape rows.

Hook	Params (default)	Behavior
`map-columns`	`rename` (map old → new); `drop` (list)	renames then drops columns
`coalesce`	`column`; `from` (list of source columns)	when the target is null/blank, takes the first non-blank source value
`defaults`	a map `column: value`	fills every listed column that is null or blank
`validate-row`	`required` (list of columns)	a row missing (or blank in) any required column is dropped and counted in the run's DLQ counter
`split-row`	`by` (column) + `sep` (`,`) — or `explode` (column holding a list)	one row per part/list element (parts trimmed), the other columns copied — the inverse of an aggregation
`aggregator`	`emit-on-change` (key column); `keep-first` (columns copied from the group's first row); `merge` (map out-column → `{collect, into: array\\|set\\|csv, dedup: false, sep: ","}`)	groups consecutive rows by key — refused on csv-file connector runs, see below

mapping:
  row:
    - map-columns:
        rename: { CUST_NM: name, CUST_CITY: city }
        drop: [internal_flag]
    - validate-row: { required: [code, name] }
    - split-row: { by: tags, sep: ";" }

aggregator exists in the engine, but csv-file runs refuse it

Connector runs are chunked transactions with a fresh mapping state per chunk — the aggregator's cross-chunk buffer would be silently dropped at every chunk boundary. lm validate / lm apply refuse the manifest instead:

aggregator is not supported on csv-file connector runs — each chunk commits separately, buffered rows would be silently lost at chunk boundaries

with the hint: Pre-aggregate in the source (e.g. a grouped export), or drop the aggregator hook.

Dataset hooks¶

Run once per chunk, before the row and cell tiers. The contract: a dataset hook may rewrite the whole chunk, never more — it never sees the full file. Local csv-file runs commit in fixed 500-row chunks (remote sftp/s3 transports read one whole file per transaction, so there the chunk is the file). Consequence: sort-by orders within a chunk, not globally.

Hook	Params (default)	Behavior
`require-headers`	`columns` (all must exist); `any-of` (at least one must exist)	fails the run honestly when the file's shape is wrong: `required header missing: <col>` / `none of required headers found: [...]`
`skip-empty-rows`	—	drops rows whose every cell is null or blank
`filter-where`	`column`; `equals`; `in` (list); `regex` (full match)	keeps rows satisfying every configured criterion (AND); a null cell compares as `""`
`sort-by`	`columns` (list); `max-rows` (100000)	lexicographic in-memory sort within the chunk (nulls sort first, as empty strings); over the cap it refuses: `sort-by max-rows exceeded: <n> > <max> (input must be pre-sorted or increase max-rows)`
`enrich-context`	a map `column: value`; the value `${now}` resolves to the ingestion instant	adds constant columns to every row (existing values are not overwritten)

mapping:
  dataset:
    - require-headers: { columns: [code, name] }
    - skip-empty-rows: {}
    - filter-where: { column: status, in: [ACTIVE, PENDING] }
    - enrich-context: { source_system: siebel, ingested_at: "${now}" }

Kotlin escape hatch¶

When no built-in fits, a kotlin cell hook runs a .kts script over the whole row:

mapping:
  cell:
    - kotlin:
        script: classpath:hooks/samples/normalize-date.kts
        context: { defaultCountry: FR }    # optional, read-only in the script

The script gets bindings["cells"] (the mutable row map) and bindings["context"]. Same trust boundary as every Kotlin hook (see the hooks guide): scripts run with full JVM trust and must live on the platform classpath — classpath: refs only, a file: path is rejected (only classpath: scripts supported in v1). A script failure fails the run.

Runnable example: docs/connectors/csv/50-kotlin-row-hook.yaml (its script source is mirrored at docs/connectors/csv/hooks/normalize-date.kts).

REST: the `transform:` enum¶

REST/GraphQL/SOAP endpoints do not use spec.mapping. Their transformation point is the promote: block of a hybrid target — each promoted column takes an optional transform:

Value	Effect
`from_epoch`	Unix epoch seconds → timestamp (UTC); for milliseconds, divide upstream
`parse_iso8601`	ISO-8601 string → timestamp
`to_lower` / `to_upper`	case conversion
`trim`	whitespace trim
`to_json_string`	serializes the extracted value to a JSON string

See the REST connector for the promote: block itself.