FIX: Address Copilot review comments in async zarr loading

- Fix typo in comment (rootshalgive -> root)
- Add max_concurrency validation (raise ValueError if < 1)
- Update open_store_async docstring to reflect actual behavior
- Add semaphore bound to _iter_zarr_groups_async
- Add create_default_indexes param to _open_groups_from_stores_async
- Add semaphore bound to create_indexes_async in api.py
- Fix whats-new.rst PR references (10742 -> 11149)

Author: AAlexxis222

doc/whats-new.rst

@@ -16,7 +16,7 @@ New Features
 
 - Added ``max_concurrency`` parameter to :py:func:`open_datatree` to control
   the maximum number of concurrent I/O operations when opening groups in parallel
-  with the Zarr backend (:pull:`10742`).
+  with the Zarr backend (:pull:`11149`).
   By `Alfonso Ladino <https://github.com/aladinor>`_.
 
 Breaking Changes
@@ -153,7 +153,7 @@ Performance
 ~~~~~~~~~~~
 
 - Improve performance of :py:func:`open_datatree` for zarr stores by using async/concurrent
-  loading of groups and indexes (:pull:`10742`).
+  loading of groups and indexes (:pull:`11149`).
   By `Alfonso Ladino <https://github.com/aladinor>`_.
 - Add a fastpath to the backend plugin system for standard engines (:issue:`10178`, :pull:`10937`).
   By `Sam Levang <https://github.com/slevang>`_.

xarray/backends/api.py

@@ -362,9 +362,17 @@ def _datatree_from_backend_datatree(
             async def create_indexes_async() -> dict[str, Dataset]:
                 import asyncio
 
+                sem = asyncio.Semaphore(10)
+
+                async def _bounded_create_index(
+                    path: str, ds: Dataset,
+                ) -> tuple[str, Dataset]:
+                    async with sem:
+                        return await _create_index_for_node(path, ds)
+
                 results: dict[str, Dataset] = {}
                 tasks = [
-                    _create_index_for_node(path, node.dataset)
+                    _bounded_create_index(path, node.dataset)
                     for path, [node] in group_subtrees(backend_tree)
                 ]
                 for fut in asyncio.as_completed(tasks):

xarray/backends/zarr.py

@@ -767,11 +767,13 @@ async def open_store_async(
         write_empty: bool | None = None,
         cache_members: bool = False,
     ):
-        """Async version of open_store using flat group discovery.
+        """Async version of open_store using asynchronous group discovery.
 
-        This method uses store.list() to discover all groups in a single
-        async call, which is significantly faster than recursive traversal
-        for stores that support listing (like icechunk).
+        This method discovers all groups by recursively traversing the store,
+        issuing asynchronous directory listing calls (via store.list_dir when
+        available) and per-directory store.get calls. As a result, group discovery
+        may involve multiple asynchronous operations and round-trips, especially
+        for large or deeply nested hierarchies.
 
         Parameters
         ----------
@@ -1960,6 +1962,7 @@ async def _open_groups_from_stores_async(
         use_cftime=None,
         decode_timedelta=None,
         max_concurrency: int | None = None,
+        create_default_indexes: bool = True,
     ) -> dict[str, Dataset]:
         """Shared async core: open datasets from pre-opened stores concurrently.
 
@@ -1981,6 +1984,10 @@ async def _open_groups_from_stores_async(
 
         if max_concurrency is None:
             max_concurrency = 10
+        elif max_concurrency < 1:
+            raise ValueError(
+                f"max_concurrency must be None or an integer >= 1, got {max_concurrency!r}"
+            )
         sem = asyncio.Semaphore(max_concurrency)
 
         async def open_one(path_group: str, store) -> tuple[str, Dataset]:
@@ -1998,7 +2005,8 @@ async def open_one(path_group: str, store) -> tuple[str, Dataset]:
                         decode_timedelta=decode_timedelta,
                     )
                 # Create indexes concurrently
-                ds = await _maybe_create_default_indexes_async(ds)
+                if create_default_indexes:
+                    ds = await _maybe_create_default_indexes_async(ds)
                 if group:
                     group_name = str(NodePath(path_group).relative_to(parent))
                 else:
@@ -2232,16 +2240,19 @@ async def discover_subgroups(dir_path: str) -> list[str]:
                 subdirs.append(subdir)
 
         # Check each subdirectory in parallel
+        _sem = asyncio.Semaphore(10)
+
         async def check_subdir(subdir: str) -> tuple[str | None, list[str]]:
             """Check if subdir is a group and discover its subgroups."""
-            if await is_zarr_group(subdir):
-                group_path = (
-                    str(parent_nodepath / subdir) if subdir else str(parent_nodepath)
-                )
-                # Recursively find subgroups
-                sub_groups = await discover_subgroups(subdir)
-                return group_path, sub_groups
-            return None, []
+            async with _sem:
+                if await is_zarr_group(subdir):
+                    group_path = (
+                        str(parent_nodepath / subdir) if subdir else str(parent_nodepath)
+                    )
+                    # Recursively find subgroups
+                    sub_groups = await discover_subgroups(subdir)
+                    return group_path, sub_groups
+                return None, []
 
         # Run all subdirectory checks in parallel
         results = await asyncio.gather(*[check_subdir(sd) for sd in subdirs])
@@ -2253,7 +2264,7 @@ async def check_subdir(subdir: str) -> tuple[str | None, list[str]]:
 
         return found_groups
 
-    # Start discovery from rootshalgive
+    # Start discovery from root
     subgroups = await discover_subgroups("")
     group_paths.extend(subgroups)