State Persistence Workflow

Relevant source files

The following files were used as context for generating this wiki page:

• src/meltano_state_backend_clickhouse/backend.py
• src/meltano_state_backend_clickhouse/__init__.py
• tests/test_backend.py
• README.md
• pyproject.toml

The State Persistence Workflow in the ClickHouse state backend for Meltano enables the reliable storage and retrieval of pipeline progress markers (such as incremental bookmarks and full-table markers). Because ClickHouse is an OLAP (Online Analytical Processing) store, this workflow implements specific strategies to handle the lack of traditional row-level updates and ACID transaction primitives typically found in OLTP databases like Postgres or MSSQL.

The core of this workflow is the ClickhouseStateStoreManager, which manages the lifecycle of pipeline state through a specialized table schema using the ReplacingMergeTree engine. This ensures that the most recent version of a state for a given state_id is always the one processed, while providing a "best-effort" advisory locking mechanism to prevent concurrent write collisions.

Sources: README.md:1-12, src/meltano_state_backend_clickhouse/backend.py:1-20

Architecture and Data Modeling

The workflow relies on two primary tables within a dedicated schema (defaulting to meltano): the state table and the lock table.

State Storage Schema

State is stored in a table (defaulting to state) that uses the ReplacingMergeTree engine. This engine is critical because ClickHouse does not support row-level UPDATE statements. Instead, new versions of state are inserted as new rows, and ClickHouse merges them in the background based on the updated_at timestamp.

The following ER diagram illustrates the storage structure:

Table Definitions

Table	Column	Type	Description
State Table	state_id	String	The unique identifier for the pipeline state.
	partial_state	Nullable(String)	JSON string representing the partial state.
	completed_state	Nullable(String)	JSON string representing the completed state.
	updated_at	DateTime64(3)	Versioning key for ReplacingMergeTree.
Lock Table	state_id	String	The ID of the state being locked.
	lock_id	String	UUID generated by the process attempting the lock.
	locked_at	DateTime	Timestamp used for TTL and versioning.

Sources: src/meltano_state_backend_clickhouse/backend.py:182-205

State Retrieval and Persistence Logic

The workflow follows a specific "read-modify-write" pattern. Due to the nature of ClickHouse, read operations must explicitly use the FINAL keyword to ensure only the latest version of the state is retrieved.

Retrieval Flow

When get(state_id) is called, the system queries the state table using FINAL. This collapses all historical rows for that state_id and returns the row with the highest updated_at value.

Persistence Flow

When set(state) is called, the system performs a standard INSERT. The new row naturally supersedes previous rows because its updated_at (defaulting to now64(3)) will be greater than existing rows.

Advisory Locking Mechanism

Since ClickHouse lacks INSERT uniqueness/conflict errors and transactional locking, the workflow implements a best-effort advisory lock. This is necessary to prevent two concurrent pipeline runs from overwriting each other's state markers.

Lock Acquisition Process

The lock acquisition uses a "settle-then-decide" strategy:

1. Cleanup: Stale locks (older than 300 seconds) are deleted.
2. Insert: A candidate row is inserted with a unique lock_id (UUID).
3. Settle: The process waits for LOCK_SETTLE_SECONDS (1.0s) to allow concurrent insertions from other processes to become visible.
4. Verify: The process queries all rows for the state_id (without FINAL).
5. Tie-break: The process with the lexicographically lowest lock_id wins the lock. Others delete their candidate rows and retry.

Configuration and Initialization

The workflow is configured via Meltano settings or environment variables. The ClickhouseStateStoreManager initializes the environment by ensuring the database and necessary tables exist upon startup.

Connection Parameters

Setting	Env Var	Default
host	MELTANO_STATE_BACKEND_CLICKHOUSE_HOST	localhost
port	MELTANO_STATE_BACKEND_CLICKHOUSE_PORT	8123
database	MELTANO_STATE_BACKEND_CLICKHOUSE_DATABASE	(Required)
schema	MELTANO_STATE_BACKEND_CLICKHOUSE_SCHEMA	meltano
table	MELTANO_STATE_BACKEND_CLICKHOUSE_TABLE	state

Sources: pyproject.toml:29-37, src/meltano_state_backend_clickhouse/backend.py:64-120

Initialization Code Snippet

The _ensure_tables method is called during initialization to set up the environment:

def _ensure_tables(self) -> None:
    self.client.command(f"CREATE DATABASE IF NOT EXISTS `{self.schema}`")
    self.client.command(
        f"""
        CREATE TABLE IF NOT EXISTS {self.state_table} (
            state_id String,
            partial_state Nullable(String),
            completed_state Nullable(String),
            updated_at DateTime64(3) DEFAULT now64(3)
        ) ENGINE = ReplacingMergeTree(updated_at)
        ORDER BY state_id
        """
    )

Sources: src/meltano_state_backend_clickhouse/backend.py:175-191

Summary

The State Persistence Workflow transforms ClickHouse into a viable state backend for Meltano by leveraging the ReplacingMergeTree engine for versioned updates and a custom settle-and-verify algorithm for advisory locking. This architecture ensures that even in a high-concurrency OLAP environment, pipeline state remains consistent and durable.

Sources: src/meltano_state_backend_clickhouse/backend.py:10-20, README.md:30-38