1use crate::{schema::CqlType, RowKey, Value};
7pub use crate::types::{CellExpiration, CellWriteMetadata};
10use base64::Engine;
11use serde::{Deserialize, Serialize};
12use serde_json::json;
13use std::collections::HashMap;
14use std::fmt;
15use tokio::sync::mpsc;
16
17fn b64(bytes: &[u8]) -> String {
19 base64::engine::general_purpose::STANDARD.encode(bytes)
20}
21
22fn row_metadata_is_populated(meta: &RowMetadata) -> bool {
24 meta.version.is_some() || meta.ttl.is_some() || !meta.tags.is_empty()
25}
26
27#[derive(Debug, Clone, Default)]
41pub struct ProjectionFlags {
42 pub include_cell_metadata: bool,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize)]
53pub struct QueryResult {
54 pub rows: Vec<QueryRow>,
56 pub rows_affected: u64,
58 pub execution_time_ms: u64,
60 pub metadata: QueryMetadata,
62}
63
64#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct QueryRow {
67 pub values: HashMap<String, Value>,
69 pub key: RowKey,
71 pub metadata: RowMetadata,
73 #[serde(skip_serializing_if = "Option::is_none", default)]
84 pub cell_metadata: Option<HashMap<String, CellWriteMetadata>>,
85}
86
87#[derive(Debug, Clone, Default, Serialize, Deserialize)]
89pub struct QueryMetadata {
90 pub columns: Vec<ColumnInfo>,
92 pub total_rows: Option<u64>,
94 pub plan_info: Option<PlanInfo>,
96 pub performance: PerformanceMetrics,
98 pub warnings: Vec<String>,
100}
101
102#[derive(Debug, Clone, Serialize, Deserialize)]
104pub struct ColumnInfo {
105 pub name: String,
107 pub data_type: crate::types::DataType,
109 pub nullable: bool,
111 pub position: usize,
113 pub table_name: Option<String>,
115 #[serde(skip_serializing_if = "Option::is_none")]
124 pub cql_type: Option<CqlType>,
125}
126
127#[derive(Debug, Clone, Default, Serialize, Deserialize)]
129pub struct RowMetadata {
130 pub version: Option<u64>,
132 pub ttl: Option<u64>,
134 pub tags: HashMap<String, String>,
136}
137
138#[derive(Debug, Clone, Serialize, Deserialize)]
140pub struct PlanInfo {
141 pub plan_type: String,
143 pub estimated_cost: f64,
145 pub actual_cost: f64,
147 pub indexes_used: Vec<String>,
149 pub steps: Vec<String>,
151 pub parallelization: Option<ParallelizationInfo>,
153}
154
155#[derive(Debug, Clone, Serialize, Deserialize)]
157pub struct ParallelizationInfo {
158 pub threads_used: usize,
160 pub effective: bool,
162 pub partitions: Vec<PartitionInfo>,
164}
165
166#[derive(Debug, Clone, Serialize, Deserialize)]
168pub struct PartitionInfo {
169 pub id: usize,
171 pub rows_processed: u64,
173 pub processing_time_ms: u64,
175}
176
177#[derive(Debug, Clone, Default, Serialize, Deserialize)]
179pub struct PerformanceMetrics {
180 pub parse_time_us: u64,
182 pub planning_time_us: u64,
184 pub execution_time_us: u64,
186 pub total_time_us: u64,
188 pub memory_usage_bytes: u64,
190 pub io_operations: u64,
192 pub cache_hits: u64,
194 pub cache_misses: u64,
196}
197
198#[derive(Debug, Clone)]
207pub struct StreamingConfig {
208 pub buffer_size: usize,
211 pub chunk_size: usize,
214}
215
216impl Default for StreamingConfig {
217 fn default() -> Self {
218 Self {
219 buffer_size: 1024, chunk_size: 10_000, }
222 }
223}
224
225impl StreamingConfig {
226 pub fn new(buffer_size: usize, chunk_size: usize) -> Self {
228 Self {
229 buffer_size,
230 chunk_size,
231 }
232 }
233
234 pub fn for_parquet() -> Self {
236 Self {
237 buffer_size: 1024,
238 chunk_size: 10_000, }
240 }
241
242 pub fn for_text_formats() -> Self {
244 Self {
245 buffer_size: 512,
246 chunk_size: 5_000, }
248 }
249}
250
251pub struct QueryResultIterator {
298 receiver: mpsc::Receiver<Result<QueryRow, crate::Error>>,
300 pub metadata: QueryMetadata,
302 pub total_rows_hint: Option<u64>,
304 rows_received: u64,
306}
307
308impl QueryResultIterator {
309 pub fn new(
311 receiver: mpsc::Receiver<Result<QueryRow, crate::Error>>,
312 metadata: QueryMetadata,
313 ) -> Self {
314 Self {
315 receiver,
316 metadata,
317 total_rows_hint: None,
318 rows_received: 0,
319 }
320 }
321
322 pub fn with_total_hint(mut self, total: u64) -> Self {
324 self.total_rows_hint = Some(total);
325 self
326 }
327
328 pub async fn next_async(&mut self) -> Option<Result<QueryRow, crate::Error>> {
332 let result = self.receiver.recv().await?;
333 if result.is_ok() {
334 self.rows_received += 1;
335 }
336 Some(result)
337 }
338
339 const MAX_CHUNK_SIZE: usize = 100_000;
341
342 pub async fn collect_chunk(&mut self, size: usize) -> Result<Vec<QueryRow>, crate::Error> {
357 let safe_size = size.min(Self::MAX_CHUNK_SIZE);
358 let mut chunk = Vec::new();
360 while chunk.len() < safe_size {
361 match self.receiver.recv().await {
362 Some(Ok(row)) => {
363 self.rows_received += 1;
364 chunk.push(row);
365 }
366 Some(Err(e)) => return Err(e),
367 None => break,
368 }
369 }
370 Ok(chunk)
371 }
372
373 pub fn rows_received(&self) -> u64 {
375 self.rows_received
376 }
377
378 pub fn progress_percent(&self) -> Option<f64> {
380 self.total_rows_hint.map(|total| {
381 if total == 0 {
382 100.0
383 } else {
384 (self.rows_received as f64 / total as f64) * 100.0
385 }
386 })
387 }
388}
389
390impl QueryResult {
391 pub fn new() -> Self {
393 Self {
394 rows: Vec::new(),
395 rows_affected: 0,
396 execution_time_ms: 0,
397 metadata: QueryMetadata::default(),
398 }
399 }
400
401 pub fn with_rows(rows: Vec<QueryRow>) -> Self {
403 Self {
404 rows,
405 ..Self::new()
406 }
407 }
408
409 pub fn with_affected_rows(rows_affected: u64) -> Self {
411 Self {
412 rows_affected,
413 ..Self::new()
414 }
415 }
416
417 pub fn row_count(&self) -> usize {
419 self.rows.len()
420 }
421
422 pub fn is_empty(&self) -> bool {
424 self.rows.is_empty()
425 }
426
427 pub fn get_row(&self, index: usize) -> Option<&QueryRow> {
429 self.rows.get(index)
430 }
431
432 pub fn columns(&self) -> &[ColumnInfo] {
434 &self.metadata.columns
435 }
436
437 pub fn column_names(&self) -> Vec<String> {
439 self.metadata
440 .columns
441 .iter()
442 .map(|c| c.name.clone())
443 .collect()
444 }
445
446 pub fn execution_time(&self) -> u64 {
448 self.execution_time_ms
449 }
450
451 pub fn performance(&self) -> &PerformanceMetrics {
453 &self.metadata.performance
454 }
455
456 pub fn warnings(&self) -> &[String] {
458 &self.metadata.warnings
459 }
460
461 pub fn add_warning(&mut self, warning: String) {
463 self.metadata.warnings.push(warning);
464 }
465
466 pub fn to_json(&self) -> serde_json::Value {
471 let rows: Vec<_> = self
472 .rows
473 .iter()
474 .map(|row| self.row_to_json_deterministic(row))
475 .collect();
476 let columns: Vec<_> = self
477 .metadata
478 .columns
479 .iter()
480 .map(ColumnInfo::to_json)
481 .collect();
482 let warnings: Vec<_> = self
483 .metadata
484 .warnings
485 .iter()
486 .cloned()
487 .map(serde_json::Value::String)
488 .collect();
489
490 json!({
491 "rows": rows,
492 "rows_affected": self.rows_affected,
493 "row_count": self.rows.len(),
494 "columns": columns,
495 "performance": self.metadata.performance.to_json(),
496 "warnings": warnings,
497 })
498 }
499
500 pub fn iter(&self) -> std::slice::Iter<'_, QueryRow> {
502 self.rows.iter()
503 }
504
505 fn row_to_json_deterministic(&self, row: &QueryRow) -> serde_json::Value {
510 let mut result = serde_json::Map::new();
511
512 if !self.metadata.columns.is_empty() {
513 for col in &self.metadata.columns {
514 let value_json = row
515 .values
516 .get(&col.name)
517 .map_or(serde_json::Value::Null, ToJson::to_json);
518 result.insert(col.name.clone(), value_json);
519 }
520 } else {
521 let mut sorted_keys: Vec<&String> = row.values.keys().collect();
522 sorted_keys.sort();
523 for key in sorted_keys {
524 if let Some(value) = row.values.get(key) {
525 result.insert(key.clone(), value.to_json());
526 }
527 }
528 }
529
530 result.insert(
531 "_key".to_string(),
532 serde_json::Value::String(format!("{:?}", row.key)),
533 );
534
535 if row_metadata_is_populated(&row.metadata) {
536 result.insert("_metadata".to_string(), row.metadata.to_json());
537 }
538
539 serde_json::Value::Object(result)
540 }
541}
542
543impl QueryRow {
544 pub fn new(key: RowKey) -> Self {
546 Self {
547 values: HashMap::new(),
548 key,
549 metadata: RowMetadata::default(),
550 cell_metadata: None,
551 }
552 }
553
554 pub fn with_values(key: RowKey, values: HashMap<String, Value>) -> Self {
556 Self {
557 values,
558 key,
559 metadata: RowMetadata::default(),
560 cell_metadata: None,
561 }
562 }
563
564 pub fn from_map(values: HashMap<String, Value>) -> Self {
569 Self {
570 values,
571 key: RowKey::new(vec![]),
572 metadata: RowMetadata::default(),
573 cell_metadata: None,
574 }
575 }
576
577 pub fn get(&self, column: &str) -> Option<&Value> {
579 self.values.get(column)
580 }
581
582 pub fn set(&mut self, column: String, value: Value) {
584 self.values.insert(column, value);
585 }
586
587 pub fn column_names(&self) -> Vec<String> {
589 self.values.keys().cloned().collect()
590 }
591
592 pub fn key(&self) -> &RowKey {
594 &self.key
595 }
596
597 pub fn metadata(&self) -> &RowMetadata {
599 &self.metadata
600 }
601
602 pub fn set_metadata(&mut self, metadata: RowMetadata) {
604 self.metadata = metadata;
605 }
606
607 pub fn set_cell_metadata(&mut self, map: HashMap<String, CellWriteMetadata>) {
614 self.cell_metadata = Some(map);
615 }
616
617 pub fn insert_cell_metadata(&mut self, column: String, meta: CellWriteMetadata) {
623 self.cell_metadata
624 .get_or_insert_with(HashMap::new)
625 .insert(column, meta);
626 }
627
628 pub fn get_cell_metadata(&self, column: &str) -> Option<&CellWriteMetadata> {
630 self.cell_metadata.as_ref()?.get(column)
631 }
632
633 pub fn to_json(&self) -> serde_json::Value {
635 let mut result = serde_json::Map::new();
636
637 for (column, value) in &self.values {
638 result.insert(column.clone(), value.to_json());
639 }
640
641 result.insert(
642 "_key".to_string(),
643 serde_json::Value::String(format!("{:?}", self.key)),
644 );
645
646 if row_metadata_is_populated(&self.metadata) {
647 result.insert("_metadata".to_string(), self.metadata.to_json());
648 }
649
650 serde_json::Value::Object(result)
651 }
652}
653
654impl ColumnInfo {
655 pub fn new(
657 name: String,
658 data_type: crate::types::DataType,
659 nullable: bool,
660 position: usize,
661 ) -> Self {
662 Self {
663 name,
664 data_type,
665 nullable,
666 position,
667 table_name: None,
668 cql_type: None,
669 }
670 }
671
672 pub fn with_table_name(mut self, table_name: String) -> Self {
674 self.table_name = Some(table_name);
675 self
676 }
677
678 pub fn with_cql_type(mut self, cql_type: CqlType) -> Self {
684 self.cql_type = Some(cql_type);
685 self
686 }
687
688 pub fn to_json(&self) -> serde_json::Value {
695 let mut map = serde_json::Map::new();
696 map.insert("name".to_string(), json!(self.name));
697 map.insert(
698 "data_type".to_string(),
699 json!(format!("{:?}", self.data_type)),
700 );
701 map.insert("nullable".to_string(), json!(self.nullable));
702 map.insert("position".to_string(), json!(self.position));
703 if let Some(table_name) = &self.table_name {
704 map.insert("table_name".to_string(), json!(table_name));
705 }
706 if let Some(cql_type) = &self.cql_type {
708 map.insert("cql_type".to_string(), json!(format_cql_type(cql_type)));
709 }
710 serde_json::Value::Object(map)
711 }
712}
713
714pub fn cql_type_to_data_type(cql_type: &CqlType) -> crate::types::DataType {
720 use crate::types::DataType;
721 match cql_type {
722 CqlType::Boolean => DataType::Boolean,
723 CqlType::TinyInt => DataType::TinyInt,
724 CqlType::SmallInt => DataType::SmallInt,
725 CqlType::Int => DataType::Integer,
726 CqlType::BigInt | CqlType::Varint | CqlType::Counter => DataType::BigInt,
727 CqlType::Float => DataType::Float32,
728 CqlType::Double | CqlType::Decimal => DataType::Float,
729 CqlType::Text | CqlType::Varchar | CqlType::Ascii => DataType::Text,
730 CqlType::Blob => DataType::Blob,
731 CqlType::Timestamp => DataType::Timestamp,
732 CqlType::Date | CqlType::Time | CqlType::Duration | CqlType::Inet => DataType::BigInt,
733 CqlType::Uuid | CqlType::TimeUuid => DataType::Uuid,
734 CqlType::List(_) => DataType::List,
735 CqlType::Set(_) => DataType::Set,
736 CqlType::Map(_, _) => DataType::Map,
737 CqlType::Tuple(_) => DataType::Tuple,
738 CqlType::Udt(_, _) => DataType::Udt,
739 CqlType::Frozen(inner) => cql_type_to_data_type(inner),
740 CqlType::Custom(_) => DataType::Blob,
741 }
742}
743
744fn format_cql_type(cql_type: &CqlType) -> String {
748 match cql_type {
749 CqlType::Boolean => "boolean".to_string(),
750 CqlType::TinyInt => "tinyint".to_string(),
751 CqlType::SmallInt => "smallint".to_string(),
752 CqlType::Int => "int".to_string(),
753 CqlType::BigInt => "bigint".to_string(),
754 CqlType::Counter => "counter".to_string(),
755 CqlType::Float => "float".to_string(),
756 CqlType::Double => "double".to_string(),
757 CqlType::Decimal => "decimal".to_string(),
758 CqlType::Text => "text".to_string(),
759 CqlType::Varchar => "varchar".to_string(),
760 CqlType::Ascii => "ascii".to_string(),
761 CqlType::Blob => "blob".to_string(),
762 CqlType::Timestamp => "timestamp".to_string(),
763 CqlType::Date => "date".to_string(),
764 CqlType::Time => "time".to_string(),
765 CqlType::Uuid => "uuid".to_string(),
766 CqlType::TimeUuid => "timeuuid".to_string(),
767 CqlType::Inet => "inet".to_string(),
768 CqlType::Duration => "duration".to_string(),
769 CqlType::Varint => "varint".to_string(),
770 CqlType::List(inner) => format!("list<{}>", format_cql_type(inner)),
771 CqlType::Set(inner) => format!("set<{}>", format_cql_type(inner)),
772 CqlType::Map(k, v) => format!("map<{}, {}>", format_cql_type(k), format_cql_type(v)),
773 CqlType::Tuple(types) => {
774 let inner: Vec<_> = types.iter().map(format_cql_type).collect();
775 format!("tuple<{}>", inner.join(", "))
776 }
777 CqlType::Udt(name, _) => name.clone(),
778 CqlType::Frozen(inner) => format!("frozen<{}>", format_cql_type(inner)),
779 CqlType::Custom(name) => name.clone(),
780 }
781}
782
783impl RowMetadata {
784 pub fn new() -> Self {
786 Self::default()
787 }
788
789 pub fn with_version(mut self, version: u64) -> Self {
791 self.version = Some(version);
792 self
793 }
794
795 pub fn with_ttl(mut self, ttl: u64) -> Self {
797 self.ttl = Some(ttl);
798 self
799 }
800
801 pub fn with_tag(mut self, key: String, value: String) -> Self {
803 self.tags.insert(key, value);
804 self
805 }
806
807 pub fn to_json(&self) -> serde_json::Value {
809 let mut map = serde_json::Map::new();
810 if let Some(version) = self.version {
811 map.insert("version".to_string(), json!(version));
812 }
813 if let Some(ttl) = self.ttl {
814 map.insert("ttl".to_string(), json!(ttl));
815 }
816 if !self.tags.is_empty() {
817 map.insert("tags".to_string(), json!(self.tags));
818 }
819 serde_json::Value::Object(map)
820 }
821}
822
823impl PerformanceMetrics {
824 pub fn new() -> Self {
826 Self::default()
827 }
828
829 pub fn total_time_ms(&self) -> u64 {
831 self.total_time_us / 1000
832 }
833
834 pub fn cache_hit_ratio(&self) -> f64 {
836 let total = self.cache_hits + self.cache_misses;
837 if total == 0 {
838 0.0
839 } else {
840 self.cache_hits as f64 / total as f64
841 }
842 }
843
844 pub fn to_json(&self) -> serde_json::Value {
846 let cache_hit_ratio = serde_json::Number::from_f64(self.cache_hit_ratio())
847 .map(serde_json::Value::Number)
848 .unwrap_or(json!(0));
849 json!({
850 "parse_time_us": self.parse_time_us,
851 "planning_time_us": self.planning_time_us,
852 "execution_time_us": self.execution_time_us,
853 "total_time_us": self.total_time_us,
854 "memory_usage_bytes": self.memory_usage_bytes,
855 "io_operations": self.io_operations,
856 "cache_hits": self.cache_hits,
857 "cache_misses": self.cache_misses,
858 "cache_hit_ratio": cache_hit_ratio,
859 })
860 }
861}
862
863fn write_border(
865 f: &mut fmt::Formatter<'_>,
866 widths: &[usize],
867 left: char,
868 sep: char,
869 right: char,
870) -> fmt::Result {
871 write!(f, "{}", left)?;
872 for (i, width) in widths.iter().enumerate() {
873 write!(f, "{}", "─".repeat(width + 2))?;
874 if i < widths.len() - 1 {
875 write!(f, "{}", sep)?;
876 }
877 }
878 writeln!(f, "{}", right)
879}
880
881impl fmt::Display for QueryResult {
882 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
883 if self.rows.is_empty() {
884 return write!(f, "Empty result set ({} rows affected)", self.rows_affected);
885 }
886
887 let column_names = self.column_names();
888 if column_names.is_empty() {
889 return write!(f, "No columns in result set");
890 }
891
892 let col_widths: Vec<usize> = column_names
894 .iter()
895 .map(|col_name| {
896 self.rows
897 .iter()
898 .filter_map(|row| row.values.get(col_name))
899 .map(|v| format!("{}", v).len())
900 .max()
901 .unwrap_or(0)
902 .max(col_name.len())
903 })
904 .collect();
905
906 write_border(f, &col_widths, '┌', '┬', '┐')?;
907
908 write!(f, "│")?;
909 for (i, (col_name, width)) in column_names.iter().zip(col_widths.iter()).enumerate() {
910 write!(f, " {:width$} ", col_name, width = width)?;
911 if i < column_names.len() - 1 {
912 write!(f, "│")?;
913 }
914 }
915 writeln!(f, "│")?;
916
917 write_border(f, &col_widths, '├', '┼', '┤')?;
918
919 for row in &self.rows {
920 write!(f, "│")?;
921 for (i, (col_name, width)) in column_names.iter().zip(col_widths.iter()).enumerate() {
922 let value = row
923 .values
924 .get(col_name)
925 .map(|v| format!("{}", v))
926 .unwrap_or_else(|| "NULL".to_string());
927 write!(f, " {:width$} ", value, width = width)?;
928 if i < column_names.len() - 1 {
929 write!(f, "│")?;
930 }
931 }
932 writeln!(f, "│")?;
933 }
934
935 write_border(f, &col_widths, '└', '┴', '┘')?;
936
937 writeln!(
938 f,
939 "{} rows returned in {}ms",
940 self.rows.len(),
941 self.execution_time_ms
942 )?;
943
944 if !self.metadata.warnings.is_empty() {
945 writeln!(f, "\nWarnings:")?;
946 for warning in &self.metadata.warnings {
947 writeln!(f, " - {}", warning)?;
948 }
949 }
950
951 Ok(())
952 }
953}
954
955impl Default for QueryResult {
956 fn default() -> Self {
957 Self::new()
958 }
959}
960
961impl IntoIterator for QueryResult {
962 type Item = QueryRow;
963 type IntoIter = std::vec::IntoIter<QueryRow>;
964
965 fn into_iter(self) -> Self::IntoIter {
966 self.rows.into_iter()
967 }
968}
969
970impl<'a> IntoIterator for &'a QueryResult {
971 type Item = &'a QueryRow;
972 type IntoIter = std::slice::Iter<'a, QueryRow>;
973
974 fn into_iter(self) -> Self::IntoIter {
975 self.rows.iter()
976 }
977}
978
979trait ToJson {
981 fn to_json(&self) -> serde_json::Value;
982}
983
984impl ToJson for Value {
985 fn to_json(&self) -> serde_json::Value {
986 fn float_to_json(x: f64) -> serde_json::Value {
988 serde_json::Number::from_f64(x)
989 .map(serde_json::Value::Number)
990 .unwrap_or(serde_json::Value::Null)
991 }
992
993 match self {
994 Value::Null => serde_json::Value::Null,
995 Value::Boolean(b) => json!(*b),
996 Value::Integer(i) => json!(*i),
997 Value::BigInt(i) => json!(*i),
998 Value::Counter(c) => json!(*c),
999 Value::TinyInt(i) => json!(*i as i64),
1000 Value::SmallInt(i) => json!(*i as i64),
1001 Value::Date(d) => json!(*d),
1002 Value::Time(t) => json!(*t),
1003 Value::Timestamp(ts) => json!(*ts),
1004 Value::Float(f) => float_to_json(*f),
1005 Value::Float32(f) => float_to_json(*f as f64),
1006 Value::Text(s) => json!(s),
1007 Value::Json(value) => value.clone(),
1008 Value::Blob(bytes) | Value::Varint(bytes) | Value::Inet(bytes) => json!(b64(bytes)),
1009 Value::Uuid(uuid) => json!(b64(uuid)),
1010 Value::List(items) | Value::Set(items) | Value::Tuple(items) => {
1011 let json_list: Vec<_> = items.iter().map(ToJson::to_json).collect();
1012 serde_json::Value::Array(json_list)
1013 }
1014 Value::Map(entries) => {
1015 let json_map: serde_json::Map<String, serde_json::Value> = entries
1016 .iter()
1017 .map(|(k, v)| (format!("{}", k), v.to_json()))
1018 .collect();
1019 serde_json::Value::Object(json_map)
1020 }
1021 Value::Udt(udt) => {
1022 let mut json_obj = serde_json::Map::new();
1023 json_obj.insert("_type".to_string(), json!(udt.type_name));
1024 for field in &udt.fields {
1025 let field_json = field
1026 .value
1027 .as_ref()
1028 .map_or(serde_json::Value::Null, ToJson::to_json);
1029 json_obj.insert(field.name.clone(), field_json);
1030 }
1031 serde_json::Value::Object(json_obj)
1032 }
1033 Value::Frozen(boxed) => boxed.to_json(),
1034 Value::Decimal { scale, unscaled } => json!({
1035 "scale": *scale,
1036 "unscaled": b64(unscaled),
1037 }),
1038 Value::Duration {
1039 months,
1040 days,
1041 nanos,
1042 } => json!({
1043 "months": *months,
1044 "days": *days,
1045 "nanos": *nanos,
1046 }),
1047 Value::Tombstone(info) => {
1048 let mut json_obj = serde_json::Map::new();
1049 json_obj.insert("type".to_string(), json!("tombstone"));
1050 json_obj.insert("deletion_time".to_string(), json!(info.deletion_time));
1051 json_obj.insert(
1052 "tombstone_type".to_string(),
1053 json!(format!("{:?}", info.tombstone_type)),
1054 );
1055 if let Some(ttl) = info.ttl {
1056 json_obj.insert("ttl".to_string(), json!(ttl));
1057 }
1058 serde_json::Value::Object(json_obj)
1059 }
1060 }
1061 }
1062}
1063
1064#[cfg(test)]
1065mod tests {
1066 use super::*;
1067 use crate::Value;
1068
1069 #[test]
1070 fn test_query_result_creation() {
1071 let result = QueryResult::new();
1072 assert!(result.is_empty());
1073 assert_eq!(result.row_count(), 0);
1074 assert_eq!(result.execution_time(), 0);
1075 }
1076
1077 #[test]
1078 fn test_query_result_with_rows() {
1079 let mut row1 = QueryRow::new(RowKey::new(vec![1]));
1080 row1.set("id".to_string(), Value::Integer(1));
1081 row1.set("name".to_string(), Value::Text("Alice".to_string()));
1082
1083 let mut row2 = QueryRow::new(RowKey::new(vec![2]));
1084 row2.set("id".to_string(), Value::Integer(2));
1085 row2.set("name".to_string(), Value::Text("Bob".to_string()));
1086
1087 let result = QueryResult::with_rows(vec![row1, row2]);
1088 assert_eq!(result.row_count(), 2);
1089 assert!(!result.is_empty());
1090
1091 let first_row = result.get_row(0).unwrap();
1092 assert_eq!(first_row.get("id"), Some(&Value::Integer(1)));
1093 assert_eq!(
1094 first_row.get("name"),
1095 Some(&Value::Text("Alice".to_string()))
1096 );
1097 }
1098
1099 #[test]
1100 fn test_query_row_operations() {
1101 let mut row = QueryRow::new(RowKey::new(vec![1]));
1102 row.set("id".to_string(), Value::Integer(42));
1103 row.set("active".to_string(), Value::Boolean(true));
1104
1105 assert_eq!(row.get("id"), Some(&Value::Integer(42)));
1106 assert_eq!(row.get("active"), Some(&Value::Boolean(true)));
1107 assert_eq!(row.get("nonexistent"), None);
1108
1109 let column_names = row.column_names();
1110 assert_eq!(column_names.len(), 2);
1111 assert!(column_names.contains(&"id".to_string()));
1112 assert!(column_names.contains(&"active".to_string()));
1113 }
1114
1115 #[test]
1116 fn test_column_info() {
1117 let column = ColumnInfo::new(
1118 "user_id".to_string(),
1119 crate::types::DataType::Integer,
1120 false,
1121 0,
1122 )
1123 .with_table_name("users".to_string());
1124
1125 assert_eq!(column.name, "user_id");
1126 assert_eq!(column.data_type, crate::types::DataType::Integer);
1127 assert!(!column.nullable);
1128 assert_eq!(column.position, 0);
1129 assert_eq!(column.table_name, Some("users".to_string()));
1130 assert!(column.cql_type.is_none());
1131 }
1132
1133 #[test]
1134 fn test_column_info_with_cql_type_scalar() {
1135 use crate::schema::CqlType;
1136 let column = ColumnInfo::new("ts".to_string(), crate::types::DataType::Timestamp, true, 1)
1137 .with_cql_type(CqlType::Timestamp);
1138
1139 assert_eq!(column.name, "ts");
1140 assert_eq!(column.cql_type, Some(CqlType::Timestamp));
1141 assert_eq!(column.data_type, crate::types::DataType::Timestamp);
1143 }
1144
1145 #[test]
1146 fn test_column_info_with_cql_type_list() {
1147 use crate::schema::CqlType;
1148 let list_type = CqlType::List(Box::new(CqlType::Int));
1149 let column = ColumnInfo::new("items".to_string(), crate::types::DataType::List, true, 2)
1150 .with_cql_type(list_type.clone());
1151
1152 assert_eq!(column.cql_type, Some(list_type));
1153 }
1154
1155 #[test]
1156 fn test_column_info_with_cql_type_map() {
1157 use crate::schema::CqlType;
1158 let map_type = CqlType::Map(Box::new(CqlType::Text), Box::new(CqlType::BigInt));
1159 let column = ColumnInfo::new("props".to_string(), crate::types::DataType::Map, true, 3)
1160 .with_cql_type(map_type.clone());
1161
1162 assert_eq!(column.cql_type, Some(map_type));
1163 }
1164
1165 #[test]
1166 fn test_column_info_with_cql_type_udt() {
1167 use crate::schema::CqlType;
1168 let udt_type = CqlType::Udt("address".to_string(), vec![]);
1169 let column = ColumnInfo::new("addr".to_string(), crate::types::DataType::Udt, true, 4)
1170 .with_cql_type(udt_type.clone());
1171
1172 assert_eq!(column.cql_type, Some(udt_type));
1173 }
1174
1175 #[test]
1176 fn test_cql_type_to_data_type_scalars() {
1177 use super::cql_type_to_data_type;
1178 use crate::schema::CqlType;
1179 use crate::types::DataType;
1180
1181 assert_eq!(cql_type_to_data_type(&CqlType::Boolean), DataType::Boolean);
1182 assert_eq!(cql_type_to_data_type(&CqlType::Int), DataType::Integer);
1183 assert_eq!(cql_type_to_data_type(&CqlType::BigInt), DataType::BigInt);
1184 assert_eq!(cql_type_to_data_type(&CqlType::Text), DataType::Text);
1185 assert_eq!(cql_type_to_data_type(&CqlType::Blob), DataType::Blob);
1186 assert_eq!(cql_type_to_data_type(&CqlType::Uuid), DataType::Uuid);
1187 assert_eq!(
1188 cql_type_to_data_type(&CqlType::Timestamp),
1189 DataType::Timestamp
1190 );
1191 }
1192
1193 #[test]
1194 fn test_cql_type_to_data_type_collections() {
1195 use super::cql_type_to_data_type;
1196 use crate::schema::CqlType;
1197 use crate::types::DataType;
1198
1199 assert_eq!(
1200 cql_type_to_data_type(&CqlType::List(Box::new(CqlType::Int))),
1201 DataType::List
1202 );
1203 assert_eq!(
1204 cql_type_to_data_type(&CqlType::Set(Box::new(CqlType::Text))),
1205 DataType::Set
1206 );
1207 assert_eq!(
1208 cql_type_to_data_type(&CqlType::Map(
1209 Box::new(CqlType::Text),
1210 Box::new(CqlType::BigInt)
1211 )),
1212 DataType::Map
1213 );
1214 }
1215
1216 #[test]
1217 fn test_cql_type_to_data_type_frozen() {
1218 use super::cql_type_to_data_type;
1219 use crate::schema::CqlType;
1220 use crate::types::DataType;
1221
1222 assert_eq!(
1224 cql_type_to_data_type(&CqlType::Frozen(Box::new(CqlType::List(Box::new(
1225 CqlType::Int
1226 ))))),
1227 DataType::List
1228 );
1229 }
1230
1231 #[test]
1232 fn test_column_info_to_json_includes_cql_type() {
1233 use crate::schema::CqlType;
1234 let column = ColumnInfo::new("items".to_string(), crate::types::DataType::List, true, 0)
1235 .with_cql_type(CqlType::List(Box::new(CqlType::Int)));
1236
1237 let json = column.to_json();
1238 let obj = json.as_object().unwrap();
1239
1240 assert_eq!(obj["name"], "items");
1242 assert!(obj.contains_key("data_type"));
1243 assert!(obj.contains_key("nullable"));
1244 assert!(obj.contains_key("position"));
1245
1246 assert!(obj.contains_key("cql_type"));
1248 assert_eq!(obj["cql_type"], "list<int>");
1249 }
1250
1251 #[test]
1252 fn test_column_info_to_json_no_cql_type() {
1253 let column = ColumnInfo::new("id".to_string(), crate::types::DataType::Integer, false, 0);
1255
1256 let json = column.to_json();
1257 let obj = json.as_object().unwrap();
1258
1259 assert!(!obj.contains_key("cql_type"));
1260 }
1261
1262 #[test]
1263 fn test_row_metadata() {
1264 let metadata = RowMetadata::new()
1265 .with_version(123)
1266 .with_ttl(3600)
1267 .with_tag("source".to_string(), "import".to_string());
1268
1269 assert_eq!(metadata.version, Some(123));
1270 assert_eq!(metadata.ttl, Some(3600));
1271 assert_eq!(metadata.tags.get("source"), Some(&"import".to_string()));
1272 }
1273
1274 #[test]
1275 fn test_performance_metrics() {
1276 let mut metrics = PerformanceMetrics::new();
1277 metrics.cache_hits = 8;
1278 metrics.cache_misses = 2;
1279 metrics.total_time_us = 5000;
1280
1281 assert_eq!(metrics.cache_hit_ratio(), 0.8);
1282 assert_eq!(metrics.total_time_ms(), 5);
1283 }
1284
1285 #[test]
1286 fn test_json_serialization() {
1287 let mut row = QueryRow::new(RowKey::new(vec![1]));
1288 row.set("id".to_string(), Value::Integer(1));
1289 row.set("name".to_string(), Value::Text("test".to_string()));
1290
1291 let json = row.to_json();
1292 assert!(json.is_object());
1293
1294 let obj = json.as_object().unwrap();
1295 assert_eq!(obj.get("id"), Some(&serde_json::Value::Number(1.into())));
1296 assert_eq!(
1297 obj.get("name"),
1298 Some(&serde_json::Value::String("test".to_string()))
1299 );
1300 }
1301
1302 #[test]
1303 fn test_result_iteration() {
1304 let row1 = QueryRow::new(RowKey::new(vec![1]));
1305 let row2 = QueryRow::new(RowKey::new(vec![2]));
1306 let result = QueryResult::with_rows(vec![row1, row2]);
1307
1308 let mut count = 0;
1309 for _row in &result {
1310 count += 1;
1311 }
1312 assert_eq!(count, 2);
1313
1314 let mut count = 0;
1315 for _row in result {
1316 count += 1;
1317 }
1318 assert_eq!(count, 2);
1319 }
1320
1321 #[test]
1328 fn test_cell_metadata_absent_by_default() {
1329 let row = QueryRow::new(RowKey::new(vec![1]));
1330 assert!(
1331 row.cell_metadata.is_none(),
1332 "cell_metadata must be None when no metadata is attached (hot-path, zero allocation)"
1333 );
1334
1335 let row2 = QueryRow::with_values(RowKey::new(vec![2]), HashMap::new());
1336 assert!(row2.cell_metadata.is_none());
1337
1338 let row3 = QueryRow::from_map(HashMap::new());
1339 assert!(row3.cell_metadata.is_none());
1340 }
1341
1342 #[test]
1344 fn test_projection_flags_default_no_metadata() {
1345 let flags = ProjectionFlags::default();
1346 assert!(
1347 !flags.include_cell_metadata,
1348 "include_cell_metadata must default to false"
1349 );
1350 }
1351
1352 #[test]
1354 fn test_cell_metadata_single_sstable_single_cell() {
1355 let mut row = QueryRow::new(RowKey::new(vec![1]));
1356 row.set("name".to_string(), Value::Text("Alice".to_string()));
1357
1358 let meta = CellWriteMetadata {
1359 write_timestamp_micros: 1_700_000_000_000_000, expiration: None,
1361 };
1362 row.insert_cell_metadata("name".to_string(), meta.clone());
1363
1364 assert!(row.cell_metadata.is_some());
1366 assert_eq!(row.get("name"), Some(&Value::Text("Alice".to_string())));
1368 let got = row
1370 .get_cell_metadata("name")
1371 .expect("metadata must be present");
1372 assert_eq!(got.write_timestamp_micros, meta.write_timestamp_micros);
1373 assert!(got.expiration.is_none());
1374 }
1375
1376 #[test]
1378 fn test_cell_metadata_with_ttl_expiration() {
1379 let mut row = QueryRow::new(RowKey::new(vec![2]));
1380 row.set("score".to_string(), Value::Integer(42));
1381
1382 let ttl_seconds = 3600_i32;
1383 let write_ts_micros = 1_700_000_000_000_000_i64;
1384 let expires_at = (write_ts_micros / 1_000_000) + ttl_seconds as i64;
1386
1387 let meta = CellWriteMetadata {
1388 write_timestamp_micros: write_ts_micros,
1389 expiration: Some(CellExpiration {
1390 ttl_seconds,
1391 expires_at_seconds: expires_at,
1392 }),
1393 };
1394 row.insert_cell_metadata("score".to_string(), meta);
1395
1396 let got = row.get_cell_metadata("score").unwrap();
1397 assert_eq!(got.write_timestamp_micros, write_ts_micros);
1398 let exp = got.expiration.as_ref().unwrap();
1399 assert_eq!(exp.ttl_seconds, 3600);
1400 assert_eq!(exp.expires_at_seconds, expires_at);
1401 }
1402
1403 #[test]
1406 fn test_cell_metadata_absent_for_null_cells() {
1407 let mut row = QueryRow::new(RowKey::new(vec![3]));
1408 row.set("id".to_string(), Value::Null);
1409 row.insert_cell_metadata(
1412 "name".to_string(),
1413 CellWriteMetadata {
1414 write_timestamp_micros: 42,
1415 expiration: None,
1416 },
1417 );
1418
1419 assert!(
1420 row.get_cell_metadata("id").is_none(),
1421 "no metadata for null column"
1422 );
1423 assert!(row.get_cell_metadata("name").is_some());
1424 }
1425
1426 #[test]
1436 fn test_cell_metadata_lww_winner_carries_newer_timestamp() {
1437 let mut older_row = QueryRow::new(RowKey::new(b"partition1".to_vec()));
1439 older_row.set("value".to_string(), Value::Integer(10));
1440 older_row.insert_cell_metadata(
1441 "value".to_string(),
1442 CellWriteMetadata {
1443 write_timestamp_micros: 1_000_000,
1444 expiration: None,
1445 },
1446 );
1447
1448 let mut newer_row = QueryRow::new(RowKey::new(b"partition1".to_vec()));
1450 newer_row.set("value".to_string(), Value::Integer(20));
1451 newer_row.insert_cell_metadata(
1452 "value".to_string(),
1453 CellWriteMetadata {
1454 write_timestamp_micros: 2_000_000,
1455 expiration: None,
1456 },
1457 );
1458
1459 let winner = if newer_row
1461 .get_cell_metadata("value")
1462 .map(|m| m.write_timestamp_micros)
1463 .unwrap_or(0)
1464 > older_row
1465 .get_cell_metadata("value")
1466 .map(|m| m.write_timestamp_micros)
1467 .unwrap_or(0)
1468 {
1469 newer_row
1470 } else {
1471 older_row
1472 };
1473
1474 assert_eq!(
1475 winner.get("value"),
1476 Some(&Value::Integer(20)),
1477 "value from the newer SSTable must be present"
1478 );
1479 assert_eq!(
1480 winner
1481 .get_cell_metadata("value")
1482 .map(|m| m.write_timestamp_micros),
1483 Some(2_000_000),
1484 "metadata must reflect the winning (newer) cell's timestamp"
1485 );
1486 }
1487
1488 #[test]
1490 fn test_set_cell_metadata_replaces_map() {
1491 let mut row = QueryRow::new(RowKey::new(vec![4]));
1492 row.insert_cell_metadata(
1493 "a".to_string(),
1494 CellWriteMetadata {
1495 write_timestamp_micros: 1,
1496 expiration: None,
1497 },
1498 );
1499
1500 let mut new_map = HashMap::new();
1501 new_map.insert(
1502 "b".to_string(),
1503 CellWriteMetadata {
1504 write_timestamp_micros: 99,
1505 expiration: None,
1506 },
1507 );
1508 row.set_cell_metadata(new_map);
1509
1510 assert!(row.get_cell_metadata("a").is_none());
1512 assert_eq!(
1513 row.get_cell_metadata("b").map(|m| m.write_timestamp_micros),
1514 Some(99)
1515 );
1516 }
1517
1518 #[test]
1520 fn test_cell_metadata_serde_round_trip() {
1521 let mut row = QueryRow::new(RowKey::new(vec![5]));
1522 row.set("x".to_string(), Value::Integer(7));
1523 row.insert_cell_metadata(
1524 "x".to_string(),
1525 CellWriteMetadata {
1526 write_timestamp_micros: 123_456_789,
1527 expiration: Some(CellExpiration {
1528 ttl_seconds: 60,
1529 expires_at_seconds: 9999,
1530 }),
1531 },
1532 );
1533
1534 let json = serde_json::to_string(&row).expect("serialise");
1535 let back: QueryRow = serde_json::from_str(&json).expect("deserialise");
1536
1537 let meta = back
1538 .get_cell_metadata("x")
1539 .expect("metadata present after round-trip");
1540 assert_eq!(meta.write_timestamp_micros, 123_456_789);
1541 let exp = meta.expiration.as_ref().unwrap();
1542 assert_eq!(exp.ttl_seconds, 60);
1543 assert_eq!(exp.expires_at_seconds, 9999);
1544 }
1545
1546 #[test]
1549 fn test_cell_metadata_none_omitted_from_json() {
1550 let row = QueryRow::new(RowKey::new(vec![6]));
1551 let json = serde_json::to_string(&row).expect("serialise");
1552 assert!(
1553 !json.contains("cell_metadata"),
1554 "cell_metadata must be absent from JSON when None (backward compat)"
1555 );
1556 }
1557}