linkrs_storage/edge/mutable_csr/
serialization.rs1use linkrs_core::{StorageError, StorageResult};
2
3use crate::persistence::read_u32_le;
4
5use super::super::{EdgeId, Nbr};
6use super::overflow::OverflowChunk;
7
8pub(crate) const MUTABLE_CSR_FORMAT_VERSION: u32 = 10;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum TopologyColumnEncoding {
18 Plain = 0,
19 BitPacked = 1,
20 Rle = 2,
21}
22
23impl TopologyColumnEncoding {
24 fn from_u8(value: u8) -> StorageResult<Self> {
25 match value {
26 0 => Ok(Self::Plain),
27 1 => Ok(Self::BitPacked),
28 2 => Ok(Self::Rle),
29 other => Err(StorageError::deserialize_error(format!(
30 "unknown topology column encoding: {}",
31 other
32 ))),
33 }
34 }
35}
36
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub struct TopologyEncodingChoice {
43 pub encoding: TopologyColumnEncoding,
44 pub plain_bytes: usize,
45 pub encoded_bytes: usize,
46}
47
48fn bit_width_for_range(range: u64) -> u8 {
49 if range == 0 {
50 1
51 } else {
52 (64 - range.leading_zeros()) as u8
53 }
54}
55
56fn rle_runs(values: &[u64]) -> Vec<(u64, u32)> {
57 let mut runs: Vec<(u64, u32)> = Vec::new();
58 for &value in values {
59 match runs.last_mut() {
60 Some(last) if last.0 == value => {
61 last.1 = last.1.saturating_add(1);
62 }
63 _ => runs.push((value, 1)),
64 }
65 }
66 runs
67}
68
69fn bitpacked_payload(values: &[u64]) -> Option<Vec<u8>> {
70 if values.is_empty() {
71 return Some(Vec::new());
72 }
73 let min = *values.iter().min()?;
74 let max = *values.iter().max()?;
75 let width = bit_width_for_range(max.saturating_sub(min)).max(1) as usize;
76 if width >= 64 {
77 return None;
78 }
79 let mut out = Vec::new();
80 out.extend_from_slice(&min.to_le_bytes());
81 out.push(width as u8);
82 let mut current: u64 = 0;
83 let mut filled: usize = 0;
84 for &value in values {
85 let adjusted = value.saturating_sub(min);
86 for bit in 0..width {
87 if (adjusted >> bit) & 1 == 1 {
88 current |= 1u64 << filled;
89 }
90 filled += 1;
91 if filled == 64 {
92 out.extend_from_slice(¤t.to_le_bytes());
93 current = 0;
94 filled = 0;
95 }
96 }
97 }
98 if filled > 0 {
99 out.extend_from_slice(¤t.to_le_bytes());
100 }
101 Some(out)
102}
103
104fn decode_bitpacked_payload(payload: &[u8], count: usize) -> StorageResult<Vec<u64>> {
105 if count == 0 {
106 if !payload.is_empty() {
107 return Err(StorageError::deserialize_error(
108 "non-empty bitpacked payload for empty topology column",
109 ));
110 }
111 return Ok(Vec::new());
112 }
113 if payload.len() < 9 {
114 return Err(StorageError::deserialize_error(
115 "bitpacked topology column too short",
116 ));
117 }
118 let min = u64::from_le_bytes(payload[0..8].try_into().map_err(|_| {
119 StorageError::deserialize_error("bitpacked topology column base too short")
120 })?);
121 let width = payload[8] as usize;
122 if width == 0 || width >= 64 {
123 return Err(StorageError::deserialize_error(format!(
124 "invalid bitpacked topology width: {}",
125 width
126 )));
127 }
128 let words: Vec<u64> = payload[9..]
129 .as_chunks::<8>()
130 .0
131 .iter()
132 .map(|chunk| u64::from_le_bytes(*chunk))
133 .collect();
134 if payload[9..].len() != words.len() * 8 {
135 return Err(StorageError::deserialize_error(
136 "bitpacked topology payload has trailing bytes",
137 ));
138 }
139 let mut out = Vec::with_capacity(count);
140 let mut bit_pos: usize = 0;
141 for _ in 0..count {
142 let mut adjusted: u64 = 0;
143 for bit in 0..width {
144 let word = bit_pos / 64;
145 let offset = bit_pos % 64;
146 let stored = words.get(word).copied().unwrap_or(0);
147 if (stored >> offset) & 1 == 1 {
148 adjusted |= 1u64 << bit;
149 }
150 bit_pos += 1;
151 }
152 out.push(min.saturating_add(adjusted));
153 }
154 let expected_words = bit_pos.div_ceil(64);
155 if words.len() != expected_words {
156 return Err(StorageError::deserialize_error(
157 "bitpacked topology payload length mismatch",
158 ));
159 }
160 Ok(out)
161}
162
163pub fn encode_topology_u64_column(values: &[u64]) -> (TopologyEncodingChoice, Vec<u8>) {
167 let plain_bytes = values.len().saturating_mul(8);
168 let mut best_encoding = TopologyColumnEncoding::Plain;
169 let mut best_payload = Vec::with_capacity(plain_bytes);
170 for &value in values {
171 best_payload.extend_from_slice(&value.to_le_bytes());
172 }
173 let mut best_bytes = plain_bytes;
174
175 if let Some(packed) = bitpacked_payload(values) {
176 if packed.len() < best_bytes {
177 best_bytes = packed.len();
178 best_encoding = TopologyColumnEncoding::BitPacked;
179 best_payload = packed;
180 }
181 }
182
183 if !values.is_empty() {
184 let runs = rle_runs(values);
185 let rle_bytes = runs.len().saturating_mul(12);
186 if rle_bytes < best_bytes {
187 let mut payload = Vec::with_capacity(rle_bytes);
188 for (value, count) in &runs {
189 payload.extend_from_slice(&value.to_le_bytes());
190 payload.extend_from_slice(&count.to_le_bytes());
191 }
192 best_bytes = rle_bytes;
193 best_encoding = TopologyColumnEncoding::Rle;
194 best_payload = payload;
195 }
196 }
197
198 let mut out = Vec::with_capacity(5 + best_payload.len());
199 out.push(best_encoding as u8);
200 out.extend_from_slice(&(values.len() as u32).to_le_bytes());
201 out.extend_from_slice(&best_payload);
202 let choice = TopologyEncodingChoice {
203 encoding: best_encoding,
204 plain_bytes,
205 encoded_bytes: best_bytes,
206 };
207 (choice, out)
208}
209
210pub fn decode_topology_u64_column(data: &[u8], offset: &mut usize) -> StorageResult<Vec<u64>> {
213 if data.len().saturating_sub(*offset) < 5 {
214 return Err(StorageError::deserialize_error(
215 "topology column too short for header",
216 ));
217 }
218 let encoding = TopologyColumnEncoding::from_u8(data[*offset])?;
219 *offset += 1;
220 let count = u32::from_le_bytes(
221 data[*offset..*offset + 4]
222 .try_into()
223 .map_err(|_| StorageError::deserialize_error("topology column count too short"))?,
224 ) as usize;
225 *offset += 4;
226 match encoding {
227 TopologyColumnEncoding::Plain => {
228 let need = count.saturating_mul(8);
229 if data.len().saturating_sub(*offset) < need {
230 return Err(StorageError::deserialize_error(
231 "plain topology column too short",
232 ));
233 }
234 let mut out = Vec::with_capacity(count);
235 for _ in 0..count {
236 let value = u64::from_le_bytes(
237 data[*offset..*offset + 8]
238 .try_into()
239 .map_err(|_| StorageError::deserialize_error("plain value too short"))?,
240 );
241 *offset += 8;
242 out.push(value);
243 }
244 Ok(out)
245 }
246 TopologyColumnEncoding::BitPacked => {
247 let remaining = &data[*offset..];
248 let min_width_len = if count == 0 { 0 } else { 9 };
249 if remaining.len() < min_width_len {
250 return Err(StorageError::deserialize_error(
251 "bitpacked topology column too short",
252 ));
253 }
254 let width = if count == 0 { 1 } else { remaining[8] as usize };
255 let payload_words = if count == 0 {
256 0
257 } else {
258 (count.saturating_mul(width)).div_ceil(64)
259 };
260 let payload_len = min_width_len + payload_words.saturating_mul(8);
261 if remaining.len() < payload_len {
262 return Err(StorageError::deserialize_error(
263 "bitpacked topology column too short",
264 ));
265 }
266 let values = decode_bitpacked_payload(&remaining[..payload_len], count)?;
267 *offset += payload_len;
268 Ok(values)
269 }
270 TopologyColumnEncoding::Rle => {
271 let mut out = Vec::with_capacity(count);
272 while out.len() < count {
273 if data.len().saturating_sub(*offset) < 12 {
274 return Err(StorageError::deserialize_error(
275 "run-length topology column too short",
276 ));
277 }
278 let value = u64::from_le_bytes(
279 data[*offset..*offset + 8]
280 .try_into()
281 .map_err(|_| StorageError::deserialize_error("run value too short"))?,
282 );
283 *offset += 8;
284 let run = u32::from_le_bytes(
285 data[*offset..*offset + 4]
286 .try_into()
287 .map_err(|_| StorageError::deserialize_error("run length too short"))?,
288 ) as usize;
289 *offset += 4;
290 if run == 0 {
291 return Err(StorageError::deserialize_error(
292 "run-length topology run has zero length",
293 ));
294 }
295 if out.len().saturating_add(run) > count {
296 return Err(StorageError::deserialize_error(
297 "run-length topology column overruns declared count",
298 ));
299 }
300 out.extend(std::iter::repeat_n(value, run));
301 }
302 Ok(out)
303 }
304 }
305}
306
307fn rle_runs_u32(values: &[u32]) -> Vec<(u32, u32)> {
308 let mut runs: Vec<(u32, u32)> = Vec::new();
309 for &value in values {
310 match runs.last_mut() {
311 Some(last) if last.0 == value => {
312 last.1 = last.1.saturating_add(1);
313 }
314 _ => runs.push((value, 1)),
315 }
316 }
317 runs
318}
319
320fn bitpacked_payload_u32(values: &[u32]) -> Option<Vec<u8>> {
321 if values.is_empty() {
322 return Some(Vec::new());
323 }
324 let min = *values.iter().min()?;
325 let max = *values.iter().max()?;
326 let range = (max as u64).saturating_sub(min as u64);
327 let width = bit_width_for_range(range).max(1) as usize;
328 if width >= 32 {
329 return None;
330 }
331 let mut out = Vec::new();
332 out.extend_from_slice(&min.to_le_bytes());
333 out.push(width as u8);
334 let mut current: u64 = 0;
335 let mut filled: usize = 0;
336 for &value in values {
337 let adjusted = (value as u64).saturating_sub(min as u64);
338 for bit in 0..width {
339 if (adjusted >> bit) & 1 == 1 {
340 current |= 1u64 << filled;
341 }
342 filled += 1;
343 if filled == 64 {
344 out.extend_from_slice(¤t.to_le_bytes());
345 current = 0;
346 filled = 0;
347 }
348 }
349 }
350 if filled > 0 {
351 out.extend_from_slice(¤t.to_le_bytes());
352 }
353 Some(out)
354}
355
356fn decode_bitpacked_payload_u32(payload: &[u8], count: usize) -> StorageResult<Vec<u32>> {
357 if count == 0 {
358 if !payload.is_empty() {
359 return Err(StorageError::deserialize_error(
360 "non-empty bitpacked payload for empty topology column",
361 ));
362 }
363 return Ok(Vec::new());
364 }
365 if payload.len() < 5 {
366 return Err(StorageError::deserialize_error(
367 "bitpacked topology column too short",
368 ));
369 }
370 let min = u32::from_le_bytes(payload[0..4].try_into().map_err(|_| {
371 StorageError::deserialize_error("bitpacked topology column base too short")
372 })?);
373 let width = payload[4] as usize;
374 if width == 0 || width >= 32 {
375 return Err(StorageError::deserialize_error(format!(
376 "invalid bitpacked topology width: {}",
377 width
378 )));
379 }
380 let words: Vec<u64> = payload[5..]
381 .as_chunks::<8>()
382 .0
383 .iter()
384 .map(|chunk| u64::from_le_bytes(*chunk))
385 .collect();
386 if payload[5..].len() != words.len() * 8 {
387 return Err(StorageError::deserialize_error(
388 "bitpacked topology payload has trailing bytes",
389 ));
390 }
391 let mut out = Vec::with_capacity(count);
392 let mut bit_pos: usize = 0;
393 for _ in 0..count {
394 let mut adjusted: u64 = 0;
395 for bit in 0..width {
396 let word = bit_pos / 64;
397 let offset = bit_pos % 64;
398 let stored = words.get(word).copied().unwrap_or(0);
399 if (stored >> offset) & 1 == 1 {
400 adjusted |= 1u64 << bit;
401 }
402 bit_pos += 1;
403 }
404 out.push((min as u64).saturating_add(adjusted) as u32);
405 }
406 let expected_words = bit_pos.div_ceil(64);
407 if words.len() != expected_words {
408 return Err(StorageError::deserialize_error(
409 "bitpacked topology payload length mismatch",
410 ));
411 }
412 Ok(out)
413}
414
415pub fn encode_topology_u32_column(values: &[u32]) -> (TopologyEncodingChoice, Vec<u8>) {
422 let plain_bytes = values.len().saturating_mul(4);
423 let mut best_encoding = TopologyColumnEncoding::Plain;
424 let mut best_payload = Vec::with_capacity(plain_bytes);
425 for &value in values {
426 best_payload.extend_from_slice(&value.to_le_bytes());
427 }
428 let mut best_bytes = plain_bytes;
429
430 if let Some(packed) = bitpacked_payload_u32(values) {
431 if packed.len() < best_bytes {
432 best_bytes = packed.len();
433 best_encoding = TopologyColumnEncoding::BitPacked;
434 best_payload = packed;
435 }
436 }
437
438 if !values.is_empty() {
439 let runs = rle_runs_u32(values);
440 let rle_bytes = runs.len().saturating_mul(8);
441 if rle_bytes < best_bytes {
442 let mut payload = Vec::with_capacity(rle_bytes);
443 for (value, count) in &runs {
444 payload.extend_from_slice(&value.to_le_bytes());
445 payload.extend_from_slice(&count.to_le_bytes());
446 }
447 best_bytes = rle_bytes;
448 best_encoding = TopologyColumnEncoding::Rle;
449 best_payload = payload;
450 }
451 }
452
453 let mut out = Vec::with_capacity(5 + best_payload.len());
454 out.push(best_encoding as u8);
455 out.extend_from_slice(&(values.len() as u32).to_le_bytes());
456 out.extend_from_slice(&best_payload);
457 let choice = TopologyEncodingChoice {
458 encoding: best_encoding,
459 plain_bytes,
460 encoded_bytes: best_bytes,
461 };
462 (choice, out)
463}
464
465pub fn decode_topology_u32_column(data: &[u8], offset: &mut usize) -> StorageResult<Vec<u32>> {
467 if data.len().saturating_sub(*offset) < 5 {
468 return Err(StorageError::deserialize_error(
469 "topology column too short for header",
470 ));
471 }
472 let encoding = TopologyColumnEncoding::from_u8(data[*offset])?;
473 *offset += 1;
474 let count = u32::from_le_bytes(
475 data[*offset..*offset + 4]
476 .try_into()
477 .map_err(|_| StorageError::deserialize_error("topology column count too short"))?,
478 ) as usize;
479 *offset += 4;
480 match encoding {
481 TopologyColumnEncoding::Plain => {
482 let need = count.saturating_mul(4);
483 if data.len().saturating_sub(*offset) < need {
484 return Err(StorageError::deserialize_error(
485 "plain topology column too short",
486 ));
487 }
488 let mut out = Vec::with_capacity(count);
489 for _ in 0..count {
490 let value = u32::from_le_bytes(
491 data[*offset..*offset + 4]
492 .try_into()
493 .map_err(|_| StorageError::deserialize_error("plain value too short"))?,
494 );
495 *offset += 4;
496 out.push(value);
497 }
498 Ok(out)
499 }
500 TopologyColumnEncoding::BitPacked => {
501 let remaining = &data[*offset..];
502 let min_width_len = if count == 0 { 0 } else { 5 };
503 if remaining.len() < min_width_len {
504 return Err(StorageError::deserialize_error(
505 "bitpacked topology column too short",
506 ));
507 }
508 let width = if count == 0 { 1 } else { remaining[4] as usize };
509 let payload_words = if count == 0 {
510 0
511 } else {
512 (count.saturating_mul(width)).div_ceil(64)
513 };
514 let payload_len = min_width_len + payload_words.saturating_mul(8);
515 if remaining.len() < payload_len {
516 return Err(StorageError::deserialize_error(
517 "bitpacked topology column too short",
518 ));
519 }
520 let values = decode_bitpacked_payload_u32(&remaining[..payload_len], count)?;
521 *offset += payload_len;
522 Ok(values)
523 }
524 TopologyColumnEncoding::Rle => {
525 let mut out = Vec::with_capacity(count);
526 while out.len() < count {
527 if data.len().saturating_sub(*offset) < 8 {
528 return Err(StorageError::deserialize_error(
529 "run-length topology column too short",
530 ));
531 }
532 let value = u32::from_le_bytes(
533 data[*offset..*offset + 4]
534 .try_into()
535 .map_err(|_| StorageError::deserialize_error("run value too short"))?,
536 );
537 *offset += 4;
538 let run = u32::from_le_bytes(
539 data[*offset..*offset + 4]
540 .try_into()
541 .map_err(|_| StorageError::deserialize_error("run length too short"))?,
542 ) as usize;
543 *offset += 4;
544 if run == 0 {
545 return Err(StorageError::deserialize_error(
546 "run-length topology run has zero length",
547 ));
548 }
549 if out.len().saturating_add(run) > count {
550 return Err(StorageError::deserialize_error(
551 "run-length topology column overruns declared count",
552 ));
553 }
554 out.extend(std::iter::repeat_n(value, run));
555 }
556 Ok(out)
557 }
558 }
559}
560
561pub fn encode_topology_i64_column(values: &[i64]) -> (TopologyEncodingChoice, Vec<u8>) {
562 let wide: Vec<u64> = values.iter().map(|&v| v as u64).collect();
563 encode_topology_u64_column(&wide)
564}
565
566pub fn decode_topology_i64_column(data: &[u8], offset: &mut usize) -> StorageResult<Vec<i64>> {
567 let wide = decode_topology_u64_column(data, offset)?;
568 Ok(wide.into_iter().map(|v| v as i64).collect())
569}
570
571pub fn encode_overflow_chunk(chunk: &OverflowChunk, out: &mut Vec<u8>) {
579 out.extend_from_slice(&(chunk.len() as u32).to_le_bytes());
580 if chunk.is_empty() {
581 return;
582 }
583 let endpoints: Vec<u32> = chunk.hot_slice().iter().map(|hot| hot.endpoint).collect();
584 let ranks: Vec<i64> = chunk.hot_slice().iter().map(|hot| hot.rank).collect();
585 let edge_ids: Vec<u64> = chunk.hot_slice().iter().map(|hot| hot.edge_id.0).collect();
586 let deletes: Vec<u64> = chunk
587 .cold_slice()
588 .iter()
589 .map(|cold| cold.delete_ts)
590 .collect();
591 let (_, payload) = encode_topology_u32_column(&endpoints);
592 out.extend_from_slice(&payload);
593 let (_, payload) = encode_topology_i64_column(&ranks);
594 out.extend_from_slice(&payload);
595 let (_, payload) = encode_topology_u64_column(&edge_ids);
596 out.extend_from_slice(&payload);
597 let (_, payload) = encode_topology_u64_column(&deletes);
598 out.extend_from_slice(&payload);
599}
600
601pub fn decode_overflow_chunk(data: &[u8], offset: &mut usize) -> StorageResult<OverflowChunk> {
604 let chunk_len = read_u32_le(data, offset)? as usize;
605 if chunk_len == 0 {
606 return Ok(OverflowChunk::default());
607 }
608 let endpoints = decode_topology_u32_column(data, offset)?;
609 let ranks = decode_topology_i64_column(data, offset)?;
610 let edge_ids = decode_topology_u64_column(data, offset)?;
611 let deletes = decode_topology_u64_column(data, offset)?;
612 if endpoints.len() != chunk_len
613 || ranks.len() != chunk_len
614 || edge_ids.len() != chunk_len
615 || deletes.len() != chunk_len
616 {
617 return Err(StorageError::deserialize_error(
618 "overflow chunk column length mismatch",
619 ));
620 }
621 let mut out = OverflowChunk::with_capacity(chunk_len);
622 for index in 0..chunk_len {
623 let nbr = Nbr::with_timestamps(
624 endpoints[index],
625 ranks[index],
626 EdgeId(edge_ids[index]),
627 deletes[index],
628 );
629 out.push(nbr);
630 }
631 Ok(out)
632}
633
634#[cfg(test)]
635mod tests {
636 use super::*;
637
638 fn roundtrip_u64(values: &[u64]) -> Vec<u64> {
639 let (_, payload) = encode_topology_u64_column(values);
640 let mut offset = 0usize;
641 decode_topology_u64_column(&payload, &mut offset).expect("decode must succeed");
642 assert_eq!(offset, payload.len());
643 decode_topology_u64_column(&payload, &mut 0usize).expect("decode must succeed")
644 }
645
646 #[test]
647 fn topology_u64_roundtrip_small_range_uses_bitpacking() {
648 let values: Vec<u64> = (1000..1100).collect();
649 let (choice, _) = encode_topology_u64_column(&values);
650 assert_eq!(choice.encoding, TopologyColumnEncoding::BitPacked);
651 assert!(choice.encoded_bytes < choice.plain_bytes);
652 assert_eq!(roundtrip_u64(&values), values);
653 }
654
655 #[test]
656 fn topology_u64_roundtrip_runs_use_rle() {
657 let values = vec![7u64; 64];
658 let (choice, _) = encode_topology_u64_column(values.as_slice());
659 assert_eq!(choice.encoding, TopologyColumnEncoding::Rle);
660 assert_eq!(roundtrip_u64(&values), values);
661 }
662
663 #[test]
664 fn topology_u64_falls_back_to_plain_without_benefit() {
665 let values = vec![u64::MIN, u64::MAX, 1, u64::MAX - 1];
666 let (choice, _) = encode_topology_u64_column(values.as_slice());
667 assert_eq!(choice.encoding, TopologyColumnEncoding::Plain);
668 assert_eq!(roundtrip_u64(&values), values);
669 }
670
671 #[test]
672 fn topology_u32_and_i64_roundtrip() {
673 let narrow = vec![0u32, 1, 2, 3, 100];
674 let (_, payload) = encode_topology_u32_column(&narrow);
675 let mut offset = 0usize;
676 assert_eq!(
677 decode_topology_u32_column(&payload, &mut offset).unwrap(),
678 narrow
679 );
680 assert_eq!(offset, payload.len());
681
682 let ranks = vec![0i64, 0, 1, 1, 2];
683 let (_, payload) = encode_topology_i64_column(&ranks);
684 let mut offset = 0usize;
685 assert_eq!(
686 decode_topology_i64_column(&payload, &mut offset).unwrap(),
687 ranks
688 );
689 assert_eq!(offset, payload.len());
690 }
691
692 #[test]
693 fn topology_column_rejects_bad_encoding_and_trailing() {
694 let (_, mut payload) = encode_topology_u64_column(&[1u64, 2, 3]);
695 payload[0] = 99;
696 assert!(decode_topology_u64_column(&payload, &mut 0usize).is_err());
697
698 let (_, mut payload) = encode_topology_u64_column(&[1u64, 2, 3]);
699 payload.push(0);
700 let mut offset = 0usize;
701 decode_topology_u64_column(&payload, &mut offset).unwrap();
702 assert_eq!(offset, payload.len() - 1);
703
704 let short = vec![0u8; 3];
705 assert!(decode_topology_u64_column(&short, &mut 0usize).is_err());
706 }
707}