Introduction to Meltano State Backends

Relevant source files

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

• README.md
• src/meltano_state_backend_postgresql/backend.py
• pyproject.toml
• tests/test_integration.py
• tests/test_backwards_compat.py
• tests/test_conninfo.py

The meltano-state-backend-postgresql package is a Meltano extension that implements a state backend using PostgreSQL. Its primary purpose is to provide a persistent storage layer for Meltano job states, allowing for incremental data integration by tracking what data has already been processed. It acts as an alternative to the default system database or other cloud-based state backends.

Sources: README.md:1-12, src/meltano_state_backend_postgresql/backend.py:1-15

Architecture and Core Components

The backend is built around the PostgreSQLStateStoreManager class, which extends Meltano's base StateStoreManager. It manages the lifecycle of the connection to a PostgreSQL instance and handles the mapping between Meltano's state objects and the underlying database schema.

Backend Logic Flow

When Meltano operations (like meltano run) require state, the manager handles connection initialization, table creation (if missing), and CRUD operations on the state data.

The diagram shows the relationship between Meltano core, the state store manager, and the PostgreSQL database operations.
Sources: src/meltano_state_backend_postgresql/backend.py:126-150, pyproject.toml:44-45

Data Model and Schema

State is stored in a single table, defaulting to the name state. The schema is designed to hold both partial and completed state information in JSON format, associated with a unique job identifier.

Table Schema Definition

The table is automatically created by the backend if it does not exist during initialization.

Field	Type	Description
state_id	VARCHAR(900)	Primary Key. The unique identifier for the Meltano job/state.
partial_state	TEXT	Stores the job's partial state as a JSON string.
completed_state	TEXT	Stores the job's completed state as a JSON string.
updated_at	TIMESTAMP	Automatic timestamp of the last update.

Sources: src/meltano_state_backend_postgresql/backend.py:176-187, tests/test_integration.py:84-106

Table Creation Logic

CREATE TABLE IF NOT EXISTS state (
    state_id VARCHAR(900) PRIMARY KEY,
    partial_state TEXT,
    completed_state TEXT,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)

Sources: src/meltano_state_backend_postgresql/backend.py:180-185

Configuration and Connectivity

The backend can be configured using a standard PostgreSQL connection URI or through individual setting definitions. The conninfo_from_params function merges URI-based configurations with explicit setting overrides.

Connection Parameters

Parameter	Type	Env Variable	Default	Description
host	String	...HOST	localhost	Server hostname.
port	Integer	...PORT	5432	Server port.
database	String	...DATABASE	-	Target database name.
user	String	...USER	-	Authentication username.
password	String	...PASSWORD	-	Authentication password.
schema	String	...SCHEMA	search_path	Target schema for the state table.
sslmode	String	...SSLMODE	prefer	SSL connection mode.
table	String	...TABLE	state	Name of the storage table.

Sources: src/meltano_state_backend_postgresql/backend.py:38-100, README.md:43-69

Connection URI Format

The URI follows the standard PostgreSQL format, with support for specific options to define the schema:
postgresql://<user>:<password>@<host>:<port>/<database>?options=-csearch_path%3D<schema>

Sources: README.md:29-33, tests/test_conninfo.py:33-40

State Concurrency and Locking

To prevent race conditions when multiple processes attempt to update the same state simultaneously, the backend implements a locking mechanism using PostgreSQL Advisory Locks.

Locking Sequence

1. The state_id is converted into a deterministic 64-bit signed integer using an MD5 hash.
2. The backend attempts to acquire a lock using pg_try_advisory_lock.
3. If the lock is held by another process, the manager retries for up to 30 seconds.
4. Once acquired, the operation is performed, and the lock is released using pg_advisory_unlock.

The sequence diagram illustrates the advisory lock acquisition and release process for a specific state_id.
Sources: src/meltano_state_backend_postgresql/backend.py:330-388, tests/test_state.py:126-160

Compatibility and Integration

The backend is designed to be compatible with standard Meltano workflows and includes specific handling for backwards compatibility with the legacy systemdb schema.

• Entry Points: The backend registers itself as a meltano.state_backends entry point named postgresql.
• Version Requirements: Requires Meltano >= 3.7. Context manager support for the backend specifically requires Meltano 4.2+.
• Migration: Testing suites demonstrate that the backend can successfully retrieve state previously set by the Meltano system database, provided the search path and schema configurations match.

Sources: pyproject.toml:21-45, tests/test_backwards_compat.py:61-105, src/meltano_state_backend_postgresql/backend.py:314-328

Summary

The PostgreSQL state backend provides a robust, relational-database-backed storage solution for Meltano. It leverages standard PostgreSQL features like schemas for organization, JSON storage for flexible state data, and advisory locks for concurrency control, ensuring reliable state management in multi-process data integration environments.