polydat 0.3.2

Polydat — a variates construction engine
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

// Toy test definition: a self-describing, hierarchic, parameterized
// dataset and test flow in one Polydat grammar file.
//
// Hierarchy:  cycle -> row -> (tenant, device, reading)
// Parameters: externs with defaults; a host may override any of them.
// Traversal:  a comprehension over phases, intervals, and partitions of
//             the row domain, bound as a producer and traversed below.
// Flow:       schema, load, read, and verify statements derived from
//             the same coordinate, so any row can be regenerated. The
//             load statement carries a JSON document rendered by a tile.

// ---- Coordinates and parameters ---------------------------------------

// The root coordinate. Each traversal activation below has its own
// `cycle`, local to its slice; this one positions the root scope.
input cycle: u64

// Runtime parameters are externs with defaults; a host may assign them
// on the command line without recompiling.
extern base_epoch_ms: u64 = 1700000000000
extern rows_total: u64 = 1000000

// ---- Self-description ------------------------------------------------

const dataset := "iot-readings-toy"
const keyspace := "toy"
const table := "readings"
const schema_stmt := "CREATE TABLE {keyspace}.{table} (tenant_id bigint, device_id text, ts bigint, doc text, PRIMARY KEY ((tenant_id, device_id), ts))"
const shape := "tenants=20 devices_per_tenant=50 readings=unbounded"

// ---- Reusable model ---------------------------------------------------

reading_model(seed: u64) -> (temp_c: f64, humidity: f64, status: str) := {
    temp_c   := normal_sample(input: seed, mean: 21.5, stddev: 2.0)
    humidity := uniform_sample(input: hash(seed), min: 30.0, max: 70.0)
    status   := weighted_strings(hash(hash(seed)), "ok:0.97;degraded:0.02;error:0.01")
}

// ---- Traversal ----------------------------------------------------------

// The test flow is a comprehension bound as a producer: every phase runs
// over every quarter of the row domain, at two reading intervals. Its
// element names are wires inside the traversal below.
flow := for phase in load,verify, interval_ms in 1000,60000, p in partitions("*/4", {rows_total})

for flow {
    // Each activation owns the slice `p` of the row domain. The cursor is
    // narrowed to `p` at activation, and the local cycle maps onto an
    // absolute row ordinal inside the slice.
    cursor rows = range(0, rows_total) over p
    row := mod_in(cycle, rows.cursor)

    // Structural parameters are literal constants in the decomposition.
    // The first two dimensions are bounded; the third is unbounded, so
    // the dataset grows with the row count.
    (tenant, device, reading) := mixed_radix(row, 20, 50, 0)

    // ---- Entities -----------------------------------------------------

    tenant_id   := hashed_id(input: tenant, bound: 1000000)
    device_key  := interleave(tenant, device)
    device_id   := hashed_uuid(device_key)
    device_kind := weighted_strings(device_key, "sensor:0.7;gateway:0.2;controller:0.1")

    reading_seed := hash(interleave(device_key, reading))
    (temp_c, humidity, status) := reading_model(reading_seed)
    ts := base_epoch_ms + reading * interval_ms

    // A coarse health signal derived from the same coordinate: nonzero
    // when the reading should be flagged by a verifier.
    flagged := if temp_c > 26.0 { 1 } else { 0 }

    // ---- Document -----------------------------------------------------

    // The reading as a JSON document. A tile is a template whose holes
    // are wires: numbers render bare and strings quoted by their types,
    // the `meta` arm is one static copy, and `samples` repeats its body
    // over a comprehension. The tile is a wire like any other.
    tile doc : json := {
        "meta": { "schema": 3, "source": "polydat", "units": { "temp": "C", "rh": "%" } },
        "tenant": ${tenant_id},
        "device": ${device_id},
        "kind": ${device_kind},
        "ts": ${ts},
        "reading": { "temp": ${temp_c | .2}, "rh": ${humidity | .1}, "status": ${status} },
        "samples": [ @for s in 0..4 { { "n": ${s}, "temp": ${temp_c + s | .2} } } ],
        "flagged": ${flagged: bool}
    }

    // ---- Test flow ----------------------------------------------------

    // The load statement carries the document raw: it is already JSON.
    tile load := <<<
INSERT INTO ${keyspace}.${table} (tenant_id, device_id, ts, doc) VALUES (${tenant_id}, '${device_id}', ${ts}, '${doc!}')
>>>
    read_stmt   := "SELECT doc FROM {keyspace}.{table} WHERE tenant_id = {tenant_id} AND device_id = '{device_id}' AND ts = {ts}"
    verify_stmt := "expect temp_c = {temp_c}, humidity = {humidity}, status = '{status}'"

    // The statement this activation executes is selected by its phase.
    stmt := select_str(str_eq(phase, "load"), load, verify_stmt)

}