Emit a state message even if no records were read

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

@@ -226,6 +226,10 @@ def _read_incremental(
         )
         logger.debug(f"Processing stream slices for {stream_name}", extra={"stream_slices": slices})
         total_records_counter = 0
+        if not slices:
+            # Safety net to ensure we always emit at least one state message even if there are no slices
+            checkpoint = self._checkpoint_state(stream_instance, stream_instance.state, connector_state)
+            yield checkpoint
         for _slice in slices:
             logger.debug("Processing stream slice", extra={"slice": _slice})
             records = stream_instance.read_records(

airbyte-cdk/python/unit_tests/sources/test_abstract_source.py

@@ -98,9 +98,13 @@ def primary_key(self) -> Optional[Union[str, List[str], List[List[str]]]]:
 class MockStreamWithState(MockStream):
     cursor_field = "cursor"
 
+    def __init__(self, inputs_and_mocked_outputs: List[Tuple[Mapping[str, Any], Iterable[Mapping[str, Any]]]], name: str, state=None):
+        super().__init__(inputs_and_mocked_outputs, name)
+        self._state = state
+
     @property
     def state(self):
-        return {}
+        return self._state
 
     @state.setter
     def state(self, value):
@@ -452,6 +456,74 @@ def test_with_slices(self, mocker):
 
         assert expected == messages
 
+    def test_no_slices(self, mocker):
+        """
+        Tests that an incremental read returns at least one state messages even if no records were read:
+            1. outputs a state message after reading the entire stream
+        """
+        slices = []
+        stream_output = [{"k1": "v1"}, {"k2": "v2"}, {"k3": "v3"}]
+        state = {"cursor": "value"}
+        s1 = MockStreamWithState(
+            [
+                (
+                    {
+                        "sync_mode": SyncMode.incremental,
+                        "stream_slice": s,
+                        "stream_state": mocker.ANY,
+                    },
+                    stream_output,
+                )
+                for s in slices
+            ],
+            name="s1",
+            state=state,
+        )
+        s2 = MockStreamWithState(
+            [
+                (
+                    {
+                        "sync_mode": SyncMode.incremental,
+                        "stream_slice": s,
+                        "stream_state": mocker.ANY,
+                    },
+                    stream_output,
+                )
+                for s in slices
+            ],
+            name="s2",
+            state=state,
+        )
+
+        mocker.patch.object(MockStreamWithState, "supports_incremental", return_value=True)
+        mocker.patch.object(MockStreamWithState, "get_json_schema", return_value={})
+        mocker.patch.object(MockStreamWithState, "stream_slices", return_value=slices)
+        mocker.patch.object(
+            MockStreamWithState,
+            "state_checkpoint_interval",
+            new_callable=mocker.PropertyMock,
+            return_value=2,
+        )
+
+        src = MockSource(streams=[s1, s2])
+        catalog = ConfiguredAirbyteCatalog(
+            streams=[
+                _configured_stream(s1, SyncMode.incremental),
+                _configured_stream(s2, SyncMode.incremental),
+            ]
+        )
+
+        expected = [
+            _state({"s1": state}),
+            _state({"s1": state, "s2": state}),
+        ]
+
+        messages = _fix_emitted_at(list(src.read(logger, {}, catalog, state=defaultdict(dict))))
+
+        print(f"expected:\n{expected}")
+        print(f"messages:\n{messages}")
+        assert expected == messages
+
     def test_with_slices_and_interval(self, mocker):
         """
         Tests that an incremental read which uses slices and a checkpoint interval: