spatialrust_records/
migrate.rs

1//! Explicit schema migration for spatial records.
2
3use spatialrust_core::{PointBuffer, PointBufferSet, PointCloud, PointField};
4
5use crate::{
6    compare_schemas, CompatVerdict, RecordsError, RecordsResult, SchemaDescriptor, SpatialRecord,
7};
8
9/// How to fill fields missing from the source cloud.
10#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
11pub enum FieldFill {
12    /// Fill numeric columns with zeros.
13    #[default]
14    Zeros,
15}
16
17/// Migration permissions when projecting a record onto a target schema.
18#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
19pub struct MigrationPolicy {
20    /// Drop source fields absent from the target.
21    pub drop_unknown: bool,
22    /// Fill target fields absent from the source.
23    pub fill_missing: Option<FieldFill>,
24}
25
26impl Default for MigrationPolicy {
27    fn default() -> Self {
28        Self { drop_unknown: true, fill_missing: Some(FieldFill::Zeros) }
29    }
30}
31
32/// Projects `record` onto `target` under an explicit migration policy.
33pub fn migrate_record(
34    record: &SpatialRecord,
35    target: &SchemaDescriptor,
36    policy: MigrationPolicy,
37) -> RecordsResult<SpatialRecord> {
38    let report = compare_schemas(target, record.schema());
39    match report.verdict {
40        CompatVerdict::Identical => {
41            // Field order may still differ; always rebuild against `target`.
42        }
43        CompatVerdict::BackwardCompatible => {
44            if !policy.drop_unknown {
45                return Err(RecordsError::SchemaMismatch(
46                    "source has extra fields; enable drop_unknown to migrate".into(),
47                ));
48            }
49        }
50        CompatVerdict::ForwardCompatible => {
51            if policy.fill_missing.is_none() {
52                return Err(RecordsError::SchemaMismatch(
53                    "source is missing fields; configure fill_missing to migrate".into(),
54                ));
55            }
56        }
57        CompatVerdict::Incompatible => {
58            return Err(RecordsError::SchemaMismatch(format!(
59                "cannot migrate incompatible schemas: missing={:?} extra={:?} conflict={:?}",
60                report.missing_in_actual, report.extra_in_actual, report.conflicting
61            )));
62        }
63    }
64
65    let len = record.cloud().len();
66    let mut buffers = PointBufferSet::new();
67    for field in target.point_schema().fields() {
68        if let Ok(source) = record.cloud().field(&field.name) {
69            buffers.insert(field.name.clone(), clone_buffer(source)?);
70        } else {
71            let fill = policy
72                .fill_missing
73                .ok_or_else(|| RecordsError::MissingField(field.name.clone()))?;
74            buffers.insert(field.name.clone(), filled_buffer(field, len, fill)?);
75        }
76    }
77    let cloud = PointCloud::try_from_parts(
78        target.point_schema().clone(),
79        buffers,
80        record.metadata().clone(),
81    )?;
82    SpatialRecord::try_new_with_provenance(target.clone(), cloud, record.provenance().clone())
83}
84
85fn clone_buffer(buffer: &PointBuffer) -> RecordsResult<PointBuffer> {
86    Ok(match buffer {
87        PointBuffer::F32(values) => PointBuffer::F32(values.clone()),
88        PointBuffer::F64(values) => PointBuffer::F64(values.clone()),
89        PointBuffer::U8(values) => PointBuffer::U8(values.clone()),
90        PointBuffer::U16(values) => PointBuffer::U16(values.clone()),
91        PointBuffer::U32(values) => PointBuffer::U32(values.clone()),
92        PointBuffer::I32(values) => PointBuffer::I32(values.clone()),
93    })
94}
95
96fn filled_buffer(field: &PointField, len: usize, fill: FieldFill) -> RecordsResult<PointBuffer> {
97    match fill {
98        FieldFill::Zeros => {
99            let mut buffer = PointBuffer::with_capacity(field.dtype, len);
100            buffer.resize(len, field)?;
101            Ok(buffer)
102        }
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::{migrate_record, FieldFill, MigrationPolicy};
109    use crate::{SchemaDescriptor, SchemaVersion, SpatialRecord};
110    use spatialrust_core::{
111        PointBuffer, PointBufferSet, PointCloud, SpatialMetadata, StandardSchemas,
112    };
113
114    fn xyz_record() -> SpatialRecord {
115        let mut buffers = PointBufferSet::new();
116        buffers.insert("x", PointBuffer::from_f32(vec![1.0, 2.0]));
117        buffers.insert("y", PointBuffer::from_f32(vec![3.0, 4.0]));
118        buffers.insert("z", PointBuffer::from_f32(vec![5.0, 6.0]));
119        let cloud = PointCloud::try_from_parts(
120            StandardSchemas::point_xyz(),
121            buffers,
122            SpatialMetadata::default(),
123        )
124        .unwrap();
125        SpatialRecord::try_from_cloud("point", SchemaVersion::new(1, 0), cloud).unwrap()
126    }
127
128    #[test]
129    fn migrates_xyz_to_xyzi_with_zero_fill() {
130        let source = xyz_record();
131        let target = SchemaDescriptor::try_new(
132            "point",
133            SchemaVersion::new(1, 1),
134            StandardSchemas::point_xyzi(),
135        )
136        .unwrap();
137        let migrated = migrate_record(
138            &source,
139            &target,
140            MigrationPolicy { drop_unknown: true, fill_missing: Some(FieldFill::Zeros) },
141        )
142        .unwrap();
143        assert_eq!(migrated.cloud().field("intensity").unwrap().as_f32().unwrap(), &[0.0, 0.0]);
144        assert_eq!(migrated.schema().version.minor, 1);
145    }
146
147    #[test]
148    fn drops_intensity_when_targeting_xyz() {
149        let mut buffers = PointBufferSet::new();
150        buffers.insert("x", PointBuffer::from_f32(vec![1.0]));
151        buffers.insert("y", PointBuffer::from_f32(vec![2.0]));
152        buffers.insert("z", PointBuffer::from_f32(vec![3.0]));
153        buffers.insert("intensity", PointBuffer::from_f32(vec![9.0]));
154        let cloud = PointCloud::try_from_parts(
155            StandardSchemas::point_xyzi(),
156            buffers,
157            SpatialMetadata::default(),
158        )
159        .unwrap();
160        let source =
161            SpatialRecord::try_from_cloud("point", SchemaVersion::new(1, 1), cloud).unwrap();
162        let target = SchemaDescriptor::try_new(
163            "point",
164            SchemaVersion::new(1, 0),
165            StandardSchemas::point_xyz(),
166        )
167        .unwrap();
168        let migrated = migrate_record(&source, &target, MigrationPolicy::default()).unwrap();
169        assert!(migrated.cloud().field("intensity").is_err());
170        assert_eq!(migrated.cloud().len(), 1);
171    }
172
173    #[test]
174    fn migration_preserves_record_provenance() {
175        let source = xyz_record().with_provenance(
176            crate::RecordProvenance::try_new("bag-1")
177                .unwrap()
178                .with_stream_id("lidar")
179                .with_sequence(Some(9)),
180        );
181        let target = SchemaDescriptor::try_new(
182            "point",
183            SchemaVersion::new(1, 1),
184            StandardSchemas::point_xyzi(),
185        )
186        .unwrap();
187        let migrated =
188            crate::migrate_record(&source, &target, crate::MigrationPolicy::default()).unwrap();
189        assert_eq!(migrated.provenance(), source.provenance());
190    }
191}
spatialrust_records/migrate.rs

spatialrust_records/
migrate.rs
