Search docs.molo17.com

    Vertica logo Vertica CDC: Epoch-based change data capture with Gluesync

    Source data from Vertica

    The Vertica source agent (agent tag vertica) reads a snapshot of your tables and then captures their changes as the difference between two epochs of the database. It creates no triggers, no shadow tables and no objects of its own, and it never writes to the source: the agent only runs SELECT statements.

    The same agent is also a target.

    How it works

    Vertica has no log that a client can read, no change stream and no row triggers. What it has is a version number for every commit, the epoch, and that is enough to compute an exact change set:

    • Every committed row carries the epoch of its commit in a hidden column named epoch. An UPDATE or a MERGE is a delete plus an insert in Vertica, so the new version of a row carries the new epoch;

    • AT EPOCH n SELECT …​ reads the database as it was when epoch n was committed. It takes no locks and works for any epoch that the Ancient History Mark (AHM) has not passed;

    • v_monitor.system reports the current (open) epoch and the AHM. The latest closed epoch is the current epoch minus one.

    The agent keeps one position per entity: the last epoch whose changes it delivered, and the number of rows the table held at that epoch. The row count is how the agent notices that the history of the table was lost (see When the history is lost).

    Snapshot

    The snapshot reads every table AT EPOCH of one epoch, pinned when the snapshot starts, so the rows it reads are one consistent picture of the table while applications keep writing. CDC starts from that same epoch: a change committed while the snapshot runs is delivered by CDC afterwards, and nothing is lost or delivered twice between the two.

    The table is read in parallel, in N hash buckets (snapshotReadParallelism, default 4): bucket i reads the rows where MOD(HASH(keys), N) = i, over the keys of the entity (a table without keys is split on its first column). Each bucket uses its own connection, so a restart of the snapshot resumes every bucket where it stopped. When you configure partitions for a table, the agent reads those ranges instead of the hash buckets, still at the pinned epoch. Set snapshotReadParallelism to 1 to read each table on one connection.

    Before reading, the agent checks that the pinned epoch has not been passed by the AHM. If it has, the entity fails with a message that names HistoryRetentionTime and the snapshot has to be restarted (see History retention).

    Change data capture

    For each entity, the agent polls the database (at most once every polling interval of the entity, pollingIntervalMilliseconds, default 500 ms) and delivers the changes committed since its position:

    1. It reads the latest closed epoch. When it is the epoch of the position, there is nothing to deliver;

    2. It checks that the history it needs is still there. If it is not, the table is reloaded or the entity fails, depending on resyncOnHistoryLoss (see When the history is lost);

    3. It counts the rows of the table at the new epoch and how many of them are older than the position. Those two numbers tell, without reading any row, how many rows that existed at the position are gone at the new epoch (deleted, or the old version of an updated row);

    4. It reads the rows whose epoch is newer than the position. When no old row is gone, they are all inserts. When some are gone, the agent finds out which keys disappeared by comparing the keys of the table at the old epoch with the keys that survived, in buckets of at most deleteDetectionBucketSize keys (default 1000000) so that memory stays bounded. A changed row whose key was gone is an update; a gone key with no new row is a delete;

    5. The changes of the window are sent to the target in chunks, and the new position is saved only with the last one. A crash in the middle of a window resumes from the position the window started from and reads the same window again: the result is the same net change, so replaying it is harmless.

    Each of these statements is a historical read: it takes no locks on your tables and writes nothing.

    What a window delivers:

    • Insert: every row whose key was not on the table at the position;

    • Update: every row whose key existed at the position and whose row changed. The target receives the full row;

    • Key change: an update that changes a key column reaches the target as a delete of the old key and an insert of the new one;

    • Delete: the key columns of the row. A row that was inserted and deleted inside the same window is never delivered.

    The agent delivers the net change of each poll window. See Limitations.

    Prerequisites

    To have Gluesync working on your Vertica instance as a source you will need:

    • Vertica 24.1. The agent is tested against Vertica 24.1;

    • History retention long enough for the window you need: HistoryRetentionTime of the database set to at least one hour. See History retention;

    • A database user that can SELECT from the tables to replicate and that has USAGE on their schemas and on the public schema. See Privileges;

    • Room for the sessions of the agent in MaxClientSessions. See Sessions;

    • Network access from Core Hub to Vertica, on its client port (default 5433).

    History retention

    CDC reads changes by comparing epochs, so the epoch of the agent’s position must still be readable. By default HistoryRetentionTime is 0: the AHM then follows the last good epoch every AdvanceAHMInterval (180 seconds by default), and a read AT EPOCH of an epoch older than the AHM fails with ERROR 3183: Epoch number out of range. With the default, a pipeline that pauses for a few minutes, or a snapshot that runs for a few minutes, loses the epoch it needs.

    A superuser keeps the history for a day (86400 seconds) with:

    ALTER DATABASE DEFAULT SET HistoryRetentionTime = 86400;

    Gluesync needs at least minHistoryRetentionSeconds (default 3600, one hour); pick a value longer than the longest pause or outage of your pipeline you want to survive without a reload of the tables. -1 keeps all the history. Vertica keeps deleted rows for as long as the history it retains, so a long retention costs some storage on tables that change a lot.

    Privileges

    The agent works with a login that is not a superuser and holds no role. It needs:

    • USAGE on each schema of the tables to replicate, and SELECT on the tables;

    • USAGE on the schema public. The geospatial functions (ST_AsText and the like) live there, and the agent reads GEOMETRY and GEOGRAPHY columns through them. Without it, a user that is not a superuser fails with Function ST_GeomFromText(unknown) does not exist, or permission is denied.

    CREATE USER gluesync IDENTIFIED BY 'password';

GRANT USAGE ON SCHEMA public TO gluesync;
GRANT USAGE ON SCHEMA sales TO gluesync;
GRANT SELECT ON ALL TABLES IN SCHEMA sales TO gluesync;

-- Tables created later in the schema inherit the privileges of the schema
ALTER SCHEMA sales DEFAULT INCLUDE PRIVILEGES;

    Sessions

    The snapshot opens one connection per bucket, so a table read with snapshotReadParallelism 4 uses four sessions, and every entity uses sessions of its own for CDC. Keep MaxClientSessions above the number of sessions that all the pipelines of the database use at the same time, with room for snapshotReadParallelism extra sessions for each table that is snapshotted at once:

    ALTER DATABASE DEFAULT SET MaxClientSessions = 100;

    Setup via Web UI

    • Hostname / IP Address: DNS record or IP Address of your Vertica server;

    • Port: Optional, defaults to 5433;

    • Database name: Name of the database to read;

    • Username: Username with the privileges listed in the prerequisites;

    • Password: Password belonging to the given username;

    • Enable TLS: Optional, defaults to false. When it is on, the agent connects with TLS; the certificate can be uploaded through Core Hub (certificatePath in the REST API);

    Custom properties

    • Use connection pool: (optional, defaults to true) Uses a pool of connections for reads. Turning it off lowers the performance.

    The other custom properties of the agent (Gluesync schema name and Disable automatic schema creation) apply to the target.

    Specific configuration

    Setting Default Description

    snapshotReadParallelism (Snapshot read parallelism)

    4

    Number of parallel buckets, and so of connections, a table is read with during the snapshot. 1 disables the parallel read. In a lab test on a single-node Vertica, four connections read about twice as fast as one and eight were no faster than four. Every bucket needs a session: see Sessions.

    resultBufferSizeBytes (JDBC result buffer size in bytes)

    8388608 (8 MiB)

    Size of the buffer of the JDBC driver (ResultBufferSize) that rows are streamed through. A larger buffer reads faster at the cost of memory for each connection.

    deleteDetectionBucketSize (Delete detection bucket size)

    1000000

    Number of keys held in memory at once when the agent works out which rows were deleted. A window that deletes more rows is split in as many buckets as needed. Lower it when Core Hub is short of memory; keep it high otherwise, because every bucket is an extra read of the table.

    resyncOnHistoryLoss (Reload the table when its history is gone)

    true

    What happens when the history the agent needs is gone (TRUNCATE, DROP_PARTITIONS, a purge, or the AHM moved past the position). When true, the table is reloaded on the target; when false, the entity fails with an error. See When the history is lost.

    minHistoryRetentionSeconds (Minimum history retention)

    3600

    The HistoryRetentionTime that the setup check of a CDC entity requires. See Setup checks.

    Technical fields

    none

    Adds TRANSACTION_ID, TRANSACTION_TIMESTAMP and TRANSACTION_OPERATION to the replicated rows. All the rows of a poll window carry the same TRANSACTION_ID and TRANSACTION_TIMESTAMP, as the agent does not know the order of the changes inside a window. See Source Agent Advanced Settings.

    For each table, a SQL Snapshot where clause filter limits the rows the snapshot reads, and the rows CDC reads, as it does on the other SQL agents.

    Setup via Rest APIs

    Here following an example of calling the Core Hub’s Rest API via curl to setup the connection for this Agent.

    Connect the agent

    curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/credentials' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
    "agentType": "SOURCE",
    "agentTag": "vertica",
    "hostCredentials": {
        "connectionName": "myAgentNickName",
        "host": "vertica-host-address",
        "port": 5433,
        "databaseName": "db_name",
        "username": "gluesync",
        "password": "your_password",
        "maxConnectionsCount": 100,
        "enableTls": false
    }
}'

    Setup specific configuration

    curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/specific' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
  "configuration": {
      "snapshotReadParallelism": 4,
      "resultBufferSizeBytes": 8388608,
      "deleteDetectionBucketSize": 1000000,
      "resyncOnHistoryLoss": true,
      "minHistoryRetentionSeconds": 3600
    }
}'

    Setup checks

    When a CDC entity is set up, the agent runs two checks. A snapshot-only entity skips both.

    Check Epoch History Retention reads HistoryRetentionTime and HistoryRetentionEpochs from the configuration parameters of the database:

    • It passes when HistoryRetentionTime is -1 or at least minHistoryRetentionSeconds;

    • When HistoryRetentionEpochs is greater than 0 it only warns: the number of epochs a database commits in a given time depends on its load, so Gluesync cannot tell how long the history lasts. The warning is posted as a notification and logged:

      Vertica keeps <n> epochs of history (HistoryRetentionEpochs). That is enough only while the database commits less than that between two reads of a table: Gluesync cannot tell how long that is.

    • When the value cannot be read (the user may not be allowed to read the settings), it warns too, instead of refusing:

      The HistoryRetentionTime of the database could not be read. Gluesync reads changes from the history of the database, which has to span at least <seconds> seconds.

    • Otherwise the setup fails with the remedy:

      The history of the database spans <seconds> seconds (HistoryRetentionTime), too little to read changes from: Gluesync needs at least <minimum>. A superuser sets it with ALTER DATABASE DEFAULT SET HistoryRetentionTime = <minimum>;

    Check CDC Keys fails for a CDC entity that has no key column, naming the table:

    Entity <name> has no key column on <table>, and Vertica changes are matched by key. Choose the key columns of the entity, or make it snapshot only.

    Keys

    CDC matches changes by key: it is how the agent tells an updated row from a deleted one. An entity that runs CDC needs at least one key column, either the primary key of the table or the keys chosen on the entity. A Vertica primary key constraint is metadata the server does not have to enforce, so a table often has none: choose its keys on the entity. A table with no key at all can only be replicated by snapshot.

    When the history is lost

    Some operations remove the history that CDC compares: TRUNCATE removes the table’s storage together with its history, DROP_PARTITIONS and a purge remove the rows of the past, and the AHM can move past the position of a pipeline that was paused for longer than HistoryRetentionTime. The agent notices because the epoch of its position is no longer readable, or because the table held at that epoch a different number of rows than the agent left it with. The agent never skips such a loss silently:

    • With resyncOnHistoryLoss true (the default), it logs a warning that names the cause and reloads the table: one TRUNCATE of the target table, followed by every row of the table at the latest epoch as an insert. Changes go on from there. The target is short of rows until the reload completes;

    • With resyncOnHistoryLoss false, the entity fails with Entity <name> cannot be read as changes from its last position: <reason>. Reload it with a snapshot, or enable resyncOnHistoryLoss to let the agent reload it. Snapshot the entity again to recover.

    A table that did not change while its history was gone is not a loss: if the AHM passed the position of a paused pipeline but the table is what the position says it was, the agent has nothing to deliver and carries on.

    The same limit applies to a snapshot: when the AHM passes the epoch a snapshot was pinned to before the snapshot finishes reading, the entity fails with a message that names HistoryRetentionTime, and the snapshot has to be restarted after raising it.

    Data types

    Vertica reports every integer type (INT, BIGINT, SMALLINT, TINYINT, INT8) as int, which is 64 bit, and every float type (FLOAT, FLOAT8, REAL, DOUBLE PRECISION) as float, which is 64 bit. MONEY is reported as numeric(18,4), DATETIME and SMALLDATETIME as timestamp, and BYTEA as varbinary.

    Vertica type Gluesync type Notes

    int

    LONG

    64 bit. The target column of a LONG stays int on a Vertica target; INT and SHORT also map to int.

    float

    DOUBLE

    64 bit, read without losing precision. NaN and +-Infinity are kept.

    numeric(p,s)

    BIG_DECIMAL

    The declared precision and scale are kept.

    boolean

    BOOLEAN

    char, varchar, long varchar

    STRING

    A CHAR(n) value keeps its padding: Vertica stores and returns a fixed-length value, so CHAR(10) holding ab arrives as ten characters, and an empty one as ten spaces.

    binary, varbinary, long varbinary

    BYTE_ARRAY

    date

    LOCAL_DATE

    time

    LOCAL_TIME

    timetz

    OFFSET_TIME

    The offset is kept.

    timestamp

    LOCAL_DATE_TIME

    timestamptz

    OFFSET_DATE_TIME

    Delivered at UTC (offset +00:00), whatever the time zone of the session.

    interval (day to second), interval year to month

    STRING

    An ISO-8601 duration with the sign on every component: P10DT11H12M13.5S, P-5DT-3H-0.000001S, PT0S, P-14Y-3M, P0M. Vertica prints -5 03:00:00.000001 with the sign on the whole value, and a target that read that text would take it as minus 5 days plus 3 hours, which is why the agent writes the exact sign of every component. A STRING column cannot be a partition column, so an interval cannot be used to partition a snapshot.

    uuid

    STRING

    geometry, geography

    STRING

    The well-known text of the value, read with ST_AsText. Needs USAGE on the schema public.

    array[…​], set[…​]

    ARRAY

    ROW(…​)

    MAP

    Vertica cannot UPDATE an ARRAY, SET or ROW column. A table changes them by deleting and inserting the row again with the same key, which the agent delivers as an update when both happen in the same poll window.

    A column of a type the agent does not know is reported as unsupported and skipped with a warning in the Core Hub log; it does not stop the entity.

    Limitations

    Epoch-based CDC has no log to read, and that shapes what it can deliver:

    • Net changes only. The agent delivers the difference between two epochs, one poll window at a time. A row that changes three times in a window is delivered once, with its last values; a row inserted and deleted inside the same window is never delivered. There are no intermediate versions and no transaction boundaries inside a window, and the technical fields TRANSACTION_ID and TRANSACTION_TIMESTAMP are the same for every row of a window;

    • A key change is a delete and an insert. An update of a key column reaches the target as a delete of the old key plus an insert of the new one, and two rows that exchange their keys in one window are delivered as two updates;

    • A table without keys is replicated by snapshot only. CDC needs a primary key or keys chosen on the entity (see Keys);

    • The cost of a poll grows with the table. Each poll counts the rows of the table at two epochs, so a table of many millions of rows costs more per poll than a small one, however few rows changed. Raise the entity’s polling interval (pollingIntervalMilliseconds) to poll large tables less often; the latency of a change is about that interval plus the time of a poll;

    • The history must cover the pause. A pipeline that is paused or down for longer than the history retention of the database reloads the table (or fails with resyncOnHistoryLoss off). Size HistoryRetentionTime for the longest outage you want to survive;

    • Single-table entities. An entity built from a query or from several tables is not supported as a source;

    • EXPORT TO PARQUET is not used. The snapshot always reads over JDBC, in parallel buckets, so no shared object storage is required.

    Troubleshooting

    The setup fails with too little to read changes from

    HistoryRetentionTime of the database is lower than minHistoryRetentionSeconds. Run the statement from the message as a superuser, see History retention.

    ERROR 3183: Epoch number out of range

    The epoch a read asked for is older than the AHM. Raise HistoryRetentionTime (see History retention). A snapshot has to be restarted; CDC reloads the table or fails, depending on resyncOnHistoryLoss, see When the history is lost.

    Function ST_GeomFromText(unknown) does not exist, or permission is denied

    The user lacks USAGE on the schema public, where the geospatial functions are. Run GRANT USAGE ON SCHEMA public TO <user>;.

    Connections are refused during a snapshot

    The database ran out of sessions. Raise MaxClientSessions or lower snapshotReadParallelism, see Sessions.

    The table is reloaded again and again

    Something removes the history of the table at each window: a job that truncates it, or a purge. A TRUNCATE removes the history of the table, so each one makes the agent reload the table. Find the job, or change the process so that it deletes rows instead of truncating.

    The changes arrive late

    Look at the cost of a poll on large tables and raise the entity’s polling interval (pollingIntervalMilliseconds) to match it: the Core Hub log, at debug level, shows the epochs and the number of rows of every window.