1use reddb_types::Value;
3use std::collections::HashMap;
4use std::sync::Arc;
5
6mod memory;
7
8#[derive(Debug, Clone)]
10pub struct ExecutionError {
11 pub message: String,
12}
13
14impl ExecutionError {
15 pub fn new(message: impl Into<String>) -> Self {
16 Self {
17 message: message.into(),
18 }
19 }
20}
21
22impl std::fmt::Display for ExecutionError {
23 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24 write!(f, "Execution error: {}", self.message)
25 }
26}
27
28impl std::error::Error for ExecutionError {}
29
30#[derive(Debug, Clone, Default)]
32pub struct UnifiedResult {
33 pub columns: Vec<String>,
35 pub records: Vec<UnifiedRecord>,
37 pub stats: QueryStats,
39 pub pre_serialized_json: Option<String>,
41}
42
43impl UnifiedResult {
44 pub fn empty() -> Self {
46 Self::default()
47 }
48
49 pub fn with_columns(columns: Vec<String>) -> Self {
51 Self {
52 columns,
53 records: Vec::new(),
54 stats: QueryStats::default(),
55 pre_serialized_json: None,
56 }
57 }
58
59 pub fn push(&mut self, record: UnifiedRecord) {
61 self.records.push(record);
62 }
63
64 pub fn len(&self) -> usize {
66 self.records.len()
67 }
68
69 pub fn is_empty(&self) -> bool {
71 self.records.is_empty()
72 }
73}
74
75#[derive(Debug, Clone)]
96pub struct UnifiedRecord {
97 schema: Arc<Vec<Arc<str>>>,
100 values: Vec<Value>,
102 overflow: Option<HashMap<Arc<str>, Value>>,
104 pub nodes: HashMap<String, MatchedNode>,
106 pub edges: HashMap<String, MatchedEdge>,
108 pub paths: Vec<GraphPath>,
110 pub vector_results: Vec<VectorSearchResult>,
112}
113
114fn empty_schema() -> Arc<Vec<Arc<str>>> {
119 static EMPTY: std::sync::OnceLock<Arc<Vec<Arc<str>>>> = std::sync::OnceLock::new();
120 Arc::clone(EMPTY.get_or_init(|| Arc::new(Vec::new())))
121}
122
123impl Default for UnifiedRecord {
124 fn default() -> Self {
125 Self {
126 schema: empty_schema(),
127 values: Vec::new(),
128 overflow: None,
129 nodes: HashMap::new(),
130 edges: HashMap::new(),
131 paths: Vec::new(),
132 vector_results: Vec::new(),
133 }
134 }
135}
136
137#[derive(Debug, Clone, Copy)]
143pub struct ColumnarRow<'a> {
144 pub schema: &'a Arc<Vec<Arc<str>>>,
145 pub values: &'a [Value],
146}
147
148pub fn sys_key_rid() -> Arc<str> {
153 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
154 Arc::clone(KEY.get_or_init(|| Arc::from("rid")))
155}
156
157pub fn sys_key_collection() -> Arc<str> {
158 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
159 Arc::clone(KEY.get_or_init(|| Arc::from("collection")))
160}
161
162pub fn sys_key_kind() -> Arc<str> {
163 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
164 Arc::clone(KEY.get_or_init(|| Arc::from("kind")))
165}
166
167pub fn sys_key_tenant() -> Arc<str> {
168 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
169 Arc::clone(KEY.get_or_init(|| Arc::from("tenant")))
170}
171
172pub fn sys_key_created_at() -> Arc<str> {
173 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
174 Arc::clone(KEY.get_or_init(|| Arc::from("created_at")))
175}
176
177pub fn sys_key_updated_at() -> Arc<str> {
178 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
179 Arc::clone(KEY.get_or_init(|| Arc::from("updated_at")))
180}
181
182pub fn sys_key_row_id() -> Arc<str> {
183 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
184 Arc::clone(KEY.get_or_init(|| Arc::from("row_id")))
185}
186
187pub fn sys_key_red_collection() -> Arc<str> {
188 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
189 Arc::clone(KEY.get_or_init(|| Arc::from("red_collection")))
190}
191
192pub fn sys_key_red_kind() -> Arc<str> {
193 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
194 Arc::clone(KEY.get_or_init(|| Arc::from("red_kind")))
195}
196
197pub fn sys_key_red_sequence_id() -> Arc<str> {
198 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
199 Arc::clone(KEY.get_or_init(|| Arc::from("red_sequence_id")))
200}
201
202pub fn sys_key_red_entity_type() -> Arc<str> {
203 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
204 Arc::clone(KEY.get_or_init(|| Arc::from("red_entity_type")))
205}
206
207pub fn sys_key_red_capabilities() -> Arc<str> {
208 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
209 Arc::clone(KEY.get_or_init(|| Arc::from("red_capabilities")))
210}
211
212impl UnifiedRecord {
213 pub fn new() -> Self {
216 Self::default()
217 }
218
219 pub fn with_capacity(capacity: usize) -> Self {
223 let mut rec = Self::default();
224 if capacity > 0 {
225 rec.overflow = Some(HashMap::with_capacity(capacity));
226 }
227 rec
228 }
229
230 pub fn with_schema(schema: Arc<Vec<Arc<str>>>, values: Vec<Value>) -> Self {
241 Self {
242 schema,
243 values,
244 overflow: None,
245 nodes: HashMap::new(),
246 edges: HashMap::new(),
247 paths: Vec::new(),
248 vector_results: Vec::new(),
249 }
250 }
251
252 #[inline]
255 pub fn from_columnar(schema: Arc<Vec<Arc<str>>>, values: Vec<Value>) -> Self {
256 Self::with_schema(schema, values)
257 }
258
259 pub fn schemaless() -> Self {
263 Self::default()
264 }
265
266 #[inline]
269 pub fn columns(&self) -> &[Arc<str>] {
270 self.schema.as_ref()
271 }
272
273 #[inline]
276 pub fn schema_values(&self) -> &[Value] {
277 &self.values
278 }
279
280 #[inline]
282 pub fn overflow(&self) -> Option<&HashMap<Arc<str>, Value>> {
283 self.overflow.as_ref()
284 }
285
286 #[inline]
293 fn schema_index(&self, column: &str) -> Option<usize> {
294 self.schema.iter().rposition(|k| &**k == column)
295 }
296
297 pub fn set(&mut self, column: &str, value: Value) {
300 if let Some(idx) = self.schema_index(column) {
301 if idx < self.values.len() {
302 self.values[idx] = value;
303 return;
304 }
305 }
306 self.overflow_mut().insert(Arc::from(column), value);
307 }
308
309 #[inline]
313 pub fn set_owned(&mut self, column: String, value: Value) {
314 if let Some(idx) = self.schema_index(&column) {
315 if idx < self.values.len() {
316 self.values[idx] = value;
317 return;
318 }
319 }
320 self.overflow_mut()
321 .insert(Arc::from(column.into_boxed_str()), value);
322 }
323
324 #[inline]
329 pub fn set_arc(&mut self, column: Arc<str>, value: Value) {
330 if let Some(idx) = self.schema_index(&column) {
331 if idx < self.values.len() {
332 self.values[idx] = value;
333 return;
334 }
335 }
336 self.overflow_mut().insert(column, value);
337 }
338
339 #[inline]
347 pub fn set_arc_if_absent(&mut self, column: Arc<str>, value: Value) {
348 if self.get(&column).is_none() {
349 self.set_arc(column, value);
350 }
351 }
352
353 pub fn get(&self, column: &str) -> Option<&Value> {
364 if let Some(idx) = self.schema_index(column) {
365 if let Some(v) = self.values.get(idx) {
366 return Some(v);
367 }
368 }
369 self.overflow.as_ref().and_then(|m| m.get(column))
370 }
371
372 pub fn field_count(&self) -> usize {
374 self.values.len() + self.overflow.as_ref().map(|m| m.len()).unwrap_or(0)
375 }
376
377 pub fn contains_column(&self, column: &str) -> bool {
379 self.get(column).is_some()
380 }
381
382 pub fn iter_fields(&self) -> Box<dyn Iterator<Item = (&Arc<str>, &Value)> + '_> {
387 let schema_iter = self.schema.iter().zip(self.values.iter());
388 let overflow_iter = self.overflow.iter().flat_map(|m| m.iter());
389 Box::new(schema_iter.chain(overflow_iter))
390 }
391
392 pub fn values_mut(&mut self) -> Box<dyn Iterator<Item = &mut Value> + '_> {
396 let schema_iter = self.values.iter_mut();
397 let overflow_iter = self.overflow.iter_mut().flat_map(|m| m.values_mut());
398 Box::new(schema_iter.chain(overflow_iter))
399 }
400
401 pub fn column_names(&self) -> Vec<Arc<str>> {
403 let mut out: Vec<Arc<str>> = Vec::with_capacity(self.field_count());
404 out.extend(self.schema.iter().cloned());
405 if let Some(m) = &self.overflow {
406 out.extend(m.keys().cloned());
407 }
408 out
409 }
410
411 #[inline]
418 pub fn columnar(&self) -> Option<ColumnarRow<'_>> {
419 if self.schema.is_empty() {
420 None
421 } else {
422 Some(ColumnarRow {
423 schema: &self.schema,
424 values: &self.values,
425 })
426 }
427 }
428
429 #[inline]
435 pub fn columnar_schema(&self) -> Option<&Arc<Vec<Arc<str>>>> {
436 if self.schema.is_empty() {
437 None
438 } else {
439 Some(&self.schema)
440 }
441 }
442
443 pub fn overflow_mut(&mut self) -> &mut HashMap<Arc<str>, Value> {
447 self.overflow.get_or_insert_with(HashMap::new)
448 }
449
450 pub fn overflow_entry_or_insert(&mut self, column: Arc<str>, value: Value) {
453 self.overflow_mut().entry(column).or_insert(value);
454 }
455
456 pub fn set_node(&mut self, alias: &str, node: MatchedNode) {
458 self.nodes.insert(alias.to_string(), node);
459 }
460
461 pub fn get_node(&self, alias: &str) -> Option<&MatchedNode> {
463 self.nodes.get(alias)
464 }
465
466 pub fn set_edge(&mut self, alias: &str, edge: MatchedEdge) {
468 self.edges.insert(alias.to_string(), edge);
469 }
470}
471
472#[derive(Debug, Clone)]
474pub struct MatchedNode {
475 pub id: String,
476 pub label: String,
477 pub node_label: String,
479 pub properties: HashMap<String, Value>,
480}
481
482impl MatchedNode {
483 pub fn from_stored(node: impl Into<Self>) -> Self {
486 node.into()
487 }
488}
489
490#[derive(Debug, Clone)]
492pub struct MatchedEdge {
493 pub from: String,
494 pub to: String,
495 pub edge_label: String,
497 pub weight: f32,
498 pub properties: HashMap<String, Value>,
499}
500
501impl MatchedEdge {
502 pub fn from_tuple(
505 source: &str,
506 edge_label: impl Into<String>,
507 target: &str,
508 weight: f32,
509 ) -> Self {
510 Self {
511 from: source.to_string(),
512 to: target.to_string(),
513 edge_label: edge_label.into(),
514 weight,
515 properties: HashMap::new(),
516 }
517 }
518}
519
520#[derive(Debug, Clone)]
522pub struct VectorSearchResult {
523 pub id: u64,
525 pub collection: String,
527 pub distance: f32,
529 pub vector: Option<Vec<f32>>,
531 pub metadata: Option<HashMap<String, Value>>,
533 pub linked_node: Option<String>,
535 pub linked_row: Option<(String, u64)>,
537}
538
539impl VectorSearchResult {
540 pub fn new(id: u64, collection: impl Into<String>, distance: f32) -> Self {
542 Self {
543 id,
544 collection: collection.into(),
545 distance,
546 vector: None,
547 metadata: None,
548 linked_node: None,
549 linked_row: None,
550 }
551 }
552
553 pub fn with_vector(mut self, vector: Vec<f32>) -> Self {
555 self.vector = Some(vector);
556 self
557 }
558
559 pub fn with_metadata(mut self, metadata: HashMap<String, Value>) -> Self {
561 self.metadata = Some(metadata);
562 self
563 }
564
565 pub fn with_linked_node(mut self, node_id: impl Into<String>) -> Self {
567 self.linked_node = Some(node_id.into());
568 self
569 }
570
571 pub fn with_linked_row(mut self, table: impl Into<String>, row_id: u64) -> Self {
573 self.linked_row = Some((table.into(), row_id));
574 self
575 }
576}
577
578impl Default for VectorSearchResult {
579 fn default() -> Self {
580 Self::new(0, String::new(), 0.0)
581 }
582}
583
584#[derive(Debug, Clone)]
586pub struct GraphPath {
587 pub nodes: Vec<String>,
589 pub edges: Vec<MatchedEdge>,
591 pub total_weight: f32,
593}
594
595impl GraphPath {
596 pub fn start(node_id: &str) -> Self {
598 Self {
599 nodes: vec![node_id.to_string()],
600 edges: Vec::new(),
601 total_weight: 0.0,
602 }
603 }
604
605 pub fn extend(&self, edge: MatchedEdge, node_id: &str) -> Self {
607 let mut new_path = self.clone();
608 new_path.total_weight += edge.weight;
609 new_path.edges.push(edge);
610 new_path.nodes.push(node_id.to_string());
611 new_path
612 }
613
614 pub fn len(&self) -> usize {
616 self.edges.len()
617 }
618
619 pub fn is_empty(&self) -> bool {
621 self.edges.is_empty()
622 }
623}
624
625#[derive(Debug, Clone, Default)]
627pub struct QueryStats {
628 pub nodes_scanned: u64,
630 pub edges_scanned: u64,
632 pub rows_scanned: u64,
634 pub segments_total: u64,
636 pub segments_scanned: u64,
638 pub segments_pruned: u64,
640 pub exec_time_us: u64,
642 pub vector: Option<VectorQueryStats>,
644}
645
646impl QueryStats {
647 pub fn merge(a: &QueryStats, b: &QueryStats) -> QueryStats {
649 QueryStats {
650 nodes_scanned: a.nodes_scanned + b.nodes_scanned,
651 edges_scanned: a.edges_scanned + b.edges_scanned,
652 rows_scanned: a.rows_scanned + b.rows_scanned,
653 segments_total: a.segments_total + b.segments_total,
654 segments_scanned: a.segments_scanned + b.segments_scanned,
655 segments_pruned: a.segments_pruned + b.segments_pruned,
656 exec_time_us: a.exec_time_us + b.exec_time_us,
657 vector: None, }
659 }
660}
661
662#[derive(Debug, Clone, Default)]
665pub struct VectorOperatorStats {
666 pub operator: String,
667 pub input_rows: u64,
668 pub output_rows: u64,
669 pub inclusive_time_us: u64,
670}
671
672#[derive(Debug, Clone, Default)]
674pub struct VectorQueryStats {
675 pub operators: Vec<VectorOperatorStats>,
676 pub mode_requested: String,
677 pub mode_executed: String,
678 pub fallback_reason: Option<String>,
679 pub approximate_distance_evaluations: u64,
680 pub peak_topk_entries: u64,
681 pub cache_hit: bool,
683 pub access_path: String,
684 pub index_used: bool,
685 pub candidates_examined: u64,
686 pub metadata_rejected: u64,
687 pub rls_rejected: u64,
688 pub visibility_rejected: u64,
689 pub exact_distance_evaluations: u64,
690 pub rows_returned: u64,
691}
692
693impl VectorQueryStats {
694 pub fn operators_json(&self) -> crate::json::Value {
695 use crate::json::{Map, Value as JsonValue};
696 JsonValue::Array(
697 self.operators
698 .iter()
699 .map(|operator| {
700 let mut fields = Map::new();
701 fields.insert(
702 "operator".into(),
703 JsonValue::String(operator.operator.clone()),
704 );
705 for (name, value) in [
706 ("input_rows", operator.input_rows),
707 ("output_rows", operator.output_rows),
708 ("inclusive_time_us", operator.inclusive_time_us),
709 ] {
710 fields.insert(name.into(), Value::UnsignedInteger(value).to_json());
711 }
712 JsonValue::Object(fields)
713 })
714 .collect(),
715 )
716 }
717}
718
719#[cfg(all(test, not(target_arch = "wasm32")))]
720mod record_layout_tests {
721 use super::*;
741 use proptest::collection::vec;
742 use proptest::prelude::*;
743 use reddb_types::Value;
744
745 fn arb_value() -> impl Strategy<Value = Value> {
746 prop_oneof![
747 Just(Value::Null),
748 any::<bool>().prop_map(Value::Boolean),
749 any::<i64>().prop_map(Value::Integer),
750 any::<f64>()
751 .prop_filter("nan-free", |v| !v.is_nan())
752 .prop_map(Value::Float),
753 "[a-zA-Z0-9 ]{0,16}".prop_map(Value::text),
754 ]
755 }
756
757 fn arb_column_name() -> impl Strategy<Value = String> {
758 "[a-d]{1,3}".prop_map(|s| s.to_string())
761 }
762
763 fn arb_schema() -> impl Strategy<Value = Vec<String>> {
764 vec(arb_column_name(), 0..6).prop_map(|mut v| {
768 v.sort();
769 v.dedup();
770 v
771 })
772 }
773
774 fn arb_writes() -> impl Strategy<Value = Vec<(String, Value)>> {
775 vec((arb_column_name(), arb_value()), 0..16)
776 }
777
778 fn record_with_schema(schema: &[String], writes: &[(String, Value)]) -> UnifiedRecord {
783 let arc_schema: Arc<Vec<Arc<str>>> =
784 Arc::new(schema.iter().map(|s| Arc::from(s.as_str())).collect());
785 let mut rec =
789 UnifiedRecord::with_schema(Arc::clone(&arc_schema), vec![Value::Null; schema.len()]);
790 for (k, v) in writes {
791 rec.set_owned(k.clone(), v.clone());
792 }
793 rec
794 }
795
796 fn hashmap_reference(
801 schema: &[String],
802 writes: &[(String, Value)],
803 ) -> std::collections::HashMap<String, Value> {
804 let mut map: std::collections::HashMap<String, Value> =
805 schema.iter().map(|c| (c.clone(), Value::Null)).collect();
806 for (k, v) in writes {
807 map.insert(k.clone(), v.clone());
808 }
809 map
810 }
811
812 proptest! {
813 #[test]
817 fn get_set_round_trip(schema in arb_schema(), writes in arb_writes()) {
818 let rec = record_with_schema(&schema, &writes);
819 let reference = hashmap_reference(&schema, &writes);
820 for (k, expected) in &reference {
821 prop_assert_eq!(rec.get(k), Some(expected), "key {}", k);
822 }
823 }
824
825 #[test]
830 fn overflow_promotion_only_for_missing_keys(
831 schema in arb_schema(),
832 writes in arb_writes(),
833 ) {
834 let rec = record_with_schema(&schema, &writes);
835
836 for col in &schema {
838 if let Some(over) = rec.overflow() {
839 prop_assert!(
840 !over.contains_key(col.as_str()),
841 "schema column {} leaked into overflow",
842 col
843 );
844 }
845 }
846
847 for (k, _) in &writes {
849 if !schema.iter().any(|c| c == k) {
850 let over = rec.overflow().expect("overflow allocated");
851 prop_assert!(
852 over.contains_key(k.as_str()),
853 "out-of-schema key {} missing from overflow",
854 k
855 );
856 }
857 }
858 }
859
860 #[test]
864 fn iter_fields_matches_hashmap(
865 schema in arb_schema(),
866 writes in arb_writes(),
867 ) {
868 let rec = record_with_schema(&schema, &writes);
869 let reference = hashmap_reference(&schema, &writes);
870
871 let mut got: Vec<(String, Value)> = rec
872 .iter_fields()
873 .map(|(k, v)| (k.as_ref().to_string(), v.clone()))
874 .collect();
875 let mut want: Vec<(String, Value)> = reference
876 .into_iter()
877 .collect();
878
879 got.sort_by(|a, b| a.0.cmp(&b.0));
880 want.sort_by(|a, b| a.0.cmp(&b.0));
881 prop_assert_eq!(got, want);
882 }
883
884 #[test]
890 fn shared_schema_scan_parity(
891 schema in arb_schema().prop_filter("non-empty schema", |s| !s.is_empty()),
892 rows in vec(arb_writes(), 0..8),
893 ) {
894 let arc_schema: Arc<Vec<Arc<str>>> =
896 Arc::new(schema.iter().map(|s| Arc::from(s.as_str())).collect());
897 let shared: Vec<UnifiedRecord> = rows
898 .iter()
899 .map(|row_writes| {
900 let mut r = UnifiedRecord::with_schema(
901 Arc::clone(&arc_schema),
902 vec![Value::Null; schema.len()],
903 );
904 for (k, v) in row_writes {
905 r.set_owned(k.clone(), v.clone());
906 }
907 r
908 })
909 .collect();
910
911 let schemaless: Vec<UnifiedRecord> = rows
914 .iter()
915 .map(|row_writes| {
916 let mut r = UnifiedRecord::schemaless();
917 for col in &schema {
920 r.set_owned(col.clone(), Value::Null);
921 }
922 for (k, v) in row_writes {
923 r.set_owned(k.clone(), v.clone());
924 }
925 r
926 })
927 .collect();
928
929 for (s_row, h_row) in shared.iter().zip(schemaless.iter()) {
933 for col in &schema {
934 prop_assert_eq!(
935 s_row.get(col),
936 h_row.get(col),
937 "column {} diverged on shared vs schemaless build",
938 col
939 );
940 }
941 }
942
943 for s_row in &shared {
946 prop_assert!(s_row.columnar().is_some() || schema.is_empty());
947 }
948 }
949
950 #[test]
953 fn duplicate_schema_last_write_wins(
954 col in arb_column_name(),
955 v1 in arb_value(),
956 v2 in arb_value(),
957 ) {
958 let arc_schema: Arc<Vec<Arc<str>>> = Arc::new(vec![
959 Arc::from(col.as_str()),
960 Arc::from(col.as_str()),
961 ]);
962 let rec = UnifiedRecord::with_schema(arc_schema, vec![v1.clone(), v2.clone()]);
963 prop_assert_eq!(rec.get(&col), Some(&v2));
965 }
966
967 #[test]
971 fn schemaless_path_uses_overflow_only(writes in arb_writes()) {
972 let mut rec = UnifiedRecord::schemaless();
973 for (k, v) in &writes {
974 rec.set_owned(k.clone(), v.clone());
975 }
976
977 prop_assert!(rec.columns().is_empty());
978 prop_assert!(rec.schema_values().is_empty());
979
980 let mut reference: std::collections::HashMap<String, Value> =
982 std::collections::HashMap::new();
983 for (k, v) in &writes {
984 reference.insert(k.clone(), v.clone());
985 }
986 for (k, v) in reference {
987 prop_assert_eq!(rec.get(&k), Some(&v));
988 }
989 }
990 }
991}