linkrs_storage/edge/csr_with_properties/
transfer.rs

1use super::{CsrWithProperties, ExportedRow};
2use linkrs_core::types::{EdgeId, Timestamp};
3use linkrs_core::{StorageResult, Value};
4use std::sync::Arc;
5
6impl CsrWithProperties {
7    /// Export one edge row for per-group property sharding.
8    ///
9    /// Returns the row visibility plus current values for every schema
10    /// column (nulls as `None`). Unknown edges yield `None`. The export
11    /// carries plain values only; version history stays memory-only and is
12    /// collapsed on checkpoint, matching the whole-file dump contract.
13    pub fn export_row(&self, edge_id: EdgeId) -> Option<ExportedRow> {
14        let pos = self.mapped_row(edge_id)?;
15        let vis = *self.visibility.get(pos)?;
16        let values = self
17            .property_schema
18            .iter()
19            .enumerate()
20            .map(|(idx, schema)| {
21                let current = self.property_columns.get(idx).and_then(|col| col.get(pos));
22                (schema.name.clone(), current)
23            })
24            .collect();
25        Some((vis.create_ts, vis.delete_ts, values))
26    }
27
28    /// Import one exported row into this store for shard merge.
29    ///
30    /// Unknown columns in `values` are skipped so shards written under an
31    /// older published schema still merge; missing columns keep their
32    /// defaults via the fresh insert path. Nulls are materialized
33    /// explicitly so a shard null never becomes a default on merge.
34    pub fn import_row(
35        &mut self,
36        edge_id: EdgeId,
37        create_ts: Timestamp,
38        delete_ts: Option<Timestamp>,
39        values: &[(Arc<str>, Option<Value>)],
40    ) -> StorageResult<()> {
41        if self.mapped_row(edge_id).is_some() {
42            return Ok(());
43        }
44        let mut present: Vec<(usize, Value)> = Vec::new();
45        let mut nulls: Vec<Arc<str>> = Vec::new();
46        for (name, opt) in values {
47            let Some(&idx) = self.column_index.get(name.as_ref()) else {
48                continue;
49            };
50            match opt {
51                Some(value) => present.push((idx, value.clone())),
52                None => nulls.push(name.clone()),
53            }
54        }
55        self.insert_for_edge_at(edge_id, &present, create_ts)?;
56        for name in nulls {
57            let _ = self.set_property_for_edge(edge_id, &name, None, create_ts);
58        }
59        if let Some(delete) = delete_ts {
60            let _ = self.mark_deleted(edge_id, delete);
61        }
62        Ok(())
63    }
64}
linkrs_storage/edge/csr_with_properties/transfer.rs

linkrs_storage/edge/csr_with_properties/
transfer.rs
