Materialized Views and Incremental Refresh

-- Basic materialized view
CREATE MATERIALIZED VIEW mv_order_summary AS
SELECT
    customer_id,
    DATE_TRUNC('day', created_at) AS day,
    COUNT(*) AS order_count,
    SUM(total) AS revenue,
    AVG(total) AS avg_order_value
FROM orders
WHERE status = 'completed'
GROUP BY customer_id, DATE_TRUNC('day', created_at);
 
-- Query like a regular table
SELECT * FROM mv_order_summary
WHERE customer_id = 42
AND day >= '2026-08-01';
 
-- Add a unique index for concurrent refresh
CREATE UNIQUE INDEX idx_mv_order_summary
    ON mv_order_summary (customer_id, day);

-- Exclusive refresh (blocks all reads)
REFRESH MATERIALIZED VIEW mv_order_summary;
-- Takes 5 minutes for a large view
-- All queries to mv_order_summary block for 5 minutes
 
-- Concurrent refresh (allows reads during refresh)
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_order_summary;
-- Still takes 5 minutes
-- But queries can read the old version while refresh runs
-- Requires unique index on the materialized view

-- Monitor refresh progress
SELECT
    matviewname,
    pg_size_pretty(pg_total_relation_size(matviewname::regclass)) AS size,
    last_refresh
FROM pg_stat_matviews;
 
-- Refresh with timing
\timing on
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_order_summary;
-- Shows how long the refresh took

-- Create a changes table to track modifications
CREATE TABLE order_changes (
    id bigserial PRIMARY KEY,
    customer_id bigint,
    day date,
    operation text,  -- 'insert', 'update', 'delete'
    created_at timestamptz DEFAULT now()
);
 
-- Populate changes table with triggers
CREATE OR REPLACE FUNCTION track_order_changes()
RETURNS trigger AS $$
BEGIN
    INSERT INTO order_changes (customer_id, day, operation)
    VALUES (
        COALESCE(NEW.customer_id, OLD.customer_id),
        DATE_TRUNC('day', COALESCE(NEW.created_at, OLD.created_at)),
        TG_OP
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
CREATE TRIGGER trg_order_changes
    AFTER INSERT OR UPDATE OR DELETE ON orders
    FOR EACH ROW EXECUTE FUNCTION track_order_changes();
 
-- Incremental refresh function
CREATE OR REPLACE FUNCTION refresh_order_summary_incremental()
RETURNS void AS $$
BEGIN
    -- Only recompute affected (customer_id, day) pairs
    INSERT INTO mv_order_summary
    SELECT
        customer_id,
        DATE_TRUNC('day', created_at) AS day,
        COUNT(*) AS order_count,
        SUM(total) AS revenue,
        AVG(total) AS avg_order_value
    FROM orders
    WHERE (customer_id, DATE_TRUNC('day', created_at)) IN (
        SELECT customer_id, day FROM order_changes
    )
    AND status = 'completed'
    GROUP BY customer_id, DATE_TRUNC('day', created_at)
    ON CONFLICT (customer_id, day)
    DO UPDATE SET
        order_count = EXCLUDED.order_count,
        revenue = EXCLUDED.revenue,
        avg_order_value = EXCLUDED.avg_order_value;
    
    -- Clear processed changes
    TRUNCATE order_changes;
END;
$$ LANGUAGE plpgsql;

-- TimescaleDB continuous aggregate
CREATE MATERIALIZED VIEW mv_hourly_metrics
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', created_at) AS bucket,
    COUNT(*) AS request_count,
    AVG(duration_ms) AS avg_duration,
    percentile_cont(0.99) WITHIN GROUP (ORDER BY duration_ms) AS p99_duration
FROM http_requests
GROUP BY time_bucket('1 hour', created_at)
WITH NO DATA;
 
-- Add a refresh policy
SELECT add_continuous_aggregate_policy('mv_hourly_metrics',
    start_offset => INTERVAL '3 hours',
    end_offset => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour'
);
 
-- Only processes data between start_offset and end_offset
-- Older data is never reprocessed

# Cron-based refresh (simple but has gaps)
# Refresh every 15 minutes
*/15 * * * * psql -c "REFRESH MATERIALIZED VIEW CONCURRENTLY mv_order_summary"
 
# pg_cron extension (PostgreSQL-native)
CREATE EXTENSION pg_cron;
SELECT cron.schedule('refresh-order-summary', '*/15 * * * *',
    $$REFRESH MATERIALIZED VIEW CONCURRENTLY mv_order_summary$$);
 
# Monitor cron jobs
SELECT * FROM cron.job;
SELECT * FROM cron.job_run_details ORDER BY start_time DESC LIMIT 10;

-- Base tables
-- orders (raw data)
-- customers (raw data)
 
-- Layer 1: Direct aggregation
CREATE MATERIALIZED VIEW mv_daily_orders AS
SELECT customer_id, DATE_TRUNC('day', created_at) AS day, COUNT(*), SUM(total)
FROM orders GROUP BY customer_id, DATE_TRUNC('day', created_at);
 
-- Layer 2: Depends on Layer 1
CREATE MATERIALIZED VIEW mv_customer_lifetime AS
SELECT customer_id, SUM(total) AS lifetime_revenue
FROM mv_daily_orders
GROUP BY customer_id;
 
-- Refresh order: Layer 1 must be refreshed before Layer 2
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_daily_orders;
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_customer_lifetime;

-- Automated refresh with dependency tracking
CREATE OR REPLACE FUNCTION refresh_materialized_views()
RETURNS void AS $$
BEGIN
    -- Refresh in dependency order
    REFRESH MATERIALIZED VIEW CONCURRENTLY mv_daily_orders;
    REFRESH MATERIALIZED VIEW CONCURRENTLY mv_customer_lifetime;
    REFRESH MATERIALIZED VIEW CONCURRENTLY mv_monthly_cohorts;
END;
$$ LANGUAGE plpgsql;