# Read change messages as inserts, updates and deletes

Acquire / file

A variant of [Read a JSON file](https://meridian.online/operations/read-a-json-file).

Read JSON change-data-capture messages so each one becomes an insert, update or delete row.

Page: <https://meridian.online/operations/read-a-json-file/read-change-messages-as-inserts-updates-and-deletes>

## Status

Runs as a SQL step.

arcform carries it as a `sql:` step today.

## The DuckDB step

read change messages as insert, update-before, update-after and delete rows

```sql
-- change-data-capture messages: op c creates, u updates (before and after images), d deletes (before image only)
COPY (SELECT * FROM (VALUES
  ('{"op":"c","ts":1,"before":null,"after":{"id":1,"customer":"Alice Ng","amount":120}}'),
  ('{"op":"u","ts":2,"before":{"id":1,"customer":"Alice Ng","amount":120},"after":{"id":1,"customer":"Alice Ng","amount":130}}'),
  ('{"op":"c","ts":3,"before":null,"after":{"id":2,"customer":"Bob Okafor","amount":50}}'),
  ('{"op":"d","ts":4,"before":{"id":2,"customer":"Bob Okafor","amount":50},"after":null}')) t(line))
  TO '<your_scratch_dir>/changes.json' (FORMAT csv, HEADER false, QUOTE '', ESCAPE '', DELIMITER '~');
CREATE TEMP TABLE result AS
WITH messages AS (SELECT json AS m FROM read_json_objects('<your_scratch_dir>/changes.json'))
SELECT (m->>'$.ts')::INTEGER AS ts, kind, rec->>'$.id' AS id, rec->>'$.customer' AS customer, (rec->>'$.amount')::INTEGER AS amount
FROM (
  SELECT m, 'insert' AS kind, m->'$.after' AS rec FROM messages WHERE m->>'$.op' = 'c'
  UNION ALL SELECT m, 'update_before' AS kind, m->'$.before' AS rec FROM messages WHERE m->>'$.op' = 'u'
  UNION ALL SELECT m, 'update_after' AS kind, m->'$.after' AS rec FROM messages WHERE m->>'$.op' = 'u'
  UNION ALL SELECT m, 'delete' AS kind, m->'$.before' AS rec FROM messages WHERE m->>'$.op' = 'd');
SELECT * FROM result ORDER BY ts, kind DESC;
SELECT CASE WHEN ((SELECT count(*) FROM result) = 5 AND (SELECT list(kind ORDER BY ts, kind DESC) FROM result) = ['insert', 'update_before', 'update_after', 'insert', 'delete'] AND (SELECT list(amount ORDER BY ts, kind DESC) FROM result) = [120, 120, 130, 50, 50]) THEN 'as expected' ELSE error('the four messages did not become an insert, the before and after rows of an update, an insert and a delete') END AS verdict;
```

Replace `<your_scratch_dir>` with a directory of your own that you can write to before running the statement.

The statement writes `changes.json` itself, in `<your_scratch_dir>`.

It needs nothing installed beyond DuckDB itself.

## What other tools call it

Meridian's classification; an independent blind rating agreed on 95.7% of 324 rows (variant, version 3).

- Apache SeaTunnel: [`Canal Format`](https://seatunnel.apache.org/docs/2.3.13/connectors/formats/canal-json) (read 2026-09-27)
- Apache SeaTunnel: [`Debezium Format`](https://seatunnel.apache.org/docs/2.3.13/connectors/formats/debezium-json) (read 2026-09-27)
- Apache SeaTunnel: [`Kafka source compatible kafka-connect-json`](https://seatunnel.apache.org/docs/2.3.13/connectors/formats/kafka-compatible-kafkaconnect-json) (read 2026-09-27)
- Apache SeaTunnel: [`MaxWell Format`](https://seatunnel.apache.org/docs/2.3.13/connectors/formats/maxwell-json) (read 2026-09-27)
- Apache SeaTunnel: [`Ogg Format`](https://seatunnel.apache.org/docs/2.3.13/connectors/formats/ogg-json) (read 2026-09-27)