# Debezium (change data capture)

Source: https://docs.rushdb.com/connect/debezium/

[`debezium-server-rushdb`](https://github.com/rush-db/debezium-server-rushdb) is a [Debezium Server](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) sink: it keeps RushDB in sync with an existing database by streaming every insert, update and delete that Debezium captures. No Kafka is needed. Any Debezium source connector works, including PostgreSQL, MySQL, MariaDB, MongoDB, SQL Server, Oracle and Db2.

It writes to **RushDB Cloud** or to a **self-hosted RushDB engine** with the same results.

## What lands in RushDB

| Source | RushDB |
|---|---|
| Table or collection `orders` | Label `ORDER` (upper case, singular; overridable) |
| Row | One record, with column names preserved |
| Nested object `address: {city}` | Property `address.city` |
| Primary key | `syncKey` property: `<topic>:<key values>`, the record's identity |
| Insert or update | Upsert by `syncKey` (`append` merges, `rewrite` mirrors the row) |
| Delete | The record with that `syncKey` is deleted |

Each record also gets `syncedAt`, the time of the change. Fields listed in `ignore-fields` (for example PII) are dropped before anything is sent.

Offsets are committed only after RushDB accepts a batch, so a failure replays the batch. Replays are idempotent because writes are keyed by `syncKey`. The sink never stores event data.

## Run it

Debezium Server registers sinks when the server is built, so the sink ships as a Debezium Server build with it included. Use the Docker image, which is the official Debezium Server image with the RushDB sink built in:

```bash
docker run --rm \
  -v $PWD/config:/debezium/config \
  -v $PWD/data:/debezium/data \
  -e RUSHDB_API_KEY=... \
  ghcr.io/rush-db/debezium-server-rushdb:latest
```

`config/application.properties` configures the source and the sink:

```properties
debezium.sink.type=rushdb

debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.topic.prefix=app
debezium.source.database.hostname=postgres
debezium.source.database.port=5432
debezium.source.database.user=postgres
debezium.source.database.password=${PG_PASSWORD}
debezium.source.database.dbname=app
debezium.source.table.include.list=public.customers,public.orders
debezium.source.plugin.name=pgoutput
debezium.source.decimal.handling.mode=double
debezium.source.offset.storage.file.filename=data/offsets.dat

debezium.format.key=json
debezium.format.value=json
debezium.format.key.schemas.enable=false
debezium.format.value.schemas.enable=false
```

The source settings are Debezium's; see the [Debezium connector documentation](https://debezium.io/documentation/reference/stable/connectors/index.html) for your database. A full example is in the repository's [`example/application.properties`](https://github.com/rush-db/debezium-server-rushdb/blob/main/example/application.properties).

## Connect

| Target | Environment variables |
|---|---|
| RushDB Cloud | `RUSHDB_API_KEY` |
| Self-hosted engine | `RUSHDB_ENGINE_URL`, `RUSHDB_ENGINE_TOKEN`, and `RUSHDB_SCOPE` on a multitenant engine |

Setting both is an error, so a deployment never picks a target by accident. Pass the key or token as an environment variable rather than in `application.properties`, because Debezium Server prints its configuration when it starts.

On a self-hosted engine, the sink declares an [ordered index](../engine/indexes-and-search/indexes.md) on `syncKey` for each label it writes to. Upserts and deletes on large tables then look records up through the index instead of reading every record of the label.

## Sink settings

Settings go in `application.properties` as `debezium.sink.rushdb.<name>`:

| Setting | Default | Description |
|---|---|---|
| `merge-strategy` | `append` | `append` keeps properties the row no longer has; `rewrite` mirrors the row exactly |
| `labels` | — | Label overrides: `orders=PURCHASE;users=APP_USER` |
| `ignore-fields` | — | Comma-separated fields never sent, matched at any depth and ignoring case and `_`/`-` |
| `sync-key-field` | `syncKey` | Name of the identity property |
| `synced-at-field` | `syncedAt` | Name of the change-time property |
| `import-chunk` | `500` | Records per write request |
| `index-sync-key` | `true` | Self-hosted engine: declare the `syncKey` index per label |
| `singularize` | `true` | `orders` becomes `ORDER` |
| `flatten-separator` | `.` | Joins nested field names |
| `reserved-prefix` | `src` | Renames source fields that start with `$` or `__`, such as `$type` to `src_type` |

## Data types

- **Dates and times** arrive as ISO-8601 strings and are stored as [datetimes](../learn/records-and-queries/labels-and-properties.mdx).
- **DECIMAL / NUMERIC** columns arrive as base64 bytes when schemas are off, which is Debezium's default `decimal.handling.mode=precise`. Set `debezium.source.decimal.handling.mode=double` to store them as numbers, or `string` to keep the exact digits as strings.
- **JSON columns** (PostgreSQL `json`/`jsonb`, MySQL `JSON`) arrive as strings and are stored as strings.
- **MongoDB** extended JSON (`$oid`, `$date`, `$numberLong`) is stored as plain values.

## Next steps

- [Find and query records](../learn/records-and-queries/find-and-query.mdx) synced from your database
- [Connect records](../learn/relationships/connect-records.mdx) across tables, for example orders to customers by `customer_id`
- [Bring your own vectors](../learn/semantic-search/bring-your-own-vectors.mdx) to search the synced data by meaning