linkrs_storage/
edge.rs

1//! Edge Storage Module
2//!
3//! Provides CSR (Compressed Sparse Row) based edge storage.
4//!
5//! ## Components
6//!
7//! - `MutableCsr`: Mutable CSR supporting dynamic edge operations
8//! - `SingleMutableCsr`: Optimized mutable CSR for single-edge scenarios
9//! - `ImmutableCsr`: Frozen packed CSR for read-mostly groups (explicit
10//!   freeze/unfreeze only, writes rejected while frozen)
11//! - `CsrVariant`: Enum wrapper for runtime CSR selection
12//! - `CsrWithProperties`: Ladybug-style columnar property storage
13//! - `CsrShardSet`: Node-group sharded topology container routing by endpoint interval
14//! - `EdgeStore`: Node-group sharded edge table combining out/in shards and property storage
15//!
16//! ## CSR Type Selection (two-level)
17//!
18//! Selection is two-level: first the record form (`RecordForm`), then the
19//! row strategy (`EdgeStrategy`) under the columnar form.
20//!
21//! Level 1 — `RecordForm` fixes bytes per edge and capability ceiling:
22//!
23//! | RecordForm | Bytes/edge | Use Case |
24//! |------------|------------|----------|
25//! | `Pure` | 12 | Pure topology: no rank, no timestamps, no properties |
26//! | `Bundled` | 20 | One inline scalar property, read-mostly, stable schema |
27//! | `Columnar` (default) | 32 + columnar properties | General multi-property edges |
28//!
29//! Level 2 — under `Columnar`, `EdgeStrategy` picks the CSR variant:
30//!
31//! | Strategy | CSR Type | Use Case | Time Complexity |
32//! |----------|----------|----------|-----------------|
33//! | `Multiple` | `MutableCsr` | General multi-edge relationships | O(degree) |
34//! | `Single` | `SingleMutableCsr` | One-to-one relationships (spouse, current_employer) | O(1) |
35//! | `None` | - | No edges stored | - |
36//!
37//! The form is locked at table creation and persisted; loading never
38//! re-derives it, and precondition-breaking changes go through explicit
39//! migration. Only `Pure` and `Columnar` are auto-derived; `Bundled` must
40//! be selected explicitly because of its capability ceiling.
41
42pub(crate) mod bundled_csr;
43pub(crate) mod csr_shared;
44pub mod csr_trait;
45pub mod csr_variant;
46pub mod csr_with_properties;
47pub mod edge_table;
48pub mod fragmentation_stats;
49pub mod immutable_csr;
50pub mod mutable_csr;
51pub mod node_group;
52pub mod property_schema;
53pub(crate) mod pure_csr;
54pub mod record_form_policy;
55pub mod schema;
56pub mod single_mutable_csr;
57pub mod slot;
58
59pub use csr_trait::{CsrBase, MutableCsrTrait};
60pub use csr_variant::{CsrRowIter, CsrVariant};
61pub use csr_with_properties::CsrWithProperties;
62pub use edge_table::core::UpdateEdgePropertyByKeyParams;
63pub use edge_table::{EdgeIndexStatus, EdgeStore, EdgeWalRecoveryMode, IncidentDeletedEdge};
64pub use fragmentation_stats::{
65    FragmentationStats, VertexFragmentation, GROUP_FRAGMENTATION_THRESHOLD,
66};
67pub use linkrs_core::types::EdgeStrategy;
68use linkrs_core::types::{EdgeId, Timestamp, VertexId};
69use linkrs_core::Value;
70pub use mutable_csr::{EdgePosition, MutableCsr, MutableCsrIterator};
71pub use node_group::{
72    region_id_for_local, region_local_range, regions_per_group, CsrShardSet, EdgeCheckpointKind,
73    FreezeBlockReason, FreezeFeasibility, GroupDirty, NodeGroupStats, RegionDirty,
74    RegionMergeScope, ShardCsrIterator, TableShardManifest, DEFAULT_NODE_GROUP_BITS,
75    GROUP_MERGE_MIN_DENSITY, LEAF_REGION_ROWS, REGION_MERGE_MIN_DENSITY,
76};
77pub use single_mutable_csr::{SingleMutableCsr, SingleMutableCsrIterator};
78
79pub use bundled_csr::{decode_scalar, encode_scalar, BundledCsr};
80pub use edge_table::checkpoint::snapshot::{
81    MappedFrozen, MappedFrozenIterator, MappedFrozenRowIter,
82};
83pub use immutable_csr::{FrozenRowIter, ImmutableCsr, ImmutableCsrIterator};
84pub use linkrs_core::types::INVALID_EDGE_ID;
85pub use pure_csr::{PureAllIter, PureRowIter, PureTopologyCsr};
86
87pub use record_form_policy::{
88    bundled_ineligibility_reason, inline_form_accepts_rank, is_bundled_eligible,
89    is_scalar_encodable, validate_record_form_target, validate_strategy_form, RecordForm,
90    RecordFormPreference, BUNDLED_RANK_REQUIRES_COLUMNAR_MSG,
91};
92pub(crate) use record_form_policy::{
93    INLINE_FORM_SCHEMA_CHANGE_MSG, NO_EDGES_STORED_MSG, ROW_POSITION_CROSS_VARIANT_MSG,
94    SCHEMA_CHANGE_PENDING_MSG, SINGLE_REQUIRES_COLUMNAR_MSG,
95};
96pub use schema::{EdgeMultiplicity, EdgeRecord, EdgeSchema, IndexConsistency, StorageDirection};
97pub use slot::{ColdStamps, HotNbr, Nbr};
98
99/// One edge batch-insert entry: `(src, dst, rank, properties, ts)`.
100pub type BatchInsertEntry<'a> = (u32, u32, i64, &'a [(std::sync::Arc<str>, Value)], Timestamp);
101
102/// Decoded neighbor for bulk puts: `(endpoint, rank, edge_id, ts)`.
103pub type EdgePut = (u32, i64, EdgeId, Timestamp);
104
105/// One source row's bulk-put batch: `(local_src, entries)`.
106pub type RowEdgeBatch = (u32, Vec<EdgePut>);
107
108#[cfg(test)]
109mod tests {
110    use super::*;
111    use crate::types::StoragePropertyDef;
112
113    #[test]
114    fn hot_half_stays_within_single_cache_line_budget() {
115        // Per-edge memory budget, pinned exactly: 24-byte hot half plus
116        // 8-byte cold half is 32 bytes per edge with no padding waste.
117        // Rank stays 64-bit because it is a caller-controlled multigraph
118        // key shared with the query layer, WAL redo records and endpoint
119        // key packing; narrowing it would reject legal inputs instead of
120        // storing them. create_ts lives in the EdgeTimestamps authority,
121        // not inline; only delete_ts is kept for row-level reclaim.
122        assert_eq!(std::mem::size_of::<HotNbr>(), 24);
123        assert_eq!(std::mem::size_of::<ColdStamps>(), 8);
124        assert_eq!(std::mem::size_of::<Nbr>(), 32);
125    }
126
127    #[test]
128    fn slot_halves_roundtrip_through_nbr() {
129        let nbr = Nbr::with_create_ts(7, 2, EdgeId(9), 100);
130        let assembled = Nbr::from_parts(nbr.hot(), nbr.cold());
131        assert_eq!(assembled, nbr);
132        assert_eq!(
133            Nbr::from_parts(HotNbr::dead_gap(), ColdStamps::dead_gap()),
134            Nbr::dead_gap()
135        );
136    }
137
138    #[test]
139    fn test_edge_schema_validation_both_none() {
140        let schema = EdgeSchema {
141            label_id: 0,
142            label_name: "invalid_edge".into(),
143            src_label: 0,
144            dst_label: 0,
145            properties: vec![],
146            oe_strategy: EdgeStrategy::None,
147            ie_strategy: EdgeStrategy::None,
148            schema_version: 1,
149            record_form: RecordForm::default(),
150        };
151
152        let result = schema.validate();
153        assert!(result.is_err());
154        assert!(result.unwrap_err().to_string().contains("at least one"));
155    }
156
157    #[test]
158    fn test_edge_schema_validation_both_enabled() {
159        let schema = EdgeSchema {
160            label_id: 0,
161            label_name: "valid_edge".into(),
162            src_label: 0,
163            dst_label: 0,
164            properties: vec![],
165            oe_strategy: EdgeStrategy::Multiple,
166            ie_strategy: EdgeStrategy::Single,
167            schema_version: 1,
168            record_form: RecordForm::default(),
169        };
170
171        let result = schema.validate();
172        assert!(result.is_ok());
173    }
174
175    #[test]
176    fn pure_target_rejection_comes_from_the_shared_constant() {
177        use linkrs_core::DataType;
178        let props = vec![StoragePropertyDef {
179            name: "p".into(),
180            data_type: DataType::Double,
181            nullable: true,
182            default_value: None,
183        }];
184        let err = validate_record_form_target(
185            &props,
186            EdgeStrategy::Multiple,
187            EdgeStrategy::Multiple,
188            RecordForm::Pure,
189        )
190        .expect_err("pure target with properties must be rejected");
191        assert!(
192            err.to_string()
193                .contains(record_form_policy::PURE_REQUIRES_ZERO_PROPERTIES_MSG),
194            "unexpected wording: {}",
195            err
196        );
197    }
198
199    #[test]
200    fn test_edge_schema_validation_ie_only() {
201        let schema = EdgeSchema {
202            label_id: 0,
203            label_name: "valid_edge".into(),
204            src_label: 0,
205            dst_label: 0,
206            properties: vec![],
207            oe_strategy: EdgeStrategy::None,
208            ie_strategy: EdgeStrategy::Multiple,
209            schema_version: 1,
210            record_form: RecordForm::default(),
211        };
212
213        let result = schema.validate();
214        assert!(result.is_ok());
215        assert_eq!(schema.storage_direction(), StorageDirection::InOnly);
216        assert!(!schema.has_out());
217        assert!(schema.has_in());
218    }
219}
linkrs_storage/edge.rs

linkrs_storage/
edge.rs
