timelake 0.7.0

The open time-series lakehouse: fast queries on your own object storage, with no database and no maintenance jobs
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

//! Helpers shared by the integration tests: a small IoT schema, generated batches, SQL shortcuts,
//! and plan inspection.

#![allow(dead_code, reason = "each test binary uses a different subset")]
#![allow(clippy::unwrap_used, reason = "test helpers")]

use std::collections::HashMap;
use std::sync::Arc;

use arrow::array::{Array, Float64Array, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
use timelake::datafusion::datasource::memory::MemorySourceConfig;
use timelake::datafusion::datasource::physical_plan::{FileGroup, FileScanConfig};
use timelake::datafusion::datasource::source::DataSourceExec;
use timelake::datafusion::physical_plan::ExecutionPlan;
use timelake::datafusion::physical_plan::empty::EmptyExec;
use timelake::{Lake, Session, TableOptions};

pub(crate) const DAY: i64 = 86_400_000_000;
pub(crate) const T0: i64 = 1_767_225_600_000_000; // 2026-01-01T00:00:00Z

pub(crate) fn schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("time", DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), false),
        Field::new("device", DataType::Int32, true),
        Field::new("site", DataType::Utf8, true),
        Field::new("value", DataType::Float64, true),
    ]))
}

/// `devices` devices, one point every `step` µs from `start`, `n` points each, in time order.
pub(crate) fn batch(start: i64, n: i64, step: i64, devices: i32, value: impl Fn(i32, i64) -> f64) -> RecordBatch {
    let points: Vec<(i64, i32)> = (0..n).flat_map(|i| (0..devices).map(move |d| (i, d))).collect();
    let times = points.iter().map(|&(i, _)| start + i * step);
    RecordBatch::try_new(
        schema(),
        vec![
            Arc::new(TimestampMicrosecondArray::from_iter_values(times).with_timezone("UTC")),
            Arc::new(Int32Array::from_iter_values(points.iter().map(|&(_, d)| d))),
            Arc::new(StringArray::from_iter_values(points.iter().map(|&(_, d)| format!("site-{}", d % 3)))),
            Arc::new(Float64Array::from_iter_values(points.iter().map(|&(i, d)| value(d, i)))),
        ],
    )
    .unwrap()
}

/// A time-series table keyed by device.
pub(crate) fn opts() -> TableOptions {
    TableOptions { time_column: Some("time".into()), series_key: vec!["device".into()], ..Default::default() }
}

/// A lake in a fresh temporary directory (deleted when the guard drops).
pub(crate) fn temp_lake() -> (tempfile::TempDir, Lake) {
    let dir = tempfile::tempdir().unwrap();
    let lake = Lake::open(dir.path().to_str().unwrap(), HashMap::new()).unwrap();
    (dir, lake)
}

pub(crate) async fn rows(s: &Session, sql: &str) -> Vec<RecordBatch> {
    s.sql(sql).await.unwrap().collect().await.unwrap()
}

/// The first value of the first column, as i64.
pub(crate) async fn scalar_i64(s: &Session, sql: &str) -> i64 {
    let b = rows(s, sql).await;
    let c = arrow::compute::cast(b[0].column(0), &DataType::Int64).unwrap();
    c.as_any().downcast_ref::<Int64Array>().unwrap().value(0)
}

/// One string column of a result, concatenated across batches.
pub(crate) fn strings(batches: &[RecordBatch], column: &str) -> Vec<String> {
    batches
        .iter()
        .flat_map(|b| {
            let c = arrow::compute::cast(b.column_by_name(column).unwrap(), &DataType::Utf8).unwrap();
            let c = c.as_any().downcast_ref::<StringArray>().unwrap().clone();
            (0..c.len()).map(move |i| c.value(i).to_string())
        })
        .collect()
}

/// What a scan in a physical plan reads.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Scan {
    /// The last-value index (an in-memory batch).
    LastValueIndex,
    /// This many data files.
    Files(usize),
    /// Nothing: every file was pruned.
    Nothing,
}

/// The scans and operator names of a query's physical plan.
#[derive(Debug)]
pub(crate) struct Plan {
    pub(crate) scans: Vec<Scan>,
    pub(crate) operators: Vec<String>,
}

impl Plan {
    pub(crate) fn has(&self, operator: &str) -> bool {
        self.operators.iter().any(|o| o == operator)
    }
}

pub(crate) async fn plan(s: &Session, sql: &str) -> Plan {
    fn walk(node: &Arc<dyn ExecutionPlan>, plan: &mut Plan) {
        plan.operators.push(node.name().to_string());
        if node.downcast_ref::<EmptyExec>().is_some() {
            plan.scans.push(Scan::Nothing);
        }
        if let Some(exec) = node.downcast_ref::<DataSourceExec>() {
            let source = exec.data_source();
            if let Some(files) = source.downcast_ref::<FileScanConfig>() {
                plan.scans.push(Scan::Files(files.file_groups.iter().map(FileGroup::len).sum()));
            } else if source.downcast_ref::<MemorySourceConfig>().is_some() {
                plan.scans.push(Scan::LastValueIndex);
            }
        }
        for child in node.children() {
            walk(child, plan);
        }
    }
    let physical = s.sql(sql).await.unwrap().create_physical_plan().await.unwrap();
    let mut plan = Plan { scans: vec![], operators: vec![] };
    walk(&physical, &mut plan);
    plan
}

/// The `value` column of a result (or its only column), concatenated across batches.
pub(crate) fn f64_values(batches: &[RecordBatch]) -> Vec<f64> {
    batches
        .iter()
        .flat_map(|b| {
            let c = b.column_by_name("value").unwrap_or_else(|| b.column(0));
            c.as_any().downcast_ref::<Float64Array>().unwrap().values().to_vec()
        })
        .collect()
}