akar_processor/physical/
common.rs

1//! Common utility functions used across physical operators.
2
3use akar_common::types::{PhysicalTypeID, Value};
4use akar_common::vector::ValueVector;
5
6#[inline]
7pub(crate) fn store_value_in_vector(v: &mut ValueVector, row: usize, val: &Value) -> Result<(), String> {
8    match val {
9        Value::Null => {
10            v.set_null(row, true);
11        }
12        Value::Bool(x) => {
13            if v.physical_type() == PhysicalTypeID::Bool {
14                v.data_mut()[row] = if *x { 1 } else { 0 };
15                v.set_null(row, false);
16            }
17        }
18        Value::Int64(x) => {
19            let offset = row * 8;
20            if offset + 8 <= v.data().len() {
21                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
22                v.set_null(row, false);
23            }
24        }
25        Value::UInt64(x) => {
26            let offset = row * 8;
27            if offset + 8 <= v.data().len() {
28                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
29                v.set_null(row, false);
30            }
31        }
32        Value::Int32(x) => {
33            let offset = row * 4;
34            if offset + 4 <= v.data().len() {
35                v.data_mut()[offset..offset + 4].copy_from_slice(&x.to_le_bytes());
36                v.set_null(row, false);
37            }
38        }
39        Value::Double(x) => {
40            let offset = row * 8;
41            if offset + 8 <= v.data().len() {
42                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
43                v.set_null(row, false);
44            }
45        }
46        Value::Float(x) => {
47            let offset = row * 4;
48            if offset + 4 <= v.data().len() {
49                v.data_mut()[offset..offset + 4].copy_from_slice(&x.to_le_bytes());
50                v.set_null(row, false);
51            }
52        }
53        Value::String(s) => {
54            let bytes = s.as_bytes();
55            if bytes.len() > 255 {
56                return Err(format!(
57                    "Cannot store string of {} bytes: inline string storage limit is 255 bytes",
58                    bytes.len()
59                ));
60            }
61            let offset = row * 256;
62            if offset < v.data().len() {
63                v.data_mut()[offset] = bytes.len() as u8;
64                if offset + 1 + bytes.len() <= v.data().len() {
65                    v.data_mut()[offset + 1..offset + 1 + bytes.len()].copy_from_slice(bytes);
66                }
67                v.set_null(row, false);
68            }
69        }
70        _ => {
71            v.set_null(row, true);
72        }
73    }
74    Ok(())
75}
76
77#[inline(always)]
78pub(crate) fn value_cmp(a: &Value, b: &Value) -> std::cmp::Ordering {
79    match (a, b) {
80        (Value::Null, Value::Null) => std::cmp::Ordering::Equal,
81        (Value::Null, _) => std::cmp::Ordering::Greater,
82        (_, Value::Null) => std::cmp::Ordering::Less,
83        (Value::Bool(x), Value::Bool(y)) => x.cmp(y),
84        (Value::Int64(x), Value::Int64(y)) => x.cmp(y),
85        (Value::Int32(x), Value::Int32(y)) => x.cmp(y),
86        (Value::Int16(x), Value::Int16(y)) => (*x as i64).cmp(&(*y as i64)),
87        (Value::Int8(x), Value::Int8(y)) => (*x as i64).cmp(&(*y as i64)),
88        (Value::UInt64(x), Value::UInt64(y)) => x.cmp(y),
89        (Value::UInt32(x), Value::UInt32(y)) => (*x as u64).cmp(&(*y as u64)),
90        (Value::Double(x), Value::Double(y)) => double_cmp(*x, *y),
91        (Value::Float(x), Value::Float(y)) => double_cmp(*x as f64, *y as f64),
92        (Value::String(x), Value::String(y)) => x.cmp(y),
93        (Value::Date(x), Value::Date(y)) => x.0.cmp(&y.0),
94        (Value::Timestamp(x), Value::Timestamp(y)) => x.0.cmp(&y.0),
95        _ => std::cmp::Ordering::Equal,
96    }
97}
98
99/// Total-order comparison for floats with a NaN convention: NaN sorts greater
100/// than every finite value, and NaN == NaN. `partial_cmp(...).unwrap_or(Equal)`
101/// would treat NaN as equal to everything, breaking ORDER BY / TOP-K sort order.
102#[inline(always)]
103pub(crate) fn double_cmp(a: f64, b: f64) -> std::cmp::Ordering {
104    if a.is_nan() {
105        if b.is_nan() {
106            std::cmp::Ordering::Equal
107        } else {
108            std::cmp::Ordering::Greater
109        }
110    } else if b.is_nan() {
111        std::cmp::Ordering::Less
112    } else {
113        a.partial_cmp(&b).unwrap_or(std::cmp::Ordering::Equal)
114    }
115}
116
117#[inline]
118pub(crate) fn value_hash(val: &Value) -> u64 {
119    use std::hash::Hasher;
120    let mut hasher = std::collections::hash_map::DefaultHasher::new();
121    hash_value_into(val, &mut hasher);
122    hasher.finish()
123}
124
125/// Write a Value's hash into an arbitrary Hasher.
126#[inline]
127pub(crate) fn hash_value_into(val: &Value, hasher: &mut impl std::hash::Hasher) {
128    use std::hash::Hash;
129    match val {
130        Value::Null => 0u8.hash(hasher),
131        Value::Bool(b) => b.hash(hasher),
132        Value::Int64(i) => i.hash(hasher),
133        Value::Int32(i) => i.hash(hasher),
134        Value::Int16(i) => i.hash(hasher),
135        Value::Int8(i) => i.hash(hasher),
136        Value::UInt64(i) => i.hash(hasher),
137        Value::UInt32(i) => i.hash(hasher),
138        Value::UInt16(i) => (*i as u64).hash(hasher),
139        Value::UInt8(i) => (*i as u64).hash(hasher),
140        Value::Int128(i) => i.hash(hasher),
141        Value::Double(f) => f.to_bits().hash(hasher),
142        Value::Float(f) => f.to_bits().hash(hasher),
143        Value::String(s) => s.hash(hasher),
144        Value::Blob(b) => b.hash(hasher),
145        Value::Date(d) => d.0.hash(hasher),
146        Value::Timestamp(t) => t.0.hash(hasher),
147        Value::TimestampTz(t) => t.0.hash(hasher),
148        Value::TimestampNs(t) => t.0.hash(hasher),
149        Value::TimestampMs(t) => t.0.hash(hasher),
150        Value::TimestampSec(t) => t.0.hash(hasher),
151        Value::Interval(i) => (i.months, i.days, i.micros).hash(hasher),
152        Value::InternalID(id) => id.offset.hash(hasher),
153        Value::UInt128(i) => i.hash(hasher),
154        Value::List(vals) => {
155            for v in vals {
156                hash_value_into(v, hasher);
157            }
158        }
159        Value::Map(kvs) => {
160            for (k, v) in kvs {
161                hash_value_into(k, hasher);
162                hash_value_into(v, hasher);
163            }
164        }
165        Value::Union(_, v) => hash_value_into(v, hasher),
166        _ => std::mem::discriminant(val).hash(hasher),
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    #[test]
175    fn store_value_in_vector_string_overflow_returns_error() {
176        let mut v = ValueVector::new(PhysicalTypeID::String, 1);
177        let err = store_value_in_vector(&mut v, 0, &Value::String("a".repeat(256))).unwrap_err();
178        assert!(err.contains("255"), "err: {err}");
179    }
180
181    #[test]
182    fn store_value_in_vector_string_255_round_trips() {
183        let mut v = ValueVector::new(PhysicalTypeID::String, 1);
184        v.resize(1);
185        let s = "b".repeat(255);
186        store_value_in_vector(&mut v, 0, &Value::String(s.clone())).unwrap();
187        assert_eq!(v.get_value(0), Some(Value::String(s)));
188    }
189}
akar_processor/physical/common.rs

akar_processor/physical/
common.rs
