Calculating Optimal chunk_interval for IoT Sensor Data

-- Conceptual calculation (execute in application layer or psql).
-- target_chunk_bytes is written as a numeric literal (500000000.0) so the
-- division is performed in floating point rather than truncating integer math.
WITH telemetry_profile AS (
  SELECT
    500000000.0 AS target_chunk_bytes,  -- 500 MB midpoint
    2500 AS avg_rows_per_sec,
    180 AS avg_row_bytes
)
SELECT
  target_chunk_bytes / (avg_rows_per_sec * 86400 * avg_row_bytes) AS chunk_interval_days
FROM telemetry_profile;

-- 1. Create hypertable idempotently
CREATE TABLE IF NOT EXISTS sensor_readings (
    time TIMESTAMPTZ NOT NULL,
    device_id UUID NOT NULL,
    metric_type TEXT NOT NULL,
    value DOUBLE PRECISION,
    payload JSONB
);

SELECT create_hypertable(
    'sensor_readings', 
    'time', 
    chunk_time_interval => INTERVAL '1 hour', 
    if_not_exists => TRUE
);

-- 2. Enable compression with idempotent policy
SELECT add_compression_policy(
    'sensor_readings', 
    INTERVAL '7 days', 
    if_not_exists => TRUE
);

-- 3. Configure continuous aggregate for 5-minute rollups
CREATE MATERIALIZED VIEW IF NOT EXISTS sensor_metrics_5m
WITH (timescaledb.continuous) AS
SELECT
    time_bucket(INTERVAL '5 minutes', time) AS bucket,
    device_id,
    metric_type,
    avg(value) AS avg_value,
    max(value) AS max_value,
    min(value) AS min_value
FROM sensor_readings
GROUP BY 1, 2, 3;

-- 4. Automate refresh and retention
SELECT add_continuous_aggregate_policy('sensor_metrics_5m',
    start_offset => INTERVAL '2 hours',
    end_offset => INTERVAL '5 minutes',
    schedule_interval => INTERVAL '15 minutes',
    if_not_exists => TRUE
);

SELECT add_retention_policy('sensor_readings', 
    drop_after => INTERVAL '90 days', 
    if_not_exists => TRUE);

import psycopg
from psycopg.rows import dict_row
import math

def calculate_optimal_chunk_interval(conn_str: str, target_mb: int = 500) -> str:
    """
    Queries live telemetry metrics and returns a PostgreSQL interval string
    optimized for chunk sizing.
    """
    with psycopg.connect(conn_str, row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    COUNT(*)::float / EXTRACT(EPOCH FROM (MAX(time) - MIN(time))) AS rps,
                    pg_column_size((SELECT row_to_json(t) FROM sensor_readings t LIMIT 1)) AS row_bytes
                FROM sensor_readings
                WHERE time > NOW() - INTERVAL '1 hour';
            """)
            row = cur.fetchone()
            if not row or row['rps'] == 0:
                return "1 hour"  # Fallback (bare interval literal for ::interval)
            
            daily_rows = row['rps'] * 86400
            daily_bytes = daily_rows * row['row_bytes']
            target_bytes = target_mb * 1_048_576
            
            interval_days = target_bytes / daily_bytes
            interval_minutes = max(15, math.ceil(interval_days * 1440))  # Floor at 15m
            
            # Round to nearest operational boundary (15m, 30m, 1h, 2h, etc.)
            boundaries = [15, 30, 60, 120, 240]
            optimal = min(boundaries, key=lambda x: abs(x - interval_minutes))
            return f"{optimal} minutes"

def apply_chunk_interval(conn_str: str, interval_sql: str) -> None:
    """
    Idempotently updates the hypertable chunk interval for new chunks.
    Existing chunks remain untouched per TimescaleDB design.
    """
    with psycopg.connect(conn_str) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT set_chunk_time_interval(
                    'sensor_readings', 
                    %s::interval
                );
            """, (interval_sql,))
            conn.commit()

# Usage
# interval = calculate_optimal_chunk_interval("dbname=iot_db user=ts_admin")
# apply_chunk_interval("dbname=iot_db user=ts_admin", interval)

-- Monitor chunk distribution and compression ratios
SELECT
    chunk_schema,
    chunk_name,
    pg_size_pretty(pg_total_relation_size(format('%I.%I', chunk_schema, chunk_name))) AS chunk_size,
    is_compressed,
    range_start,
    range_end
FROM timescaledb_information.chunks
WHERE hypertable_name = 'sensor_readings'
ORDER BY range_start DESC
LIMIT 10;

-- Verify background job execution history
SELECT
    j.job_id,
    j.proc_schema || '.' || j.proc_name AS job_name,
    js.last_run_status,
    js.last_run_started_at,
    j.next_start
FROM timescaledb_information.jobs j
LEFT JOIN timescaledb_information.job_stats js ON js.job_id = j.job_id
WHERE j.hypertable_name = 'sensor_readings'
   OR j.config::text LIKE '%sensor_readings%';