1use std::collections::BTreeMap;
7
8use marsdb_storage::{ReadableMultimapTable, ReadableTable, Txn};
9use serde::{Deserialize, Serialize};
10
11use crate::error::GraphError;
12use crate::labels::lookup_label_id;
13use crate::model::{NodeId, PropertyValue};
14use crate::props::lookup_prop_id;
15use crate::write_ctx::WriteCtx;
16
17#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
18pub struct IndexDef {
19 pub unique: bool,
20}
21
22fn index_prefix(label_id: u32, prop_id: u32) -> [u8; 8] {
26 let mut out = [0u8; 8];
27 out[0..4].copy_from_slice(&label_id.to_be_bytes());
28 out[4..8].copy_from_slice(&prop_id.to_be_bytes());
29 out
30}
31
32pub(crate) fn encode_index_value(v: &PropertyValue) -> Vec<u8> {
41 match v {
42 PropertyValue::Null => vec![0x00],
43 PropertyValue::Bool(b) => vec![0x01, u8::from(*b)],
44 PropertyValue::Int(i) => {
49 let mut out = vec![0x02];
50 out.extend_from_slice(&((*i as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
51 out
52 }
53 PropertyValue::Float(f) => {
57 let bits = f.to_bits();
58 let sortable = if bits & 0x8000_0000_0000_0000 != 0 {
59 !bits
60 } else {
61 bits | 0x8000_0000_0000_0000
62 };
63 let mut out = vec![0x03];
64 out.extend_from_slice(&sortable.to_be_bytes());
65 out
66 }
67 PropertyValue::String(s) => {
71 let mut out = vec![0x04];
72 out.extend_from_slice(s.as_bytes());
73 out
74 }
75 PropertyValue::Date(days) => {
76 let mut out = vec![0x05];
77 out.extend_from_slice(&((*days as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
78 out
79 }
80 PropertyValue::Duration {
81 months,
82 days,
83 seconds,
84 nanos,
85 } => {
86 let mut out = vec![0x06];
87 out.extend_from_slice(&months.to_be_bytes());
88 out.extend_from_slice(&days.to_be_bytes());
89 out.extend_from_slice(&seconds.to_be_bytes());
90 out.extend_from_slice(&nanos.to_be_bytes());
91 out
92 }
93 PropertyValue::LocalTime(nanos_of_day) => {
96 let mut out = vec![0x07];
97 out.extend_from_slice(&nanos_of_day.to_be_bytes());
98 out
99 }
100 PropertyValue::Time {
106 nanos_of_day,
107 offset_seconds,
108 } => {
109 let instant = nanos_of_day - *offset_seconds as i64 * 1_000_000_000;
110 let mut out = vec![0x08];
111 out.extend_from_slice(&((instant as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
112 out
113 }
114 PropertyValue::LocalDateTime {
115 epoch_seconds,
116 nanos,
117 } => {
118 let mut out = vec![0x09];
119 out.extend_from_slice(&((*epoch_seconds as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
120 out.extend_from_slice(&nanos.to_be_bytes());
121 out
122 }
123 PropertyValue::DateTime {
127 epoch_seconds,
128 nanos,
129 ..
130 } => {
131 let mut out = vec![0x0A];
132 out.extend_from_slice(&((*epoch_seconds as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
133 out.extend_from_slice(&nanos.to_be_bytes());
134 out
135 }
136 PropertyValue::List(items) => {
146 let mut out = vec![0x0B];
147 for item in items {
148 let encoded = encode_index_value(item);
149 out.extend_from_slice(&(encoded.len() as u32).to_be_bytes());
150 out.extend_from_slice(&encoded);
151 }
152 out
153 }
154 PropertyValue::Map(_) => {
159 unreachable!("PropertyValue::Map is never a real stored/indexed property value")
160 }
161 }
162}
163
164fn index_key(label_id: u32, prop_id: u32, value: &PropertyValue) -> Vec<u8> {
165 let mut out = index_prefix(label_id, prop_id).to_vec();
166 out.extend_from_slice(&encode_index_value(value));
167 out
168}
169
170pub(crate) fn create_index(
177 ctx: &mut WriteCtx,
178 label: &str,
179 prop: &str,
180 unique: bool,
181) -> Result<(), GraphError> {
182 let label_id = crate::labels::intern_label(ctx, label)?;
183 let prop_id = crate::props::intern_prop(ctx, prop)?;
184 let prefix = index_prefix(label_id, prop_id);
185 if ctx.index_defs()?.get(prefix.as_slice())?.is_some() {
186 return Err(GraphError::CorruptData(format!(
187 "index on label {label:?} property {prop:?} already exists"
188 )));
189 }
190
191 let node_ids: Vec<u64> = ctx
198 .node_label_index()?
199 .get(label_id)?
200 .map(|entry| entry.map(|value| value.value()).map_err(GraphError::from))
201 .collect::<Result<Vec<_>, GraphError>>()?;
202 let mut entries: Vec<(Vec<u8>, u64)> = Vec::with_capacity(node_ids.len());
203 for node_id in &node_ids {
204 let Some(guard) = ctx.nodes()?.get(*node_id)? else {
205 continue;
206 };
207 if let Some(raw) = crate::encode::node_prop_raw(guard.value(), prop_id)? {
211 let value = crate::encode::decode_value(raw)?;
212 entries.push((index_key(label_id, prop_id, &value), *node_id));
213 }
214 }
215 if unique {
216 let mut seen = std::collections::HashSet::with_capacity(entries.len());
217 for (key, _) in &entries {
218 if !seen.insert(key.clone()) {
219 return Err(GraphError::UniqueConstraintViolation {
220 label: label.to_string(),
221 property: prop.to_string(),
222 });
223 }
224 }
225 }
226
227 let encoded = postcard::to_allocvec(&IndexDef { unique })?;
228 ctx.index_defs()?
229 .insert(prefix.as_slice(), encoded.as_slice())?;
230 for (key, node_id) in entries {
231 ctx.property_index()?.insert(key.as_slice(), node_id)?;
232 }
233 Ok(())
234}
235
236pub fn lookup_index_def(txn: Txn, label: &str, prop: &str) -> Result<Option<IndexDef>, GraphError> {
238 let Some(label_id) = lookup_label_id(txn, label)? else {
239 return Ok(None);
240 };
241 let Some(prop_id) = lookup_prop_id(txn, prop)? else {
242 return Ok(None);
243 };
244 let prefix = index_prefix(label_id, prop_id);
245 let defs = txn.open_table(marsdb_storage::tables::INDEX_DEFS)?;
246 let found = defs
247 .get(prefix.as_slice())?
248 .map(|guard| guard.value().to_vec());
249 drop(defs);
250 match found {
251 Some(bytes) => Ok(Some(postcard::from_bytes(&bytes)?)),
252 None => Ok(None),
253 }
254}
255
256pub fn lookup_range(
283 txn: Txn,
284 label: &str,
285 prop: &str,
286 lo: Option<(&PropertyValue, bool)>,
287 hi: Option<(&PropertyValue, bool)>,
288 limit: Option<usize>,
289) -> Result<Vec<NodeId>, GraphError> {
290 let Some(mut cursor) = IndexRangeCursor::new(txn, label, prop, lo, hi)? else {
291 return Ok(Vec::new());
292 };
293 let mut out = Vec::new();
294 loop {
295 let want = match limit {
296 Some(l) => {
297 if out.len() >= l {
298 return Ok(out);
299 }
300 l - out.len()
301 }
302 None => usize::MAX,
303 };
304 let chunk = cursor.next_chunk(txn, want.min(4096))?;
305 if chunk.is_empty() {
306 return Ok(out);
307 }
308 out.extend(chunk);
309 }
310}
311
312type KeyRegion = (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>);
320
321pub struct IndexRangeCursor {
322 regions: Vec<KeyRegion>,
323 region_index: usize,
324 resume: Option<(Vec<u8>, u64)>,
328}
329
330impl IndexRangeCursor {
331 pub fn new(
333 txn: Txn,
334 label: &str,
335 prop: &str,
336 lo: Option<(&PropertyValue, bool)>,
337 hi: Option<(&PropertyValue, bool)>,
338 ) -> Result<Option<Self>, GraphError> {
339 let Some(label_id) = lookup_label_id(txn, label)? else {
340 return Ok(None);
341 };
342 let Some(prop_id) = lookup_prop_id(txn, prop)? else {
343 return Ok(None);
344 };
345 let prefix = index_prefix(label_id, prop_id);
346 Ok(Some(Self {
347 regions: range_regions(&prefix, lo, hi),
348 region_index: 0,
349 resume: None,
350 }))
351 }
352
353 pub fn next_chunk(&mut self, txn: Txn, chunk_size: usize) -> Result<Vec<NodeId>, GraphError> {
354 use std::ops::Bound;
355 if chunk_size == 0 {
356 return Ok(Vec::new());
357 }
358 let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
359 let mut out = Vec::new();
360 while self.region_index < self.regions.len() {
361 let (region_start, region_end) = &self.regions[self.region_index];
362 let start_owned;
366 let start_bound: Bound<&[u8]> = match &self.resume {
367 Some((key, _)) => {
368 start_owned = key.clone();
369 Bound::Included(start_owned.as_slice())
370 }
371 None => match region_start {
372 Bound::Included(k) => Bound::Included(k.as_slice()),
373 Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
374 Bound::Unbounded => Bound::Unbounded,
375 },
376 };
377 let end_bound: Bound<&[u8]> = match region_end {
378 Bound::Included(k) => Bound::Included(k.as_slice()),
379 Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
380 Bound::Unbounded => Bound::Unbounded,
381 };
382 for entry in index.range::<&[u8]>((start_bound, end_bound))? {
383 let (key, values) = entry?;
384 let key_bytes = key.value().to_vec();
385 let skip_through = match &self.resume {
386 Some((resume_key, resume_val)) if *resume_key == key_bytes => Some(*resume_val),
387 _ => None,
388 };
389 for value in values {
390 let node = value?.value();
391 if skip_through.is_some_and(|last| node <= last) {
392 continue;
393 }
394 out.push(NodeId(node));
395 self.resume = Some((key_bytes.clone(), node));
396 if out.len() >= chunk_size {
397 return Ok(out);
398 }
399 }
400 }
401 self.region_index += 1;
403 self.resume = None;
404 }
405 Ok(out)
406 }
407}
408
409fn range_regions(
413 prefix: &[u8],
414 lo: Option<(&PropertyValue, bool)>,
415 hi: Option<(&PropertyValue, bool)>,
416) -> Vec<KeyRegion> {
417 use std::ops::Bound;
418 let key = |value: &PropertyValue| {
419 let mut k = prefix.to_vec();
420 k.extend_from_slice(&encode_index_value(value));
421 k
422 };
423 let tag_start = |tag: u8| {
424 let mut k = prefix.to_vec();
425 k.push(tag);
426 Bound::Included(k)
427 };
428 let tag_end = |tag: u8| {
429 let mut k = prefix.to_vec();
430 k.push(tag + 1);
431 Bound::Excluded(k)
432 };
433 let numeric = |v: &PropertyValue| matches!(v, PropertyValue::Int(_) | PropertyValue::Float(_));
434
435 let is_numeric = lo.map(|(v, _)| numeric(v)).unwrap_or(true)
436 && hi.map(|(v, _)| numeric(v)).unwrap_or(true)
437 && (lo.is_some() || hi.is_some())
438 && (lo.is_some_and(|(v, _)| numeric(v)) || hi.is_some_and(|(v, _)| numeric(v)));
439 if is_numeric {
440 let int_bound = |side_lo: bool, bound: Option<(&PropertyValue, bool)>| match bound {
444 None => {
445 if side_lo {
446 tag_start(0x02)
447 } else {
448 tag_end(0x02)
449 }
450 }
451 Some((PropertyValue::Int(i), inclusive)) => {
452 let k = key(&PropertyValue::Int(*i));
453 if inclusive {
454 Bound::Included(k)
455 } else {
456 Bound::Excluded(k)
457 }
458 }
459 Some((PropertyValue::Float(f), _)) => {
460 let widened = if side_lo { f.floor() } else { f.ceil() };
463 let clamped = widened.clamp(i64::MIN as f64, i64::MAX as f64) as i64;
464 Bound::Included(key(&PropertyValue::Int(clamped)))
465 }
466 Some(_) => unreachable!("numeric region only built for numeric bounds"),
467 };
468 let float_bound = |side_lo: bool, bound: Option<(&PropertyValue, bool)>| match bound {
469 None => {
470 if side_lo {
471 tag_start(0x03)
472 } else {
473 tag_end(0x03)
474 }
475 }
476 Some((PropertyValue::Float(f), inclusive)) => {
477 let k = key(&PropertyValue::Float(*f));
478 if inclusive {
479 Bound::Included(k)
480 } else {
481 Bound::Excluded(k)
482 }
483 }
484 Some((PropertyValue::Int(i), _)) => {
485 let f = *i as f64;
491 let step = f.abs() * (2.0 * f64::EPSILON) + f64::MIN_POSITIVE;
492 let widened = if side_lo { f - step } else { f + step };
493 Bound::Included(key(&PropertyValue::Float(widened)))
494 }
495 Some(_) => unreachable!("numeric region only built for numeric bounds"),
496 };
497 return vec![
498 (int_bound(true, lo), int_bound(false, hi)),
499 (float_bound(true, lo), float_bound(false, hi)),
500 ];
501 }
502
503 let tag = lo
508 .or(hi)
509 .map(|(v, _)| encode_index_value(v)[0])
510 .unwrap_or(0x00);
511 let start = match lo {
512 None => tag_start(tag),
513 Some((v, true)) => Bound::Included(key(v)),
514 Some((v, false)) => Bound::Excluded(key(v)),
515 };
516 let end = match hi {
517 None => tag_end(tag),
518 Some((v, true)) => Bound::Included(key(v)),
519 Some((v, false)) => Bound::Excluded(key(v)),
520 };
521 vec![(start, end)]
522}
523
524pub fn lookup_exact(
525 txn: Txn,
526 label: &str,
527 prop: &str,
528 value: &PropertyValue,
529 limit: Option<usize>,
530) -> Result<Vec<NodeId>, GraphError> {
531 let Some(label_id) = lookup_label_id(txn, label)? else {
532 return Ok(Vec::new());
533 };
534 let Some(prop_id) = lookup_prop_id(txn, prop)? else {
535 return Ok(Vec::new());
536 };
537 let key = index_key(label_id, prop_id, value);
538 let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
539 let iter = index.get(key.as_slice())?;
540 let ids: Vec<NodeId> = match limit {
541 Some(limit) => iter
542 .take(limit)
543 .map(|entry| {
544 entry
545 .map(|value| NodeId(value.value()))
546 .map_err(GraphError::from)
547 })
548 .collect::<Result<Vec<_>, GraphError>>()?,
549 None => iter
550 .map(|entry| {
551 entry
552 .map(|value| NodeId(value.value()))
553 .map_err(GraphError::from)
554 })
555 .collect::<Result<Vec<_>, GraphError>>()?,
556 };
557 drop(index);
558 Ok(ids)
559}
560
561pub fn match_count(
570 txn: Txn,
571 label: &str,
572 prop: &str,
573 value: &PropertyValue,
574) -> Result<u64, GraphError> {
575 let Some(label_id) = lookup_label_id(txn, label)? else {
576 return Ok(0);
577 };
578 let Some(prop_id) = lookup_prop_id(txn, prop)? else {
579 return Ok(0);
580 };
581 let key = index_key(label_id, prop_id, value);
582 let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
583 let count = index.get(key.as_slice())?.len();
584 Ok(count)
585}
586
587fn indexes_for_labels(
593 ctx: &mut WriteCtx,
594 label_ids: &[u32],
595) -> Result<Vec<(u32, u32, String, IndexDef)>, GraphError> {
596 let raw: Vec<(u32, u32, IndexDef)> = {
602 let mut raw = Vec::new();
603 for entry in ctx.index_defs()?.iter()? {
604 let (key, value) = entry?;
605 let key_bytes = key.value();
606 let label_id = u32::from_be_bytes(
607 key_bytes[0..4]
608 .try_into()
609 .expect("index key prefix is 8 bytes"),
610 );
611 if !label_ids.contains(&label_id) {
612 continue;
613 }
614 let prop_id = u32::from_be_bytes(
615 key_bytes[4..8]
616 .try_into()
617 .expect("index key prefix is 8 bytes"),
618 );
619 let def: IndexDef = postcard::from_bytes(value.value())?;
620 raw.push((label_id, prop_id, def));
621 }
622 raw
623 };
624 raw.into_iter()
625 .map(|(label_id, prop_id, def)| {
626 let prop_name = resolve_prop_ctx(ctx, prop_id)?;
627 Ok((label_id, prop_id, prop_name, def))
628 })
629 .collect()
630}
631
632pub(crate) fn resolve_label_ctx(ctx: &mut WriteCtx, label_id: u32) -> Result<String, GraphError> {
640 let value = ctx.id_to_label()?.get(label_id)?.ok_or_else(|| {
641 GraphError::CorruptData(format!("label id {label_id} has no interned string"))
642 })?;
643 Ok(value.value().to_string())
644}
645
646pub(crate) fn resolve_prop_ctx(ctx: &mut WriteCtx, prop_id: u32) -> Result<String, GraphError> {
647 let value = ctx.id_to_prop()?.get(prop_id)?.ok_or_else(|| {
648 GraphError::CorruptData(format!("prop id {prop_id} has no interned string"))
649 })?;
650 Ok(value.value().to_string())
651}
652
653struct IndexTarget<'a> {
658 label_id: u32,
659 prop_id: u32,
660 label: &'a str,
661 prop: &'a str,
662}
663
664fn insert_entry(
665 ctx: &mut WriteCtx,
666 target: &IndexTarget<'_>,
667 value: &PropertyValue,
668 node_id: u64,
669 unique: bool,
670) -> Result<(), GraphError> {
671 let key = index_key(target.label_id, target.prop_id, value);
672 if unique && ctx.property_index()?.get(key.as_slice())?.next().is_some() {
673 return Err(GraphError::UniqueConstraintViolation {
674 label: target.label.to_string(),
675 property: target.prop.to_string(),
676 });
677 }
678 ctx.property_index()?.insert(key.as_slice(), node_id)?;
679 Ok(())
680}
681
682fn remove_entry(
683 ctx: &mut WriteCtx,
684 label_id: u32,
685 prop_id: u32,
686 value: &PropertyValue,
687 node_id: u64,
688) -> Result<(), GraphError> {
689 let key = index_key(label_id, prop_id, value);
690 ctx.property_index()?.remove(key.as_slice(), node_id)?;
691 Ok(())
692}
693
694pub(crate) fn on_node_created(
701 ctx: &mut WriteCtx,
702 node_id: u64,
703 label_ids: &[u32],
704 props: &BTreeMap<String, PropertyValue>,
705) -> Result<(), GraphError> {
706 for (label_id, prop_id, prop_name, def) in indexes_for_labels(ctx, label_ids)? {
707 if let Some(value) = props.get(&prop_name) {
708 let label = resolve_label_ctx(ctx, label_id)?;
709 let target = IndexTarget {
710 label_id,
711 prop_id,
712 label: &label,
713 prop: &prop_name,
714 };
715 insert_entry(ctx, &target, value, node_id, def.unique)?;
716 }
717 }
718 Ok(())
719}
720
721pub(crate) fn on_node_deleted(
727 ctx: &mut WriteCtx,
728 node_id: u64,
729 label_ids: &[u32],
730 props: &BTreeMap<String, PropertyValue>,
731) -> Result<(), GraphError> {
732 for (label_id, prop_id, prop_name, _def) in indexes_for_labels(ctx, label_ids)? {
733 if let Some(value) = props.get(&prop_name) {
734 remove_entry(ctx, label_id, prop_id, value, node_id)?;
735 }
736 }
737 Ok(())
738}
739
740pub(crate) fn on_node_prop_changed(
750 ctx: &mut WriteCtx,
751 node_id: u64,
752 label_ids: &[u32],
753 prop: &str,
754 old_value: Option<&PropertyValue>,
755 new_value: Option<&PropertyValue>,
756) -> Result<(), GraphError> {
757 for (label_id, prop_id, prop_name, def) in indexes_for_labels(ctx, label_ids)? {
758 if prop_name != prop {
759 continue;
760 }
761 if let Some(old) = old_value {
762 remove_entry(ctx, label_id, prop_id, old, node_id)?;
763 }
764 if let Some(new) = new_value {
765 let label = resolve_label_ctx(ctx, label_id)?;
766 let target = IndexTarget {
767 label_id,
768 prop_id,
769 label: &label,
770 prop: &prop_name,
771 };
772 insert_entry(ctx, &target, new, node_id, def.unique)?;
773 }
774 }
775 Ok(())
776}