[fix] use deepcopy for BatchMeta init

Co-authored-by: 0oshowero0<o0shower0o@outlook.com>



# message auto-generated for no-merge-commit merge:
!2 merge main into main

[fix] use deepcopy for BatchMeta init

Created-by: hanzhenyu8
Commit-by: 0oshowero0
Merged-by: ascend-robot
Description: ## Background

We nofity there might be some error during `BatchMeta.chunk()` due to shallow copy.

Reproduction script:
```python3
from transfer_queue.metadata import BatchMeta, FieldMeta, SampleMeta
from transfer_queue.utils.utils import ProductionStatus
import pickle

samples = [
    SampleMeta(
        partition_id="p1",
        global_index=0,
        fields={"image": FieldMeta("image", None, None)},
    ),
    SampleMeta(
        partition_id="p1",
        global_index=1,
        fields={"image": FieldMeta("image", None, None)},
    ),
    SampleMeta(
        partition_id="p1",
        global_index=2,
        fields={"image": FieldMeta("image", None, None)},
    ),
    SampleMeta(
        partition_id="p1",
        global_index=3,
        fields={"image": FieldMeta("image", None, None)},
    ),
]

extra_info = {"key1": [1, 2], "key2": {"nested": "value"}}
batch_meta = BatchMeta(samples=samples, extra_info=extra_info)


with open('before_chunk', "wb") as f:
    pickle.dump(batch_meta, f)


# Execute chunk
chunks = batch_meta.chunk(2)
import pickle
with open('after_chunk', "wb") as f:
    pickle.dump(batch_meta, f)
```

```bash
# You will see different md5 values!
md5sum before_chunk
md5sum after_chunk

```

This happens because during the `__post_init__` of `BatchMeta`, we set the attributes of the underlying `self.samples`:

```python3
class BatchMeta:
    """Records the metadata of a batch of data samples."""

    samples: list[SampleMeta]
    extra_info: dict[str, Any] = dataclasses.field(default_factory=dict)

    def __post_init__(self):
        """Initialize all computed properties during initialization"""
        # Basic properties
        object.__setattr__(self, "_size", len(self.samples))
        object.__setattr__(self, "_is_ready", all(sample.is_ready for sample in self.samples))

        # Pre-compute all list properties for better performance
        if self.samples:
            for idx, sample in enumerate(self.samples):
                object.__setattr__(sample, "_batch_index", idx)  # Ensure batch_index is set correctly
        ...
```

## Solution

During `__post_init__`, we deepcopy the input `samples` and `extra_info`.

```python3
    def __post_init__(self):
        """Initialize all computed properties during initialization"""
        self.samples = copy.deepcopy(self.samples)
        self.extra_info = copy.deepcopy(self.extra_info)

```

See merge request: Ascend/TransferQueue!2

Author: ascend-robot

transfer_queue/metadata.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
 import dataclasses
 import itertools
 import logging
@@ -176,6 +177,9 @@ class BatchMeta:
 
     def __post_init__(self):
         """Initialize all computed properties during initialization"""
+        self.samples = copy.deepcopy(self.samples)
+        self.extra_info = copy.deepcopy(self.extra_info)
+
         # Basic properties
         object.__setattr__(self, "_size", len(self.samples))
         object.__setattr__(self, "_is_ready", all(sample.is_ready for sample in self.samples))
@@ -253,7 +257,7 @@ def has_extra_info(self, key: str) -> bool:
 
     def get_all_extra_info(self) -> dict[str, Any]:
         """Get all extra info as a dictionary"""
-        return self.extra_info.copy()
+        return copy.deepcopy(self.extra_info)
 
     def add_fields(self, tensor_dict: TensorDict, set_all_ready: bool = True) -> "BatchMeta":
         """
@@ -296,7 +300,7 @@ def select_samples(self, sample_indices: list[int]) -> "BatchMeta":
         selected_samples = [self.samples[i] for i in sample_indices]
 
         # construct new BatchMeta instance
-        selected_batch_meta = BatchMeta(samples=selected_samples, extra_info=self.extra_info.copy())
+        selected_batch_meta = BatchMeta(samples=selected_samples, extra_info=self.extra_info)
 
         return selected_batch_meta
 
@@ -315,7 +319,7 @@ def select_fields(self, field_names: list[str]) -> "BatchMeta":
         new_samples = [sample.select_fields(field_names=field_names) for sample in self.samples]
 
         # construct new BatchMeta instance
-        new_batch_meta = BatchMeta(samples=new_samples, extra_info=self.extra_info.copy())
+        new_batch_meta = BatchMeta(samples=new_samples, extra_info=self.extra_info)
 
         return new_batch_meta
 
@@ -326,7 +330,7 @@ def __len__(self) -> int:
     def __getitem__(self, item):
         if isinstance(item, int | np.integer):
             sample_meta = self.samples[item] if self.samples else []
-            return BatchMeta(samples=[sample_meta], extra_info=self.extra_info.copy())
+            return BatchMeta(samples=[sample_meta], extra_info=self.extra_info)
         else:
             raise TypeError(f"Indexing with {type(item)} is not supported now!")
 
@@ -359,7 +363,7 @@ def chunk(self, chunks: int) -> list["BatchMeta"]:
             current_chunk_size = base_size + 1 if i < remainder else base_size
             end = start + current_chunk_size
             chunk_samples = self.samples[start:end]
-            chunk = BatchMeta(samples=chunk_samples, extra_info=self.extra_info.copy())
+            chunk = BatchMeta(samples=chunk_samples, extra_info=self.extra_info)
             chunk_list.append(chunk)
             start = end
         return chunk_list