[feat] Support multi-step kv_update with partial values hidden

[OutstanderWang]

Problem

kv_update (#182) merges new values into stored fields, which suits fields built in several steps, such as a response appended chunk by chunk during rollout. But two problems get in the way:

• Half-built values are visible. Every update marks its fields ready in the controller, so consumers see the field after the first chunk. That includes StreamingDataLoader / sampler fetches, kv_batch_get and kv_retrieve_meta, and they can read an intermediate value.
• The first step has to be a kv_put. kv_update rejects keys or fields that don't exist yet, so callers need a different code path for the first chunk.

Summary

• finalize on kv_update / kv_batch_update (and their async_ variants), default True. With finalize=False, the value is stored and its metadata recorded, but the controller keeps the cells not ready. The field stays hidden from every consumer until an update with finalize=True.
• kv_update creates missing keys and fields. A field the key doesn't hold yet is stored as given, without calling merge_fn. Unlike kv_put, it takes no tags or data_parser. kv_put and restore don't take finalize.
• Dump/restore keeps readiness. A dump records each row's unfinalized fields, and restore keeps them hidden. The dump format becomes version 4; version 3 dumps still load, with every field ready.

Usage

import torch
import transfer_queue as tq

concat = lambda old, new: torch.cat([old, new])

# The first chunk creates the key and field; they stay hidden.
tq.kv_update(key="s0", partition_id="train", fields={"response": torch.tensor([1, 2])},
             merge_fn=concat, finalize=False)
tq.kv_update(key="s0", partition_id="train", fields={"response": torch.tensor([3])},
             merge_fn=concat, finalize=False)

tq.kv_describe(keys="s0", partition_id="train")["s0"]["unfinalized"]  # ['response']
tq.kv_batch_get(keys="s0", partition_id="train", select_fields="response")  # ValueError: not finalized yet

# The last chunk publishes the whole value.
tq.kv_update(key="s0", partition_id="train", fields={"response": torch.tensor([4])}, merge_fn=concat)
# response -> tensor([1, 2, 3, 4]), now visible to kv_batch_get and the dataloader

Implementation

The path is async_kv_batch_update → TransferQueueController.kv_prepare_update → TransferQueueClient.async_update → AsyncSimpleStorageManager.update_data → StorageUnitData.apply_update on each owner unit → notify_data_update(ready=finalize) → DataPartitionStatus.update_production_status(ready=...).

• Readiness stays in the controller. The new ready flag in the update notification sets production_status to 0 or 1. Field metadata (dtype, shapes) is recorded either way. Storage never decides visibility, so the control-plane/data-plane split is unchanged.
• The controller decides which rows merge. kv_batch_update sends one KV_PREPARE_UPDATE request in place of kv_retrieve_meta. It registers missing keys and returns, for each updated field, the column type and a held mask aligned with the batch. The mask comes from FieldMeta.global_indexes, so hidden cells count as held. A storage unit may still hold values the controller dropped, e.g. after a failed clear, so it never decides on its own. The manager slices the mask per unit, the same way it slices the values.
• Merge, overwrite and type check in one write. Each unit merges the held rows and stores the others as given, overwriting any leftover. A held row the unit lacks raises. Every final value is checked against the column type (tensor/non-tensor kind and dtype). All merges and checks finish before the single put_data, so a failure leaves that unit unchanged and nothing is marked ready. kv_retrieve_meta and BatchMeta are unchanged from main.
• Dump/restore. kv_describe reports each row's unfinalized fields, and the dump stores them in the row index. On restore, each unit publishes finalized and unfinalized fields in separate updates with the matching ready flag.
• No silent results.
  • kv_batch_get and kv_batch_get_by_meta raise when a selected field isn't ready in every key; before, they returned an empty TensorDict.
  • New kv_describe / async_kv_describe return {key: {"fields", "unfinalized"}}: each key's fields, and the subset not ready yet.
  • kv_batch_update rejects duplicate keys before registering anything. A repeated missing key used to orphan rows, and a repeated existing key used to lose a merge.
  • update_production_status checks dtypes before writing anything. A conflicting write used to return False with its cells already flipped (ready, or hidden for ready=False) for rows FieldMeta didn't list.

Compatibility

• Existing kv_update calls keep their behavior, since finalize defaults to True. The one change: a missing key or field is now created instead of raising.
• kv_batch_get / kv_batch_get_by_meta with a select_fields entry that isn't ready in every key now raise ValueError instead of returning an empty result.
• Dumps are written as version 4. Builds that only read version 3 reject them instead of restoring hidden fields as ready.
• A dtype-conflicting notification leaves readiness unchanged, also for kv_put. Before, it was rejected but still marked its cells ready.
• kv_update remains SimpleStorage-only. KV backends still raise NotImplementedError.
• Client and storage-manager signatures. TransferQueueClient.(async_)update and StorageManager.update_data take a new required merge_plan argument. It comes from the new (async_)kv_prepare_update. The only in-tree caller is kv_batch_update.

Known limitation

Updating a field that is already ready happens in place. This covers finalize=True, kv_put overwrites, and finalize=False on a ready field, so re-hiding a ready field is not guaranteed. Readers that already hold ready metadata can see the new value, or a mix of old and new across storage units. The normal streaming flow isn't affected, because its fields stay hidden until the final update. See the review thread for why that window is narrow and why hiding first wouldn't close it.

Updates to one key must come from one writer at a time: readiness follows the last notification the controller receives, so interleaved writers can re-hide a field the other one finalized. Use kv_global_lock when several processes write the same key.

Tests

• Unit tests:
  • tests/test_kv_update.py: rows that aren't held are stored without merge_fn and overwrite local leftovers; a held row missing on the unit raises before writing; column-type checks run before writing; duplicate keys are rejected.
  • tests/test_controller.py: kv_prepare_update counts hidden cells as held, but not new keys or new fields.
  • tests/test_controller_data_partitions.py: an unfinalized write records metadata but stays hidden from scans; a dtype-conflicting write leaves readiness and FieldMeta unchanged.
  • tests/test_data_dump.py
• e2e (tests/e2e/test_kv_interface_e2e.py, tests/e2e/test_data_dump_e2e.py):
  • unfinalized updates stay hidden until finalized, including the kv_batch_get error and kv_describe;
  • missing keys and fields are created;
  • a new row must match a column type held only by another key's hidden write;
  • storage values the controller doesn't hold are overwritten, not merged: after a failed field clear, and when retrying a first chunk the controller never recorded;
  • restore keeps unfinalized fields hidden;
  • a version 3 dump restores as ready.
• pre-commit run --all-files and python -m pytest -q pass, with the Yuanrong e2e test excluded locally because of an environment-specific TORCH_LIBRARY registration error.

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-13T11:34:56.362785Z	936657f	PR opened

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 936657f949

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Hide fields before writing unfinalized payloads

When finalize=False updates a field that is currently ready, all storage-unit writes have already completed before this notification flips the controller's readiness. A concurrent kv_batch_get in that window can therefore receive ready metadata and read the intermediate payload; if the notification fails, the field can remain visible indefinitely. Hide the cells in the controller before dispatching the write (and account for already in-flight reads), then publish readiness only for a successful finalizing write.

AGENTS.md reference: AGENTS.md:L13-L14

Useful? React with 👍 / 👎.

[OutstanderWang]

Thanks. The window is real, but narrow, so I'm leaving the write order as it is in this PR.

• The flow finalize is meant for has no window. The first finalize=False chunk writes a field whose cells are already not ready (new, or still hidden), so nothing becomes visible early. The controller sets the ready bit only on the finalizing write, after every storage unit has committed.
• A window only exists in two cases. (a) A caller uses finalize=False to hide a field that is currently ready again. (b) The controller notification fails after storage committed. In case (b) the call raises "Storage update committed, but the controller did not accept its metadata" instead of failing silently.
• Hiding first wouldn't close the gap. Readers that already hold ready metadata still read the new value; you point this out yourself ("account for already in-flight reads"). The same exposure exists for any in-place update of a ready field: finalize=True updates, and the existing kv_put overwrite on main. In both, readers can see the new value or an old/new mix across units. A pre-hide RPC would add a round trip and a new state transition but only shrink the window.

Reader isolation for in-place updates would need versioned or leased reads, which is out of scope here. If it becomes a requirement, I'll handle it in a separate PR.

[chatgpt-codex-connector]

Validate missing-row values against the partition schema

When an update adds field f to a row while f already exists elsewhere in the partition with another tensor dtype, present is empty for this row, so its incoming value is stored as-is and only merged rows are type-checked. After the payload is committed, _build_update_field_schema reports the new dtype and the controller rejects it against FieldMeta, but the controller's readiness assignment has already run; the call raises while the conflicting value can be marked ready under the stale schema. Prevalidate missing-row values against the controller's partition schema before writing them.

AGENTS.md reference: AGENTS.md:L13-L14

Useful? React with 👍 / 👎.

[OutstanderWang]

Fixed in 3891aa4. kv_update now passes include_fields to kv_retrieve_meta, so the metadata carries the column schema for every updated field the partition already knows, including unfinalized fields and fields only some keys hold. The manager sends these column_types to each unit, and StorageUnitData.apply_update checks every new row's tensor/non-tensor kind and dtype before the single put_data. A conflict now fails before anything is stored or marked ready. Covered by test_new_row_must_match_column_type_held_by_other_keys (e2e) and test_apply_update_checks_new_rows_against_column_type_before_writing.

[chatgpt-codex-connector]

Reject duplicate keys before creating missing rows

When keys repeats a previously absent key (for example, ['k', 'k']), create=True makes the controller allocate two global indexes and then overwrite keys_mapping['k'] with the second one while retaining both global and reverse mappings. Both rows are written and become eligible for sampling, but only the second remains reachable by key or clear operations, leaking capacity and potentially emitting the orphaned first value; reject duplicate batch keys before registration.

Useful? React with 👍 / 👎.

[OutstanderWang]

Fixed in 63a751d. async_kv_batch_update now raises ValueError on repeated keys before any controller or storage call, so nothing is registered or written. This also covers a repeated existing key: there, both merges would read the same old value and one would be silently lost. Covered by test_batch_update_rejects_duplicate_keys_before_registering. kv_batch_put has the same create-path behavior on main, so a controller-side fix would be a separate change.

[chatgpt-codex-connector]

Bump the dump version for unfinalized readiness

When a dump contains an unfinalized field and is loaded by the immediately previous build, it still advertises DUMP_FORMAT_VERSION = 3, so that reader accepts it but omits unfinalized from load records and publishes every restored field as ready. A downgrade or mixed-version restore therefore silently exposes hidden intermediate data; write a new format version while allowing the current reader to treat legacy v3 dumps as fully finalized.

Useful? React with 👍 / 👎.

[OutstanderWang]

Fixed in 3891aa4. DUMP_FORMAT_VERSION is now 4. This build still reads version 3 dumps, which have no unfinalized, so every field restores as ready. Builds that only read version 3 reject a version 4 dump instead of exposing hidden fields. Covered by test_restores_a_version_3_dump_as_ready; docs/data_dump.md is updated.

[0oshowero0]

[AI Review] kv_update merges into storage values the controller no longer tracks

Problem

StorageUnitData.apply_update chooses between merging and storing from the unit's local state (idx in stored, simple_storage.py#L262-L277). But storage can hold values the controller has already dropped. AsyncSimpleStorageManager.clear_data logs per-unit failures and returns normally (simple_storage_manager.py#L918-L922). The controller then forgets the cell, and for kv_clear(keys) it releases the global index into the pool shared by all partitions.

On main these leftovers are unreachable: kv_update requires keys and fields the controller knows, and kv_put overwrites. With this PR, the next kv_update folds the new value into them:

• Same key: kv_clear(fields="response") fails in storage. kv_describe then lists only prompt, yet kv_update(response=[1]) stores [100, 101, 1].
• Another partition: kv_clear of old in pa fails in storage. The first kv_update of a new key in pb reuses that global index and stores [200, 201, 9], i.e. the other partition's payload.

The same rule makes retrying a first chunk unsafe: after a partial multi-unit failure, the units that already committed merge the chunk a second time.

This is separate from the in-place visibility window discussed here: the stored value itself is wrong, and it stays wrong for every later reader.

Repro (SimpleStorage with one storage unit, after tq.init):

from transfer_queue import interface

def concat(old, new):
    return torch.cat([old, new])

async def fail_clear(*_args, **_kwargs):
    raise RuntimeError("simulated storage unit timeout")

manager = interface._maybe_create_tq_client().storage_manager
real_clear = manager._clear_single_storage_unit

tq.kv_put(key="k", partition_id="p", fields={"response": torch.tensor([100, 101]), "prompt": torch.tensor([7])})
manager._clear_single_storage_unit = fail_clear
tq.kv_clear(keys="k", partition_id="p", fields="response")  # returns normally; the failure is only logged
manager._clear_single_storage_unit = real_clear

tq.kv_update(key="k", partition_id="p", fields={"response": torch.tensor([1])}, merge_fn=concat)
tq.kv_batch_get(keys="k", partition_id="p", select_fields="response")["response"][0]  # tensor([100, 101, 1])

Suggestion: let the controller decide which rows to merge

In async_kv_batch_update, replace kv_retrieve_meta(create=True, include_fields=...) with one controller request specific to kv_update, e.g. kv_prepare_update(keys, partition_id, fields). In a single actor call it registers missing keys and returns the usual BatchMeta. For each updated field it also returns the column type and the batch rows that hold the field. These rows come from FieldMeta.global_indexes, which also covers hidden finalize=False cells.

def kv_prepare_update(self, keys: list[str], partition_id: str, fields: list[str]):
    """Register missing keys and tell kv_update, per field, which rows to merge into."""
    metadata = self.kv_retrieve_meta(keys, partition_id, create=True)
    partition = self._get_partition(partition_id)
    batch_rows = set(metadata.global_indexes)
    columns = {}
    for name in fields:
        meta = partition.field_metadata.get(name)
        columns[name] = {
            "dtype": meta.dtype if meta else None,
            "is_non_tensor": meta.is_non_tensor if meta else None,
            "present": sorted(meta.global_indexes & batch_rows) if meta else [],
        }
    return metadata, columns

The manager passes each unit its slice of present, and apply_update merges exactly those rows:

held = set(columns[field]["present"])
present = [i for i, idx in enumerate(global_indexes) if idx in held]  # was: idx in stored

Example:

• s0 (index 3) holds a hidden response=[1, 2].
• s1 (index 4) holds only prompt, but its unit keeps a leftover response=[200, 201].
• s2 is new and is registered as index 7.

For a batch update of response = [3], [5], [9], the controller returns {"response": {"dtype": torch.int64, "is_non_tensor": False, "present": [3]}}. Row 3 merges to [1, 2, 3], row 7 stores [9], and row 4 stores [5] over the leftover instead of [200, 201, 5].

What this changes:

• Leftovers are overwritten, never merged.
• Retrying a first chunk whose write committed but was never recorded by the controller now overwrites instead of merging twice. Retries of later chunks still must not be blind.
• If the controller holds a row that the unit lacks (e.g. it was cleared concurrently), the update fails with StorageKeyNotFoundError instead of storing the value as new.

Cost:

• No extra RPC: the request replaces the existing kv_retrieve_meta call.
• The controller does one set lookup per row per updated field, which is tens of microseconds for a few thousand rows. The mask is a few KB.
• include_fields and the manager's column_types construction go away. kv_retrieve_meta keeps its "fields ready in every row" meaning, and BatchMeta is unchanged.
• The return value is unaffected, because apply_field_schema already adds the updated fields to field_names.

[OutstanderWang]

Agreed, the unit's local state can't decide what to merge. Fixed in bfdbd7a.

• One controller request instead of kv_retrieve_meta. kv_batch_update now sends KV_PREPARE_UPDATE, which registers missing keys and returns, per field, the column type and a held mask aligned with the batch. The mask comes from FieldMeta.global_indexes, so hidden finalize=False cells count as held. The manager slices the mask by batch_positions, like the values.
• Units merge exactly the held rows. apply_update overwrites every other row, so leftovers from a failed clear (same key or a reused index from another partition) are overwritten, never merged. A retried first chunk that storage committed but the controller never recorded is rewritten, not merged twice. A held row the unit lacks raises StorageKeyNotFoundError before put_data.
• Simpler type check. Every final value is now checked against the controller's column type, which replaces the separate check against the old value. include_fields and the manager's column_types are gone, so kv_retrieve_meta and BatchMeta match main again.

Tests:

• test_storage_values_the_controller_does_not_hold_are_overwritten (e2e): your same-key repro (now [1], was [100, 101, 1] before this commit) and a first-chunk retry after the controller rejects the notification.
• test_apply_update_overwrites_a_local_value_the_controller_does_not_hold and test_apply_update_rejects_held_row_missing_locally_before_writing (unit).
• test_controller_kv_prepare_update_reports_held_rows_from_field_metadata: a hidden cell is held; a new key and a new field are not.

I didn't add the cross-partition case as an e2e test, because which index gets reused depends on the shared pool's order. It's the same code path as the unit-level overwrite test. clear_data swallowing unit failures is unchanged on main; with this fix it only leaks space, so I'll leave it for a separate PR.

[OutstanderWang]

Follow-up in 33ac43d:

• README: the kv_update paragraph now says that the controller's record of which rows hold a field (hidden ones included) decides what merges. Every other row overwrites any storage leftover.
• kv_update docstring: says the same thing. It also now lists merge and type-check failures as the RuntimeError callers actually get; it used to say TypeError.
• PR description: Implementation, Compatibility and Tests are updated. Compatibility now notes the new required columns argument on TransferQueueClient.(async_)update and StorageManager.update_data.

[OutstanderWang]

On the root cause: AsyncSimpleStorageManager.clear_data collects per-unit failures with gather(return_exceptions=True) and only logs them. That is pre-existing behavior on main, not something this PR introduces, so it should be fixed in a separate PR against main. This PR doesn't change it.

The two problems are independent:

• This PR: bfdbd7a makes kv_update stop trusting the unit's local state, so storage leftovers are overwritten and can no longer corrupt merged values.
• The main issue: after bfdbd7a, a failed clear only leaks storage space, but kv_clear still reports success while storage still holds the data. It also affects kv_clear and clear_samples without any kv_update involved.

Fixing that means changing the error contract of clear_data and every caller's handling of it, so it needs its own review.

[0oshowero0]

[AI Review] Narrow what the public kv_describe returns

async_kv_describe returns the controller's row records unchanged: {"global_index", "fields", "unfinalized", "tag"}.

• global_index is internal. Every other KV API addresses rows by key, and an index is reused across partitions once its key is cleared, so callers should not hold on to it. Once it is public, it is hard to take back.
• tag duplicates kv_list.
• unfinalized is inferred, not recorded. It lists the fields the row holds whose production_status is 0 (controller.py#L1846-L1850). But kv_clear(keys) also zeroes the whole row in mark_clearing (controller.py#L1540), before storage is cleared and the metadata is dropped. A kv_describe in that window reports every field of the row as unfinalized, and a dump taken in that window restores those rows as hidden. The window is narrow, but it is a reason not to promise more than "fields that are not ready yet".

Suggestion: keep the controller and client shape for the dump, and return only what callers need from the public API. The tests only read fields and unfinalized.

description = await tq_client.async_kv_describe(list(dict.fromkeys(keys)), partition_id)
return {key: {"fields": row["fields"], "unfinalized": row["unfinalized"]} for key, row in description["rows"].items()}

[OutstanderWang]

Fixed in 9830d9b. The public kv_describe / async_kv_describe now return {key: {"fields", "unfinalized"}}, and the docstring describes unfinalized only as "the subset that is not ready yet". The dump still reads the full controller rows through client.kv_describe, so it keeps global_index and tag. The e2e test now asserts the whole dict, so an extra key would fail it.

[0oshowero0]

[AI Review] Document that chunked updates to one key need a single writer

Readiness follows the last notification the controller receives. When two writers update the same key, storage commits and controller notifications can interleave. A late finalize=False notification then hides a field that the other writer already finalized, and it stays hidden until someone finalizes it again. The chunk order in the merged value also depends on commit order. This paragraph only describes the single-writer flow.

Suggestion: add one sentence, e.g. "Updates to one key must come from one writer at a time; use kv_global_lock if several processes write the same key."

Related: test_unfinalized_updates_stay_hidden_until_finalize asserts that finalize=False hides a field that is already ready (test_kv_interface_e2e.py#L1283-L1287). That is the in-place path that the Known limitation says readers can observe mid-update. Asserting it in a test turns it into a contract. I'd drop that part of the test, or document re-hiding as not guaranteed.

[OutstanderWang]

Fixed in deb3275. The README now says updates to one key must come from one writer at a time, with kv_global_lock for several processes. I dropped the re-hide part of test_unfinalized_updates_stay_hidden_until_finalize and changed the finalize docstring: it no longer says fields "become" hidden, and states that hiding an already ready field is not guaranteed for readers that already hold its metadata. The Known limitation in the PR description now says the same and includes the single-writer rule.

[0oshowero0]

[AI Review] kv_update is not a superset of kv_put yet

This line, the kv_update docstring (interface.py#L652) and the PR description call kv_update a superset of kv_put. Three gaps remain:

• kv_update takes no tags.
• It takes no data_parser.
• It only runs on SimpleStorage, while kv_put runs on every backend.

A caller who replaces kv_put with kv_update on that promise loses these.

Being a real superset sounds like a good goal, and it doesn't need to block this PR. The gaps could be closed in follow-up PRs, tracked in one issue so the order is visible: tags first, then data_parser, then the KV backends. Until then, the README and the docstring could name what is still missing, e.g. "a superset of kv_put except tags, data_parser and non-SimpleStorage backends (follow-up: #...)".

[OutstanderWang]

Fixed in 2179afb, by dropping the claim rather than tracking the gaps. The README and the kv_update docstring no longer call it a superset of kv_put. The docstring says it takes no tags or data_parser, and the README entry already says SimpleStorage only. The PR description is updated the same way. If we decide to close the gaps later, that can be its own issue.

[0oshowero0]

[AI Review] Check the dtype before writing readiness (please fix in this PR)

update_production_status writes production_status before _update_field_metadata, and the latter raises on a dtype conflict. The call then returns False, but the cells are already flipped (ready for ready=True, hidden for ready=False), and FieldMeta does not list the row:

p = DataPartitionStatus(partition_id="p")
f32 = {"x": {"dtype": torch.float32, "shape": (2,), "is_nested": False, "is_non_tensor": False}}
i64 = {"x": {"dtype": torch.int64, "shape": (2,), "is_nested": False, "is_non_tensor": False}}
assert p.update_production_status([0], ["x"], f32)
assert not p.update_production_status([1], ["x"], i64)
p.production_status[1, p.field_name_mapping["x"]]  # 1: ready, but row 1 is not in FieldMeta["x"]

This ordering already exists on main (controller.py#L546-L549 at a55e3dd), and a plain kv_put with another dtype reaches it. This PR's column-type check already protects kv_update whenever the controller knows the column. The remaining cases are concurrent first writes of a new field with different dtypes, and kv_put.

This PR adds the ready=False path through this function, so could you fix it here? Run the dtype check that FieldMeta.update does before anything is written. Then a conflicting write changes nothing, not even the row capacity or the field mapping. For example:

        try:
            for name, incoming in field_schema.items():
                existing = self.field_metadata.get(name)
                dtype = incoming.get("dtype")
                if existing is not None and None not in (existing.dtype, dtype) and existing.dtype != dtype:
                    raise ValueError(f"Field {name!r} dtype mismatch: {existing.dtype} != {dtype}")

Please don't reuse validate_field_schema here. It also rejects a tensor/non-tensor mix within one column, which puts support: with it, test_dense_field_with_a_wrapped_later_row_roundtrips and test_chunks_with_a_wrapped_last_row_roundtrip in tests/e2e/test_data_dump_e2e.py fail. Please run that file with the fix.

A regression test could put x as float32 on rows 0 and 1, then check that a conflicting int64 write with ready=False to row 1 returns False and leaves row 1 ready.

[OutstanderWang]

Fixed in f0234cb. update_production_status now runs the same dtype check as FieldMeta.update before it touches anything, so a conflicting write returns False and leaves readiness, row capacity, the field mapping and FieldMeta unchanged. It also covers a multi-field schema where a later field conflicts. Tensor/non-tensor mixing stays allowed, so I didn't reuse validate_field_schema; tests/e2e/test_data_dump_e2e.py passes (45/45). Regression test: test_dtype_conflict_leaves_readiness_unchanged puts pv as float32 on rows 0 and 1, then checks that an int64 write with ready=False to row 1 returns False and row 1 stays ready. It fails without the fix.

[0oshowero0]

[AI Review] nit: rename columns to merge_plan

Next to values, which is a TensorDict (that is, the data columns), columns reads like more column data or a list of column names. What it actually carries is the controller's plan for this update:

• held decides which rows go through merge_fn.
• dtype and is_non_tensor check the results.

merge_plan says this at every call site, e.g. apply_update(global_indexes, new_data, merge_fn, merge_plan), and it pairs with kv_prepare_update. held can stay as it is.

The rename is mechanical. It covers the parameters of update, update_data, _update_to_single_storage_unit and apply_update, the "columns" key in the unit request, and the tests.

[OutstanderWang]

Done in 16b2180. columns is now merge_plan in the parameters of update, update_data, _update_to_single_storage_unit and apply_update, in the unit request and the KV_PREPARE_UPDATE response, and in the tests. The PR description's Compatibility section is updated too.