chronix 0.7.0

#![allow(clippy::unwrap_used, clippy::expect_used)] // test code may unwrap
//! `last_value()` must return the newest point for a series, including
//! when a late (out-of-order) write has landed in the memtable behind an
//! already-flushed newer point.

use std::collections::BTreeMap;

use chronix::prelude::*;
use chronix::{Chronix, fields, tags};

const HOUR_NS: i64 = 3_600_000_000_000;

fn open_without_lvc(dir: &tempfile::TempDir) -> Chronix {
    // The last-value cache answers correctly on its own; this exercises the
    // memtable/segment fallback that runs whenever the cache misses (cold
    // start, eviction, or a measurement outside `lvc_measurements`).
    let config = ChronixConfig::builder()
        .data_dir(dir.path())
        .enable_last_value_cache(false)
        .build()
        .unwrap();
    Chronix::open(config).unwrap()
}

fn value(db: &Chronix, ts_expected: i64) {
    let tags: BTreeMap<String, String> = BTreeMap::from([("h".to_string(), "a".to_string())]);
    let got = db.last_value("m", &tags).unwrap().expect("a point");
    assert_eq!(
        got.timestamp(),
        ts_expected,
        "last_value returned a stale point"
    );
}

#[test]
fn late_write_does_not_shadow_a_newer_flushed_point() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_without_lvc(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();

    // Newest point, flushed to a segment.
    let newest = 2 * HOUR_NS;
    db.insert(&Point::new(key.clone(), fields! { "v" => 2.0 }, newest).unwrap())
        .unwrap();
    db.flush().unwrap();

    // A late arrival within the out-of-order tolerance stays in the memtable.
    let late = HOUR_NS;
    db.insert(&Point::new(key.clone(), fields! { "v" => 1.0 }, late).unwrap())
        .unwrap();

    value(&db, newest);
}

#[test]
fn memtable_point_wins_when_it_is_actually_newer() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_without_lvc(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();

    db.insert(&Point::new(key.clone(), fields! { "v" => 1.0 }, HOUR_NS).unwrap())
        .unwrap();
    db.flush().unwrap();
    db.insert(&Point::new(key.clone(), fields! { "v" => 2.0 }, 2 * HOUR_NS).unwrap())
        .unwrap();

    value(&db, 2 * HOUR_NS);
}

#[test]
fn segment_only_series_reads_back() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_without_lvc(&dir);
    let key = SeriesKey::new("m", tags! { "h" => "a" }).unwrap();

    db.insert(&Point::new(key.clone(), fields! { "v" => 1.0 }, HOUR_NS).unwrap())
        .unwrap();
    db.insert(&Point::new(key.clone(), fields! { "v" => 2.0 }, 2 * HOUR_NS).unwrap())
        .unwrap();
    db.flush().unwrap();

    value(&db, 2 * HOUR_NS);
}

/// A pre-1970 timestamp is a legal timestamp.
///
/// `Timestamp` is a signed nanosecond epoch and `ShardId::from_timestamp`
/// uses `div_euclid` so that negative values shard correctly — a deliberate
/// choice, made once and then contradicted by the one read path that scanned
/// the memtable from `0`. The result was `last_value()` answering `None` for
/// a series whose rows a query returns.
#[test]
fn last_value_finds_a_pre_epoch_point() {
    let dir = tempfile::tempdir().unwrap();
    let db = chronix::Chronix::open(
        chronix::prelude::ChronixConfig::builder()
            .data_dir(dir.path())
            .build()
            .unwrap(),
    )
    .unwrap();

    let key = chronix::prelude::SeriesKey::new("cpu", chronix::tags! { "host" => "a" }).unwrap();
    let ts = -3_600_000_000_000; // one hour before the epoch
    db.insert(
        &chronix::prelude::Point::new(key.clone(), chronix::fields! { "v" => 7.0 }, ts).unwrap(),
    )
    .unwrap();

    let plan = db
        .query()
        .measurement("cpu")
        .range(i64::MIN, i64::MAX)
        .build()
        .unwrap();
    assert_eq!(
        db.execute(&plan).unwrap().num_rows(),
        1,
        "a query sees the row"
    );

    let last = db
        .last_value(
            "cpu",
            &std::collections::BTreeMap::from([("host".to_owned(), "a".to_owned())]),
        )
        .unwrap()
        .expect("last_value must see it too");
    assert_eq!(last.timestamp(), ts);
    db.close().unwrap();
}

/// Label discovery answers from the memtable as well as from the segments.
///
/// `tag_keys()` and `tag_values()` read the inverted tag index, which is
/// built at **flush**. A freshly started database therefore reported no
/// labels at all, and a tag that only appears in recent data was invisible
/// for as long as it stayed unflushed. The one test that covered these
/// flushed five times first, so it never saw it.
#[test]
fn tag_discovery_sees_unflushed_series() {
    let dir = tempfile::tempdir().unwrap();
    let db = chronix::Chronix::open(
        chronix::prelude::ChronixConfig::builder()
            .data_dir(dir.path())
            .build()
            .unwrap(),
    )
    .unwrap();

    let key =
        chronix::prelude::SeriesKey::new("cpu", chronix::tags! { "host" => "a", "region" => "eu" })
            .unwrap();
    db.insert(&chronix::prelude::Point::new(key, chronix::fields! { "v" => 1.0 }, 1_000).unwrap())
        .unwrap();

    // Nothing has been flushed.
    assert!(db.tag_keys().contains(&"host".to_owned()));
    assert!(db.tag_keys().contains(&"region".to_owned()));
    assert_eq!(db.tag_values("host"), vec!["a".to_owned()]);

    // And the answer does not change when the data moves to a segment.
    db.flush().unwrap();
    assert!(db.tag_keys().contains(&"host".to_owned()));
    assert_eq!(db.tag_values("region"), vec!["eu".to_owned()]);

    // A second series only in memory joins the answer.
    let key2 =
        chronix::prelude::SeriesKey::new("cpu", chronix::tags! { "host" => "b", "region" => "us" })
            .unwrap();
    db.insert(&chronix::prelude::Point::new(key2, chronix::fields! { "v" => 2.0 }, 2_000).unwrap())
        .unwrap();
    assert_eq!(db.tag_values("host"), vec!["a".to_owned(), "b".to_owned()]);
    db.close().unwrap();
}