chore: remove cursor.is_greater_than_or_equal (#674)

Author: maxi297

airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -429,17 +429,6 @@ def _send_log(self, level: Level, message: str) -> None:
                 )
             )
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        cursor_field = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
-        first_cursor_value = first.get(cursor_field)
-        second_cursor_value = second.get(cursor_field)
-        if first_cursor_value and second_cursor_value:
-            return self.parse_date(first_cursor_value) >= self.parse_date(second_cursor_value)
-        elif first_cursor_value:
-            return True
-        else:
-            return False
-
     def set_runtime_lookback_window(self, lookback_window_in_seconds: int) -> None:
         """
         Updates the lookback window based on a given number of seconds if the new duration

airbyte_cdk/sources/declarative/incremental/global_substream_cursor.py

@@ -338,12 +338,6 @@ def get_request_body_json(
     def should_be_synced(self, record: Record) -> bool:
         return self._stream_cursor.should_be_synced(self._convert_record_to_cursor_record(record))
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        return self._stream_cursor.is_greater_than_or_equal(
-            self._convert_record_to_cursor_record(first),
-            self._convert_record_to_cursor_record(second),
-        )
-
     @staticmethod
     def _convert_record_to_cursor_record(record: Record) -> Record:
         return Record(

airbyte_cdk/sources/declarative/incremental/per_partition_cursor.py

@@ -315,21 +315,6 @@ def should_be_synced(self, record: Record) -> bool:
             self._convert_record_to_cursor_record(record)
         )
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        if not first.associated_slice or not second.associated_slice:
-            raise ValueError(
-                f"Both records should have an associated slice but got {first.associated_slice} and {second.associated_slice}"
-            )
-        if first.associated_slice.partition != second.associated_slice.partition:
-            raise ValueError(
-                f"To compare records, partition should be the same but got {first.associated_slice.partition} and {second.associated_slice.partition}"
-            )
-
-        return self._get_cursor(first).is_greater_than_or_equal(
-            self._convert_record_to_cursor_record(first),
-            self._convert_record_to_cursor_record(second),
-        )
-
     @staticmethod
     def _convert_record_to_cursor_record(record: Record) -> Record:
         return Record(

airbyte_cdk/sources/declarative/incremental/per_partition_with_global.py

@@ -195,6 +195,3 @@ def get_request_body_json(
 
     def should_be_synced(self, record: Record) -> bool:
         return self._get_active_cursor().should_be_synced(record)
-
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        return self._global_cursor.is_greater_than_or_equal(first, second)

airbyte_cdk/sources/declarative/incremental/resumable_full_refresh_cursor.py

@@ -42,12 +42,6 @@ def should_be_synced(self, record: Record) -> bool:
         """
         return True
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        """
-        RFR record don't have ordering to be compared between one another.
-        """
-        return False
-
     def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
         # A top-level RFR cursor only manages the state of a single partition
         return self._cursor

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -41,7 +41,6 @@
 from airbyte_cdk.sources.declarative.requesters.requester import Requester
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
-from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.streams.core import StreamData
 from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState
@@ -528,35 +527,13 @@ def read_records(
                 if self.cursor and current_record:
                     self.cursor.observe(_slice, current_record)
 
-                # Latest record read, not necessarily within slice boundaries.
-                # TODO Remove once all custom components implement `observe` method.
-                # https://github.com/airbytehq/airbyte-internal-issues/issues/6955
-                most_recent_record_from_slice = self._get_most_recent_record(
-                    most_recent_record_from_slice, current_record, _slice
-                )
                 yield stream_data
 
             if self.cursor:
-                self.cursor.close_slice(_slice, most_recent_record_from_slice)
+                self.cursor.close_slice(_slice)
         return
 
-    def _get_most_recent_record(
-        self,
-        current_most_recent: Optional[Record],
-        current_record: Optional[Record],
-        stream_slice: StreamSlice,
-    ) -> Optional[Record]:
-        if self.cursor and current_record:
-            if not current_most_recent:
-                return current_record
-            else:
-                return (
-                    current_most_recent
-                    if self.cursor.is_greater_than_or_equal(current_most_recent, current_record)
-                    else current_record
-                )
-        else:
-            return None
+    # FIXME based on the comment above in SimpleRetriever.read_records, it seems like we can tackle https://github.com/airbytehq/airbyte-internal-issues/issues/6955 and remove this
 
     def _extract_record(
         self, stream_data: StreamData, stream_slice: StreamSlice

airbyte_cdk/sources/streams/checkpoint/cursor.py

@@ -62,12 +62,6 @@ def should_be_synced(self, record: Record) -> bool:
         Evaluating if a record should be synced allows for filtering and stop condition on pagination
         """
 
-    @abstractmethod
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        """
-        Evaluating which record is greater in terms of cursor. This is used to avoid having to capture all the records to close a slice
-        """
-
     @abstractmethod
     def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
         """

airbyte_cdk/sources/streams/checkpoint/resumable_full_refresh_cursor.py

@@ -40,12 +40,6 @@ def should_be_synced(self, record: Record) -> bool:
         """
         return True
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        """
-        RFR record don't have ordering to be compared between one another.
-        """
-        return False
-
     def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
         # A top-level RFR cursor only manages the state of a single partition
         return self._cursor

airbyte_cdk/sources/streams/checkpoint/substream_resumable_full_refresh_cursor.py

@@ -89,12 +89,6 @@ def should_be_synced(self, record: Record) -> bool:
         """
         return True
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        """
-        RFR record don't have ordering to be compared between one another.
-        """
-        return False
-
     def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
         if not stream_slice:
             raise ValueError("A partition needs to be provided in order to extract a state")

unit_tests/sources/declarative/incremental/test_datetime_based_cursor.py

@@ -1205,57 +1205,5 @@ def test_given_record_without_cursor_value_when_should_be_synced_then_return_tru
     assert cursor.should_be_synced(Record({"record without cursor value": "any"}, ANY_SLICE))
 
 
-def test_given_first_greater_than_second_then_return_true():
-    cursor = DatetimeBasedCursor(
-        start_datetime=MinMaxDatetime("3000-01-01", parameters={}),
-        cursor_field="cursor_field",
-        datetime_format="%Y-%m-%d",
-        config=config,
-        parameters={},
-    )
-    assert cursor.is_greater_than_or_equal(
-        Record({"cursor_field": "2023-01-01"}, {}), Record({"cursor_field": "2021-01-01"}, {})
-    )
-
-
-def test_given_first_lesser_than_second_then_return_false():
-    cursor = DatetimeBasedCursor(
-        start_datetime=MinMaxDatetime("3000-01-01", parameters={}),
-        cursor_field="cursor_field",
-        datetime_format="%Y-%m-%d",
-        config=config,
-        parameters={},
-    )
-    assert not cursor.is_greater_than_or_equal(
-        Record({"cursor_field": "2021-01-01"}, {}), Record({"cursor_field": "2023-01-01"}, {})
-    )
-
-
-def test_given_no_cursor_value_for_second_than_second_then_return_true():
-    cursor = DatetimeBasedCursor(
-        start_datetime=MinMaxDatetime("3000-01-01", parameters={}),
-        cursor_field="cursor_field",
-        datetime_format="%Y-%m-%d",
-        config=config,
-        parameters={},
-    )
-    assert cursor.is_greater_than_or_equal(
-        Record({"cursor_field": "2021-01-01"}, {}), Record({}, {})
-    )
-
-
-def test_given_no_cursor_value_for_first_than_second_then_return_false():
-    cursor = DatetimeBasedCursor(
-        start_datetime=MinMaxDatetime("3000-01-01", parameters={}),
-        cursor_field="cursor_field",
-        datetime_format="%Y-%m-%d",
-        config=config,
-        parameters={},
-    )
-    assert not cursor.is_greater_than_or_equal(
-        Record({}, {}), Record({"cursor_field": "2021-01-01"}, {})
-    )
-
-
 if __name__ == "__main__":
     unittest.main()