State Management & Incremental Syncs
Relevant source files
The following files were used as context for generating this wiki page:
State Management is a critical component of the Meltano Singer SDK that allows taps to track progress and perform incremental synchronizations. By maintaining "bookmarks," the SDK ensures that only new or updated data is extracted from source systems, significantly improving performance and reducing resource consumption. This system handles everything from simple replication keys to complex partitioned state for hierarchical data structures.
Sources: singer_sdk/streams/core.py, cookiecutter/tap-template/{{cookiecutter.tap_id}}/AGENTS.md
Core Architecture
The state management system is centered around the StreamStateManager and the Stream class. The StreamStateManager encapsulates logic for partition-level state handling, replication key tracking, and progress markers. This allows the state logic to be decoupled from the actual data extraction process.
Components of State Management
| Component | Description |
|---|---|
| Tap State | A shared, writable dictionary representing the state for the entire tap plugin. |
| Stream State | A specific section of the tap state dedicated to an individual stream. |
| Replication Key | The field (e.g., updated_at or id) used to filter source data for incremental syncs. |
| Bookmark Value | The stored value of the replication key from the last successful sync. |
| Signpost | A "high-water mark" (usually current time) used to prevent data loss from late-arriving records. |
| Progress Markers | Temporary tags in state that indicate a sync is in progress and not yet resumable if interrupted. |
Sources: singer_sdk/streams/_state.py:23-53, singer_sdk/streams/core.py:465-515, cookiecutter/tap-template/{{cookiecutter.tap_id}}/AGENTS.md
State Data Flow
The following diagram illustrates how state is initialized, updated during record processing, and finally emitted back to the Singer ecosystem.
Incremental Sync Logic
Incremental synchronization relies on the replication_key and replication_method. When a stream is configured for incremental replication, the SDK uses the stored state to determine where to begin the next sync.
Replication Key Management
The SDK supports both timestamp-based and non-timestamp replication keys. For timestamps, it automatically handles ISO 8601 parsing and timezone awareness.
- Starting Value:
get_starting_replication_key_value()retrieves the bookmark from state. - Starting Timestamp:
get_starting_timestamp()is used specifically for date/datetime keys, incorporatingstart_datefrom configuration if no state exists. - Signposts:
get_replication_key_signpost()provides a maximum allowable bookmark value (usuallyutc_now()) to bound the sync.
Sources: singer_sdk/streams/core.py:255-325, singer_sdk/streams/core.py:420-445
State Partitioning
For complex sources, state can be "partitioned." This means a single stream can have multiple bookmarks based on unique context keys (e.g., one bookmark per project_id).
Resumable vs. Non-Resumable State
The SDK implements a safety mechanism using PROGRESS_MARKERS. While a stream is actively syncing, it may be in a "non-resumable" state.
- In-Progress: When records are being emitted, the state is marked as needing a flush.
- Abortion Handling: If a sync is aborted (e.g., via
MaxRecordsLimitException), the SDK checks if the state is resumable.Full Tablesyncs or streams with active progress markers that haven't reached a checkpoint are considered non-resumable. - Finalization: Once a stream finishes,
finalize_state_progress_markers()clears these markers, making the state safe for the next run.
Sources: singer_sdk/streams/core.py:698-732, singer_sdk/streams/_state.py:230-245
Implementation Example
Developers enable incremental sync by setting the replication_key and defining how records are filtered in get_records().
class MyStream(RESTStream):
name = "orders"
primary_keys = ["id"]
replication_key = "updated_at" # Enables incremental sync
def get_records(self, context):
# Retrieve the last saved timestamp or config start_date
start_date = self.get_starting_timestamp(context)
# Use start_date in API request parameters
params = {"updated_since": start_date.isoformat()}
# ... yield records
Sources: cookiecutter/tap-template/{{cookiecutter.tap_id}}/AGENTS.md, singer_sdk/streams/core.py:279-310
Target State Handling
Singer targets receive state messages from taps but do not typically modify them. Their role is to forward state messages to indicate that data up to a certain point has been successfully persisted.
- Processing: Targets call
process_state_message()when receiving a state message. - Persistence: Targets should emit the state message back to stdout only after the associated data batches have been successfully loaded into the destination.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Summary
The state management system in the Meltano Singer SDK provides a robust framework for efficient data extraction. By automating the tracking of bookmarks, handling partitions, and managing progress markers, it ensures that taps can safely resume interrupted syncs and only process new data. This architecture supports both simple API integrations and complex, hierarchical data structures through its partitioned state logic.