Data integration

Keep a live copy of tables from PostgreSQL or MySQL in a KiviDB database. A pipeline copies each table once, then applies every change made in the source within seconds, so your application can read from KiviDB at memory speed while the source stays the system of record. This is often called change data capture (CDC).

Each row becomes one key, named by a template you choose: with the template customer:{id}, the row whose id is 42 is stored at customer:42, as a hash (one field per column) or a JSON document. An insert or update rewrites the key; a delete removes it; a row whose key columns change moves to its new key.

Sources

Source	How changes are read
PostgreSQL	logical replication: a publication and a pgoutput replication slot the pipeline creates
MySQL and MariaDB	the binary log, read as a replica would

We test against PostgreSQL 16 and MySQL 8.4. Managed services built on them (Amazon RDS and Aurora, Google Cloud SQL, Azure Database) work the same way once logical replication or the row-based binary log is turned on, as below; tell us if one does not.

Preparing PostgreSQL

1. Set wal_level = logical and restart. On Amazon RDS and Aurora, set the parameter rds.logical_replication = 1 in the parameter group and reboot.
2. Create a user with the REPLICATION attribute (on RDS, grant rds_replication) that can read the tables and create a publication for them; the simplest is the tables' owner.
3. Give every table a primary key. The pipeline pages through a table by its primary key, and a delete carries only the key's columns, so use primary key columns in the key template, or set REPLICA IDENTITY FULL on the table.

The pipeline creates a publication and a replication slot named kividb_ plus part of its id. A replication slot makes the source keep its write-ahead log until the pipeline has read it: while a pipeline is stopped, the source's disk use grows with every write. Deleting the pipeline drops the slot and the publication.

Preparing MySQL or MariaDB

1. Turn on the binary log with binlog_format = ROW and binlog_row_image = FULL (the defaults on MySQL 8.0 and later). On RDS, set a binlog retention period long enough to cover an outage, for example CALL mysql.rds_set_configuration('binlog retention hours', 24).
2. Recommended on MySQL 8.0 and later: binlog_row_metadata = FULL.
3. Create a user with REPLICATION SLAVE, REPLICATION CLIENT and SELECT on the tables.
4. Give every table a primary key.

Network

The source must have a public address, and accept connections from KiviDB Cloud. Addresses in private ranges, loopback and cloud metadata addresses are refused. Use TLS (the default); choose TLS, certificate checked to verify the server's certificate against public certificate authorities.

How a pipeline runs

1. Checking. When you save a pipeline, KiviDB connects with its settings and reads one row of each table, so a mistake shows up immediately.
2. Copying. It records where the source's change log is, then copies each table in primary key order, 500 rows at a time.
3. Streaming. It applies every change committed since that point, in order, and keeps going.

Writes replace whole keys, so applying a change twice has no effect. That lets a pipeline resume from its last saved position after a restart without missing or duplicating anything. A failing pipeline shows the error and retries on its own, waiting longer each time, up to 30 minutes.

What it does not do yet: schema changes are not followed (add a column, then save the pipeline again to copy everything afresh); there is no write-back from KiviDB to the source; and key templates take column values as they are, without expressions.

Price

A pipeline costs $0.04 for each hour, or part of an hour, it runs: about $29.20 a month for one that runs all month. Each pipeline may also write 10 GB to its KiviDB database in a calendar month; after that, data is $0.10 per GB, counted from the keys, field names and values it writes, whether copying or streaming changes. A stopped pipeline is charged nothing. Both show on the usage page as Data integration and Data integration transfer.

In the console

The Data integration section lists every pipeline in your organization; each database's Data integration tab lists the pipelines writing into it. A pipeline shows its state (copying, streaming, failing, stopped), rows copied, changes applied, how far behind the source it is, and the last error, and can be stopped, started, made to copy everything again, or deleted.

API

Request
GET /data-integration	every pipeline in the organization
GET /instances/{id}/data-integration	the pipelines writing into a database
PUT /data-integration/{name}	create or change a pipeline (the source is checked first)
GET /data-integration/{name}	one pipeline, with its progress
POST /data-integration/{name}/start	run it, or retry now after a failure
POST /data-integration/{name}/stop	stop it
POST /data-integration/{name}/resnapshot	copy every table again, then carry on streaming
DELETE /data-integration/{name}	delete it and drop its replication slot (?force=true deletes it even when the source cannot be reached, leaving the slot for you to drop)

curl -X PUT https://api.kividb.io/data-integration/orders-cache \
  -H "Authorization: Bearer $KIVIDB_API_KEY" -H "Content-Type: application/json" \
  -d '{
    "instance_id": "'"$DB"'",
    "source_type": "postgres",
    "source_host": "orders.abc123.eu-west-1.rds.amazonaws.com",
    "source_port": 5432,
    "source_database": "shop",
    "source_user": "kividb_cdc",
    "source_password": "'"$PG_PASSWORD"'",
    "tables": [
      { "table": "public.customers", "key": "customer:{id}", "type": "hash", "columns": ["id", "name", "tier"] },
      { "table": "public.orders", "key": "order:{id}", "type": "json", "rename": { "customer_id": "customer" }, "ttl_seconds": 604800 }
    ]
  }'

Table field
table	schema.table for PostgreSQL (public when left out), the table's name for MySQL
key	the key template: text with {column} placeholders, at least one
type	hash (default) or json
columns	copy only these columns; every column when left out
rename	store a column under another name
ttl_seconds	expire each key this long after it was last written

The source password is stored encrypted and never returned.