chronix_query/
lib.rs

1//! # chronix-query
2//!
3//! Query engine for the Chronix time-series database.
4//!
5//! This crate implements the full query pipeline:
6//!
7//! 1. **Planning** — [`QueryPlan`] and [`QueryBuilder`] for fluent query construction
8//! 2. **Pruning** — Multi-level segment elimination (time, bloom, stats)
9//! 3. **Projection** — Read only the columns needed by the query
10//! 4. **Filtering** — Vectorized predicate evaluation on Arrow arrays
11//! 5. **Deduplication** — Sort-merge dedup for overlapping segments
12//! 6. **Aggregation** — count, sum, min, max, avg, first, last
13//! 7. **Downsampling** — Time-bucket aggregation
14//! 8. **Window functions** — row_number, rank, lead, lag, delta, rate, irate,
15//!    moving average, cumulative sum, increase
16//!
17//! All results are returned as Apache Arrow `RecordBatch` for zero-copy
18//! interoperability with data processing ecosystems.
19
20#![warn(missing_docs)]
21#![deny(unsafe_code)]
22#![allow(clippy::module_name_repetitions)]
23
24pub mod aggregate;
25pub mod convert;
26pub mod dedup;
27pub mod downsample;
28pub mod encoded_pushdown;
29pub mod error;
30pub mod filter;
31pub mod memory;
32pub mod plan;
33pub mod pruning;
34pub mod window;
35
36pub use aggregate::AggFn;
37pub use aggregate::{AggResult, DecimalTrack, Num};
38pub use convert::points_to_record_batch;
39pub use encoded_pushdown::{ColumnPredicate, EncodedDomainEvaluator, RowGroupVerdict};
40pub use error::QueryError;
41pub use memory::MemoryTracker;
42pub use plan::{FieldPredicate, QueryBuilder, QueryPlan, StatsOp};
43pub use window::WindowFn;
44
45/// The Arrow type a schema column decodes to — one answer, in one place.
46///
47/// It was four copies of this match (the scan schema, the SQL table
48/// provider, the server's schema endpoint, the segment reader), each ending
49/// in a catch-all that mapped anything it did not name to a string. That is
50/// harmless while every unnamed type *is* a string, and turns a decimal
51/// column into `Utf8` the moment one is not.
52///
53/// `Timestamp` maps to `Int64`, the storage representation. The SQL layer
54/// presents the time column as `Timestamp(Nanosecond)` and converts on the
55/// way out; see `chronix::sql`.
56#[must_use]
57pub fn column_type_to_arrow(ct: chronix_core::ColumnType) -> arrow::datatypes::DataType {
58    use arrow::datatypes::DataType;
59    use chronix_core::ColumnType;
60    match ct {
61        ColumnType::Timestamp | ColumnType::I64 => DataType::Int64,
62        ColumnType::U64 => DataType::UInt64,
63        ColumnType::F64 => DataType::Float64,
64        ColumnType::Bool => DataType::Boolean,
65        ColumnType::String => DataType::Utf8,
66        // A native histogram is a composite sample with no Arrow equivalent:
67        // the value is a `postcard` blob the `histogram_*` functions decode.
68        // `Binary` keeps the column's Arrow type stable regardless of the
69        // histogram's resolution, which a struct-of-lists encoding would not.
70        ColumnType::Histogram => DataType::Binary,
71        ColumnType::Decimal { scale } => DataType::Decimal128(
72            chronix_core::DECIMAL_PRECISION,
73            i8::try_from(scale).unwrap_or(0),
74        ),
75    }
76}
77
78/// The inverse of [`column_type_to_arrow`], for one cell.
79///
80/// `Ok(None)` means the cell is null. `Err` means the column's Arrow type is
81/// not one the storage layer can hold — which is a *schema* mistake by the
82/// caller, not a value, and must not be mistaken for "no data".
83///
84/// Callers *store* what this produces — a distributed read, and a Raft region
85/// snapshot — so it is one total conversion rather than a `match` at each
86/// site. A quiet arm there drops a field from a point with nothing to say so.
87///
88/// # Errors
89///
90/// Returns [`QueryError::Validation`] when the Arrow type has no
91/// [`ColumnType`](chronix_core::ColumnType), or when a decimal carries a
92/// negative scale — legal in Arrow, unrepresentable in storage.
93pub fn arrow_cell_to_field_value(
94    col: &dyn arrow::array::Array,
95    index: usize,
96) -> error::Result<Option<chronix_core::FieldValue>> {
97    use arrow::array::{
98        BooleanArray, Decimal128Array, Float64Array, Int64Array, StringArray, UInt64Array,
99    };
100    use arrow::datatypes::DataType;
101    use chronix_core::FieldValue;
102
103    if col.is_null(index) {
104        return Ok(None);
105    }
106
107    let bad = |what: &str| {
108        QueryError::Validation(format!(
109            "column has {what}, which no Chronix column type can hold"
110        ))
111    };
112    let mismatch = || bad("an array that disagrees with its own Arrow type");
113
114    // The six storage types, and nothing else. `Timestamp` shares `Int64`
115    // with `I64`; a timestamp *field* is an `I64` value here, which is what
116    // the schema layer stores it as.
117    let v = match col.data_type() {
118        DataType::Float64 => FieldValue::F64(
119            col.as_any()
120                .downcast_ref::<Float64Array>()
121                .ok_or_else(mismatch)?
122                .value(index),
123        ),
124        DataType::Int64 => FieldValue::I64(
125            col.as_any()
126                .downcast_ref::<Int64Array>()
127                .ok_or_else(mismatch)?
128                .value(index),
129        ),
130        DataType::UInt64 => FieldValue::U64(
131            col.as_any()
132                .downcast_ref::<UInt64Array>()
133                .ok_or_else(mismatch)?
134                .value(index),
135        ),
136        DataType::Boolean => FieldValue::Bool(
137            col.as_any()
138                .downcast_ref::<BooleanArray>()
139                .ok_or_else(mismatch)?
140                .value(index),
141        ),
142        DataType::Utf8 => FieldValue::String(
143            col.as_any()
144                .downcast_ref::<StringArray>()
145                .ok_or_else(mismatch)?
146                .value(index)
147                .to_owned(),
148        ),
149        DataType::Decimal128(_, scale) => {
150            let arr = col
151                .as_any()
152                .downcast_ref::<Decimal128Array>()
153                .ok_or_else(mismatch)?;
154            let scale = u8::try_from(*scale).map_err(|_| bad("a decimal with a negative scale"))?;
155            FieldValue::Decimal(
156                chronix_core::Decimal::new(arr.value(index), scale)
157                    .map_err(|e| QueryError::Validation(format!("decimal cell: {e}")))?,
158            )
159        }
160        other => return Err(bad(&format!("Arrow type {other}"))),
161    };
162    Ok(Some(v))
163}
164
165/// Consolidated Arrow column → `f64` extraction.
166///
167/// Supports `Float64Array`, `Int64Array`, `UInt64Array` and
168/// `Decimal128Array`. Returns `None` for null values or unsupported types.
169///
170/// # The one place a decimal becomes approximate
171///
172/// Storing or aggregating a `Decimal128` column keeps it exact: the segment
173/// holds the mantissa, `sum`/`min`/`max`/`first`/`last` fold in `i128`, and
174/// SQL sees `Decimal128(38, s)`.
175///
176/// What comes through here is the other kind of question — a moving average,
177/// a percentile, a forecast, an anomaly score, a chart's downsample, a
178/// PromQL sample. Those are defined in floating point and their answers are
179/// estimates whatever the input was, so converting is honest. Keeping it in
180/// one named function leaves one boundary to audit rather than an `as f64`
181/// in every module.
182#[allow(clippy::cast_precision_loss)]
183pub fn extract_f64(col: &dyn arrow::array::Array, index: usize) -> Option<f64> {
184    use arrow::array::{Decimal128Array, Float64Array, Int64Array, UInt64Array};
185    if col.is_null(index) {
186        return None;
187    }
188    col.as_any()
189        .downcast_ref::<Float64Array>()
190        .map(|a| a.value(index))
191        .or_else(|| {
192            col.as_any()
193                .downcast_ref::<Int64Array>()
194                .map(|a| a.value(index) as f64)
195        })
196        .or_else(|| {
197            col.as_any()
198                .downcast_ref::<UInt64Array>()
199                .map(|a| a.value(index) as f64)
200        })
201        .or_else(|| {
202            col.as_any().downcast_ref::<Decimal128Array>().map(|a| {
203                let scale = u32::try_from(a.scale()).unwrap_or(0);
204                a.value(index) as f64 / chronix_core::pow10(scale).unwrap_or(1) as f64
205            })
206        })
207}
208
209#[cfg(test)]
210#[allow(clippy::unwrap_used, clippy::expect_used)]
211mod cell_tests {
212    use super::{arrow_cell_to_field_value, column_type_to_arrow};
213    use arrow::array::{
214        BooleanArray, Decimal128Array, Float64Array, Int64Array, StringArray, UInt64Array,
215    };
216    use chronix_core::{ColumnType, FieldValue};
217
218    #[test]
219    fn every_storage_type_survives_the_round_trip() {
220        // The property that matters: whatever `column_type_to_arrow` can
221        // produce, this can read back. A type missing here is a field that
222        // disappears from a replicated region.
223        let cases: Vec<(ColumnType, arrow::array::ArrayRef, FieldValue)> = vec![
224            (
225                ColumnType::F64,
226                std::sync::Arc::new(Float64Array::from(vec![1.5])),
227                FieldValue::F64(1.5),
228            ),
229            (
230                ColumnType::I64,
231                std::sync::Arc::new(Int64Array::from(vec![i64::MIN])),
232                FieldValue::I64(i64::MIN),
233            ),
234            (
235                ColumnType::U64,
236                std::sync::Arc::new(UInt64Array::from(vec![u64::MAX])),
237                FieldValue::U64(u64::MAX),
238            ),
239            (
240                ColumnType::Bool,
241                std::sync::Arc::new(BooleanArray::from(vec![true])),
242                FieldValue::Bool(true),
243            ),
244            (
245                ColumnType::String,
246                std::sync::Arc::new(StringArray::from(vec!["s"])),
247                FieldValue::String("s".to_string()),
248            ),
249            (
250                ColumnType::Decimal { scale: 4 },
251                std::sync::Arc::new(
252                    Decimal128Array::from(vec![12_345_i128])
253                        .with_precision_and_scale(chronix_core::DECIMAL_PRECISION, 4)
254                        .unwrap(),
255                ),
256                FieldValue::Decimal(chronix_core::Decimal::new(12_345, 4).unwrap()),
257            ),
258        ];
259
260        for (ct, arr, want) in cases {
261            assert_eq!(
262                arr.data_type(),
263                &column_type_to_arrow(ct),
264                "test array does not match {ct:?}"
265            );
266            let got = arrow_cell_to_field_value(arr.as_ref(), 0).unwrap();
267            assert_eq!(got, Some(want), "round trip failed for {ct:?}");
268        }
269    }
270
271    #[test]
272    fn a_null_cell_is_absent_not_an_error() {
273        let arr = Float64Array::from(vec![None::<f64>]);
274        assert_eq!(arrow_cell_to_field_value(&arr, 0).unwrap(), None);
275    }
276
277    #[test]
278    fn a_type_storage_cannot_hold_is_an_error_not_a_silent_drop() {
279        // The defect this exists for: `_ => None` read as "no value here"
280        // for a column that in fact held one.
281        let arr = arrow::array::Int32Array::from(vec![1]);
282        let err = arrow_cell_to_field_value(&arr, 0).unwrap_err();
283        assert!(
284            err.to_string().contains("Int32"),
285            "the error should name the type: {err}"
286        );
287    }
288
289    #[test]
290    fn a_negative_scale_is_an_error_not_a_dropped_column() {
291        // Legal in Arrow, unrepresentable in storage. It used to be read
292        // into a `u8`, fail, and take the whole cell with it.
293        let arr = Decimal128Array::from(vec![123_i128])
294            .with_precision_and_scale(10, -2)
295            .unwrap();
296        let err = arrow_cell_to_field_value(&arr, 0).unwrap_err();
297        assert!(
298            err.to_string().contains("negative scale"),
299            "the error should say why: {err}"
300        );
301    }
302}
chronix_query/lib.rs

chronix_query/
lib.rs
