akar_processor/processor/
chunk_helpers.rs

1use akar_common::types::{PhysicalTypeID, Value};
2use akar_common::vector::ValueVector;
3
4pub fn extract_all_rows_from_chunks(chunks: &[akar_common::vector::DataChunk]) -> Vec<Vec<Value>> {
5    let mut all_rows = Vec::new();
6    for chunk in chunks {
7        let rows = extract_all_rows(chunk);
8        all_rows.extend(rows);
9    }
10    all_rows
11}
12
13pub fn extract_all_rows(chunk: &akar_common::vector::DataChunk) -> Vec<Vec<Value>> {
14    let num_rows = chunk.size;
15    let num_cols = chunk.fields.len();
16    let mut rows: Vec<Vec<Value>> = Vec::with_capacity(num_rows);
17    for row in 0..num_rows {
18        let mut values: Vec<Value> = Vec::with_capacity(num_cols);
19        for col in 0..num_cols {
20            let val = chunk.get_value(col, row).unwrap_or(Value::Null);
21            values.push(val);
22        }
23        rows.push(values);
24    }
25    rows
26}
27
28pub fn rows_to_columns(rows: &[Vec<Value>]) -> (Vec<arrow::array::ArrayRef>, Vec<PhysicalTypeID>) {
29    if rows.is_empty() {
30        return (Vec::new(), Vec::new());
31    }
32    let num_cols = rows[0].len();
33    let num_rows = rows.len();
34
35    let mut fields = Vec::with_capacity(num_cols);
36    let mut field_types = Vec::with_capacity(num_cols);
37
38    for col in 0..num_cols {
39        // Infer the column type from the first non-null value so that columns
40        // whose leading rows are null (e.g. OPTIONAL MATCH null padding) still
41        // carry their real values instead of being dropped as untypable.
42        let first_val = rows
43            .iter()
44            .map(|r| &r[col])
45            .find(|v| !matches!(v, Value::Null))
46            .unwrap_or(&rows[0][col]);
47        let phys_type = value_to_physical_type(first_val);
48        let mut vec = ValueVector::new(phys_type, num_rows.max(1));
49        for (row_idx, row) in rows.iter().enumerate() {
50            let val = &row[col];
51            let _ = vec.set_value(row_idx, val);
52        }
53        vec.resize(num_rows);
54        fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&vec).array);
55        field_types.push(phys_type);
56    }
57    (fields, field_types)
58}
59
60pub fn value_to_physical_type(val: &Value) -> PhysicalTypeID {
61    match val {
62        Value::Null => PhysicalTypeID::Any,
63        Value::Bool(_) => PhysicalTypeID::Bool,
64        Value::Int64(_) => PhysicalTypeID::Int64,
65        Value::Int32(_) => PhysicalTypeID::Int32,
66        Value::Int16(_) => PhysicalTypeID::Int16,
67        Value::Int8(_) => PhysicalTypeID::Int8,
68        Value::UInt64(_) => PhysicalTypeID::UInt64,
69        Value::UInt32(_) => PhysicalTypeID::UInt32,
70        Value::UInt16(_) => PhysicalTypeID::UInt16,
71        Value::UInt8(_) => PhysicalTypeID::UInt8,
72        Value::Int128(_) => PhysicalTypeID::Int128,
73        Value::Double(_) => PhysicalTypeID::Double,
74        Value::Float(_) => PhysicalTypeID::Float,
75        Value::String(_) | Value::Blob(_) => PhysicalTypeID::String,
76        Value::Date(_)
77        | Value::Timestamp(_)
78        | Value::TimestampTz(_)
79        | Value::TimestampNs(_)
80        | Value::TimestampMs(_)
81        | Value::TimestampSec(_)
82        | Value::Interval(_) => PhysicalTypeID::Int64,
83        Value::InternalID(_) => PhysicalTypeID::Int64,
84        Value::UInt128(_) => PhysicalTypeID::Int128,
85        Value::Json(_) => PhysicalTypeID::String,
86        Value::DTime(_) => PhysicalTypeID::Int64,
87        Value::Union(_, _) => PhysicalTypeID::Struct,
88        Value::List(_) => PhysicalTypeID::List,
89        Value::Map(_) => PhysicalTypeID::Struct,
90        Value::Struct(_) => PhysicalTypeID::Struct,
91    }
92}
akar_processor/processor/chunk_helpers.rs

akar_processor/processor/
chunk_helpers.rs
