Meltano Plugin Integration

Relevant source files

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

• src/meltano_state_backend_postgresql/__init__.py
• pyproject.toml
• src/meltano_state_backend_postgresql/backend.py
• README.md
• tests/test_meltano.py
• tests/test_integration.py

The meltano-state-backend-postgresql package is a Meltano extension designed to provide a dedicated PostgreSQL-based state backend. Its primary purpose is to allow Meltano to store, retrieve, and manage incremental replication state data within a PostgreSQL database rather than using the default system database or local filesystem.

This integration functions as a plugin by registering itself through Python entry points, allowing Meltano to discover the PostgreSQLStateStoreManager class. It leverages the psycopg library to interact with the database and supports advanced configuration including SSL modes, custom schemas, and advisory locking for state concurrency.
Sources: README.md:1-12, pyproject.toml:28-40, src/meltano_state_backend_postgresql/backend.py:126-130

Plugin Registration and Discovery

The integration is registered via pyproject.toml using standard Python entry points. This registration allows Meltano to identify the settings available for the plugin and the specific class responsible for managing state operations.

Entry Points

The plugin defines two main types of entry points:

1. meltano.settings: Maps configuration keys to SettingDefinition objects in the backend module.
2. meltano.state_backends: Maps the postgresql identifier to the PostgreSQLStateStoreManager class.

Sources: pyproject.toml:28-40

Configuration Schema

The plugin exposes several configuration options through SettingDefinition objects. These settings can be configured in meltano.yml or via environment variables.

Setting Name	Kind	Description
state_backend.postgresql.host	STRING	PostgreSQL server hostname
state_backend.postgresql.port	INTEGER	PostgreSQL server port
state_backend.postgresql.database	STRING	PostgreSQL database name
state_backend.postgresql.user	STRING	PostgreSQL username
state_backend.postgresql.password	STRING	PostgreSQL password (Sensitive)
state_backend.postgresql.schema	STRING	PostgreSQL schema name
state_backend.postgresql.sslmode	STRING	PostgreSQL SSL mode (e.g., prefer, require)
state_backend.postgresql.table	STRING	Table name for state storage (default: state)

Sources: src/meltano_state_backend_postgresql/backend.py:44-105, tests/test_meltano.py:35-71

Core Architecture

The architecture centers on the PostgreSQLStateStoreManager, which inherits from Meltano's base StateStoreManager. It manages the lifecycle of the database connection and provides the implementation for state persistence.

Data Flow for State Management

When Meltano performs a state operation (Set, Get, or Delete), the flow moves through the PostgreSQLStateStoreManager to the database using psycopg.

Connection Logic

The plugin constructs a connection string (conninfo) by merging a base URI with individual configuration overrides. It requires user, password, and dbname to be present either in the URI or as explicit settings.
Sources: src/meltano_state_backend_postgresql/backend.py:108-144

Database Schema and Persistence

The plugin automatically manages its own storage structure. Upon initialization, it ensures the existence of the state table.

State Table Structure

The default table name is state, though this can be overridden. The schema is defined as follows:

Column	Type	Description
state_id	VARCHAR(900)	Primary Key; unique identifier for the job state
partial_state	TEXT	JSON string containing incomplete/incremental state
completed_state	TEXT	JSON string containing confirmed/flushed state
updated_at	TIMESTAMP	Last update time (Defaults to CURRENT_TIMESTAMP)

Sources: src/meltano_state_backend_postgresql/backend.py:212-223

State Operations Logic

• Set: Uses an INSERT ... ON CONFLICT (state_id) DO UPDATE pattern to ensure atomicity. Partial and completed states are serialized to JSON strings before insertion.
• Get: Retrieves state data and deserializes the JSON strings back into Python dictionaries for the MeltanoState object.
• List: Supports glob-style patterns (e.g., job_*) by converting them to SQL LIKE patterns (job_%).

Sources: src/meltano_state_backend_postgresql/backend.py:225-325, tests/test_integration.py:146-161

Concurrency and Locking

To prevent race conditions when multiple Meltano instances attempt to update the same state ID, the plugin implements a locking mechanism using PostgreSQL Advisory Locks.

Locking Flow

The plugin converts the state_id into a deterministic 64-bit integer using an MD5 hash to use as a lock key.

Integration Lifecycle Management

The plugin supports cleanup and resource management, particularly for different versions of Meltano.

• Context Management: For Meltano versions < 4.2.0, the backend provides its own __enter__ and __exit__ methods to ensure the database connection is closed after use.
• Cleanup: The clear_all() method allows for truncating the entire state table, returning a count of the records removed.
• Backwards Compatibility: The plugin is designed to work alongside existing Meltano system databases by allowing specific schema targeting, ensuring it can read state even if migrated from a different schema.

Sources: src/meltano_state_backend_postgresql/backend.py:277-293, src/meltano_state_backend_postgresql/backend.py:334-346, tests/test_backwards_compat.py:58-100

Conclusion

The Meltano Plugin Integration for PostgreSQL provides a robust, production-ready state storage solution. By utilizing PostgreSQL's native features like ON CONFLICT updates and advisory locks, it ensures data integrity and high concurrency support. The integration is seamless through Meltano's extension points, requiring only standard connection parameters to move state management into a managed relational database environment.