1use rustc_hash::FxHashMap;
19
20use crate::traits::{Scanner, TypedBatchFallbackReason, TypedColumnBatch};
21use radixdb_core::value::NULL_VALUE;
22use radixdb_core::CompactArc;
23use radixdb_core::{Result, Row, Value};
24
25#[doc(hidden)]
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum DeferredColumnSource {
33 Left(usize),
34 Right(usize),
35}
36
37#[doc(hidden)]
44#[derive(Debug, Clone)]
45pub enum DeferredRow {
46 Owned(Row),
47 Shared {
48 rows: CompactArc<Vec<Row>>,
49 row_index: usize,
50 },
51 Projected {
52 left: Box<DeferredRow>,
53 right: Box<DeferredRow>,
54 columns: CompactArc<[DeferredColumnSource]>,
55 },
56 Remapped {
57 row: Box<DeferredRow>,
58 columns: CompactArc<[usize]>,
59 },
60}
61
62impl DeferredRow {
63 #[inline]
64 pub fn owned(row: Row) -> Self {
65 Self::Owned(row)
66 }
67
68 #[inline]
69 pub fn shared(rows: CompactArc<Vec<Row>>, row_index: usize) -> Self {
70 assert!(
71 row_index < rows.len(),
72 "deferred shared row index outside batch"
73 );
74 Self::Shared { rows, row_index }
75 }
76
77 #[inline]
78 pub fn projected(
79 left: DeferredRow,
80 right: DeferredRow,
81 columns: CompactArc<[DeferredColumnSource]>,
82 ) -> Self {
83 Self::Projected {
84 left: Box::new(left),
85 right: Box::new(right),
86 columns,
87 }
88 }
89
90 #[inline]
91 pub fn remapped(row: DeferredRow, columns: CompactArc<[usize]>) -> Self {
92 Self::Remapped {
93 row: Box::new(row),
94 columns,
95 }
96 }
97
98 #[inline]
99 pub fn len(&self) -> usize {
100 match self {
101 Self::Owned(row) => row.len(),
102 Self::Shared { rows, row_index } => rows[*row_index].len(),
103 Self::Projected { columns, .. } => columns.len(),
104 Self::Remapped { columns, .. } => columns.len(),
105 }
106 }
107
108 #[inline]
109 pub fn is_empty(&self) -> bool {
110 self.len() == 0
111 }
112
113 #[inline]
114 pub fn is_deferred(&self) -> bool {
115 !matches!(self, Self::Owned(_))
116 }
117
118 pub fn estimated_retained_bytes(&self) -> usize {
126 fn values_bytes(values: &[Value]) -> usize {
127 values.iter().fold(0_usize, |total, value| {
128 let payload = match value {
129 Value::Text(text) => text.len(),
130 Value::Extension(bytes) => bytes.len(),
131 _ => 0,
132 };
133 total
134 .saturating_add(std::mem::size_of::<Value>())
135 .saturating_add(payload)
136 })
137 }
138
139 match self {
140 Self::Owned(row) => {
141 std::mem::size_of::<Self>().saturating_add(values_bytes(row.as_slice()))
142 }
143 Self::Shared { .. } => std::mem::size_of::<Self>(),
144 Self::Projected {
145 left,
146 right,
147 columns,
148 } => std::mem::size_of::<Self>()
149 .saturating_add(left.estimated_retained_bytes())
150 .saturating_add(right.estimated_retained_bytes())
151 .saturating_add(
152 columns
153 .len()
154 .saturating_mul(std::mem::size_of::<DeferredColumnSource>()),
155 ),
156 Self::Remapped { row, columns } => std::mem::size_of::<Self>()
157 .saturating_add(row.estimated_retained_bytes())
158 .saturating_add(columns.len().saturating_mul(std::mem::size_of::<usize>())),
159 }
160 }
161
162 #[inline]
163 pub fn get(&self, index: usize) -> Option<&Value> {
164 match self {
165 Self::Owned(row) => row.get(index),
166 Self::Shared { rows, row_index } => rows[*row_index].get(index),
167 Self::Projected {
168 left,
169 right,
170 columns,
171 } => match columns.get(index)? {
172 DeferredColumnSource::Left(source_index) => left.get(*source_index),
173 DeferredColumnSource::Right(source_index) => right.get(*source_index),
174 },
175 Self::Remapped { row, columns } => row.get(*columns.get(index)?),
176 }
177 }
178
179 pub fn to_owned(&self) -> Row {
180 let row = match self {
181 Self::Owned(row) => row.clone(),
182 Self::Shared { rows, row_index } => rows[*row_index].clone(),
183 Self::Projected { columns, .. } => {
184 let mut row = Row::with_capacity(columns.len());
185 for index in 0..columns.len() {
186 row.push(self.get(index).cloned().unwrap_or(NULL_VALUE));
187 }
188 row
189 }
190 Self::Remapped { columns, .. } => {
191 let mut row = Row::with_capacity(columns.len());
192 for index in 0..columns.len() {
193 row.push(self.get(index).cloned().unwrap_or(NULL_VALUE));
194 }
195 row
196 }
197 };
198 crate::instrumentation::record_join_value_copies(row.as_slice());
199 row
200 }
201
202 #[inline]
203 pub fn into_owned(self) -> Row {
204 match self {
205 Self::Owned(row) => row,
206 Self::Shared { rows, row_index } => {
207 let row = rows[row_index].clone();
208 crate::instrumentation::record_join_value_copies(row.as_slice());
209 row
210 }
211 deferred @ (Self::Projected { .. } | Self::Remapped { .. }) => deferred.to_owned(),
212 }
213 }
214}
215
216pub trait QueryResult: Send {
235 fn columns(&self) -> &[String];
239
240 fn columns_arc(&self) -> Option<CompactArc<Vec<String>>> {
245 None
246 }
247
248 fn next(&mut self) -> bool;
252
253 fn scan(&self, dest: &mut [Value]) -> Result<()>;
258
259 fn row(&self) -> &Row;
264
265 fn take_row(&mut self) -> Row {
271 self.row().clone()
272 }
273
274 #[doc(hidden)]
280 fn take_deferred_row(&mut self) -> DeferredRow {
281 DeferredRow::owned(self.take_row())
282 }
283
284 #[doc(hidden)]
291 fn preserves_deferred_rows(&self) -> bool {
292 false
293 }
294
295 #[doc(hidden)]
302 fn ascending_nulls_last_ordering(&self) -> Option<Vec<usize>> {
303 None
304 }
305
306 fn close(&mut self) -> Result<()> {
310 Ok(())
311 }
312
313 fn rows_affected(&self) -> i64;
316
317 fn last_insert_id(&self) -> i64;
320
321 fn try_into_arc_rows(&mut self) -> Option<CompactArc<Vec<Row>>> {
327 None
328 }
329
330 fn estimated_count(&self) -> Option<usize> {
335 None
336 }
337
338 fn supports_typed_batches(&self) -> bool {
345 false
346 }
347
348 fn typed_batch_fallback_reason(&self) -> Option<TypedBatchFallbackReason> {
352 Some(TypedBatchFallbackReason::UnsupportedResultShape)
353 }
354
355 fn next_typed_batch(&mut self) -> Result<Option<TypedColumnBatch>> {
361 Ok(None)
362 }
363
364 fn last_error(&mut self) -> Option<radixdb_core::Error> {
370 None
371 }
372
373 fn with_aliases(self: Box<Self>, aliases: FxHashMap<String, String>) -> Box<dyn QueryResult>;
378}
379
380pub struct AliasedResult {
382 inner: Box<dyn QueryResult>,
383 aliased_columns: Vec<String>,
384}
385
386impl AliasedResult {
387 pub fn new(inner: Box<dyn QueryResult>, aliases: FxHashMap<String, String>) -> Self {
388 let aliased_columns = inner
389 .columns()
390 .iter()
391 .map(|column| {
392 aliases
393 .iter()
394 .find(|(_, original)| *original == column)
395 .map_or_else(|| column.clone(), |(alias, _)| alias.clone())
396 })
397 .collect();
398 Self {
399 inner,
400 aliased_columns,
401 }
402 }
403}
404
405impl QueryResult for AliasedResult {
406 fn columns(&self) -> &[String] {
407 &self.aliased_columns
408 }
409
410 fn next(&mut self) -> bool {
411 self.inner.next()
412 }
413
414 fn scan(&self, dest: &mut [Value]) -> Result<()> {
415 self.inner.scan(dest)
416 }
417
418 fn row(&self) -> &Row {
419 self.inner.row()
420 }
421
422 fn take_row(&mut self) -> Row {
423 self.inner.take_row()
424 }
425
426 fn take_deferred_row(&mut self) -> DeferredRow {
427 self.inner.take_deferred_row()
428 }
429
430 fn preserves_deferred_rows(&self) -> bool {
431 self.inner.preserves_deferred_rows()
432 }
433
434 fn ascending_nulls_last_ordering(&self) -> Option<Vec<usize>> {
435 self.inner.ascending_nulls_last_ordering()
436 }
437
438 fn close(&mut self) -> Result<()> {
439 self.inner.close()
440 }
441
442 fn rows_affected(&self) -> i64 {
443 self.inner.rows_affected()
444 }
445
446 fn last_insert_id(&self) -> i64 {
447 self.inner.last_insert_id()
448 }
449
450 fn last_error(&mut self) -> Option<radixdb_core::Error> {
451 self.inner.last_error()
452 }
453
454 fn with_aliases(self: Box<Self>, aliases: FxHashMap<String, String>) -> Box<dyn QueryResult> {
455 Box::new(Self::new(self, aliases))
456 }
457}
458
459pub struct ScannerResult {
464 scanner: Box<dyn Scanner>,
465 columns: Vec<String>,
466 current_row: Row,
467 has_current: bool,
468}
469
470impl ScannerResult {
471 pub fn new(scanner: Box<dyn Scanner>, columns: Vec<String>) -> Self {
472 Self {
473 scanner,
474 columns,
475 current_row: Row::new(),
476 has_current: false,
477 }
478 }
479}
480
481impl QueryResult for ScannerResult {
482 fn columns(&self) -> &[String] {
483 &self.columns
484 }
485
486 fn next(&mut self) -> bool {
487 if self.scanner.next() {
488 self.current_row = self.scanner.take_row();
489 self.has_current = true;
490 true
491 } else {
492 self.has_current = false;
493 false
494 }
495 }
496
497 fn scan(&self, dest: &mut [Value]) -> Result<()> {
498 if !self.has_current {
499 return Err(radixdb_core::Error::internal(
500 "scan() called without successful next()",
501 ));
502 }
503 if dest.len() != self.current_row.len() {
504 return Err(radixdb_core::Error::internal(format!(
505 "scan destination has {} values but row has {} columns",
506 dest.len(),
507 self.current_row.len()
508 )));
509 }
510 for (dest, value) in dest.iter_mut().zip(self.current_row.iter()) {
511 *dest = value.clone();
512 }
513 Ok(())
514 }
515
516 fn row(&self) -> &Row {
517 assert!(self.has_current, "row() called without successful next()");
518 &self.current_row
519 }
520
521 fn take_row(&mut self) -> Row {
522 assert!(
523 self.has_current,
524 "take_row() called without successful next()"
525 );
526 std::mem::take(&mut self.current_row)
527 }
528
529 fn close(&mut self) -> Result<()> {
530 self.scanner.close()
531 }
532
533 fn last_error(&mut self) -> Option<radixdb_core::Error> {
534 self.scanner.err().cloned()
535 }
536
537 fn rows_affected(&self) -> i64 {
538 0
539 }
540
541 fn last_insert_id(&self) -> i64 {
542 0
543 }
544
545 fn estimated_count(&self) -> Option<usize> {
546 self.scanner.estimated_count()
547 }
548
549 fn supports_typed_batches(&self) -> bool {
550 !self.has_current && self.scanner.supports_typed_batches()
551 }
552
553 fn typed_batch_fallback_reason(&self) -> Option<TypedBatchFallbackReason> {
554 if self.supports_typed_batches() {
555 None
556 } else if self.has_current {
557 Some(TypedBatchFallbackReason::RowAlreadyFetched)
558 } else {
559 self.scanner.typed_batch_fallback_reason()
560 }
561 }
562
563 fn next_typed_batch(&mut self) -> Result<Option<TypedColumnBatch>> {
564 if self.has_current {
565 return Err(radixdb_core::Error::internal(
566 "typed batch requested after row-oriented scanner advance",
567 ));
568 }
569 self.scanner.next_typed_batch()
570 }
571
572 fn with_aliases(self: Box<Self>, aliases: FxHashMap<String, String>) -> Box<dyn QueryResult> {
573 Box::new(AliasedResult::new(self, aliases))
574 }
575}
576
577pub struct MemoryResult {
579 columns: Vec<String>,
580 rows: Vec<Row>,
581 current_index: Option<usize>,
582 rows_affected: i64,
583 last_insert_id: i64,
584 closed: bool,
585}
586
587impl MemoryResult {
588 pub fn new(columns: Vec<String>) -> Self {
590 Self {
591 columns,
592 rows: Vec::new(),
593 current_index: None,
594 rows_affected: 0,
595 last_insert_id: 0,
596 closed: false,
597 }
598 }
599
600 pub fn with_rows(columns: Vec<String>, rows: Vec<Row>) -> Self {
602 Self {
603 columns,
604 rows,
605 current_index: None,
606 rows_affected: 0,
607 last_insert_id: 0,
608 closed: false,
609 }
610 }
611
612 pub fn for_modification(rows_affected: i64, last_insert_id: i64) -> Self {
614 Self {
615 columns: Vec::new(),
616 rows: Vec::new(),
617 current_index: None,
618 rows_affected,
619 last_insert_id,
620 closed: false,
621 }
622 }
623
624 pub fn add_row(&mut self, row: Row) {
626 self.rows.push(row);
627 }
628
629 pub fn set_rows_affected(&mut self, count: i64) {
631 self.rows_affected = count;
632 }
633
634 pub fn set_last_insert_id(&mut self, id: i64) {
636 self.last_insert_id = id;
637 }
638}
639
640impl QueryResult for MemoryResult {
641 fn columns(&self) -> &[String] {
642 &self.columns
643 }
644
645 fn estimated_count(&self) -> Option<usize> {
646 Some(self.rows.len())
647 }
648
649 fn next(&mut self) -> bool {
650 if self.closed {
651 return false;
652 }
653
654 let next_index = match self.current_index {
655 None => 0,
656 Some(i) => i + 1,
657 };
658
659 if next_index < self.rows.len() {
660 self.current_index = Some(next_index);
661 true
662 } else {
663 false
664 }
665 }
666
667 fn scan(&self, dest: &mut [Value]) -> Result<()> {
668 let row = self.row();
669
670 if dest.len() != row.len() {
671 return Err(radixdb_core::Error::internal(format!(
672 "scan destination has {} values but row has {} columns",
673 dest.len(),
674 row.len()
675 )));
676 }
677
678 for (i, value) in row.iter().enumerate() {
679 dest[i] = value.clone();
680 }
681
682 Ok(())
683 }
684
685 fn row(&self) -> &Row {
686 match self.current_index {
687 Some(i) if i < self.rows.len() => &self.rows[i],
688 _ => panic!("row() called without successful next()"),
689 }
690 }
691
692 fn take_row(&mut self) -> Row {
694 match self.current_index {
695 Some(i) if i < self.rows.len() => std::mem::take(&mut self.rows[i]),
696 _ => panic!("take_row() called without successful next()"),
697 }
698 }
699
700 fn close(&mut self) -> Result<()> {
701 self.closed = true;
702 Ok(())
703 }
704
705 fn rows_affected(&self) -> i64 {
706 self.rows_affected
707 }
708
709 fn last_insert_id(&self) -> i64 {
710 self.last_insert_id
711 }
712
713 fn with_aliases(
714 mut self: Box<Self>,
715 aliases: FxHashMap<String, String>,
716 ) -> Box<dyn QueryResult> {
717 for col in &mut self.columns {
719 for (alias, original) in &aliases {
721 if col == original {
722 *col = alias.clone();
723 break;
724 }
725 }
726 }
727 self
728 }
729}
730
731pub struct EmptyResult {
733 columns: Vec<String>,
734 rows_affected: i64,
735 last_insert_id: i64,
736}
737
738impl EmptyResult {
739 pub fn new() -> Self {
741 Self {
742 columns: Vec::new(),
743 rows_affected: 0,
744 last_insert_id: 0,
745 }
746 }
747
748 pub fn for_modification(rows_affected: i64, last_insert_id: i64) -> Self {
750 Self {
751 columns: Vec::new(),
752 rows_affected,
753 last_insert_id,
754 }
755 }
756}
757
758impl Default for EmptyResult {
759 fn default() -> Self {
760 Self::new()
761 }
762}
763
764impl QueryResult for EmptyResult {
765 fn columns(&self) -> &[String] {
766 &self.columns
767 }
768
769 fn next(&mut self) -> bool {
770 false
771 }
772
773 fn scan(&self, _dest: &mut [Value]) -> Result<()> {
774 Err(radixdb_core::Error::internal(
775 "scan() called on empty result",
776 ))
777 }
778
779 fn row(&self) -> &Row {
780 panic!("row() called on empty result")
781 }
782
783 fn close(&mut self) -> Result<()> {
784 Ok(())
785 }
786
787 fn rows_affected(&self) -> i64 {
788 self.rows_affected
789 }
790
791 fn last_insert_id(&self) -> i64 {
792 self.last_insert_id
793 }
794
795 fn with_aliases(self: Box<Self>, _aliases: FxHashMap<String, String>) -> Box<dyn QueryResult> {
796 self
797 }
798}
799
800#[cfg(test)]
801mod tests {
802 use super::*;
803 use crate::traits::VecScanner;
804
805 #[test]
806 fn test_memory_result_empty() {
807 let mut result = MemoryResult::new(vec!["id".to_string(), "name".to_string()]);
808
809 assert_eq!(result.columns(), &["id", "name"]);
810 assert!(!result.next());
811 assert_eq!(result.rows_affected(), 0);
812 assert_eq!(result.last_insert_id(), 0);
813 }
814
815 #[test]
816 fn test_memory_result_with_rows() {
817 let rows = vec![
818 Row::from_values(vec![Value::Integer(1), Value::text("Alice")]),
819 Row::from_values(vec![Value::Integer(2), Value::text("Bob")]),
820 ];
821
822 let mut result = MemoryResult::with_rows(vec!["id".to_string(), "name".to_string()], rows);
823
824 assert!(result.next());
825 assert_eq!(result.row().get(0), Some(&Value::Integer(1)));
826
827 assert!(result.next());
828 assert_eq!(result.row().get(0), Some(&Value::Integer(2)));
829
830 assert!(!result.next());
831 }
832
833 #[test]
834 fn test_memory_result_scan() {
835 let rows = vec![Row::from_values(vec![
836 Value::Integer(42),
837 Value::text("test"),
838 ])];
839
840 let mut result = MemoryResult::with_rows(vec!["id".to_string(), "name".to_string()], rows);
841
842 assert!(result.next());
843
844 let mut dest = vec![Value::null_unknown(), Value::null_unknown()];
845 result.scan(&mut dest).unwrap();
846
847 assert_eq!(dest[0], Value::Integer(42));
848 assert_eq!(dest[1], Value::text("test"));
849 }
850
851 #[test]
852 fn test_memory_result_for_modification() {
853 let result = MemoryResult::for_modification(5, 100);
854
855 assert_eq!(result.rows_affected(), 5);
856 assert_eq!(result.last_insert_id(), 100);
857 }
858
859 #[test]
860 fn test_memory_result_close() {
861 let rows = vec![Row::from_values(vec![Value::Integer(1)])];
862 let mut result = MemoryResult::with_rows(vec!["id".to_string()], rows);
863
864 assert!(result.next());
865 assert!(result.close().is_ok());
866 assert!(!result.next()); }
868
869 #[test]
870 fn test_memory_result_with_aliases() {
871 let rows = vec![Row::from_values(vec![Value::Integer(1)])];
872 let result = Box::new(MemoryResult::with_rows(vec!["user_id".to_string()], rows));
873
874 let mut aliases = FxHashMap::default();
875 aliases.insert("id".to_string(), "user_id".to_string());
876
877 let aliased = result.with_aliases(aliases);
878 assert_eq!(aliased.columns(), &["id"]);
879 }
880
881 #[test]
882 fn test_empty_result() {
883 let mut result = EmptyResult::new();
884
885 assert!(result.columns().is_empty());
886 assert!(!result.next());
887 assert_eq!(result.rows_affected(), 0);
888 assert!(result.close().is_ok());
889 }
890
891 #[test]
892 fn test_empty_result_for_modification() {
893 let result = EmptyResult::for_modification(10, 0);
894
895 assert_eq!(result.rows_affected(), 10);
896 assert_eq!(result.last_insert_id(), 0);
897 }
898
899 #[test]
900 fn scanner_result_streams_rows_and_preserves_alias_wrapper_contract() {
901 let scanner = VecScanner::new(vec![
902 Row::from_values(vec![Value::Integer(1), Value::text("one")]),
903 Row::from_values(vec![Value::Integer(2), Value::text("two")]),
904 ]);
905 let result: Box<dyn QueryResult> = Box::new(ScannerResult::new(
906 Box::new(scanner),
907 vec!["id".to_string(), "payload".to_string()],
908 ));
909 let mut aliases = FxHashMap::default();
910 aliases.insert("value".to_string(), "payload".to_string());
911 let mut result = result.with_aliases(aliases);
912
913 assert_eq!(result.columns(), &["id", "value"]);
914 assert!(result.next());
915 assert_eq!(result.take_row().get(0), Some(&Value::Integer(1)));
916 assert!(result.next());
917 assert_eq!(result.row().get(1), Some(&Value::text("two")));
918 assert!(!result.next());
919 assert!(result.last_error().is_none());
920 result.close().unwrap();
921 }
922
923 #[test]
924 fn scanner_result_surfaces_scanner_terminal_error() {
925 let scanner = VecScanner::with_error(radixdb_core::Error::internal("scanner terminal"));
926 let mut result = ScannerResult::new(Box::new(scanner), vec!["id".to_string()]);
927
928 assert!(!result.next());
929 assert!(result
930 .last_error()
931 .is_some_and(|error| error.to_string().contains("scanner terminal")));
932 }
933}