feat: query from realtime view

Author: EvolveArt

Cargo.lock

@@ -1,4 +1,2 @@
 -- Your SQL goes here
-SELECT create_hypertable('entries', by_range('timestamp'));
-SELECT * FROM add_dimension('entries', by_hash('pair_id', 4))
-SELECT * FROM add_dimension('entries', by_hash('source', 4))
+SELECT create_hypertable('entries', 'timestamp');

pragma-entities/migrations/2023-12-29-052754_add_hypertable_entries/up.sql

@@ -2,23 +2,45 @@
 CREATE MATERIALIZED VIEW price_1_min_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
 AS SELECT source,
-    time_bucket('1 min'::interval, ts) as bucket,
+    pair_id,
+    time_bucket('1 min'::interval, timestamp) as bucket,
     percentile_agg(price)
 FROM entries
-GROUP BY source, bucket;
+GROUP BY source, pair_id, bucket
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('price_1_min_agg',
+  start_offset => INTERVAL '3 min',
+  end_offset => INTERVAL '1 min',
+  schedule_interval => INTERVAL '1 min');
 
 CREATE MATERIALIZED VIEW price_15_min_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
 AS SELECT source,
-    time_bucket('15 min'::interval, ts) as bucket,
+    pair_id,
+    time_bucket('15 min'::interval, timestamp) as bucket,
     percentile_agg(price)
 FROM entries
-GROUP BY source, bucket;
+GROUP BY source, pair_id, bucket
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('price_15_min_agg',
+  start_offset => INTERVAL '45 min',
+  end_offset => INTERVAL '15 min',
+  schedule_interval => INTERVAL '15 min');
 
 CREATE MATERIALIZED VIEW price_1_h_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
 AS SELECT source,
-    time_bucket('1 h'::interval, ts) as bucket,
+    pair_id,
+    time_bucket('1 hour'::interval, timestamp) as bucket,
     percentile_agg(price)
 FROM entries
-GROUP BY source, bucket;
+GROUP BY source, pair_id, bucket
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('price_1_h_agg',
+  start_offset => INTERVAL '3 hours',
+  end_offset => INTERVAL '1 hour',
+  schedule_interval => INTERVAL '1 hour');
+

pragma-entities/migrations/2024-01-11-123510_add_continuous_aggregates/up.sql

@@ -86,17 +86,15 @@ pub async fn get_median_entries(
     let conn = pool.get().await.map_err(adapt_infra_error)?;
 
     let raw_sql = r#"
-        -- select the latest entry for every publisher,source combination
+        -- query the materialized realtime view
         SELECT
             source,
-            MAX(timestamp) AS time,
-            (approx_percentile(0.5, uddsketch(100, 0.01, price)) WITHIN GROUP (ORDER BY price))::numeric AS median_price
+            MAX(bucket) AS time,
+            approx_percentile(0.5, percentile_agg) AS median_price
         FROM
-            entries
+            price_1_min_agg
         WHERE
             pair_id = $1
-        GROUP BY
-            source
         ORDER BY
             source;
     "#;