fix: implement RowIdMeta.asdict() and from_dict() for stable row ID serialization (#6405)

Fixes #6403

`FragmentMetadata.to_json()` raised `NotImplementedError` when
`row_id_meta` was present (`enable_stable_row_ids=True`), because
`PyRowIdMeta.asdict()` was stubbed out. This also broke
`FragmentMetadata.from_json()` for the round-trip, since
`RowIdMeta(**dict)` doesn't work on a PyO3 class.

### Repro

```python
import lance, pyarrow as pa

uri = "/tmp/repro.lance"
ds = lance.write_dataset(pa.table({"x": [1, 2, 3]}), uri, enable_stable_row_ids=True)
ds.get_fragments()[0].metadata.to_json()  # NotImplementedError: PyRowIdMeta.asdict is not yet supported.
```

### Fix

- Implement `PyRowIdMeta.asdict()` via `pythonize` (Rust struct → Python
dict)
- Add `PyRowIdMeta.from_dict()` via `depythonize` (Python dict → Rust
struct)
- Update `FragmentMetadata.from_json()` to use `RowIdMeta.from_dict()`
instead of `RowIdMeta(**dict)`
- Add JSON round-trip test to the existing
`test_fragment_metadata_pickle` parametrized test

Author: pengw0048

python/python/lance/fragment.py

@@ -143,7 +143,7 @@ def from_json(json_data: str) -> FragmentMetadata:
 
         row_id_meta = json_data.get("row_id_meta")
         if row_id_meta is not None:
-            row_id_meta = RowIdMeta(**row_id_meta)
+            row_id_meta = RowIdMeta.from_dict(row_id_meta)
 
         created_at_version_meta = json_data.get("created_at_version_meta")
         if created_at_version_meta is not None:

python/python/tests/test_fragment.py

@@ -472,9 +472,16 @@ def test_fragment_metadata_pickle(tmp_path: Path, enable_stable_row_ids: bool):
 
     # Pickle and unpickle the fragment metadata
     round_trip = pickle.loads(pickle.dumps(frag_meta))
-
     assert frag_meta == round_trip
 
+    # JSON round-trip
+    json_data = frag_meta.to_json()
+    json_round_trip = FragmentMetadata.from_json(json.dumps(json_data))
+    assert frag_meta.id == json_round_trip.id
+    assert frag_meta.physical_rows == json_round_trip.physical_rows
+    if enable_stable_row_ids:
+        assert json_round_trip.row_id_meta is not None
+
 
 def test_deletion_file_with_base_id_serialization():
     """Test that DeletionFile with base_id serializes correctly."""

python/src/fragment.rs

@@ -627,10 +627,10 @@ pub struct PyRowDatasetVersionMeta(pub RowDatasetVersionMeta);
 
 #[pymethods]
 impl PyRowIdMeta {
-    fn asdict(&self) -> PyResult<Bound<'_, PyDict>> {
-        Err(PyNotImplementedError::new_err(
-            "PyRowIdMeta.asdict is not yet supported.s",
-        ))
+    fn asdict(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+        pythonize::pythonize(py, &self.0)
+            .map(|b| b.unbind())
+            .map_err(|err| PyValueError::new_err(format!("Could not convert RowIdMeta: {}", err)))
     }
 
     pub fn json(&self) -> PyResult<String> {
@@ -650,6 +650,13 @@ impl PyRowIdMeta {
         Ok(Self(row_id_meta))
     }
 
+    #[staticmethod]
+    pub fn from_dict(dict: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let row_id_meta: RowIdMeta = pythonize::depythonize(dict)
+            .map_err(|err| PyValueError::new_err(format!("Could not load RowIdMeta: {}", err)))?;
+        Ok(Self(row_id_meta))
+    }
+
     fn __reduce__(&self, py: Python<'_>) -> PyResult<(Py<PyAny>, Py<PyAny>)> {
         let state = self.json()?;
         let state = PyTuple::new(py, vec![state])?.extract()?;