re_data_store 0.14.1

An in-memory time series database for Rerun log data, based on Apache Arrow
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

//! Tests running assertions on the internal state of the datastore.
//!
//! They're awful, but sometimes you just have to…

use re_data_store::{DataStore, DataStoreConfig};
use re_log_types::{build_frame_nr, DataRow, EntityPath, RowId, TimePoint};
use re_types::{components::InstanceKey, datagen::build_some_instances};
use re_types_core::Loggable as _;

// --- Internals ---

// TODO(cmc): One should _never_ run assertions on the internal state of the datastore, this
// is a recipe for disaster.
//
// The contract that needs to be asserted here, from the point of view of the actual user,
// is performance: getting the datastore into a pathological topology should show up in
// integration query benchmarks.
//
// In the current state of things, though, it is much easier to test for it that way… so we
// make an exception, for now…
#[test]
fn pathological_bucket_topology() {
    re_log::setup_logging();

    let mut store_forward = DataStore::new(
        re_log_types::StoreId::random(re_log_types::StoreKind::Recording),
        InstanceKey::name(),
        DataStoreConfig {
            indexed_bucket_num_rows: 10,
            ..Default::default()
        },
    );
    let mut store_backward = DataStore::new(
        re_log_types::StoreId::random(re_log_types::StoreKind::Recording),
        InstanceKey::name(),
        DataStoreConfig {
            indexed_bucket_num_rows: 10,
            ..Default::default()
        },
    );

    fn store_repeated_frame(
        frame_nr: i64,
        num: usize,
        store_forward: &mut DataStore,
        store_backward: &mut DataStore,
    ) {
        let ent_path = EntityPath::from("this/that");
        let num_instances = 1;

        let timepoint = TimePoint::from([build_frame_nr(frame_nr.into())]);
        for _ in 0..num {
            let row = DataRow::from_cells1_sized(
                RowId::new(),
                ent_path.clone(),
                timepoint.clone(),
                num_instances,
                build_some_instances(num_instances as _),
            )
            .unwrap();
            store_forward.insert_row(&row).unwrap();

            let row = DataRow::from_cells1_sized(
                RowId::new(),
                ent_path.clone(),
                timepoint.clone(),
                num_instances,
                build_some_instances(num_instances as _),
            )
            .unwrap();
            store_backward.insert_row(&row).unwrap();
        }
    }

    fn store_frame_range(
        range: core::ops::RangeInclusive<i64>,
        store_forward: &mut DataStore,
        store_backward: &mut DataStore,
    ) {
        let ent_path = EntityPath::from("this/that");
        let num_instances = 1;

        let rows = range
            .map(|frame_nr| {
                let timepoint = TimePoint::from([build_frame_nr(frame_nr.into())]);
                DataRow::from_cells1_sized(
                    RowId::new(),
                    ent_path.clone(),
                    timepoint,
                    num_instances,
                    build_some_instances(num_instances as _),
                )
                .unwrap()
            })
            .collect::<Vec<_>>();

        for row in &rows {
            store_forward.insert_row(row).unwrap();
        }

        rows.iter().rev().for_each(|row| {
            store_backward.insert_row(row).unwrap();
        });
    }

    store_repeated_frame(1000, 10, &mut store_forward, &mut store_backward);
    store_frame_range(970..=979, &mut store_forward, &mut store_backward);
    store_frame_range(990..=999, &mut store_forward, &mut store_backward);
    store_frame_range(980..=989, &mut store_forward, &mut store_backward);
    store_repeated_frame(1000, 7, &mut store_forward, &mut store_backward);
    store_frame_range(1000..=1009, &mut store_forward, &mut store_backward);
    store_repeated_frame(975, 10, &mut store_forward, &mut store_backward);

    {
        let num_buckets = store_forward
            .iter_indices()
            .flat_map(|(_, table)| table.buckets.values())
            .count();
        assert_eq!(
            7usize,
            num_buckets,
            "pathological topology (forward): {}",
            {
                store_forward.sort_indices_if_needed();
                store_forward
            }
        );
    }
    {
        let num_buckets = store_backward
            .iter_indices()
            .flat_map(|(_, table)| table.buckets.values())
            .count();
        assert_eq!(
            8usize,
            num_buckets,
            "pathological topology (backward): {}",
            {
                store_backward.sort_indices_if_needed();
                store_backward
            }
        );
    }
}