chronix 0.7.0

Chronix — an embedded, analytics-optimised time-series database.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

#![allow(clippy::unwrap_used, clippy::expect_used)] // test code may unwrap
//! D-NULL regression tests: absent fields must read back as SQL `NULL`.
//!
//! Before `.csx` v2 an absent field was stored as a type sentinel (`0`, `""`,
//! `false`) with only an aggregate `null_count` surviving, so `IS NULL` never
//! matched, `= 0` matched rows that had no value, and `AVG`/`MIN`/`COUNT`
//! folded sentinels into their results.

use arrow::array::Array;
use chronix::prelude::*;
use chronix::{Chronix, fields, tags};
use std::sync::Arc;

fn open(dir: &tempfile::TempDir) -> Arc<Chronix> {
    let config = ChronixConfig::builder()
        .data_dir(dir.path())
        .build()
        .unwrap();
    Arc::new(Chronix::open(config).unwrap())
}

async fn rows(ctx: &datafusion::prelude::SessionContext, q: &str) -> usize {
    ctx.sql(q)
        .await
        .unwrap()
        .collect()
        .await
        .unwrap()
        .iter()
        .map(arrow::record_batch::RecordBatch::num_rows)
        .sum()
}

async fn one_f64(ctx: &datafusion::prelude::SessionContext, q: &str) -> Option<f64> {
    let b = ctx.sql(q).await.unwrap().collect().await.unwrap();
    let col = b[0]
        .column(0)
        .as_any()
        .downcast_ref::<arrow::array::Float64Array>()
        .unwrap();
    col.is_valid(0).then(|| col.value(0))
}

#[tokio::test]
async fn absent_numeric_field_is_null_not_zero() {
    let dir = tempfile::tempdir().unwrap();
    let db = open(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();
    db.insert(&Point::new(key.clone(), fields! { "power" => 10.0 }, 1000).unwrap())
        .unwrap();
    db.insert(&Point::new(key.clone(), fields! { "other" => 1.0 }, 2000).unwrap())
        .unwrap();
    db.flush().unwrap();

    let ctx = chronix::sql::create_session_context(db);

    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE power IS NULL").await, 1);
    assert_eq!(
        rows(&ctx, "SELECT * FROM m WHERE power IS NOT NULL").await,
        1
    );
    assert_eq!(
        rows(&ctx, "SELECT * FROM m WHERE power = 0").await,
        0,
        "an absent field must not compare equal to 0"
    );

    // Aggregates must ignore the absent row entirely.
    assert_eq!(one_f64(&ctx, "SELECT avg(power) FROM m").await, Some(10.0));
    assert_eq!(one_f64(&ctx, "SELECT min(power) FROM m").await, Some(10.0));
    assert_eq!(one_f64(&ctx, "SELECT sum(power) FROM m").await, Some(10.0));
    assert_eq!(rows(&ctx, "SELECT power FROM m WHERE power > 5").await, 1);

    let counts = ctx
        .sql("SELECT count(power) FROM m")
        .await
        .unwrap()
        .collect()
        .await
        .unwrap();
    let c = counts[0]
        .column(0)
        .as_any()
        .downcast_ref::<arrow::array::Int64Array>()
        .unwrap()
        .value(0);
    assert_eq!(c, 1, "count() must skip the absent row");
}

#[tokio::test]
async fn absent_string_bool_and_int_fields_are_null() {
    let dir = tempfile::tempdir().unwrap();
    let db = open(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();
    db.insert(
        &Point::new(
            key.clone(),
            fields! { "s" => "hello", "b" => true, "i" => 7_i64 },
            1000,
        )
        .unwrap(),
    )
    .unwrap();
    db.insert(&Point::new(key.clone(), fields! { "z" => 1.0 }, 2000).unwrap())
        .unwrap();
    db.flush().unwrap();

    let ctx = chronix::sql::create_session_context(db);
    for col in ["s", "b", "i"] {
        assert_eq!(
            rows(&ctx, &format!("SELECT * FROM m WHERE {col} IS NULL")).await,
            1,
            "absent {col} must be NULL"
        );
    }
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE s = ''").await, 0);
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE b = false").await, 0);
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE i = 0").await, 0);
}

/// Dense data must be unaffected — and must store no bitmap at all.
#[tokio::test]
async fn dense_data_has_no_nulls() {
    let dir = tempfile::tempdir().unwrap();
    let db = open(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();
    for i in 0..100i64 {
        db.insert(&Point::new(key.clone(), fields! { "v" => i as f64 }, i * 1000).unwrap())
            .unwrap();
    }
    db.flush().unwrap();

    let ctx = chronix::sql::create_session_context(db);
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE v IS NULL").await, 0);
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE v IS NOT NULL").await, 100);
    assert_eq!(one_f64(&ctx, "SELECT sum(v) FROM m").await, Some(4950.0));
}

/// The memtable path always built proper Arrow null buffers while the segment
/// path stored sentinels, so the *same* query returned different answers
/// before and after a flush. `.csx` v2 makes the two agree.
#[tokio::test]
async fn unflushed_and_flushed_data_agree_on_nulls() {
    let dir = tempfile::tempdir().unwrap();
    let db = open(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();
    db.insert(&Point::new(key.clone(), fields! { "power" => 10.0 }, 1000).unwrap())
        .unwrap();
    db.insert(&Point::new(key.clone(), fields! { "other" => 1.0 }, 2000).unwrap())
        .unwrap();

    let ctx = chronix::sql::create_session_context(db.clone());

    // Read from the memtable (nothing flushed yet).
    let before_null = rows(&ctx, "SELECT * FROM m WHERE power IS NULL").await;
    let before_avg = one_f64(&ctx, "SELECT avg(power) FROM m").await;

    db.flush().unwrap();

    // Same queries, now served from a segment.
    let after_null = rows(&ctx, "SELECT * FROM m WHERE power IS NULL").await;
    let after_avg = one_f64(&ctx, "SELECT avg(power) FROM m").await;

    assert_eq!(before_null, after_null, "IS NULL changed across flush");
    assert_eq!(before_avg, after_avg, "avg() changed across flush");
    assert_eq!(after_null, 1);
    assert_eq!(after_avg, Some(10.0));
}

/// Compaction rewrites segments through the batch encode path. Nulls must
/// survive that round-trip, not silently revert to sentinels.
#[tokio::test]
async fn nulls_survive_compaction() {
    let dir = tempfile::tempdir().unwrap();
    let db = open(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();

    // Several small segments, each alternating which field is present.
    for i in 0..8i64 {
        let p = if i % 2 == 0 {
            Point::new(key.clone(), fields! { "power" => i as f64 }, i * 1000).unwrap()
        } else {
            Point::new(key.clone(), fields! { "other" => i as f64 }, i * 1000).unwrap()
        };
        db.insert(&p).unwrap();
        db.flush().unwrap();
    }

    let ctx = chronix::sql::create_session_context(db.clone());
    let before = rows(&ctx, "SELECT * FROM m WHERE power IS NULL").await;
    let before_sum = one_f64(&ctx, "SELECT sum(power) FROM m").await;
    assert_eq!(before, 4, "4 rows have no power field");

    db.compact().unwrap();

    assert_eq!(
        rows(&ctx, "SELECT * FROM m WHERE power IS NULL").await,
        before,
        "compaction lost null information"
    );
    assert_eq!(
        one_f64(&ctx, "SELECT sum(power) FROM m").await,
        before_sum,
        "compaction changed sum(power)"
    );
    assert_eq!(rows(&ctx, "SELECT * FROM m WHERE power = 0").await, 1);
}