From bdac43d9bde4ce6d150942043121ab4e1e08dae1 Mon Sep 17 00:00:00 2001
From: Jason Petersen <jason@tembo.io>
Date: Sun, 12 May 2024 22:45:30 -0600
Subject: [PATCH] Add LOCF and gap-fill logic

---
 sql/timeseries.sql | 55 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 55 insertions(+)

diff --git a/sql/timeseries.sql b/sql/timeseries.sql
index 3d83545..370dc94 100644
--- a/sql/timeseries.sql
+++ b/sql/timeseries.sql
@@ -496,3 +496,58 @@ CREATE OR REPLACE AGGREGATE last(value anyelement, rank anycompatible) (
     FINALFUNC = endpoint_final,
     FINALFUNC_EXTRA
 );
+
+-- When provided with a table, time source, data source,
+-- stride, and overall time range, this function will
+-- bin all data within the specified range into rows
+-- spaced according to the stride.
+--
+-- Time bins which lack data in the input table will still
+-- receive a single row of output with a NULL in the data
+-- column
+CREATE OR REPLACE FUNCTION @extschema@.date_bin_table(target_table_id regclass,
+  time_source name,
+  time_stride interval,
+  time_range tstzrange,
+  data_source name)
+ RETURNS TABLE(binned_date timestamp with time zone,
+               stat numeric)
+ LANGUAGE plpgsql
+AS $function$
+DECLARE
+BEGIN
+  RETURN QUERY EXECUTE format($$
+      WITH data AS (
+        SELECT %I as stat,
+          date_bin($1, %I, $2) binned_date
+        WHERE %I BETWEEN $2 AND $3
+        FROM %I ORDER BY 2
+      )
+      SELECT date_series.date, data.stat
+      FROM generate_series($2, $3, $1) date_series(date)
+      LEFT JOIN data
+      ON data.binned_date = date_series.date;$$,
+    data_source,
+    time_source,
+    target_table_id)
+  USING time_stride, lower(time_range), upper(time_range);
+END;
+$function$;
+
+-- Function implementation for LOCF: last-observed carry-forward.
+-- Intended for use on the output of date_bin_table in order to
+-- fill NULL rows with the last observed value.
+CREATE OR REPLACE FUNCTION public.locf_agg(state anyelement, value anyelement)
+ RETURNS anyelement
+ LANGUAGE plpgsql
+AS $function$
+BEGIN
+  RETURN COALESCE(value, state);
+END;
+$function$;
+
+-- Aggregate for LOCF. For use in a WINDOW clause
+CREATE AGGREGATE locf(anyelement) (
+  SFUNC = locf_agg,
+  STYPE = anyelement
+);