Skip to main content

linkrs_storage/edge/edge_table/
remap.rs

1//! Vertex ID remapping.
2//!
3//! When a vertex table is compacted, surviving vertices receive new internal
4//! IDs (densified). Edge shards index rows by internal IDs and store neighbors
5//! as encoded `(internal_id, rank)` keys, so the old-to-new mapping must be
6//! propagated here or every edge reference to compacted vertices breaks.
7//!
8//! Entries are collected per group (including tombstoned ones, preserving
9//! snapshot visibility), translated, and routed into fresh shard sets: a
10//! translated endpoint may land in a different group than its source.
11//! Trailing empty groups are dropped. Rebuilding also drops the
12//! pre-compaction group space left behind by deleted vertices.
13//!
14//! Vertex internal ID spaces are per-label, so an edge table must be given
15//! two separate mappings: one for its `src_label` space and one for its
16//! `dst_label` space. Out rows and in neighbors live in the src space; in
17//! rows and out neighbors live in the dst space.
18//!
19//! Observability: row and neighbor mapping misses are counted and debug
20//! logged. Unmapped endpoints keep their original value so sparse rebuilds
21//! stay total; every miss increments the returned counters for the caller
22//! to observe.
23//!
24//! Edge-id counter discipline: `next_edge_id` only guarantees monotonicity
25//! without collision, never crash-to-crash stability. Empty-table reload falls
26//! back to max id plus one; tests must assert monotonicity, never exact values.
27//!
28//! Complexity: linear in physical rows including tombstones. Per-group drain
29//! keeps peak memory bounded to one group; large-table compaction belongs in a
30//! maintenance window. Property index rebuild reuses the streaming build.
31
32use super::core::EdgeStore;
33use crate::edge::csr_trait::MutableCsrTrait;
34use crate::edge::{CsrShardSet, EdgeStrategy, Nbr, RecordForm};
35use linkrs_core::types::{Timestamp, VertexId};
36use linkrs_core::StorageResult;
37use std::collections::HashMap;
38
39/// Remap miss counters for one rebuild call.
40#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
41pub struct RemapStats {
42    pub row_misses: usize,
43    pub neighbor_misses: usize,
44    pub entries: usize,
45}
46
47fn remapped_row_counted(id: u32, mapping: Option<&HashMap<u32, u32>>, misses: &mut usize) -> u32 {
48    match mapping {
49        Some(m) => match m.get(&id).copied() {
50            Some(new_id) => new_id,
51            None => {
52                *misses += 1;
53                id
54            }
55        },
56        None => id,
57    }
58}
59
60fn remap_endpoint_key_counted(
61    key: VertexId,
62    mapping: Option<&HashMap<u32, u32>>,
63    misses: &mut usize,
64) -> VertexId {
65    let Some((endpoint, rank)) = EdgeStore::try_decode_edge_endpoint(key) else {
66        *misses += 1;
67        return key;
68    };
69    match endpoint.as_internal_u32() {
70        Some(id) => match mapping {
71            Some(m) => match m.get(&id).copied() {
72                Some(new_id) => EdgeStore::edge_endpoint_key(new_id, rank),
73                None => {
74                    *misses += 1;
75                    key
76                }
77            },
78            None => key,
79        },
80        None => {
81            *misses += 1;
82            key
83        }
84    }
85}
86
87/// Target shard layout for one remap rebuild direction.
88#[derive(Clone, Copy)]
89struct RebuildLayout {
90    strategy: EdgeStrategy,
91    group_bits: u32,
92    overflow_chunk_edges: usize,
93    form: RecordForm,
94}
95
96/// Rebuild one direction into a fresh shard set with translated rows and
97/// neighbors. Tombstoned entries are re-marked so snapshot visibility is
98/// preserved. Trailing empty groups are dropped; non-empty strategies keep
99/// at least one group. Miss counters observe unmapped rows and neighbors.
100fn remap_direction(
101    old: &CsrShardSet,
102    row_mapping: Option<&HashMap<u32, u32>>,
103    neighbor_mapping: Option<&HashMap<u32, u32>>,
104    layout: RebuildLayout,
105    stats: &mut RemapStats,
106) -> StorageResult<CsrShardSet> {
107    let RebuildLayout {
108        strategy,
109        group_bits,
110        overflow_chunk_edges,
111        form,
112    } = layout;
113    let mut rebuilt = CsrShardSet::new(strategy, group_bits, overflow_chunk_edges, form)?;
114    if strategy == EdgeStrategy::None {
115        return Ok(rebuilt);
116    }
117    // Drain per group so a single group never has to hold the whole table;
118    // routing decides the target group of each translated entry, which may
119    // differ from its source group after densification. Sparse holes are
120    // never visited.
121    for gid in old.existing_group_ids() {
122        let entries: Vec<(u32, Nbr)> = old
123            .group_variant(gid)
124            .map(|variant| {
125                variant
126                    .iter_all()
127                    .map(|(src, nbr)| {
128                        let base = crate::edge::node_group::group_base(gid, old.group_bits());
129                        let global = src.as_internal_u32().unwrap_or(0).saturating_add(base);
130                        (global, nbr)
131                    })
132                    .collect()
133            })
134            .unwrap_or_default();
135        for (src, nbr) in entries {
136            let mut row_miss = 0usize;
137            let mut nbr_miss = 0usize;
138            let new_src = remapped_row_counted(src, row_mapping, &mut row_miss);
139            let new_neighbor =
140                remap_endpoint_key_counted(nbr.to_vertex_id(), neighbor_mapping, &mut nbr_miss);
141            let new_nbr =
142                match new_neighbor
143                    .try_decode_edge_endpoint()
144                    .and_then(|(ep_vid, ep_rank)| {
145                        ep_vid.as_internal_u32().map(|endpoint| (endpoint, ep_rank))
146                    }) {
147                    Some((endpoint, rank)) => Nbr {
148                        endpoint,
149                        rank,
150                        ..nbr
151                    },
152                    None => {
153                        nbr_miss += 1;
154                        nbr
155                    }
156                };
157            stats.row_misses += row_miss;
158            stats.neighbor_misses += nbr_miss;
159            stats.entries += 1;
160            // Bundled values ride along keyed by edge id; other forms carry
161            // no inline state.
162            let inline_value = if form == RecordForm::Bundled {
163                let base = crate::edge::node_group::group_base(gid, old.group_bits());
164                let old_local = src.saturating_sub(base);
165                old.group_variant(gid)
166                    .and_then(|variant| variant.bundled_value_by_edge_id(old_local, nbr.edge_id))
167                    .and_then(|(raw, valid)| valid.then_some(raw))
168            } else {
169                None
170            };
171            if form == RecordForm::Bundled {
172                rebuilt.bundled_insert_with_value(
173                    new_src,
174                    new_nbr.to_vertex_id(),
175                    new_nbr.edge_id,
176                    0,
177                    inline_value,
178                )?;
179            } else {
180                rebuilt.insert_edge(new_src, new_nbr.to_vertex_id(), new_nbr.edge_id, 0)?;
181            }
182            if new_nbr.delete_ts != Timestamp::MAX {
183                let _ = rebuilt.delete_edge(new_src, new_nbr.edge_id, new_nbr.delete_ts);
184                rebuilt.mark_reclaim_hint_for(new_src);
185            }
186        }
187    }
188    // Rebuilt rows must reach the next checkpoint: mark every surviving
189    // group dirty so incremental flush cannot skip them. Reclaim hints are
190    // preserved: tombstoned entries survived the rebuild and the next
191    // reclaim pass must inspect their groups once. The rebuild-time append
192    // logs are dropped because the delete dirt above forces a base merge
193    // that carries the same states.
194    rebuilt.mark_all_dirty();
195    rebuilt.clear_all_append_logs();
196    rebuilt.truncate_trailing_empty_groups();
197    Ok(rebuilt)
198}
199
200/// Offline reshard statistics for one width change.
201#[derive(Debug, Clone, Copy, PartialEq, Eq)]
202pub struct ReshardStats {
203    pub old_bits: u32,
204    pub new_bits: u32,
205    pub out_groups: usize,
206    pub in_groups: usize,
207    pub edges: u64,
208}
209
210impl EdgeStore {
211    /// Offline group-width change: the only adjustment outlet for the locked
212    /// address width. Reads the old width group by group and writes the new
213    /// width, then switches the manifest on the next checkpoint. There is no
214    /// online width-change branch: callers must hold exclusive access and
215    /// checkpoint after a successful reshard. A successful width change arms
216    /// the mandatory-checkpoint fence, so writes are rejected until the next
217    /// checkpoint persists the new width; the equal-width no-op arms nothing.
218    /// Vertex ids are unchanged, so the property index needs no rebuild.
219    /// Widths outside `1..=20` are rejected; the current width is a no-op success.
220    ///
221    /// Operation guide: run offline with no other writer, flush with
222    /// `flush_incremental` after success, then reload to verify. The edge
223    /// logical set is unchanged and the audit must report no drift.
224    /// Selection guide: small tables tolerating downtime use offline reshard
225    /// for any width change; large tables tolerating no downtime keep the
226    /// current width and densify vertex ids offline first; small width
227    /// adjustments on idle tables prefer reshard, while online record-form
228    /// migration only changes the inline versus columnar layout and never
229    /// changes the address width.
230    pub fn reshard(&mut self, new_group_bits: u32) -> StorageResult<ReshardStats> {
231        crate::edge::node_group::validate_group_bits(new_group_bits)?;
232        let old_bits = self.config.node_group_bits;
233        if new_group_bits == old_bits {
234            return Ok(ReshardStats {
235                old_bits,
236                new_bits: new_group_bits,
237                out_groups: self.out_csr.group_count(),
238                in_groups: self.in_csr.group_count(),
239                edges: self.edge_count(),
240            });
241        }
242        let mut stats = RemapStats::default();
243        self.out_csr = remap_direction(
244            &self.out_csr,
245            None,
246            None,
247            RebuildLayout {
248                strategy: self.schema.oe_strategy,
249                group_bits: new_group_bits,
250                overflow_chunk_edges: self.config.overflow_chunk_edges,
251                form: self.out_csr.record_form(),
252            },
253            &mut stats,
254        )?;
255        self.in_csr = remap_direction(
256            &self.in_csr,
257            None,
258            None,
259            RebuildLayout {
260                strategy: self.schema.ie_strategy,
261                group_bits: new_group_bits,
262                overflow_chunk_edges: self.config.overflow_chunk_edges,
263                form: self.in_csr.record_form(),
264            },
265            &mut stats,
266        )?;
267        self.config.node_group_bits = new_group_bits;
268        let owner_stats = self.rebuild_owner_map_with_stats();
269        if let Some(stats) = &self.stats_manager {
270            stats.add_value_with_amount(
271                linkrs_metrics::MetricType::EdgeRelocatedOrphans,
272                owner_stats.relocated_orphans as u64,
273            );
274        }
275        let drift = self.audit_copy_drift();
276        if !drift.is_empty() {
277            return Err(linkrs_core::StorageError::db_error(format!(
278                "reshard drifted authority and projection: {}",
279                drift.join("; ")
280            )));
281        }
282        // The new width is memory-only until the next checkpoint persists
283        // it: fence writes the same way a record-form switch does, so a
284        // crash before the checkpoint recovers to the pre-reshard base with
285        // no mixed-width state ever built on top.
286        self.migration_pending_checkpoint = true;
287        log::debug!(
288            "EdgeTable[{}] resharded width {} -> {}; out_groups={}, in_groups={}, edges={}",
289            self.label,
290            old_bits,
291            new_group_bits,
292            self.out_csr.group_count(),
293            self.in_csr.group_count(),
294            self.edge_count(),
295        );
296        Ok(ReshardStats {
297            old_bits,
298            new_bits: new_group_bits,
299            out_groups: self.out_csr.group_count(),
300            in_groups: self.in_csr.group_count(),
301            edges: self.edge_count(),
302        })
303    }
304
305    /// Propagate vertex compaction old-to-new internal ID mappings into this
306    /// edge table.
307    ///
308    /// Must run inside the compaction commit barrier held by the maintenance
309    /// layer (vertex remap and edge rewrite share one barrier and one
310    /// journal): the caller journaled the vertex mapping before this call,
311    /// and an empty mapping short-circuits to a no-op so stable row IDs
312    /// (free-stack reuse below the hole-rate watermark) produce zero edge
313    /// writes.
314    ///
315    /// The table references two vertex label ID spaces:
316    /// - `src_mapping` applies to out shard rows and in shard neighbor keys
317    /// - `dst_mapping` applies to in shard rows and out shard neighbor keys
318    ///
319    /// Both directions are rebuilt group by group with trailing empty groups
320    /// dropped. The property index encodes (src, dst) internal IDs in its keys
321    /// and is rebuilt from the remapped data when enabled. Unmapped endpoints
322    /// keep their original value; misses are debug logged only.
323    pub fn remap_vertex_ids(
324        &mut self,
325        src_mapping: Option<&HashMap<u32, u32>>,
326        dst_mapping: Option<&HashMap<u32, u32>>,
327    ) -> StorageResult<()> {
328        self.remap_vertex_ids_with_stats(src_mapping, dst_mapping)
329            .map(|_| ())
330    }
331
332    /// Same as `remap_vertex_ids` but returns miss counters for tests.
333    pub fn remap_vertex_ids_with_stats(
334        &mut self,
335        src_mapping: Option<&HashMap<u32, u32>>,
336        dst_mapping: Option<&HashMap<u32, u32>>,
337    ) -> StorageResult<RemapStats> {
338        if src_mapping.is_none() && dst_mapping.is_none() {
339            return Ok(RemapStats::default());
340        }
341        let src_empty = src_mapping.is_none_or(|m| m.is_empty());
342        let dst_empty = dst_mapping.is_none_or(|m| m.is_empty());
343        if src_empty && dst_empty {
344            return Ok(RemapStats::default());
345        }
346
347        let mut stats = RemapStats::default();
348        self.out_csr = remap_direction(
349            &self.out_csr,
350            src_mapping,
351            dst_mapping,
352            RebuildLayout {
353                strategy: self.schema.oe_strategy,
354                group_bits: self.config.node_group_bits,
355                overflow_chunk_edges: self.config.overflow_chunk_edges,
356                form: self.out_csr.record_form(),
357            },
358            &mut stats,
359        )?;
360        self.in_csr = remap_direction(
361            &self.in_csr,
362            dst_mapping,
363            src_mapping,
364            RebuildLayout {
365                strategy: self.schema.ie_strategy,
366                group_bits: self.config.node_group_bits,
367                overflow_chunk_edges: self.config.overflow_chunk_edges,
368                form: self.in_csr.record_form(),
369            },
370            &mut stats,
371        )?;
372
373        // The property index encodes (src, dst) internal IDs in its keys;
374        // rebuild it from the remapped data when enabled.
375        if self.property_index.is_some() {
376            let pool_capacity = self
377                .property_index
378                .as_ref()
379                .map(|idx| idx.pool_capacity())
380                .unwrap_or(1024);
381            self.build_property_index(pool_capacity)?;
382        }
383        let owner_stats = self.rebuild_owner_map_with_stats();
384        if let Some(metrics) = &self.stats_manager {
385            metrics.add_value_with_amount(
386                linkrs_metrics::MetricType::EdgeRelocatedOrphans,
387                owner_stats.relocated_orphans as u64,
388            );
389        }
390        let drift = self.audit_copy_drift();
391        if !drift.is_empty() {
392            return Err(linkrs_core::StorageError::db_error(format!(
393                "remap drifted authority and projection: {}",
394                drift.join("; ")
395            )));
396        }
397        let (_, _, live_orphans) = self.copy_audit();
398        if let Some(stats) = &self.stats_manager {
399            stats.add_value_with_amount(
400                linkrs_metrics::MetricType::EdgeLiveAuthorityOrphans,
401                live_orphans as u64,
402            );
403        }
404
405        log::debug!(
406            "EdgeTable[{}] remapped vertex IDs (src_mapping={}, dst_mapping={}); out_groups={}, in_groups={}, row_misses={}, neighbor_misses={}, entries={}",
407            self.label,
408            src_mapping.map(|m| m.len()).unwrap_or(0),
409            dst_mapping.map(|m| m.len()).unwrap_or(0),
410            self.out_csr.group_count(),
411            self.in_csr.group_count(),
412            stats.row_misses,
413            stats.neighbor_misses,
414            stats.entries,
415        );
416
417        Ok(stats)
418    }
419}
420
421#[cfg(test)]
422mod tests {
423    use super::*;
424    use crate::edge::edge_table::config::EdgeTableConfig;
425    use crate::edge::{EdgeSchema, EdgeStrategy, RecordForm};
426    use crate::types::StoragePropertyDef;
427    use linkrs_core::Value;
428
429    fn make_table() -> EdgeStore {
430        let schema = EdgeSchema {
431            label_id: 0,
432            label_name: "knows".into(),
433            src_label: 0,
434            dst_label: 0,
435            properties: vec![StoragePropertyDef {
436                name: "weight".into(),
437                data_type: linkrs_core::types::DataType::Double,
438                nullable: false,
439                default_value: Some(Value::Double(0.0)),
440            }],
441            oe_strategy: EdgeStrategy::Multiple,
442            ie_strategy: EdgeStrategy::Multiple,
443            schema_version: 1,
444            record_form: RecordForm::default(),
445        };
446        EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
447    }
448
449    fn mapping_from_removals(live: &[u32]) -> HashMap<u32, u32> {
450        // Simulates a vertex compaction: live ids (sorted) become dense 0..n.
451        let mut mapping = HashMap::new();
452        for (new_id, &old_id) in live.iter().enumerate() {
453            if old_id != new_id as u32 {
454                mapping.insert(old_id, new_id as u32);
455            }
456        }
457        mapping
458    }
459
460    fn make_single_table() -> EdgeStore {
461        let schema = EdgeSchema {
462            label_id: 0,
463            label_name: "single".into(),
464            src_label: 0,
465            dst_label: 0,
466            properties: vec![],
467            oe_strategy: EdgeStrategy::Single,
468            ie_strategy: EdgeStrategy::Single,
469            schema_version: 1,
470            record_form: RecordForm::default(),
471        };
472        EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
473    }
474
475    #[test]
476    fn test_remap_rows_and_neighbors() {
477        let mut table = make_table();
478        // Rows 0..5, gaps at 2 and 4 (deleted vertices).
479        table.insert_edge(0, 1, 0, &[], 100).unwrap();
480        table.insert_edge(1, 3, 1, &[], 100).unwrap();
481        table.insert_edge(3, 5, 2, &[], 100).unwrap();
482        table.insert_edge(5, 0, 3, &[], 100).unwrap();
483
484        // Vertex compaction keeps rows {0, 1, 3, 5} → dense {0, 1, 2, 3}.
485        let mapping = mapping_from_removals(&[0, 1, 3, 5]);
486
487        table
488            .remap_vertex_ids(Some(&mapping), Some(&mapping))
489            .expect("remap should succeed");
490
491        // src 0 -> 0, src 1 -> 1, src 3 -> 2, src 5 -> 3
492        assert_eq!(table.out_edges(0, 200).len(), 1);
493        assert_eq!(table.out_edges(1, 200).len(), 1);
494        assert_eq!(table.out_edges(2, 200).len(), 1);
495        assert_eq!(table.out_edges(3, 200).len(), 1);
496
497        // Neighbors remapped: 1 -> 1, 3 -> 2, 5 -> 3, 0 -> 0
498        assert!(table.get_edge(0, 1, 0, 200).is_some());
499        assert!(table.get_edge(1, 2, 1, 200).is_some());
500        assert!(table.get_edge(2, 3, 2, 200).is_some());
501        assert!(table.get_edge(3, 0, 3, 200).is_some());
502
503        // Single-group table: one group per direction, trailing groups gone.
504        assert_eq!(table.out_csr.group_count(), 1);
505        assert_eq!(table.in_csr.group_count(), 1);
506        assert!(table.get_edge(4, 0, 0, 200).is_none());
507        assert_eq!(table.out_edges(4, 200).len(), 0);
508    }
509
510    #[test]
511    fn test_remap_preserves_tombstoned_entries() {
512        let mut table = make_table();
513        table.insert_edge(0, 1, 0, &[], 100).unwrap();
514        table.insert_edge(3, 4, 1, &[], 100).unwrap();
515        // Edge 0->1 is deleted at ts=150; must survive remapping.
516        table.delete_edge(0, 1, 0, 150).unwrap();
517
518        // Vertex 2 removed; live {0, 1, 3, 4} → dense {0, 1, 2, 3}.
519        let mapping = mapping_from_removals(&[0, 1, 3, 4]);
520
521        table
522            .remap_vertex_ids(Some(&mapping), Some(&mapping))
523            .unwrap();
524
525        // Live at ts before deletion: still visible via remapped rows.
526        let before = table.out_edges(0, 149);
527        assert_eq!(before.len(), 1);
528        assert_eq!(before[0].dst_vid.as_int64(), Some(1));
529        // Live at ts after deletion: hidden, but entry preserved.
530        assert_eq!(table.out_edges(0, 200).len(), 0);
531        // Second edge: src 3 -> 2, dst 4 -> 3 (dense).
532        assert!(table.get_edge(2, 3, 1, 200).is_some());
533        // Highest edge-bearing row is 2; a single group remains.
534        assert_eq!(table.out_csr.group_count(), 1);
535    }
536
537    #[test]
538    fn test_remap_dst_only_affects_neighbors_out() {
539        let mut table = make_table();
540        table.insert_edge(1, 4, 0, &[], 100).unwrap();
541
542        // Only dst label compacted: rows stay, out neighbors change.
543        let mapping = HashMap::from([(4u32, 0u32)]);
544
545        table.remap_vertex_ids(None, Some(&mapping)).unwrap();
546
547        assert!(table.get_edge(1, 0, 0, 200).is_some());
548        assert!(table.get_edge(1, 4, 0, 200).is_none());
549        assert_eq!(table.out_csr.group_count(), 1);
550    }
551
552    #[test]
553    fn test_remap_src_only_affects_rows() {
554        let mut table = make_table();
555        table.insert_edge(4, 1, 0, &[], 100).unwrap();
556
557        // Only src label compacted: rows change, out neighbors stay.
558        let mapping = HashMap::from([(4u32, 0u32)]);
559
560        table.remap_vertex_ids(Some(&mapping), None).unwrap();
561
562        assert!(table.get_edge(0, 1, 0, 200).is_some());
563        assert!(table.get_edge(4, 1, 0, 200).is_none());
564        assert_eq!(table.out_csr.group_count(), 1);
565        assert_eq!(table.in_csr.group_count(), 1);
566    }
567
568    #[test]
569    fn test_remap_empty_mapping_is_noop() {
570        let mut table = make_table();
571        table.insert_edge(0, 1, 0, &[], 100).unwrap();
572        let groups_before = table.out_csr.group_count();
573        table.remap_vertex_ids(Some(&HashMap::new()), None).unwrap();
574        assert_eq!(table.out_csr.group_count(), groups_before);
575        assert!(table.get_edge(0, 1, 0, 200).is_some());
576    }
577
578    #[test]
579    fn test_remap_moves_edges_across_groups() {
580        let mut table = make_table();
581        table.insert_edge(5000, 6000, 0, &[], 100).unwrap();
582        assert_eq!(table.out_csr.group_count(), 2);
583
584        // Densify 5000 -> 1 and 6000 -> 2: the edge moves into group 0.
585        let src_mapping = HashMap::from([(5000u32, 1u32)]);
586        let dst_mapping = HashMap::from([(6000u32, 2u32)]);
587        table
588            .remap_vertex_ids(Some(&src_mapping), Some(&dst_mapping))
589            .unwrap();
590
591        assert!(table.get_edge(1, 2, 0, 200).is_some());
592        assert!(table.get_edge(5000, 6000, 0, 200).is_none());
593        assert_eq!(table.out_csr.group_count(), 1);
594    }
595
596    #[test]
597    fn test_remap_flush_load_roundtrip() {
598        let mut table = make_table();
599        table.insert_edge(0, 1, 0, &[], 100).unwrap();
600        table.insert_edge(5000, 6000, 0, &[], 100).unwrap();
601        let dir = tempfile::tempdir().expect("temporary edge table directory");
602        table
603            .flush(
604                dir.path(),
605                crate::compression::CompressionType::Zstd { level: 3 },
606            )
607            .expect("baseline flush should succeed");
608
609        let src_mapping = HashMap::from([(5000u32, 1u32)]);
610        let dst_mapping = HashMap::from([(6000u32, 2u32)]);
611        table
612            .remap_vertex_ids(Some(&src_mapping), Some(&dst_mapping))
613            .unwrap();
614        assert!(!table.out_csr.dirty_group_ids().is_empty());
615        table
616            .flush(
617                dir.path(),
618                crate::compression::CompressionType::Zstd { level: 3 },
619            )
620            .expect("post-remap flush should succeed");
621
622        let mut loaded = make_table();
623        loaded.load(dir.path()).expect("load should succeed");
624        assert!(loaded.get_edge(0, 1, 0, 200).is_some());
625        assert!(loaded.get_edge(1, 2, 0, 200).is_some());
626        assert!(loaded.get_edge(5000, 6000, 0, 200).is_none());
627        assert_eq!(loaded.edge_count(), 2);
628    }
629
630    #[test]
631    fn test_remap_compaction_reports_zero_misses() {
632        let mut table = make_table();
633        table.insert_edge(0, 1, 0, &[], 100).unwrap();
634        table.insert_edge(1, 0, 0, &[], 100).unwrap();
635        let mapping = HashMap::from([(0u32, 0u32), (1u32, 1u32)]);
636        let stats = table
637            .remap_vertex_ids_with_stats(Some(&mapping), Some(&mapping))
638            .unwrap();
639        assert_eq!(stats.row_misses, 0);
640        assert_eq!(stats.neighbor_misses, 0);
641        assert!(table.get_edge(0, 1, 0, 200).is_some());
642        assert!(table.get_edge(1, 0, 0, 200).is_some());
643        assert_eq!(table.live_authority_orphans(), 0);
644    }
645
646    #[test]
647    fn test_remap_partial_mapping_reports_expected_misses() {
648        let mut table = make_table();
649        table.insert_edge(0, 1, 0, &[], 100).unwrap();
650        let mapping = HashMap::from([(99u32, 100u32)]);
651        let stats = table
652            .remap_vertex_ids_with_stats(Some(&mapping), Some(&mapping))
653            .unwrap();
654        assert!(stats.row_misses > 0);
655        assert!(stats.neighbor_misses > 0);
656        assert!(table.get_edge(0, 1, 0, 200).is_some());
657    }
658
659    #[test]
660    fn test_single_strategy_remap_into_fresh_shards() {
661        let mut table = make_single_table();
662        table.insert_edge(0, 1, 0, &[], 100).unwrap();
663        let mapping = HashMap::from([(0u32, 0u32), (1u32, 1u32)]);
664        let stats = table
665            .remap_vertex_ids_with_stats(Some(&mapping), Some(&mapping))
666            .unwrap();
667        assert_eq!(stats.row_misses, 0);
668        assert_eq!(stats.neighbor_misses, 0);
669        assert!(table.get_edge(0, 1, 0, 200).is_some());
670        assert_eq!(table.live_authority_orphans(), 0);
671    }
672
673    #[test]
674    fn test_reshard_equal_width_is_noop_success_and_bad_width_rejected() {
675        let mut table = make_table();
676        table.insert_edge(0, 1, 0, &[], 100).unwrap();
677        let current = table.config.node_group_bits;
678        let stats = table.reshard(current).expect("equal width must succeed");
679        assert_eq!(stats.old_bits, current);
680        assert_eq!(stats.new_bits, current);
681        assert!(table.get_edge(0, 1, 0, 200).is_some());
682        assert!(table.audit_copy_drift().is_empty());
683        assert!(table.reshard(0).is_err());
684        assert!(table.reshard(21).is_err());
685        assert!(table.get_edge(0, 1, 0, 200).is_some());
686    }
687
688    #[test]
689    fn test_reshard_arms_mandatory_checkpoint_fence() {
690        let mut table = make_table();
691        table.insert_edge(0, 1, 0, &[], 100).unwrap();
692        let current = table.config.node_group_bits;
693        assert!(current > 1);
694        table.reshard(current - 1).expect("width change succeeds");
695        // Reads keep serving the new width, but writes stay fenced until
696        // the mandatory checkpoint persists it.
697        assert!(table.is_migration_checkpoint_required());
698        assert!(table.get_edge(0, 1, 0, 200).is_some());
699        assert!(table.insert_edge(2, 3, 0, &[], 200).is_err());
700        assert!(table.audit_copy_drift().is_empty());
701        let dir = tempfile::tempdir().expect("temporary edge table directory");
702        table
703            .flush(
704                dir.path(),
705                crate::compression::CompressionType::Zstd { level: 3 },
706            )
707            .expect("mandatory checkpoint");
708        assert!(!table.is_migration_checkpoint_required());
709        table
710            .insert_edge(2, 3, 0, &[], 200)
711            .expect("write after checkpoint");
712        assert!(table.get_edge(2, 3, 0, 300).is_some());
713        assert!(table.audit_copy_drift().is_empty());
714    }
715
716    #[test]
717    fn test_edge_id_counter_monotonic_across_reload() {
718        let mut table = make_table();
719        table.insert_edge(0, 1, 0, &[], 100).unwrap();
720        table.insert_edge(0, 2, 0, &[], 110).unwrap();
721        table.delete_edge(0, 2, 0, 150).unwrap();
722        let before = table.next_edge_id;
723        let dir = tempfile::tempdir().expect("temporary edge table directory");
724        table
725            .flush(
726                dir.path(),
727                crate::compression::CompressionType::Zstd { level: 3 },
728            )
729            .expect("flush should succeed");
730        let mut loaded = make_table();
731        loaded.load(dir.path()).expect("load should succeed");
732        assert!(loaded.next_edge_id >= before);
733        let mut out_seen = std::collections::HashSet::new();
734        for (_, nbr) in loaded.out_csr.iter_all() {
735            assert!(out_seen.insert(nbr.edge_id));
736        }
737        let mut in_seen = std::collections::HashSet::new();
738        for (_, nbr) in loaded.in_csr.iter_all() {
739            assert!(in_seen.insert(nbr.edge_id));
740        }
741        assert_eq!(out_seen, in_seen);
742    }
743}