MySQL / MariaDB connector¶

Connects to a MySQL or MariaDB server as both a source (incremental cursor SELECT or binlog CDC) and a batched INSERT sink, exchanging rows as JSON-object strings.

Overview¶

This connector integrates a MySQL or MariaDB server over the synchronous mariadb-connector-c C API. It provides three factories on the std::string channel: a polling cursor source (mysql_source) that issues incremental SELECT statements, a binlog CDC source (mysql_cdc_source) that streams row-level changes from the master's binary log, and a sink (mysql_sink) that writes records as batched multi-row INSERT statements. Every record on the wire is a JSON-object string keyed by column name; the SQL Row path bridges these via json_string_to_row (source) and row_to_json_string (sink). The cursor source uses the MySQL text protocol, so every cell is delivered as a JSON string and the downstream Row bridge coerces it to the declared column type. The CDC source emits flat JSON change rows carrying __op, __table, __lsn and __xid metadata.

Dependency and version¶

Component	Provenance	Version
mariadb-connector-c (libmariadb)	System package via apt (Debian `libmariadb-dev`) / brew (macOS `mariadb-connector-c`)	Not pinned by clink

TLS support is built into mariadb-connector-c, so no separate TLS dependency is required. This connector does not use Apache Arrow on its data path; records are JSON-object strings.

Enabling it¶

The module is gated by the CMake option CLINK_WITH_MYSQL, which defaults to AUTO:

AUTO (default): the connector is built only if mariadb-connector-c is discovered.
ON: build is required; CMake fails with a fatal error if the client library is not found.
OFF: the target is not defined.

The client library is discovered three ways, most portable first: a pkg-config imported target (libmariadb, as shipped by apt's libmariadb-dev), the mariadb CONFIG package, then a manual header and library probe (Homebrew mariadb-connector-c or a bare install). On Debian, install the dependency with the connector deps script or directly:

apt-get install -y libmariadb-dev

cmake -S . -B build -DCLINK_WITH_MYSQL=ON

The target compiles only where the client library is present, defining CLINK_HAS_MYSQL.

Factories¶

Factory name	Direction	Record type
`mysql_source`	Source	`std::string` (JSON object per row)
`mysql_cdc_source`	Source	`std::string` (flat JSON change row)
`mysql_sink`	Sink	`std::string` (JSON object per row)
`mysql_upsert_sink`	Sink	`Row` (changelog upsert/delete by PRIMARY KEY)

Configuration¶

Connection options (shared by all three factories)¶

Option	Required	Default	Description
`host`	No	`localhost`	Server host.
`port`	No	`3306`	Server port.
`user`	No	`` (empty)	Connection user.
`password`	No	`` (empty)	Connection password; an empty value connects without a password.
`database`	No	`` (empty)	Database selected in-band at connect.
`ssl`	No	`false`	`ssl='true'` encrypts the connection (the server is required to support TLS; a plaintext server is refused).
`ssl_ca`	No	`` (empty)	Optional PEM path to the CA used to verify the server certificate.
`ssl_cert`	No	`` (empty)	Optional PEM path to the client certificate (mutual TLS).
`ssl_key`	No	`` (empty)	Optional PEM path to the client key (mutual TLS).
`ssl_verify`	No	`true`	`ssl_verify='false'` skips server-certificate verification (encrypted but not authenticated, for example a self-signed development certificate).

Source: conn_options_from in impls/mysql/src/register_factories.cpp and the ConnectOptions struct in impls/mysql/include/clink/mysql/mysql_client.hpp.

`mysql_sink`¶

Option	Required	Default	Description
`table`	Yes	(none)	Target table.
`columns`	See note	(none)	Comma-separated projection and column order. On the SQL path it defaults to the declared table schema (`schema_columns`), so it is only required when the sink is built outside SQL. Each column is looked up by name in the row's JSON object; an absent or null value becomes SQL NULL.
`on_duplicate`	No	`` (empty)	Empty performs a plain `INSERT`. `on_duplicate='update'` appends `ON DUPLICATE KEY UPDATE` for an idempotent insert-or-update by primary key on replay. This is not clink `mode='upsert'` (a changelog contract with delete tombstones, which this sink does not implement).
`update_columns`	No	`` (empty)	Comma-separated `ON DUPLICATE KEY UPDATE` SET list; empty means all columns.
`batch_records`	No	`1000`	Flush threshold by record count; `0` is coerced to `1`.
`max_bytes`	No	`0`	Byte-based flush threshold; `0` disables it.
`linger_ms`	No	`0`	Linger before flushing a partial batch, in milliseconds; `0` disables it.
`timestamp_columns`	No	derived	Comma-separated `<column>:ts` or `<column>:tstz`: the columns written as timestamps. On the SQL path the planner names every `TIMESTAMP` column. Taken by `mysql_upsert_sink` too.

Timestamps. The SQL Row path carries a TIMESTAMP as epoch milliseconds, and the sink writes a column named in timestamp_columns as its time in UTC ('2026-10-24 17:41:28.105') instead of that number. Both MySQL sinks run their session with time_zone = '+00:00', so a DATETIME column holds the UTC wall time and a TIMESTAMP column the same instant whatever the server's time zone. A value outside years 1 to 9999 fails the batch. To store epoch milliseconds in a BIGINT column, declare the SQL column BIGINT.

Source: the mysql_sink lambda in impls/mysql/src/register_factories.cpp and MysqlSinkOptions in impls/mysql/include/clink/mysql/mysql_sink.hpp.

`mysql_source` (incremental cursor)¶

Option	Required	Default	Description
`table`	Yes	(none)	Source table.
`cursor_column`	Yes	(none)	Ascending, NOT NULL cursor column. Must be unique unless `id_column` is set. The cursor comparison is exclusive (`>`), so the boundary row is never re-emitted.
`id_column`	No	`` (empty)	Optional unique tie-breaker for keyset pagination, so a non-unique `cursor_column` does not drop rows sharing a boundary value at a page (LIMIT) boundary.
`initial_cursor`	No	`` (empty)	Cold-start cursor value; empty re-scans from the beginning until the first checkpoint.
`batch_size`	No	`1000`	Rows per poll (`LIMIT`).
`poll_ms`	No	`1000`	Poll interval in milliseconds.
`jitter_frac`	No	`0`	Fractional jitter applied to the poll interval.
`bounded` / `mode`	No	`false`	`bounded='true'` or `mode='snapshot'` reads the table once and then finishes, rather than tailing it forever.

Source: the mysql_source lambda in impls/mysql/src/register_factories.cpp and MysqlPollOptions in impls/mysql/include/clink/mysql/mysql_source.hpp.

`mysql_cdc_source` (binlog CDC)¶

Option	Required	Default	Description
`server_id`	Yes	`0`	Replica server-id; must be non-zero and unique against the master. A value of `0` is rejected.
`tables`	No	`` (empty)	Comma-separated `db.table` allowlist; empty means all tables. Must be fully qualified `db.table` when `initial_snapshot` is enabled.
`start_file`	No	`` (empty)	Cold-start binlog file; empty starts from the current master head.
`start_pos`	No	`4`	Cold-start binlog position (4 is just past the magic header).
`initial_snapshot`	No	`false`	`initial_snapshot='true'` bootstraps existing rows of `tables` as Insert changes before streaming, captured at a consistent binlog point so existing data and ongoing changes form one gap-free stream.
`snapshot_lock`	No	`true`	With `true`, briefly takes `FLUSH TABLES WITH READ LOCK` (needs the RELOAD privilege) for an exact, overlap-free snapshot point. `snapshot_lock='false'` skips the lock, making the snapshot-to-stream boundary at-least-once.
`heartbeat_ms`	No	`1000`	Master heartbeat period in milliseconds; bounds cancel and checkpoint latency on an idle stream.

The CDC source requires the master to run binlog_format=ROW with binlog_row_image=FULL, and the connecting user to hold REPLICATION SLAVE and REPLICATION CLIENT. Column names and signedness are resolved from information_schema. A binlog stream is single-reader: only subtask 0 streams, and other subtasks are dormant.

Source: the mysql_cdc_source lambda in impls/mysql/src/register_factories.cpp and MysqlCdcOptions in impls/mysql/include/clink/mysql/mysql_cdc_source.hpp.

SQL usage¶

The connector is mapped under connector='mysql' in src/sql/physical_plan.cpp. The source factory is selected by the mode property: mode='cdc' selects mysql_cdc_source, otherwise mysql_source. The sink maps to mysql_sink.

Source table (incremental cursor):

CREATE TABLE orders (
    id BIGINT,
    customer VARCHAR,
    amount DOUBLE
) WITH (
    connector = 'mysql',
    host = 'db.internal',
    port = '3306',
    user = 'reader',
    password = 'secret',
    database = 'shop',
    table = 'orders',
    cursor_column = 'id',
    batch_size = '500',
    poll_ms = '1000'
);

Source table (binlog CDC):

CREATE TABLE orders_cdc (
    id BIGINT,
    customer VARCHAR,
    amount DOUBLE
) WITH (
    connector = 'mysql',
    mode = 'cdc',
    host = 'db.internal',
    user = 'repl',
    password = 'secret',
    server_id = '42',
    tables = 'shop.orders'
);

Sink table:

CREATE TABLE orders_out (
    id BIGINT,
    customer VARCHAR,
    amount DOUBLE
) WITH (
    connector = 'mysql',
    host = 'db.internal',
    user = 'writer',
    password = 'secret',
    database = 'shop',
    table = 'orders_out',
    on_duplicate = 'update'
);

On the SQL path, columns defaults to the declared table schema, so it does not need to be repeated. mode='upsert' with a PRIMARY KEY selects the changelog-aware upsert sink (mysql_upsert_sink, see Delivery semantics), which lets a retracting query maintain the table; on_duplicate='update' is the lighter append-stream idempotent-by-key option on the default sink. Exactly-once (XA 2PC) delivery is not supported.

A NaN or infinite double in a row reaches this sink through the row_to_json_string bridge as null, or as "NaN", "Infinity" or "-Infinity" when the sink table sets json_non_finite='string'; the sink handles it as it handles any null or string value. See NaN and infinity in JSON output.

Example¶

The sink and cursor source can be constructed directly from their typed Options, as the live test does in impls/mysql/tests/test_mysql_live.cpp:

#include "clink/mysql/mysql_sink.hpp"
#include "clink/mysql/mysql_source.hpp"

using namespace clink::mysql;

ConnectOptions conn;
conn.host = "db.internal";
conn.user = "writer";
conn.password = "secret";
conn.database = "shop";

// Sink: batched INSERT of JSON-object rows.
MysqlSinkOptions sink_opts;
sink_opts.conn = conn;
sink_opts.table = "orders_out";
sink_opts.columns = {"id", "val"};
sink_opts.upsert = true;  // INSERT ... ON DUPLICATE KEY UPDATE
MysqlSink sink(std::move(sink_opts));
sink.open();
// sink.on_data(batch); ... flushed on count/byte/linger thresholds and barriers.

// Source: incremental cursor SELECT.
MysqlPollOptions src_opts;
src_opts.conn = conn;
src_opts.table = "orders";
src_opts.cursor_column = "id";
src_opts.batch_size = 500;
auto source = make_mysql_poll_source(src_opts);

The factories are also reachable through the registry on the string channel as mysql_sink, mysql_source and mysql_cdc_source once clink::mysql::install(registry) has run.

Delivery semantics¶

mysql_sink: at-least-once. A replay after a failed checkpoint re-runs the INSERT. A plain insert appends duplicates on replay; with on_duplicate='update' (and a PRIMARY KEY or UNIQUE index) the insert-or-update is idempotent and so effectively-once by key. Buffered rows are flushed on the count, byte and linger thresholds and on every checkpoint barrier. A flush failure drops the connection, clears the pending buffer and rethrows so the job replays from the last checkpoint. Exactly-once (XA 2PC) is not supported.
mysql_upsert_sink (mode='upsert'): effectively-once on the sink table for a stable PRIMARY KEY and a deterministic defining query. Consumes the changelog by key - insert/update_after INSERT ... ON DUPLICATE KEY UPDATE, delete/update_before DELETE ... WHERE <pk> IN (...) - netted by key within a flush and applied in one transaction, so a replay converges the table to the same state. Lets a retracting query (GROUP BY, TOP-N, outer join) maintain a MySQL table. Not two-phase commit.
mysql_source: at-least-once. The cursor is checkpointed as operator state (via PollingSource) and replayed on restart; the exclusive > comparison avoids re-emitting the boundary row. A NULL cursor or id value in a delivered row is fatal rather than silently stalling the cursor.
mysql_cdc_source: at-least-once for the decodable change stream, with the checkpoint cursor being (binlog_file, position). An undecodable row event (schema drift or an unseen table-map) is dropped and counted, so it is at-most-once for those events; it is never emitted half-populated. With initial_snapshot enabled and snapshot_lock=true the snapshot-to-stream boundary is exact; with snapshot_lock=false it is at-least-once. The snapshot is not checkpointed mid-way, so a crash during it re-snapshots from scratch on restart.

Limitations¶

Each sink must be the only sink on its chain: at every checkpoint barrier it writes out what it received before the barrier, which holds only while it owns its chain's checkpoint, so Dag::add_sink refuses another sink beside it, whichever is added first. SQL and job graphs plan every sink as a subtask of its own; with the Dag API, put the other sink in a separate Dag (see checkpointing).
The cursor source uses the MySQL text protocol, so every cell is delivered as a string; numeric and date typing depends on the downstream Row bridge coercing to declared column types.
A VARCHAR cursor_column orders lexically (so '10' < '9'); choose a numeric or zero-padded key.
A non-unique cursor_column can drop rows at a page boundary unless id_column is set as a unique tie-breaker.
mysql_cdc_source requires binlog_format=ROW and binlog_row_image=FULL on the master, and REPLICATION SLAVE plus REPLICATION CLIENT privileges; column names are resolved from information_schema.
A binlog stream is single-reader: only subtask 0 streams in a parallel CDC job; other subtasks emit nothing.
CDC initial_snapshot requires tables to be a non-empty, fully-qualified db.table list, and a consistent snapshot relies on InnoDB MVCC (a non-transactional engine such as MyISAM has no consistent-snapshot view).
The sink does not implement clink mode='upsert' (changelog delete tombstones); for idempotent insert-or-update use on_duplicate='update'.
Exactly-once delivery is not supported by the sink.
The channel is std::string only (JSON-object rows); there is no typed int64 channel and no Arrow or Parquet encoding on this connector's data path.

Testing¶

In-process tests cover factory registration, sink and source SQL building, CDC decode and snapshot logic without a server (impls/mysql/tests/test_factory_registration.cpp, test_mysql_sink_logic.cpp, test_mysql_source_logic.cpp, test_mysql_cdc_decode.cpp, test_mysql_snapshot.cpp).

The live integration tests (test_mysql_live.cpp and test_mysql_cdc_live.cpp) are skipped unless the environment variable CLINK_MYSQL_TEST_DSN is set; each is a space-separated key=value DSN, for example:

export CLINK_MYSQL_TEST_DSN="host=localhost port=3306 user=root password=mysql database=test"

The fixture server is the mysql:8.0 image defined in docker/integration-services.yml, started with --server-id=1 --log-bin=mysql-bin --binlog-format=ROW --binlog-row-image=FULL so the binlog CDC source can stream changes. Run the live tests with:

ctest --test-dir build -L mysql -R Live --output-on-failure