fix: Stop dataset iterators from skipping items when unwind is used

src/apify_client/_pagination.py

@@ -19,9 +19,10 @@
 class HasItems(Protocol[T]):
     """Structural contract for a single page of results from a paginated API endpoint.
 
-    Implementations must expose `items`. They may optionally expose `count` - the number of items scanned by the API for
-    this page, which can exceed `len(items)` when filters drop items from the response. The iterator helpers consult
-    `count` opportunistically via `getattr` for offset bookkeeping and fall back to `len(items)` when it is absent.
+    Implementations must expose `items`. They may optionally expose `count` - the number of rows the API scanned to
+    produce this page, which `len(items)` can land below (filters drop items) or above (`unwind` splits one row into
+    several items). The iterator helpers consult `count` opportunistically via `getattr` for offset bookkeeping and
+    fall back to `len(items)` when it is absent.
     """
 
     items: list[T]
@@ -38,18 +39,17 @@ def get_items_iterator(
 
     The `callback` is invoked lazily to fetch each page from the API. It must accept `limit` and `offset` keyword
     arguments and return an object whose `items` attribute is a list. If the object also exposes a `count` attribute, it
-    is used for offset bookkeeping (the Apify API's `count` reflects items scanned, which can exceed items returned when
-    filters are applied).
+    is used for offset bookkeeping - `_page_scanned_rows` describes how the next offset is derived.
 
-    Iteration stops when a page scans no items (`count` is `0`, or `items` is empty when `count` is absent) or when the
-    user-requested `limit` is reached. A page can scan items while returning none - filters like `clean` drop items from
-    `items` but still count toward `count` - so terminating on scanned rather than returned items keeps the iterator
-    advancing across fully-filtered pages. The `total` field is intentionally not consulted, because it can change
-    between calls.
+    Iteration stops when a page scans no rows or when the user-requested `limit` is reached. A page can scan rows while
+    returning no items - filters like `clean` drop items from `items` but still count toward `count` - so terminating on
+    scanned rather than returned rows keeps the iterator advancing across fully-filtered pages. The `total` field is
+    intentionally not consulted, because it can change between calls.
 
     Args:
         callback: Function returning a single page of items.
-        limit: Maximum total number of items to yield across all pages. `None` or `0` means no limit.
+        limit: Maximum total number of scanned rows across all pages - `unwind` can turn those into more yielded items.
+            `None` or `0` means no limit.
         offset: Starting offset for the first page.
         chunk_size: Maximum number of items requested per API call. `None` or `0` lets the API decide.
     """
@@ -59,13 +59,14 @@ def get_items_iterator(
     fetched_items = 0
 
     while True:
+        page_limit = _next_page_limit(initial_limit, fetched_items, effective_chunk)
         current_page = callback(
-            limit=_next_page_limit(initial_limit, fetched_items, effective_chunk),
+            limit=page_limit,
             offset=initial_offset + fetched_items,
         )
         yield from current_page.items
 
-        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = _page_scanned_rows(current_page, page_limit)
         fetched_items += page_scanned
 
         if not page_scanned or (initial_limit and fetched_items >= initial_limit):
@@ -89,14 +90,15 @@ async def get_items_iterator_async(
     fetched_items = 0
 
     while True:
+        page_limit = _next_page_limit(initial_limit, fetched_items, effective_chunk)
         current_page = await callback(
-            limit=_next_page_limit(initial_limit, fetched_items, effective_chunk),
+            limit=page_limit,
             offset=initial_offset + fetched_items,
         )
         for item in current_page.items:
             yield item
 
-        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = _page_scanned_rows(current_page, page_limit)
         fetched_items += page_scanned
 
         if not page_scanned or (initial_limit and fetched_items >= initial_limit):
@@ -130,10 +132,9 @@ def get_cursor_iterator(
 
     Cursor pagination is restricted to the two API responses that expose it: `ListOfKeys` (for key-value store keys) and
     `ListOfRequests` (for request queue requests). Iteration ends when the next cursor is `None` or the user-requested
-    `limit` is reached. Emptiness alone does not stop iteration: server-side filters (such as the request-queue state
-    `filter`) can drop every item on a page while a live cursor still points at more data, so termination relies on the
-    cursor, not on whether a page returned items. Unlike offset responses, cursor responses expose no scanned-item
-    `count`, so `count` cannot be used to detect a fully-filtered page here.
+    `limit` is reached; an empty page on its own does not stop it. Both endpoints derive the cursor from the page they
+    return - the key-value store hands back the last key of the page, the request queue hands back a cursor only once a
+    page came back full - so termination can rest on the cursor alone.
 
     Args:
         callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
@@ -218,3 +219,19 @@ def _next_page_limit(initial_limit: int, fetched_items: int, effective_chunk: in
     if not effective_chunk:
         return remaining
     return min(remaining, effective_chunk)
+
+
+def _page_scanned_rows(page: HasItems[T], requested_limit: int) -> int:
+    """Compute how far the offset advances past `page`, in dataset rows.
+
+    Neither reported number is right on its own. `count` follows the rows the API scanned, but it is derived from a
+    dataset's item count, which is incremented by a throttled write and so lags a fresh push. `len(items)` counts the
+    items the API shaped out of those rows: filters (`clean`, `skip_empty`, `skip_hidden`) drop some, and `unwind`
+    splits one row into several. Taking the larger of the two covers a lagging `count`, and capping it at the rows the
+    call asked for keeps an unwound page from advancing past rows the next call would then never read. The cap is a
+    valid bound because the endpoint applies the `limit` it is sent verbatim: a page covering fewer rows than the call
+    asked for has reached the end of the dataset, where an overshoot costs nothing. A cap of `0` means the call sent no
+    limit, leaving the advance unbounded.
+    """
+    scanned_rows = max(getattr(page, 'count', 0), len(page.items))
+    return min(scanned_rows, requested_limit) if requested_limit else scanned_rows

src/apify_client/_resource_clients/dataset.py

@@ -40,7 +40,7 @@ class DatasetItemsPage:
     """The offset of the first item in this page."""
 
     count: int
-    """Number of items in this page."""
+    """Number of dataset rows the API scanned for this page, or the number of items returned when that is larger."""
 
     limit: int
     """The limit that was used for this request."""
@@ -237,7 +237,8 @@ def iterate_items(
 
         Args:
             offset: Number of items that should be skipped at the start. The default value is 0.
-            limit: Maximum number of items to return. By default there is no limit.
+            limit: Maximum number of dataset rows to scan. Filters leave fewer items than that, `unwind` more.
+                By default there is no limit.
             desc: By default, results are returned in the same order as they were stored. To reverse the order,
                 set this parameter to True.
             clean: If True, returns only non-empty items and skips hidden fields (i.e. fields starting with
@@ -796,7 +797,8 @@ def iterate_items(
 
         Args:
             offset: Number of items that should be skipped at the start. The default value is 0.
-            limit: Maximum number of items to return. By default there is no limit.
+            limit: Maximum number of dataset rows to scan. Filters leave fewer items than that, `unwind` more.
+                By default there is no limit.
             desc: By default, results are returned in the same order as they were stored. To reverse the order,
                 set this parameter to True.
             clean: If True, returns only non-empty items and skips hidden fields (i.e. fields starting with

tests/unit/test_client_pagination.py

@@ -100,6 +100,7 @@
 NORMAL_ITEMS = 2500
 EXTRA_ITEMS_UNNAMED = 100
 MAX_ITEMS_PER_PAGE = 1000
+UNWIND_PARTS = 3
 
 # Inner list models whose `items: list[<specific schema>]` is relaxed to `list[dict]`. Point of these tests is
 # pagination mechanism, not internal object validation.
@@ -179,6 +180,11 @@ def create_items(start: int, end: int, step: int | None = None) -> list[dict[str
     return [{'id': i} for i in range(start, end, step)]
 
 
+def create_unwound_items(start: int, end: int, step: int | None = None) -> list[dict[str, int]]:
+    """Create the items the simulated `unwind` produces for the given index range."""
+    return [{**item, 'part': part} for item in create_items(start, end, step) for part in range(UNWIND_PARTS)]
+
+
 def _is_true(value: str | None) -> bool:
     """Match the `'true'` wire form produced by the client's bool->string serialization."""
     return value == 'true'
@@ -191,9 +197,14 @@ def _parse_int_param(value: str | None) -> int:
 def _handle_offset_pagination(request: Request) -> Response:
     """Serve an offset-paginated Apify API response.
 
-    The simulated platform holds 2500 items normally and an additional 100 when `unnamed=true` is requested. Pages are
-    capped at 1000 items regardless of the requested limit, mirroring the real API. The dataset items endpoint returns
-    items as a raw list; all other endpoints wrap them in `{'data': {...}}`.
+    The simulated platform holds 2500 items normally and an additional 100 when `unnamed=true` is requested. The
+    collection endpoints cap a page at 1000 items regardless of the requested limit, mirroring the real API, while the
+    dataset items endpoint applies the requested limit verbatim and returns its items as a raw list; all other
+    endpoints wrap them in `{'data': {...}}`.
+
+    The `x-apify-pagination-count` header reports the rows the API scanned, which `offset` and `limit` pick before the
+    result is shaped: the filters drop items from the page and `unwind` multiplies them, so `len(items)` lands below or
+    above the header.
     """
     params = request.args
 
@@ -206,15 +217,22 @@ def _handle_offset_pagination(request: Request) -> Response:
     desc = _is_true(params.get('desc'))
     items = create_items(total_items, 0) if desc else create_items(0, total_items)
 
+    is_dataset_items = request.path.endswith(f'/datasets/{ID_PLACEHOLDER}/items')
+    page_size = total_items if is_dataset_items else MAX_ITEMS_PER_PAGE
+
     lower_index = min(offset, total_items)
     upper_index = min(offset + (limit or total_items), total_items)
-    count = min(max(upper_index - lower_index, 0), MAX_ITEMS_PER_PAGE)
-    selected_items = items[lower_index : min(upper_index, lower_index + MAX_ITEMS_PER_PAGE)]
+    count = min(max(upper_index - lower_index, 0), page_size)
+    selected_items = items[lower_index : min(upper_index, lower_index + page_size)]
 
     # Every second item is filtered out when `skipEmpty=true`, `skipHidden=true`, or `clean=true`.
     if _is_true(params.get('skipEmpty')) or _is_true(params.get('skipHidden')) or _is_true(params.get('clean')):
         selected_items = selected_items[::2]
 
+    # `unwind` splits each item into `UNWIND_PARTS` records, so the page carries more items than the rows it scanned.
+    if params.get('unwind'):
+        selected_items = [{**item, 'part': part} for item in selected_items for part in range(UNWIND_PARTS)]
+
     headers = {
         'x-apify-pagination-count': str(count),
         'x-apify-pagination-total': str(total_items),
@@ -224,7 +242,7 @@ def _handle_offset_pagination(request: Request) -> Response:
         'content-type': 'application/json',
     }
 
-    if request.path.endswith(f'/datasets/{ID_PLACEHOLDER}/items'):
+    if is_dataset_items:
         body: Any = selected_items
     else:
         body = {
@@ -443,6 +461,25 @@ def __hash__(self) -> int:
         create_items(0, 1500, 2),
         DATASET_CLIENTS,
     ),
+    _PaginationCase(
+        'Unwind',
+        {'unwind': ['parts']},
+        create_unwound_items(0, 2500),
+        DATASET_CLIENTS,
+    ),
+    _PaginationCase(
+        'Unwind, limit, chunk_size',
+        # `limit` counts the rows the API scans, matching a single `list_items` call, so `unwind` yields more items.
+        {'unwind': ['parts'], 'limit': 150, 'chunk_size': 100},
+        create_unwound_items(0, 150),
+        DATASET_CLIENTS,
+    ),
+    _PaginationCase(
+        'Unwind, chunk_size above the collection page cap',
+        {'unwind': ['parts'], 'chunk_size': 2000},
+        create_unwound_items(0, 2500),
+        DATASET_CLIENTS,
+    ),
     _PaginationCase(
         'Exclusive start key',
         {'exclusive_start_key': '1000'},
@@ -629,7 +666,7 @@ async def test_rq_list_requests_iterable_async(
 
 
 class FakeOffsetPage:
-    """Offset-paginated page whose `count` (items scanned) may exceed `len(items)` when filters drop items."""
+    """Offset-paginated page whose `count` (rows scanned) and `len(items)` can differ in either direction."""
 
     def __init__(self, items: list[dict[str, int]], count: int) -> None:
         self.items = items
@@ -662,8 +699,62 @@ async def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffse
     assert [item async for item in get_items_iterator_async(callback, chunk_size=1000)] == [{'id': 1}, {'id': 2}]
 
 
-def test_cursor_iterator_continues_past_fully_filtered_page() -> None:
-    """A fully-filtered page (`items=[]`) with a live cursor must not stop the cursor iterator."""
+def test_items_iterator_advances_by_items_when_count_lags() -> None:
+    """A `count` lagging behind the items returned (`count=0`) must still advance the offset iterator."""
+    pages = {
+        0: FakeOffsetPage(items=create_items(0, 1000), count=0),
+        1000: FakeOffsetPage(items=create_items(1000, 1500), count=0),
+    }
+
+    def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
+
+    assert list(get_items_iterator(callback, chunk_size=1000)) == create_items(0, 1500)
+
+
+async def test_items_iterator_async_advances_by_items_when_count_lags() -> None:
+    """A `count` lagging behind the items returned (`count=0`) must still advance the async offset iterator."""
+    pages = {
+        0: FakeOffsetPage(items=create_items(0, 1000), count=0),
+        1000: FakeOffsetPage(items=create_items(1000, 1500), count=0),
+    }
+
+    async def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
+
+    collected = [item async for item in get_items_iterator_async(callback, chunk_size=1000)]
+    assert collected == create_items(0, 1500)
+
+
+def test_items_iterator_advances_by_scanned_rows_when_unwind_inflates_items() -> None:
+    """An unwound page holding more items than the rows it scanned must advance the offset by the rows alone."""
+    pages = {
+        0: FakeOffsetPage(items=create_unwound_items(0, 1000), count=1000),
+        1000: FakeOffsetPage(items=create_unwound_items(1000, 1500), count=500),
+    }
+
+    def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
+
+    assert list(get_items_iterator(callback, chunk_size=1000)) == create_unwound_items(0, 1500)
+
+
+async def test_items_iterator_async_advances_by_scanned_rows_when_unwind_inflates_items() -> None:
+    """An unwound page holding more items than the rows it scanned must advance the async offset iterator by rows."""
+    pages = {
+        0: FakeOffsetPage(items=create_unwound_items(0, 1000), count=1000),
+        1000: FakeOffsetPage(items=create_unwound_items(1000, 1500), count=500),
+    }
+
+    async def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
+
+    collected = [item async for item in get_items_iterator_async(callback, chunk_size=1000)]
+    assert collected == create_unwound_items(0, 1500)
+
+
+def test_cursor_iterator_continues_past_empty_page() -> None:
+    """An empty page with a live cursor must not stop the cursor iterator."""
     pages = {
         None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
         'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),
@@ -675,8 +766,8 @@ def callback(*, cursor: str | None = None, **_kwargs: object) -> ListOfRequests:
     assert list(get_cursor_iterator(callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]
 
 
-async def test_cursor_iterator_async_continues_past_fully_filtered_page() -> None:
-    """A fully-filtered page (`items=[]`) with a live cursor must not stop the async cursor iterator."""
+async def test_cursor_iterator_async_continues_past_empty_page() -> None:
+    """An empty page with a live cursor must not stop the async cursor iterator."""
     pages = {
         None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
         'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),