diff --git a/docs/02_concepts/08_pagination.mdx b/docs/02_concepts/08_pagination.mdx
index e68179ee..757f51c2 100644
--- a/docs/02_concepts/08_pagination.mdx
+++ b/docs/02_concepts/08_pagination.mdx
@@ -25,6 +25,8 @@ Most methods named `list` or `list_something` in the Apify client return a page
 - `count` - The number of items in the current page.
 - `limit` - The maximum number of items per page.
 
+On <ApiLink to="class/DatasetItemsPage">`DatasetItemsPage`</ApiLink>, `count` reports how many dataset rows the API scanned for the page. Filters drop items from a page and `unwind` multiplies them, so a page can hold fewer or more items than `count`.
+
 Some methods paginate differently. For example, <ApiLink to="class/RequestQueueClient#list_requests">`RequestQueueClient.list_requests`</ApiLink> returns a cursor-based <ApiLink to="class/ListOfRequests">`ListOfRequests`</ApiLink> without the `total`, `offset`, and `count` fields. To fetch the next page, pass its `next_cursor` value back as the `cursor` parameter. Other examples include `list_keys` and `list_head`. Regardless, the primary results are always stored under the `items` field, and the `limit` field can be used to control the number of results returned.
 
 The following example shows how to fetch all items from a dataset using pagination:
diff --git a/src/apify_client/_pagination.py b/src/apify_client/_pagination.py
index 4facbe58..46637719 100644
--- a/src/apify_client/_pagination.py
+++ b/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,20 +39,19 @@ 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 rows scanned across all pages. On the dataset items endpoint `unwind` can
+            turn one row into several items, so more items than this can be yielded. `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.
+        chunk_size: Per-page cap, sent to the API as its `limit`. `None` or `0` lets the API decide.
     """
     effective_chunk = chunk_size or 0
     initial_offset = offset or 0
@@ -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):
@@ -126,20 +128,30 @@ def get_cursor_iterator(
     limit: int | None = None,
     chunk_size: int | None = None,
 ) -> Iterator[KeyValueStoreKey] | Iterator[Request]:
-    """Yield individual items from cursor-paginated API responses.
+    """Yield individual items from a cursor-paginated API response.
+
+    This iterator supports the two API responses that use cursor pagination. `ListOfKeys` is used for key-value store
+    keys, while `ListOfRequests` is used for request queue requests.
+
+    Pagination continues until either:
+
+    - the API returns no next cursor, or
+    - the requested `limit` is reached.
+
+    An empty page does not explicitly stop the iteration. In practice, both supported endpoints return a next cursor
+    only when the current page contains items, so an empty page always has a `None` cursor and naturally ends the
+    iteration.
 
-    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.
+    The endpoints determine the next cursor differently:
+
+    - For key-value store keys, the cursor is the last key returned on the current page.
+    - For request queue requests, a cursor is returned only when the current page is full.
 
     Args:
-        callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
-        cursor: Value of the cursor for the first request, or `None` to start from the beginning.
+        callback: Function that returns one page of items and accepts `cursor` and `limit` keyword arguments.
+        cursor: Cursor to use for the first request. If `None`, iteration starts from the beginning.
         limit: Maximum total number of items to yield across all pages.
-        chunk_size: Maximum number of items requested per API call.
+        chunk_size: Maximum number of items to request in a single API call.
     """
     effective_chunk = chunk_size or 0
     initial_limit = limit or 0
@@ -218,3 +230,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. The larger of the two absorbs a `count` that lags behind the items returned, 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; on a page
+    covering fewer rows than that, the advance can still overshoot into rows a concurrent push appends afterwards. A
+    `requested_limit` 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
diff --git a/src/apify_client/_resource_clients/dataset.py b/src/apify_client/_resource_clients/dataset.py
index 194e6572..1f772dde 100644
--- a/src/apify_client/_resource_clients/dataset.py
+++ b/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."""
@@ -204,8 +204,8 @@ def list_items(
             items=items,
             total=int(response.headers['x-apify-pagination-total']),
             offset=int(response.headers['x-apify-pagination-offset']),
-            # x-apify-pagination-count returns count of processed items, not count of returned items
-            # This makes difference when items were filtered using hidden/empty
+            # The header counts the rows the API scanned, which `unwind` and a lagging dataset item count can
+            # both leave below the number of items returned.
             count=max(int(response.headers['x-apify-pagination-count']), len(items)),
             # API returns 999999999999 when no limit is used
             limit=int(response.headers['x-apify-pagination-limit']),
@@ -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. Fewer items are yielded when filters drop some, more
+                when `unwind` splits a row into several. 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
@@ -260,7 +261,7 @@ def iterate_items(
             skip_hidden: If True, then hidden fields are skipped from the output, i.e. fields starting with
                 the # character.
             signature: Signature used to access the items.
-            chunk_size: Maximum number of items requested per API call when iterating across pages.
+            chunk_size: Maximum number of dataset rows requested per API call when iterating across pages.
             timeout: Timeout for the API HTTP request.
 
         Yields:
@@ -763,8 +764,8 @@ async def list_items(
             items=items,
             total=int(response.headers['x-apify-pagination-total']),
             offset=int(response.headers['x-apify-pagination-offset']),
-            # x-apify-pagination-count returns count of processed items, not count of returned items
-            # This makes difference when items were filtered using hidden/empty
+            # The header counts the rows the API scanned, which `unwind` and a lagging dataset item count can
+            # both leave below the number of items returned.
             count=max(int(response.headers['x-apify-pagination-count']), len(items)),
             # API returns 999999999999 when no limit is used
             limit=int(response.headers['x-apify-pagination-limit']),
@@ -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. Fewer items are yielded when filters drop some, more
+                when `unwind` splits a row into several. 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
@@ -819,7 +821,7 @@ def iterate_items(
             skip_hidden: If True, then hidden fields are skipped from the output, i.e. fields starting with
                 the # character.
             signature: Signature used to access the items.
-            chunk_size: Maximum number of items requested per API call when iterating across pages.
+            chunk_size: Maximum number of dataset rows requested per API call when iterating across pages.
             timeout: Timeout for the API HTTP request.
 
         Yields:
diff --git a/tests/integration/test_dataset.py b/tests/integration/test_dataset.py
index 333c7229..79fc5816 100644
--- a/tests/integration/test_dataset.py
+++ b/tests/integration/test_dataset.py
@@ -596,6 +596,47 @@ async def get_items() -> DatasetItemsPage:
         await maybe_await(dataset_client.delete())
 
 
+async def test_dataset_iterate_items_unwound(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None:
+    """Test iterate_items with `unwind`, where a page carries more items than the rows it scanned."""
+    dataset_name = get_random_resource_name('dataset')
+    created_dataset = await maybe_await(client.datasets().get_or_create(name=dataset_name))
+    assert isinstance(created_dataset, Dataset)
+    dataset_client = client.dataset(created_dataset.id)
+
+    try:
+        items_to_push = [{'idx': i, 'parts': [{'part': p} for p in range(3)]} for i in range(12)]
+        await maybe_await(dataset_client.push_items(items_to_push))
+
+        # Poll until all 12 rows are visible (eventual consistency) so the chunked iteration sees every page
+        async def get_items() -> DatasetItemsPage:
+            page = await maybe_await(dataset_client.list_items(limit=12))
+            assert isinstance(page, DatasetItemsPage)
+            return page
+
+        await poll_until_condition(get_items, lambda page: len(page.items) == 12)
+
+        # chunk_size=5 caps a page at 5 rows, which `unwind` expands into 15 items
+        iterator = dataset_client.iterate_items(unwind=['parts'], chunk_size=5)
+        collected: list[dict] = []
+        if is_async:
+            assert isinstance(iterator, AsyncIterator)
+            async for item in iterator:
+                assert isinstance(item, dict)
+                collected.append(item)
+        else:
+            assert isinstance(iterator, Iterator)
+            for item in iterator:
+                assert isinstance(item, dict)
+                collected.append(item)
+
+        # Every part of every row arrives exactly once: no page is skipped and none is read twice.
+        assert sorted((item['idx'], item['part']) for item in collected) == [
+            (idx, part) for idx in range(12) for part in range(3)
+        ]
+    finally:
+        await maybe_await(dataset_client.delete())
+
+
 async def test_dataset_iterate_items_with_fields(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None:
     """Test iterate_items with `fields` filter."""
     dataset_name = get_random_resource_name('dataset')
diff --git a/tests/unit/test_client_pagination.py b/tests/unit/test_client_pagination.py
index 63c9b7c1..4b99ed11 100644
--- a/tests/unit/test_client_pagination.py
+++ b/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) -> 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) 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 1000',
+        {'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),