From 78d2b3ed7eaa0e06e033f89764ba293c1bc840d7 Mon Sep 17 00:00:00 2001 From: ANIRUDDHA ADAK Date: Mon, 28 Sep 2026 23:39:28 +0530 Subject: [PATCH 1/4] fix(indexing): align in-memory indexer for empty ndarray keys (#11626) --- doc/whats-new.rst | 7 ++++++ xarray/core/indexing.py | 14 ++++++++++- xarray/tests/test_backends.py | 15 ++++++++++++ xarray/tests/test_indexing.py | 46 +++++++++++++++++++++++++++++++++++ 4 files changed, 81 insertions(+), 1 deletion(-) diff --git a/doc/whats-new.rst b/doc/whats-new.rst index bfbff532c17..5540de966c6 100644 --- a/doc/whats-new.rst +++ b/doc/whats-new.rst @@ -129,6 +129,13 @@ Bug Fixes - Fixed a bug that caused rechunking a multi-dimensional cftime array along a subset of its dimensions to raise an error (:issue:`11567`, :pull:`11576`). By `Spencer Clark `_. +- Fixed indexing with an empty indexer array. An empty indexer array is now always + turned into an empty slice for the backend, so that the in-memory part of the + decomposed indexer stays aligned with the axes of the loaded array. Previously the + ``h5netcdf`` and ``scipy`` engines raised ``IndexError`` for multi-dimensional + variables, the ``netCDF4`` engine silently returned a wrongly-sized array, and + ``pydap`` raised ``ValueError`` (:issue:`9075`, :issue:`11625`, :pull:`11626`). + By `Aniruddha Adak `_. .. _`pandas-dev/pandas#64793`: https://github.com/pandas-dev/pandas/pull/64793 diff --git a/xarray/core/indexing.py b/xarray/core/indexing.py index a96b102f928..c41519f39fb 100644 --- a/xarray/core/indexing.py +++ b/xarray/core/indexing.py @@ -1411,6 +1411,10 @@ def _decompose_outer_indexer( # empty np.ndarray key is converted to empty slice # see https://github.com/pydata/xarray/issues/10867 backend_indexer.append(slice(0, 0)) + # an empty slice does not drop the dimension (unlike an integer + # key), so np_indexer needs a matching entry to stay aligned + # with the axes of the loaded array + np_indexer.append(slice(None)) elif isinstance(k, np.ndarray) and i != array_index: # np.ndarray key is converted to slice that covers the entire # entries of this key. @@ -1437,6 +1441,10 @@ def _decompose_outer_indexer( bk_slice, np_slice = _decompose_slice(k, s) backend_indexer.append(bk_slice) np_indexer.append(np_slice) + elif isinstance(k, np.ndarray) and k.size == 0: + # empty np.ndarray key is converted to empty slice + backend_indexer.append(slice(0, 0)) + np_indexer.append(slice(None)) elif isinstance(k, integer_types): backend_indexer.append(k) elif isinstance(k, np.ndarray) and (np.diff(k) >= 0).all(): @@ -1454,7 +1462,11 @@ def _decompose_outer_indexer( assert indexing_support == IndexingSupport.BASIC for k, s in zip(indexer_elems, shape, strict=False): - if isinstance(k, np.ndarray): + if isinstance(k, np.ndarray) and k.size == 0: + # empty np.ndarray key is converted to empty slice + backend_indexer.append(slice(0, 0)) + np_indexer.append(slice(None)) + elif isinstance(k, np.ndarray): # np.ndarray key is converted to slice that covers the entire # entries of this key. backend_indexer.append(slice(np.min(k), np.max(k) + 1)) diff --git a/xarray/tests/test_backends.py b/xarray/tests/test_backends.py index c52c4674ea7..10c7b746399 100644 --- a/xarray/tests/test_backends.py +++ b/xarray/tests/test_backends.py @@ -1043,6 +1043,21 @@ def test_empty_isel(self) -> None: actual = on_disk.isel(x=[]) assert_identical(expected, actual) + def test_empty_isel_multidim(self) -> None: + # Empty indexers must also work for multi-dimensional variables, where + # the in-memory part of the decomposed indexer still has to line up with + # the axes of the loaded array. + # GH:issue:11625, GH:issue:9075 + in_memory = xr.Dataset( + {"a": (("x", "y"), np.arange(12.0).reshape(4, 3))}, + coords={"x": np.arange(4), "y": np.arange(3)}, + ) + with self.roundtrip(in_memory) as on_disk: + for x_indexer, y_indexer in (([], [0, 2]), ([1, 3], []), ([], [])): + expected = in_memory.isel(x=x_indexer, y=y_indexer) + actual = on_disk.isel(x=x_indexer, y=y_indexer) + assert_identical(expected, actual) + def validate_array_type(self, ds): # Make sure that only NumpyIndexingAdapter stores a bare np.ndarray. diff --git a/xarray/tests/test_indexing.py b/xarray/tests/test_indexing.py index c8193ac1e64..3ca3a34e1a7 100644 --- a/xarray/tests/test_indexing.py +++ b/xarray/tests/test_indexing.py @@ -889,6 +889,52 @@ def test_decompose_indexers(shape, indexer_mode, indexing_support) -> None: np.testing.assert_array_equal(expected, array) +@pytest.mark.parametrize("shape", [(10, 5), (10, 5, 8)]) +@pytest.mark.parametrize("empty_axis", ["first", "last"]) +@pytest.mark.parametrize( + "indexing_support", + [ + indexing.IndexingSupport.BASIC, + indexing.IndexingSupport.OUTER, + indexing.IndexingSupport.OUTER_1VECTOR, + indexing.IndexingSupport.VECTORIZED, + ], +) +def test_decompose_indexer_empty_ndarray(shape, empty_axis, indexing_support) -> None: + # An empty indexer array must be decomposable for backends that do not + # support full vectorized indexing. The empty array is turned into an empty + # slice, which keeps the dimension, so the in-memory indexer needs a + # matching entry for it. + # See https://github.com/pydata/xarray/issues/11625 + # and https://github.com/pydata/xarray/issues/9075 + data = np.random.randn(*shape) + key = [np.arange(s)[::-1] for s in shape] + key[0 if empty_axis == "first" else -1] = np.array([], dtype=np.int64) + indexer = indexing.OuterIndexer(tuple(key)) + + expected = indexing.NumpyIndexingAdapter(data).oindex[indexer] + backend_ind, np_ind = indexing.decompose_indexer(indexer, shape, indexing_support) + indexing_adapter = indexing.NumpyIndexingAdapter(data) + + if isinstance(backend_ind, indexing.VectorizedIndexer): + array = indexing_adapter.vindex[backend_ind] + elif isinstance(backend_ind, indexing.OuterIndexer): + array = indexing_adapter.oindex[backend_ind] + else: + array = indexing_adapter[backend_ind] + + if len(np_ind.tuple) > 0: + array_indexing_adapter = indexing.NumpyIndexingAdapter(array) + if isinstance(np_ind, indexing.VectorizedIndexer): + array = array_indexing_adapter.vindex[np_ind] + elif isinstance(np_ind, indexing.OuterIndexer): + array = array_indexing_adapter.oindex[np_ind] + else: + array = array_indexing_adapter[np_ind] + + assert_array_equal(expected, array) + + def test_implicit_indexing_adapter() -> None: array = np.arange(10, dtype=np.int64) implicit = indexing.ImplicitToExplicitIndexingAdapter( From dfd25c7252e71a47e789aa24b8a06dd911a56461 Mon Sep 17 00:00:00 2001 From: Matthias Schabel Date: Mon, 28 Sep 2026 11:57:02 -0700 Subject: [PATCH 2/4] Fix broadcast with an index spanning several dimensions (#11615) Co-authored-by: Claude Opus 5.5 Co-authored-by: Matthias Schabel Co-authored-by: Deepak Cherian --- doc/whats-new.rst | 3 +++ xarray/structure/alignment.py | 8 +++++--- xarray/tests/test_dataarray.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/doc/whats-new.rst b/doc/whats-new.rst index 5540de966c6..dc15b4ccdc4 100644 --- a/doc/whats-new.rst +++ b/doc/whats-new.rst @@ -53,6 +53,9 @@ Deprecations Bug Fixes ~~~~~~~~~ +- Fix :py:func:`broadcast` failing on objects with an index spanning several + dimensions, such as a custom index set on both ``x`` and ``y`` (:pull:`11615`). + By `Matthias Schabel `_. - :py:meth:`DataArray.roll` and :py:meth:`Dataset.roll` now return unchanged empty results when rolling an empty dimension, including with ``roll_coords=True``, instead of raising ``ZeroDivisionError`` (:pull:`11613`). diff --git a/xarray/structure/alignment.py b/xarray/structure/alignment.py index 1a61afd3e31..eebceddde97 100644 --- a/xarray/structure/alignment.py +++ b/xarray/structure/alignment.py @@ -1136,10 +1136,12 @@ def _get_broadcast_dims_map_common_coords(args, exclude): dims_map = {} for arg in args: for dim in arg.dims: - if dim not in common_coords and dim not in exclude: + if dim in exclude: + continue + if dim not in dims_map: dims_map[dim] = arg.sizes[dim] - if dim in arg._indexes: - common_coords.update(arg.xindexes.get_all_coords(dim)) + if dim in arg._indexes and dim not in common_coords: + common_coords.update(arg.xindexes.get_all_coords(dim)) return dims_map, common_coords diff --git a/xarray/tests/test_dataarray.py b/xarray/tests/test_dataarray.py index f0dc2de2a3e..70e7cf6bf8a 100644 --- a/xarray/tests/test_dataarray.py +++ b/xarray/tests/test_dataarray.py @@ -66,6 +66,7 @@ requires_sparse, source_ndarray, ) +from xarray.tests.indexes import XYIndex try: from pandas.errors import UndefinedVariableError @@ -3598,6 +3599,35 @@ def test_broadcast_arrays(self) -> None: assert_identical(expected_x2, x2) assert_identical(expected_y2, y2) + def test_broadcast_arrays_multi_coordinate_index(self) -> None: + array = DataArray( + np.arange(6).reshape(2, 3), + dims=("y", "x"), + coords={"y": [0, 1], "x": [10, 20, 30]}, + ) + array = array.drop_indexes(["y", "x"]).set_xindex(["y", "x"], XYIndex) + other = DataArray([1, 2], dims="channel") + assert isinstance(array.xindexes["x"], XYIndex) + assert array.xindexes["x"] is array.xindexes["y"] + + result, other_result = broadcast(array, other) + + assert result.dims == ("y", "x", "channel") + assert_array_equal( + result.values, np.broadcast_to(array.values[..., None], result.shape) + ) + + other_x = DataArray([1, 2, 3], dims="x") + other_y = DataArray([1, 2], dims="y") + for broadcast_result in ( + result, + other_result, + *broadcast(other_x, other_y, array), + ): + assert set(broadcast_result.coords) == {"x", "y"} + assert_array_equal(broadcast_result.x, [10, 20, 30]) + assert_array_equal(broadcast_result.y, [0, 1]) + def test_broadcast_arrays_misaligned(self) -> None: # broadcast on misaligned coords must auto-align x = DataArray([[1, 2], [3, 4]], coords=[("a", [-1, -2]), ("b", [3, 4])]) From 84b9d3d7bfeb8b0bc3b041074ab69e79181539a2 Mon Sep 17 00:00:00 2001 From: Michael Niklas Date: Mon, 28 Sep 2026 23:46:04 +0200 Subject: [PATCH 3/4] Fix CombinedLock deadlocks & flaky tests (#11622) * Release held locks when a non-blocking CombinedLock.acquire fails * CI: time out hung test jobs instead of waiting 6 hours * Acquire the locks of a CombinedLock in a consistent order * Keep files open while in use when they are evicted from the file cache --------- Co-authored-by: Claude Co-authored-by: Deepak Cherian --- .github/workflows/ci.yaml | 6 +- doc/whats-new.rst | 12 ++- xarray/backends/file_manager.py | 76 ++++++++++++++++--- xarray/backends/h5netcdf_.py | 63 ++++++++-------- xarray/backends/locks.py | 40 ++++++++-- xarray/backends/netCDF4_.py | 46 +++++------ xarray/backends/scipy_.py | 2 +- xarray/tests/test_backends.py | 8 +- xarray/tests/test_backends_file_manager.py | 62 ++++++++++++++- xarray/tests/test_backends_locks.py | 88 +++++++++++++++++++++- 10 files changed, 326 insertions(+), 77 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index a48e8485352..cfacfd489ea 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -64,6 +64,8 @@ jobs: name: "${{ matrix.os }} | ${{ matrix.pixi-env }}${{ matrix.pytest-addopts && format(' ({0})', matrix.pytest-addopts) || '' }}" runs-on: ${{ matrix.os }} needs: cache-pixi-lock + # The slowest jobs take ~20 minutes; a hung test should not burn the 6h default. + timeout-minutes: 60 defaults: run: shell: bash -l {0} @@ -131,7 +133,9 @@ jobs: then if [[ "${{ matrix.pytest-addopts }}" == "flaky" ]] ; then - echo "PYTEST_ADDOPTS=-m 'flaky or network' --run-flaky --run-network-tests -W default" >> $GITHUB_ENV + # The thread timeout method also kills tests stuck in C code or deadlocked + # threads, which the default signal method cannot interrupt. + echo "PYTEST_ADDOPTS=-m 'flaky or network' --run-flaky --run-network-tests --timeout-method=thread -W default" >> $GITHUB_ENV elif [[ "${{ matrix.pytest-addopts }}" == "mypy" ]] ; then echo "PYTEST_ADDOPTS=-n 1 -m 'mypy' --run-mypy -W default" >> $GITHUB_ENV diff --git a/doc/whats-new.rst b/doc/whats-new.rst index dc15b4ccdc4..1cf9f1a386b 100644 --- a/doc/whats-new.rst +++ b/doc/whats-new.rst @@ -53,9 +53,20 @@ Deprecations Bug Fixes ~~~~~~~~~ +- Fix deadlocks when reading and writing netCDF files with dask at the same + time. Combined locks now always acquire their locks in the same order, which + previously depended on memory addresses, and a failed non-blocking acquire, + e.g. while garbage collecting an unclosed file, no longer leaves some of its + locks held forever (:pull:`11622`). + By `Michael Niklas `_. - Fix :py:func:`broadcast` failing on objects with an index spanning several dimensions, such as a custom index set on both ``x`` and ``y`` (:pull:`11615`). By `Matthias Schabel `_. +- Fix errors and crashes in :py:func:`open_mfdataset` with ``parallel=True`` + when opening more files than ``file_cache_maxsize``. Files evicted from the + file cache by another thread are no longer closed while they are still being + read (:pull:`11622`). + By `Michael Niklas `_. - :py:meth:`DataArray.roll` and :py:meth:`Dataset.roll` now return unchanged empty results when rolling an empty dimension, including with ``roll_coords=True``, instead of raising ``ZeroDivisionError`` (:pull:`11613`). @@ -64,7 +75,6 @@ Bug Fixes index instead of an empty index when the interval direction conflicts with the sign of ``step`` (:pull:`11623`). By `Ahmet Kamer Çivi `_. - - Fix async zarr tests using ``wraps`` with ``autospec=True`` on async methods, which caused ``AsyncMock`` objects to leak through instead of real array data (:pull:`11232`). diff --git a/xarray/backends/file_manager.py b/xarray/backends/file_manager.py index ec31b8aa04b..4dd20dab50c 100644 --- a/xarray/backends/file_manager.py +++ b/xarray/backends/file_manager.py @@ -14,12 +14,35 @@ from xarray.core.options import OPTIONS from xarray.core.types import Closable, Lock +# Files are pinned while in use inside CachingFileManager.acquire_context(). +# A pinned file evicted from FILE_CACHE by another thread is not closed right +# away, as that thread may still be reading from it. Instead it is parked in +# _EVICTED_PINNED until it is acquired again or the last user unpins it. +_PIN_COUNTS: dict[Any, int] = {} +_EVICTED_PINNED: dict[Any, Closable] = {} + + +def _close_unless_pinned(key: Any, file: Closable) -> None: + with _PIN_LOCK: + if _PIN_COUNTS.get(key): + _EVICTED_PINNED[key] = file + return + file.close() + + # Global cache for storing open files. FILE_CACHE: LRUCache[Any, Closable] = LRUCache( - maxsize=OPTIONS["file_cache_maxsize"], on_evict=lambda k, v: v.close() + maxsize=OPTIONS["file_cache_maxsize"], on_evict=_close_unless_pinned ) assert FILE_CACHE.maxsize, "file cache must be at least size one" +# Guards the pin state. It is the reentrant lock of FILE_CACHE, as evicting a +# file from the cache updates the pin state while holding the cache's lock, and +# garbage collection can run a manager's __del__, which closes its file and +# removes it from the cache, while a thread holds either. Separate locks could +# deadlock there. +_PIN_LOCK = FILE_CACHE._lock + T_File = TypeVar("T_File", bound=Closable) REF_COUNTS: dict[Any, int] = {} @@ -203,14 +226,37 @@ def acquire(self, needs_lock: bool = True) -> T_File: @contextmanager def acquire_context(self, needs_lock: bool = True) -> Iterator[T_File]: - """Context manager for acquiring a file.""" - file, cached = self._acquire_with_cache_info(needs_lock) + """Context manager for acquiring a file. + + The file stays open until the context exits, even if it is evicted + from the cache in the meantime. + """ + with self._pinned(): + file, cached = self._acquire_with_cache_info(needs_lock) + try: + yield file + except Exception: + if not cached: + self.close(needs_lock) + raise + + @contextmanager + def _pinned(self) -> Iterator[None]: + """Keep the file open while in use, even if it is evicted meanwhile.""" + with _PIN_LOCK: + _PIN_COUNTS[self._key] = _PIN_COUNTS.get(self._key, 0) + 1 try: - yield file - except Exception: - if not cached: - self.close(needs_lock) - raise + yield + finally: + with _PIN_LOCK: + count = _PIN_COUNTS.pop(self._key) - 1 + if count: + _PIN_COUNTS[self._key] = count + file = None + else: + file = _EVICTED_PINNED.pop(self._key, None) + if file is not None: + file.close() def _acquire_with_cache_info(self, needs_lock: bool = True) -> tuple[T_File, bool]: """Acquire a file, returning the file and whether it was cached.""" @@ -218,6 +264,13 @@ def _acquire_with_cache_info(self, needs_lock: bool = True) -> tuple[T_File, boo try: file = self._cache[self._key] except KeyError: + with _PIN_LOCK: + evicted = _EVICTED_PINNED.pop(self._key, None) + if evicted is not None: + # still open because it is in use, so reuse it + file = cast(T_File, evicted) + self._cache[self._key] = file + return file, True kwargs = self._kwargs if self._mode is not _OMIT_MODE: kwargs = kwargs.copy() @@ -238,8 +291,11 @@ def close(self, needs_lock: bool = True) -> None: with self._optional_lock(needs_lock): default = None file = self._cache.pop(self._key, default) - if file is not None: - file.close() + with _PIN_LOCK: + evicted = _EVICTED_PINNED.pop(self._key, None) + for f in (file, evicted): + if f is not None: + f.close() def __del__(self) -> None: # If we're the only CachingFileManger referencing an unclosed file, diff --git a/xarray/backends/h5netcdf_.py b/xarray/backends/h5netcdf_.py index 445c694c1fb..8e7174f5142 100644 --- a/xarray/backends/h5netcdf_.py +++ b/xarray/backends/h5netcdf_.py @@ -149,7 +149,8 @@ def __init__( self.format = format or "NETCDF4" # todo: utilizing find_root_and_group seems a bit clunky # making filename available on h5netcdf.Group seems better - self._filename = find_root_and_group(self.ds)[0].filename + with manager.acquire_context(): + self._filename = find_root_and_group(self.ds)[0].filename self.is_remote = is_remote_uri(self._filename) self.lock = ensure_lock(lock) self.autoclose = autoclose @@ -570,16 +571,17 @@ def open_dataset( store_entrypoint = StoreBackendEntrypoint() - ds = store_entrypoint.open_dataset( - store, - mask_and_scale=mask_and_scale, - decode_times=decode_times, - concat_characters=concat_characters, - decode_coords=decode_coords, - drop_variables=drop_variables, - use_cftime=use_cftime, - decode_timedelta=decode_timedelta, - ) + with store._manager.acquire_context(): + ds = store_entrypoint.open_dataset( + store, + mask_and_scale=mask_and_scale, + decode_times=decode_times, + concat_characters=concat_characters, + decode_coords=decode_coords, + drop_variables=drop_variables, + use_cftime=use_cftime, + decode_timedelta=decode_timedelta, + ) # only warn if phony_dims exist in file # remove together with the above check @@ -686,26 +688,27 @@ def open_groups_as_dict( manager = store._manager groups_dict = {} - for path_group in _iter_nc_groups(store.ds, parent=parent): - group_store = H5NetCDFStore(manager, group=path_group, **kwargs) - store_entrypoint = StoreBackendEntrypoint() - with close_on_error(group_store): - group_ds = store_entrypoint.open_dataset( - group_store, - mask_and_scale=mask_and_scale, - decode_times=decode_times, - concat_characters=concat_characters, - decode_coords=decode_coords, - drop_variables=drop_variables, - use_cftime=use_cftime, - decode_timedelta=decode_timedelta, - ) + with manager.acquire_context(): + for path_group in _iter_nc_groups(store.ds, parent=parent): + group_store = H5NetCDFStore(manager, group=path_group, **kwargs) + store_entrypoint = StoreBackendEntrypoint() + with close_on_error(group_store): + group_ds = store_entrypoint.open_dataset( + group_store, + mask_and_scale=mask_and_scale, + decode_times=decode_times, + concat_characters=concat_characters, + decode_coords=decode_coords, + drop_variables=drop_variables, + use_cftime=use_cftime, + decode_timedelta=decode_timedelta, + ) - if group: - group_name = str(NodePath(path_group).relative_to(parent)) - else: - group_name = str(NodePath(path_group)) - groups_dict[group_name] = group_ds + if group: + group_name = str(NodePath(path_group).relative_to(parent)) + else: + group_name = str(NodePath(path_group)) + groups_dict[group_name] = group_ds # only warn if phony_dims exist in file # remove together with the above check diff --git a/xarray/backends/locks.py b/xarray/backends/locks.py index 1c6d673cdad..52f13b5f302 100644 --- a/xarray/backends/locks.py +++ b/xarray/backends/locks.py @@ -207,6 +207,16 @@ def acquire(lock, blocking=True): return lock.acquire(blocking) +def _lock_identity(lock: Lock) -> int: + """Identity of the lock that is actually acquired. + + Unpickled SerializableLocks are new objects wrapping the same threading.Lock. + """ + if isinstance(lock, SerializableLock): + return id(lock.lock) + return id(lock) + + class CombinedLock(Lock): """A combination of multiple locks. @@ -215,22 +225,38 @@ class CombinedLock(Lock): """ def __init__(self, locks: Sequence[Lock]): - self.locks = tuple(set(locks)) # remove duplicates + # Remove duplicates and always acquire in one global order. If not careful, + # CombinedLocks sharing locks could acquire them in opposite orders and + # deadlock each other. + unique = {_lock_identity(lock): lock for lock in locks} + self.locks = tuple(lock for _, lock in sorted(unique.items())) + + def __reduce__(self): + # The order depends on the ids of the locks, which differ between + # processes, so sort again when unpickling. + return (type(self), (list(self.locks),)) def acquire(self, blocking=True): - return all(acquire(lock, blocking=blocking) for lock in self.locks) + acquired = [] + for lock in self.locks: + if not acquire(lock, blocking=blocking): + # Release the locks we already hold, otherwise a failed + # non-blocking acquire leaves them locked forever. + for held in reversed(acquired): + held.release() + return False + acquired.append(lock) + return True def release(self): - for lock in self.locks: + for lock in reversed(self.locks): lock.release() def __enter__(self): - for lock in self.locks: - lock.__enter__() + self.acquire() def __exit__(self, *args): - for lock in self.locks: - lock.__exit__(*args) + self.release() def locked(self): return any(lock.locked() for lock in self.locks) diff --git a/xarray/backends/netCDF4_.py b/xarray/backends/netCDF4_.py index af9804669c9..ef7971f4387 100644 --- a/xarray/backends/netCDF4_.py +++ b/xarray/backends/netCDF4_.py @@ -451,8 +451,9 @@ def __init__( self._manager = manager self._group = group self._mode = mode - self.format = self.ds.data_model - self._filename = self.ds.filepath() + with manager.acquire_context(): + self.format = self.ds.data_model + self._filename = self.ds.filepath() self.is_remote = is_remote_uri(self._filename) self.lock = ensure_lock(lock) self.autoclose = autoclose @@ -807,7 +808,7 @@ def open_dataset( ) store_entrypoint = StoreBackendEntrypoint() - with close_on_error(store): + with close_on_error(store), store._manager.acquire_context(): ds = store_entrypoint.open_dataset( store, mask_and_scale=mask_and_scale, @@ -908,25 +909,26 @@ def open_groups_as_dict( manager = store._manager groups_dict = {} - for path_group in _iter_nc_groups(store.ds, parent=parent): - group_store = NetCDF4DataStore(manager, group=path_group, **kwargs) - store_entrypoint = StoreBackendEntrypoint() - with close_on_error(group_store): - group_ds = store_entrypoint.open_dataset( - group_store, - mask_and_scale=mask_and_scale, - decode_times=decode_times, - concat_characters=concat_characters, - decode_coords=decode_coords, - drop_variables=drop_variables, - use_cftime=use_cftime, - decode_timedelta=decode_timedelta, - ) - if group: - group_name = str(NodePath(path_group).relative_to(parent)) - else: - group_name = str(NodePath(path_group)) - groups_dict[group_name] = group_ds + with manager.acquire_context(): + for path_group in _iter_nc_groups(store.ds, parent=parent): + group_store = NetCDF4DataStore(manager, group=path_group, **kwargs) + store_entrypoint = StoreBackendEntrypoint() + with close_on_error(group_store): + group_ds = store_entrypoint.open_dataset( + group_store, + mask_and_scale=mask_and_scale, + decode_times=decode_times, + concat_characters=concat_characters, + decode_coords=decode_coords, + drop_variables=drop_variables, + use_cftime=use_cftime, + decode_timedelta=decode_timedelta, + ) + if group: + group_name = str(NodePath(path_group).relative_to(parent)) + else: + group_name = str(NodePath(path_group)) + groups_dict[group_name] = group_ds return groups_dict diff --git a/xarray/backends/scipy_.py b/xarray/backends/scipy_.py index 9d5f33e8947..55df7bf9370 100644 --- a/xarray/backends/scipy_.py +++ b/xarray/backends/scipy_.py @@ -446,7 +446,7 @@ def open_dataset( ) store_entrypoint = StoreBackendEntrypoint() - with close_on_error(store): + with close_on_error(store), store._manager.acquire_context(): ds = store_entrypoint.open_dataset( store, mask_and_scale=mask_and_scale, diff --git a/xarray/tests/test_backends.py b/xarray/tests/test_backends.py index 10c7b746399..e94edeb1722 100644 --- a/xarray/tests/test_backends.py +++ b/xarray/tests/test_backends.py @@ -5461,12 +5461,14 @@ def tmp_store(request, tmp_path): @requires_dask @pytest.mark.filterwarnings("ignore:use make_scale(name) instead") -@pytest.mark.skip( - reason="Flaky test which can cause the worker to crash (so don't xfail). Very open to contributions fixing this" -) def test_open_mfdataset_manyfiles( readengine, nfiles, parallel, chunks, file_cache_maxsize ): + if readengine == "netcdf4" and parallel: + pytest.skip( + "netCDF4 reads metadata without holding the netCDF-C lock and can " + "crash the worker, see GH9779" + ) randdata = np.random.randn(nfiles) original = Dataset({"foo": ("x", randdata)}) # test standard open_mfdataset approach with too many files diff --git a/xarray/tests/test_backends_file_manager.py b/xarray/tests/test_backends_file_manager.py index 7b5fbee6309..189406e9a4e 100644 --- a/xarray/tests/test_backends_file_manager.py +++ b/xarray/tests/test_backends_file_manager.py @@ -7,7 +7,12 @@ import pytest -from xarray.backends.file_manager import CachingFileManager, PickleableFileManager +from xarray.backends import file_manager +from xarray.backends.file_manager import ( + CachingFileManager, + PickleableFileManager, + _close_unless_pinned, +) from xarray.backends.lru_cache import LRUCache from xarray.core.options import set_options from xarray.tests import assert_no_warnings @@ -264,6 +269,43 @@ class AcquisitionError(Exception): manager.close() +def test_file_manager_acquire_context_survives_eviction() -> None: + # A file evicted by another manager while in use must stay open until the + # context exits, and be reused rather than opened a second time. + cache: LRUCache = LRUCache(maxsize=1, on_evict=_close_unless_pinned) + file1, file2 = mock.Mock(), mock.Mock() + opener1 = mock.Mock(return_value=file1) + manager1 = CachingFileManager(opener1, "file1", cache=cache) + manager2 = CachingFileManager(mock.Mock(return_value=file2), "file2", cache=cache) + + with manager1.acquire_context() as f: + assert f is file1 + manager2.acquire() # evicts file1 + file1.close.assert_not_called() + assert manager1.acquire() is file1 # evicts file2 + file2.close.assert_called_once_with() + manager2.acquire() # evicts file1 again + file1.close.assert_called_once_with() + opener1.assert_called_once() + + manager1.close() + manager2.close() + + +def test_file_manager_close_while_pinned_and_evicted() -> None: + cache: LRUCache = LRUCache(maxsize=1, on_evict=_close_unless_pinned) + file1 = mock.Mock() + manager1 = CachingFileManager(mock.Mock(return_value=file1), "file1", cache=cache) + manager2 = CachingFileManager(mock.Mock(), "file2", cache=cache) + + with manager1.acquire_context(): + manager2.acquire() # evicts file1 + manager1.close() + file1.close.assert_called_once_with() + file1.close.assert_called_once_with() + manager2.close() + + def test_pickleable_file_manager_write_pickle(tmpdir) -> None: path = str(tmpdir.join("testing.txt")) manager = PickleableFileManager(open, path, mode="w") @@ -289,3 +331,21 @@ def test_pickleable_file_manager_preserves_closed(tmpdir) -> None: manager2 = pickle.loads(pickle.dumps(manager)) assert manager2._closed assert repr(manager2) == "" + + +def test_file_manager_del_while_pin_lock_held() -> None: + # Garbage collection can run a manager's __del__ while the same thread is + # updating the pin state, so closing from __del__ must not deadlock. + manager = CachingFileManager(mock.Mock(return_value=mock.Mock()), cache={}) + manager.acquire() + done = threading.Event() + + def delete_while_locked(): + nonlocal manager + with file_manager._PIN_LOCK: + del manager # runs __del__, which closes the file + done.set() + + thread = threading.Thread(target=delete_while_locked, daemon=True) + thread.start() + assert done.wait(timeout=5) diff --git a/xarray/tests/test_backends_locks.py b/xarray/tests/test_backends_locks.py index 8c1f8f6b12d..7d209ebd21b 100644 --- a/xarray/tests/test_backends_locks.py +++ b/xarray/tests/test_backends_locks.py @@ -1,9 +1,10 @@ from __future__ import annotations +import pickle import threading from xarray.backends import locks -from xarray.backends.locks import CombinedLock, SerializableLock +from xarray.backends.locks import CombinedLock, SerializableLock, combine_locks def test_threaded_lock() -> None: @@ -76,6 +77,26 @@ def test_combined_lock_locked_with_serializable_locks() -> None: assert combined.locked() is False +def test_combined_lock_failed_nonblocking_acquire_releases_held_locks() -> None: + """A failed non-blocking acquire must not leave any constituent lock held.""" + lock1 = SerializableLock() + lock2 = SerializableLock() + combined = CombinedLock([lock1, lock2]) + + # The acquisition order is an implementation detail, so try both. + for busy, other in [(lock1, lock2), (lock2, lock1)]: + busy.acquire() + try: + assert combined.acquire(blocking=False) is False + assert other.locked() is False + finally: + busy.release() + + assert combined.acquire(blocking=False) is True + combined.release() + assert combined.locked() is False + + def test_combined_lock_locked_with_context_manager() -> None: """CombinedLock.locked() should reflect state when using context manager.""" lock1 = threading.Lock() @@ -88,3 +109,68 @@ def test_combined_lock_locked_with_context_manager() -> None: assert combined.locked() is True assert combined.locked() is False + + +def test_combined_lock_order_independent_of_construction_order() -> None: + """Combined locks sharing locks must acquire them in the same order.""" + lock_a = SerializableLock() + lock_b = SerializableLock() + lock_c = threading.Lock() + + assert CombinedLock([lock_a, lock_b]).locks == CombinedLock([lock_b, lock_a]).locks + forward = CombinedLock([lock_a, lock_b, lock_c]).locks + backward = CombinedLock([lock_c, lock_b, lock_a]).locks + assert forward == backward + + +class _HashedLock(SerializableLock): + """SerializableLock with a fixed hash, to control set iteration order.""" + + def __init__(self, hash_value: int): + super().__init__() + self.hash_value = hash_value + + def __hash__(self) -> int: + return self.hash_value + + +def test_combined_lock_reader_and_writer_share_order() -> None: + """Mimic the netCDF4 backend: a reader lock and a writer lock built from it. + + Deduplicating with a set made the order depend on hashes (memory addresses), + so with these hash values a reader acquired the two shared locks in the + opposite order of the writer, and the two deadlocked each other (GH11622). + """ + netcdfc, hdf5, write_lock = _HashedLock(3), _HashedLock(11), _HashedLock(0) + reader = combine_locks([netcdfc, hdf5]) + writer = combine_locks([reader, write_lock]) + assert isinstance(reader, CombinedLock) + assert isinstance(writer, CombinedLock) + + shared = [lock for lock in writer.locks if lock in (netcdfc, hdf5)] + assert shared == list(reader.locks) + + +def test_combined_lock_deduplicates_unpickled_serializable_lock() -> None: + """An unpickled SerializableLock wraps the same lock and must not be held twice.""" + lock = SerializableLock() + copy = pickle.loads(pickle.dumps(lock)) + assert copy is not lock + assert copy.lock is lock.lock + + combined = CombinedLock([lock, copy]) + assert len(combined.locks) == 1 + with combined: + assert lock.locked() + assert not lock.locked() + + +def test_combined_lock_is_sorted_again_when_unpickled() -> None: + # The lock order depends on ids, which differ between processes, so a + # CombinedLock pickled in another process must be sorted again. + combined = CombinedLock([SerializableLock(), SerializableLock()]) + local_order = [locks._lock_identity(lock) for lock in combined.locks] + combined.locks = combined.locks[::-1] # as if sorted in another process + + unpickled = pickle.loads(pickle.dumps(combined)) + assert [locks._lock_identity(lock) for lock in unpickled.locks] == local_order From ad79ed8ee9d9d5d35cb4aea2dd743f6754f68df4 Mon Sep 17 00:00:00 2001 From: imam Date: Tue, 29 Sep 2026 04:37:21 +0600 Subject: [PATCH 4/4] Raise error in diff() for a non-existent dimension (#11628) Co-authored-by: dsk0425-sketch Co-authored-by: Claude --- doc/whats-new.rst | 5 +++++ xarray/core/dataset.py | 4 ++++ xarray/tests/test_dataarray.py | 7 +++++++ xarray/tests/test_dataset.py | 10 ++++++++++ 4 files changed, 26 insertions(+) diff --git a/doc/whats-new.rst b/doc/whats-new.rst index 1cf9f1a386b..4cf481f618e 100644 --- a/doc/whats-new.rst +++ b/doc/whats-new.rst @@ -142,6 +142,11 @@ Bug Fixes - Fixed a bug that caused rechunking a multi-dimensional cftime array along a subset of its dimensions to raise an error (:issue:`11567`, :pull:`11576`). By `Spencer Clark `_. +- :py:meth:`Dataset.diff` and :py:meth:`DataArray.diff` now raise a ``ValueError`` + when ``dim`` is not an existing dimension, instead of silently returning the + object unchanged. This matches the behavior of other methods such as + :py:meth:`Dataset.differentiate` and reductions like ``mean`` (:issue:`7748`). + By `imam2004i `_. - Fixed indexing with an empty indexer array. An empty indexer array is now always turned into an empty slice for the backend, so that the in-memory part of the decomposed indexer stays aligned with the axes of the loaded array. Previously the diff --git a/xarray/core/dataset.py b/xarray/core/dataset.py index 2706e5eea12..0a46ff6d5e0 100644 --- a/xarray/core/dataset.py +++ b/xarray/core/dataset.py @@ -8016,6 +8016,10 @@ def diff( -------- Dataset.differentiate """ + if dim not in self.dims: + raise ValueError( + f"Dimension {dim!r} not found in data dimensions {tuple(self.dims)}" + ) if n == 0: return self if n < 0: diff --git a/xarray/tests/test_dataarray.py b/xarray/tests/test_dataarray.py index 70e7cf6bf8a..3d3d388287a 100644 --- a/xarray/tests/test_dataarray.py +++ b/xarray/tests/test_dataarray.py @@ -4329,6 +4329,13 @@ def test_dataarray_diff_n1(self) -> None: expected = DataArray(np.diff(da.values, axis=1), dims=["x", "y"]) assert_equal(expected, actual) + def test_dataarray_diff_exception_invalid_dim(self) -> None: + # GH7748: diff along a non-existent dimension should raise instead of + # silently returning the array unchanged. + da = DataArray(np.arange(10), dims=["a"]) + with pytest.raises(ValueError, match=r"not found in data dimensions"): + da.diff("b") + def test_coordinate_diff(self) -> None: # regression test for GH634 arr = DataArray(range(0, 20, 2), dims=["lon"], coords=[range(10)]) diff --git a/xarray/tests/test_dataset.py b/xarray/tests/test_dataset.py index 1f622d9b05a..c9001c95141 100644 --- a/xarray/tests/test_dataset.py +++ b/xarray/tests/test_dataset.py @@ -7006,6 +7006,16 @@ def test_dataset_diff_exception_label_str(self) -> None: with pytest.raises(ValueError, match=r"'label' argument has to"): ds.diff("dim2", label="raise_me") # type: ignore[arg-type] + def test_dataset_diff_exception_invalid_dim(self) -> None: + # GH7748: diff along a non-existent dimension should raise instead of + # silently returning the object unchanged. + ds = create_test_data(seed=1) + with pytest.raises(ValueError, match=r"not found in data dimensions"): + ds.diff("not_a_dim") + # the check runs before the ``n == 0`` short-circuit + with pytest.raises(ValueError, match=r"not found in data dimensions"): + ds.diff("not_a_dim", n=0) + @pytest.mark.parametrize("fill_value", [dtypes.NA, 2, 2.0, {"foo": -10}]) def test_shift(self, fill_value) -> None: coords = {"bar": ("x", list("abc")), "x": [-4, 3, 2]}