Add prefetch_pages to _get_multiple for overlapped page fetching

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Abel Milash

src/PowerPlatform/Dataverse/data/_odata.py

@@ -840,6 +840,7 @@ def _get_multiple(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Iterable[List[Dict[str, Any]]]:
         """Iterate records from an entity set, yielding one page (list of dicts) at a time.
 
@@ -861,6 +862,11 @@ def _get_multiple(
         :type count: ``bool``
         :param include_annotations: OData annotation pattern for the ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
         :type include_annotations: ``str`` | ``None``
+        :param prefetch_pages: Number of pages to pre-fetch ahead of the caller. ``0`` (default) is
+            fully sequential. ``1`` submits the next HTTP request immediately after receiving the
+            current page, overlapping network I/O with the caller's processing time. Values above
+            ``1`` are clamped to ``1``.
+        :type prefetch_pages: ``int``
 
         :return: Iterator yielding pages (each page is a ``list`` of record dicts).
         :rtype: ``Iterable[list[dict[str, Any]]]``
@@ -905,21 +911,38 @@ def _do_request(url: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[st
         if count:
             params["$count"] = "true"
 
-        data = _do_request(base_url, params=params)
-        items = data.get("value") if isinstance(data, dict) else None
-        if isinstance(items, list) and items:
-            yield [x for x in items if isinstance(x, dict)]
-
-        next_link = None
-        if isinstance(data, dict):
-            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink")
-
-        while next_link:
-            data = _do_request(next_link)
+        if prefetch_pages <= 0:
+            data = _do_request(base_url, params=params)
             items = data.get("value") if isinstance(data, dict) else None
             if isinstance(items, list) and items:
                 yield [x for x in items if isinstance(x, dict)]
-            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink") if isinstance(data, dict) else None
+
+            next_link = None
+            if isinstance(data, dict):
+                next_link = data.get("@odata.nextLink") or data.get("odata.nextLink")
+
+            while next_link:
+                data = _do_request(next_link)
+                items = data.get("value") if isinstance(data, dict) else None
+                if isinstance(items, list) and items:
+                    yield [x for x in items if isinstance(x, dict)]
+                next_link = data.get("@odata.nextLink") or data.get("odata.nextLink") if isinstance(data, dict) else None
+        else:
+            # Submit the next request before yielding the current page so
+            # network I/O overlaps with the caller's processing time.
+            ctx = copy_context()
+            executor = ThreadPoolExecutor(max_workers=1)
+            try:
+                pending = executor.submit(ctx.run, _do_request, base_url, params=params)
+                while pending is not None:
+                    data = pending.result()
+                    items = data.get("value") if isinstance(data, dict) else None
+                    next_link = (data.get("@odata.nextLink") or data.get("odata.nextLink")) if isinstance(data, dict) else None
+                    pending = executor.submit(ctx.run, _do_request, next_link) if next_link else None
+                    if isinstance(items, list) and items:
+                        yield [x for x in items if isinstance(x, dict)]
+            finally:
+                executor.shutdown(wait=False, cancel_futures=True)
 
     # --------------------------- SQL Custom API -------------------------
     def _query_sql(self, sql: str) -> list[dict[str, Any]]:

src/PowerPlatform/Dataverse/operations/records.py

@@ -301,6 +301,7 @@ def get(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Iterable[List[Record]]:
         """Fetch multiple records from a Dataverse table with pagination.
 
@@ -338,6 +339,11 @@ def get(
             ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or
             ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
         :type include_annotations: :class:`str` or None
+        :param prefetch_pages: When ``1``, the next page is fetched in a
+            background thread while the caller processes the current page,
+            overlapping network I/O with processing. ``0`` (default) is fully
+            sequential. Values above ``1`` are clamped to ``1``.
+        :type prefetch_pages: :class:`int`
 
         :return: Generator yielding pages, where each page is a list of
             :class:`~PowerPlatform.Dataverse.models.record.Record` objects.
@@ -370,6 +376,7 @@ def get(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Union[Record, Iterable[List[Record]]]:
         """Fetch a single record by ID, or fetch multiple records with pagination.
 
@@ -485,6 +492,7 @@ def _paged() -> Iterable[List[Record]]:
                     page_size=page_size,
                     count=count,
                     include_annotations=include_annotations,
+                    prefetch_pages=prefetch_pages,
                 ):
                     yield [Record.from_api_response(table, row) for row in page]
 

tests/perf/prefetch_pages.py

@@ -0,0 +1,143 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Performance comparison: prefetch_pages=0 (sequential) vs prefetch_pages=1.
+
+Usage:
+    python tests/perf/prefetch_pages.py
+
+Set PYTHONPATH=src before running. A browser login prompt will appear.
+
+Environment variables (all optional):
+    DATAVERSE_URL   Org URL — if omitted, you will be prompted interactively
+    PAGE_SIZE       Records per page (default: 10)
+    RUNS            Repetitions per mode for averaging (default: 3)
+"""
+
+import os
+import sys
+import time
+import statistics
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src"))
+
+from azure.identity import InteractiveBrowserCredential
+from PowerPlatform.Dataverse.client import DataverseClient
+
+_TABLE = "new_PrefetchPerfTest"
+_RECORD_COUNT = 50   # 50 records ÷ PAGE_SIZE pages for pagination
+
+
+def _backoff(op, delays=(0, 2, 5, 10, 20)):
+    last = None
+    for d in delays:
+        if d:
+            time.sleep(d)
+        try:
+            return op()
+        except Exception as ex:  # noqa: BLE001
+            last = ex
+    raise last
+
+
+def _setup(client: DataverseClient) -> None:
+    """Create table and populate with enough records to paginate."""
+    print(f"\n-> client.tables.get('{_TABLE}')")
+    info = _backoff(lambda: client.tables.get(_TABLE))
+    if info:
+        print(f"[OK] Table already exists: {info['table_schema_name']}")
+    else:
+        print(f"-> client.tables.create('{_TABLE}', ...)")
+        _backoff(lambda: client.tables.create(_TABLE, {"new_Label": "string"}))
+        print(f"[OK] Created table: {_TABLE}")
+
+    print(f"-> client.records.create('{_TABLE}', [{_RECORD_COUNT} records])")
+    records = [{"new_Label": f"record-{i:04d}"} for i in range(_RECORD_COUNT)]
+    ids = _backoff(lambda: client.records.create(_TABLE, records))
+    print(f"[OK] Inserted {len(ids)} records")
+
+
+def _cleanup(client: DataverseClient) -> None:
+    print(f"\n-> client.tables.delete('{_TABLE}')")
+    try:
+        _backoff(lambda: client.tables.delete(_TABLE))
+        print(f"[OK] Deleted table: {_TABLE}")
+    except Exception as ex:  # noqa: BLE001
+        print(f"[WARN] Cleanup failed: {ex}")
+
+
+def _consume(client: DataverseClient, page_size: int, prefetch: int) -> tuple[int, int, float]:
+    """Return (total_records, page_count, elapsed_seconds)."""
+    total = 0
+    pages = 0
+    t0 = time.perf_counter()
+    for page in client.records.get(_TABLE, page_size=page_size, prefetch_pages=prefetch):
+        total += len(page)
+        pages += 1
+        time.sleep(0.05)  # simulate per-page processing (e.g. db write, transform)
+    return total, pages, time.perf_counter() - t0
+
+
+def _run(client: DataverseClient, page_size: int, runs: int) -> None:
+    # Dry run to confirm page count
+    _, page_count, _ = _consume(client, page_size, prefetch=0)
+    print(f"\nPage size   : {page_size}")
+    print(f"Pages found : {page_count}")
+    print(f"Runs        : {runs}\n")
+
+    results: dict[int, list[float]] = {0: [], 1: []}
+
+    for run in range(1, runs + 1):
+        for mode in (0, 1):
+            _, _, elapsed = _consume(client, page_size, prefetch=mode)
+            results[mode].append(elapsed)
+            label = "sequential" if mode == 0 else "prefetch=1"
+            print(f"  Run {run}  {label:12s}  {elapsed:.3f}s")
+
+    print()
+    print("=" * 50)
+    print(f"{'Mode':<14}  {'Mean':>7}  {'Median':>7}  {'Min':>7}  {'Max':>7}")
+    print("-" * 50)
+    for mode, label in ((0, "sequential"), (1, "prefetch=1")):
+        t = results[mode]
+        print(f"{label:<14}  {statistics.mean(t):>6.3f}s  {statistics.median(t):>6.3f}s  {min(t):>6.3f}s  {max(t):>6.3f}s")
+    print("=" * 50)
+
+    mean_seq = statistics.mean(results[0])
+    mean_pre = statistics.mean(results[1])
+    if mean_seq > 0:
+        delta = (mean_seq - mean_pre) / mean_seq * 100
+        direction = "faster" if delta > 0 else "slower"
+        print(f"\nprefetch=1 is {abs(delta):.1f}% {direction} on average.")
+
+
+def main():
+    print("=" * 60)
+    print("prefetch_pages performance comparison")
+    print("=" * 60)
+
+    base_url = os.environ.get("DATAVERSE_URL", "").rstrip("/")
+    if not base_url:
+        base_url = input("Enter Dataverse org URL: ").strip().rstrip("/")
+    if not base_url:
+        print("No URL provided; exiting.")
+        sys.exit(1)
+
+    page_size = int(os.environ.get("PAGE_SIZE", "10"))
+    runs = int(os.environ.get("RUNS", "3"))
+
+    print(f"\n-> InteractiveBrowserCredential()")
+    credential = InteractiveBrowserCredential()
+
+    print(f"-> DataverseClient('{base_url}', ...)")
+    with DataverseClient(base_url=base_url, credential=credential) as client:
+        print(f"[OK] Connected to: {base_url}")
+        try:
+            _setup(client)
+            _run(client, page_size, runs)
+        finally:
+            _cleanup(client)
+
+
+if __name__ == "__main__":
+    main()

tests/unit/data/test_odata_internal.py

@@ -1086,6 +1086,81 @@ def test_empty_value_list_yields_nothing(self):
         self.assertEqual(pages, [])
 
 
+class TestGetMultiplePrefetch(unittest.TestCase):
+    """Behavioral tests for _get_multiple with prefetch_pages=1."""
+
+    def setUp(self):
+        self.od = _make_odata_client()
+        self.od._entity_set_from_schema_name = MagicMock(return_value="accounts")
+
+    def _two_page_responses(self):
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        page2 = _mock_response(
+            json_data={"value": [{"accountid": "id-2"}]},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, page2]
+
+    def test_prefetch_returns_same_pages_as_sequential(self):
+        """prefetch_pages=1 yields the same records in the same order as sequential mode."""
+        self._two_page_responses()
+        pages = list(self.od._get_multiple("account", prefetch_pages=1))
+        self.assertEqual(len(pages), 2)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.assertEqual(pages[1][0]["accountid"], "id-2")
+
+    def test_prefetch_values_above_one_clamped_to_one(self):
+        """prefetch_pages=2 (or any value > 1) behaves identically to prefetch_pages=1."""
+        self._two_page_responses()
+        pages = list(self.od._get_multiple("account", prefetch_pages=2))
+        self.assertEqual(len(pages), 2)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.assertEqual(pages[1][0]["accountid"], "id-2")
+        # Both pages should have been fetched (max_workers=1 means sequential requests)
+        self.assertEqual(self.od._request.call_count, 2)
+
+    def test_prefetch_exception_propagates_to_caller(self):
+        """An exception raised by the background request surfaces when the caller consumes the next page."""
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, RuntimeError("network failure")]
+        gen = self.od._get_multiple("account", prefetch_pages=1)
+        first = next(gen)
+        self.assertEqual(first[0]["accountid"], "id-1")
+        with self.assertRaises(RuntimeError):
+            next(gen)
+
+    def test_prefetch_early_close_does_not_hang(self):
+        """Closing the generator early (break) does not deadlock or raise an error."""
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        page2 = _mock_response(
+            json_data={"value": [{"accountid": "id-2"}]},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, page2]
+        gen = self.od._get_multiple("account", prefetch_pages=1)
+        first = next(gen)
+        self.assertEqual(first[0]["accountid"], "id-1")
+        gen.close()  # should not raise or hang
+
+    def test_prefetch_single_page_no_nextlink(self):
+        """prefetch_pages=1 with a single page (no nextLink) yields exactly one page."""
+        data = {"value": [{"accountid": "id-1"}]}
+        self.od._request.return_value = _mock_response(json_data=data, text=str(data))
+        pages = list(self.od._get_multiple("account", prefetch_pages=1))
+        self.assertEqual(len(pages), 1)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.od._request.assert_called_once()
+
+
 class TestQuerySql(unittest.TestCase):
     """Unit tests for _ODataClient._query_sql."""
 

tests/unit/test_client.py

@@ -128,6 +128,7 @@ def test_get_multiple(self):
             page_size=None,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
         self.assertEqual(len(results), 1)
         self.assertEqual(len(results[0]), 2)

tests/unit/test_client_dataframe.py

@@ -112,6 +112,7 @@ def test_get_passes_all_parameters(self):
             page_size=25,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
 
 

tests/unit/test_client_deprecations.py

@@ -132,6 +132,7 @@ def test_get_multiple_warns(self):
             page_size=None,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
 
     # ------------------------------------------------------------- query_sql

tests/unit/test_dataframe_operations.py

@@ -96,6 +96,7 @@ def test_get_passes_all_params(self):
             page_size=25,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
 
     def test_get_record_id_with_query_params_raises(self):

tests/unit/test_query_operations.py

@@ -83,6 +83,7 @@ def test_builder_execute_flat_default(self):
             page_size=None,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
         self.assertEqual(len(records), 1)
         self.assertEqual(records[0]["name"], "Test")
@@ -134,6 +135,7 @@ def test_builder_execute_all_params(self):
             page_size=25,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
 
     def test_builder_execute_with_where(self):

tests/unit/test_records_operations.py

@@ -245,6 +245,7 @@ def test_get_paginated_with_all_params(self):
             page_size=25,
             count=False,
             include_annotations=None,
+            prefetch_pages=0,
         )
 
     # ------------------------------------------------------------------ upsert