Core Backend Implementation

Relevant source files

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

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

The meltano-state-backend-postgresql extension implements a custom state storage mechanism for Meltano using a PostgreSQL database. It allows Meltano to persist job state information (incremental markers, bookmarks, etc.) in a structured relational format instead of the default local filesystem or system database.

The implementation centers around the PostgreSQLStateStoreManager class, which extends Meltano's base StateStoreManager. It utilizes the psycopg (version 3) library for high-performance, asynchronous-capable database communication and provides full support for Meltano's state management lifecycle, including state retrieval, updates, locking, and pattern-based searching.

Sources: backend.py:127-133, README.md:1-10, pyproject.toml:28-32

Architecture and Class Hierarchy

The core architecture leverages inheritance from Meltano's core state store abstractions to ensure compatibility with the Meltano ecosystem.

The PostgreSQLStateStoreManager manages its own connection lifecycle and ensures that the necessary database schema exists upon initialization.
Sources: backend.py:127-142, pyproject.toml:44-45

Database Schema and Initialization

Upon instantiation, the manager automatically executes a CREATE TABLE IF NOT EXISTS statement to prepare the state storage table. By default, the table is named state.

State Table Structure

Field	Type	Description
state_id	VARCHAR(900)	Primary Key. The unique identifier for the Meltano job state.
partial_state	TEXT	JSON-encoded string containing incremental or partial state data.
completed_state	TEXT	JSON-encoded string containing state from completed runs.
updated_at	TIMESTAMP	Automatic timestamp set to CURRENT_TIMESTAMP on creation or update.

Sources: backend.py:214-224, README.md:38-40

Connection Management

The backend constructs a connection string (conninfo) by merging a base URI with explicit configuration overrides.

State Operations Logic

Setting and Retrieving State

The set operation uses an "upsert" (ON CONFLICT) strategy to ensure that existing states are updated rather than duplicated. JSON serialization is handled explicitly before sending data to the TEXT columns.

# From src/meltano_state_backend_postgresql/backend.py:236-249
query = SQL(
    """\
    INSERT INTO {table_identifier}
    (state_id, partial_state, completed_state, updated_at)
    VALUES (%s, %s, %s, CURRENT_TIMESTAMP)
    ON CONFLICT (state_id)
    DO UPDATE SET
        partial_state = EXCLUDED.partial_state,
        completed_state = EXCLUDED.completed_state,
        updated_at = CURRENT_TIMESTAMP
    """
).format(table_identifier=self.table_identifier)

When retrieving state via get(), the backend deserializes the JSON strings back into Python dictionaries. If a column is NULL in the database, it defaults to an empty dictionary {} to satisfy Meltano's internal requirements.
Sources: backend.py:231-250, backend.py:269-281, tests/test_state.py:75-92

State Locking Mechanism

To prevent race conditions during concurrent Meltano runs, the backend implements acquire_lock using PostgreSQL Advisory Locks.

1. Key Generation: The state_id string is hashed using MD5. The first 8 bytes are converted into a 64-bit signed integer.
2. Acquisition: The backend calls pg_try_advisory_lock(key). If unsuccessful, it retries for up to 30 seconds.
3. Release: The lock is released via pg_advisory_unlock(key) within a finally block of a context manager.

Sources: backend.py:355-407, tests/test_state.py:167-185

Configuration Settings

The backend exposes several settings through Meltano entry points. These can be configured via meltano.yml or environment variables.

Setting Name	Environment Variable	Description	Default
state_backend.uri	MELTANO_STATE_BACKEND_URI	Full Postgres URI	N/A
host	...POSTGRESQL_HOST	Server hostname	localhost
port	...POSTGRESQL_PORT	Server port	5432
user	...POSTGRESQL_USER	Auth username	N/A
password	...POSTGRESQL_PASSWORD	Auth password	N/A
schema	...POSTGRESQL_SCHEMA	Target schema	Server search path
table	...POSTGRESQL_TABLE	Target table name	state
sslmode	...POSTGRESQL_SSLMODE	SSL Mode (e.g., require)	prefer

Sources: backend.py:38-88, pyproject.toml:44-55, README.md:42-70

Integration and Compatibility

Backwards Compatibility

The implementation is designed to be compatible with existing Meltano system database schemas. It can point to the same schema used by the Meltano system database by configuring the search_path in the URI options (e.g., ?options=-csearch_path%3Dmy_schema).
Sources: README.md:28-32, tests/test_integration.py:46-68

Meltano Version Support

The backend requires meltano>=3.7. For Meltano versions prior to 4.2.0, the PostgreSQLStateStoreManager explicitly implements the __enter__ and __exit__ context manager methods to ensure proper connection closing, as this behavior was not natively handled in older versions of the core StateStoreManager.
Sources: pyproject.toml:28, backend.py:338-353

Summary

The Core Backend Implementation provides a robust, SQL-based state management layer for Meltano. By leveraging PostgreSQL's native JSON handling (via TEXT storage and application-side parsing) and advisory locks, it ensures data integrity and concurrency control across distributed Meltano environments.