Extract edge properties from an EdgeStore into a PropertyColumns aligned with node IDs. For each edge, extracts specified property names from MessagePack-encoded properties. Properties are associated with the **source** node of each edge. This is a bulk extraction — O(E) — intended for pre-computation before running algorithms, not for hot-path queries.
(
edge_store: &crate::engine::graph::edge_store::EdgeStore,
csr: &crate::engine::graph::csr::CsrIndex,
tid: nodedb_types::TenantId,
property_names: &[&str],
)
| 258 | /// This is a bulk extraction — O(E) — intended for pre-computation before |
| 259 | /// running algorithms, not for hot-path queries. |
| 260 | pub fn extract_edge_properties( |
| 261 | edge_store: &crate::engine::graph::edge_store::EdgeStore, |
| 262 | csr: &crate::engine::graph::csr::CsrIndex, |
| 263 | tid: nodedb_types::TenantId, |
| 264 | property_names: &[&str], |
| 265 | ) -> Result<PropertyColumns, crate::Error> { |
| 266 | let n = csr.node_count(); |
| 267 | let mut columns = PropertyColumns::new(n); |
| 268 | |
| 269 | // Current-state view only — Ceiling-resolved, tombstones filtered, |
| 270 | // properties already decoded out of `EdgeValuePayload`. |
| 271 | let tenant_edges: Vec<_> = edge_store |
| 272 | .scan_all_edges_decoded(None)? |
| 273 | .into_iter() |
| 274 | .filter(|(rec_tid, _, _, _, _, _)| *rec_tid == tid) |
| 275 | .collect(); |
| 276 | |
| 277 | for (_tid, _coll, src_name, _label, _dst, properties) in &tenant_edges { |
| 278 | if properties.is_empty() { |
| 279 | continue; |
| 280 | } |
| 281 | |
| 282 | let Some(src_id) = csr.node_id_raw(src_name) else { |
| 283 | continue; |
| 284 | }; |
| 285 | let edge_properties = properties; |
| 286 | |
| 287 | // Parse MessagePack properties. |
| 288 | let Ok(val) = rmpv::decode::read_value(&mut edge_properties.as_slice()) else { |
| 289 | continue; |
| 290 | }; |
| 291 | |
| 292 | if let rmpv::Value::Map(entries) = val { |
| 293 | for (k, v) in entries { |
| 294 | if let rmpv::Value::String(ref key_str) = k { |
| 295 | let key = key_str.as_str().unwrap_or(""); |
| 296 | if !property_names.contains(&key) { |
| 297 | continue; |
| 298 | } |
| 299 | |
| 300 | match v { |
| 301 | rmpv::Value::F64(f) => { |
| 302 | columns.f64_column(key).set(src_id, f); |
| 303 | } |
| 304 | rmpv::Value::F32(f) => { |
| 305 | columns.f64_column(key).set(src_id, f as f64); |
| 306 | } |
| 307 | rmpv::Value::Integer(i) => { |
| 308 | if let Some(val) = i.as_i64() { |
| 309 | columns.i64_column(key).set(src_id, val); |
| 310 | } |
| 311 | } |
| 312 | rmpv::Value::String(ref s) => { |
| 313 | if let Some(s) = s.as_str() { |
| 314 | columns.string_column(key).set(src_id, s); |
| 315 | } |
| 316 | } |
| 317 | _ => {} |