Managing Incremental Bookmarks
Relevant source files
The following files were used as context for generating this wiki page:
The meltano-state-backend-clickhouse extension provides a mechanism for Meltano to persist pipeline state, such as incremental bookmarks and full-table markers, within a ClickHouse warehouse. This allows for reliable data synchronization by tracking the progress of data integration pipelines across successive runs.
Because ClickHouse is an OLAP column store, it lacks traditional transactional primitives like row-level updates or primary key conflict constraints. To address this, the backend utilizes specific ClickHouse engines and a custom advisory locking mechanism to ensure state consistency and prevent data corruption during concurrent operations.
Sources: README.md:1-12, src/meltano_state_backend_clickhouse/backend.py:1-15
State Storage Architecture
State management is built around a specialized ClickHouse table structure that accommodates the immutability of OLAP data. The system uses the ReplacingMergeTree engine, which handles "updates" by keeping the latest version of a record based on a timestamp.
Data Schema
The state is stored in a table (default meltano.state) defined with the following fields:
| Field | Type | Description |
|---|---|---|
state_id | String | Unique identifier for the pipeline state. |
partial_state | Nullable(String) | JSON string containing in-progress or partial bookmark data. |
completed_state | Nullable(String) | JSON string containing the final state from a successful run. |
updated_at | DateTime64(3) | Versioning timestamp used by the merge engine. |
Sources: src/meltano_state_backend_clickhouse/backend.py:195-207, README.md:27-30
Logic Flow for State Retrieval
When reading state, the backend applies the FINAL modifier to the SQL query. This ensures that ClickHouse collapses all superseded versions of a state_id and returns only the row with the highest updated_at value.
The use of FINAL provides the functional equivalent of a row-level update visibility in a system that otherwise only supports appends.
Sources: src/meltano_state_backend_clickhouse/backend.py:228-243, tests/test_backend.py:61-66
Advisory Locking Mechanism
To manage concurrency and prevent multiple processes from updating the same incremental bookmarks simultaneously, the backend implements a "best-effort" advisory lock. Since ClickHouse does not support INSERT uniqueness errors, a multi-step "settle-then-verify" pattern is used.
Lock Table Schema
A dedicated lock table (default meltano.state_lock) tracks active locks:
| Field | Type | Description |
|---|---|---|
state_id | String | The state identifier being locked. |
lock_id | String | A unique UUID generated by the process attempting to lock. |
locked_at | DateTime | Timestamp used for TTL and stale lock cleanup. |
Sources: src/meltano_state_backend_clickhouse/backend.py:209-218
The Settle-then-Decide Pattern
The acquisition process involves inserting a candidate row, waiting for a "settle" period to account for network jitter, and then verifying if the current process owns the earliest lock for that state ID.
Configuration and Connectivity
The backend is configured primarily via a URI, with optional environment variable overrides for specific ClickHouse settings.
Configuration Parameters
These settings are registered as Meltano entry points and can be defined in meltano.yml or via environment variables.
| Meltano Setting | Env Var | Default |
|---|---|---|
state_backend.clickhouse.host | CLICKHOUSE_HOST | localhost |
state_backend.clickhouse.port | CLICKHOUSE_PORT | 8123 |
state_backend.clickhouse.database | CLICKHOUSE_DATABASE | - |
state_backend.clickhouse.schema | CLICKHOUSE_SCHEMA | meltano |
state_backend.clickhouse.table | CLICKHOUSE_TABLE | state |
Sources: pyproject.toml:27-35, src/meltano_state_backend_clickhouse/backend.py:65-126
Connection String Example
state_backend:
uri: clickhouse://user:password@host:8123/meltano_db
clickhouse:
table: custom_state_table
Sources: README.md:19-25
Key Implementation Components
ClickhouseStateStoreManager
The primary class implementing the Meltano StateStoreManager interface. It handles table initialization, state persistence, and lock coordination.
set(state): Encodes state as JSON and performs a ClickHouseINSERT.get(state_id): Queries the state table usingFINALand decodes the JSON response.acquire_lock(state_id): A context manager that implements the advisory locking logic._ensure_tables(): Automatically creates the database, state table, and lock table if they do not exist.
Sources: src/meltano_state_backend_clickhouse/backend.py:148-226, src/meltano_state_backend_clickhouse/__init__.py:5-7
ReplacingMergeTree Logic
The backend relies on ClickHouse's background merge process to clean up older state versions. The TTL (Time To Live) directive on the lock table is used as a backstop to reap locks from crashed processes after 300 seconds (5 minutes).
Sources: src/meltano_state_backend_clickhouse/backend.py:53-54, src/meltano_state_backend_clickhouse/backend.py:217
Summary
Managing incremental bookmarks in ClickHouse requires a departure from standard relational database patterns. By leveraging the ReplacingMergeTree engine with FINAL queries and implementing a settle-then-decide advisory lock, the backend provides a reliable, distributed state management system suitable for high-performance OLAP environments.