1use crate::query::{ColumnInfo, QueryRow};
34use crate::schema::CqlType;
35use crate::types::{DataType, Value};
36use crate::util::value_fmt::ValueFormatter;
37use arrow::array::{
38 ArrayRef, BinaryArray, BooleanArray, Date32Array, Float32Array, Float64Array, Int16Array,
39 Int32Array, Int64Array, Int8Array, ListArray, MapArray, StringArray, StructArray,
40 Time64NanosecondArray, TimestampMillisecondArray,
41};
42use arrow::buffer::{NullBuffer, OffsetBuffer};
43use arrow::datatypes::{DataType as ArrowDataType, Field, Fields, Schema, TimeUnit};
44use arrow::record_batch::RecordBatch;
45use std::collections::HashMap;
46use std::sync::Arc;
47use thiserror::Error;
48
49pub(crate) const DECIMAL_FIXED_SCALE: i32 = 9;
59
60pub(crate) const DECIMAL_MAX_PRECISION: u8 = 38;
62
63pub(crate) const ARROW_EXTENSION_NAME_KEY: &str = "ARROW:extension:name";
67pub(crate) const ARROW_UUID_EXTENSION_NAME: &str = "arrow.uuid";
68
69#[derive(Debug, Error)]
79pub enum ArrowConvertError {
80 #[error("Arrow error: {0}")]
82 Arrow(#[from] arrow::error::ArrowError),
83 #[error("{0}")]
85 InvalidValue(String),
86}
87
88pub fn build_arrow_schema(columns: &[ColumnInfo]) -> Result<Schema, ArrowConvertError> {
100 let fields: Vec<Field> = columns.iter().map(column_to_field).collect();
101 Ok(Schema::new(fields))
102}
103
104pub fn rows_to_record_batch(
114 columns: &[ColumnInfo],
115 rows: &[QueryRow],
116) -> Result<RecordBatch, ArrowConvertError> {
117 let schema = build_arrow_schema(columns)?;
118 let arrays = convert_to_arrays(columns, rows)?;
119 let batch = RecordBatch::try_new(Arc::new(schema), arrays)?;
120 Ok(batch)
121}
122
123pub(crate) fn bigint_to_i128(n: &num_bigint::BigInt) -> Result<i128, ArrowConvertError> {
134 let tc_bytes = n.to_signed_bytes_be();
135 if tc_bytes.len() > 16 {
136 return Err(ArrowConvertError::InvalidValue(
137 "BigInt value requires more than 16 bytes; cannot fit in i128".to_string(),
138 ));
139 }
140 let pad: u8 = if n.sign() == num_bigint::Sign::Minus {
142 0xFF
143 } else {
144 0x00
145 };
146 let mut buf = [pad; 16];
147 buf[16 - tc_bytes.len()..].copy_from_slice(&tc_bytes);
149 Ok(i128::from_be_bytes(buf))
150}
151
152pub(crate) fn column_to_field(col: &ColumnInfo) -> Field {
162 if let Some(cql_type) = &col.cql_type {
163 if let Some(field) = cql_type_to_arrow_field(&col.name, cql_type, col.nullable) {
164 return field;
165 }
166 }
167 let arrow_type = data_type_to_arrow(&col.data_type);
168 Field::new(&col.name, arrow_type, col.nullable)
169}
170
171pub(crate) fn cql_type_to_arrow_field(
180 name: &str,
181 cql_type: &CqlType,
182 nullable: bool,
183) -> Option<Field> {
184 match cql_type {
185 CqlType::Date => Some(Field::new(name, ArrowDataType::Date32, nullable)),
186 CqlType::Time => Some(Field::new(
187 name,
188 ArrowDataType::Time64(TimeUnit::Nanosecond),
189 nullable,
190 )),
191 CqlType::Decimal => Some(Field::new(
192 name,
193 ArrowDataType::Decimal128(DECIMAL_MAX_PRECISION, DECIMAL_FIXED_SCALE as i8),
194 nullable,
195 )),
196 CqlType::Varint => {
197 Some(Field::new(
201 name,
202 ArrowDataType::Decimal128(DECIMAL_MAX_PRECISION, 0),
203 nullable,
204 ))
205 }
206 CqlType::Duration => {
207 Some(Field::new(name, ArrowDataType::Utf8, nullable))
219 }
220 CqlType::Uuid | CqlType::TimeUuid => {
221 let mut meta = HashMap::new();
224 meta.insert(
225 ARROW_EXTENSION_NAME_KEY.to_string(),
226 ARROW_UUID_EXTENSION_NAME.to_string(),
227 );
228 Some(Field::new(name, ArrowDataType::FixedSizeBinary(16), nullable).with_metadata(meta))
229 }
230 CqlType::Inet => Some(Field::new(name, ArrowDataType::Utf8, nullable)),
231 CqlType::Counter => Some(Field::new(name, ArrowDataType::Int64, nullable)),
232 CqlType::List(inner) | CqlType::Set(inner) => {
235 let item_type = cql_type_to_arrow_data_type(inner);
236 let item_field = Arc::new(Field::new("item", item_type, true));
237 Some(Field::new(name, ArrowDataType::List(item_field), nullable))
238 }
239 CqlType::Frozen(inner) => cql_type_to_arrow_field(name, inner, nullable),
241 CqlType::Map(key_type, val_type) => {
245 let key_arrow = cql_type_to_arrow_data_type(key_type);
246 let val_arrow = cql_type_to_arrow_data_type(val_type);
247 let entries_field = Arc::new(Field::new(
248 "entries",
249 ArrowDataType::Struct(Fields::from(vec![
250 Field::new("key", key_arrow, false),
251 Field::new("value", val_arrow, true),
252 ])),
253 false,
254 ));
255 Some(Field::new(
256 name,
257 ArrowDataType::Map(entries_field, false),
258 nullable,
259 ))
260 }
261 CqlType::Tuple(element_types) => {
264 if element_types.is_empty() {
265 return Some(Field::new(name, ArrowDataType::Utf8, nullable));
266 }
267 let struct_type = cql_type_to_arrow_data_type(cql_type);
268 Some(Field::new(name, struct_type, nullable))
269 }
270 CqlType::Udt(_udt_name, udt_fields) => {
273 if udt_fields.is_empty() {
274 return Some(Field::new(name, ArrowDataType::Utf8, nullable));
275 }
276 let struct_type = cql_type_to_arrow_data_type(cql_type);
277 Some(Field::new(name, struct_type, nullable))
278 }
279 _ => None,
282 }
283}
284
285pub(crate) fn cql_type_to_arrow_data_type(cql_type: &CqlType) -> ArrowDataType {
302 match cql_type {
303 CqlType::Boolean => ArrowDataType::Boolean,
305 CqlType::TinyInt => ArrowDataType::Int8,
306 CqlType::SmallInt => ArrowDataType::Int16,
307 CqlType::Int => ArrowDataType::Int32,
308 CqlType::BigInt => ArrowDataType::Int64,
309 CqlType::Counter => ArrowDataType::Int64,
310 CqlType::Float => ArrowDataType::Float32,
311 CqlType::Double => ArrowDataType::Float64,
312 CqlType::Text | CqlType::Ascii | CqlType::Varchar => ArrowDataType::Utf8,
313 CqlType::Blob => ArrowDataType::Binary,
314 CqlType::Timestamp => ArrowDataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
315 CqlType::Date => ArrowDataType::Date32,
316 CqlType::Time => ArrowDataType::Time64(TimeUnit::Nanosecond),
317 CqlType::Decimal => {
318 ArrowDataType::Decimal128(DECIMAL_MAX_PRECISION, DECIMAL_FIXED_SCALE as i8)
319 }
320 CqlType::Varint => ArrowDataType::Decimal128(DECIMAL_MAX_PRECISION, 0),
321 CqlType::Duration => ArrowDataType::Utf8,
323 CqlType::Uuid | CqlType::TimeUuid => ArrowDataType::FixedSizeBinary(16),
324 CqlType::Inet => ArrowDataType::Utf8,
326 CqlType::List(inner) | CqlType::Set(inner) => {
329 let item_type = cql_type_to_arrow_data_type(inner);
330 ArrowDataType::List(Arc::new(Field::new("item", item_type, true)))
331 }
332 CqlType::Frozen(inner) => cql_type_to_arrow_data_type(inner),
334 CqlType::Map(key_type, val_type) => {
338 let key_arrow = cql_type_to_arrow_data_type(key_type);
339 let val_arrow = cql_type_to_arrow_data_type(val_type);
340 ArrowDataType::Map(
341 Arc::new(Field::new(
342 "entries",
343 ArrowDataType::Struct(Fields::from(vec![
344 Field::new("key", key_arrow, false),
345 Field::new("value", val_arrow, true),
346 ])),
347 false,
348 )),
349 false,
350 )
351 }
352 CqlType::Tuple(element_types) => {
355 if element_types.is_empty() {
356 return ArrowDataType::Utf8;
357 }
358 let struct_fields: Vec<Field> = element_types
359 .iter()
360 .enumerate()
361 .map(|(i, t)| {
362 Field::new(
363 format!("field_{i}"),
364 cql_type_to_arrow_data_type(t),
365 true, )
367 })
368 .collect();
369 ArrowDataType::Struct(Fields::from(struct_fields))
370 }
371 CqlType::Udt(_udt_name, udt_fields) => {
374 if udt_fields.is_empty() {
375 return ArrowDataType::Utf8;
376 }
377 let struct_fields: Vec<Field> = udt_fields
378 .iter()
379 .map(|(field_name, field_type)| {
380 Field::new(
381 field_name.as_str(),
382 cql_type_to_arrow_data_type(field_type),
383 true, )
385 })
386 .collect();
387 ArrowDataType::Struct(Fields::from(struct_fields))
388 }
389 CqlType::Custom(_) => ArrowDataType::Utf8,
391 }
392}
393
394pub(crate) fn build_typed_value_array(
407 cql_type: &CqlType,
408 values: &[Option<&Value>],
409) -> Result<ArrayRef, ArrowConvertError> {
410 let effective_type = unwrap_frozen_type(cql_type);
412
413 match effective_type {
414 CqlType::Boolean => {
418 let arr: Vec<Option<bool>> = values
419 .iter()
420 .filter_map(|opt| {
421 let v = unwrap_frozen_value(*opt)?;
422 Some(match v {
423 Value::Boolean(b) => Ok(Some(*b)),
424 Value::Null => Ok(None),
425 other => Err(ArrowConvertError::InvalidValue(format!(
426 "expected Boolean value in element, got {:?}",
427 other
428 ))),
429 })
430 })
431 .collect::<Result<Vec<Option<bool>>, ArrowConvertError>>()?;
432 Ok(Arc::new(BooleanArray::from(arr)))
433 }
434 CqlType::TinyInt => {
435 let arr: Vec<Option<i8>> = values
436 .iter()
437 .filter_map(|opt| {
438 let v = unwrap_frozen_value(*opt)?;
439 Some(match v {
440 Value::TinyInt(i) => Ok(Some(*i)),
441 Value::Null => Ok(None),
442 other => Err(ArrowConvertError::InvalidValue(format!(
443 "expected TinyInt value in element, got {:?}",
444 other
445 ))),
446 })
447 })
448 .collect::<Result<Vec<Option<i8>>, ArrowConvertError>>()?;
449 Ok(Arc::new(Int8Array::from(arr)))
450 }
451 CqlType::SmallInt => {
452 let arr: Vec<Option<i16>> = values
453 .iter()
454 .filter_map(|opt| {
455 let v = unwrap_frozen_value(*opt)?;
456 Some(match v {
457 Value::SmallInt(i) => Ok(Some(*i)),
458 Value::Null => Ok(None),
459 other => Err(ArrowConvertError::InvalidValue(format!(
460 "expected SmallInt value in element, got {:?}",
461 other
462 ))),
463 })
464 })
465 .collect::<Result<Vec<Option<i16>>, ArrowConvertError>>()?;
466 Ok(Arc::new(Int16Array::from(arr)))
467 }
468 CqlType::Int => {
469 let arr: Vec<Option<i32>> = values
470 .iter()
471 .filter_map(|opt| {
472 let v = unwrap_frozen_value(*opt)?;
473 Some(match v {
474 Value::Integer(i) => Ok(Some(*i)),
475 Value::Null => Ok(None),
476 other => Err(ArrowConvertError::InvalidValue(format!(
477 "expected Int value in element, got {:?}",
478 other
479 ))),
480 })
481 })
482 .collect::<Result<Vec<Option<i32>>, ArrowConvertError>>()?;
483 Ok(Arc::new(Int32Array::from(arr)))
484 }
485 CqlType::BigInt => {
486 let arr: Vec<Option<i64>> = values
487 .iter()
488 .filter_map(|opt| {
489 let v = unwrap_frozen_value(*opt)?;
490 Some(match v {
491 Value::BigInt(i) => Ok(Some(*i)),
492 Value::Null => Ok(None),
493 other => Err(ArrowConvertError::InvalidValue(format!(
494 "expected BigInt value in element, got {:?}",
495 other
496 ))),
497 })
498 })
499 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
500 Ok(Arc::new(Int64Array::from(arr)))
501 }
502 CqlType::Counter => {
503 let arr: Vec<Option<i64>> = values
504 .iter()
505 .filter_map(|opt| {
506 let v = unwrap_frozen_value(*opt)?;
507 Some(match v {
508 Value::Counter(c) => Ok(Some(*c)),
509 Value::BigInt(i) => Ok(Some(*i)),
510 Value::Null => Ok(None),
511 other => Err(ArrowConvertError::InvalidValue(format!(
512 "expected Counter value in element, got {:?}",
513 other
514 ))),
515 })
516 })
517 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
518 Ok(Arc::new(Int64Array::from(arr)))
519 }
520 CqlType::Float => {
521 let arr: Vec<Option<f32>> = values
522 .iter()
523 .filter_map(|opt| {
524 let v = unwrap_frozen_value(*opt)?;
525 Some(match v {
526 Value::Float32(f) => Ok(Some(*f)),
527 Value::Float(f) => Ok(Some(*f as f32)),
532 Value::Null => Ok(None),
533 other => Err(ArrowConvertError::InvalidValue(format!(
534 "expected Float value in element, got {:?}",
535 other
536 ))),
537 })
538 })
539 .collect::<Result<Vec<Option<f32>>, ArrowConvertError>>()?;
540 Ok(Arc::new(Float32Array::from(arr)))
541 }
542 CqlType::Double => {
543 let arr: Vec<Option<f64>> = values
544 .iter()
545 .filter_map(|opt| {
546 let v = unwrap_frozen_value(*opt)?;
547 Some(match v {
548 Value::Float(f) => Ok(Some(*f)),
549 Value::Float32(f) => Ok(Some(*f as f64)),
550 Value::Null => Ok(None),
551 other => Err(ArrowConvertError::InvalidValue(format!(
552 "expected Double value in element, got {:?}",
553 other
554 ))),
555 })
556 })
557 .collect::<Result<Vec<Option<f64>>, ArrowConvertError>>()?;
558 Ok(Arc::new(Float64Array::from(arr)))
559 }
560 CqlType::Text | CqlType::Ascii | CqlType::Varchar => {
561 let arr: Vec<Option<String>> = values
562 .iter()
563 .filter_map(|opt| {
564 let v = unwrap_frozen_value(*opt)?;
565 Some(match v {
566 Value::Text(s) => Ok(Some(s.clone())),
567 Value::Null => Ok(None),
568 other => Err(ArrowConvertError::InvalidValue(format!(
569 "expected Text value in element, got {:?}",
570 other
571 ))),
572 })
573 })
574 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
575 Ok(Arc::new(StringArray::from(arr)))
576 }
577 CqlType::Blob => {
578 let byte_slices: Vec<Option<Vec<u8>>> = values
579 .iter()
580 .filter_map(|opt| {
581 let v = unwrap_frozen_value(*opt)?;
582 Some(match v {
583 Value::Blob(b) => Ok(Some(b.clone())),
584 Value::Null => Ok(None),
585 other => Err(ArrowConvertError::InvalidValue(format!(
586 "expected Blob value in element, got {:?}",
587 other
588 ))),
589 })
590 })
591 .collect::<Result<Vec<Option<Vec<u8>>>, ArrowConvertError>>()?;
592 let refs: Vec<Option<&[u8]>> = byte_slices.iter().map(|o| o.as_deref()).collect();
593 Ok(Arc::new(BinaryArray::from(refs)))
594 }
595 CqlType::Timestamp => {
596 let arr: Vec<Option<i64>> = values
597 .iter()
598 .filter_map(|opt| {
599 let v = unwrap_frozen_value(*opt)?;
600 Some(match v {
601 Value::Timestamp(ts) => Ok(Some(*ts)),
602 Value::Null => Ok(None),
603 other => Err(ArrowConvertError::InvalidValue(format!(
604 "expected Timestamp value in element, got {:?}",
605 other
606 ))),
607 })
608 })
609 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
610 Ok(Arc::new(
611 TimestampMillisecondArray::from(arr).with_timezone("UTC"),
612 ))
613 }
614 CqlType::Date => {
615 let arr: Vec<Option<i32>> = values
616 .iter()
617 .filter_map(|opt| {
618 let v = unwrap_frozen_value(*opt)?;
619 Some(match v {
620 Value::Date(d) => Ok(Some(*d)),
621 Value::Null => Ok(None),
622 other => Err(ArrowConvertError::InvalidValue(format!(
623 "expected Date value in element, got {:?}",
624 other
625 ))),
626 })
627 })
628 .collect::<Result<Vec<Option<i32>>, ArrowConvertError>>()?;
629 Ok(Arc::new(Date32Array::from(arr)))
630 }
631 CqlType::Time => {
632 let arr: Vec<Option<i64>> = values
633 .iter()
634 .filter_map(|opt| {
635 let v = unwrap_frozen_value(*opt)?;
636 Some(match v {
637 Value::Time(t) => Ok(Some(*t)),
638 Value::Null => Ok(None),
639 other => Err(ArrowConvertError::InvalidValue(format!(
640 "expected Time value in element, got {:?}",
641 other
642 ))),
643 })
644 })
645 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
646 Ok(Arc::new(Time64NanosecondArray::from(arr)))
647 }
648 CqlType::Decimal => {
649 let mut builder = arrow::array::Decimal128Builder::new()
650 .with_precision_and_scale(DECIMAL_MAX_PRECISION, DECIMAL_FIXED_SCALE as i8)?;
651 for opt in values {
652 let v = unwrap_frozen_value(*opt);
653 match v {
654 Some(Value::Decimal { scale, unscaled }) => {
655 let rescaled = rescale_decimal(*scale, unscaled)?;
656 builder.append_value(rescaled);
657 }
658 Some(Value::Null) | None => builder.append_null(),
659 Some(other) => {
660 return Err(ArrowConvertError::InvalidValue(format!(
661 "expected Decimal value in element, got {:?}",
662 other
663 )));
664 }
665 }
666 }
667 Ok(Arc::new(builder.finish()))
668 }
669 CqlType::Varint => {
670 use num_bigint::BigInt;
671 let mut builder = arrow::array::Decimal128Builder::new()
672 .with_precision_and_scale(DECIMAL_MAX_PRECISION, 0)?;
673 for opt in values {
674 let v = unwrap_frozen_value(*opt);
675 match v {
676 Some(Value::Varint(bytes)) => {
677 if bytes.is_empty() {
678 builder.append_value(0);
679 } else {
680 let bigint = BigInt::from_signed_bytes_be(bytes);
681 let max_abs = BigInt::from(10i64).pow(38u32) - BigInt::from(1i64);
682 let abs_val = if bigint.sign() == num_bigint::Sign::Minus {
683 -bigint.clone()
684 } else {
685 bigint.clone()
686 };
687 if abs_val > max_abs {
688 return Err(ArrowConvertError::InvalidValue(
689 "varint element exceeds Decimal128(38, 0) range".to_string(),
690 ));
691 }
692 let i128_val = bigint_to_i128(&bigint)?;
693 builder.append_value(i128_val);
694 }
695 }
696 Some(Value::Null) | None => builder.append_null(),
697 Some(other) => {
698 return Err(ArrowConvertError::InvalidValue(format!(
699 "expected Varint value in element, got {:?}",
700 other
701 )));
702 }
703 }
704 }
705 Ok(Arc::new(builder.finish()))
706 }
707 CqlType::Duration => {
708 let arr: Vec<Option<String>> = values
710 .iter()
711 .filter_map(|opt| {
712 let v = unwrap_frozen_value(*opt)?;
713 Some(match v {
714 Value::Duration { .. } => Ok(Some(ValueFormatter::format_value(v))),
715 Value::Null => Ok(None),
716 other => Err(ArrowConvertError::InvalidValue(format!(
717 "expected Duration value in element, got {:?}",
718 other
719 ))),
720 })
721 })
722 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
723 Ok(Arc::new(StringArray::from(arr)))
724 }
725 CqlType::Uuid | CqlType::TimeUuid => {
726 let mut builder = arrow::array::FixedSizeBinaryBuilder::new(16);
727 for opt in values {
728 let v = unwrap_frozen_value(*opt);
729 match v {
730 Some(Value::Uuid(bytes)) => builder.append_value(bytes)?,
731 Some(Value::Null) | None => builder.append_null(),
732 Some(other) => {
733 return Err(ArrowConvertError::InvalidValue(format!(
734 "expected Uuid value in element, got {:?}",
735 other
736 )));
737 }
738 }
739 }
740 Ok(Arc::new(builder.finish()))
741 }
742 CqlType::Inet => {
743 let arr: Vec<Option<String>> = values
744 .iter()
745 .filter_map(|opt| {
746 let v = unwrap_frozen_value(*opt)?;
747 Some(match v {
748 Value::Inet(bytes) => Ok(Some(ValueFormatter::format_value(&Value::Inet(
749 bytes.clone(),
750 )))),
751 Value::Null => Ok(None),
752 other => Err(ArrowConvertError::InvalidValue(format!(
753 "expected Inet value in element, got {:?}",
754 other
755 ))),
756 })
757 })
758 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
759 Ok(Arc::new(StringArray::from(arr)))
760 }
761 CqlType::List(inner) | CqlType::Set(inner) => {
766 let element_type = cql_type_to_arrow_data_type(inner);
767 let item_field = Arc::new(Field::new("item", element_type, true));
768
769 let mut offsets: Vec<i32> = vec![0];
772 let mut flat_elements: Vec<Option<&Value>> = Vec::new();
773 let mut null_bitmap: Vec<bool> = Vec::new();
774
775 for opt in values {
776 let v = unwrap_frozen_value(*opt);
777 match v {
778 Some(Value::List(items)) | Some(Value::Set(items)) => {
779 null_bitmap.push(true);
780 for item in items {
781 flat_elements.push(Some(item));
782 }
783 offsets.push(flat_elements.len() as i32);
784 }
785 Some(Value::Null) | None => {
786 null_bitmap.push(false);
787 offsets.push(flat_elements.len() as i32);
788 }
789 Some(other) => {
790 return Err(ArrowConvertError::InvalidValue(format!(
791 "expected List/Set value, got {:?}",
792 other
793 )));
794 }
795 }
796 }
797
798 let elements_array = build_typed_value_array(inner, &flat_elements)?;
800
801 let offset_buffer = OffsetBuffer::new(offsets.into());
802 let null_buffer = NullBuffer::from(null_bitmap);
803
804 Ok(Arc::new(ListArray::new(
805 item_field,
806 offset_buffer,
807 elements_array,
808 Some(null_buffer),
809 )))
810 }
811 CqlType::Frozen(inner) => build_typed_value_array(inner, values),
814 CqlType::Map(key_type, val_type) => {
829 let key_arrow = cql_type_to_arrow_data_type(key_type);
830 let val_arrow = cql_type_to_arrow_data_type(val_type);
831
832 let mut offsets: Vec<i32> = vec![0];
833 let mut flat_keys: Vec<Option<&Value>> = Vec::new();
834 let mut flat_vals: Vec<Option<&Value>> = Vec::new();
835 let mut null_bitmap: Vec<bool> = Vec::new();
836
837 for opt in values {
838 let v = unwrap_frozen_value(*opt);
839 match v {
840 Some(Value::Map(pairs)) => {
841 null_bitmap.push(true);
842 for (k, val) in pairs {
843 if matches!(k, Value::Null) {
845 return Err(ArrowConvertError::InvalidValue(
846 "null key in map is not allowed in Arrow MapArray".to_string(),
847 ));
848 }
849 flat_keys.push(Some(k));
850 flat_vals.push(Some(val));
851 }
852 offsets.push(flat_keys.len() as i32);
853 }
854 Some(Value::Null) | None => {
855 null_bitmap.push(false);
856 offsets.push(flat_keys.len() as i32);
857 }
858 Some(other) => {
859 return Err(ArrowConvertError::InvalidValue(format!(
860 "expected Map value, got {:?}",
861 other
862 )));
863 }
864 }
865 }
866
867 let key_array = build_typed_value_array(key_type, &flat_keys)?;
869 let val_array = build_typed_value_array(val_type, &flat_vals)?;
870
871 let struct_fields = Fields::from(vec![
873 Field::new("key", key_arrow, false),
874 Field::new("value", val_arrow, true),
875 ]);
876 let entries_array =
877 StructArray::new(struct_fields.clone(), vec![key_array, val_array], None);
878
879 let map_field = Arc::new(Field::new(
880 "entries",
881 ArrowDataType::Struct(struct_fields),
882 false,
883 ));
884 let offset_buffer = OffsetBuffer::new(offsets.into());
885 let null_buffer = NullBuffer::from(null_bitmap);
886
887 Ok(Arc::new(MapArray::new(
888 map_field,
889 offset_buffer,
890 entries_array,
891 Some(null_buffer),
892 false,
893 )))
894 }
895 CqlType::Tuple(element_types) => {
906 if element_types.is_empty() {
907 let arr: Vec<Option<String>> = values
910 .iter()
911 .map(|opt| match unwrap_frozen_value(*opt) {
912 Some(Value::Null) | None => Ok(None),
913 Some(v @ Value::Tuple(_)) => Ok(Some(ValueFormatter::format_value(v))),
914 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
915 "expected Tuple value, got {:?}",
916 other
917 ))),
918 })
919 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
920 return Ok(Arc::new(StringArray::from(arr)));
921 }
922
923 let n_rows = values.len();
924 let n_fields = element_types.len();
925
926 let unwrapped: Vec<Option<&Value>> =
928 values.iter().map(|opt| unwrap_frozen_value(*opt)).collect();
929
930 for v in unwrapped.iter() {
934 match v {
935 Some(Value::Tuple(_)) | Some(Value::Null) | None => {}
936 Some(other) => {
937 return Err(ArrowConvertError::InvalidValue(format!(
938 "expected Tuple value, got {:?}",
939 other
940 )));
941 }
942 }
943 }
944
945 let null_bitmap: Vec<bool> = unwrapped
947 .iter()
948 .map(|v| !matches!(v, Some(Value::Null) | None))
949 .collect();
950
951 let null_sentinel = Value::Null;
962 let mut child_arrays: Vec<ArrayRef> = Vec::with_capacity(n_fields);
963 for (field_idx, element_type) in element_types.iter().enumerate() {
964 let child_values: Vec<Option<&Value>> = (0..n_rows)
965 .map(|row_idx| {
966 match unwrapped[row_idx] {
967 Some(Value::Tuple(items)) => {
968 Some(
970 items
971 .get(field_idx)
972 .map(|v| v as &Value)
973 .unwrap_or(&null_sentinel),
974 )
975 }
976 _ => Some(&null_sentinel),
979 }
980 })
981 .collect();
982 let child_arr = build_typed_value_array(element_type, &child_values)?;
983 child_arrays.push(child_arr);
984 }
985
986 let struct_fields: Fields = Fields::from(
987 element_types
988 .iter()
989 .enumerate()
990 .map(|(i, t)| {
991 Field::new(format!("field_{i}"), cql_type_to_arrow_data_type(t), true)
992 })
993 .collect::<Vec<_>>(),
994 );
995
996 let null_buffer = NullBuffer::from(null_bitmap);
997 Ok(Arc::new(StructArray::new(
998 struct_fields,
999 child_arrays,
1000 Some(null_buffer),
1001 )))
1002 }
1003 CqlType::Udt(_udt_name, udt_fields) => {
1014 if udt_fields.is_empty() {
1015 let arr: Vec<Option<String>> = values
1019 .iter()
1020 .map(|opt| match unwrap_frozen_value(*opt) {
1021 Some(Value::Null) | None => Ok(None),
1022 Some(v @ Value::Udt(_)) => Ok(Some(ValueFormatter::format_value(v))),
1023 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1024 "expected Udt value, got {:?}",
1025 other
1026 ))),
1027 })
1028 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
1029 return Ok(Arc::new(StringArray::from(arr)));
1030 }
1031
1032 let n_rows = values.len();
1033
1034 let unwrapped: Vec<Option<&Value>> =
1036 values.iter().map(|opt| unwrap_frozen_value(*opt)).collect();
1037
1038 for v in unwrapped.iter() {
1042 match v {
1043 Some(Value::Udt(_)) | Some(Value::Null) | None => {}
1044 Some(other) => {
1045 return Err(ArrowConvertError::InvalidValue(format!(
1046 "expected Udt value, got {:?}",
1047 other
1048 )));
1049 }
1050 }
1051 }
1052
1053 let null_bitmap: Vec<bool> = unwrapped
1055 .iter()
1056 .map(|v| !matches!(v, Some(Value::Null) | None))
1057 .collect();
1058
1059 let null_sentinel = Value::Null;
1065 let mut child_arrays: Vec<ArrayRef> = Vec::with_capacity(udt_fields.len());
1066 for (field_name, field_type) in udt_fields.iter() {
1067 let child_values: Vec<Option<&Value>> = (0..n_rows)
1068 .map(|row_idx| match unwrapped[row_idx] {
1069 Some(Value::Udt(udt_val)) => {
1070 Some(
1072 udt_val
1073 .fields
1074 .iter()
1075 .find(|f| &f.name == field_name)
1076 .and_then(|f| f.value.as_ref().map(|v| v as &Value))
1077 .unwrap_or(&null_sentinel),
1078 )
1079 }
1080 _ => Some(&null_sentinel),
1083 })
1084 .collect();
1085 let child_arr = build_typed_value_array(field_type, &child_values)?;
1086 child_arrays.push(child_arr);
1087 }
1088
1089 let struct_fields: Fields = Fields::from(
1090 udt_fields
1091 .iter()
1092 .map(|(field_name, field_type)| {
1093 Field::new(
1094 field_name.as_str(),
1095 cql_type_to_arrow_data_type(field_type),
1096 true,
1097 )
1098 })
1099 .collect::<Vec<_>>(),
1100 );
1101
1102 let null_buffer = NullBuffer::from(null_bitmap);
1103 Ok(Arc::new(StructArray::new(
1104 struct_fields,
1105 child_arrays,
1106 Some(null_buffer),
1107 )))
1108 }
1109 CqlType::Custom(_) => {
1110 let arr: Vec<Option<String>> = values
1111 .iter()
1112 .map(|opt| match opt {
1113 Some(Value::Null) | None => None,
1114 Some(v) => Some(ValueFormatter::format_value(v)),
1115 })
1116 .collect();
1117 Ok(Arc::new(StringArray::from(arr)))
1118 }
1119 }
1120}
1121
1122pub(crate) fn unwrap_frozen_type(cql_type: &CqlType) -> &CqlType {
1127 let mut t = cql_type;
1128 while let CqlType::Frozen(inner) = t {
1129 t = inner.as_ref();
1130 }
1131 t
1132}
1133
1134pub(crate) fn unwrap_frozen_value(v: Option<&Value>) -> Option<&Value> {
1139 match v {
1140 Some(Value::Frozen(inner)) => Some(inner.as_ref()),
1141 other => other,
1142 }
1143}
1144
1145pub(crate) fn data_type_to_arrow(data_type: &DataType) -> ArrowDataType {
1149 match data_type {
1150 DataType::Null => ArrowDataType::Null,
1151 DataType::Boolean => ArrowDataType::Boolean,
1152 DataType::TinyInt => ArrowDataType::Int8,
1153 DataType::SmallInt => ArrowDataType::Int16,
1154 DataType::Integer => ArrowDataType::Int32,
1155 DataType::BigInt => ArrowDataType::Int64,
1156 DataType::Float32 => ArrowDataType::Float32,
1157 DataType::Float => ArrowDataType::Float64,
1158 DataType::Text => ArrowDataType::Utf8,
1159 DataType::Blob => ArrowDataType::Binary,
1160 DataType::Timestamp => ArrowDataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
1161 DataType::Uuid => ArrowDataType::FixedSizeBinary(16),
1162 DataType::Json => ArrowDataType::Utf8,
1163 DataType::List => {
1164 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::Utf8, true)))
1165 }
1166 DataType::Set => {
1167 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::Utf8, true)))
1168 }
1169 DataType::Map => ArrowDataType::Map(
1170 Arc::new(Field::new(
1171 "entries",
1172 ArrowDataType::Struct(Fields::from(vec![
1173 Field::new("key", ArrowDataType::Utf8, false),
1174 Field::new("value", ArrowDataType::Utf8, true),
1175 ])),
1176 false,
1177 )),
1178 false,
1179 ),
1180 DataType::Tuple => ArrowDataType::Utf8, DataType::Udt => ArrowDataType::Utf8, DataType::Frozen => ArrowDataType::Utf8,
1183 DataType::Tombstone => ArrowDataType::Utf8,
1184 }
1185}
1186
1187pub(crate) fn convert_to_arrays(
1193 columns: &[ColumnInfo],
1194 rows: &[QueryRow],
1195) -> Result<Vec<ArrayRef>, ArrowConvertError> {
1196 columns
1197 .iter()
1198 .map(|col| convert_column_to_array(col, rows))
1199 .collect()
1200}
1201
1202pub(crate) fn convert_column_to_array(
1215 col: &ColumnInfo,
1216 rows: &[QueryRow],
1217) -> Result<ArrayRef, ArrowConvertError> {
1218 if let Some(cql_type) = &col.cql_type {
1220 let effective = unwrap_frozen_type(cql_type);
1222 match effective {
1223 CqlType::Date => return build_date32_array(col, rows),
1224 CqlType::Time => return build_time64_ns_array(col, rows),
1225 CqlType::Decimal => return build_decimal128_array(col, rows),
1226 CqlType::Varint => return build_varint_as_decimal128_array(col, rows),
1227 CqlType::Duration => return build_duration_utf8_array(col, rows),
1228 CqlType::Uuid | CqlType::TimeUuid => return build_uuid_fixed_binary_array(col, rows),
1229 CqlType::Inet => return build_inet_utf8_array(col, rows),
1230 CqlType::Counter => return build_int64_array(col, rows),
1231 CqlType::List(_)
1233 | CqlType::Set(_)
1234 | CqlType::Map(_, _)
1235 | CqlType::Tuple(_)
1236 | CqlType::Udt(_, _) => {
1237 let column_values: Vec<Option<&Value>> = rows
1238 .iter()
1239 .map(|row| row.values.get(col.name.as_str()))
1240 .collect();
1241 return build_typed_value_array(cql_type, &column_values);
1242 }
1243 _ => {}
1245 }
1246 }
1247
1248 match &col.data_type {
1250 DataType::Boolean => build_boolean_array(col, rows),
1251 DataType::TinyInt => build_int8_array(col, rows),
1252 DataType::SmallInt => build_int16_array(col, rows),
1253 DataType::Integer => build_int32_array(col, rows),
1254 DataType::BigInt => build_int64_array(col, rows),
1255 DataType::Float32 => build_float32_array(col, rows),
1256 DataType::Float => build_float64_array(col, rows),
1257 DataType::Text | DataType::Json => build_string_array(col, rows),
1258 DataType::Blob => build_binary_array(col, rows),
1259 DataType::Timestamp => build_timestamp_array(col, rows),
1260 DataType::Uuid => build_uuid_array(col, rows),
1261 DataType::List | DataType::Set => build_list_array(col, rows),
1262 DataType::Map => build_map_array(col, rows),
1263 DataType::Tuple
1264 | DataType::Udt
1265 | DataType::Frozen
1266 | DataType::Tombstone
1267 | DataType::Null => {
1268 build_string_array(col, rows) }
1270 }
1271}
1272
1273pub(crate) fn rescale_decimal(scale: i32, unscaled: &[u8]) -> Result<i128, ArrowConvertError> {
1289 use num_bigint::BigInt;
1290
1291 if unscaled.is_empty() {
1292 return Ok(0i128);
1293 }
1294
1295 if scale > DECIMAL_FIXED_SCALE {
1300 return Err(ArrowConvertError::InvalidValue(format!(
1301 "decimal scale {scale} exceeds the fixed export scale {DECIMAL_FIXED_SCALE}; \
1302 refusing to truncate (would lose precision)"
1303 )));
1304 }
1305
1306 let bigint = BigInt::from_signed_bytes_be(unscaled);
1308
1309 let delta = DECIMAL_FIXED_SCALE - scale;
1312
1313 let rescaled = if delta == 0 {
1314 bigint
1315 } else {
1316 let factor = BigInt::from(10i64).pow(delta as u32);
1318 bigint * factor
1319 };
1320
1321 let max_abs = BigInt::from(10i64).pow(38u32) - BigInt::from(1i64);
1324 let abs_rescaled = if rescaled.sign() == num_bigint::Sign::Minus {
1325 -rescaled.clone()
1326 } else {
1327 rescaled.clone()
1328 };
1329 if abs_rescaled > max_abs {
1330 return Err(ArrowConvertError::InvalidValue(format!(
1331 "Decimal value exceeds Decimal128(38, {DECIMAL_FIXED_SCALE}) range after rescaling"
1332 )));
1333 }
1334
1335 bigint_to_i128(&rescaled)
1336}
1337
1338fn build_boolean_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1343 let values: Vec<Option<bool>> = rows
1344 .iter()
1345 .map(
1346 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1347 None => Ok(None),
1348 Some(Value::Boolean(b)) => Ok(Some(*b)),
1349 Some(Value::Null) => Ok(None),
1350 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1351 "column '{}': expected Boolean value, got {:?}",
1352 col.name, other
1353 ))),
1354 },
1355 )
1356 .collect::<Result<Vec<Option<bool>>, ArrowConvertError>>()?;
1357 Ok(Arc::new(BooleanArray::from(values)))
1358}
1359
1360fn build_int8_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1361 let values: Vec<Option<i8>> = rows
1362 .iter()
1363 .map(
1364 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1365 None => Ok(None),
1366 Some(Value::TinyInt(i)) => Ok(Some(*i)),
1367 Some(Value::Null) => Ok(None),
1368 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1369 "column '{}': expected TinyInt value, got {:?}",
1370 col.name, other
1371 ))),
1372 },
1373 )
1374 .collect::<Result<Vec<Option<i8>>, ArrowConvertError>>()?;
1375 Ok(Arc::new(Int8Array::from(values)))
1376}
1377
1378fn build_int16_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1379 let values: Vec<Option<i16>> = rows
1380 .iter()
1381 .map(
1382 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1383 None => Ok(None),
1384 Some(Value::SmallInt(i)) => Ok(Some(*i)),
1385 Some(Value::Null) => Ok(None),
1386 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1387 "column '{}': expected SmallInt value, got {:?}",
1388 col.name, other
1389 ))),
1390 },
1391 )
1392 .collect::<Result<Vec<Option<i16>>, ArrowConvertError>>()?;
1393 Ok(Arc::new(Int16Array::from(values)))
1394}
1395
1396fn build_int32_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1397 let allow_compat = col.cql_type.is_none();
1402 let values: Vec<Option<i32>> = rows
1403 .iter()
1404 .map(
1405 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1406 None => Ok(None),
1407 Some(Value::Integer(i)) => Ok(Some(*i)),
1408 Some(Value::Date(d)) if allow_compat => Ok(Some(*d)), Some(Value::Null) => Ok(None),
1410 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1411 "column '{}': expected Int value, got {:?}",
1412 col.name, other
1413 ))),
1414 },
1415 )
1416 .collect::<Result<Vec<Option<i32>>, ArrowConvertError>>()?;
1417 Ok(Arc::new(Int32Array::from(values)))
1418}
1419
1420fn build_int64_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1421 let effective = col.cql_type.as_ref().map(unwrap_frozen_type);
1427 let allow_counter = matches!(effective, None | Some(CqlType::Counter));
1428 let allow_compat = effective.is_none();
1429 let values: Vec<Option<i64>> = rows
1430 .iter()
1431 .map(
1432 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1433 None => Ok(None),
1434 Some(Value::BigInt(i)) => Ok(Some(*i)),
1435 Some(Value::Counter(c)) if allow_counter => Ok(Some(*c)),
1436 Some(Value::Time(t)) if allow_compat => Ok(Some(*t)), Some(Value::Null) => Ok(None),
1438 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1439 "column '{}': expected BigInt value, got {:?}",
1440 col.name, other
1441 ))),
1442 },
1443 )
1444 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
1445 Ok(Arc::new(Int64Array::from(values)))
1446}
1447
1448fn build_float32_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1449 let values: Vec<Option<f32>> = rows
1450 .iter()
1451 .map(
1452 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1453 None => Ok(None),
1454 Some(Value::Float32(f)) => Ok(Some(*f)),
1455 Some(Value::Float(f)) => Ok(Some(*f as f32)),
1459 Some(Value::Null) => Ok(None),
1460 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1461 "column '{}': expected Float value, got {:?}",
1462 col.name, other
1463 ))),
1464 },
1465 )
1466 .collect::<Result<Vec<Option<f32>>, ArrowConvertError>>()?;
1467 Ok(Arc::new(Float32Array::from(values)))
1468}
1469
1470fn build_float64_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1471 let values: Vec<Option<f64>> = rows
1472 .iter()
1473 .map(
1474 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1475 None => Ok(None),
1476 Some(Value::Float(f)) => Ok(Some(*f)),
1477 Some(Value::Float32(f)) => Ok(Some(*f as f64)),
1478 Some(Value::Null) => Ok(None),
1479 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1480 "column '{}': expected Double value, got {:?}",
1481 col.name, other
1482 ))),
1483 },
1484 )
1485 .collect::<Result<Vec<Option<f64>>, ArrowConvertError>>()?;
1486 Ok(Arc::new(Float64Array::from(values)))
1487}
1488
1489fn build_string_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1490 let strict_text = matches!(
1496 col.cql_type.as_ref().map(unwrap_frozen_type),
1497 Some(CqlType::Text | CqlType::Ascii | CqlType::Varchar)
1498 );
1499 let values: Vec<Option<String>> = rows
1500 .iter()
1501 .map(
1502 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1503 None => Ok(None),
1504 Some(Value::Null) => Ok(None),
1505 Some(Value::Text(s)) => Ok(Some(s.clone())),
1506 Some(Value::Json(j)) if !strict_text => Ok(Some(j.to_string())),
1509 Some(other) if strict_text => Err(ArrowConvertError::InvalidValue(format!(
1510 "column '{}': expected Text value, got {:?}",
1511 col.name, other
1512 ))),
1513 Some(other) => Ok(Some(ValueFormatter::format_value(other))),
1515 },
1516 )
1517 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
1518 Ok(Arc::new(StringArray::from(values)))
1519}
1520
1521fn build_binary_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1522 let values: Vec<Option<&[u8]>> = rows
1523 .iter()
1524 .map(
1525 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1526 None => Ok(None),
1527 Some(Value::Blob(b)) => Ok(Some(b.as_slice())),
1528 Some(Value::Null) => Ok(None),
1529 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1530 "column '{}': expected Blob value, got {:?}",
1531 col.name, other
1532 ))),
1533 },
1534 )
1535 .collect::<Result<Vec<Option<&[u8]>>, ArrowConvertError>>()?;
1536 Ok(Arc::new(BinaryArray::from(values)))
1537}
1538
1539fn build_timestamp_array(
1540 col: &ColumnInfo,
1541 rows: &[QueryRow],
1542) -> Result<ArrayRef, ArrowConvertError> {
1543 let values: Vec<Option<i64>> = rows
1544 .iter()
1545 .map(
1546 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1547 None => Ok(None),
1548 Some(Value::Timestamp(ts)) => Ok(Some(*ts)),
1549 Some(Value::Null) => Ok(None),
1550 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1551 "column '{}': expected Timestamp value, got {:?}",
1552 col.name, other
1553 ))),
1554 },
1555 )
1556 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
1557 Ok(Arc::new(
1558 TimestampMillisecondArray::from(values).with_timezone("UTC"),
1559 ))
1560}
1561
1562fn build_uuid_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1563 let values: Vec<Option<[u8; 16]>> = rows
1564 .iter()
1565 .map(
1566 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1567 None => Ok(None),
1568 Some(Value::Uuid(uuid)) => Ok(Some(*uuid)),
1569 Some(Value::Null) => Ok(None),
1570 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1571 "column '{}': expected Uuid value, got {:?}",
1572 col.name, other
1573 ))),
1574 },
1575 )
1576 .collect::<Result<Vec<Option<[u8; 16]>>, ArrowConvertError>>()?;
1577
1578 let mut builder = arrow::array::FixedSizeBinaryBuilder::new(16);
1579 for opt in values {
1580 match opt {
1581 Some(uuid) => builder.append_value(uuid)?,
1582 None => builder.append_null(),
1583 }
1584 }
1585 Ok(Arc::new(builder.finish()))
1586}
1587
1588fn build_date32_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1594 let values: Vec<Option<i32>> = rows
1595 .iter()
1596 .map(
1597 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1598 None => Ok(None),
1599 Some(Value::Date(days)) => Ok(Some(*days)),
1600 Some(Value::Null) => Ok(None),
1601 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1602 "column '{}': expected Date value, got {:?}",
1603 col.name, other
1604 ))),
1605 },
1606 )
1607 .collect::<Result<Vec<Option<i32>>, ArrowConvertError>>()?;
1608 Ok(Arc::new(Date32Array::from(values)))
1609}
1610
1611fn build_time64_ns_array(
1613 col: &ColumnInfo,
1614 rows: &[QueryRow],
1615) -> Result<ArrayRef, ArrowConvertError> {
1616 let values: Vec<Option<i64>> = rows
1617 .iter()
1618 .map(
1619 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1620 None => Ok(None),
1621 Some(Value::Time(nanos)) => Ok(Some(*nanos)),
1622 Some(Value::Null) => Ok(None),
1623 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1624 "column '{}': expected Time value, got {:?}",
1625 col.name, other
1626 ))),
1627 },
1628 )
1629 .collect::<Result<Vec<Option<i64>>, ArrowConvertError>>()?;
1630 Ok(Arc::new(Time64NanosecondArray::from(values)))
1631}
1632
1633fn build_decimal128_array(
1636 col: &ColumnInfo,
1637 rows: &[QueryRow],
1638) -> Result<ArrayRef, ArrowConvertError> {
1639 let mut builder = arrow::array::Decimal128Builder::new()
1640 .with_precision_and_scale(DECIMAL_MAX_PRECISION, DECIMAL_FIXED_SCALE as i8)?;
1641
1642 for row in rows {
1643 match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1644 Some(Value::Decimal { scale, unscaled }) => {
1645 let rescaled = rescale_decimal(*scale, unscaled).map_err(|e| {
1646 ArrowConvertError::InvalidValue(format!("Column '{}': {e}", col.name))
1647 })?;
1648 builder.append_value(rescaled);
1649 }
1650 Some(Value::Null) | None => {
1651 builder.append_null();
1652 }
1653 Some(other) => {
1654 return Err(ArrowConvertError::InvalidValue(format!(
1655 "Column '{}': expected Decimal value, got {:?}",
1656 col.name, other
1657 )));
1658 }
1659 }
1660 }
1661 Ok(Arc::new(builder.finish()))
1662}
1663
1664fn build_varint_as_decimal128_array(
1666 col: &ColumnInfo,
1667 rows: &[QueryRow],
1668) -> Result<ArrayRef, ArrowConvertError> {
1669 use num_bigint::BigInt;
1670
1671 let mut builder = arrow::array::Decimal128Builder::new()
1672 .with_precision_and_scale(DECIMAL_MAX_PRECISION, 0)?;
1673
1674 for row in rows {
1675 match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1676 Some(Value::Varint(bytes)) => {
1677 if bytes.is_empty() {
1678 builder.append_value(0);
1679 } else {
1680 let bigint = BigInt::from_signed_bytes_be(bytes);
1681
1682 let max_abs = BigInt::from(10i64).pow(38u32) - BigInt::from(1i64);
1684 let abs_val = if bigint.sign() == num_bigint::Sign::Minus {
1685 -bigint.clone()
1686 } else {
1687 bigint.clone()
1688 };
1689 if abs_val > max_abs {
1690 return Err(ArrowConvertError::InvalidValue(format!(
1691 "Column '{}': varint value exceeds Decimal128(38, 0) range",
1692 col.name
1693 )));
1694 }
1695
1696 let i128_val = bigint_to_i128(&bigint).map_err(|e| {
1697 ArrowConvertError::InvalidValue(format!("Column '{}': {e}", col.name))
1698 })?;
1699 builder.append_value(i128_val);
1700 }
1701 }
1702 Some(Value::Null) | None => {
1703 builder.append_null();
1704 }
1705 Some(other) => {
1706 return Err(ArrowConvertError::InvalidValue(format!(
1707 "Column '{}': expected Varint value, got {:?}",
1708 col.name, other
1709 )));
1710 }
1711 }
1712 }
1713 Ok(Arc::new(builder.finish()))
1714}
1715
1716fn build_duration_utf8_array(
1718 col: &ColumnInfo,
1719 rows: &[QueryRow],
1720) -> Result<ArrayRef, ArrowConvertError> {
1721 let values: Vec<Option<String>> = rows
1722 .iter()
1723 .map(
1724 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1725 None => Ok(None),
1726 Some(v @ Value::Duration { .. }) => Ok(Some(ValueFormatter::format_value(v))),
1727 Some(Value::Null) => Ok(None),
1728 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1729 "column '{}': expected Duration value, got {:?}",
1730 col.name, other
1731 ))),
1732 },
1733 )
1734 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
1735 Ok(Arc::new(StringArray::from(values)))
1736}
1737
1738fn build_uuid_fixed_binary_array(
1740 col: &ColumnInfo,
1741 rows: &[QueryRow],
1742) -> Result<ArrayRef, ArrowConvertError> {
1743 let mut builder = arrow::array::FixedSizeBinaryBuilder::new(16);
1744 for row in rows {
1745 match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1746 Some(Value::Uuid(bytes)) => builder.append_value(bytes)?,
1747 Some(Value::Null) | None => builder.append_null(),
1748 Some(other) => {
1749 return Err(ArrowConvertError::InvalidValue(format!(
1750 "Column '{}': expected Uuid value, got {:?}",
1751 col.name, other
1752 )));
1753 }
1754 }
1755 }
1756 Ok(Arc::new(builder.finish()))
1757}
1758
1759fn build_inet_utf8_array(
1761 col: &ColumnInfo,
1762 rows: &[QueryRow],
1763) -> Result<ArrayRef, ArrowConvertError> {
1764 let values: Vec<Option<String>> = rows
1765 .iter()
1766 .map(
1767 |row| match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1768 None => Ok(None),
1769 Some(Value::Inet(bytes)) => Ok(Some(ValueFormatter::format_value(&Value::Inet(
1770 bytes.clone(),
1771 )))),
1772 Some(Value::Null) => Ok(None),
1773 Some(other) => Err(ArrowConvertError::InvalidValue(format!(
1774 "column '{}': expected Inet value, got {:?}",
1775 col.name, other
1776 ))),
1777 },
1778 )
1779 .collect::<Result<Vec<Option<String>>, ArrowConvertError>>()?;
1780 Ok(Arc::new(StringArray::from(values)))
1781}
1782
1783fn build_list_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1784 let mut offsets: Vec<i32> = vec![0];
1786 let mut values: Vec<Option<String>> = Vec::new();
1787 let mut null_bitmap: Vec<bool> = Vec::new();
1788
1789 for row in rows {
1790 match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1791 Some(Value::List(items)) | Some(Value::Set(items)) => {
1792 null_bitmap.push(true);
1793 for item in items {
1794 values.push(Some(ValueFormatter::format_value(item)));
1795 }
1796 offsets.push(values.len() as i32);
1797 }
1798 Some(Value::Null) | None => {
1799 null_bitmap.push(false);
1800 offsets.push(values.len() as i32);
1801 }
1802 Some(other) => {
1803 return Err(ArrowConvertError::InvalidValue(format!(
1804 "column '{}': expected List/Set value, got {:?}",
1805 col.name, other
1806 )));
1807 }
1808 }
1809 }
1810
1811 let values_array = Arc::new(StringArray::from(values)) as ArrayRef;
1812 let field = Arc::new(Field::new("item", ArrowDataType::Utf8, true));
1813 let offset_buffer = OffsetBuffer::new(offsets.into());
1814 let null_buffer = NullBuffer::from(null_bitmap);
1815
1816 Ok(Arc::new(ListArray::new(
1817 field,
1818 offset_buffer,
1819 values_array,
1820 Some(null_buffer),
1821 )))
1822}
1823
1824fn build_map_array(col: &ColumnInfo, rows: &[QueryRow]) -> Result<ArrayRef, ArrowConvertError> {
1825 let mut offsets: Vec<i32> = vec![0];
1827 let mut keys: Vec<Option<String>> = Vec::new();
1828 let mut values: Vec<Option<String>> = Vec::new();
1829 let mut null_bitmap: Vec<bool> = Vec::new();
1830
1831 for row in rows {
1832 match unwrap_frozen_value(row.values.get(col.name.as_str())) {
1833 Some(Value::Map(pairs)) => {
1834 null_bitmap.push(true);
1835 for (k, v) in pairs {
1836 keys.push(Some(ValueFormatter::format_value(k)));
1837 values.push(Some(ValueFormatter::format_value(v)));
1838 }
1839 offsets.push(keys.len() as i32);
1840 }
1841 Some(Value::Null) | None => {
1842 null_bitmap.push(false);
1843 offsets.push(keys.len() as i32);
1844 }
1845 Some(other) => {
1846 return Err(ArrowConvertError::InvalidValue(format!(
1847 "column '{}': expected Map value, got {:?}",
1848 col.name, other
1849 )));
1850 }
1851 }
1852 }
1853
1854 let key_array = Arc::new(StringArray::from(keys)) as ArrayRef;
1856 let value_array = Arc::new(StringArray::from(values)) as ArrayRef;
1857
1858 let struct_fields = Fields::from(vec![
1859 Field::new("key", ArrowDataType::Utf8, false),
1860 Field::new("value", ArrowDataType::Utf8, true),
1861 ]);
1862
1863 let entries_array = StructArray::new(struct_fields.clone(), vec![key_array, value_array], None);
1864
1865 let map_field = Arc::new(Field::new(
1866 "entries",
1867 ArrowDataType::Struct(struct_fields),
1868 false,
1869 ));
1870 let offset_buffer = OffsetBuffer::new(offsets.into());
1871 let null_buffer = NullBuffer::from(null_bitmap);
1872
1873 Ok(Arc::new(MapArray::new(
1874 map_field,
1875 offset_buffer,
1876 entries_array,
1877 Some(null_buffer),
1878 false,
1879 )))
1880}
1881
1882#[cfg(test)]
1887mod tests {
1888 use super::*;
1889 use crate::query::{ColumnInfo, QueryRow};
1890 use crate::schema::CqlType;
1891 use crate::types::{DataType, Value};
1892 use crate::RowKey;
1893 use arrow::array::{Array, Int32Array};
1894
1895 fn col(name: &str, data_type: DataType, cql_type: Option<CqlType>) -> ColumnInfo {
1897 ColumnInfo {
1898 name: name.to_string(),
1899 data_type,
1900 nullable: true,
1901 position: 0,
1902 table_name: None,
1903 cql_type,
1904 }
1905 }
1906
1907 fn row_one(name: &str, value: Value) -> QueryRow {
1909 let mut values: HashMap<Arc<str>, Value> = HashMap::new();
1910 values.insert(name.into(), value);
1911 QueryRow {
1912 values,
1913 key: RowKey::new(Vec::new()),
1914 metadata: Default::default(),
1915 cell_metadata: None,
1916 }
1917 }
1918
1919 fn row_absent() -> QueryRow {
1921 QueryRow {
1922 values: HashMap::new(),
1923 key: RowKey::new(Vec::new()),
1924 metadata: Default::default(),
1925 cell_metadata: None,
1926 }
1927 }
1928
1929 fn is_invalid_value(res: Result<arrow::record_batch::RecordBatch, ArrowConvertError>) -> bool {
1930 matches!(res, Err(ArrowConvertError::InvalidValue(_)))
1931 }
1932
1933 #[test]
1936 fn typed_scalar_type_mismatch_is_error() {
1937 let columns = vec![col("d", DataType::Timestamp, Some(CqlType::Date))];
1938 let rows = vec![row_one("d", Value::Text("not-a-date".into()))];
1939 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
1940 }
1941
1942 #[test]
1945 fn flat_builder_type_mismatch_is_error() {
1946 let columns = vec![col("n", DataType::Integer, None)];
1947 let rows = vec![row_one("n", Value::Text("nope".into()))];
1948 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
1949 }
1950
1951 #[test]
1954 fn collection_expected_list_got_scalar_is_error() {
1955 let columns = vec![col(
1956 "l",
1957 DataType::List,
1958 Some(CqlType::List(Box::new(CqlType::Int))),
1959 )];
1960 let rows = vec![row_one("l", Value::Integer(5))];
1961 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
1962 }
1963
1964 #[test]
1967 fn collection_mistyped_element_is_error() {
1968 let columns = vec![col(
1969 "l",
1970 DataType::List,
1971 Some(CqlType::List(Box::new(CqlType::Int))),
1972 )];
1973 let rows = vec![row_one(
1974 "l",
1975 Value::List(vec![Value::Integer(1), Value::Text("bad".into())]),
1976 )];
1977 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
1978 }
1979
1980 #[test]
1983 fn collection_expected_map_got_scalar_is_error() {
1984 let columns = vec![col(
1985 "m",
1986 DataType::Map,
1987 Some(CqlType::Map(
1988 Box::new(CqlType::Text),
1989 Box::new(CqlType::Int),
1990 )),
1991 )];
1992 let rows = vec![row_one("m", Value::Integer(7))];
1993 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
1994 }
1995
1996 #[test]
1999 fn null_and_absent_still_build_ok() {
2000 let columns = vec![col("n", DataType::Integer, None)];
2001 let rows = vec![row_one("n", Value::Null), row_absent()];
2002 let batch = rows_to_record_batch(&columns, &rows).expect("null/absent must build");
2003 assert_eq!(batch.num_rows(), 2);
2004 assert_eq!(batch.column(0).null_count(), 2);
2005 }
2006
2007 #[test]
2009 fn correctly_typed_value_builds_ok() {
2010 let columns = vec![col("n", DataType::Integer, None)];
2011 let rows = vec![row_one("n", Value::Integer(42))];
2012 let batch = rows_to_record_batch(&columns, &rows).expect("well-typed value must build");
2013 let arr = batch
2014 .column(0)
2015 .as_any()
2016 .downcast_ref::<Int32Array>()
2017 .expect("Int32Array");
2018 assert_eq!(arr.value(0), 42);
2019 assert_eq!(arr.null_count(), 0);
2020 }
2021
2022 #[test]
2026 fn decimal_scale_above_fixed_is_error() {
2027 let columns = vec![col("d", DataType::Blob, Some(CqlType::Decimal))];
2028 let unscaled = num_bigint::BigInt::from(123_456_789_012i64).to_signed_bytes_be();
2030 let rows = vec![row_one(
2031 "d",
2032 Value::Decimal {
2033 scale: 12,
2034 unscaled,
2035 },
2036 )];
2037 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2038 }
2039
2040 #[test]
2043 fn decimal_scale_within_fixed_builds_ok() {
2044 use arrow::array::Decimal128Array;
2045 let columns = vec![col("d", DataType::Blob, Some(CqlType::Decimal))];
2046 let unscaled = num_bigint::BigInt::from(123_456i64).to_signed_bytes_be();
2048 let rows = vec![row_one("d", Value::Decimal { scale: 3, unscaled })];
2049 let batch = rows_to_record_batch(&columns, &rows).expect("in-range decimal must build");
2050 let arr = batch
2051 .column(0)
2052 .as_any()
2053 .downcast_ref::<Decimal128Array>()
2054 .expect("Decimal128Array");
2055 assert_eq!(arr.value(0), 123_456_000_000i128);
2056 assert_eq!(arr.null_count(), 0);
2057 }
2058
2059 #[test]
2062 fn decimal_null_and_absent_still_null() {
2063 let columns = vec![col("d", DataType::Blob, Some(CqlType::Decimal))];
2064 let rows = vec![row_one("d", Value::Null), row_absent()];
2065 let batch = rows_to_record_batch(&columns, &rows).expect("null/absent decimal must build");
2066 assert_eq!(batch.num_rows(), 2);
2067 assert_eq!(batch.column(0).null_count(), 2);
2068 }
2069
2070 #[test]
2076 fn float32_column_accepts_wide_float_value() {
2077 let flat = vec![col("h", DataType::Float32, None)];
2079 let rows = vec![row_one("h", Value::Float(1.84f32 as f64))];
2080 let batch = rows_to_record_batch(&flat, &rows).expect("wide float must narrow, not error");
2081 let arr = batch
2082 .column(0)
2083 .as_any()
2084 .downcast_ref::<Float32Array>()
2085 .expect("Float32Array");
2086 assert_eq!(arr.value(0), 1.84f32);
2087 assert_eq!(arr.null_count(), 0);
2088
2089 let typed = vec![col("h", DataType::Float32, Some(CqlType::Float))];
2091 let rows = vec![row_one("h", Value::Float(1.84f32 as f64))];
2092 let batch =
2093 rows_to_record_batch(&typed, &rows).expect("wide float (typed) must narrow, not error");
2094 let arr = batch
2095 .column(0)
2096 .as_any()
2097 .downcast_ref::<Float32Array>()
2098 .expect("Float32Array");
2099 assert_eq!(arr.value(0), 1.84f32);
2100 }
2101
2102 #[test]
2105 fn tuple_expected_tuple_got_scalar_is_error() {
2106 let columns = vec![col(
2107 "t",
2108 DataType::Text,
2109 Some(CqlType::Tuple(vec![CqlType::Int, CqlType::Text])),
2110 )];
2111 let rows = vec![row_one("t", Value::Text("not-a-tuple".into()))];
2112 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2113 }
2114
2115 #[test]
2118 fn tuple_null_and_absent_still_build_ok() {
2119 let columns = vec![col(
2120 "t",
2121 DataType::Text,
2122 Some(CqlType::Tuple(vec![CqlType::Int, CqlType::Text])),
2123 )];
2124 let rows = vec![row_one("t", Value::Null), row_absent()];
2125 let batch = rows_to_record_batch(&columns, &rows).expect("null/absent tuple must build");
2126 assert_eq!(batch.num_rows(), 2);
2127 assert_eq!(batch.column(0).null_count(), 2);
2128 }
2129
2130 #[test]
2133 fn udt_expected_udt_got_scalar_is_error() {
2134 let columns = vec![col(
2135 "u",
2136 DataType::Text,
2137 Some(CqlType::Udt(
2138 "my_type".into(),
2139 vec![("a".into(), CqlType::Int), ("b".into(), CqlType::Text)],
2140 )),
2141 )];
2142 let rows = vec![row_one("u", Value::Integer(9))];
2143 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2144 }
2145
2146 #[test]
2149 fn udt_null_and_absent_still_build_ok() {
2150 let columns = vec![col(
2151 "u",
2152 DataType::Text,
2153 Some(CqlType::Udt(
2154 "my_type".into(),
2155 vec![("a".into(), CqlType::Int), ("b".into(), CqlType::Text)],
2156 )),
2157 )];
2158 let rows = vec![row_one("u", Value::Null), row_absent()];
2159 let batch = rows_to_record_batch(&columns, &rows).expect("null/absent UDT must build");
2160 assert_eq!(batch.num_rows(), 2);
2161 assert_eq!(batch.column(0).null_count(), 2);
2162 }
2163
2164 #[test]
2168 fn empty_field_udt_expected_udt_got_scalar_is_error() {
2169 let columns = vec![col(
2170 "u",
2171 DataType::Text,
2172 Some(CqlType::Udt("unresolved".into(), vec![])),
2173 )];
2174 let rows = vec![row_one("u", Value::Integer(9))];
2175 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2176 }
2177
2178 #[test]
2181 fn empty_field_tuple_expected_tuple_got_scalar_is_error() {
2182 let columns = vec![col("t", DataType::Text, Some(CqlType::Tuple(vec![])))];
2183 let rows = vec![row_one("t", Value::Text("nope".into()))];
2184 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2185 }
2186
2187 #[test]
2191 fn authoritative_text_column_type_mismatch_is_error() {
2192 for cql in [CqlType::Text, CqlType::Ascii, CqlType::Varchar] {
2193 let columns = vec![col("s", DataType::Text, Some(cql))];
2194 let rows = vec![row_one("s", Value::Integer(1))];
2195 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2196 }
2197 }
2198
2199 #[test]
2202 fn authoritative_text_column_rejects_json() {
2203 let columns = vec![col("s", DataType::Text, Some(CqlType::Text))];
2204 let rows = vec![row_one("s", Value::Json(serde_json::json!({"a": 1})))];
2205 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2206 }
2207
2208 #[test]
2212 fn frozen_wrapped_scalar_values_build_ok() {
2213 let text_cols = vec![col(
2215 "s",
2216 DataType::Text,
2217 Some(CqlType::Frozen(Box::new(CqlType::Text))),
2218 )];
2219 let text_rows = vec![row_one(
2220 "s",
2221 Value::Frozen(Box::new(Value::Text("hi".into()))),
2222 )];
2223 let batch =
2224 rows_to_record_batch(&text_cols, &text_rows).expect("frozen<text> value must build");
2225 let arr = batch
2226 .column(0)
2227 .as_any()
2228 .downcast_ref::<StringArray>()
2229 .expect("StringArray");
2230 assert_eq!(arr.value(0), "hi");
2231
2232 let date_cols = vec![col(
2234 "d",
2235 DataType::Integer,
2236 Some(CqlType::Frozen(Box::new(CqlType::Date))),
2237 )];
2238 let date_rows = vec![row_one("d", Value::Frozen(Box::new(Value::Date(19_000))))];
2239 let batch =
2240 rows_to_record_batch(&date_cols, &date_rows).expect("frozen<date> value must build");
2241 assert_eq!(batch.num_rows(), 1);
2242 assert_eq!(batch.column(0).null_count(), 0);
2243 }
2244
2245 #[test]
2248 fn authoritative_text_column_builds_ok() {
2249 let columns = vec![col("s", DataType::Text, Some(CqlType::Text))];
2250 let rows = vec![
2251 row_one("s", Value::Text("hi".into())),
2252 row_one("s", Value::Null),
2253 row_absent(),
2254 ];
2255 let batch = rows_to_record_batch(&columns, &rows).expect("well-typed text must build");
2256 let arr = batch
2257 .column(0)
2258 .as_any()
2259 .downcast_ref::<StringArray>()
2260 .expect("StringArray");
2261 assert_eq!(arr.value(0), "hi");
2262 assert_eq!(arr.null_count(), 2);
2263 }
2264
2265 #[test]
2268 fn authoritative_int_column_rejects_date() {
2269 let columns = vec![col("n", DataType::Integer, Some(CqlType::Int))];
2270 let rows = vec![row_one("n", Value::Date(19_000))];
2271 assert!(is_invalid_value(rows_to_record_batch(&columns, &rows)));
2272 }
2273
2274 #[test]
2277 fn opaque_int_column_accepts_date() {
2278 let columns = vec![col("n", DataType::Integer, None)];
2279 let rows = vec![row_one("n", Value::Date(19_000))];
2280 let batch = rows_to_record_batch(&columns, &rows).expect("opaque int accepts Date");
2281 let arr = batch
2282 .column(0)
2283 .as_any()
2284 .downcast_ref::<Int32Array>()
2285 .expect("Int32Array");
2286 assert_eq!(arr.value(0), 19_000);
2287 }
2288
2289 #[test]
2293 fn authoritative_bigint_counter_reject_mismatch() {
2294 let bigint_time = vec![col("b", DataType::BigInt, Some(CqlType::BigInt))];
2295 assert!(is_invalid_value(rows_to_record_batch(
2296 &bigint_time,
2297 &[row_one("b", Value::Time(123))]
2298 )));
2299
2300 let counter_time = vec![col("c", DataType::BigInt, Some(CqlType::Counter))];
2301 assert!(is_invalid_value(rows_to_record_batch(
2302 &counter_time,
2303 &[row_one("c", Value::Time(123))]
2304 )));
2305
2306 let bigint_counter = vec![col("b", DataType::BigInt, Some(CqlType::BigInt))];
2307 assert!(is_invalid_value(rows_to_record_batch(
2308 &bigint_counter,
2309 &[row_one("b", Value::Counter(7))]
2310 )));
2311 }
2312
2313 #[test]
2315 fn authoritative_counter_column_accepts_counter() {
2316 let columns = vec![col("c", DataType::BigInt, Some(CqlType::Counter))];
2317 let rows = vec![row_one("c", Value::Counter(42))];
2318 let batch = rows_to_record_batch(&columns, &rows).expect("counter accepts Counter");
2319 assert_eq!(batch.num_rows(), 1);
2320 assert_eq!(batch.column(0).null_count(), 0);
2321 }
2322}