surreal_sync_runtime/pipeline/
flatten_id.rs

1//! Built-in in-place transform: flatten Array record IDs to Text.
2
3use anyhow::Result;
4use std::collections::HashMap;
5use surreal_sync_core::InPlaceTransform;
6use surreal_sync_core::{flatten_composite_id, Relation, RelationChange, Value};
7
8/// Default separator matching relation FK flattening and historical Snowflake IDs.
9pub const DEFAULT_FLATTEN_ID_SEPARATOR: &str = ":";
10
11/// Rewrite `Value::Array` record IDs to Text joined with [`Self::separator`].
12///
13/// Scalar IDs are left unchanged. Relation **edge** IDs are also flattened when
14/// they are Arrays (endpoint Thing refs are flattened the same way so links stay
15/// consistent if a source emitted Array endpoint IDs).
16#[derive(Debug, Clone)]
17pub struct FlattenId {
18    /// Joiner between composite key parts (default `:`).
19    pub separator: String,
20}
21
22impl Default for FlattenId {
23    fn default() -> Self {
24        Self {
25            separator: DEFAULT_FLATTEN_ID_SEPARATOR.to_string(),
26        }
27    }
28}
29
30impl FlattenId {
31    /// Create with an explicit separator.
32    pub fn new(separator: impl Into<String>) -> Self {
33        Self {
34            separator: separator.into(),
35        }
36    }
37}
38
39impl InPlaceTransform for FlattenId {
40    fn transform(
41        &self,
42        _table: &str,
43        id: &mut Value,
44        _fields: Option<&mut HashMap<String, Value>>,
45    ) -> Result<()> {
46        *id = flatten_composite_id(std::mem::replace(id, Value::Null), &self.separator);
47        Ok(())
48    }
49
50    fn transform_relation(&self, relation: &mut Relation) -> Result<()> {
51        relation.id = flatten_composite_id(
52            std::mem::replace(&mut relation.id, Value::Null),
53            &self.separator,
54        );
55        relation.input.id = flatten_composite_id(
56            std::mem::replace(&mut relation.input.id, Value::Null),
57            &self.separator,
58        );
59        relation.output.id = flatten_composite_id(
60            std::mem::replace(&mut relation.output.id, Value::Null),
61            &self.separator,
62        );
63        Ok(())
64    }
65
66    fn transform_relation_change(&self, change: &mut RelationChange) -> Result<()> {
67        self.transform_relation(&mut change.relation)?;
68        Ok(())
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use super::*;
75    use surreal_sync_core::{Row, Type, Value};
76
77    #[test]
78    fn flattens_array_row_id() {
79        let t = FlattenId::default();
80        let mut row = Row::new(
81            "ledger",
82            0,
83            Value::Array {
84                elements: vec![Value::Int32(1), Value::Int32(2)],
85                element_type: Box::new(Type::Int32),
86            },
87            Default::default(),
88        );
89        t.transform_row(&mut row).unwrap();
90        assert_eq!(row.id, Value::Text("1:2".into()));
91    }
92
93    #[test]
94    fn leaves_scalar_unchanged() {
95        let t = FlattenId::new("_");
96        let mut row = Row::new("t", 0, Value::Int64(9), Default::default());
97        t.transform_row(&mut row).unwrap();
98        assert_eq!(row.id, Value::Int64(9));
99    }
100
101    #[test]
102    fn pipeline_from_flatten_id_toml_applies_to_rows() {
103        use crate::pipeline::{parse_transforms_toml, Pipeline};
104
105        let cfg = parse_transforms_toml(
106            r#"
107[[transforms]]
108type = "flatten_id"
109separator = ":"
110"#,
111        )
112        .unwrap();
113        let pipeline = Pipeline::from_config(&cfg).unwrap();
114        let rows = pipeline
115            .apply_rows(vec![Row::new(
116                "ledger",
117                0,
118                Value::Array {
119                    elements: vec![Value::Text("a".into()), Value::Text("b".into())],
120                    element_type: Box::new(Type::Text),
121                },
122                Default::default(),
123            )])
124            .unwrap();
125        assert_eq!(rows[0].id, Value::Text("a:b".into()));
126    }
127}
surreal_sync_runtime/pipeline/flatten_id.rs

surreal_sync_runtime/pipeline/
flatten_id.rs
