timeseries-table-format 0.5.0

Append-only time-series table format with gap/overlap tracking
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

//! Integration test for latest snapshot helpers on TimeSeriesTable.
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

use crate::coverage::EntityIdentity;
use crate::metadata::logical_schema::{
    LogicalDataType, LogicalField, LogicalSchema, LogicalTimestampUnit,
};
use crate::metadata::segments::{FileFormat, SegmentEntityLayout, SegmentMeta};
use crate::metadata::{
    index::{IndexKind, IndexSpec, TimeIndexGranularity},
    table::TableMeta,
};
use crate::storage::TableLocation;
use crate::table::TimeSeriesTable;
use crate::transaction_log::{LogAction, TransactionLogStore};
use chrono::{TimeZone, Utc};
use tempfile::TempDir;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn make_basic_table_meta() -> TableMeta {
    let index = IndexSpec {
        column: "ts".to_string(),
        entity_columns: vec!["symbol".to_string()],
        kind: IndexKind::Timestamp {
            index_granularity: TimeIndexGranularity::Minutes(1),
            timezone: None,
        },
    };

    TableMeta::new_time_series_with_schema(
        index,
        LogicalSchema::new(vec![
            LogicalField {
                name: "ts".to_string(),
                data_type: LogicalDataType::Timestamp {
                    unit: LogicalTimestampUnit::Millis,
                    timezone: None,
                },
                nullable: false,
            },
            LogicalField {
                name: "symbol".to_string(),
                data_type: LogicalDataType::Utf8,
                nullable: false,
            },
        ])
        .expect("valid latest-snapshot test schema"),
    )
}

#[tokio::test]
async fn load_latest_state_sees_new_commits() -> TestResult {
    let tmp = TempDir::new()?;
    let location = TableLocation::local(tmp.path());

    let meta = make_basic_table_meta();
    let _writer = TimeSeriesTable::create(location.clone(), meta).await?;

    // Open a "stale" handle (version == 1).
    let stale = TimeSeriesTable::open(location.clone()).await?;

    // Commit a new segment directly through the log (version 2).
    let log = TransactionLogStore::new(location.clone());
    let seg = SegmentMeta {
        path: "data/seg_0001.parquet".to_string(),
        format: FileFormat::Parquet,
        entity_layout: SegmentEntityLayout::Single(EntityIdentity::try_new(vec!["A".into()])?),
        index_min: (Utc.timestamp_opt(10, 0).single().unwrap()).into(),
        index_max: (Utc.timestamp_opt(20, 0).single().unwrap()).into(),
        row_count: 1,
        file_size: None,
        coverage_path: None,
    };

    log.commit_with_expected_version(1, vec![LogAction::AddSegment(seg.clone())])
        .await?;

    // Stale in-memory snapshot is still old.
    assert_eq!(stale.state().version, 1);

    // But latest helpers see new version and segment.
    let v = stale.current_version().await?;
    assert_eq!(v, 2);

    let latest = stale.load_latest_state().await?;
    assert_eq!(latest.version, 2);
    assert!(latest.segments.contains_key(&seg.path));
    let latest_seg = latest.segments.get(&seg.path).expect("segment present");
    assert_eq!(latest_seg.index_min, seg.index_min);
    assert_eq!(latest_seg.index_max, seg.index_max);
    assert!(latest.table_coverage.is_none());

    Ok(())
}

#[tokio::test]
async fn load_latest_state_no_change_returns_current_snapshot() -> TestResult {
    let tmp = TempDir::new()?;
    let location = TableLocation::local(tmp.path());

    let meta = make_basic_table_meta();
    let table = TimeSeriesTable::create(location.clone(), meta).await?;

    let v = table.current_version().await?;
    assert_eq!(v, table.state().version);

    let latest = table.load_latest_state().await?;
    assert_eq!(latest.version, table.state().version);
    assert!(latest.segments.is_empty());
    assert!(latest.table_coverage.is_none());
    match latest.table_meta.kind() {
        crate::metadata::table::TableKind::TimeSeries(_) => {}
        other => panic!("expected time series table kind, got {other:?}"),
    }

    Ok(())
}