fix(low-code): close ConcurrentCursor end-window and clamping gaps

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1651,6 +1651,7 @@ def create_concurrent_cursor_from_datetime_based_cursor(
             slice_range=step_length,
             cursor_granularity=cursor_granularity,
             clamping_strategy=clamping_strategy,
+            is_compare_strictly=datetime_based_cursor_model.is_compare_strictly or False,
         )
 
     def create_concurrent_cursor_from_incrementing_count_cursor(

airbyte_cdk/sources/streams/concurrent/clamping.py

@@ -37,7 +37,7 @@ def __init__(self, is_ceiling: bool = True) -> None:
 
     def clamp(self, value: datetime) -> datetime:  # type: ignore  # datetime implements method from CursorValueType
         return_value = value.replace(hour=0, minute=0, second=0, microsecond=0)
-        if self._is_ceiling:
+        if self._is_ceiling and return_value != value:
             return return_value + timedelta(days=1)
         return return_value
 
@@ -93,7 +93,7 @@ def clamp(self, value: datetime) -> datetime:  # type: ignore  # datetime implem
         )
         delta = (
             timedelta(days_diff_to_ceiling)
-            if self._is_ceiling
+            if self._is_ceiling or days_diff_to_ceiling == 0
             else timedelta(days_diff_to_ceiling - 7)
         )
         return value.replace(hour=0, minute=0, second=0, microsecond=0) + delta

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -195,6 +195,7 @@ def copy_without_state(self) -> "ConcurrentCursor":
             slice_range=self._slice_range,
             cursor_granularity=self._cursor_granularity,
             clamping_strategy=self._clamping_strategy,
+            is_compare_strictly=self._is_compare_strictly,
         )
 
     def __init__(
@@ -213,6 +214,7 @@ def __init__(
         slice_range: Optional[GapType] = None,
         cursor_granularity: Optional[GapType] = None,
         clamping_strategy: ClampingStrategy = NoClamping(),
+        is_compare_strictly: bool = False,
     ) -> None:
         self._stream_name = stream_name
         self._stream_namespace = stream_namespace
@@ -235,6 +237,7 @@ def __init__(
         # Flag to track if the logger has been triggered (per stream)
         self._should_be_synced_logger_triggered = False
         self._clamping_strategy = clamping_strategy
+        self._is_compare_strictly = is_compare_strictly
         self._is_ascending_order = True
 
         # A lock is required when closing a partition because updating the cursor's concurrent_state is
@@ -483,20 +486,30 @@ def _calculate_lower_boundary_of_last_slice(
     def _split_per_slice_range(
         self, lower: CursorValueType, upper: CursorValueType, upper_is_end: bool
     ) -> Iterable[StreamSlice]:
-        if lower >= upper:
+        if lower > upper or (lower == upper and not self._cursor_granularity):
             return
 
         if self._start and upper < self._start:
             return
 
         lower = max(lower, self._start) if self._start else lower
+        # Clamp first, else a step shorter than the clamping period collapses the first window
+        lower = self._clamping_strategy.clamp(lower)
+        # With a granularity the last window includes upper, so lower == upper is one unit to read,
+        # unless the API rejects start == end. In a gap, the next slice already reads upper.
+        if lower > upper or (lower == upper and (self._is_compare_strictly or not upper_is_end)):
+            return
         if not self._slice_range or self._evaluate_upper_safely(lower, self._slice_range) >= upper:
-            clamped_lower = self._clamping_strategy.clamp(lower)
-            clamped_upper = self._clamping_strategy.clamp(upper)
+            # With a granularity upper is the last unit to read: clamping it would read past the end
+            clamped_upper = (
+                upper
+                if upper_is_end and self._cursor_granularity
+                else self._clamping_strategy.clamp(upper)
+            )
             start_value, end_value = (
-                (clamped_lower, clamped_upper - self._cursor_granularity)
+                (lower, clamped_upper - self._cursor_granularity)
                 if self._cursor_granularity and not upper_is_end
-                else (clamped_lower, clamped_upper)
+                else (lower, clamped_upper)
             )
             yield StreamSlice(
                 partition={},
@@ -516,13 +529,15 @@ def _split_per_slice_range(
                 current_upper_boundary = min(
                     self._evaluate_upper_safely(current_lower_boundary, self._slice_range), upper
                 )
-                has_reached_upper_boundary = current_upper_boundary >= upper
-
                 clamped_upper = (
                     self._clamping_strategy.clamp(current_upper_boundary)
                     if current_upper_boundary != upper
                     else current_upper_boundary
                 )
+                # A boundary clamped onto upper makes this the last window, else upper is dropped
+                has_reached_upper_boundary = (
+                    current_upper_boundary >= upper or clamped_upper == upper
+                )
                 clamped_lower = self._clamping_strategy.clamp(current_lower_boundary)
                 if clamped_lower >= clamped_upper:
                     # clamping collapsed both values which means that it is time to stop processing
@@ -546,7 +561,7 @@ def _split_per_slice_range(
                     },
                 )
                 current_lower_boundary = clamped_upper
-                if current_upper_boundary >= upper:
+                if has_reached_upper_boundary:
                     stop_processing = True
 
     def _evaluate_upper_safely(self, lower: CursorValueType, step: GapType) -> CursorValueType:

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -5403,6 +5403,41 @@ def test_incrementing_count_cursor_with_partition_router_raises_error():
         )
 
 
+@pytest.mark.parametrize(
+    "is_compare_strictly, expected_slices",
+    [
+        pytest.param(
+            False, [{"start_time": "2024-09-01", "end_time": "2024-09-01"}], id="reads_the_one_day"
+        ),
+        pytest.param(True, [], id="skipped_when_compare_strictly"),
+    ],
+)
+def test_create_concurrent_cursor_from_datetime_based_cursor_given_start_equals_end(
+    is_compare_strictly, expected_slices
+):
+    concurrent_cursor = (
+        ModelToComponentFactory().create_concurrent_cursor_from_datetime_based_cursor(
+            model_type=DatetimeBasedCursorModel,
+            component_definition={
+                "type": "DatetimeBasedCursor",
+                "cursor_field": "updated_at",
+                "datetime_format": "%Y-%m-%d",
+                "start_datetime": "2024-09-01",
+                "end_datetime": "2024-09-01",
+                "step": "P30D",
+                "cursor_granularity": "P1D",
+                "is_compare_strictly": is_compare_strictly,
+            },
+            stream_name="test",
+            stream_namespace=None,
+            stream_state={},
+            config={},
+        )
+    )
+
+    assert list(concurrent_cursor.stream_slices()) == expected_slices
+
+
 def test_create_concurrent_cursor_uses_min_max_datetime_format_if_defined():
     """
     Validates a special case for when the start_time.datetime_format and end_time.datetime_format are defined, the date to

unit_tests/sources/streams/concurrent/test_clamping.py

@@ -24,9 +24,14 @@ def test_when_clamp_then_remove_every_unit_smaller_than_days(self) -> None:
         assert result.microsecond == 0
 
     def test_given_last_day_of_month_when_clamp_then_result_is_next_month(self) -> None:
-        result = self._strategy.clamp(datetime(2024, 1, 31))
+        result = self._strategy.clamp(datetime(2024, 1, 31, 20, 23, 3, 2039))
         assert result == datetime(2024, 2, 1)
 
+    def test_given_midnight_when_clamp_then_return_same_date(self) -> None:
+        midnight = datetime(2024, 1, 1)
+        result = self._strategy.clamp(midnight)
+        assert result == midnight
+
     def test_given_is_not_ceiling_when_clamp_then_just_remove_unit_smaller_than_days(self) -> None:
         strategy = DayClampingStrategy(is_ceiling=False)
         result = strategy.clamp(datetime(2024, 1, 1, 20, 23, 3, 1029))
@@ -101,6 +106,11 @@ def test_given_not_weekday_after_target_when_clamp_then_return_next_occurrence_o
             _DATETIME_ON_WEDNESDAY.day + 2,
         )
 
+    def test_given_not_ceiling_and_same_weekday_when_clamp_then_return_same_date(self) -> None:
+        strategy = WeekClampingStrategy(Weekday.TUESDAY, is_ceiling=False)
+        result = strategy.clamp(_DATETIME_ON_TUESDAY)
+        assert result == _DATETIME_ON_TUESDAY
+
     def test_given_not_ceiling_when_clamp_then_round_down(self) -> None:
         strategy = WeekClampingStrategy(Weekday.FRIDAY, is_ceiling=False)
         result = strategy.clamp(_DATETIME_ON_WEDNESDAY)