[PostgreSQL]PostgreSQL+TimescaleDB：时序算法数据存储方案

-- InfluxQL无法表达复杂逻辑
SELECT "device_id", MEAN("temperature") as avg_temp 
FROM "sensors" 
WHERE time > now() - 30d 
GROUP BY "device_id";
-- 需导出到Python计算3σ，再二次查询，耗时47分钟

-- 期望在数据库内完成算法计算
WITH stats AS (
    SELECT device_id, 
           AVG(temperature) as mean,
           STDDEV(temperature) as std
    FROM sensor_data
    WHERE ts BETWEEN NOW() - INTERVAL '30 days' AND NOW()
    GROUP BY device_id
)
SELECT d.device_id, 
       d.ts, 
       d.temperature,
       (d.temperature - s.mean) / s.std as z_score
FROM sensor_data d
JOIN stats s ON d.device_id = s.device_id
WHERE (d.temperature - s.mean) / s.std > 3;

-- 将普通表转为超表，自动按时间分区
SELECT create_hypertable('sensor_data', by_range('ts', INTERVAL '1 day'));
-- 底层自动创建1000+个子分区，对应用透明

-- 预计算小时级统计，自动刷新
CREATE MATERIALIZED VIEW sensor_hourly
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 hour', ts) as bucket,
    device_id,
    AVG(temperature) as avg_temp,
    MAX(temperature) as max_temp
FROM sensor_data
GROUP BY bucket, device_id;

-- 自动删除90天前数据
SELECT add_retention_policy('sensor_data', INTERVAL '90 days');
-- 底层使用高效DROP PARTITION，非DELETE

# /etc/sysctl.conf
# I. 内存与文件系统
vm.swappiness = 1
vm.dirty_ratio = 15
vm.dirty_background_ratio = 5
fs.file-max = 2097152

# II. 网络
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
net.ipv4.tcp_congestion_control = bbr

# 应用
sysctl -p

# 使用XFS，禁用atime
mkfs.xfs -f -i size=512 /dev/nvme0n1
mount -o noatime,nodiratime /dev/nvme0n1 /data/pgsql

# 调整预读
blockdev --setra 65536 /dev/nvme0n1

# Ubuntu 22.04
apt update && apt install -y \
  build-essential \
  libreadline-dev \
  zlib1g-dev \
  libssl-dev \
  libxml2-dev \
  libxslt1-dev \
  libsystemd-dev \
  llvm-14-dev \
  clang-14 \
  python3-dev \
  postgresql-server-dev-15

wget https://ftp.postgresql.org/pub/source/v15.3/postgresql-15.3.tar.gz
tar xzf postgresql-15.3.tar.gz && cd postgresql-15.3

# 配置（关键优化项）
./configure \
  --prefix=/usr/local/pg15 \
  --with-systemd \
  --with-ssl=openssl \
  --with-llvm \
  --with-python \
  --enable-nls \
  --enable-thread-safety \
  CFLAGS="-O3 -march=native"

# 并行编译
make -j 128 world
make install-world

git clone https://github.com/timescale/timescaledb.git -b 2.10.1
cd timescaledb

# 使用PGXS编译
export PG_CONFIG=/usr/local/pg15/bin/pg_config
./bootstrap -DCMAKE_BUILD_TYPE=Release
cd build && make -j 128
make install

# 验证安装
/usr/local/pg15/bin/pg_config --version  # 应显示15.3
ls /usr/local/pg15/lib/timescaledb*.so    # 应看到扩展文件

useradd postgres
mkdir -p /data/pgsql/{data,wal,log}
chown -R postgres:postgres /data/pgsql

# 初始化
/usr/local/pg15/bin/initdb -D /data/pgsql/data \
  --encoding=UTF8 --locale=C --auth-local=peer

# 配置环境变量
echo 'export PATH=/usr/local/pg15/bin:$PATH' >> /home/postgres/.bashrc
echo 'export PGDATA=/data/pgsql/data' >> /home/postgres/.bashrc

-- postgresql.conf 关键配置

# I. 共享内存
shared_buffers = 64GB                    # 25%内存
effective_cache_size = 192GB             # 75%内存
maintenance_work_mem = 8GB               # 维护操作内存

# II. 写入优化
wal_buffers = 128MB
checkpoint_timeout = 15min
max_wal_size = 16GB
min_wal_size = 4GB

# III. 并行查询
max_worker_processes = 64
max_parallel_workers = 48
max_parallel_workers_per_gather = 16
parallel_setup_cost = 100
parallel_tuple_cost = 0.01

# IV. TimescaleDB专用
timescaledb.max_background_workers = 16  # 连续聚合worker
timescaledb.max_open_chunks_per_insert = 10  # 批量写入优化
timescaledb.max_cached_chunks_per_hypertable = 50

# V. 查询优化
random_page_cost = 1.1       # NVMe优化
effective_io_concurrency = 256
work_mem = 256MB

# VI. 日志
log_min_duration_statement = '1s'
log_line_prefix = '%t [%p-%l] %q%u@%d '
log_checkpoints = on

# 应用配置
SELECT pg_reload_conf();

-- 以设备传感器数据为例
CREATE TABLE sensor_data (
    ts TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    temperature DOUBLE PRECISION,
    humidity DOUBLE PRECISION,
    voltage DOUBLE PRECISION,
    status TEXT
);

-- I. 转换为超表（按天分区）
SELECT create_hypertable(
    'sensor_data', 
    'ts', 
    chunk_time_interval => INTERVAL '1 day',
    if_not_exists => TRUE
);

-- II. 添加设备ID分区（可选，加速设备级查询）
SELECT set_chunk_time_interval('sensor_data', INTERVAL '1 day');

-- III. 设置数据保留策略（90天）
SELECT add_retention_policy(
    'sensor_data', 
    INTERVAL '90 days',
    if_not_exists => TRUE
);

-- IV. 启用压缩（90天后压缩）
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id',  -- 按设备分组压缩
    timescaledb.compress_orderby = 'ts DESC'       -- 时间排序
);

SELECT add_compression_policy(
    'sensor_data', 
    compress_after => '90 days'::interval,
    if_not_exists => TRUE
);

-- V. 验证配置
SELECT * FROM timescaledb_information.hypertables 
WHERE hypertable_name = 'sensor_data';

-- 输出：
-- hypertable_schema | hypertable_name |   owner    | num_dimensions | num_chunks | compression_enabled | retention_enabled
-- -------------------+-----------------+------------+----------------+------------+---------------------+-------------------
-- public            | sensor_data     | postgres   |              1 |         92 | t                   | t

-- 正确：时间戳在最前，支持高效范围扫描
CREATE TABLE metrics (
    ts TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    metric_name TEXT,
    value DOUBLE PRECISION
);

-- 错误：device_id在前，时间范围查询性能差
CREATE TABLE metrics_bad (
    device_id INT NOT NULL,
    ts TIMESTAMPTZ NOT NULL,
    metric_name TEXT,
    value DOUBLE PRECISION
);

-- 压缩时按device_id分组，同一设备数据物理相邻
ALTER TABLE sensor_data SET (
    timescaledb.compress_segmentby = 'device_id'
);

-- 查询优势：按设备聚合只需扫描少量块
SELECT device_id, AVG(temperature)
FROM sensor_data
WHERE device_id = 12345  -- 仅扫描1-2个chunk
  AND ts > NOW() - INTERVAL '7 days';

-- 创建分钟级连续聚合
CREATE MATERIALIZED VIEW sensor_minute
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 minute', ts) as bucket,
    device_id,
    AVG(temperature) as avg_temp,
    MAX(temperature) as max_temp,
    MIN(temperature) as min_temp,
    COUNT(*) as sample_cnt
FROM sensor_data
GROUP BY bucket, device_id;

-- 查询最近1小时数据（从2秒降至20毫秒）
SELECT * FROM sensor_minute
WHERE bucket > NOW() - INTERVAL '1 hour';

-- 仅按时间分区，chunk数少
SELECT create_hypertable('metrics_global', 'ts', 
                         chunk_time_interval => INTERVAL '7 days');
-- 优点：元数据少，写入快
-- 缺点：设备级查询需扫描所有分区

-- 按时间分区，device_id作为segmentby
CREATE TABLE device_metrics (
    ts TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    metric_name TEXT,
    value DOUBLE PRECISION
);

SELECT create_hypertable('device_metrics', 'ts', 
                         chunk_time_interval => INTERVAL '1 day');

-- 添加设备维度索引
CREATE INDEX idx_device_metrics_device ON device_metrics (device_id, ts DESC);

-- 使用PostGIS + TimescaleDB
CREATE TABLE spatial_sensor_data (
    ts TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    location GEOGRAPHY(POINT, 4326),
    temperature DOUBLE PRECISION
);

SELECT create_hypertable('spatial_sensor_data', 'ts', 
                         chunk_time_interval => INTERVAL '1 day');

-- 空间索引
CREATE INDEX idx_spatial_location ON spatial_sensor_data USING GIST (location);

-- 时空范围查询（10公里内最近1小时）
SELECT * FROM spatial_sensor_data
WHERE ts > NOW() - INTERVAL '1 hour'
  AND ST_DWithin(location, ST_MakePoint(116.40, 39.90)::geography, 10000);

-- I. 创建原始超表
CREATE TABLE device_telemetry (
    ts TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    cpu_percent DOUBLE PRECISION,
    memory_percent DOUBLE PRECISION,
    disk_io_ps BIGINT
);

SELECT create_hypertable('device_telemetry', 'ts', 
                         chunk_time_interval => INTERVAL '1 day');

-- II. 创建多级连续聚合

-- 1分钟级（实时）
CREATE MATERIALIZED VIEW device_telemetry_minute
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 minute', ts) as bucket,
    device_id,
    AVG(cpu_percent) as avg_cpu,
    MAX(cpu_percent) as max_cpu,
    AVG(memory_percent) as avg_memory,
    SUM(disk_io_ps) as total_io
FROM device_telemetry
GROUP BY bucket, device_id
WITH DATA;

-- 配置刷新策略（1分钟滞后）
SELECT add_continuous_aggregate_policy('device_telemetry_minute',
    start_offset => INTERVAL '2 hours',
    end_offset => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute',
    if_not_exists => TRUE
);

-- 1小时级（历史分析）
CREATE MATERIALIZED VIEW device_telemetry_hourly
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 hour', bucket) as hour_bucket,
    device_id,
    AVG(avg_cpu) as avg_cpu_hourly,
    MAX(max_cpu) as max_cpu_hourly,
    SUM(total_io) as total_io_hourly
FROM device_telemetry_minute
GROUP BY hour_bucket, device_id
WITH DATA;

-- III. 查询优化器自动路由
-- 查询最近1小时：自动从device_telemetry_minute读取
-- 查询最近7天：自动从device_telemetry_hourly读取
-- 查询超过30天：自动从压缩chunk读取

-- IV. 预计算特征（用于机器学习）
CREATE MATERIALIZED VIEW device_features
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 day', ts) as day,
    device_id,
    -- 统计特征
    AVG(cpu_percent) as cpu_mean,
    STDDEV(cpu_percent) as cpu_std,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY cpu_percent) as cpu_p95,
    -- 时序特征
    AVG(cpu_percent) - LAG(AVG(cpu_percent)) OVER w as cpu_trend,
    COUNT(*) as uptime_hours
FROM device_telemetry
GROUP BY day, device_id
WINDOW w AS (PARTITION BY device_id ORDER BY time_bucket(INTERVAL '1 day', ts))
WITH DATA;

-- 刷新策略（每日凌晨2点）
SELECT add_continuous_aggregate_policy('device_features',
    start_offset => INTERVAL '30 days',
    end_offset => INTERVAL '1 day',
    schedule_interval => INTERVAL '1 day',
    initial_start => '02:00:00'
);

import pandas as pd

def moving_average_python(df, window=30):
    """
    Python中计算移动平均，需全表加载到内存
    耗时：约15秒/设备（100万行）
    """
    df['ma_30'] = df['temperature'].rolling(window=30).mean()
    return df

# 全表处理（1万台设备）≈ 15秒 × 10000 = 41小时！

-- I. 使用窗口函数（快1000倍）
CREATE OR REPLACE FUNCTION detect_temperature_trend(
    device_id INT,
    start_time TIMESTAMPTZ,
    end_time TIMESTAMPTZ
)
RETURNS TABLE (
    ts TIMESTAMPTZ,
    temperature DOUBLE PRECISION,
    ma_30 DOUBLE PRECISION,
    trend TEXT
) AS $$
BEGIN
    RETURN QUERY
    WITH daily_stats AS (
        SELECT 
            time_bucket(INTERVAL '1 hour', ts) as hour,
            AVG(temperature) as avg_temp
        FROM sensor_data
        WHERE device_id = $1 
          AND ts BETWEEN $2 AND $3
        GROUP BY hour
        ORDER BY hour
    ),
    moving_avg AS (
        SELECT 
            hour,
            avg_temp,
            AVG(avg_temp) OVER (
                ORDER BY hour 
                ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
            ) as ma_30,
            -- 计算二阶导数判断趋势
            AVG(avg_temp) OVER (
                ORDER BY hour 
                ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
            ) - AVG(avg_temp) OVER (
                ORDER BY hour 
                ROWS BETWEEN 59 PRECEDING AND 30 PRECEDING
            ) as trend_change
        FROM daily_stats
    )
    SELECT 
        hour,
        avg_temp,
        ma_30,
        CASE 
            WHEN trend_change > 0.5 THEN 'up'
            WHEN trend_change < -0.5 THEN 'down'
            ELSE 'stable'
        END as trend
    FROM moving_avg;
END;
$$ LANGUAGE plpgsql PARALLEL SAFE;

-- II. 查询调用（毫秒级）
SELECT * FROM detect_temperature_trend(
    device_id => 12345,
    start_time => NOW() - INTERVAL '7 days',
    end_time => NOW()
);

-- III. 批量处理所有设备（并行执行）
SELECT 
    device_id,
    AVG(CASE WHEN trend = 'up' THEN 1 ELSE 0 END) as up_ratio,
    MAX(ma_30) as max_temp_ma
FROM detect_temperature_trend(
    device_id => sd.device_id,
    start_time => NOW() - INTERVAL '7 days',
    end_time => NOW()
)
FROM (SELECT DISTINCT device_id FROM sensor_data WHERE ts > NOW() - INTERVAL '7 days') sd
GROUP BY device_id;

-- I. 创建异常检测函数
CREATE OR REPLACE FUNCTION detect_anomalies_zscore(
    metric_table REGCLASS,
    time_col TEXT,
    value_col TEXT,
    device_id_val INT,
    lookback_interval INTERVAL,
    threshold FLOAT DEFAULT 3.0
)
RETURNS TABLE (
    ts TIMESTAMPTZ,
    value DOUBLE PRECISION,
    z_score DOUBLE PRECISION,
    is_anomaly BOOLEAN
) AS $$
DECLARE
    sql TEXT;
    mean_val DOUBLE PRECISION;
    std_val DOUBLE PRECISION;
BEGIN
    -- 动态SQL构建
    sql := format(
        'SELECT AVG(%I), STDDEV(%I) FROM %s 
         WHERE device_id = %L 
           AND %I BETWEEN NOW() - %L AND NOW()',
        value_col, value_col, metric_table, device_id_val, 
        time_col, lookback_interval
    );
    
    EXECUTE sql INTO mean_val, std_val;
    
    -- 主查询
    RETURN QUERY EXECUTE format(
        'SELECT 
            %I,
            %I,
            (%I - %s) / NULLIF(%s, 0) as z_score,
            ABS((%I - %s) / NULLIF(%s, 0)) > %L as is_anomaly
         FROM %s
         WHERE device_id = %L
           AND %I BETWEEN NOW() - %L AND NOW()
         ORDER BY %I',
        time_col, value_col, value_col, mean_val, std_val,
        value_col, mean_val, std_val, threshold,
        metric_table, device_id_val, time_col, lookback_interval, time_col
    );
END;
$$ LANGUAGE plpgsql PARALLEL SAFE;

-- II. 调用示例
SELECT * FROM detect_anomalies_zscore(
    'sensor_data', 'ts', 'temperature', 12345, 
    INTERVAL '24 hours', 3.0
);

-- III. 创建物化视图存储结果（增量更新）
CREATE MATERIALIZED VIEW temperature_anomalies
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '5 minute', sd.ts) as bucket,
    sd.device_id,
    AVG(sd.temperature) as avg_temp,
    STDDEV(sd.temperature) as std_temp,
    -- 计算Z-score
    (MAX(sd.temperature) - AVG(sd.temperature)) / NULLIF(STDDEV(sd.temperature), 0) as max_zscore
FROM sensor_data sd
WHERE sd.ts > NOW() - INTERVAL '1 hour'
GROUP BY bucket, sd.device_id
HAVING (MAX(sd.temperature) - AVG(sd.temperature)) / NULLIF(STDDEV(sd.temperature), 0) > 3.0;

-- I. 创建趋势预测函数
CREATE OR REPLACE FUNCTION forecast_prophet_like(
    device_id INT,
    forecast_days INT DEFAULT 7
)
RETURNS TABLE (
    ds DATE,
    trend DOUBLE PRECISION,
    seasonal DOUBLE PRECISION,
    yhat DOUBLE PRECISION
) AS $$
BEGIN
    RETURN QUERY
    WITH base AS (
        SELECT 
            ts::date as ds,
            AVG(temperature) as y
        FROM sensor_data
        WHERE device_id = $1
          AND ts BETWEEN NOW() - INTERVAL '30 days' AND NOW()
        GROUP BY ds
        ORDER BY ds
    ),
    -- 趋势：线性回归
    trend AS (
        SELECT 
            ds,
            y,
            regr_slope(y, EXTRACT(EPOCH FROM ds)::bigint) OVER (
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
            ) as slope,
            regr_intercept(y, EXTRACT(EPOCH FROM ds)::bigint) OVER (
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
            ) as intercept
        FROM base
    ),
    -- 季节性：周周期
    seasonal AS (
        SELECT 
            ds,
            y,
            AVG(y) OVER (
                PARTITION BY EXTRACT(DOW FROM ds)
                ROWS BETWEEN 3 PRECEDING AND 3 FOLLOWING
            ) as weekly_seasonal
        FROM base
    ),
    -- 合并
    combined AS (
        SELECT 
            t.ds,
            t.intercept + t.slope * EXTRACT(EPOCH FROM t.ds)::bigint as trend,
            s.weekly_seasonal - AVG(t.y) OVER () as seasonal
        FROM trend t
        JOIN seasonal s ON t.ds = s.ds
    )
    SELECT 
        ds,
        trend,
        seasonal,
        trend + seasonal as yhat
    FROM combined;
END;
$$ LANGUAGE plpgsql PARALLEL SAFE;

-- II. 执行预测
SELECT * FROM forecast_prophet_like(12345, 7);

-- III. 与实际值对比（评估准确率）
SELECT 
    f.ds,
    f.yhat as forecasted,
    AVG(sd.temperature) as actual,
    ABS(f.yhat - AVG(sd.temperature)) as mae
FROM forecast_prophet_like(12345, 7) f
LEFT JOIN sensor_data sd ON sd.ts::date = f.ds
GROUP BY f.ds, f.yhat
ORDER BY f.ds;

# 耗时：3.5小时（全表扫描+Pandas处理）
def engineer_features(df):
    # 时滞特征
    for lag in [1, 6, 12, 24]:
        df[f'temp_lag_{lag}h'] = df.groupby('device_id')['temperature'].shift(lag)
    
    # 滚动统计
    df['temp_rolling_mean_6h'] = df.groupby('device_id')['temperature'].rolling(6).mean()
    df['temp_rolling_std_6h'] = df.groupby('device_id')['temperature'].rolling(6).std()
    
    # 差分特征
    df['temp_diff'] = df.groupby('device_id')['temperature'].diff()
    
    return df

-- I. 创建特征工程视图
CREATE MATERIALIZED VIEW device_features_ml
WITH (timescaledb.continuous) AS
SELECT 
    time_bucket(INTERVAL '1 hour', ts) as hour,
    device_id,
    
    -- 时滞特征
    LAG(AVG(temperature), 1) OVER w as temp_lag_1h,
    LAG(AVG(temperature), 6) OVER w as temp_lag_6h,
    LAG(AVG(temperature), 12) OVER w as temp_lag_12h,
    LAG(AVG(temperature), 24) OVER w as temp_lag_24h,
    
    -- 滚动统计
    AVG(AVG(temperature)) OVER w_6h as temp_rolling_mean_6h,
    STDDEV(AVG(temperature)) OVER w_6h as temp_rolling_std_6h,
    
    -- 差分特征
    AVG(temperature) - LAG(AVG(temperature), 1) OVER w as temp_diff
    
FROM sensor_data
GROUP BY hour, device_id
WINDOW 
    w AS (PARTITION BY device_id ORDER BY time_bucket(INTERVAL '1 hour', ts)),
    w_6h AS (PARTITION BY device_id ORDER BY time_bucket(INTERVAL '1 hour', ts) 
             ROWS BETWEEN 5 PRECEDING AND CURRENT ROW);

-- II. 查询特征（直接用于训练）
SELECT * FROM device_features_ml
WHERE hour BETWEEN '2024-01-01' AND '2024-01-31';

-- III. 导出到Python（增量）
COPY (
    SELECT * FROM device_features_ml
    WHERE hour > (SELECT MAX(hour) FROM ml_features_cache)
) TO '/tmp/features_increment.csv' WITH CSV HEADER;

-- 错误：逐条插入（100条/秒）
INSERT INTO sensor_data VALUES (NOW(), 1, 25.3, 60.1, 12.5);
INSERT INTO sensor_data VALUES (NOW(), 2, 26.1, 58.2, 12.3);
-- ...

-- 正确：批量COPY（10万条/秒）
COPY sensor_data (ts, device_id, temperature, humidity, voltage) 
FROM STDIN WITH (FORMAT CSV);
2024-01-15 10:00:00,1,25.3,60.1,12.5
2024-01-15 10:00:00,2,26.1,58.2,12.3
...

-- Python驱动批量写入
import psycopg2
from psycopg2.extras import execute_values

def batch_insert(conn, data_list):
    """
    使用execute_values批量插入
    data_list: [(ts, device_id, temp, hum, volt), ...]
    """
    with conn.cursor() as cur:
        execute_values(
            cur,
            "INSERT INTO sensor_data (ts, device_id, temperature, humidity, voltage) VALUES %s",
            data_list,
            page_size=5000  # 每批5000条
        )
    conn.commit()

-- postgresql.conf
synchronous_commit = off          # 异步提交，提升写入
wal_writer_delay = 10ms           # WAL刷新间隔
wal_compression = on              # WAL压缩

-- 应用层设置（会话级）
SET LOCAL synchronous_commit TO off;

-- III. 使用UNLOGGED表（临时高速写入）
CREATE UNLOGGED TABLE sensor_data_temp (LIKE sensor_data);
-- 批量写入到临时表
COPY sensor_data_temp FROM '/tmp/batch.csv';
-- 再合并到主表
INSERT INTO sensor_data SELECT * FROM sensor_data_temp;

-- 主体索引（自动创建）
CREATE INDEX idx_sensor_data_ts_device ON sensor_data (ts DESC, device_id);

-- 设备级查询优化
CREATE INDEX idx_device_data_lookup ON sensor_data (device_id, ts DESC);

-- 覆盖索引（Index Only Scan）
CREATE INDEX idx_sensor_lookup ON sensor_data 
(device_id, ts DESC) 
INCLUDE (temperature, humidity);

-- BRIN索引（超大规模顺序数据）
CREATE INDEX idx_sensor_brin ON sensor_data 
USING BRIN (ts) WITH (pages_per_range = 128);

-- 索引效果验证
EXPLAIN (ANALYZE, BUFFERS)
SELECT device_id, AVG(temperature)
FROM sensor_data
WHERE ts BETWEEN '2024-01-01' AND '2024-01-02'
GROUP BY device_id;

-- 优化前：Seq Scan，耗时2.3秒
-- 优化后：Index Only Scan，耗时12毫秒

-- 开启JIT（对复杂表达式显著加速）
ALTER SYSTEM SET jit = on;
ALTER SYSTEM SET jit_above_cost = 5000;
SELECT pg_reload_conf();

-- 验证效果
EXPLAIN (ANALYZE, TIMING)
SELECT 
    device_id,
    AVG(temperature),
    STDDEV(temperature),
    PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY temperature)
FROM sensor_data
WHERE ts > NOW() - INTERVAL '1 day'
GROUP BY device_id;

-- 输出应包含:
-- JIT:
--   Functions: 24
--   Options: Inlining true, Optimization true, Expressions true
--   Timing: Generation 1.2ms, Inlining 2.1ms, Optimization 45ms, Emission 12ms

-- 强制并行（会话级）
SET max_parallel_workers_per_gather = 16;
SET parallel_setup_cost = 100;  # 降低并行门槛
SET parallel_tuple_cost = 0.01;

-- 查询所有设备的24小时统计（并行执行）
EXPLAIN (ANALYZE, VERBOSE)
SELECT 
    device_id,
    AVG(temperature) as avg_temp,
    STDDEV(temperature) as std_temp
FROM sensor_data
WHERE ts > NOW() - INTERVAL '1 day'
GROUP BY device_id;

-- 执行计划应显示:
-- Finalize GroupAggregate
--   ->  Gather Merge
--         Workers Planned: 16
--         ->  Partial GroupAggregate
--               ->  Parallel Index Only Scan

-- 查看压缩前大小
SELECT pg_size_pretty(pg_total_relation_size('sensor_data'));
-- 输出: 450 GB

-- 启用压缩（90天后）
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id',
    timescaledb.compress_orderby = 'ts DESC'
);

-- 添加压缩策略
SELECT add_compression_policy('sensor_data', 
    compress_after => '90 days'::interval);

-- 手动触发压缩（首次）
SELECT compress_chunk(c, if_not_compressed => true)
FROM show_chunks('sensor_data', older_than => '90 days'::interval) c;

-- 查看压缩后大小
SELECT pg_size_pretty(pg_total_relation_size('sensor_data'));
-- 输出: 38 GB (压缩率91.5%)

-- 查看压缩元数据
SELECT 
    chunk_name,
    pg_size_pretty(before_compression_total_bytes) as before,
    pg_size_pretty(after Compression_total_bytes) as after,
    round(100 * (1 - after_compression_total_bytes::numeric / before_compression_total_bytes), 2) as compress_ratio
FROM chunk_compression_stats('sensor_data');

-- 输出:
-- chunk_name          | before   | after   | compress_ratio
-- --------------------+----------+---------+----------------
-- _hyper_1_1_chunk    | 5.1 GB   | 480 MB  |          90.58
-- _hyper_1_2_chunk    | 5.2 GB   | 470 MB  |          90.96

-- 查询压缩数据（自动解压）
SELECT device_id, AVG(temperature)
FROM sensor_data
WHERE ts BETWEEN '2023-10-01' AND '2023-10-02';  -- 90天前数据

-- 性能对比：
-- 未压缩chunk: 45毫秒（Index Scan）
-- 压缩chunk: 89毫秒（Decompress+Scan）
-- 开销仅2倍，但节省90%存储

-- 创建监控视图
CREATE OR REPLACE VIEW v_chunk_health AS
SELECT 
    hypertable_schema,
    hypertable_name,
    chunk_schema,
    chunk_name,
    pg_size_pretty(chunk_total_bytes) as total_size,
    pg_size_pretty(chunk_index_bytes) as index_size,
    pg_size_pretty(chunk_toast_bytes) as toast_size,
    is_compressed,
    compression_ratio,
    node_name,
    ranges
FROM chunks_detailed_size('sensor_data')
JOIN timescaledb_information.chunks USING (chunk_name)
ORDER BY chunk_total_bytes DESC;

-- 监控告警：chunk超过10GB
SELECT * FROM v_chunk_health 
WHERE chunk_total_bytes > 10 * 1024^3;

-- 检查CAGG刷新延迟
SELECT 
    view_name,
    last_run_duration,
    last_run_status,
    total_runs,
    total_failures,
    (EXTRACT(EPOCH FROM (NOW() - last_run_started_at))/60)::int as lag_minutes
FROM timescaledb_information.jobs
WHERE proc_name = 'policy_continuous_aggregate';

-- 告警：滞后超过5分钟
DO $$
BEGIN
    IF EXISTS (
        SELECT 1 FROM timescaledb_information.jobs
        WHERE proc_name = 'policy_continuous_aggregate'
          AND EXTRACT(EPOCH FROM (NOW() - last_run_started_at)) > 300
    ) THEN
        RAISE WARNING 'Continuous aggregate lag detected!';
    END IF;
END $$;

-- 监控压缩率下降趋势
SELECT 
    chunk_name,
    before_compression_total_bytes,
    after_compression_total_bytes,
    100 * (1 - after_compression_total_bytes::numeric / before_compression_total_bytes) as ratio,
    CASE 
        WHEN ratio < 80 THEN 'ALERT: Low compression'
        WHEN ratio < 85 THEN 'WARNING: Compression degrading'
        ELSE 'OK'
    END as status
FROM chunk_compression_stats('sensor_data')
WHERE is_compressed = true;

# /etc/prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'postgres'
    static_configs:
      - targets: ['pg-prod:9187']
    metric_relpath: '/metrics'

  - job_name: 'timescaledb'
    static_configs:
      - targets: ['localhost:9201']

{
  "dashboard": {
    "title": "TimescaleDB监控",
    "panels": [
      {
        "title": "Chunk健康度",
        "targets": [
          {
            "expr": "sum(timescaledb_chunk_size_bytes) by (chunk_name)",
            "legendFormat": "{{chunk_name}}"
          }
        ],
        "type": "table",
        "transformations": [
          {
            "id": "organize",
            "options": {
              "include": {
                "chunk_name": true,
                "size": true,
                "is_compressed": true
              }
            }
          }
        ]
      },
      {
        "title": "连续聚合延迟",
        "targets": [
          {
            "expr": "timescaledb_cagg_lag_seconds",
            "thresholds": [
              {
                "colorMode": "critical",
                "op": "gt",
                "value": 300
              }
            ]
          }
        ],
        "type": "stat"
      }
    ]
  }
}

#!/usr/bin/env python3
# auto_resize_chunks.py
import psycopg2
import os

def check_and_resize_chunk(conn):
    """
    监控chunk大小，超过阈值自动拆分
    """
    cur = conn.cursor()
    
    # 查询大chunk
    cur.execute("""
        SELECT chunk_schema, chunk_name, chunk_total_bytes
        FROM chunks_detailed_size('sensor_data')
        WHERE chunk_total_bytes > 10 * 1024^3  -- 10GB
    """)
    
    for schema, name, size in cur.fetchall():
        print(f"大chunk发现: {schema}.{name} ({size/1024**3:.2f}GB)")
        
        # 执行压缩
        cur.execute(f"SELECT compress_chunk('{schema}.{name}')")
        print(f"已压缩: {name}")
        
        # 如果仍然过大，手动拆分
        if size > 20 * 1024**3:
            cur.execute(f"SELECT decompress_chunk('{schema}.{name}')")
            # 拆分逻辑...
    
    conn.commit()

if __name__ == '__main__':
    conn = psycopg2.connect(os.getenv('PG_DSN'))
    check_and_resize_chunk(conn)
    conn.close()

-- 创建函数：终止超过5分钟的查询
CREATE OR REPLACE FUNCTION kill_long_queries()
RETURNS INT AS $$
DECLARE
    killed_count INT := 0;
BEGIN
    SELECT COUNT(*) INTO killed_count
    FROM pg_stat_activity
    WHERE state = 'active'
      AND query LIKE '%sensor_data%'
      AND NOW() - query_start > INTERVAL '5 minutes'
      AND pid <> pg_backend_pid();
    
    -- 终止查询
    PERFORM pg_terminate_backend(pid)
    FROM pg_stat_activity
    WHERE state = 'active'
      AND query LIKE '%sensor_data%'
      AND NOW() - query_start > INTERVAL '5 minutes'
      AND pid <> pg_backend_pid();
    
    RETURN killed_count;
END;
$$ LANGUAGE plpgsql;

-- 定时任务（每分钟执行）
SELECT cron.schedule('kill-long-queries', '* * * * *', 'SELECT kill_long_queries()');

# 日志片段
2024-01-15 14:23:12.123 UTC [12345] FATAL:  out of memory
2024-01-15 14:23:12.123 UTC [12345] DETAIL:  Failed on request of size 1073741824 in memory context "ExecutorState".

-- 诊断
SELECT chunk_name, pg_size_pretty(chunk_total_bytes)
FROM chunks_detailed_size('sensor_data')
ORDER BY chunk_total_bytes DESC LIMIT 5;

-- 结果：_hyper_1_12_chunk | 50 GB | 未压缩

-- 1. 紧急压缩（在线操作）
SELECT compress_chunk('_timescaledb_internal._hyper_1_12_chunk', 
                      if_not_compressed => true);

-- 2. 调整压缩策略（提前压缩）
SELECT remove_compression_policy('sensor_data');
SELECT add_compression_policy('sensor_data', 
    compress_after => '30 days'::interval);  -- 从90天改为30天

-- 3. 设置chunk大小限制（预防）
ALTER TABLE sensor_data SET (
    timescaledb.compress_chunk_time_interval = '7 days'
);

SELECT * FROM timescaledb_information.job_stats 
WHERE proc_name = 'policy_continuous_aggregate';
-- 输出：failure_reason = "deadlock detected"

-- 1. 调整刷新窗口（避开写入高峰）
SELECT remove_continuous_aggregate_policy('sensor_minute');
SELECT add_continuous_aggregate_policy('sensor_minute',
    start_offset => INTERVAL '3 hours',  -- 从2小时改为3小时
    end_offset => INTERVAL '1 minute',
    schedule_interval => INTERVAL '30 seconds',  -- 更频繁但更快
    initial_start => '02:00:00'  -- 凌晨2点执行
);

-- 2. 使用并发刷新（TimescaleDB 2.10+）
SELECT alter_job(job_id, config => '{"concurrent": true}')
FROM timescaledb_information.jobs
WHERE proc_name = 'policy_continuous_aggregate';

-- 1. 紧急终止长事务
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE state = 'idle in transaction'
  AND state_change < NOW() - INTERVAL '1 hour';

-- 2. 手动FREEZE
VACUUM FREEZE sensor_data;

-- 3. 设置告警阈值
ALTER SYSTEM SET vacuum_freeze_min_age = 10000000;  -- 提前FREEZE
SELECT pg_reload_conf();

#!/bin/bash
# backup.sh - 物理备份脚本

# 全量备份（每周日）
pg_basebackup -D /backup/full_$(date +%Y%m%d) -Ft -z -P -X stream

# 增量备份（基于WAL归档）
# postgresql.conf配置
archive_mode = on
archive_command = 'cp %p /backup/wal/%f'

# 使用pgBackRest（推荐）
pgbackrest --stanza=timescaledb backup --type=full

# 恢复
pgbackrest --stanza=timescaledb restore

-- TimescaleDB专用备份（保留超表结构）
-- 备份元数据
pg_dump -d algorithm_db --section=pre-data > /backup/metadata.sql

-- 备份数据（跳过chunk，使用COPY）
SELECT format('COPY %s FROM PROGRAM ''zcat /backup/%s.csv.gz''', 
              chunk_name, chunk_name)
FROM timescaledb_information.chunks
WHERE hypertable_name = 'sensor_data'
ORDER BY range_start;

# 恢复到指定时间点
pgbackrest --stanza=timescaledb restore --set=20240115-020000F \
  --target-time="2024-01-15 14:00:00"

# 启动recovery
pg_ctl -D /data/pgsql/data start

# 连接与认证
max_connections = 500
shared_preload_libraries = 'timescaledb,pg_stat_statements'

# 内存
shared_buffers = 64GB
effective_cache_size = 192GB
maintenance_work_mem = 8GB
work_mem = 256MB

# 写入
wal_buffers = 128MB
synchronous_commit = off
wal_compression = on
checkpoint_timeout = 15min
max_wal_size = 16GB

# 并行
max_worker_processes = 64
max_parallel_workers = 48
max_parallel_workers_per_gather = 16
parallel_setup_cost = 100
parallel_tuple_cost = 0.01

# TimescaleDB
timescaledb.max_background_workers = 16
timescaledb.max_open_chunks_per_insert = 10
timescaledb.max_cached_chunks_per_hypertable = 50

# 查询
random_page_cost = 1.1
effective_io_concurrency = 256
jit = on
jit_above_cost = 5000

# 日志
log_min_duration_statement = '1s'
log_line_prefix = '%t [%p-%l] %q%u@%d '
log_checkpoints = on
log_connections = on
log_temp_files = 0

# 自动维护
autovacuum = on
autovacuum_max_workers = 8
autovacuum_naptime = 30s
autovacuum_vacuum_scale_factor = 0.05
vacuum_freeze_min_age = 10000000

#!/bin/bash
# deploy_timescaledb.sh
set -e

PG_VERSION=${1:-15}
TS_VERSION=${2:-2.10.1}

# 1. 安装PG
apt update && apt install -y postgresql-${PG_VERSION} postgresql-server-dev-${PG_VERSION}

# 2. 安装TimescaleDB
wget https://packagecloud.io/timescale/timescaledb/gpgkey
apt-key add gpgkey
echo "deb https://packagecloud.io/timescale/timescaledb/ubuntu/ jammy main" > /etc/apt/sources.list.d/timescaledb.list
apt update
apt install -y timescaledb-2-postgresql-${PG_VERSION}=${TS_VERSION}

# 3. 配置
timescaledb-tune --quiet --yes

# 4. 重启
systemctl restart postgresql@${PG_VERSION}-main
systemctl enable postgresql@${PG_VERSION}-main

echo "TimescaleDB ${TS_VERSION} on PostgreSQL ${PG_VERSION} installed!"