Debezium (change data capture)

debezium-server-rushdb is a Debezium Server 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:

Shell

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 for your database. A full example is in the repository's 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 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.
• 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 synced from your database
• Connect records across tables, for example orders to customers by customer_id
• Bring your own vectors to search the synced data by meaning