airbyte_cdk.sources.streams.concurrent.cursor

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