diff --git a/docs/faq.md b/docs/faq.md index 63dcb675e..cded23bbc 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -26,3 +26,12 @@ class MyStream(Stream): super().__init__(tap) self.conn... ``` + +## I'm seeing `Note: Progress is not resumable if interrupted.` in my state files + +If the stream attribute [`is_sorted`](singer_sdk.Stream.is_sorted) (default: `False`) is not set to `True`, the records are assumed not to be monotonically increasing, i.e. there's no guarantee that newer records always come later and the tap has to run to completion so the state can safely reflect the largest replication value seen. If you know that the records are monotonically increasing, you can set `is_sorted` to `True` and the sync will be resumable if it's interrupted at any point and the state file will reflect this. + +```python +class MyStream(Stream): + is_sorted = True +``` diff --git a/singer_sdk/helpers/_state.py b/singer_sdk/helpers/_state.py index f20f40703..e265be97b 100644 --- a/singer_sdk/helpers/_state.py +++ b/singer_sdk/helpers/_state.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import typing as t from singer_sdk.exceptions import InvalidStreamSortException @@ -17,6 +18,8 @@ SIGNPOST_MARKER = "replication_key_signpost" STARTING_MARKER = "starting_replication_value" +logger = logging.getLogger("singer_sdk") + def get_state_if_exists( tap_state: dict, @@ -207,6 +210,11 @@ def increment_state( stream_or_partition_state[PROGRESS_MARKERS] = { PROGRESS_MARKER_NOTE: "Progress is not resumable if interrupted.", } + logger.warning( + "Stream is assumed to be unsorted, progress is not resumable if " + "interrupted", + extra={"replication_key": replication_key}, + ) progress_dict = stream_or_partition_state[PROGRESS_MARKERS] old_rk_value = to_json_compatible(progress_dict.get("replication_key_value")) new_rk_value = to_json_compatible(latest_record[replication_key]) diff --git a/tests/core/test_state_handling.py b/tests/core/test_state_handling.py index 339df1f4c..3f4fff148 100644 --- a/tests/core/test_state_handling.py +++ b/tests/core/test_state_handling.py @@ -103,3 +103,27 @@ def test_state_finalize(dirty_state, finalized_state): for partition_state in stream_state.get("partitions", []): _state.finalize_state_progress_markers(partition_state) assert state == finalized_state + + +def test_irresumable_state(): + stream_state = {} + latest_record = {"updated_at": "2021-05-17T20:41:16Z"} + replication_key = "updated_at" + is_sorted = False + check_sorted = False + + _state.increment_state( + stream_state, + latest_record=latest_record, + replication_key=replication_key, + is_sorted=is_sorted, + check_sorted=check_sorted, + ) + + assert stream_state == { + "progress_markers": { + "Note": "Progress is not resumable if interrupted.", + "replication_key": "updated_at", + "replication_key_value": "2021-05-17T20:41:16Z", + }, + }