1use reddb_types::Value;
3use std::collections::HashMap;
4use std::sync::Arc;
5
6#[derive(Debug, Clone)]
8pub struct ExecutionError {
9 pub message: String,
10}
11
12impl ExecutionError {
13 pub fn new(message: impl Into<String>) -> Self {
14 Self {
15 message: message.into(),
16 }
17 }
18}
19
20impl std::fmt::Display for ExecutionError {
21 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22 write!(f, "Execution error: {}", self.message)
23 }
24}
25
26impl std::error::Error for ExecutionError {}
27
28#[derive(Debug, Clone, Default)]
30pub struct UnifiedResult {
31 pub columns: Vec<String>,
33 pub records: Vec<UnifiedRecord>,
35 pub stats: QueryStats,
37 pub pre_serialized_json: Option<String>,
39}
40
41impl UnifiedResult {
42 pub fn empty() -> Self {
44 Self::default()
45 }
46
47 pub fn with_columns(columns: Vec<String>) -> Self {
49 Self {
50 columns,
51 records: Vec::new(),
52 stats: QueryStats::default(),
53 pre_serialized_json: None,
54 }
55 }
56
57 pub fn push(&mut self, record: UnifiedRecord) {
59 self.records.push(record);
60 }
61
62 pub fn len(&self) -> usize {
64 self.records.len()
65 }
66
67 pub fn is_empty(&self) -> bool {
69 self.records.is_empty()
70 }
71}
72
73#[derive(Debug, Clone)]
94pub struct UnifiedRecord {
95 schema: Arc<Vec<Arc<str>>>,
98 values: Vec<Value>,
100 overflow: Option<HashMap<Arc<str>, Value>>,
102 pub nodes: HashMap<String, MatchedNode>,
104 pub edges: HashMap<String, MatchedEdge>,
106 pub paths: Vec<GraphPath>,
108 pub vector_results: Vec<VectorSearchResult>,
110}
111
112fn empty_schema() -> Arc<Vec<Arc<str>>> {
117 static EMPTY: std::sync::OnceLock<Arc<Vec<Arc<str>>>> = std::sync::OnceLock::new();
118 Arc::clone(EMPTY.get_or_init(|| Arc::new(Vec::new())))
119}
120
121impl Default for UnifiedRecord {
122 fn default() -> Self {
123 Self {
124 schema: empty_schema(),
125 values: Vec::new(),
126 overflow: None,
127 nodes: HashMap::new(),
128 edges: HashMap::new(),
129 paths: Vec::new(),
130 vector_results: Vec::new(),
131 }
132 }
133}
134
135#[derive(Debug, Clone, Copy)]
141pub struct ColumnarRow<'a> {
142 pub schema: &'a Arc<Vec<Arc<str>>>,
143 pub values: &'a [Value],
144}
145
146pub fn sys_key_rid() -> Arc<str> {
151 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
152 Arc::clone(KEY.get_or_init(|| Arc::from("rid")))
153}
154
155pub fn sys_key_collection() -> Arc<str> {
156 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
157 Arc::clone(KEY.get_or_init(|| Arc::from("collection")))
158}
159
160pub fn sys_key_kind() -> Arc<str> {
161 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
162 Arc::clone(KEY.get_or_init(|| Arc::from("kind")))
163}
164
165pub fn sys_key_tenant() -> Arc<str> {
166 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
167 Arc::clone(KEY.get_or_init(|| Arc::from("tenant")))
168}
169
170pub fn sys_key_created_at() -> Arc<str> {
171 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
172 Arc::clone(KEY.get_or_init(|| Arc::from("created_at")))
173}
174
175pub fn sys_key_updated_at() -> Arc<str> {
176 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
177 Arc::clone(KEY.get_or_init(|| Arc::from("updated_at")))
178}
179
180pub fn sys_key_row_id() -> Arc<str> {
181 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
182 Arc::clone(KEY.get_or_init(|| Arc::from("row_id")))
183}
184
185pub fn sys_key_red_collection() -> Arc<str> {
186 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
187 Arc::clone(KEY.get_or_init(|| Arc::from("red_collection")))
188}
189
190pub fn sys_key_red_kind() -> Arc<str> {
191 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
192 Arc::clone(KEY.get_or_init(|| Arc::from("red_kind")))
193}
194
195pub fn sys_key_red_sequence_id() -> Arc<str> {
196 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
197 Arc::clone(KEY.get_or_init(|| Arc::from("red_sequence_id")))
198}
199
200pub fn sys_key_red_entity_type() -> Arc<str> {
201 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
202 Arc::clone(KEY.get_or_init(|| Arc::from("red_entity_type")))
203}
204
205pub fn sys_key_red_capabilities() -> Arc<str> {
206 static KEY: std::sync::OnceLock<Arc<str>> = std::sync::OnceLock::new();
207 Arc::clone(KEY.get_or_init(|| Arc::from("red_capabilities")))
208}
209
210impl UnifiedRecord {
211 pub fn new() -> Self {
214 Self::default()
215 }
216
217 pub fn with_capacity(capacity: usize) -> Self {
221 let mut rec = Self::default();
222 if capacity > 0 {
223 rec.overflow = Some(HashMap::with_capacity(capacity));
224 }
225 rec
226 }
227
228 pub fn with_schema(schema: Arc<Vec<Arc<str>>>, values: Vec<Value>) -> Self {
239 Self {
240 schema,
241 values,
242 overflow: None,
243 nodes: HashMap::new(),
244 edges: HashMap::new(),
245 paths: Vec::new(),
246 vector_results: Vec::new(),
247 }
248 }
249
250 #[inline]
253 pub fn from_columnar(schema: Arc<Vec<Arc<str>>>, values: Vec<Value>) -> Self {
254 Self::with_schema(schema, values)
255 }
256
257 pub fn schemaless() -> Self {
261 Self::default()
262 }
263
264 #[inline]
267 pub fn columns(&self) -> &[Arc<str>] {
268 self.schema.as_ref()
269 }
270
271 #[inline]
274 pub fn schema_values(&self) -> &[Value] {
275 &self.values
276 }
277
278 #[inline]
280 pub fn overflow(&self) -> Option<&HashMap<Arc<str>, Value>> {
281 self.overflow.as_ref()
282 }
283
284 #[inline]
291 fn schema_index(&self, column: &str) -> Option<usize> {
292 self.schema.iter().rposition(|k| &**k == column)
293 }
294
295 pub fn set(&mut self, column: &str, value: Value) {
298 if let Some(idx) = self.schema_index(column) {
299 if idx < self.values.len() {
300 self.values[idx] = value;
301 return;
302 }
303 }
304 self.overflow_mut().insert(Arc::from(column), value);
305 }
306
307 #[inline]
311 pub fn set_owned(&mut self, column: String, value: Value) {
312 if let Some(idx) = self.schema_index(&column) {
313 if idx < self.values.len() {
314 self.values[idx] = value;
315 return;
316 }
317 }
318 self.overflow_mut()
319 .insert(Arc::from(column.into_boxed_str()), value);
320 }
321
322 #[inline]
327 pub fn set_arc(&mut self, column: Arc<str>, value: Value) {
328 if let Some(idx) = self.schema_index(&column) {
329 if idx < self.values.len() {
330 self.values[idx] = value;
331 return;
332 }
333 }
334 self.overflow_mut().insert(column, value);
335 }
336
337 #[inline]
345 pub fn set_arc_if_absent(&mut self, column: Arc<str>, value: Value) {
346 if self.get(&column).is_none() {
347 self.set_arc(column, value);
348 }
349 }
350
351 pub fn get(&self, column: &str) -> Option<&Value> {
362 if let Some(idx) = self.schema_index(column) {
363 if let Some(v) = self.values.get(idx) {
364 return Some(v);
365 }
366 }
367 self.overflow.as_ref().and_then(|m| m.get(column))
368 }
369
370 pub fn field_count(&self) -> usize {
372 self.values.len() + self.overflow.as_ref().map(|m| m.len()).unwrap_or(0)
373 }
374
375 pub fn contains_column(&self, column: &str) -> bool {
377 self.get(column).is_some()
378 }
379
380 pub fn iter_fields(&self) -> Box<dyn Iterator<Item = (&Arc<str>, &Value)> + '_> {
385 let schema_iter = self.schema.iter().zip(self.values.iter());
386 let overflow_iter = self.overflow.iter().flat_map(|m| m.iter());
387 Box::new(schema_iter.chain(overflow_iter))
388 }
389
390 pub fn values_mut(&mut self) -> Box<dyn Iterator<Item = &mut Value> + '_> {
394 let schema_iter = self.values.iter_mut();
395 let overflow_iter = self.overflow.iter_mut().flat_map(|m| m.values_mut());
396 Box::new(schema_iter.chain(overflow_iter))
397 }
398
399 pub fn column_names(&self) -> Vec<Arc<str>> {
401 let mut out: Vec<Arc<str>> = Vec::with_capacity(self.field_count());
402 out.extend(self.schema.iter().cloned());
403 if let Some(m) = &self.overflow {
404 out.extend(m.keys().cloned());
405 }
406 out
407 }
408
409 #[inline]
416 pub fn columnar(&self) -> Option<ColumnarRow<'_>> {
417 if self.schema.is_empty() {
418 None
419 } else {
420 Some(ColumnarRow {
421 schema: &self.schema,
422 values: &self.values,
423 })
424 }
425 }
426
427 #[inline]
433 pub fn columnar_schema(&self) -> Option<&Arc<Vec<Arc<str>>>> {
434 if self.schema.is_empty() {
435 None
436 } else {
437 Some(&self.schema)
438 }
439 }
440
441 pub fn overflow_mut(&mut self) -> &mut HashMap<Arc<str>, Value> {
445 self.overflow.get_or_insert_with(HashMap::new)
446 }
447
448 pub fn overflow_entry_or_insert(&mut self, column: Arc<str>, value: Value) {
451 self.overflow_mut().entry(column).or_insert(value);
452 }
453
454 pub fn set_node(&mut self, alias: &str, node: MatchedNode) {
456 self.nodes.insert(alias.to_string(), node);
457 }
458
459 pub fn get_node(&self, alias: &str) -> Option<&MatchedNode> {
461 self.nodes.get(alias)
462 }
463
464 pub fn set_edge(&mut self, alias: &str, edge: MatchedEdge) {
466 self.edges.insert(alias.to_string(), edge);
467 }
468}
469
470#[derive(Debug, Clone)]
472pub struct MatchedNode {
473 pub id: String,
474 pub label: String,
475 pub node_label: String,
477 pub properties: HashMap<String, Value>,
478}
479
480impl MatchedNode {
481 pub fn from_stored(node: impl Into<Self>) -> Self {
484 node.into()
485 }
486}
487
488#[derive(Debug, Clone)]
490pub struct MatchedEdge {
491 pub from: String,
492 pub to: String,
493 pub edge_label: String,
495 pub weight: f32,
496 pub properties: HashMap<String, Value>,
497}
498
499impl MatchedEdge {
500 pub fn from_tuple(
503 source: &str,
504 edge_label: impl Into<String>,
505 target: &str,
506 weight: f32,
507 ) -> Self {
508 Self {
509 from: source.to_string(),
510 to: target.to_string(),
511 edge_label: edge_label.into(),
512 weight,
513 properties: HashMap::new(),
514 }
515 }
516}
517
518#[derive(Debug, Clone)]
520pub struct VectorSearchResult {
521 pub id: u64,
523 pub collection: String,
525 pub distance: f32,
527 pub vector: Option<Vec<f32>>,
529 pub metadata: Option<HashMap<String, Value>>,
531 pub linked_node: Option<String>,
533 pub linked_row: Option<(String, u64)>,
535}
536
537impl VectorSearchResult {
538 pub fn new(id: u64, collection: impl Into<String>, distance: f32) -> Self {
540 Self {
541 id,
542 collection: collection.into(),
543 distance,
544 vector: None,
545 metadata: None,
546 linked_node: None,
547 linked_row: None,
548 }
549 }
550
551 pub fn with_vector(mut self, vector: Vec<f32>) -> Self {
553 self.vector = Some(vector);
554 self
555 }
556
557 pub fn with_metadata(mut self, metadata: HashMap<String, Value>) -> Self {
559 self.metadata = Some(metadata);
560 self
561 }
562
563 pub fn with_linked_node(mut self, node_id: impl Into<String>) -> Self {
565 self.linked_node = Some(node_id.into());
566 self
567 }
568
569 pub fn with_linked_row(mut self, table: impl Into<String>, row_id: u64) -> Self {
571 self.linked_row = Some((table.into(), row_id));
572 self
573 }
574}
575
576impl Default for VectorSearchResult {
577 fn default() -> Self {
578 Self::new(0, String::new(), 0.0)
579 }
580}
581
582#[derive(Debug, Clone)]
584pub struct GraphPath {
585 pub nodes: Vec<String>,
587 pub edges: Vec<MatchedEdge>,
589 pub total_weight: f32,
591}
592
593impl GraphPath {
594 pub fn start(node_id: &str) -> Self {
596 Self {
597 nodes: vec![node_id.to_string()],
598 edges: Vec::new(),
599 total_weight: 0.0,
600 }
601 }
602
603 pub fn extend(&self, edge: MatchedEdge, node_id: &str) -> Self {
605 let mut new_path = self.clone();
606 new_path.total_weight += edge.weight;
607 new_path.edges.push(edge);
608 new_path.nodes.push(node_id.to_string());
609 new_path
610 }
611
612 pub fn len(&self) -> usize {
614 self.edges.len()
615 }
616
617 pub fn is_empty(&self) -> bool {
619 self.edges.is_empty()
620 }
621}
622
623#[derive(Debug, Clone, Default)]
625pub struct QueryStats {
626 pub nodes_scanned: u64,
628 pub edges_scanned: u64,
630 pub rows_scanned: u64,
632 pub segments_total: u64,
634 pub segments_scanned: u64,
636 pub segments_pruned: u64,
638 pub exec_time_us: u64,
640 pub vector: Option<VectorQueryStats>,
642}
643
644impl QueryStats {
645 pub fn merge(a: &QueryStats, b: &QueryStats) -> QueryStats {
647 QueryStats {
648 nodes_scanned: a.nodes_scanned + b.nodes_scanned,
649 edges_scanned: a.edges_scanned + b.edges_scanned,
650 rows_scanned: a.rows_scanned + b.rows_scanned,
651 segments_total: a.segments_total + b.segments_total,
652 segments_scanned: a.segments_scanned + b.segments_scanned,
653 segments_pruned: a.segments_pruned + b.segments_pruned,
654 exec_time_us: a.exec_time_us + b.exec_time_us,
655 vector: None, }
657 }
658}
659
660#[derive(Debug, Clone, Default)]
663pub struct VectorOperatorStats {
664 pub operator: String,
665 pub input_rows: u64,
666 pub output_rows: u64,
667 pub inclusive_time_us: u64,
668}
669
670#[derive(Debug, Clone, Default)]
672pub struct VectorQueryStats {
673 pub operators: Vec<VectorOperatorStats>,
674 pub mode_requested: String,
675 pub mode_executed: String,
676 pub fallback_reason: Option<String>,
677 pub approximate_distance_evaluations: u64,
678 pub peak_topk_entries: u64,
679 pub cache_hit: bool,
681 pub access_path: String,
682 pub index_used: bool,
683 pub candidates_examined: u64,
684 pub metadata_rejected: u64,
685 pub rls_rejected: u64,
686 pub visibility_rejected: u64,
687 pub exact_distance_evaluations: u64,
688 pub rows_returned: u64,
689}
690
691impl VectorQueryStats {
692 pub fn operators_json(&self) -> crate::json::Value {
693 use crate::json::{Map, Value as JsonValue};
694 JsonValue::Array(
695 self.operators
696 .iter()
697 .map(|operator| {
698 let mut fields = Map::new();
699 fields.insert(
700 "operator".into(),
701 JsonValue::String(operator.operator.clone()),
702 );
703 for (name, value) in [
704 ("input_rows", operator.input_rows),
705 ("output_rows", operator.output_rows),
706 ("inclusive_time_us", operator.inclusive_time_us),
707 ] {
708 fields.insert(name.into(), Value::UnsignedInteger(value).to_json());
709 }
710 JsonValue::Object(fields)
711 })
712 .collect(),
713 )
714 }
715}
716
717#[cfg(all(test, not(target_arch = "wasm32")))]
718mod record_layout_tests {
719 use super::*;
739 use proptest::collection::vec;
740 use proptest::prelude::*;
741 use reddb_types::Value;
742
743 fn arb_value() -> impl Strategy<Value = Value> {
744 prop_oneof![
745 Just(Value::Null),
746 any::<bool>().prop_map(Value::Boolean),
747 any::<i64>().prop_map(Value::Integer),
748 any::<f64>()
749 .prop_filter("nan-free", |v| !v.is_nan())
750 .prop_map(Value::Float),
751 "[a-zA-Z0-9 ]{0,16}".prop_map(Value::text),
752 ]
753 }
754
755 fn arb_column_name() -> impl Strategy<Value = String> {
756 "[a-d]{1,3}".prop_map(|s| s.to_string())
759 }
760
761 fn arb_schema() -> impl Strategy<Value = Vec<String>> {
762 vec(arb_column_name(), 0..6).prop_map(|mut v| {
766 v.sort();
767 v.dedup();
768 v
769 })
770 }
771
772 fn arb_writes() -> impl Strategy<Value = Vec<(String, Value)>> {
773 vec((arb_column_name(), arb_value()), 0..16)
774 }
775
776 fn record_with_schema(schema: &[String], writes: &[(String, Value)]) -> UnifiedRecord {
781 let arc_schema: Arc<Vec<Arc<str>>> =
782 Arc::new(schema.iter().map(|s| Arc::from(s.as_str())).collect());
783 let mut rec =
787 UnifiedRecord::with_schema(Arc::clone(&arc_schema), vec![Value::Null; schema.len()]);
788 for (k, v) in writes {
789 rec.set_owned(k.clone(), v.clone());
790 }
791 rec
792 }
793
794 fn hashmap_reference(
799 schema: &[String],
800 writes: &[(String, Value)],
801 ) -> std::collections::HashMap<String, Value> {
802 let mut map: std::collections::HashMap<String, Value> =
803 schema.iter().map(|c| (c.clone(), Value::Null)).collect();
804 for (k, v) in writes {
805 map.insert(k.clone(), v.clone());
806 }
807 map
808 }
809
810 proptest! {
811 #[test]
815 fn get_set_round_trip(schema in arb_schema(), writes in arb_writes()) {
816 let rec = record_with_schema(&schema, &writes);
817 let reference = hashmap_reference(&schema, &writes);
818 for (k, expected) in &reference {
819 prop_assert_eq!(rec.get(k), Some(expected), "key {}", k);
820 }
821 }
822
823 #[test]
828 fn overflow_promotion_only_for_missing_keys(
829 schema in arb_schema(),
830 writes in arb_writes(),
831 ) {
832 let rec = record_with_schema(&schema, &writes);
833
834 for col in &schema {
836 if let Some(over) = rec.overflow() {
837 prop_assert!(
838 !over.contains_key(col.as_str()),
839 "schema column {} leaked into overflow",
840 col
841 );
842 }
843 }
844
845 for (k, _) in &writes {
847 if !schema.iter().any(|c| c == k) {
848 let over = rec.overflow().expect("overflow allocated");
849 prop_assert!(
850 over.contains_key(k.as_str()),
851 "out-of-schema key {} missing from overflow",
852 k
853 );
854 }
855 }
856 }
857
858 #[test]
862 fn iter_fields_matches_hashmap(
863 schema in arb_schema(),
864 writes in arb_writes(),
865 ) {
866 let rec = record_with_schema(&schema, &writes);
867 let reference = hashmap_reference(&schema, &writes);
868
869 let mut got: Vec<(String, Value)> = rec
870 .iter_fields()
871 .map(|(k, v)| (k.as_ref().to_string(), v.clone()))
872 .collect();
873 let mut want: Vec<(String, Value)> = reference
874 .into_iter()
875 .collect();
876
877 got.sort_by(|a, b| a.0.cmp(&b.0));
878 want.sort_by(|a, b| a.0.cmp(&b.0));
879 prop_assert_eq!(got, want);
880 }
881
882 #[test]
888 fn shared_schema_scan_parity(
889 schema in arb_schema().prop_filter("non-empty schema", |s| !s.is_empty()),
890 rows in vec(arb_writes(), 0..8),
891 ) {
892 let arc_schema: Arc<Vec<Arc<str>>> =
894 Arc::new(schema.iter().map(|s| Arc::from(s.as_str())).collect());
895 let shared: Vec<UnifiedRecord> = rows
896 .iter()
897 .map(|row_writes| {
898 let mut r = UnifiedRecord::with_schema(
899 Arc::clone(&arc_schema),
900 vec![Value::Null; schema.len()],
901 );
902 for (k, v) in row_writes {
903 r.set_owned(k.clone(), v.clone());
904 }
905 r
906 })
907 .collect();
908
909 let schemaless: Vec<UnifiedRecord> = rows
912 .iter()
913 .map(|row_writes| {
914 let mut r = UnifiedRecord::schemaless();
915 for col in &schema {
918 r.set_owned(col.clone(), Value::Null);
919 }
920 for (k, v) in row_writes {
921 r.set_owned(k.clone(), v.clone());
922 }
923 r
924 })
925 .collect();
926
927 for (s_row, h_row) in shared.iter().zip(schemaless.iter()) {
931 for col in &schema {
932 prop_assert_eq!(
933 s_row.get(col),
934 h_row.get(col),
935 "column {} diverged on shared vs schemaless build",
936 col
937 );
938 }
939 }
940
941 for s_row in &shared {
944 prop_assert!(s_row.columnar().is_some() || schema.is_empty());
945 }
946 }
947
948 #[test]
951 fn duplicate_schema_last_write_wins(
952 col in arb_column_name(),
953 v1 in arb_value(),
954 v2 in arb_value(),
955 ) {
956 let arc_schema: Arc<Vec<Arc<str>>> = Arc::new(vec![
957 Arc::from(col.as_str()),
958 Arc::from(col.as_str()),
959 ]);
960 let rec = UnifiedRecord::with_schema(arc_schema, vec![v1.clone(), v2.clone()]);
961 prop_assert_eq!(rec.get(&col), Some(&v2));
963 }
964
965 #[test]
969 fn schemaless_path_uses_overflow_only(writes in arb_writes()) {
970 let mut rec = UnifiedRecord::schemaless();
971 for (k, v) in &writes {
972 rec.set_owned(k.clone(), v.clone());
973 }
974
975 prop_assert!(rec.columns().is_empty());
976 prop_assert!(rec.schema_values().is_empty());
977
978 let mut reference: std::collections::HashMap<String, Value> =
980 std::collections::HashMap::new();
981 for (k, v) in &writes {
982 reference.insert(k.clone(), v.clone());
983 }
984 for (k, v) in reference {
985 prop_assert_eq!(rec.get(&k), Some(&v));
986 }
987 }
988 }
989}