TimescaleDB Time Series

# TimescaleDB Time-Series

You are an expert in TimescaleDB for time-series data, IoT analytics, and real-time monitoring applications.

## Key Principles
- Convert regular tables to hypertables for automatic time-based partitioning
- Use continuous aggregates for efficient pre-computed analytics
- Implement compression policies for storage optimization
- Leverage retention policies for automatic data lifecycle management
- Design schemas optimized for time-series query patterns

## Hypertable Setup

```sql
-- Enable TimescaleDB extension
CREATE EXTENSION IF NOT EXISTS timescaledb;

-- Create a regular table first
CREATE TABLE sensor_data (
    time TIMESTAMPTZ NOT NULL,
    sensor_id TEXT NOT NULL,
    location TEXT,
    temperature DOUBLE PRECISION,
    humidity DOUBLE PRECISION,
    pressure DOUBLE PRECISION,
    battery_level DOUBLE PRECISION
);

-- Convert to hypertable with automatic partitioning
SELECT create_hypertable(
    'sensor_data',
    'time',
    chunk_time_interval => INTERVAL '1 day',
    if_not_exists => TRUE
);

-- Create indexes for common query patterns
CREATE INDEX idx_sensor_data_sensor_time 
    ON sensor_data (sensor_id, time DESC);
CREATE INDEX idx_sensor_data_location 
    ON sensor_data (location, time DESC);

-- Metrics table with space partitioning
CREATE TABLE metrics (
    time TIMESTAMPTZ NOT NULL,
    device_id INTEGER NOT NULL,
    metric_name TEXT NOT NULL,
    value DOUBLE PRECISION,
    tags JSONB
);

SELECT create_hypertable(
    'metrics',
    'time',
    partitioning_column => 'device_id',
    number_partitions => 4,
    chunk_time_interval => INTERVAL '6 hours'
);
```

## Continuous Aggregates for Real-time Analytics

```sql
-- Hourly aggregates with automatic refresh
CREATE MATERIALIZED VIEW sensor_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS bucket,
    sensor_id,
    location,
    AVG(temperature) AS avg_temp,
    MIN(temperature) AS min_temp,
    MAX(temperature) AS max_temp,
    AVG(humidity) AS avg_humidity,
    COUNT(*) AS sample_count
FROM sensor_data
GROUP BY bucket, sensor_id, location
WITH NO DATA;

-- Add refresh policy
SELECT add_continuous_aggregate_policy('sensor_hourly',
    start_offset => INTERVAL '3 hours',
    end_offset => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour'
);

-- Daily rollups from hourly data (hierarchical aggregates)
CREATE MATERIALIZED VIEW sensor_daily
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 day', bucket) AS day,
    sensor_id,
    location,
    AVG(avg_temp) AS avg_temp,
    MIN(min_temp) AS min_temp,
    MAX(max_temp) AS max_temp,
    AVG(avg_humidity) AS avg_humidity,
    SUM(sample_count) AS total_samples
FROM sensor_hourly
GROUP BY day, sensor_id, location
WITH NO DATA;

-- Real-time aggregate queries
SELECT
    bucket,
    sensor_id,
    avg_temp,
    avg_humidity
FROM sensor_hourly
WHERE bucket >= NOW() - INTERVAL '24 hours'
    AND sensor_id = 'sensor-001'
ORDER BY bucket DESC;
```

## Compression and Retention Policies

```sql
-- Enable compression on hypertable
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'sensor_id',
    timescaledb.compress_orderby = 'time DESC'
);

-- Add compression policy (compress data older than 7 days)
SELECT add_compression_policy('sensor_data', INTERVAL '7 days');

-- Add retention policy (drop data older than 90 days)
SELECT add_retention_policy('sensor_data', INTERVAL '90 days');

-- Check compression stats
SELECT
    hypertable_name,
    before_compression_total_bytes,
    after_compression_total_bytes,
    ROUND((1 - after_compression_total_bytes::numeric / 
           before_compression_total_bytes::numeric) * 100, 2) AS compression_ratio
FROM hypertable_compression_stats('sensor_data');

-- Manually compress specific chunks
SELECT compress_chunk(c.chunk_name)
FROM timescaledb_information.chunks c
WHERE c.hypertable_name = 'sensor_data'
    AND NOT c.is_compressed
    AND c.range_end < NOW() - INTERVAL '7 days';
```

## Advanced Time-Series Queries

```sql
-- Gap filling with interpolation
SELECT
    time_bucket_gapfill('5 minutes', time) AS bucket,
    sensor_id,
    locf(AVG(temperature)) AS temperature, -- last observation carried forward
    interpolate(AVG(humidity)) AS humidity  -- linear interpolation
FROM sensor_data
WHERE time >= NOW() - INTERVAL '1 hour'
    AND sensor_id = 'sensor-001'
GROUP BY bucket, sensor_id
ORDER BY bucket;

-- Moving averages and window functions
SELECT
    time,
    sensor_id,
    temperature,
    AVG(temperature) OVER (
        PARTITION BY sensor_id 
        ORDER BY time 
        ROWS BETWEEN 5 PRECEDING AND CURRENT ROW
    ) AS moving_avg_6,
    LAG(temperature) OVER (
        PARTITION BY sensor_id ORDER BY time
    ) AS prev_temperature
FROM sensor_data
WHERE time >= NOW() - INTERVAL '1 hour'
ORDER BY time DESC;

-- Percentile calculations
SELECT
    time_bucket('1 hour', time) AS hour,
    sensor_id,
    percentile_cont(0.50) WITHIN GROUP (ORDER BY temperature) AS median_temp,
    percentile_cont(0.95) WITHIN GROUP (ORDER BY temperature) AS p95_temp,
    percentile_cont(0.99) WITHIN GROUP (ORDER BY temperature) AS p99_temp
FROM sensor_data
WHERE time >= NOW() - INTERVAL '24 hours'
GROUP BY hour, sensor_id
ORDER BY hour DESC;

-- Anomaly detection with delta
SELECT
    time,
    sensor_id,
    temperature,
    temperature - LAG(temperature) OVER w AS temp_delta,
    CASE 
        WHEN ABS(temperature - AVG(temperature) OVER w) > 
             2 * STDDEV(temperature) OVER w 
        THEN true 
        ELSE false 
    END AS is_anomaly
FROM sensor_data
WHERE time >= NOW() - INTERVAL '1 hour'
WINDOW w AS (PARTITION BY sensor_id ORDER BY time ROWS BETWEEN 100 PRECEDING AND CURRENT ROW)
ORDER BY time DESC;
```

## Node.js Integration

```typescript
import { Pool } from 'pg';

const pool = new Pool({
  connectionString: process.env.TIMESCALE_URL,
});

// Batch insert for high throughput
async function insertSensorData(readings: SensorReading[]) {
  const values = readings.map((r, i) => {
    const offset = i * 6;
    return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}, $${offset + 6})`;
  }).join(', ');
  
  const params = readings.flatMap(r => [
    r.time, r.sensorId, r.location, 
    r.temperature, r.humidity, r.pressure
  ]);

  await pool.query(`
    INSERT INTO sensor_data (time, sensor_id, location, temperature, humidity, pressure)
    VALUES ${values}
  `, params);
}

// Real-time dashboard query
async function getDashboardData(sensorId: string, hours: number = 24) {
  const result = await pool.query(`
    SELECT
      time_bucket('15 minutes', time) AS bucket,
      AVG(temperature) AS avg_temp,
      AVG(humidity) AS avg_humidity,
      COUNT(*) AS samples
    FROM sensor_data
    WHERE sensor_id = $1
      AND time >= NOW() - $2 * INTERVAL '1 hour'
    GROUP BY bucket
    ORDER BY bucket DESC
  `, [sensorId, hours]);
  
  return result.rows;
}

// Alerting query for threshold breaches
async function checkAlerts(threshold: number) {
  const result = await pool.query(`
    SELECT DISTINCT ON (sensor_id)
      sensor_id,
      time,
      temperature,
      location
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '5 minutes'
      AND temperature > $1
    ORDER BY sensor_id, time DESC
  `, [threshold]);
  
  return result.rows;
}
```

## Best Practices
- Choose chunk intervals based on query patterns (smaller for recent data queries)
- Use continuous aggregates for dashboard and reporting queries
- Implement compression for storage cost reduction (typically 90%+ savings)
- Set appropriate retention policies to manage data lifecycle
- Use time_bucket_gapfill for visualization-ready time-series data
- Create indexes on commonly filtered columns alongside time