State & Bookmark Management
Relevant source files
The following files were used as context for generating this wiki page:
State and Bookmark Management in the flext-tap-oracle-wms project ensures that data extraction from the Oracle Warehouse Management System (WMS) is both efficient and reliable. By maintaining a persistent record of the last successfully processed data point (a "bookmark"), the system can perform incremental replication, avoiding the need to re-extract entire datasets during subsequent syncs.
The system complies with the Singer protocol, utilizing STATE messages to communicate progress between the Tap and the Target. This architecture supports high-volume data environments where full table refreshes are impractical.
Sources: src/flext_tap_oracle_wms/README.md:92-95, meltano.yml:204-209, tests/e2e/README.md:120-130
Replication Methods
The Tap supports two primary replication methods defined within the stream metadata. These methods dictate how the system utilizes state to filter outgoing requests to the Oracle WMS API.
| Method | Description | Bookmark Dependency |
|---|---|---|
FULL_TABLE | Extracts all records from the source entity regardless of previous syncs. | None |
INCREMENTAL | Extracts only records created or updated after the value stored in the state bookmark. | High (Requires replication-key) |
Sources: meltano.yml:210-229, tests/integration/test_functional.py:214-232
Incremental Logic Flow
When a stream is configured for incremental replication, it relies on a specific field—typically a timestamp—designated as the replication-key. The following diagram illustrates how state is used to filter data during an operation.
The system tracks the highest value of the replication key encountered during a sync to update the bookmark for the next session.
Sources: meltano.yml:210-229, tests/integration/test_functional.py:246-260, tests/e2e/README.md:120-130
Implementation Details
State in Catalog Metadata
State management is driven by the Singer catalog metadata. In tap.py, the discovercatalog_typed method initializes default metadata for discovered entities, setting the forced-replication-method and inclusion status.
# src/flext_tap_oracle_wms/tap.py:270-280
streams.append(
entry_result.value.model_copy(
update={
"metadata": [
m.Meltano.SingerCatalogMetadata(
breadcrumb=[],
metadata={
"inclusion": "available",
"forced-replication-method": "FULL_TABLE",
"table-key-properties": ["id"],
},
),
]
}
)
)
Sources: src/flext_tap_oracle_wms/tap.py:270-280
Filtering via Operation Kwargs
During data extraction, the stream class uses the _build_operation_kwargs method to transform the current state into API request parameters. If a replication_key_value is present in the context (derived from state), it is converted into a filter, such as a timestamp comparison.
# tests/integration/test_functional.py:254-256
context = {"replication_key_value": "2024-01-01T00:00:00Z"}
url_params = stream._build_operation_kwargs(page=1, context=context)
kwargs_filter = url_params.get("filter")
Sources: tests/integration/test_functional.py:254-259
Configuration and Scheduling
The meltano.yml file defines how state and bookmarks are utilized in a production environment. It maps specific replication keys to entities and sets a global start_date for initial syncs when no bookmark is available.
Entity Configuration Table
| Entity | Replication Method | Replication Key |
|---|---|---|
inventory | INCREMENTAL | last_updated_ts |
order_hdr | INCREMENTAL | last_updated_ts |
order_dtl | INCREMENTAL | last_updated_ts |
item | FULL_TABLE | N/A |
location | FULL_TABLE | N/A |
Sources: meltano.yml:210-229
Sequential Extraction Sequence
The following sequence diagram represents the interaction between the Tap and the state management process during a multi-sync lifecycle.
Summary
State & Bookmark Management provides the foundation for incremental data synchronization in the flext-tap-oracle-wms project. By leveraging the Singer protocol's state messages and specific WMS replication keys (like last_updated_ts), the system ensures data consistency while optimizing network and compute resources.
Sources: src/flext_tap_oracle_wms/README.md:92-95, meltano.yml:210-229