fix: address PR Ascend#39 review comments (round 1 & 2)

Round 1 (Copilot Ascend#5, Ascend#9, Ascend#10, Ascend#17, Ascend#20):
- fix(serial_utils): preserve scalar shape in round-trip serialization
- fix(zmq_utils): add strict=True to zip calls
- fix(metadata): use warnings.warn for parse_dtype fallback
- refactor(metadata): remove dead _convert_legacy_sample_meta code
- fix(simple_backend): downgrade storage unit log to DEBUG

Round 2 (Copilot Ascend#24, 0oshowero0 Ascend#27, Ascend#28):
- perf(simple_backend): track active keys with _active_keys set for O(1)
  capacity check, replacing O(K×F) existing_keys scan in put_data
- docs(tutorial): merge demonstrate_batch_meta_construction() and
  demonstrate_batch_meta() into demonstrate_batch_meta_operations(),
  eliminating 3 duplicate demos and fixing misleading function name

Signed-off-by: 看我72遍 <m.pb@msn.com>

Author: mpb159753

tests/test_metadata.py

@@ -368,6 +368,44 @@ def test_chunk_concat_roundtrip_preserves_extra_info(self):
         assert len(restored) == 6
         assert restored.global_indexes == list(range(6))
 
+    def test_to_dict_from_dict_scalar_shape_roundtrip(self):
+        """to_dict/from_dict must preserve scalar tensor shape () (empty tuple).
+
+        () is falsy in Python; `if meta.get('shape')` would incorrectly
+        serialize it as None. The fix uses `is not None`.
+        """
+        batch = BatchMeta(
+            global_indexes=[0],
+            partition_ids=["p0"],
+            field_schema={
+                "scalar_field": {
+                    "dtype": torch.float32,
+                    "shape": (),  # scalar shape — falsy!
+                    "is_nested": False,
+                    "is_non_tensor": False,
+                }
+            },
+            production_status=np.ones(1, dtype=np.int8),
+        )
+        d = batch.to_dict()
+        # shape must be serialized as [] (empty list), NOT None
+        assert d["field_schema"]["scalar_field"]["shape"] == []
+
+        restored = BatchMeta.from_dict(d)
+        # shape must be restored as () (empty tuple), NOT None
+        assert restored.field_schema["scalar_field"]["shape"] == ()
+
+    def test_parse_dtype_unknown_string_logs_warning(self, caplog):
+        """_parse_dtype must log a warning when returning a raw string fallback."""
+        import logging
+
+        from transfer_queue.metadata import _parse_dtype
+
+        with caplog.at_level(logging.WARNING, logger="transfer_queue.metadata"):
+            result = _parse_dtype("<class 'int'>")
+        assert result == "<class 'int'>"
+        assert "Unknown dtype string" in caplog.text
+
 
 # ==============================================================================
 # KVBatchMeta Tests (all migrated from main with no modification)
@@ -674,3 +712,21 @@ def test_kv_batch_meta_concat_extra_info_conflict_raises(self):
         )
         with pytest.raises(ValueError, match="conflicting"):
             KVBatchMeta.concat([kv1, kv2])
+
+
+# ==============================================================================
+# StorageUnitData Tests
+# ==============================================================================
+
+
+class TestStorageUnitDataStrict:
+    """Tests for StorageUnitData length validation."""
+
+    def test_put_data_length_mismatch_raises(self):
+        """put_data must raise when global_indexes and field values have different lengths."""
+        from transfer_queue.storage.simple_backend import StorageUnitData
+
+        sud = StorageUnitData(storage_size=10)
+        # 3 indexes but only 2 values — must raise, not silently drop
+        with pytest.raises(ValueError, match="length mismatch"):
+            sud.put_data({"field_a": [1, 2]}, global_indexes=[0, 1, 2])

tests/test_simple_storage_unit.py

@@ -467,3 +467,54 @@ def test_storage_unit_data_partial_consume_safety():
     storage.put_data({"f": [torch.tensor([9.0])]}, global_indexes=[1])
     torch.testing.assert_close(storage.field_data["f"][0], torch.tensor([0.0]))
     torch.testing.assert_close(storage.field_data["f"][1], torch.tensor([9.0]))
+
+
+def test_storage_unit_data_active_keys_tracking():
+    """_active_keys must track put/clear lifecycle without scanning field_data."""
+    from transfer_queue.storage.simple_backend import StorageUnitData
+
+    storage = StorageUnitData(storage_size=10)
+
+    # Initially empty
+    assert hasattr(storage, "_active_keys"), "StorageUnitData must have _active_keys attribute"
+    assert storage._active_keys == set()
+
+    # After put, _active_keys must include the new keys
+    storage.put_data(
+        {"f1": [1, 2, 3], "f2": ["a", "b", "c"]},
+        global_indexes=[10, 20, 30],
+    )
+    assert storage._active_keys == {10, 20, 30}
+
+    # After clearing some keys, _active_keys must shrink
+    storage.clear(keys=[20])
+    assert storage._active_keys == {10, 30}
+
+    # Re-put overlapping key should not duplicate
+    storage.put_data({"f1": [99], "f2": ["z"]}, global_indexes=[10])
+    assert storage._active_keys == {10, 30}
+
+    # After clearing all, _active_keys must be empty
+    storage.clear(keys=[10, 30])
+    assert storage._active_keys == set()
+
+
+def test_storage_unit_data_capacity_uses_active_keys():
+    """Capacity check must use _active_keys, not scan field_data."""
+    from transfer_queue.storage.simple_backend import StorageUnitData
+
+    storage = StorageUnitData(storage_size=3)
+
+    # Fill to capacity
+    storage.put_data({"f": [1, 2, 3]}, global_indexes=[0, 1, 2])
+    assert len(storage._active_keys) == 3
+
+    # Exceeding capacity must raise
+    with pytest.raises(ValueError, match="Storage capacity exceeded"):
+        storage.put_data({"f": [4]}, global_indexes=[3])
+
+    # After clearing one key, adding one more should succeed
+    storage.clear(keys=[2])
+    assert len(storage._active_keys) == 2
+    storage.put_data({"f": [4]}, global_indexes=[3])
+    assert storage._active_keys == {0, 1, 3}

transfer_queue/metadata.py

@@ -62,6 +62,7 @@ def _parse_dtype(dtype_str: str) -> Any:
     except TypeError:
         pass
     # Fallback: return as-is (e.g. plain Python type repr like "<class 'int'>")
+    logger.warning("Unknown dtype string '%s', returning as-is", dtype_str)
     return dtype_str
 
 
@@ -744,7 +745,7 @@ def to_dict(self) -> dict:
             dtype = meta.get("dtype")
             serialized_schema[field_name] = {
                 "dtype": str(dtype) if dtype is not None else None,
-                "shape": list(meta["shape"]) if meta.get("shape") else None,
+                "shape": list(meta["shape"]) if meta.get("shape") is not None else None,
                 "is_nested": meta.get("is_nested", False),
                 "is_non_tensor": meta.get("is_non_tensor", False),
             }
@@ -773,7 +774,7 @@ def from_dict(cls, data: dict) -> "BatchMeta":
             dtype = _parse_dtype(dtype_str) if dtype_str is not None else None
             field_schema[field_name] = {
                 "dtype": dtype,
-                "shape": tuple(meta["shape"]) if meta.get("shape") else None,
+                "shape": tuple(meta["shape"]) if meta.get("shape") is not None else None,
                 "is_nested": meta.get("is_nested", False),
                 "is_non_tensor": meta.get("is_non_tensor", False),
             }

transfer_queue/storage/simple_backend.py

@@ -61,6 +61,8 @@ def __init__(self, storage_size: int):
         self.field_data: dict[str, dict] = {}
         # Capacity upper bound (not pre-allocated list length)
         self.storage_size = storage_size
+        # Track active global_index keys for O(1) capacity checks
+        self._active_keys: set = set()
 
     def get_data(self, fields: list[str], global_indexes: list) -> dict[str, list]:
         """Get data by global index keys.
@@ -92,20 +94,23 @@ def put_data(self, field_data: dict[str, Any], global_indexes: list) -> None:
             global_indexes: Global indexes to use as dict keys.
         """
         # Capacity is enforced per unique sample key, not counted per-field
-        existing_keys: set = set()
-        for fd in self.field_data.values():
-            existing_keys.update(fd.keys())
-        new_global_keys = [k for k in global_indexes if k not in existing_keys]
-        if len(existing_keys) + len(new_global_keys) > self.storage_size:
+        new_global_keys = [k for k in global_indexes if k not in self._active_keys]
+        if len(self._active_keys) + len(new_global_keys) > self.storage_size:
             raise ValueError(
-                f"Storage capacity exceeded: {len(existing_keys)} existing + "
+                f"Storage capacity exceeded: {len(self._active_keys)} existing + "
                 f"{len(new_global_keys)} new > {self.storage_size}"
             )
         for f, values in field_data.items():
+            if len(values) != len(global_indexes):
+                raise ValueError(
+                    f"StorageUnitData put_data: field '{f}' values length {len(values)} "
+                    f"!= global_indexes length {len(global_indexes)}, length mismatch"
+                )
             if f not in self.field_data:
                 self.field_data[f] = {}
-            for key, val in zip(global_indexes, values, strict=False):
+            for key, val in zip(global_indexes, values, strict=True):
                 self.field_data[f][key] = val
+        self._active_keys.update(global_indexes)
 
     def clear(self, keys: list[int]) -> None:
         """Remove data at given global index keys, immediately freeing memory.
@@ -116,6 +121,7 @@ def clear(self, keys: list[int]) -> None:
         for f in self.field_data:
             for key in keys:
                 self.field_data[f].pop(key, None)
+        self._active_keys -= set(keys)
 
 
 @ray.remote(num_cpus=1)

transfer_queue/utils/serial_utils.py

@@ -411,7 +411,7 @@ def encode_with_fallback(obj: Any) -> list[bytestr]:
     try:
         return list(_encoder.encode(obj))
     except (TypeError, ValueError) as e:
-        logger.info(
+        logger.debug(
             "encode_with_fallback: msgpack failed (%s), falling back to pickle.",
             type(e).__name__,
         )

transfer_queue/utils/zmq_utils.py

@@ -182,9 +182,6 @@ def deserialize(cls, frames: list) -> "ZMQMessage":
             raise ValueError("Empty frames received")
 
         result = decode_with_fallback(frames)
-        # Pickle fallback path: serialize() pickled the ZMQMessage directly.
-        if isinstance(result, cls):
-            return result
         return cls(
             request_type=ZMQRequestType(result["request_type"]),
             sender_id=result["sender_id"],