19class AvailabilityStrategy(ABC):
20    """
21    Abstract base class for checking stream availability.
22    """
23
24    @abstractmethod
25    def check_availability(
26        self, stream: Stream, logger: logging.Logger, source: Optional["Source"] = None
27    ) -> Tuple[bool, Optional[str]]:
28        """
29        Checks stream availability.
30
31        :param stream: stream
32        :param logger: source logger
33        :param source: (optional) source
34        :return: A tuple of (boolean, str). If boolean is true, then the stream
35          is available, and no str is required. Otherwise, the stream is unavailable
36          for some reason and the str should describe what went wrong and how to
37          resolve the unavailability, if possible.
38        """
39
40    @staticmethod
41    def get_first_stream_slice(stream: Stream) -> Optional[Mapping[str, Any]]:
42        """
43        Gets the first stream_slice from a given stream's stream_slices.
44        :param stream: stream
45        :raises StopIteration: if there is no first slice to return (the stream_slices generator is empty)
46        :return: first stream slice from 'stream_slices' generator (`None` is a valid stream slice)
47        """
48        # We wrap the return output of stream_slices() because some implementations return types that are iterable,
49        # but not iterators such as lists or tuples
50        slices = iter(
51            stream.stream_slices(
52                cursor_field=stream.cursor_field,  # type: ignore[arg-type]
53                sync_mode=SyncMode.full_refresh,
54            )
55        )
56        return next(slices)
57
58    @staticmethod
59    def get_first_record_for_slice(
60        stream: Stream, stream_slice: Optional[Mapping[str, Any]]
61    ) -> StreamData:
62        """
63        Gets the first record for a stream_slice of a stream.
64
65        :param stream: stream instance from which to read records
66        :param stream_slice: stream_slice parameters for slicing the stream
67        :raises StopIteration: if there is no first record to return (the read_records generator is empty)
68        :return: StreamData containing the first record in the slice
69        """
70        # Store the original value of exit_on_rate_limit
71        original_exit_on_rate_limit = stream.exit_on_rate_limit
72
73        try:
74            # Ensure exit_on_rate_limit is safely set to True if possible
75            stream.exit_on_rate_limit = True
76
77            # We wrap the return output of read_records() because some implementations return types that are iterable,
78            # but not iterators such as lists or tuples
79            records_for_slice = iter(
80                stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream_slice)
81            )
82
83            return next(records_for_slice)
84        finally:
85            # Restore the original exit_on_rate_limit value
86            stream.exit_on_rate_limit = original_exit_on_rate_limit

18class HttpAvailabilityStrategy(AvailabilityStrategy):
19    def check_availability(
20        self, stream: Stream, logger: logging.Logger, source: Optional["Source"] = None
21    ) -> Tuple[bool, Optional[str]]:
22        """
23        Check stream availability by attempting to read the first record of the
24        stream.
25
26        :param stream: stream
27        :param logger: source logger
28        :param source: (optional) source
29        :return: A tuple of (boolean, str). If boolean is true, then the stream
30          is available, and no str is required. Otherwise, the stream is unavailable
31          for some reason and the str should describe what went wrong and how to
32          resolve the unavailability, if possible.
33        """
34        reason: Optional[str]
35        try:
36            # Some streams need a stream slice to read records (e.g. if they have a SubstreamPartitionRouter)
37            # Streams that don't need a stream slice will return `None` as their first stream slice.
38            stream_slice = self.get_first_stream_slice(stream)
39        except StopIteration:
40            # If stream_slices has no `next()` item (Note - this is different from stream_slices returning [None]!)
41            # This can happen when a substream's `stream_slices` method does a `for record in parent_records: yield <something>`
42            # without accounting for the case in which the parent stream is empty.
43            reason = f"Cannot attempt to connect to stream {stream.name} - no stream slices were found, likely because the parent stream is empty."
44            return False, reason
45        except AirbyteTracedException as error:
46            return False, error.message
47
48        try:
49            self.get_first_record_for_slice(stream, stream_slice)
50            return True, None
51        except StopIteration:
52            logger.info(f"Successfully connected to stream {stream.name}, but got 0 records.")
53            return True, None
54        except AirbyteTracedException as error:
55            return False, error.message

11@dataclass
12class ResumableFullRefreshCursor(Cursor):
13    """
14    Cursor that allows for the checkpointing of sync progress according to a synthetic cursor based on the pagination state
15    of the stream. Resumable full refresh syncs are only intended to retain state in between sync attempts of the same job
16    with the platform responsible for removing said state.
17    """
18
19    def __init__(self) -> None:
20        self._cursor: StreamState = {}
21
22    def get_stream_state(self) -> StreamState:
23        return self._cursor
24
25    def set_initial_state(self, stream_state: StreamState) -> None:
26        self._cursor = stream_state
27
28    def observe(self, stream_slice: StreamSlice, record: Record) -> None:
29        """
30        Resumable full refresh manages state using a page number so it does not need to update state by observing incoming records.
31        """
32        pass
33
34    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
35        self._cursor = stream_slice.cursor_slice
36
37    def should_be_synced(self, record: Record) -> bool:
38        """
39        Unlike date-based cursors which filter out records outside slice boundaries, resumable full refresh records exist within pages
40        that don't have filterable bounds. We should always return them.
41        """
42        return True
43
44    def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
45        # A top-level RFR cursor only manages the state of a single partition
46        return self._cursor

179class ConcurrentCursor(Cursor):
180    _START_BOUNDARY = 0
181    _END_BOUNDARY = 1
182
183    def copy_without_state(self) -> "ConcurrentCursor":
184        return self.__class__(
185            stream_name=self._stream_name,
186            stream_namespace=self._stream_namespace,
187            stream_state={},
188            message_repository=NoopMessageRepository(),
189            connector_state_manager=ConnectorStateManager(),
190            connector_state_converter=self._connector_state_converter,
191            cursor_field=self._cursor_field,
192            slice_boundary_fields=self._slice_boundary_fields,
193            start=self._start,
194            end_provider=self._end_provider,
195            lookback_window=self._lookback_window,
196            slice_range=self._slice_range,
197            cursor_granularity=self._cursor_granularity,
198            clamping_strategy=self._clamping_strategy,
199            is_compare_strictly=self._is_compare_strictly,
200        )
201
202    def __init__(
203        self,
204        stream_name: str,
205        stream_namespace: Optional[str],
206        stream_state: Any,
207        message_repository: MessageRepository,
208        connector_state_manager: ConnectorStateManager,
209        connector_state_converter: AbstractStreamStateConverter,
210        cursor_field: CursorField,
211        slice_boundary_fields: Optional[Tuple[str, str]],
212        start: Optional[CursorValueType],
213        end_provider: Callable[[], CursorValueType],
214        lookback_window: Optional[GapType] = None,
215        slice_range: Optional[GapType] = None,
216        cursor_granularity: Optional[GapType] = None,
217        clamping_strategy: ClampingStrategy = NoClamping(),
218        is_compare_strictly: bool = False,
219    ) -> None:
220        self._stream_name = stream_name
221        self._stream_namespace = stream_namespace
222        self._message_repository = message_repository
223        self._connector_state_converter = connector_state_converter
224        self._connector_state_manager = connector_state_manager
225        self._cursor_field = cursor_field
226        # To see some example where the slice boundaries might not be defined, check https://github.com/airbytehq/airbyte/blob/1ce84d6396e446e1ac2377362446e3fb94509461/airbyte-integrations/connectors/source-stripe/source_stripe/streams.py#L363-L379
227        self._slice_boundary_fields = slice_boundary_fields
228        self._start = start
229        self._end_provider = end_provider
230        self.start, self._concurrent_state = self._get_concurrent_state(stream_state)
231        self._lookback_window = lookback_window
232        self._slice_range = slice_range
233        self._most_recent_cursor_value_per_partition: MutableMapping[
234            Union[StreamSlice, Mapping[str, Any], None], Any
235        ] = {}
236        self._has_closed_at_least_one_slice = False
237        self._cursor_granularity = cursor_granularity
238        # Flag to track if the logger has been triggered (per stream)
239        self._should_be_synced_logger_triggered = False
240        self._clamping_strategy = clamping_strategy
241        self._is_compare_strictly = is_compare_strictly
242        self._is_ascending_order = True
243
244        # A lock is required when closing a partition because updating the cursor's concurrent_state is
245        # not thread safe. When multiple partitions are being closed by the cursor at the same time, it is
246        # possible for one partition to update concurrent_state after a second partition has already read
247        # the previous state. This can lead to the second partition overwriting the previous one's state.
248        self._lock = threading.Lock()
249
250    @property
251    def state(self) -> MutableMapping[str, Any]:
252        return self._connector_state_converter.convert_to_state_message(
253            self.cursor_field, self._concurrent_state
254        )
255
256    @property
257    def cursor_field(self) -> CursorField:
258        return self._cursor_field
259
260    @property
261    def _slice_boundary_fields_wrapper(self) -> Tuple[str, str]:
262        return (
263            self._slice_boundary_fields
264            if self._slice_boundary_fields
265            else (
266                self._connector_state_converter.START_KEY,
267                self._connector_state_converter.END_KEY,
268            )
269        )
270
271    def _get_concurrent_state(
272        self, state: MutableMapping[str, Any]
273    ) -> Tuple[CursorValueType, MutableMapping[str, Any]]:
274        if self._connector_state_converter.is_state_message_compatible(state):
275            partitioned_state = self._connector_state_converter.deserialize(state)
276            slices_from_partitioned_state = partitioned_state.get("slices", [])
277
278            value_from_partitioned_state = None
279            if slices_from_partitioned_state:
280                # We assume here that the slices have been already merged
281                first_slice = slices_from_partitioned_state[0]
282                value_from_partitioned_state = (
283                    first_slice[self._connector_state_converter.MOST_RECENT_RECORD_KEY]
284                    if self._connector_state_converter.MOST_RECENT_RECORD_KEY in first_slice
285                    else first_slice[self._connector_state_converter.END_KEY]
286                )
287            return (
288                value_from_partitioned_state
289                or self._start
290                or self._connector_state_converter.zero_value,
291                partitioned_state,
292            )
293        return self._connector_state_converter.convert_from_sequential_state(
294            self._cursor_field, state, self._start
295        )
296
297    def observe(self, record: Record) -> None:
298        # Because observe writes to the most_recent_cursor_value_per_partition mapping,
299        # it is not thread-safe. However, this shouldn't lead to concurrency issues because
300        # observe() is only invoked by PartitionReader.process_partition(). Since the map is
301        # broken down according to partition, concurrent threads processing only read/write
302        # from different keys which avoids any conflicts.
303        #
304        # If we were to add thread safety, we should implement a lock per-partition
305        # which is instantiated during stream_slices()
306        most_recent_cursor_value = self._most_recent_cursor_value_per_partition.get(
307            record.associated_slice
308        )
309        try:
310            cursor_value = self._extract_cursor_value(record)
311
312            if most_recent_cursor_value is None or most_recent_cursor_value < cursor_value:
313                self._most_recent_cursor_value_per_partition[record.associated_slice] = cursor_value
314            elif most_recent_cursor_value > cursor_value:
315                self._is_ascending_order = False
316        except ValueError:
317            self._log_for_record_without_cursor_value()
318
319    def _extract_cursor_value(self, record: Record) -> Any:
320        return self._connector_state_converter.parse_value(self._cursor_field.extract_value(record))
321
322    def close_partition(self, partition: Partition) -> None:
323        with self._lock:
324            slice_count_before = len(self._concurrent_state.get("slices", []))
325            self._add_slice_to_state(partition)
326            if slice_count_before < len(
327                self._concurrent_state["slices"]
328            ):  # only emit if at least one slice has been processed
329                self._merge_partitions()
330                self._emit_state_message()
331        self._has_closed_at_least_one_slice = True
332
333    def _add_slice_to_state(self, partition: Partition) -> None:
334        most_recent_cursor_value = self._most_recent_cursor_value_per_partition.get(
335            partition.to_slice()
336        )
337
338        if self._slice_boundary_fields:
339            if "slices" not in self._concurrent_state:
340                raise RuntimeError(
341                    f"The state for stream {self._stream_name} should have at least one slice to delineate the sync start time, but no slices are present. This is unexpected. Please contact Support."
342                )
343            self._concurrent_state["slices"].append(
344                {
345                    self._connector_state_converter.START_KEY: self._extract_from_slice(
346                        partition, self._slice_boundary_fields[self._START_BOUNDARY]
347                    ),
348                    self._connector_state_converter.END_KEY: self._extract_from_slice(
349                        partition, self._slice_boundary_fields[self._END_BOUNDARY]
350                    ),
351                    self._connector_state_converter.MOST_RECENT_RECORD_KEY: most_recent_cursor_value,
352                }
353            )
354        elif most_recent_cursor_value:
355            if self._has_closed_at_least_one_slice:
356                # If we track state value using records cursor field, we can only do that if there is one partition. This is because we save
357                # the state every time we close a partition. We assume that if there are multiple slices, they need to be providing
358                # boundaries. There are cases where partitions could not have boundaries:
359                # * The cursor should be per-partition
360                # * The stream state is actually the parent stream state
361                # There might be other cases not listed above. Those are not supported today hence the stream should not use this cursor for
362                # state management. For the specific user that was affected with this issue, we need to:
363                # * Fix state tracking (which is currently broken)
364                # * Make the new version available
365                # * (Probably) ask the user to reset the stream to avoid data loss
366                raise ValueError(
367                    "Given that slice_boundary_fields is not defined and that per-partition state is not supported, only one slice is "
368                    "expected. Please contact the Airbyte team."
369                )
370
371            self._concurrent_state["slices"].append(
372                {
373                    self._connector_state_converter.START_KEY: self.start,
374                    self._connector_state_converter.END_KEY: most_recent_cursor_value,
375                    self._connector_state_converter.MOST_RECENT_RECORD_KEY: most_recent_cursor_value,
376                }
377            )
378
379    def _emit_state_message(self) -> None:
380        self._connector_state_manager.update_state_for_stream(
381            self._stream_name,
382            self._stream_namespace,
383            self.state,
384        )
385        state_message = self._connector_state_manager.create_state_message(
386            self._stream_name, self._stream_namespace
387        )
388        self._message_repository.emit_message(state_message)
389
390    def _merge_partitions(self) -> None:
391        self._concurrent_state["slices"] = self._connector_state_converter.merge_intervals(
392            self._concurrent_state["slices"]
393        )
394
395    def _extract_from_slice(self, partition: Partition, key: str) -> CursorValueType:
396        try:
397            _slice = partition.to_slice()
398            if not _slice:
399                raise KeyError(f"Could not find key `{key}` in empty slice")
400            return self._connector_state_converter.parse_value(_slice[key])  # type: ignore  # we expect the devs to specify a key that would return a CursorValueType
401        except KeyError as exception:
402            raise KeyError(
403                f"Partition is expected to have key `{key}` but could not be found"
404            ) from exception
405
406    def ensure_at_least_one_state_emitted(self) -> None:
407        """
408        The platform expect to have at least one state message on successful syncs. Hence, whatever happens, we expect this method to be
409        called.
410        """
411        self._emit_state_message()
412
413    def stream_slices(self) -> Iterable[StreamSlice]:
414        """
415        Generating slices based on a few parameters:
416        * lookback_window: Buffer to remove from END_KEY of the highest slice
417        * slice_range: Max difference between two slices. If the difference between two slices is greater, multiple slices will be created
418        * start: `_split_per_slice_range` will clip any value to `self._start which means that:
419          * if upper is less than self._start, no slices will be generated
420          * if lower is less than self._start, self._start will be used as the lower boundary (lookback_window will not be considered in that case)
421
422        Note that the slices will overlap at their boundaries. We therefore expect to have at least the lower or the upper boundary to be
423        inclusive in the API that is queried.
424        """
425        self._merge_partitions()
426
427        if self._start is not None and self._is_start_before_first_slice():
428            yield from self._split_per_slice_range(
429                self._start,
430                self._concurrent_state["slices"][0][self._connector_state_converter.START_KEY],
431                False,
432            )
433
434        if len(self._concurrent_state["slices"]) == 1:
435            yield from self._split_per_slice_range(
436                self._calculate_lower_boundary_of_last_slice(
437                    self._concurrent_state["slices"][0][self._connector_state_converter.END_KEY]
438                ),
439                self._end_provider(),
440                True,
441            )
442        elif len(self._concurrent_state["slices"]) > 1:
443            for i in range(len(self._concurrent_state["slices"]) - 1):
444                if self._cursor_granularity:
445                    yield from self._split_per_slice_range(
446                        self._concurrent_state["slices"][i][self._connector_state_converter.END_KEY]
447                        + self._cursor_granularity,
448                        self._concurrent_state["slices"][i + 1][
449                            self._connector_state_converter.START_KEY
450                        ],
451                        False,
452                    )
453                else:
454                    yield from self._split_per_slice_range(
455                        self._concurrent_state["slices"][i][
456                            self._connector_state_converter.END_KEY
457                        ],
458                        self._concurrent_state["slices"][i + 1][
459                            self._connector_state_converter.START_KEY
460                        ],
461                        False,
462                    )
463            yield from self._split_per_slice_range(
464                self._calculate_lower_boundary_of_last_slice(
465                    self._concurrent_state["slices"][-1][self._connector_state_converter.END_KEY]
466                ),
467                self._end_provider(),
468                True,
469            )
470        else:
471            raise ValueError("Expected at least one slice")
472
473    def _is_start_before_first_slice(self) -> bool:
474        return (
475            self._start is not None
476            and self._start
477            < self._concurrent_state["slices"][0][self._connector_state_converter.START_KEY]
478        )
479
480    def _calculate_lower_boundary_of_last_slice(
481        self, lower_boundary: CursorValueType
482    ) -> CursorValueType:
483        if self._lookback_window:
484            return lower_boundary - self._lookback_window
485        return lower_boundary
486
487    def _split_per_slice_range(
488        self, lower: CursorValueType, upper: CursorValueType, upper_is_end: bool
489    ) -> Iterable[StreamSlice]:
490        if lower > upper or (lower == upper and not self._cursor_granularity):
491            return
492
493        if self._start and upper < self._start:
494            return
495
496        lower = max(lower, self._start) if self._start else lower
497        # Clamp first, else a step shorter than the clamping period collapses the first window
498        lower = self._clamping_strategy.clamp(lower)
499        # With a granularity the last window includes upper, so lower == upper is one unit to read,
500        # unless the API rejects start == end. In a gap, the next slice already reads upper.
501        if lower > upper or (lower == upper and (self._is_compare_strictly or not upper_is_end)):
502            return
503        if not self._slice_range or self._evaluate_upper_safely(lower, self._slice_range) >= upper:
504            # With a granularity upper is the last unit to read: clamping it would read past the end
505            clamped_upper = (
506                upper
507                if upper_is_end and self._cursor_granularity
508                else self._clamping_strategy.clamp(upper)
509            )
510            start_value, end_value = (
511                (lower, clamped_upper - self._cursor_granularity)
512                if self._cursor_granularity and not upper_is_end
513                else (lower, clamped_upper)
514            )
515            yield StreamSlice(
516                partition={},
517                cursor_slice={
518                    self._slice_boundary_fields_wrapper[
519                        self._START_BOUNDARY
520                    ]: self._connector_state_converter.output_format(start_value),
521                    self._slice_boundary_fields_wrapper[
522                        self._END_BOUNDARY
523                    ]: self._connector_state_converter.output_format(end_value),
524                },
525            )
526        else:
527            stop_processing = False
528            current_lower_boundary = lower
529            while not stop_processing:
530                current_upper_boundary = min(
531                    self._evaluate_upper_safely(current_lower_boundary, self._slice_range), upper
532                )
533                clamped_upper = (
534                    self._clamping_strategy.clamp(current_upper_boundary)
535                    if current_upper_boundary != upper
536                    else current_upper_boundary
537                )
538                # A boundary clamped onto upper makes this the last window, else upper is dropped
539                has_reached_upper_boundary = (
540                    current_upper_boundary >= upper or clamped_upper == upper
541                )
542                clamped_lower = self._clamping_strategy.clamp(current_lower_boundary)
543                if clamped_lower >= clamped_upper:
544                    # clamping collapsed both values which means that it is time to stop processing
545                    # FIXME should this be replace by proper end_provider
546                    break
547                start_value, end_value = (
548                    (clamped_lower, clamped_upper - self._cursor_granularity)
549                    if self._cursor_granularity
550                    and (not upper_is_end or not has_reached_upper_boundary)
551                    else (clamped_lower, clamped_upper)
552                )
553                yield StreamSlice(
554                    partition={},
555                    cursor_slice={
556                        self._slice_boundary_fields_wrapper[
557                            self._START_BOUNDARY
558                        ]: self._connector_state_converter.output_format(start_value),
559                        self._slice_boundary_fields_wrapper[
560                            self._END_BOUNDARY
561                        ]: self._connector_state_converter.output_format(end_value),
562                    },
563                )
564                current_lower_boundary = clamped_upper
565                if has_reached_upper_boundary:
566                    stop_processing = True
567
568    def _evaluate_upper_safely(self, lower: CursorValueType, step: GapType) -> CursorValueType:
569        """
570        Given that we set the default step at datetime.timedelta.max, we will generate an OverflowError when evaluating the next start_date
571        This method assumes that users would never enter a step that would generate an overflow. Given that would be the case, the code
572        would have broken anyway.
573        """
574        try:
575            return lower + step
576        except OverflowError:
577            return self._end_provider()
578
579    def should_be_synced(self, record: Record) -> bool:
580        """
581        Determines if a record should be synced based on its cursor value.
582        :param record: The record to evaluate
583
584        :return: True if the record's cursor value falls within the sync boundaries
585        """
586        try:
587            record_cursor_value: CursorValueType = self._extract_cursor_value(record)
588        except ValueError:
589            self._log_for_record_without_cursor_value()
590            return True
591        return self._lowest_cursor_value_to_sync() <= record_cursor_value <= self._end_provider()
592
593    def _lowest_cursor_value_to_sync(self) -> CursorValueType:
594        # Keep the records the lookback window re-reads, never below the configured start. If the state is below the
595        # configured start, keep records from the state rather than raising the floor to the configured start.
596        if not self._lookback_window:
597            return self.start
598        try:
599            lower = self.start - self._lookback_window
600        except OverflowError:
601            return self.start
602        return max(lower, min(self.start, self._start)) if self._start else lower
603
604    def _log_for_record_without_cursor_value(self) -> None:
605        if not self._should_be_synced_logger_triggered:
606            LOGGER.warning(
607                f"Could not find cursor field `{self.cursor_field.cursor_field_key}` in record for stream {self._stream_name}. The incremental sync will assume it needs to be synced"
608            )
609            self._should_be_synced_logger_triggered = True
610
611    def reduce_slice_range(self, stream_slice: StreamSlice) -> StreamSlice:
612        # In theory, we might be more flexible here meaning that it doesn't need to be in ascending order but it just
613        # needs to be ordered. For now though, we will only support ascending order.
614        if not self._is_ascending_order:
615            LOGGER.warning(
616                "Attempting to reduce slice while records are not returned in incremental order might lead to missing records"
617            )
618
619        if stream_slice in self._most_recent_cursor_value_per_partition:
620            return StreamSlice(
621                partition=stream_slice.partition,
622                cursor_slice={
623                    self._slice_boundary_fields_wrapper[
624                        self._START_BOUNDARY
625                    ]: self._connector_state_converter.output_format(
626                        self._most_recent_cursor_value_per_partition[stream_slice]
627                    ),
628                    self._slice_boundary_fields_wrapper[
629                        self._END_BOUNDARY
630                    ]: stream_slice.cursor_slice[
631                        self._slice_boundary_fields_wrapper[self._END_BOUNDARY]
632                    ],
633                },
634                extra_fields=stream_slice.extra_fields,
635            )
636        else:
637            return stream_slice
638
639    def split_request_window(
640        self, stream_slice: StreamSlice, min_split_window: Optional[datetime.timedelta] = None
641    ) -> Optional[List[StreamSlice]]:
642        """
643        Split `stream_slice` in half along this cursor's own boundary fields, granularity, and output format.
644
645        Returns `None` when the slice cannot be split any further: either it spans at most one
646        `cursor_granularity` unit already (the floor), it is already at or below `min_split_window` (a
647        connector-configured floor independent of and typically looser than `cursor_granularity`), the
648        boundaries are reversed, or no `cursor_granularity` was configured at all (there is then no way to know
649        the smallest addressable unit, or how to keep two children from overlapping at their shared edge).
650        Raises `AirbyteTracedException` instead if a boundary is missing or unparseable: that is a bug in how
651        the slice was produced, not the API rejecting a window, so it must not be reported as one. This is the
652        implementation `SimpleRetriever.request_window_splitter` binds to (see
653        `airbyte_cdk.sources.declarative.retrievers.simple_retriever`) that reuses the same
654        parsing/formatting/comparison operations `_split_per_slice_range` already relies on, so it works for
655        any `CursorValueType`/`GapType` pair this cursor was built with (datetime/timedelta, or int/int),
656        without a second, duplicate implementation of that logic living outside the cursor.
657        """
658        if not self._cursor_granularity:
659            return None
660
661        start_field, end_field = self._slice_boundary_fields_wrapper
662        try:
663            start_value = self._connector_state_converter.parse_value(
664                stream_slice.cursor_slice[start_field]
665            )
666            end_value = self._connector_state_converter.parse_value(
667                stream_slice.cursor_slice[end_field]
668            )
669        except (KeyError, ValueError, TypeError) as exception:
670            # A missing or unparseable boundary means this slice was produced (or passed in) wrong - a bug in
671            # slice generation, not something the API did. Raising here, rather than returning `None` (which
672            # `SimpleRetriever` reports as "the API kept rejecting..."), keeps that distinction visible instead
673            # of blaming the API for our own state.
674            raise AirbyteTracedException(
675                internal_message=(
676                    f"Cannot split {stream_slice}: boundary field {start_field!r} or {end_field!r} is missing "
677                    f"or not parseable: {exception}"
678                ),
679                message=(
680                    f"Stream {self._stream_name} produced a request window with a missing or invalid "
681                    f"{start_field!r}/{end_field!r} boundary and could not split it further."
682                ),
683                failure_type=FailureType.system_error,
684            ) from exception
685
686        if start_value >= end_value:
687            # Reversed or degenerate boundaries: nothing safe to split.
688            return None
689
690        span = end_value - start_value  # type: ignore[operator]  # concretely a GapType (timedelta/int) at runtime
691        if min_split_window is not None and span <= min_split_window:  # type: ignore[operator]
692            return None
693
694        total_units = span // self._cursor_granularity + 1  # type: ignore[operator]  # inclusive count of granularity units
695        if total_units <= 1:
696            # Already at the minimum granularity and still rejected; nothing left to split.
697            return None
698
699        half_units = total_units // 2
700        first_end = (
701            start_value + (half_units * self._cursor_granularity) - self._cursor_granularity  # type: ignore[operator]
702        )
703        second_start = first_end + self._cursor_granularity
704
705        if not (start_value <= first_end < second_start <= end_value):
706            # No-progress guard, independent of the granularity check above: refuse to return children that
707            # are not strictly ordered and non-degenerate, rather than risk an identical or overlapping split.
708            return None
709
710        return [
711            StreamSlice(
712                partition=stream_slice.partition,
713                cursor_slice={
714                    start_field: self._connector_state_converter.output_format(start_value),
715                    end_field: self._connector_state_converter.output_format(first_end),
716                },
717                extra_fields=stream_slice.extra_fields,
718            ),
719            StreamSlice(
720                partition=stream_slice.partition,
721                cursor_slice={
722                    start_field: self._connector_state_converter.output_format(second_start),
723                    end_field: self._connector_state_converter.output_format(end_value),
724                },
725                extra_fields=stream_slice.extra_fields,
726            ),
727        ]
728
729    def get_cursor_datetime_from_state(
730        self, stream_state: Mapping[str, Any]
731    ) -> datetime.datetime | None:
732        """Extract and parse the cursor datetime from the given stream state.
733
734        For concurrent cursors, the state can be in two formats:
735        1. Sequential/legacy format: {cursor_field: cursor_value}
736        2. Concurrent format: {state_type: "date-range", slices: [...]}
737
738        Returns the cursor datetime if present and parseable, otherwise returns None.
739        """
740        # Check if state is in concurrent format (need to convert to dict for type compatibility)
741        mutable_state: MutableMapping[str, Any] = dict(stream_state)
742        if self._connector_state_converter.is_state_message_compatible(mutable_state):
743            slices = stream_state.get("slices", [])
744            if not slices:
745                return None
746            # Get the most recent cursor value from the first slice (after merging)
747            first_slice = slices[0]
748            cursor_value = first_slice.get(
749                self._connector_state_converter.MOST_RECENT_RECORD_KEY
750            ) or first_slice.get(self._connector_state_converter.END_KEY)
751            if not cursor_value:
752                return None
753            try:
754                parsed_value = self._connector_state_converter.parse_value(cursor_value)
755                if isinstance(parsed_value, datetime.datetime):
756                    return parsed_value
757                return None
758            except (ValueError, TypeError):
759                return None
760
761        # Sequential/legacy format: {cursor_field: cursor_value}
762        cursor_value = stream_state.get(self._cursor_field.cursor_field_key)
763        if not cursor_value:
764            return None
765        try:
766            parsed_value = self._connector_state_converter.parse_value(cursor_value)
767            if isinstance(parsed_value, datetime.datetime):
768                return parsed_value
769            return None
770        except (ValueError, TypeError):
771            return None