Skip to content

Commit 4b9f82c

Browse files
d-v-bclaude
andcommitted
fix(group): store a group's documents only; keep upgraded members as stored
Group writes (attributes, `update_attributes`, deleting a member, `consolidate_metadata`, `create_hierarchy`) read each upgraded member of the consolidated metadata again from its own document and stored its upgrade. Those member writes raced concurrent deletions: two concurrent `delitem`s on a consolidated group recreated a deleted array's metadata document in most runs, which zarr 3.4.0 never did. A group write now stores only the group's own documents and reads none. Array metadata read from a document that had to be upgraded keeps that document (`_stored_document`, set by `mark_upgraded`, replacing the `_stored_document_upgraded` flag), and consolidated metadata stores such a member as it was stored. Every reader of the consolidated metadata then reads the member as upgraded again, and the array's own first chunk write stores the upgrade of its current document (or refuses a changed chunk grid) as before: the only write that upgrades a member document. Removed: the refresh and adoption of consolidated members in `save_metadata` (`_refresh_consolidated`, `_refresh_array`, `_Refreshed`), and `Group.update_attributes_async`'s detour through `save_metadata`. Tests pinning the refresh are replaced by tests that a group write reads nothing and writes only the group's documents, stores the member's consolidated copy byte for byte as stored, that concurrent deletions leave no member, and that the first write through consolidated metadata stores the member's upgrade or refuses a changed grid. Assisted-by: ClaudeCode:claude-opus-5-5 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent cddb877 commit 4b9f82c

9 files changed

Lines changed: 137 additions & 226 deletions

File tree

‎changes/4334.bugfix.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,4 +4,4 @@ A chunk edge length is now always at least 1, while an array extent may be 0. Ev
44

55
Stored metadata with a regular chunk size of 0 or JSON `false` (Zarr format 2 `chunks`, Zarr format 3 `regular` `chunk_shape`) is now read as one chunk spanning the axis (a multiple of the inner chunk size for sharded arrays, which previously failed to open). zarr-python wrote such sizes for arrays created with a zero-length axis until 3.4, and, from 2.18.7 to 3.2.1, for an explicit chunk size of 0 or `False` on an axis of any length, as in `chunks=(0,)` (those arrays could store no data). On a zero-length axis this opens silently, as before; appending to such an axis then stores chunks of size 1. On an axis of positive length it opens with a `ZarrUserWarning`, which says that the axis holds only the fill value and how to store valid metadata: `array.update_attributes({})`, then `zarr.consolidate_metadata` if the metadata is consolidated. JSON `true`, as zarr-python 3.0 and 3.2 wrote for a chunk size of `True`, is read as 1, silently, in the chunk grid (regular, and the explicit edges and run-length encoded sizes of a rectilinear one) and in the inner chunk shape of every sharding codec, nested or not. A stored rectilinear chunk grid whose edge lengths are integral JSON floats (`[[4.0, 2]]`), as zarr-python 3.2 wrote for float edges, is read with those edges as integers, silently; a float anywhere no release wrote one (a regular chunk shape, the inner chunk shape of a sharding codec, a run-length repeat count, an edge below 1) is rejected.
66

7-
Writing data to an array read from such metadata (other than an empty selection) first stores the upgrade of the metadata the store then holds, unless another writer has stored valid metadata since, so that other readers find the chunks written; in a `ZipStore` this adds a second entry for the metadata document, as every metadata update does. If the metadata the store then holds lays out chunks differently (another writer resized the array keeping the chunk size of 0, say), the write raises a `ValueError` asking to reopen the array, and stores nothing. Storing a group's metadata, which `zarr.consolidate_metadata` and every change to a group with consolidated metadata do, likewise first stores the upgrade of each such member's metadata and consolidates the member's metadata as the store then holds it.
7+
Writing data to an array read from such metadata (other than an empty selection) first stores the upgrade of the metadata the store then holds, unless another writer has stored valid metadata since, so that other readers find the chunks written; in a `ZipStore` this adds a second entry for the metadata document, as every metadata update does. If the metadata the store then holds lays out chunks differently (another writer resized the array keeping the chunk size of 0, say), the write raises a `ValueError` asking to reopen the array, and stores nothing. This is the only operation that stores the upgrade: a group's consolidated metadata keeps such an array's metadata as it was stored (`zarr.consolidate_metadata` copies it as the array's own document holds it), and changing a group reads and writes no metadata of its members.

‎src/zarr/core/array.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1644,7 +1644,7 @@ async def _store_upgraded_document(self) -> None:
16441644
upgrade. Storing the same upgrade twice is harmless, so concurrent callers need
16451645
no coordination.
16461646
"""
1647-
if not self.metadata._stored_document_upgraded:
1647+
if self.metadata._stored_document is None:
16481648
return
16491649
zarr_format = self.metadata.zarr_format
16501650
documents = await read_documents(self.store_path, ARRAY_DOCUMENTS[zarr_format])
@@ -1658,9 +1658,9 @@ async def _store_upgraded_document(self) -> None:
16581658
f"The metadata stored for the array at {str(self.store_path)!r} has "
16591659
"changed since this array was opened: reopen the array to write to it."
16601660
)
1661-
if current._stored_document_upgraded:
1661+
if current._stored_document is not None:
16621662
await upsert_metadata(self.store_path, current, documents)
1663-
object.__setattr__(self.metadata, "_stored_document_upgraded", False)
1663+
object.__setattr__(self.metadata, "_stored_document", None)
16641664

16651665
async def _set_selection(
16661666
self,

‎src/zarr/core/group.py‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
import zarr.api.asynchronous as async_api
1515
from zarr.abc.metadata import Metadata
16+
from zarr.abc.store import Store, set_or_delete
1617
from zarr.core._info import GroupInfo
1718
from zarr.core._json import buffer_to_json_object, json_to_buffer
1819
from zarr.core.array import (
@@ -74,7 +75,6 @@
7475
)
7576
from typing import Any
7677

77-
from zarr.abc.store import Store
7878
from zarr.core.array_spec import ArrayConfigLike
7979
from zarr.core.buffer import Buffer, BufferPrototype
8080
from zarr.core.chunk_key_encodings import ChunkKeyEncodingLike
@@ -147,11 +147,16 @@ class ConsolidatedMetadata:
147147
must_understand: Literal[False] = False
148148

149149
def to_dict(self) -> dict[str, JSON]:
150+
"""The consolidated metadata document. An array read from a stored document that
151+
had to be upgraded is written as that document was stored: only the array's
152+
own first chunk write stores its upgrade."""
150153
return {
151154
"kind": self.kind,
152155
"must_understand": self.must_understand,
153156
"metadata": {
154157
k: v.to_dict()
158+
if isinstance(v, GroupMetadata) or v._stored_document is None
159+
else dict(v._stored_document)
155160
for k, v in sorted(
156161
self.flattened_metadata.items(),
157162
key=lambda item: (
@@ -2115,7 +2120,9 @@ async def update_attributes_async(self, new_attributes: dict[str, Any]) -> Group
21152120
new_metadata = replace(self.metadata, attributes=new_attributes)
21162121

21172122
# Write new metadata
2118-
await save_metadata(self.store_path, new_metadata)
2123+
to_save = new_metadata.to_buffer_dict(default_buffer_prototype())
2124+
awaitables = [set_or_delete(self.store_path / key, value) for key, value in to_save.items()]
2125+
await asyncio.gather(*awaitables)
21192126

21202127
async_group = replace(self._async_group, metadata=new_metadata)
21212128
return replace(self, _async_group=async_group)
@@ -3032,8 +3039,6 @@ async def create_hierarchy(
30323039
```{'': GroupMetadata, 'a': GroupMetadata, 'b': Groupmetadata}```
30333040
30343041
After input parsing, this function then creates all the nodes in the hierarchy concurrently.
3035-
The metadata of each node is stored as given: the consolidated metadata of a group is
3036-
stored as it is, without reading the documents of its members.
30373042
30383043
Arrays and Groups are yielded in the order they are created. This order is not stable and
30393044
should not be relied on.

‎src/zarr/core/metadata/io.py‎

Lines changed: 3 additions & 96 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
from __future__ import annotations
22

33
import asyncio
4-
from dataclasses import replace
54
from enum import Enum
65
from itertools import zip_longest
76
from typing import TYPE_CHECKING, Final, NamedTuple
@@ -12,7 +11,6 @@
1211
from zarr.core.buffer.cpu import buffer_prototype as cpu_buffer_prototype
1312
from zarr.core.common import ZARR_JSON, ZARRAY_JSON, ZATTRS_JSON
1413
from zarr.core.metadata.upgrades import mark_upgraded, upgrade_array_document
15-
from zarr.core.metadata.v3 import RectilinearChunksDisabledError
1614
from zarr.errors import ArrayNotFoundError, ContainsArrayError
1715
from zarr.storage._common import StorePath, ensure_no_existing_node
1816

@@ -134,87 +132,7 @@ def parse_stored_array(documents: Mapping[str, Buffer], zarr_format: ZarrFormat)
134132
else:
135133
raise ArrayNotFoundError(f"No Zarr format {zarr_format} array metadata document.")
136134
upgraded, readings = upgrade_array_document(stored, zarr_format)
137-
return mark_upgraded(parse_array_metadata(dict(upgraded)), [None] * len(readings), None)
138-
139-
140-
async def _refresh_array(store_path: StorePath, member: ArrayMetadata) -> ArrayMetadata:
141-
"""The metadata of the array document stored at `store_path` as it now is, after
142-
storing its upgrade if it needs one, for `member`, a member of consolidated metadata
143-
that was read from a document that had to be upgraded; `member` itself if no array
144-
document that can be read is stored there (it was deleted, or replaced by a group)."""
145-
documents = await read_documents(store_path, ARRAY_DOCUMENTS[member.zarr_format])
146-
try:
147-
current = parse_stored_array(documents, member.zarr_format)
148-
except RectilinearChunksDisabledError:
149-
raise # A document that can be read, but only with the flag.
150-
except (KeyError, TypeError, ValueError):
151-
return member
152-
if current._stored_document_upgraded:
153-
await upsert_metadata(store_path, current, documents)
154-
object.__setattr__(current, "_stored_document_upgraded", False)
155-
return current
156-
157-
158-
class _Refreshed(NamedTuple):
159-
"""A member of consolidated metadata that `_refresh_consolidated` read again from the
160-
member's own document."""
161-
162-
members: dict[str, ArrayMetadata | GroupMetadata]
163-
"""The consolidated metadata that holds the member, by member name."""
164-
name: str
165-
stale: ArrayMetadata
166-
current: ArrayMetadata
167-
168-
def adopt(self) -> None:
169-
"""Replace the stale member with the current one in place, so every handle that
170-
shares this consolidated metadata sees it; unless the member was deleted or
171-
replaced since it was read."""
172-
if self.members.get(self.name) is self.stale:
173-
self.members[self.name] = self.current
174-
175-
176-
async def _refresh_consolidated(
177-
store_path: StorePath, metadata: GroupMetadata
178-
) -> tuple[GroupMetadata, list[_Refreshed]]:
179-
"""`metadata` with each member of its consolidated metadata that was read from a
180-
document that had to be upgraded, at any depth, replaced by the metadata of the
181-
member's own document as it now is (see `_refresh_array`), read concurrently, and
182-
the members it replaced: the member document may have changed since, so the
183-
consolidated copy is never stored as if it were valid. `metadata` is left as it is,
184-
for the group to adopt the members (see `_Refreshed.adopt`) once they are stored."""
185-
from zarr.core.group import GroupMetadata
186-
187-
consolidated = metadata.consolidated_metadata
188-
if consolidated is None:
189-
return metadata, []
190-
groups = {
191-
name: member
192-
for name, member in consolidated.metadata.items()
193-
if isinstance(member, GroupMetadata)
194-
}
195-
arrays = {
196-
name: member
197-
for name, member in consolidated.metadata.items()
198-
if not isinstance(member, GroupMetadata) and member._stored_document_upgraded
199-
}
200-
nested, current = await asyncio.gather(
201-
asyncio.gather(*(_refresh_consolidated(store_path / n, m) for n, m in groups.items())),
202-
asyncio.gather(*(_refresh_array(store_path / n, m) for n, m in arrays.items())),
203-
)
204-
refreshed = [
205-
_Refreshed(consolidated.metadata, name, stale, new)
206-
for (name, stale), new in zip(arrays.items(), current, strict=True)
207-
if new is not stale
208-
]
209-
members = dict(consolidated.metadata)
210-
members.update({member.name: member.current for member in refreshed})
211-
for name, (group, inner) in zip(groups, nested, strict=True):
212-
members[name] = group
213-
refreshed.extend(inner)
214-
if not refreshed:
215-
return metadata, []
216-
consolidated = replace(consolidated, metadata=members)
217-
return replace(metadata, consolidated_metadata=consolidated), refreshed
135+
return mark_upgraded(parse_array_metadata(dict(upgraded)), stored, [None] * len(readings), None)
218136

219137

220138
def _build_parents(store_path: StorePath, zarr_format: ZarrFormat) -> dict[str, GroupMetadata]:
@@ -239,9 +157,7 @@ def _build_parents(store_path: StorePath, zarr_format: ZarrFormat) -> dict[str,
239157
async def save_metadata(
240158
store_path: StorePath, metadata: ArrayMetadata | GroupMetadata, ensure_parents: bool = False
241159
) -> None:
242-
"""Asynchronously save the array or group metadata. A group adopts, in place, the
243-
members of its consolidated metadata that had to be read again to store it (see
244-
`_refresh_consolidated`), once they are stored.
160+
"""Asynchronously save the array or group metadata.
245161
246162
Parameters
247163
----------
@@ -256,14 +172,8 @@ async def save_metadata(
256172
------
257173
ValueError
258174
"""
259-
from zarr.core.group import GroupMetadata
260-
261-
refreshed: list[_Refreshed] = []
262-
if isinstance(metadata, GroupMetadata):
263-
# The one place group metadata is stored, and with it consolidated metadata.
264-
metadata, refreshed = await _refresh_consolidated(store_path, metadata)
265175
to_save = metadata.to_buffer_dict(default_buffer_prototype())
266-
set_awaitables = [store_documents(store_path, to_save)]
176+
set_awaitables = [set_or_delete(store_path / key, value) for key, value in to_save.items()]
267177

268178
if ensure_parents:
269179
# To enable zarr.create(store, path="a/b/c"), we need to create all the intermediate groups.
@@ -301,6 +211,3 @@ async def save_metadata(
301211
) from e
302212

303213
await asyncio.gather(*set_awaitables)
304-
# Only once stored: a group whose store failed keeps the members flagged, to read again.
305-
for member in refreshed:
306-
member.adopt()

‎src/zarr/core/metadata/upgrades.py‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
from __future__ import annotations
2020

21+
import copy
2122
import json
2223
import warnings
2324
from collections.abc import Callable, Iterable, Mapping, Sequence
@@ -43,14 +44,17 @@
4344
)
4445

4546

46-
def mark_upgraded[M](metadata: M, readings: Sequence[str | None], path: str | None) -> M:
47-
"""Record that `metadata` was read from a stored document that needed the upgrades
48-
whose `readings` `upgrade_array_document` returned, if any: set
49-
`_stored_document_upgraded` on `metadata`, so the array stores the upgrade before
50-
it writes chunks under it, and warn once with the readings that are warnings,
51-
naming the array at `path` when the caller knows it."""
47+
def mark_upgraded[M](
48+
metadata: M, stored: ArrayDocument, readings: Sequence[str | None], path: str | None
49+
) -> M:
50+
"""Record that `metadata` was read from the document `stored`, which needed the
51+
upgrades whose `readings` `upgrade_array_document` returned, if any: keep a copy of
52+
`stored` as `_stored_document` on `metadata`, so the array stores the upgrade before
53+
it writes chunks under it and consolidated metadata stores it as it was stored, and
54+
warn once with the readings that are warnings, naming the array at `path` when the
55+
caller knows it."""
5256
if readings:
53-
object.__setattr__(metadata, "_stored_document_upgraded", True)
57+
object.__setattr__(metadata, "_stored_document", copy.deepcopy(stored))
5458
if messages := [reading for reading in readings if reading is not None]:
5559
subject = "" if path is None else f"Array {path!r}: "
5660
# The synchronous API parses metadata on zarr's IO thread, whose stack holds no

‎src/zarr/core/metadata/v2.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
TBaseScalar,
2626
ZDType,
2727
)
28+
from zarr.core.metadata.upgrades import ArrayDocument
2829

2930
from dataclasses import dataclass, field, fields, replace
3031

@@ -72,10 +73,9 @@ class ArrayV2Metadata(Metadata):
7273
compressor: Numcodec | None
7374
attributes: dict[str, JSON] = field(default_factory=dict)
7475
zarr_format: Literal[2] = field(init=False, default=2)
75-
_stored_document_upgraded: ClassVar[bool] = False
76-
"""Whether `from_dict` read this metadata from a stored document it had to upgrade
77-
(set on the instance by `mark_upgraded`), so the store may still hold that invalid
78-
document."""
76+
_stored_document: ClassVar[ArrayDocument | None] = None
77+
"""The stored document `from_dict` read this metadata from, if it had to upgrade it
78+
(set on the instance by `mark_upgraded`): the store may still hold it."""
7979

8080
def __init__(
8181
self,
@@ -209,7 +209,7 @@ def from_dict(cls, data: dict[str, Any], *, path: str | None = None) -> ArrayV2M
209209

210210
_data = {k: v for k, v in _data.items() if k in expected}
211211

212-
return mark_upgraded(cls(**_data), readings, path)
212+
return mark_upgraded(cls(**_data), data, readings, path)
213213

214214
def to_dict(self) -> dict[str, JSON]:
215215
zarray_dict = super().to_dict()

‎src/zarr/core/metadata/v3.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
from zarr.core.buffer import Buffer, BufferPrototype
5353
from zarr.core.chunk_grids import ChunkGrid
5454
from zarr.core.dtype.wrapper import TBaseDType, TBaseScalar
55+
from zarr.core.metadata.upgrades import ArrayDocument
5556

5657

5758
def parse_zarr_format(data: object) -> Literal[3]:
@@ -472,10 +473,9 @@ class ArrayV3Metadata(Metadata):
472473
node_type: Literal["array"] = field(default="array", init=False)
473474
storage_transformers: tuple[dict[str, JSON], ...]
474475
extra_fields: dict[str, AllowedExtraField]
475-
_stored_document_upgraded: ClassVar[bool] = False
476-
"""Whether `from_dict` read this metadata from a stored document it had to upgrade
477-
(set on the instance by `mark_upgraded`), so the store may still hold that invalid
478-
document."""
476+
_stored_document: ClassVar[ArrayDocument | None] = None
477+
"""The stored document `from_dict` read this metadata from, if it had to upgrade it
478+
(set on the instance by `mark_upgraded`): the store may still hold it."""
479479

480480
def __init__(
481481
self,
@@ -676,7 +676,7 @@ def from_dict(cls, data: dict[str, JSON], *, path: str | None = None) -> Self:
676676
extra_fields=allowed_extra_fields,
677677
storage_transformers=_data_typed.get("storage_transformers", ()), # type: ignore[arg-type]
678678
)
679-
return mark_upgraded(metadata, readings, path)
679+
return mark_upgraded(metadata, data, readings, path)
680680

681681
def to_dict(self) -> dict[str, JSON]:
682682
out_dict = super().to_dict()

‎tests/test_array_stateful.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ def _open(self) -> zarr.Array[Any]:
172172
warned = any(issubclass(w.category, ZarrUserWarning) for w in record)
173173
must_warn = any(self.shape[axis] > 0 for axis in self.legacy_axes)
174174
assert warned is must_warn, [str(w.message) for w in record]
175-
assert arr.metadata._stored_document_upgraded is bool(self.legacy_axes)
175+
assert (arr.metadata._stored_document is not None) is bool(self.legacy_axes)
176176
return arr
177177

178178
# ----------------------------------------------------------------- model

0 commit comments

Comments
 (0)