refactor: rename num_partitions_started to num_partitions_in_progress (computed)

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -188,7 +188,6 @@ def __init__(
         # Partitioned stream status tracking for progress estimation.
         # These counters are per-sync only and intentionally NOT restored from persisted state
         # (_set_initial_state does not read them back). On resume, they reset to 0.
-        self._num_partitions_started: int = 0
         self._num_partitions_completed: int = 0
         self._is_partition_discovery_complete: bool = False
 
@@ -225,7 +224,8 @@ def state(self) -> MutableMapping[str, Any]:
         if self._parent_state is not None:
             state["parent_state"] = self._parent_state
         state["partitioned_stream_status"] = {
-            "num_partitions_started": self._num_partitions_started,
+            "num_partitions_in_progress": self._generated_partitions_count
+            - self._num_partitions_completed,
             "num_partitions_completed": self._num_partitions_completed,
             "num_partitions_expected": self._generated_partitions_count,
             "is_partition_discovery_complete": self._is_partition_discovery_complete,
@@ -367,7 +367,6 @@ def _generate_slices_from_partition(
         with self._lock:
             seq = self._generated_partitions_count
             self._generated_partitions_count += 1
-            self._num_partitions_started += 1
             self._processing_partitions_indexes.append(seq)
             self._partition_key_to_index[partition_key] = seq
 

unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py

@@ -400,17 +400,14 @@ def run_mocked_test(
         assert partitioned_status is not None, (
             "partitioned_stream_status must always be present in state"
         )
-        assert "num_partitions_started" in partitioned_status
+        assert "num_partitions_in_progress" in partitioned_status
         assert "num_partitions_completed" in partitioned_status
         assert "num_partitions_expected" in partitioned_status
         assert "is_partition_discovery_complete" in partitioned_status
-        assert (
-            partitioned_status["num_partitions_started"]
-            >= partitioned_status["num_partitions_completed"]
-        )
+        assert partitioned_status["num_partitions_in_progress"] >= 0
         assert (
             partitioned_status["num_partitions_expected"]
-            >= partitioned_status["num_partitions_started"]
+            >= partitioned_status["num_partitions_completed"]
         )
         _strip_partitioned_stream_status(final_state_dict)
         assert final_state_dict == expected_state
@@ -3694,7 +3691,7 @@ def test_given_no_partitions_processed_when_close_partition_then_no_state_update
     state = cursor.state
     partitioned_status = state.pop("partitioned_stream_status", None)
     assert partitioned_status is not None
-    assert partitioned_status["num_partitions_started"] == 0
+    assert partitioned_status["num_partitions_in_progress"] == 0
     assert partitioned_status["num_partitions_completed"] == 0
     assert partitioned_status["num_partitions_expected"] == 0
     assert partitioned_status["is_partition_discovery_complete"] is True
@@ -3788,7 +3785,7 @@ def test_given_unfinished_first_parent_partition_no_parent_state_update():
     state = cursor.state
     partitioned_status = state.pop("partitioned_stream_status", None)
     assert partitioned_status is not None
-    assert partitioned_status["num_partitions_started"] == 2
+    assert partitioned_status["num_partitions_in_progress"] == 1
     assert partitioned_status["num_partitions_completed"] == 1
     assert partitioned_status["num_partitions_expected"] == 2
     assert partitioned_status["is_partition_discovery_complete"] is True
@@ -3890,7 +3887,7 @@ def test_given_unfinished_last_parent_partition_with_partial_parent_state_update
     state = cursor.state
     partitioned_status = state.pop("partitioned_stream_status", None)
     assert partitioned_status is not None
-    assert partitioned_status["num_partitions_started"] == 2
+    assert partitioned_status["num_partitions_in_progress"] == 1
     assert partitioned_status["num_partitions_completed"] == 1
     assert partitioned_status["num_partitions_expected"] == 2
     assert partitioned_status["is_partition_discovery_complete"] is True

unit_tests/sources/declarative/incremental/test_per_partition_cursor_integration.py

@@ -344,7 +344,7 @@ def test_given_record_for_partition_when_read_then_update_state(caplog):
             },
         ],
         "partitioned_stream_status": {
-            "num_partitions_started": 2,
+            "num_partitions_in_progress": 0,
             "num_partitions_completed": 2,
             "num_partitions_expected": 2,
             "is_partition_discovery_complete": True,
@@ -588,7 +588,7 @@ def test_perpartition_with_fallback(caplog):
         "state": {"cursor_field": "2022-02-19"},
         "lookback_window": 1,
         "partitioned_stream_status": {
-            "num_partitions_started": 6,
+            "num_partitions_in_progress": 0,
             "num_partitions_completed": 6,
             "num_partitions_expected": 6,
             "is_partition_discovery_complete": True,
@@ -776,7 +776,7 @@ def test_per_partition_cursor_within_limit(caplog):
             },
         ],
         "partitioned_stream_status": {
-            "num_partitions_started": 3,
+            "num_partitions_in_progress": 0,
             "num_partitions_completed": 3,
             "num_partitions_expected": 3,
             "is_partition_discovery_complete": True,