Merge branch 'main' into 2045-validate-records-with-fastjsonschema

codecov.yml

@@ -0,0 +1,15 @@
+coverage:
+  precision: 2
+  range: "70...100"
+  round: down
+  status:
+    project:
+      default:
+        target: auto
+        threshold: 1%
+        base: auto
+    patch:
+      default:
+        target: 100%
+        threshold: 1%
+        base: auto

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
+```

pyproject.toml

@@ -45,7 +45,7 @@ click = "~=8.0"
 cryptography = ">=3.4.6"
 fastjsonschema = ">=2.19.0"
 fs = ">=2.4.16"
-importlib-metadata = {version = "<7.0.0", python = "<3.12"}
+importlib-metadata = {version = "<8.0.0", python = "<3.12"}
 importlib-resources = {version = ">=5.12.0", python = "<3.9"}
 inflection = ">=0.5.1"
 joblib = ">=1.0.1"

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])