perf(graph-db): store relation identities once, on the locator

crates/tracedecay-graph-db/Cargo.toml

@@ -66,6 +66,7 @@ grafeo-common.workspace = true
 grafeo-core = { workspace = true, features = ["compact-store"] }
 grafeo-engine = { workspace = true, features = ["mmap", "compact-store"] }
 grafeo-storage.workspace = true
+base64 = "0.22"
 hex = "0.4"
 hotpath.workspace = true
 parking_lot = { version = "0.12", features = ["arc_lock", "send_guard"] }

crates/tracedecay-graph-db/src/location.rs

@@ -20,7 +20,7 @@ pub struct GraphFormatVersion(u32);
 impl GraphFormatVersion {
     #[must_use]
     pub const fn current() -> Self {
-        Self(3)
+        Self(4)
     }
 
     #[cfg(any(test, feature = "test-helpers", feature = "eval-helpers"))]

crates/tracedecay-graph-db/src/projection.rs

@@ -12,6 +12,7 @@ use crate::limits::{
     MAX_GRAPH_IDENTIFIER_BYTES, MAX_GRAPH_PROPERTIES, MAX_GRAPH_PROPERTY_AGGREGATE_BYTES,
     MAX_GRAPH_PROPERTY_VALUE_BYTES,
 };
+use crate::schema::COMPACT_IDENTITY_MARKER;
 use crate::{GraphBudgetKind, GraphDbError};
 
 const RESERVED_PREFIX: &str = "__tracedecay_graph_db_";
@@ -98,6 +99,11 @@ fn validate_opaque(kind: &str, value: &str) -> Result<(), GraphDbError> {
             "{kind} exceeds {MAX_GRAPH_IDENTIFIER_BYTES} bytes"
         )));
     }
+    if value.starts_with(COMPACT_IDENTITY_MARKER) {
+        return Err(GraphDbError::invalid(format!(
+            "{kind} starts with the reserved compact identity marker"
+        )));
+    }
     if value.starts_with(RESERVED_PREFIX) {
         return Err(GraphDbError::invalid(format!(
             "{kind} uses the reserved graph database prefix"

crates/tracedecay-graph-db/src/projection_identity_index.rs

@@ -24,12 +24,11 @@ use std::collections::HashMap;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, RwLock};
 
-use grafeo_common::types::Value;
 use grafeo_engine::GrafeoDB;
 
 use crate::projection::check_cancelled;
 use crate::projection_read::IdentityScope;
-use crate::schema::{has_native_label, nodes_with_label};
+use crate::schema::{decode_identity, has_native_label, nodes_with_label};
 use crate::{GraphCancellation, GraphDbError};
 
 /// Identity bytes one cached index may retain. A projection whose identities
@@ -200,12 +199,7 @@ fn build_identity_index(
         if !has_native_label(&record, record_label) {
             continue;
         }
-        let identity = record
-            .get_property(identity_property)
-            .and_then(Value::as_str)
-            .ok_or_else(|| GraphDbError::Corrupt {
-                message: format!("projection query returned a non-string `{identity_property}`"),
-            })?;
+        let identity = decode_identity(record.get_property(identity_property), identity_property)?;
         identity_bytes = identity_bytes.saturating_add(identity.len());
         if identity_bytes > MAX_IDENTITY_INDEX_BYTES {
             return Ok(None);

crates/tracedecay-graph-db/src/projection_read.rs

@@ -2,12 +2,11 @@ use std::collections::BTreeSet;
 use std::fmt;
 use std::sync::Arc;
 
-use grafeo_common::types::Value;
 use grafeo_engine::GrafeoDB;
 
 use crate::projection::check_cancelled;
 use crate::schema::{
-    ENTITY_ID_PROPERTY, ENTITY_LABEL, RELATION_ID_PROPERTY, RELATION_LABEL,
+    ENTITY_ID_PROPERTY, ENTITY_LABEL, RELATION_ID_PROPERTY, RELATION_LABEL, decode_identity,
     entity_projection_label, relation_projection_label,
 };
 use crate::state::{labeled_projection_nodes, latest_projection, load_entity, load_relation};
@@ -411,25 +410,17 @@ fn streaming_identity_page(
         let Some(record) = store.get_node(node) else {
             continue;
         };
-        let identity = record
-            .get_property(identity_property)
-            .and_then(Value::as_str)
-            .ok_or_else(|| GraphDbError::Corrupt {
-                message: format!("projection query returned a non-string `{identity_property}`"),
-            })?;
-        if after.is_some_and(|after| identity <= after) {
+        let identity = decode_identity(record.get_property(identity_property), identity_property)?;
+        if after.is_some_and(|after| identity.as_str() <= after) {
             continue;
         }
         if page.len() == limit {
-            if page
-                .last()
-                .is_some_and(|widest| identity >= widest.as_str())
-            {
+            if page.last().is_some_and(|widest| identity >= *widest) {
                 continue;
             }
             page.pop_last();
         }
-        page.insert(identity.to_owned());
+        page.insert(identity);
     }
     Ok(page.into_iter().collect())
 }