1use crate::util::udt_json::udt_to_json_object;
7use crate::{schema::CqlType, RowKey, Value};
8pub use crate::types::{CellExpiration, CellWriteMetadata};
11use base64::Engine;
12use serde::{Deserialize, Serialize};
13use serde_json::json;
14use std::collections::HashMap;
15use std::fmt;
16use std::sync::Arc;
17use tokio::sync::mpsc;
18
19fn b64(bytes: &[u8]) -> String {
21 base64::engine::general_purpose::STANDARD.encode(bytes)
22}
23
24fn row_metadata_is_populated(meta: &RowMetadata) -> bool {
26 meta.version.is_some() || meta.ttl.is_some() || !meta.tags.is_empty()
27}
28
29#[derive(Debug, Clone, Default)]
43pub struct ProjectionFlags {
44 pub include_cell_metadata: bool,
51}
52
53#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct QueryResult {
56 pub rows: Vec<QueryRow>,
58 pub rows_affected: u64,
60 pub execution_time_ms: u64,
62 pub metadata: QueryMetadata,
64}
65
66#[derive(Debug, Clone, Serialize, Deserialize)]
68pub struct QueryRow {
69 pub values: HashMap<Arc<str>, Value>,
79 pub key: RowKey,
81 pub metadata: RowMetadata,
83 #[serde(skip_serializing_if = "Option::is_none", default)]
94 pub cell_metadata: Option<HashMap<String, CellWriteMetadata>>,
95}
96
97#[derive(Debug, Clone, Default, Serialize, Deserialize)]
99pub struct QueryMetadata {
100 pub columns: Vec<ColumnInfo>,
102 pub total_rows: Option<u64>,
104 pub plan_info: Option<PlanInfo>,
106 pub performance: PerformanceMetrics,
108 pub warnings: Vec<String>,
110 #[serde(skip_serializing_if = "Option::is_none", default)]
118 pub access_path: Option<crate::query::access_path::AccessPath>,
119}
120
121#[derive(Debug, Clone, Serialize, Deserialize)]
123pub struct ColumnInfo {
124 pub name: String,
126 pub data_type: crate::types::DataType,
128 pub nullable: bool,
130 pub position: usize,
132 pub table_name: Option<String>,
134 #[serde(skip_serializing_if = "Option::is_none")]
143 pub cql_type: Option<CqlType>,
144}
145
146#[derive(Debug, Clone, Default, Serialize, Deserialize)]
148pub struct RowMetadata {
149 pub version: Option<u64>,
151 pub ttl: Option<u64>,
153 pub tags: HashMap<String, String>,
155}
156
157#[derive(Debug, Clone, Serialize, Deserialize)]
159pub struct PlanInfo {
160 pub plan_type: String,
162 pub estimated_cost: f64,
164 pub actual_cost: f64,
166 pub indexes_used: Vec<String>,
168 pub steps: Vec<String>,
170 pub parallelization: Option<ParallelizationInfo>,
172}
173
174#[derive(Debug, Clone, Serialize, Deserialize)]
176pub struct ParallelizationInfo {
177 pub threads_used: usize,
179 pub effective: bool,
181 pub partitions: Vec<PartitionInfo>,
183}
184
185#[derive(Debug, Clone, Serialize, Deserialize)]
187pub struct PartitionInfo {
188 pub id: usize,
190 pub rows_processed: u64,
192 pub processing_time_ms: u64,
194}
195
196#[derive(Debug, Clone, Default, Serialize, Deserialize)]
198pub struct PerformanceMetrics {
199 pub parse_time_us: u64,
201 pub planning_time_us: u64,
203 pub execution_time_us: u64,
205 pub total_time_us: u64,
207 pub memory_usage_bytes: u64,
209 pub io_operations: u64,
211 pub cache_hits: u64,
213 pub cache_misses: u64,
215}
216
217#[derive(Debug, Clone)]
226pub struct StreamingConfig {
227 pub buffer_size: usize,
230 pub chunk_size: usize,
233}
234
235impl Default for StreamingConfig {
236 fn default() -> Self {
237 Self {
238 buffer_size: 1024, chunk_size: 10_000, }
241 }
242}
243
244impl StreamingConfig {
245 pub fn new(buffer_size: usize, chunk_size: usize) -> Self {
247 Self {
248 buffer_size,
249 chunk_size,
250 }
251 }
252
253 pub fn for_parquet() -> Self {
255 Self {
256 buffer_size: 1024,
257 chunk_size: 10_000, }
259 }
260
261 pub fn for_text_formats() -> Self {
263 Self {
264 buffer_size: 512,
265 chunk_size: 5_000, }
267 }
268}
269
270pub struct QueryResultIterator {
317 receiver: mpsc::Receiver<Result<QueryRow, crate::Error>>,
319 pub metadata: QueryMetadata,
321 pub total_rows_hint: Option<u64>,
323 rows_received: u64,
325}
326
327impl QueryResultIterator {
328 pub fn new(
330 receiver: mpsc::Receiver<Result<QueryRow, crate::Error>>,
331 metadata: QueryMetadata,
332 ) -> Self {
333 Self {
334 receiver,
335 metadata,
336 total_rows_hint: None,
337 rows_received: 0,
338 }
339 }
340
341 pub fn with_total_hint(mut self, total: u64) -> Self {
343 self.total_rows_hint = Some(total);
344 self
345 }
346
347 pub async fn next_async(&mut self) -> Option<Result<QueryRow, crate::Error>> {
351 let result = self.receiver.recv().await?;
352 if result.is_ok() {
353 self.rows_received += 1;
354 }
355 Some(result)
356 }
357
358 const MAX_CHUNK_SIZE: usize = 100_000;
360
361 pub async fn collect_chunk(&mut self, size: usize) -> Result<Vec<QueryRow>, crate::Error> {
376 let safe_size = size.min(Self::MAX_CHUNK_SIZE);
377 let mut chunk = Vec::new();
379 while chunk.len() < safe_size {
380 match self.receiver.recv().await {
381 Some(Ok(row)) => {
382 self.rows_received += 1;
383 chunk.push(row);
384 }
385 Some(Err(e)) => return Err(e),
386 None => break,
387 }
388 }
389 Ok(chunk)
390 }
391
392 pub fn rows_received(&self) -> u64 {
394 self.rows_received
395 }
396
397 pub fn progress_percent(&self) -> Option<f64> {
399 self.total_rows_hint.map(|total| {
400 if total == 0 {
401 100.0
402 } else {
403 (self.rows_received as f64 / total as f64) * 100.0
404 }
405 })
406 }
407}
408
409impl QueryResult {
410 pub fn new() -> Self {
412 Self {
413 rows: Vec::new(),
414 rows_affected: 0,
415 execution_time_ms: 0,
416 metadata: QueryMetadata::default(),
417 }
418 }
419
420 pub fn with_rows(rows: Vec<QueryRow>) -> Self {
422 Self {
423 rows,
424 ..Self::new()
425 }
426 }
427
428 pub fn with_affected_rows(rows_affected: u64) -> Self {
430 Self {
431 rows_affected,
432 ..Self::new()
433 }
434 }
435
436 pub fn row_count(&self) -> usize {
438 self.rows.len()
439 }
440
441 pub fn is_empty(&self) -> bool {
443 self.rows.is_empty()
444 }
445
446 pub fn get_row(&self, index: usize) -> Option<&QueryRow> {
448 self.rows.get(index)
449 }
450
451 pub fn columns(&self) -> &[ColumnInfo] {
453 &self.metadata.columns
454 }
455
456 pub fn column_names(&self) -> Vec<String> {
458 self.metadata
459 .columns
460 .iter()
461 .map(|c| c.name.clone())
462 .collect()
463 }
464
465 pub fn execution_time(&self) -> u64 {
467 self.execution_time_ms
468 }
469
470 pub fn performance(&self) -> &PerformanceMetrics {
472 &self.metadata.performance
473 }
474
475 pub fn warnings(&self) -> &[String] {
477 &self.metadata.warnings
478 }
479
480 pub fn add_warning(&mut self, warning: String) {
482 self.metadata.warnings.push(warning);
483 }
484
485 pub fn to_json(&self) -> serde_json::Value {
490 let rows: Vec<_> = self
491 .rows
492 .iter()
493 .map(|row| self.row_to_json_deterministic(row))
494 .collect();
495 let columns: Vec<_> = self
496 .metadata
497 .columns
498 .iter()
499 .map(ColumnInfo::to_json)
500 .collect();
501 let warnings: Vec<_> = self
502 .metadata
503 .warnings
504 .iter()
505 .cloned()
506 .map(serde_json::Value::String)
507 .collect();
508
509 json!({
510 "rows": rows,
511 "rows_affected": self.rows_affected,
512 "row_count": self.rows.len(),
513 "columns": columns,
514 "performance": self.metadata.performance.to_json(),
515 "warnings": warnings,
516 })
517 }
518
519 pub fn iter(&self) -> std::slice::Iter<'_, QueryRow> {
521 self.rows.iter()
522 }
523
524 fn row_to_json_deterministic(&self, row: &QueryRow) -> serde_json::Value {
529 let mut result = serde_json::Map::new();
530
531 if !self.metadata.columns.is_empty() {
532 for col in &self.metadata.columns {
533 let value_json = row
534 .values
535 .get(col.name.as_str())
536 .map_or(serde_json::Value::Null, ToJson::to_json);
537 result.insert(col.name.clone(), value_json);
538 }
539 } else {
540 let mut sorted_keys: Vec<&Arc<str>> = row.values.keys().collect();
541 sorted_keys.sort();
542 for key in sorted_keys {
543 if let Some(value) = row.values.get(key.as_ref()) {
544 result.insert(key.to_string(), value.to_json());
545 }
546 }
547 }
548
549 result.insert(
550 "_key".to_string(),
551 serde_json::Value::String(format!("{:?}", row.key)),
552 );
553
554 if row_metadata_is_populated(&row.metadata) {
555 result.insert("_metadata".to_string(), row.metadata.to_json());
556 }
557
558 serde_json::Value::Object(result)
559 }
560}
561
562impl QueryRow {
563 pub fn new(key: RowKey) -> Self {
565 Self {
566 values: HashMap::new(),
567 key,
568 metadata: RowMetadata::default(),
569 cell_metadata: None,
570 }
571 }
572
573 pub fn with_values(key: RowKey, values: HashMap<String, Value>) -> Self {
582 Self {
583 values: values.into_iter().map(|(k, v)| (Arc::from(k), v)).collect(),
584 key,
585 metadata: RowMetadata::default(),
586 cell_metadata: None,
587 }
588 }
589
590 pub fn with_interned_values(key: RowKey, values: HashMap<Arc<str>, Value>) -> Self {
597 Self {
598 values,
599 key,
600 metadata: RowMetadata::default(),
601 cell_metadata: None,
602 }
603 }
604
605 pub fn from_map(values: HashMap<String, Value>) -> Self {
612 Self {
613 values: values.into_iter().map(|(k, v)| (Arc::from(k), v)).collect(),
614 key: RowKey::new(vec![]),
615 metadata: RowMetadata::default(),
616 cell_metadata: None,
617 }
618 }
619
620 pub fn get(&self, column: &str) -> Option<&Value> {
622 self.values.get(column)
623 }
624
625 pub fn set(&mut self, column: impl Into<Arc<str>>, value: Value) {
630 self.values.insert(column.into(), value);
631 }
632
633 pub fn column_names(&self) -> Vec<String> {
635 self.values.keys().map(|k| k.to_string()).collect()
636 }
637
638 pub fn key(&self) -> &RowKey {
640 &self.key
641 }
642
643 pub fn metadata(&self) -> &RowMetadata {
645 &self.metadata
646 }
647
648 pub fn set_metadata(&mut self, metadata: RowMetadata) {
650 self.metadata = metadata;
651 }
652
653 pub fn set_cell_metadata(&mut self, map: HashMap<String, CellWriteMetadata>) {
660 self.cell_metadata = Some(map);
661 }
662
663 pub fn insert_cell_metadata(&mut self, column: String, meta: CellWriteMetadata) {
669 self.cell_metadata
670 .get_or_insert_with(HashMap::new)
671 .insert(column, meta);
672 }
673
674 pub fn get_cell_metadata(&self, column: &str) -> Option<&CellWriteMetadata> {
676 self.cell_metadata.as_ref()?.get(column)
677 }
678
679 pub fn to_json(&self) -> serde_json::Value {
681 let mut result = serde_json::Map::new();
682
683 for (column, value) in &self.values {
684 result.insert(column.to_string(), value.to_json());
685 }
686
687 result.insert(
688 "_key".to_string(),
689 serde_json::Value::String(format!("{:?}", self.key)),
690 );
691
692 if row_metadata_is_populated(&self.metadata) {
693 result.insert("_metadata".to_string(), self.metadata.to_json());
694 }
695
696 serde_json::Value::Object(result)
697 }
698}
699
700impl ColumnInfo {
701 pub fn new(
703 name: String,
704 data_type: crate::types::DataType,
705 nullable: bool,
706 position: usize,
707 ) -> Self {
708 Self {
709 name,
710 data_type,
711 nullable,
712 position,
713 table_name: None,
714 cql_type: None,
715 }
716 }
717
718 pub fn with_table_name(mut self, table_name: String) -> Self {
720 self.table_name = Some(table_name);
721 self
722 }
723
724 pub fn with_cql_type(mut self, cql_type: CqlType) -> Self {
730 self.cql_type = Some(cql_type);
731 self
732 }
733
734 pub fn to_json(&self) -> serde_json::Value {
741 let mut map = serde_json::Map::new();
742 map.insert("name".to_string(), json!(self.name));
743 map.insert(
744 "data_type".to_string(),
745 json!(format!("{:?}", self.data_type)),
746 );
747 map.insert("nullable".to_string(), json!(self.nullable));
748 map.insert("position".to_string(), json!(self.position));
749 if let Some(table_name) = &self.table_name {
750 map.insert("table_name".to_string(), json!(table_name));
751 }
752 if let Some(cql_type) = &self.cql_type {
754 map.insert("cql_type".to_string(), json!(format_cql_type(cql_type)));
755 }
756 serde_json::Value::Object(map)
757 }
758}
759
760pub fn cql_type_to_data_type(cql_type: &CqlType) -> crate::types::DataType {
766 use crate::types::DataType;
767 match cql_type {
768 CqlType::Boolean => DataType::Boolean,
769 CqlType::TinyInt => DataType::TinyInt,
770 CqlType::SmallInt => DataType::SmallInt,
771 CqlType::Int => DataType::Integer,
772 CqlType::BigInt | CqlType::Varint | CqlType::Counter => DataType::BigInt,
773 CqlType::Float => DataType::Float32,
774 CqlType::Double | CqlType::Decimal => DataType::Float,
775 CqlType::Text | CqlType::Varchar | CqlType::Ascii => DataType::Text,
776 CqlType::Blob => DataType::Blob,
777 CqlType::Timestamp => DataType::Timestamp,
778 CqlType::Date | CqlType::Time | CqlType::Duration | CqlType::Inet => DataType::BigInt,
779 CqlType::Uuid | CqlType::TimeUuid => DataType::Uuid,
780 CqlType::List(_) | CqlType::Vector(_, _) => DataType::List,
781 CqlType::Set(_) => DataType::Set,
782 CqlType::Map(_, _) => DataType::Map,
783 CqlType::Tuple(_) => DataType::Tuple,
784 CqlType::Udt(_, _) => DataType::Udt,
785 CqlType::Frozen(inner) => cql_type_to_data_type(inner),
786 CqlType::Custom(_) => DataType::Blob,
787 }
788}
789
790fn format_cql_type(cql_type: &CqlType) -> String {
794 match cql_type {
795 CqlType::Boolean => "boolean".to_string(),
796 CqlType::TinyInt => "tinyint".to_string(),
797 CqlType::SmallInt => "smallint".to_string(),
798 CqlType::Int => "int".to_string(),
799 CqlType::BigInt => "bigint".to_string(),
800 CqlType::Counter => "counter".to_string(),
801 CqlType::Float => "float".to_string(),
802 CqlType::Double => "double".to_string(),
803 CqlType::Decimal => "decimal".to_string(),
804 CqlType::Text => "text".to_string(),
805 CqlType::Varchar => "varchar".to_string(),
806 CqlType::Ascii => "ascii".to_string(),
807 CqlType::Blob => "blob".to_string(),
808 CqlType::Timestamp => "timestamp".to_string(),
809 CqlType::Date => "date".to_string(),
810 CqlType::Time => "time".to_string(),
811 CqlType::Uuid => "uuid".to_string(),
812 CqlType::TimeUuid => "timeuuid".to_string(),
813 CqlType::Inet => "inet".to_string(),
814 CqlType::Duration => "duration".to_string(),
815 CqlType::Varint => "varint".to_string(),
816 CqlType::List(inner) => format!("list<{}>", format_cql_type(inner)),
817 CqlType::Set(inner) => format!("set<{}>", format_cql_type(inner)),
818 CqlType::Map(k, v) => format!("map<{}, {}>", format_cql_type(k), format_cql_type(v)),
819 CqlType::Tuple(types) => {
820 let inner: Vec<_> = types.iter().map(format_cql_type).collect();
821 format!("tuple<{}>", inner.join(", "))
822 }
823 CqlType::Udt(name, _) => name.clone(),
824 CqlType::Frozen(inner) => format!("frozen<{}>", format_cql_type(inner)),
825 CqlType::Vector(e, n) => format!("vector<{}, {n}>", format_cql_type(e)),
826 CqlType::Custom(name) => name.clone(),
827 }
828}
829
830impl RowMetadata {
831 pub fn new() -> Self {
833 Self::default()
834 }
835
836 pub fn with_version(mut self, version: u64) -> Self {
838 self.version = Some(version);
839 self
840 }
841
842 pub fn with_ttl(mut self, ttl: u64) -> Self {
844 self.ttl = Some(ttl);
845 self
846 }
847
848 pub fn with_tag(mut self, key: String, value: String) -> Self {
850 self.tags.insert(key, value);
851 self
852 }
853
854 pub fn to_json(&self) -> serde_json::Value {
856 let mut map = serde_json::Map::new();
857 if let Some(version) = self.version {
858 map.insert("version".to_string(), json!(version));
859 }
860 if let Some(ttl) = self.ttl {
861 map.insert("ttl".to_string(), json!(ttl));
862 }
863 if !self.tags.is_empty() {
864 map.insert("tags".to_string(), json!(self.tags));
865 }
866 serde_json::Value::Object(map)
867 }
868}
869
870impl PerformanceMetrics {
871 pub fn new() -> Self {
873 Self::default()
874 }
875
876 pub fn total_time_ms(&self) -> u64 {
878 self.total_time_us / 1000
879 }
880
881 pub fn cache_hit_ratio(&self) -> f64 {
883 let total = self.cache_hits + self.cache_misses;
884 if total == 0 {
885 0.0
886 } else {
887 self.cache_hits as f64 / total as f64
888 }
889 }
890
891 pub fn to_json(&self) -> serde_json::Value {
893 let cache_hit_ratio = serde_json::Number::from_f64(self.cache_hit_ratio())
894 .map(serde_json::Value::Number)
895 .unwrap_or(json!(0));
896 json!({
897 "parse_time_us": self.parse_time_us,
898 "planning_time_us": self.planning_time_us,
899 "execution_time_us": self.execution_time_us,
900 "total_time_us": self.total_time_us,
901 "memory_usage_bytes": self.memory_usage_bytes,
902 "io_operations": self.io_operations,
903 "cache_hits": self.cache_hits,
904 "cache_misses": self.cache_misses,
905 "cache_hit_ratio": cache_hit_ratio,
906 })
907 }
908}
909
910fn write_border(
912 f: &mut fmt::Formatter<'_>,
913 widths: &[usize],
914 left: char,
915 sep: char,
916 right: char,
917) -> fmt::Result {
918 write!(f, "{}", left)?;
919 for (i, width) in widths.iter().enumerate() {
920 write!(f, "{}", "─".repeat(width + 2))?;
921 if i < widths.len() - 1 {
922 write!(f, "{}", sep)?;
923 }
924 }
925 writeln!(f, "{}", right)
926}
927
928impl fmt::Display for QueryResult {
929 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
930 if self.rows.is_empty() {
931 return write!(f, "Empty result set ({} rows affected)", self.rows_affected);
932 }
933
934 let column_names = self.column_names();
935 if column_names.is_empty() {
936 return write!(f, "No columns in result set");
937 }
938
939 let col_widths: Vec<usize> = column_names
941 .iter()
942 .map(|col_name| {
943 self.rows
944 .iter()
945 .filter_map(|row| row.values.get(col_name.as_str()))
946 .map(|v| format!("{}", v).len())
947 .max()
948 .unwrap_or(0)
949 .max(col_name.len())
950 })
951 .collect();
952
953 write_border(f, &col_widths, '┌', '┬', '┐')?;
954
955 write!(f, "│")?;
956 for (i, (col_name, width)) in column_names.iter().zip(col_widths.iter()).enumerate() {
957 write!(f, " {:width$} ", col_name, width = width)?;
958 if i < column_names.len() - 1 {
959 write!(f, "│")?;
960 }
961 }
962 writeln!(f, "│")?;
963
964 write_border(f, &col_widths, '├', '┼', '┤')?;
965
966 for row in &self.rows {
967 write!(f, "│")?;
968 for (i, (col_name, width)) in column_names.iter().zip(col_widths.iter()).enumerate() {
969 let value = row
970 .values
971 .get(col_name.as_str())
972 .map(|v| format!("{}", v))
973 .unwrap_or_else(|| "NULL".to_string());
974 write!(f, " {:width$} ", value, width = width)?;
975 if i < column_names.len() - 1 {
976 write!(f, "│")?;
977 }
978 }
979 writeln!(f, "│")?;
980 }
981
982 write_border(f, &col_widths, '└', '┴', '┘')?;
983
984 writeln!(
985 f,
986 "{} rows returned in {}ms",
987 self.rows.len(),
988 self.execution_time_ms
989 )?;
990
991 if !self.metadata.warnings.is_empty() {
992 writeln!(f, "\nWarnings:")?;
993 for warning in &self.metadata.warnings {
994 writeln!(f, " - {}", warning)?;
995 }
996 }
997
998 Ok(())
999 }
1000}
1001
1002impl Default for QueryResult {
1003 fn default() -> Self {
1004 Self::new()
1005 }
1006}
1007
1008impl IntoIterator for QueryResult {
1009 type Item = QueryRow;
1010 type IntoIter = std::vec::IntoIter<QueryRow>;
1011
1012 fn into_iter(self) -> Self::IntoIter {
1013 self.rows.into_iter()
1014 }
1015}
1016
1017impl<'a> IntoIterator for &'a QueryResult {
1018 type Item = &'a QueryRow;
1019 type IntoIter = std::slice::Iter<'a, QueryRow>;
1020
1021 fn into_iter(self) -> Self::IntoIter {
1022 self.rows.iter()
1023 }
1024}
1025
1026trait ToJson {
1028 fn to_json(&self) -> serde_json::Value;
1029}
1030
1031fn json_map_key(key: &Value) -> String {
1057 match key {
1058 Value::Empty(_) => String::new(),
1059 other => format!("{other}"),
1060 }
1061}
1062
1063impl ToJson for Value {
1064 fn to_json(&self) -> serde_json::Value {
1065 fn float_to_json(x: f64) -> serde_json::Value {
1067 serde_json::Number::from_f64(x)
1068 .map(serde_json::Value::Number)
1069 .unwrap_or(serde_json::Value::Null)
1070 }
1071
1072 match self {
1073 Value::Null => serde_json::Value::Null,
1074 Value::Empty(_) => json!(""),
1084 Value::Boolean(b) => json!(*b),
1085 Value::Integer(i) => json!(*i),
1086 Value::BigInt(i) => json!(*i),
1087 Value::Counter(c) => json!(*c),
1088 Value::TinyInt(i) => json!(*i as i64),
1089 Value::SmallInt(i) => json!(*i as i64),
1090 Value::Date(d) => json!(*d),
1091 Value::Time(t) => json!(*t),
1092 Value::Timestamp(ts) => json!(*ts),
1093 Value::Float(f) => float_to_json(*f),
1094 Value::Float32(f) => float_to_json(*f as f64),
1095 Value::Text(s) => json!(std::str::from_utf8(s).unwrap_or_default()),
1096 Value::Json(value) => (**value).clone(),
1097 Value::Blob(bytes) | Value::Varint(bytes) | Value::Inet(bytes) => json!(b64(bytes)),
1098 Value::Uuid(uuid) => json!(b64(uuid)),
1099 Value::List(items) | Value::Set(items) | Value::Tuple(items) => {
1100 let json_list: Vec<_> = items.iter().map(ToJson::to_json).collect();
1101 serde_json::Value::Array(json_list)
1102 }
1103 Value::Map(entries) => {
1104 let json_map: serde_json::Map<String, serde_json::Value> = entries
1105 .iter()
1106 .map(|(k, v)| (json_map_key(k), v.to_json()))
1107 .collect();
1108 serde_json::Value::Object(json_map)
1109 }
1110 Value::Udt(udt) => udt_to_json_object(udt, ToJson::to_json),
1114 Value::Frozen(boxed) => boxed.to_json(),
1115 Value::Decimal { scale, unscaled } => json!({
1116 "scale": *scale,
1117 "unscaled": b64(unscaled),
1118 }),
1119 Value::Duration {
1120 months,
1121 days,
1122 nanos,
1123 } => json!({
1124 "months": *months,
1125 "days": *days,
1126 "nanos": *nanos,
1127 }),
1128 Value::Tombstone(info) => {
1129 let mut json_obj = serde_json::Map::new();
1130 json_obj.insert("type".to_string(), json!("tombstone"));
1131 json_obj.insert("deletion_time".to_string(), json!(info.deletion_time));
1132 json_obj.insert(
1133 "tombstone_type".to_string(),
1134 json!(format!("{:?}", info.tombstone_type)),
1135 );
1136 if let Some(ttl) = info.ttl {
1137 json_obj.insert("ttl".to_string(), json!(ttl));
1138 }
1139 serde_json::Value::Object(json_obj)
1140 }
1141 }
1142 }
1143}
1144
1145#[cfg(test)]
1146mod tests {
1147 use super::*;
1148 use crate::Value;
1149
1150 #[test]
1151 fn test_query_result_creation() {
1152 let result = QueryResult::new();
1153 assert!(result.is_empty());
1154 assert_eq!(result.row_count(), 0);
1155 assert_eq!(result.execution_time(), 0);
1156 }
1157
1158 #[test]
1159 fn test_query_result_with_rows() {
1160 let mut row1 = QueryRow::new(RowKey::new(vec![1]));
1161 row1.set("id".to_string(), Value::Integer(1));
1162 row1.set("name".to_string(), Value::text("Alice".to_string()));
1163
1164 let mut row2 = QueryRow::new(RowKey::new(vec![2]));
1165 row2.set("id".to_string(), Value::Integer(2));
1166 row2.set("name".to_string(), Value::text("Bob".to_string()));
1167
1168 let result = QueryResult::with_rows(vec![row1, row2]);
1169 assert_eq!(result.row_count(), 2);
1170 assert!(!result.is_empty());
1171
1172 let first_row = result.get_row(0).unwrap();
1173 assert_eq!(first_row.get("id"), Some(&Value::Integer(1)));
1174 assert_eq!(
1175 first_row.get("name"),
1176 Some(&Value::text("Alice".to_string()))
1177 );
1178 }
1179
1180 #[test]
1181 fn test_query_row_operations() {
1182 let mut row = QueryRow::new(RowKey::new(vec![1]));
1183 row.set("id".to_string(), Value::Integer(42));
1184 row.set("active".to_string(), Value::Boolean(true));
1185
1186 assert_eq!(row.get("id"), Some(&Value::Integer(42)));
1187 assert_eq!(row.get("active"), Some(&Value::Boolean(true)));
1188 assert_eq!(row.get("nonexistent"), None);
1189
1190 let column_names = row.column_names();
1191 assert_eq!(column_names.len(), 2);
1192 assert!(column_names.contains(&"id".to_string()));
1193 assert!(column_names.contains(&"active".to_string()));
1194 }
1195
1196 #[test]
1197 fn test_column_info() {
1198 let column = ColumnInfo::new(
1199 "user_id".to_string(),
1200 crate::types::DataType::Integer,
1201 false,
1202 0,
1203 )
1204 .with_table_name("users".to_string());
1205
1206 assert_eq!(column.name, "user_id");
1207 assert_eq!(column.data_type, crate::types::DataType::Integer);
1208 assert!(!column.nullable);
1209 assert_eq!(column.position, 0);
1210 assert_eq!(column.table_name, Some("users".to_string()));
1211 assert!(column.cql_type.is_none());
1212 }
1213
1214 #[test]
1215 fn test_column_info_with_cql_type_scalar() {
1216 use crate::schema::CqlType;
1217 let column = ColumnInfo::new("ts".to_string(), crate::types::DataType::Timestamp, true, 1)
1218 .with_cql_type(CqlType::Timestamp);
1219
1220 assert_eq!(column.name, "ts");
1221 assert_eq!(column.cql_type, Some(CqlType::Timestamp));
1222 assert_eq!(column.data_type, crate::types::DataType::Timestamp);
1224 }
1225
1226 #[test]
1227 fn test_column_info_with_cql_type_list() {
1228 use crate::schema::CqlType;
1229 let list_type = CqlType::List(Box::new(CqlType::Int));
1230 let column = ColumnInfo::new("items".to_string(), crate::types::DataType::List, true, 2)
1231 .with_cql_type(list_type.clone());
1232
1233 assert_eq!(column.cql_type, Some(list_type));
1234 }
1235
1236 #[test]
1237 fn test_column_info_with_cql_type_map() {
1238 use crate::schema::CqlType;
1239 let map_type = CqlType::Map(Box::new(CqlType::Text), Box::new(CqlType::BigInt));
1240 let column = ColumnInfo::new("props".to_string(), crate::types::DataType::Map, true, 3)
1241 .with_cql_type(map_type.clone());
1242
1243 assert_eq!(column.cql_type, Some(map_type));
1244 }
1245
1246 #[test]
1247 fn test_column_info_with_cql_type_udt() {
1248 use crate::schema::CqlType;
1249 let udt_type = CqlType::Udt("address".to_string(), vec![]);
1250 let column = ColumnInfo::new("addr".to_string(), crate::types::DataType::Udt, true, 4)
1251 .with_cql_type(udt_type.clone());
1252
1253 assert_eq!(column.cql_type, Some(udt_type));
1254 }
1255
1256 #[test]
1257 fn test_cql_type_to_data_type_scalars() {
1258 use super::cql_type_to_data_type;
1259 use crate::schema::CqlType;
1260 use crate::types::DataType;
1261
1262 assert_eq!(cql_type_to_data_type(&CqlType::Boolean), DataType::Boolean);
1263 assert_eq!(cql_type_to_data_type(&CqlType::Int), DataType::Integer);
1264 assert_eq!(cql_type_to_data_type(&CqlType::BigInt), DataType::BigInt);
1265 assert_eq!(cql_type_to_data_type(&CqlType::Text), DataType::Text);
1266 assert_eq!(cql_type_to_data_type(&CqlType::Blob), DataType::Blob);
1267 assert_eq!(cql_type_to_data_type(&CqlType::Uuid), DataType::Uuid);
1268 assert_eq!(
1269 cql_type_to_data_type(&CqlType::Timestamp),
1270 DataType::Timestamp
1271 );
1272 }
1273
1274 #[test]
1275 fn test_cql_type_to_data_type_collections() {
1276 use super::cql_type_to_data_type;
1277 use crate::schema::CqlType;
1278 use crate::types::DataType;
1279
1280 assert_eq!(
1281 cql_type_to_data_type(&CqlType::List(Box::new(CqlType::Int))),
1282 DataType::List
1283 );
1284 assert_eq!(
1285 cql_type_to_data_type(&CqlType::Set(Box::new(CqlType::Text))),
1286 DataType::Set
1287 );
1288 assert_eq!(
1289 cql_type_to_data_type(&CqlType::Map(
1290 Box::new(CqlType::Text),
1291 Box::new(CqlType::BigInt)
1292 )),
1293 DataType::Map
1294 );
1295 }
1296
1297 #[test]
1298 fn test_cql_type_to_data_type_frozen() {
1299 use super::cql_type_to_data_type;
1300 use crate::schema::CqlType;
1301 use crate::types::DataType;
1302
1303 assert_eq!(
1305 cql_type_to_data_type(&CqlType::Frozen(Box::new(CqlType::List(Box::new(
1306 CqlType::Int
1307 ))))),
1308 DataType::List
1309 );
1310 }
1311
1312 #[test]
1313 fn test_column_info_to_json_includes_cql_type() {
1314 use crate::schema::CqlType;
1315 let column = ColumnInfo::new("items".to_string(), crate::types::DataType::List, true, 0)
1316 .with_cql_type(CqlType::List(Box::new(CqlType::Int)));
1317
1318 let json = column.to_json();
1319 let obj = json.as_object().unwrap();
1320
1321 assert_eq!(obj["name"], "items");
1323 assert!(obj.contains_key("data_type"));
1324 assert!(obj.contains_key("nullable"));
1325 assert!(obj.contains_key("position"));
1326
1327 assert!(obj.contains_key("cql_type"));
1329 assert_eq!(obj["cql_type"], "list<int>");
1330 }
1331
1332 #[test]
1333 fn test_column_info_to_json_no_cql_type() {
1334 let column = ColumnInfo::new("id".to_string(), crate::types::DataType::Integer, false, 0);
1336
1337 let json = column.to_json();
1338 let obj = json.as_object().unwrap();
1339
1340 assert!(!obj.contains_key("cql_type"));
1341 }
1342
1343 #[test]
1344 fn test_row_metadata() {
1345 let metadata = RowMetadata::new()
1346 .with_version(123)
1347 .with_ttl(3600)
1348 .with_tag("source".to_string(), "import".to_string());
1349
1350 assert_eq!(metadata.version, Some(123));
1351 assert_eq!(metadata.ttl, Some(3600));
1352 assert_eq!(metadata.tags.get("source"), Some(&"import".to_string()));
1353 }
1354
1355 #[test]
1356 fn test_performance_metrics() {
1357 let mut metrics = PerformanceMetrics::new();
1358 metrics.cache_hits = 8;
1359 metrics.cache_misses = 2;
1360 metrics.total_time_us = 5000;
1361
1362 assert_eq!(metrics.cache_hit_ratio(), 0.8);
1363 assert_eq!(metrics.total_time_ms(), 5);
1364 }
1365
1366 #[test]
1367 fn test_json_serialization() {
1368 let mut row = QueryRow::new(RowKey::new(vec![1]));
1369 row.set("id".to_string(), Value::Integer(1));
1370 row.set("name".to_string(), Value::text("test".to_string()));
1371
1372 let json = row.to_json();
1373 assert!(json.is_object());
1374
1375 let obj = json.as_object().unwrap();
1376 assert_eq!(obj.get("id"), Some(&serde_json::Value::Number(1.into())));
1377 assert_eq!(
1378 obj.get("name"),
1379 Some(&serde_json::Value::String("test".to_string()))
1380 );
1381 }
1382
1383 #[test]
1384 fn test_result_iteration() {
1385 let row1 = QueryRow::new(RowKey::new(vec![1]));
1386 let row2 = QueryRow::new(RowKey::new(vec![2]));
1387 let result = QueryResult::with_rows(vec![row1, row2]);
1388
1389 let mut count = 0;
1390 for _row in &result {
1391 count += 1;
1392 }
1393 assert_eq!(count, 2);
1394
1395 let mut count = 0;
1396 for _row in result {
1397 count += 1;
1398 }
1399 assert_eq!(count, 2);
1400 }
1401
1402 #[test]
1409 fn test_cell_metadata_absent_by_default() {
1410 let row = QueryRow::new(RowKey::new(vec![1]));
1411 assert!(
1412 row.cell_metadata.is_none(),
1413 "cell_metadata must be None when no metadata is attached (hot-path, zero allocation)"
1414 );
1415
1416 let row2 = QueryRow::with_values(RowKey::new(vec![2]), HashMap::new());
1417 assert!(row2.cell_metadata.is_none());
1418
1419 let row3 = QueryRow::from_map(HashMap::new());
1420 assert!(row3.cell_metadata.is_none());
1421 }
1422
1423 #[test]
1425 fn test_projection_flags_default_no_metadata() {
1426 let flags = ProjectionFlags::default();
1427 assert!(
1428 !flags.include_cell_metadata,
1429 "include_cell_metadata must default to false"
1430 );
1431 }
1432
1433 #[test]
1435 fn test_cell_metadata_single_sstable_single_cell() {
1436 let mut row = QueryRow::new(RowKey::new(vec![1]));
1437 row.set("name".to_string(), Value::text("Alice".to_string()));
1438
1439 let meta = CellWriteMetadata {
1440 write_timestamp_micros: 1_700_000_000_000_000, expiration: None,
1442 };
1443 row.insert_cell_metadata("name".to_string(), meta.clone());
1444
1445 assert!(row.cell_metadata.is_some());
1447 assert_eq!(row.get("name"), Some(&Value::text("Alice".to_string())));
1449 let got = row
1451 .get_cell_metadata("name")
1452 .expect("metadata must be present");
1453 assert_eq!(got.write_timestamp_micros, meta.write_timestamp_micros);
1454 assert!(got.expiration.is_none());
1455 }
1456
1457 #[test]
1459 fn test_cell_metadata_with_ttl_expiration() {
1460 let mut row = QueryRow::new(RowKey::new(vec![2]));
1461 row.set("score".to_string(), Value::Integer(42));
1462
1463 let ttl_seconds = 3600_i32;
1464 let write_ts_micros = 1_700_000_000_000_000_i64;
1465 let expires_at = (write_ts_micros / 1_000_000) + ttl_seconds as i64;
1467
1468 let meta = CellWriteMetadata {
1469 write_timestamp_micros: write_ts_micros,
1470 expiration: Some(CellExpiration {
1471 ttl_seconds,
1472 expires_at_seconds: expires_at,
1473 }),
1474 };
1475 row.insert_cell_metadata("score".to_string(), meta);
1476
1477 let got = row.get_cell_metadata("score").unwrap();
1478 assert_eq!(got.write_timestamp_micros, write_ts_micros);
1479 let exp = got.expiration.as_ref().unwrap();
1480 assert_eq!(exp.ttl_seconds, 3600);
1481 assert_eq!(exp.expires_at_seconds, expires_at);
1482 }
1483
1484 #[test]
1487 fn test_cell_metadata_absent_for_null_cells() {
1488 let mut row = QueryRow::new(RowKey::new(vec![3]));
1489 row.set("id".to_string(), Value::Null);
1490 row.insert_cell_metadata(
1493 "name".to_string(),
1494 CellWriteMetadata {
1495 write_timestamp_micros: 42,
1496 expiration: None,
1497 },
1498 );
1499
1500 assert!(
1501 row.get_cell_metadata("id").is_none(),
1502 "no metadata for null column"
1503 );
1504 assert!(row.get_cell_metadata("name").is_some());
1505 }
1506
1507 #[test]
1517 fn test_cell_metadata_lww_winner_carries_newer_timestamp() {
1518 let mut older_row = QueryRow::new(RowKey::new(b"partition1".to_vec()));
1520 older_row.set("value".to_string(), Value::Integer(10));
1521 older_row.insert_cell_metadata(
1522 "value".to_string(),
1523 CellWriteMetadata {
1524 write_timestamp_micros: 1_000_000,
1525 expiration: None,
1526 },
1527 );
1528
1529 let mut newer_row = QueryRow::new(RowKey::new(b"partition1".to_vec()));
1531 newer_row.set("value".to_string(), Value::Integer(20));
1532 newer_row.insert_cell_metadata(
1533 "value".to_string(),
1534 CellWriteMetadata {
1535 write_timestamp_micros: 2_000_000,
1536 expiration: None,
1537 },
1538 );
1539
1540 let winner = if newer_row
1542 .get_cell_metadata("value")
1543 .map(|m| m.write_timestamp_micros)
1544 .unwrap_or(0)
1545 > older_row
1546 .get_cell_metadata("value")
1547 .map(|m| m.write_timestamp_micros)
1548 .unwrap_or(0)
1549 {
1550 newer_row
1551 } else {
1552 older_row
1553 };
1554
1555 assert_eq!(
1556 winner.get("value"),
1557 Some(&Value::Integer(20)),
1558 "value from the newer SSTable must be present"
1559 );
1560 assert_eq!(
1561 winner
1562 .get_cell_metadata("value")
1563 .map(|m| m.write_timestamp_micros),
1564 Some(2_000_000),
1565 "metadata must reflect the winning (newer) cell's timestamp"
1566 );
1567 }
1568
1569 #[test]
1571 fn test_set_cell_metadata_replaces_map() {
1572 let mut row = QueryRow::new(RowKey::new(vec![4]));
1573 row.insert_cell_metadata(
1574 "a".to_string(),
1575 CellWriteMetadata {
1576 write_timestamp_micros: 1,
1577 expiration: None,
1578 },
1579 );
1580
1581 let mut new_map = HashMap::new();
1582 new_map.insert(
1583 "b".to_string(),
1584 CellWriteMetadata {
1585 write_timestamp_micros: 99,
1586 expiration: None,
1587 },
1588 );
1589 row.set_cell_metadata(new_map);
1590
1591 assert!(row.get_cell_metadata("a").is_none());
1593 assert_eq!(
1594 row.get_cell_metadata("b").map(|m| m.write_timestamp_micros),
1595 Some(99)
1596 );
1597 }
1598
1599 #[test]
1601 fn test_cell_metadata_serde_round_trip() {
1602 let mut row = QueryRow::new(RowKey::new(vec![5]));
1603 row.set("x".to_string(), Value::Integer(7));
1604 row.insert_cell_metadata(
1605 "x".to_string(),
1606 CellWriteMetadata {
1607 write_timestamp_micros: 123_456_789,
1608 expiration: Some(CellExpiration {
1609 ttl_seconds: 60,
1610 expires_at_seconds: 9999,
1611 }),
1612 },
1613 );
1614
1615 let json = serde_json::to_string(&row).expect("serialise");
1616 let back: QueryRow = serde_json::from_str(&json).expect("deserialise");
1617
1618 let meta = back
1619 .get_cell_metadata("x")
1620 .expect("metadata present after round-trip");
1621 assert_eq!(meta.write_timestamp_micros, 123_456_789);
1622 let exp = meta.expiration.as_ref().unwrap();
1623 assert_eq!(exp.ttl_seconds, 60);
1624 assert_eq!(exp.expires_at_seconds, 9999);
1625 }
1626
1627 #[test]
1630 fn test_cell_metadata_none_omitted_from_json() {
1631 let row = QueryRow::new(RowKey::new(vec![6]));
1632 let json = serde_json::to_string(&row).expect("serialise");
1633 assert!(
1634 !json.contains("cell_metadata"),
1635 "cell_metadata must be absent from JSON when None (backward compat)"
1636 );
1637 }
1638
1639 #[test]
1646 fn test_row_values_addressable_by_str_key() {
1647 let mut row = QueryRow::new(RowKey::new(vec![1]));
1648 row.set("id", Value::Integer(7)); row.set("name".to_string(), Value::text("Zoe".to_string())); assert_eq!(row.get("id"), Some(&Value::Integer(7)));
1653 assert_eq!(row.get("name"), Some(&Value::text("Zoe".to_string())));
1654 assert_eq!(row.get("absent"), None);
1655
1656 let key: &Arc<str> = row.values.keys().next().expect("a key");
1658 let _: &str = key; }
1660
1661 #[test]
1665 fn test_query_result_serde_round_trip_preserves_names_and_values() {
1666 let mut row = QueryRow::new(RowKey::new(vec![1]));
1667 row.set("id", Value::Integer(42));
1668 row.set("name", Value::text("Alice".to_string()));
1669 let result = QueryResult::with_rows(vec![row]);
1670
1671 let json = serde_json::to_value(&result).expect("serialise QueryResult");
1677 let values_obj = json["rows"][0]["values"]
1678 .as_object()
1679 .expect("values is a JSON object keyed by column name");
1680 assert!(values_obj.contains_key("id"), "object keyed by column name");
1681 assert!(
1682 values_obj.contains_key("name"),
1683 "object keyed by column name"
1684 );
1685
1686 let back: QueryResult = serde_json::from_value(json).expect("deserialise QueryResult");
1688 let r = &back.rows[0];
1689 assert_eq!(r.get("id"), Some(&Value::Integer(42)));
1690 assert_eq!(r.get("name"), Some(&Value::text("Alice".to_string())));
1691 }
1692}