Skip to content

feat(heartbeat): add average and stddev functions to heartbeat agg #770

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ollz272 wants to merge 9 commits into
base: main
Choose a base branch
from
Open

feat(heartbeat): add average and stddev functions to heartbeat agg #770

Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a60c331
feat(heartbeat): add average and stddev functions to heartbeat agg
ollz272 Sep 6, 2023
e7e6d76
fix: formatting
ollz272 Sep 6, 2023
05fb7d8
fix: mean error
ollz272 Sep 6, 2023
898a416
fix: some errors
ollz272 Sep 6, 2023
33352ba
fix: some errors
ollz272 Sep 6, 2023
0e1921a
fix: errors
ollz272 Sep 6, 2023
769330b
fix: working now..
ollz272 Sep 6, 2023
b99911c
fix: formatting
ollz272 Sep 6, 2023
68cee55
fix: test fix
ollz272 Sep 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 102 additions & 24 deletions extension/src/heartbeat_agg.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -395,6 +395,84 @@ pub fn uptime(agg: HeartbeatAgg<'static>) -> Interval {
agg.sum_live_intervals().into()
}

#[pg_extern]
pub fn average_downtime(agg: HeartbeatAgg<'static>) -> Option<f64> {
(agg.end_time - agg.start_time - agg.sum_live_intervals()).into() / agg.num_intervals
}

#[pg_extern]
pub fn average_uptime(agg: HeartbeatAgg<'static>) -> Option<f64> {
(agg.sum_live_intervals()).into() / agg.num_intervals
}

#[pg_extern]
pub fn stddev_downtime(agg: HeartbeatAgg<'static>) -> Option<f64> {
// Dead ranges are the opposite of the intervals stored in the aggregate
let mut starts = agg.interval_ends.clone().into_vec();
let mut ends = agg.interval_starts.clone().into_vec();

// Fix the first point depending on whether the aggregate starts in a live or dead range
if ends[0] == agg.start_time {
ends.remove(0);
} else {
starts.insert(0, agg.start_time);
}

// Fix the last point depending on whether the aggregate starts in a live or dead range
if *starts.last().unwrap() == agg.end_time {
starts.pop();
} else {
ends.push(agg.end_time);
}

let mut data = vec![];
for i in 0..agg.num_intervals as usize {
vec![].push(ends[i] - starts[i]);
}
match (mean(data), data.len()) {
(Some(data_mean), count) if count > 0 => {
let variance = data
.iter()
.map(|value| {
let diff = data_mean - (*value as f32);

diff * diff
})
.sum::<f64>()
/ count as f64;

Some(variance.sqrt())
}
_ => None,
}
}

#[pg_extern]
pub fn stddev_uptime(agg: HeartbeatAgg<'static>) -> Option<f64> {
let starts = agg.interval_starts.as_slice();
let ends = agg.interval_ends.as_slice();
let mut data = vec![];
for i in 0..agg.num_intervals as usize {
vec![].push(ends[i] - starts[i]);
}
match (mean(data), data.len()) {
(Some(data_mean), count) if count > 0 => {
let variance = data
.iter()
.map(|value| {
let diff = data_mean - (*value as f32);

diff * diff
})
.sum::<f64>()
/ count as f64;

Some(variance.sqrt())
}
_ => None,
}
}

#[pg_operator(immutable, parallel_safe)]
#[opname(->)]
pub fn arrow_heartbeat_agg_uptime(
Expand Down Expand Up @@ -1014,7 +1092,7 @@ mod tests {
let (result1, result2, result3) =
client.update(
"WITH agg AS (SELECT heartbeat_agg(heartbeat, '01-01-2020 UTC', '2h', '10m') AS agg FROM liveness)
SELECT live_at(agg, '01-01-2020 00:01:00 UTC')::TEXT,
SELECT live_at(agg, '01-01-2020 00:01:00 UTC')::TEXT,
live_at(agg, '01-01-2020 00:05:00 UTC')::TEXT,
live_at(agg, '01-01-2020 00:30:00 UTC')::TEXT FROM agg", None, None)
.unwrap().first()
Expand All @@ -1035,7 +1113,7 @@ mod tests {
let (result1, result2, result3) =
client.update(
"WITH agg AS (SELECT heartbeat_agg(heartbeat, '01-01-2020 UTC', '2h', '10m') AS agg FROM liveness)
SELECT (agg -> live_at('01-01-2020 00:01:00 UTC'))::TEXT,
SELECT (agg -> live_at('01-01-2020 00:01:00 UTC'))::TEXT,
(agg -> live_at('01-01-2020 00:05:00 UTC'))::TEXT,
(agg -> live_at('01-01-2020 00:30:00 UTC'))::TEXT FROM agg", None, None)
.unwrap().first()
Expand Down Expand Up @@ -1110,7 +1188,7 @@ mod tests {
.update(
"WITH aggs AS (
SELECT heartbeat_agg(time, batch, '1h', '1m')
FROM heartbeats
FROM heartbeats
GROUP BY batch
) SELECT rollup(heartbeat_agg)::TEXT FROM aggs",
None,
Expand Down Expand Up @@ -1348,12 +1426,12 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start),
t AS (
SELECT start,
interpolate(agg, LAG (agg) OVER (ORDER BY start)) AS agg
interpolate(agg, LAG (agg) OVER (ORDER BY start)) AS agg
FROM s)
SELECT downtime(agg)::TEXT FROM t;",
None,
Expand Down Expand Up @@ -1395,12 +1473,12 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start),
t AS (
SELECT start,
interpolate(agg, LAG (agg) OVER (ORDER BY start)) AS agg
interpolate(agg, LAG (agg) OVER (ORDER BY start)) AS agg
FROM s)
SELECT live_ranges(agg)::TEXT FROM t;",
None,
Expand Down Expand Up @@ -1455,12 +1533,12 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start),
t AS (
SELECT start,
agg -> interpolate(LAG (agg) OVER (ORDER BY start)) AS agg
agg -> interpolate(LAG (agg) OVER (ORDER BY start)) AS agg
FROM s)
SELECT live_ranges(agg)::TEXT FROM t;",
None,
Expand Down Expand Up @@ -1515,8 +1593,8 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start)
SELECT interpolated_uptime(agg, LAG (agg) OVER (ORDER BY start))::TEXT
FROM s",
Expand Down Expand Up @@ -1559,8 +1637,8 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start)
SELECT (agg -> interpolated_uptime(LAG (agg) OVER (ORDER BY start)))::TEXT
FROM s",
Expand Down Expand Up @@ -1603,8 +1681,8 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start)
SELECT interpolated_downtime(agg, LAG (agg) OVER (ORDER BY start))::TEXT
FROM s",
Expand Down Expand Up @@ -1647,8 +1725,8 @@ mod tests {
.update(
"WITH s AS (
SELECT start,
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
heartbeat_agg(heartbeat, start, '30m', '10m') AS agg
FROM liveness
GROUP BY start)
SELECT (agg -> interpolated_downtime(LAG (agg) OVER (ORDER BY start)))::TEXT
FROM s",
Expand Down Expand Up @@ -1836,12 +1914,12 @@ mod tests {
let output = client
.update(
"WITH rollups AS (
SELECT heartbeat_agg(ts, batch, '2h', '20m')
FROM poc
GROUP BY batch
SELECT heartbeat_agg(ts, batch, '2h', '20m')
FROM poc
GROUP BY batch
ORDER BY batch
)
SELECT live_ranges(rollup(heartbeat_agg))::TEXT
SELECT live_ranges(rollup(heartbeat_agg))::TEXT
FROM rollups",
None,
None,
Expand Down