use std::{slice::from_ref, sync::Arc};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::Schema;
use reifydb_codec::{
frame::{decode::decode_frames, encode::encode_frames, format::Encoding, options::EncodeOptions},
json::{
from::frames_from_json,
to::frames_to_json,
wire_type::{from_json, to_json},
},
key::serializer::KeySerializer,
reader::Reader,
tag::ValueKind,
typeinfo::{decode_value_type, encode_value_type},
value::{decode_value, encode_value},
};
use reifydb_value::value::{
Value,
column_view::ColumnView,
container::digest_array::digest_array,
digest::Digest,
duration::Duration,
frame::frame::Frame,
value_type::{ValueType, field::named},
};
use serde_json::json;
const ACCURACY: u32 = 10_000;
const SUPPORTED_INNER: [ValueType; 13] = [
ValueType::Float4,
ValueType::Float8,
ValueType::Int1,
ValueType::Int2,
ValueType::Int4,
ValueType::Int8,
ValueType::Int16,
ValueType::Uint1,
ValueType::Uint2,
ValueType::Uint4,
ValueType::Uint8,
ValueType::Uint16,
ValueType::Duration,
];
fn digest_type(inner: ValueType, accuracy: u32) -> ValueType {
ValueType::Digest {
inner: Box::new(inner),
accuracy,
}
}
fn float_digest(values: &[f64]) -> Digest {
let mut digest = Digest::new(ValueType::Float8, ACCURACY).unwrap();
for value in values {
digest.add_value(&Value::float8(*value)).unwrap();
}
digest
}
fn duration_digest(millis: &[i64]) -> Digest {
let mut digest = Digest::new(ValueType::Duration, ACCURACY).unwrap();
for ms in millis {
digest.add_value(&Value::Duration(Duration::from_milliseconds(*ms).unwrap())).unwrap();
}
digest
}
fn digest_column(rows: Vec<Option<Digest>>, inner: ValueType) -> (ValueType, ArrayRef) {
let value_type = match rows.iter().all(Option::is_some) {
true => digest_type(inner, ACCURACY),
false => ValueType::Option(Box::new(digest_type(inner, ACCURACY))),
};
(value_type, Arc::new(digest_array(rows)))
}
fn frame((value_type, array): (ValueType, ArrayRef)) -> Frame {
let (field, array) = named("d", value_type.into(), array);
Frame::from(RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![array]).unwrap())
}
fn view(frame: &Frame) -> ColumnView<'_> {
ColumnView::try_from((frame.batch.column(0), frame.batch.schema_ref().field(0))).unwrap()
}
fn sample_rows() -> Vec<Option<Digest>> {
vec![
Some(float_digest(&[1.0, 2.0, 3.0])),
None,
Some(float_digest(&[])),
Some(float_digest(&[-5.0, 0.0, f64::INFINITY, 1e9])),
]
}
fn assert_same_cells(expected: &ColumnView, actual: &ColumnView) {
assert_eq!(expected.len(), actual.len(), "row count changed");
for row in 0..expected.len() {
assert_eq!(expected.get_value(row), actual.get_value(row), "row {row} changed");
}
}
fn find(haystack: &[u8], needle: &[u8]) -> Vec<usize> {
haystack.windows(needle.len()).enumerate().filter(|(_, w)| *w == needle).map(|(i, _)| i).collect()
}
#[test]
fn digest_kind_is_appended_last_as_tag_30() {
assert_eq!(ValueKind::Digest.byte(), 30);
assert_eq!(ValueKind::ALL.len(), 31);
assert_eq!(ValueKind::ALL[30], ValueKind::Digest);
assert_eq!(ValueKind::from_byte(30), Some(ValueKind::Digest));
assert_eq!(ValueKind::from_byte(31), None);
assert_eq!(ValueKind::of_type(&digest_type(ValueType::Float8, ACCURACY)), ValueKind::Digest);
}
#[test]
fn digest_inner_tag_equals_value_kind_for_every_supported_inner_type() {
for inner in SUPPORTED_INNER {
let bytes = Digest::new(inner.clone(), ACCURACY).unwrap().encode();
assert_eq!(bytes[1], ValueKind::of_type(&inner).byte(), "digest inner tag for {inner}");
}
}
#[test]
fn typeinfo_carries_inner_kind_and_accuracy_ppm_little_endian() {
let mut buf = Vec::new();
encode_value_type(&digest_type(ValueType::Duration, 12_345), &mut buf).unwrap();
assert_eq!(buf, vec![30, 18, 0x39, 0x30, 0, 0]);
let mut buf = Vec::new();
encode_value_type(&ValueType::Option(Box::new(digest_type(ValueType::Float8, ACCURACY))), &mut buf).unwrap();
assert_eq!(buf, vec![(1 << 6) | 30, 3, 0x10, 0x27, 0, 0]);
}
#[test]
fn typeinfo_round_trips_every_inner_type_under_option_layers() {
for inner in SUPPORTED_INNER {
for depth in [0, 1, 3, 4] {
let ty = (0..depth)
.fold(digest_type(inner.clone(), 1_000), |ty, _| ValueType::Option(Box::new(ty)));
let mut buf = Vec::new();
encode_value_type(&ty, &mut buf).unwrap();
let mut reader = Reader::new(&buf);
assert_eq!(decode_value_type(&mut reader).unwrap(), ty, "{ty} did not survive typeinfo");
}
}
}
#[test]
fn typeinfo_rejects_an_unsupported_inner_type_or_accuracy_in_both_directions() {
let mut buf = Vec::new();
assert!(encode_value_type(&digest_type(ValueType::Utf8, ACCURACY), &mut buf).is_err());
assert!(encode_value_type(&digest_type(ValueType::Float8, 999), &mut Vec::new()).is_err());
let utf8_inner = [32, 9, 0x10, 0x27, 0, 0];
assert!(decode_value_type(&mut Reader::new(&utf8_inner)).is_err());
let accuracy_too_fine = [32, 3, 0xE7, 0x03, 0, 0];
assert!(decode_value_type(&mut Reader::new(&accuracy_too_fine)).is_err());
let truncated = [32, 3, 0x10];
assert!(decode_value_type(&mut Reader::new(&truncated)).is_err());
}
#[test]
fn value_codec_round_trips_a_digest_and_a_typed_none_of_digest() {
let digest = Value::Digest(Box::new(float_digest(&[1.0, 2.0, 3.0])));
assert_eq!(decode_value(&encode_value(&digest).unwrap()).unwrap(), digest);
let duration = Value::Digest(Box::new(duration_digest(&[5, 10, 250])));
assert_eq!(decode_value(&encode_value(&duration).unwrap()).unwrap(), duration);
let none = Value::none_of(digest_type(ValueType::Duration, ACCURACY));
assert_eq!(decode_value(&encode_value(&none).unwrap()).unwrap(), none);
}
#[test]
fn frame_round_trip_keeps_digests_nones_and_column_params_for_every_compression() {
let data = frame(digest_column(sample_rows(), ValueType::Float8));
for options in [
EncodeOptions::default(),
EncodeOptions::none(),
EncodeOptions::fast(),
EncodeOptions::max(),
EncodeOptions::forced(Encoding::Dict),
EncodeOptions::forced(Encoding::Rle),
] {
let bytes = encode_frames(from_ref(&data), &options).unwrap();
let decoded = decode_frames(&bytes).unwrap();
let column = view(&decoded[0]);
assert_same_cells(&view(&data), &column);
assert!(
column.is_nullable(),
"a digest column with a none row must decode as an option, got {:?}",
column.get_type()
);
assert_eq!(
(0..column.len()).map(|i| !column.none_at(i)).collect::<Vec<_>>(),
vec![true, false, true, true]
);
assert_eq!(
column.get_type(),
ValueType::Option(Box::new(digest_type(ValueType::Float8, ACCURACY))),
"the inner column must stay a digest column"
);
}
}
#[test]
fn frame_round_trip_keeps_a_duration_digest_column_without_nones() {
let data = frame(digest_column(
vec![Some(duration_digest(&[1, 2, 3])), Some(duration_digest(&[-7, 90_000]))],
ValueType::Duration,
));
let bytes = encode_frames(from_ref(&data), &EncodeOptions::default()).unwrap();
let decoded = decode_frames(&bytes).unwrap();
assert_same_cells(&view(&data), &view(&decoded[0]));
assert_eq!(view(&decoded[0]).get_type(), digest_type(ValueType::Duration, ACCURACY));
}
#[test]
fn a_decoded_frame_renders_a_digest_as_its_count() {
let data = digest_column(vec![Some(float_digest(&[1.0, 2.0, 3.0])), None], ValueType::Float8);
let bytes = encode_frames(&[frame(data)], &EncodeOptions::default()).unwrap();
let rendered = decode_frames(&bytes).unwrap()[0].to_string();
assert!(rendered.contains("digest(n: 3)"), "rendered frame lacks the digest count:\n{rendered}");
assert!(rendered.contains("none"), "rendered frame lacks the none row:\n{rendered}");
}
#[test]
fn frame_decode_rejects_a_row_digest_whose_params_differ_from_the_column() {
let data = digest_column(vec![Some(float_digest(&[1.0]))], ValueType::Float8);
let mut bytes = encode_frames(&[frame(data)], &EncodeOptions::default()).unwrap();
let column_params = [3, 0x10, 0x27, 0, 0];
let at = find(&bytes, &column_params);
assert_eq!(at.len(), 1, "column params must appear exactly once in the frame");
bytes[at[0] + 1..at[0] + 5].copy_from_slice(&20_000u32.to_le_bytes());
assert!(decode_frames(&bytes).is_err(), "a mismatched row digest was accepted");
}
#[test]
fn frame_decode_rejects_digest_column_params_that_are_not_a_valid_digest_type() {
let data = digest_column(vec![Some(float_digest(&[1.0]))], ValueType::Float8);
let mut bytes = encode_frames(&[frame(data)], &EncodeOptions::default()).unwrap();
let at = find(&bytes, &[3, 0x10, 0x27, 0, 0]);
assert_eq!(at.len(), 1);
bytes[at[0]] = 9;
assert!(decode_frames(&bytes).is_err(), "a utf8 digest column was accepted");
}
#[test]
fn json_round_trip_keeps_digests_nones_and_column_params() {
let data = frame(digest_column(sample_rows(), ValueType::Float8));
let json = frames_to_json(from_ref(&data)).unwrap();
let decoded = frames_from_json(&json).unwrap();
let column = view(&decoded[0]);
assert_same_cells(&view(&data), &column);
assert_eq!(column.get_type(), view(&data).get_type());
}
#[test]
fn json_wire_type_names_digest_with_inner_type_and_accuracy() {
let ty = digest_type(ValueType::Duration, ACCURACY);
let rendered = to_json(&ty);
assert_eq!(rendered, json!({"id": "Digest", "underlying": {"id": "Duration"}, "accuracy": 10000}));
assert_eq!(from_json(&rendered), Ok(ty));
let option = ValueType::Option(Box::new(digest_type(ValueType::Int4, 1_000)));
assert_eq!(from_json(&to_json(&option)), Ok(option));
}
#[test]
fn json_wire_type_rejects_a_digest_without_valid_params() {
assert!(from_json(&json!({"id": "Digest", "underlying": {"id": "Float8"}})).is_err());
assert!(from_json(&json!({"id": "Digest", "underlying": {"id": "Float8"}, "accuracy": -1})).is_err());
assert!(from_json(&json!({"id": "Digest", "underlying": {"id": "Float8"}, "accuracy": 5_000_000_000u64}))
.is_err());
assert!(from_json(&json!({"id": "Digest", "underlying": {"id": "Utf8"}, "accuracy": 10000})).is_err());
assert!(from_json(&json!({"id": "Digest", "underlying": {"id": "Float8"}, "accuracy": 500_000})).is_err());
}
#[test]
fn key_serializer_returns_an_error_for_a_digest_instead_of_panicking() {
let mut serializer = KeySerializer::new();
let err = serializer.try_extend_value(&Value::Digest(Box::new(float_digest(&[1.0])))).err().unwrap();
assert_eq!(err.diagnostic().code, "SERDE_003");
let mut serializer = KeySerializer::new();
let none = Value::none_of(digest_type(ValueType::Float8, ACCURACY));
let err = serializer.try_extend_value(&none).err().unwrap();
assert_eq!(err.diagnostic().code, "SERDE_003");
let mut serializer = KeySerializer::new();
let nested = Value::List(vec![Value::Int4(1), Value::Digest(Box::new(float_digest(&[1.0])))]);
assert!(serializer.try_extend_value(&nested).is_err(), "a digest inside a list reached the key");
}