Meltano State Backend Contract
Relevant source files
The following files were used as context for generating this wiki page:
The Meltano State Backend for ClickHouse provides a mechanism for Meltano to persist pipeline state—such as incremental bookmarks and full-table markers—within a ClickHouse warehouse. This implementation allows ClickHouse to function as a state storage provider alongside other supported backends like Postgres or Snowflake.
Sources: README.md:3-8, src/meltano_state_backend_clickhouse/backend.py:1-5
Because ClickHouse is an OLAP (Online Analytical Processing) store, it lacks traditional row-level updates and transaction primitives. The backend manages this by utilizing the ReplacingMergeTree engine for state versioning and implementing a custom "best-effort" advisory locking mechanism to handle concurrent state access.
Sources: src/meltano_state_backend_clickhouse/backend.py:8-16, README.md:33-35
Architecture and Components
The core of the system is the ClickhouseStateStoreManager, which inherits from Meltano's base StateStoreManager. It handles connection management, table initialization, and the CRUD operations required by the Meltano state contract.
Sources: src/meltano_state_backend_clickhouse/backend.py:192-196, src/meltano_state_backend_clickhouse/init.py:7-9
Class Structure
The following diagram illustrates the relationship between the manager and the ClickHouse client.
Data Model and Storage
The backend utilizes two primary tables: a state table for persisting pipeline data and a lock table for managing concurrency. Both tables are created within a specified schema (defaulting to meltano).
Sources: src/meltano_state_backend_clickhouse/backend.py:246-250, README.md:29-31
State Table
The state table uses the ReplacingMergeTree engine. This engine allows multiple versions of a state for the same state_id to exist; however, queries use the FINAL modifier to ensure only the most recent version (highest updated_at) is retrieved.
Sources: src/meltano_state_backend_clickhouse/backend.py:254-263, README.md:37-38
| Field | Type | Description |
|---|---|---|
state_id | String | Unique identifier for the pipeline state. |
partial_state | Nullable(String) | JSON string containing incremental state data. |
completed_state | Nullable(String) | JSON string containing finalized state data. |
updated_at | DateTime64(3) | Versioning timestamp used by ReplacingMergeTree. |
| Sources: src/meltano_state_backend_clickhouse/backend.py:256-261 |
Lock Table
The lock table manages advisory locks for state IDs. It includes a TTL (Time To Live) to automatically reap stale locks after a defined period (default 300 seconds).
Sources: src/meltano_state_backend_clickhouse/backend.py:270-280, src/meltano_state_backend_clickhouse/backend.py:53
| Field | Type | Description |
|---|---|---|
state_id | String | The ID being locked. |
lock_id | String | Unique UUID for the lock attempt. |
locked_at | DateTime | Timestamp when the lock was requested. |
| Sources: src/meltano_state_backend_clickhouse/backend.py:272-274 |
Concurrency Control (Best-effort Locking)
ClickHouse does not support standard primary key violation errors or row-level locking. To compensate, the backend implements a "settle-then-decide" protocol during lock acquisition.
Sources: src/meltano_state_backend_clickhouse/backend.py:12-16, src/meltano_state_backend_clickhouse/backend.py:382-390
Lock Acquisition Flow
The acquisition process involves inserting a candidate row, waiting for a "settle" period, and then verifying if the current process owns the lowest lock_id for that state_id.
Configuration and Settings
The backend can be configured via an environment variable MELTANO_STATE_BACKEND_URI or directly in meltano.yml. The URI format is clickhouse://user:password@host:port/database.
Sources: README.md:17-27
Supported Settings
The following settings are registered as Meltano entry points.
Sources: pyproject.toml:31-40, src/meltano_state_backend_clickhouse/backend.py:73-138
| Setting | Description | Default |
|---|---|---|
host | ClickHouse server hostname | localhost |
port | HTTP interface port | 8123 |
database | Target ClickHouse database | Required |
user | Connection username | Required |
password | Connection password | "" |
secure | Use HTTPS/TLS | False |
schema | Database/Schema for state storage | meltano |
table | Table name for state storage | state |
Technical Implementation Details
State Retrieval
Retrieval logic uses the FINAL keyword to collapse duplicates. The backend automatically handles JSON serialization and deserialization for the state payloads.
result = self.client.query(
f"SELECT partial_state, completed_state FROM {self.state_table} FINAL "
"WHERE state_id = {state_id:String} LIMIT 1",
parameters={"state_id": state_id},
)
Sources: src/meltano_state_backend_clickhouse/backend.py:307-311
State Updates
Since ClickHouse does not support UPDATE, the set method performs a standard INSERT. The ReplacingMergeTree engine ensures that during background merges or FINAL queries, the record with the most recent updated_at timestamp is treated as the current state.
Sources: src/meltano_state_backend_clickhouse/backend.py:291-301, src/meltano_state_backend_clickhouse/backend.py:8-11
Conclusion
The Meltano State Backend for ClickHouse effectively adapts Meltano's state management contract to the OLAP constraints of ClickHouse. By leveraging ReplacingMergeTree and a custom distributed locking algorithm involving a settle period, it provides a reliable state storage solution that supports concurrent pipeline operations.
Sources: src/meltano_state_backend_clickhouse/backend.py:382-408, tests/test_backend.py:101-118