完整测试代码分享，含详细注释，可直接运行
前言
上一篇文章我们介绍了 TimescaleDB 的核心概念和架构，今天带来完整的性能对比测试代码。本文将展示如何：
1. 创建普通表 vs TimescaleDB 超表 2. 生成 200 万条模拟传感器数据 3. 启用压缩并优化存储 4. 创建连续聚合视图 5. 运行 4 种典型查询场景的性能对比
测试环境准备
第 1 步：清理环境并创建表
-- 删除已存在的表和函数
DROP TABLE IF EXISTS temperatures_regular CASCADE;
DROP TABLE IF EXISTS temperatures_hypertable CASCADE;
DROP FUNCTION IF EXISTS generate_sensor_data(TIMESTAMPTZ, TIMESTAMPTZ, INT, TEXT) CASCADE;
DROP FUNCTION IF EXISTS run_performance_test() CASCADE;
-- 创建普通表
CREATE TABLE temperatures_regular (
time TIMESTAMPTZ NOT NULL,
sensor_id INT NOT NULL,
temperature DOUBLE PRECISION NOT NULL,
humidity DOUBLE PRECISION,
battery_voltage FLOAT,
location_id INT
);
-- 创建 TimescaleDB 超表
CREATE TABLE temperatures_hypertable (
time TIMESTAMPTZ NOT NULL,
sensor_id INT NOT NULL,
temperature DOUBLE PRECISION NOT NULL,
humidity DOUBLE PRECISION,
battery_voltage FLOAT,
location_id INT
);
-- 将表转换为超表（支持空间分区）
SELECT create_hypertable(
'temperatures_hypertable', 'time',
partitioning_column => 'sensor_id',
number_partitions => 8
);
-- 创建索引
CREATE INDEX idx_regular_time_sensor ON temperatures_regular (time DESC, sensor_id);
CREATE INDEX idx_hypertable_time_sensor ON temperatures_hypertable (time DESC, sensor_id);
CREATE INDEX idx_regular_location ON temperatures_regular (location_id, time DESC);
CREATE INDEX idx_hypertable_location ON temperatures_hypertable (location_id, time DESC);
💡 关键点：超表支持空间分区（
partitioning_column
），可以将sensor_id
作为二次分区键，实现更细粒度的数据分布。
第 2 步：创建 200 万数据生成函数
CREATE OR REPLACE FUNCTION generate_sensor_data(
start_time TIMESTAMPTZ,
end_time TIMESTAMPTZ,
num_sensors INT,
target_table TEXT
) RETURNS INT AS $$
DECLARE
curr_time TIMESTAMPTZ;
sensor_id INT;
temp_val DOUBLE PRECISION;
humid_val DOUBLE PRECISION;
total_inserted INT := 0;
batch_size INT := 10000;
BEGIN
curr_time := start_time;
WHILE curr_time < end_time LOOP
FOR sensor_id IN 1..num_sensors LOOP
-- 生成模拟数据：温度 + 正弦波动 + 随机噪声
temp_val := 20 + SIN(EXTRACT(HOUR FROM curr_time) * 15 * 3.14159/180)
+ (sensor_id % 5) * 0.5 + random();
humid_val := 60 - (temp_val - 20) * 2 + random() * 5;
IF target_table = 'regular' THEN
INSERT INTO temperatures_regular VALUES
(curr_time, sensor_id, temp_val, humid_val, random() * 0.3, (sensor_id % 10) + 1);
ELSE
INSERT INTO temperatures_hypertable VALUES
(curr_time, sensor_id, temp_val, humid_val, random() * 0.3, (sensor_id % 10) + 1);
END IF;
total_inserted := total_inserted + 1;
END LOOP;
curr_time := curr_time + INTERVAL '5 seconds';
END LOOP;
RETURN total_inserted;
END;
$$ LANGUAGE plpgsql;
📊 数据生成逻辑：
• 20 个传感器 × 6 天数据 × 每 5 秒一条 ≈ 200 万条记录 • 温度模拟：基础 20°C + 正弦日夜波动 + 传感器差异 + 随机噪声
第 3 步：生成测试数据
-- 向普通表插入数据
SELECT generate_sensor_data(
'2026-03-19 00:00:00'::TIMESTAMPTZ,
'2026-03-25 00:00:00'::TIMESTAMPTZ,
20,
'regular'
);
-- 向 TimescaleDB 超表插入数据
SELECT generate_sensor_data(
'2026-03-19 00:00:00'::TIMESTAMPTZ,
'2026-03-25 00:00:00'::TIMESTAMPTZ,
20,
'hypertable'
);
第 4 步：启用压缩并优化
-- 启用压缩
ALTER TABLE temperatures_hypertable SET (
timescaledb.compress,
timescaledb.compress_segmentby = 'sensor_id, location_id',
timescaledb.compress_orderby = 'time DESC'
);
-- 压缩所有旧数据（超过 1 小时的都压缩）
SELECT compress_chunk(chunk)
FROM show_chunks('temperatures_hypertable', older_than => INTERVAL '1 hour');
-- 更新统计信息
ANALYZE temperatures_regular;
ANALYZE temperatures_hypertable;
🔬 压缩配置说明：
• compress_segmentby
：按传感器 ID 和位置 ID 分段压缩，优化按设备查询• compress_orderby
：按时间倒序排列，利用时间局部性提升压缩比
第 5 步：创建完整的性能测试函数
CREATE OR REPLACE FUNCTION run_performance_test()
RETURNS TABLE(
test_case TEXT,
regular_time_ms NUMERIC,
hypertable_time_ms NUMERIC,
regular_rows BIGINT,
hypertable_rows BIGINT,
speedup_factor NUMERIC
) AS $$
DECLARE
start_time TIMESTAMPTZ;
end_time TIMESTAMPTZ;
regular_count BIGINT;
hypertable_count BIGINT;
regular_duration INTERVAL;
hypertable_duration INTERVAL;
BEGIN
-- ========================================
-- 测试 1：最新 1 小时数据查询
-- ========================================
start_time := clock_timestamp();
SELECT COUNT(*) INTO regular_count
FROM temperatures_regular
WHERE time > '2026-03-20 00:00:00'::TIMESTAMPTZ - INTERVAL '1 hour'
AND sensor_id = 5;
regular_duration := clock_timestamp() - start_time;
regular_time_ms := EXTRACT(EPOCH FROM regular_duration) * 1000;
start_time := clock_timestamp();
SELECT COUNT(*) INTO hypertable_count
FROM temperatures_hypertable
WHERE time > '2026-03-20 00:00:00'::TIMESTAMPTZ - INTERVAL '1 hour'
AND sensor_id = 5;
hypertable_duration := clock_timestamp() - start_time;
hypertable_time_ms := EXTRACT(EPOCH FROM hypertable_duration) * 1000;
test_case := '最新1小时数据查询';
regular_rows := regular_count;
hypertable_rows := hypertable_count;
speedup_factor := ROUND(regular_time_ms / NULLIF(hypertable_time_ms, 0), 2);
RETURN NEXT;
-- ========================================
-- 测试 2：特定传感器全天数据查询
-- ========================================
start_time := clock_timestamp();
SELECT COUNT(*), AVG(temperature), MAX(temperature), MIN(temperature)
INTO regular_count
FROM temperatures_regular
WHERE time >= '2026-03-18 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-19 00:00:00'::TIMESTAMPTZ
AND sensor_id = 10;
regular_duration := clock_timestamp() - start_time;
regular_time_ms := EXTRACT(EPOCH FROM regular_duration) * 1000;
start_time := clock_timestamp();
SELECT COUNT(*), AVG(temperature), MAX(temperature), MIN(temperature)
INTO hypertable_count
FROM temperatures_hypertable
WHERE time >= '2026-03-18 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-19 00:00:00'::TIMESTAMPTZ
AND sensor_id = 10;
hypertable_duration := clock_timestamp() - start_time;
hypertable_time_ms := EXTRACT(EPOCH FROM hypertable_duration) * 1000;
test_case := '特定传感器全天数据查询';
regular_rows := regular_count;
hypertable_rows := hypertable_count;
speedup_factor := ROUND(regular_time_ms / NULLIF(hypertable_time_ms, 0), 2);
RETURN NEXT;
-- ========================================
-- 测试 3：按小时聚合查询
-- ========================================
start_time := clock_timestamp();
SELECT COUNT(*) INTO regular_count
FROM (
SELECT
time_bucket('1 hour', time) AS hour_bucket,
sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp
FROM temperatures_regular
WHERE time >= '2026-03-15 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-17 00:00:00'::TIMESTAMPTZ
GROUP BY hour_bucket, sensor_id
) subq;
regular_duration := clock_timestamp() - start_time;
regular_time_ms := EXTRACT(EPOCH FROM regular_duration) * 1000;
start_time := clock_timestamp();
SELECT COUNT(*) INTO hypertable_count
FROM (
SELECT
time_bucket('1 hour', time) AS hour_bucket,
sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp
FROM temperatures_hypertable
WHERE time >= '2026-03-15 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-17 00:00:00'::TIMESTAMPTZ
GROUP BY hour_bucket, sensor_id
) subq;
hypertable_duration := clock_timestamp() - start_time;
hypertable_time_ms := EXTRACT(EPOCH FROM hypertable_duration) * 1000;
test_case := '按小时聚合查询';
regular_rows := regular_count;
hypertable_rows := hypertable_count;
speedup_factor := ROUND(regular_time_ms / NULLIF(hypertable_time_ms, 0), 2);
RETURN NEXT;
-- ========================================
-- 测试 4：多传感器统计分析
-- ========================================
start_time := clock_timestamp();
SELECT COUNT(*) INTO regular_count
FROM (
SELECT
sensor_id,
location_id,
AVG(temperature) as avg_temp,
STDDEV(temperature) as stddev_temp,
AVG(humidity) as avg_humidity,
COUNT(*) as reading_count
FROM temperatures_regular
WHERE time >= '2026-03-14 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-20 00:00:00'::TIMESTAMPTZ
AND sensor_id IN (1, 3, 5, 7, 9)
GROUP BY sensor_id, location_id
HAVING COUNT(*) > 1000
) subq;
regular_duration := clock_timestamp() - start_time;
regular_time_ms := EXTRACT(EPOCH FROM regular_duration) * 1000;
start_time := clock_timestamp();
SELECT COUNT(*) INTO hypertable_count
FROM (
SELECT
sensor_id,
location_id,
AVG(temperature) as avg_temp,
STDDEV(temperature) as stddev_temp,
AVG(humidity) as avg_humidity,
COUNT(*) as reading_count
FROM temperatures_hypertable
WHERE time >= '2026-03-14 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-20 00:00:00'::TIMESTAMPTZ
AND sensor_id IN (1, 3, 5, 7, 9)
GROUP BY sensor_id, location_id
HAVING COUNT(*) > 1000
) subq;
hypertable_duration := clock_timestamp() - start_time;
hypertable_time_ms := EXTRACT(EPOCH FROM hypertable_duration) * 1000;
test_case := '多传感器统计分析';
regular_rows := regular_count;
hypertable_rows := hypertable_count;
speedup_factor := ROUND(regular_time_ms / NULLIF(hypertable_time_ms, 0), 2);
RETURN NEXT;
RETURN;
END;
$$ LANGUAGE plpgsql;
第 6 步：创建连续聚合视图
-- 创建连续聚合视图
CREATE MATERIALIZED VIEW temperatures_hourly_stats
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 hour', time) AS bucket,
sensor_id,
location_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp,
AVG(humidity) as avg_humidity,
COUNT(*) as readings_count
FROM temperatures_hypertable
GROUP BY bucket, sensor_id, location_id;
-- 设置自动刷新策略
SELECT add_continuous_aggregate_policy(
'temperatures_hourly_stats',
start_offset => INTERVAL '2 days',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '1 hour'
);
⚡ 连续聚合是 TimescaleDB 的核心特性之一：
• 自动增量更新，无需每次全量计算 • 查询速度提升 100-1000 倍 • 后台自动刷新，不影响写入性能
第 7 步：运行性能测试
-- 运行完整的性能测试
SELECT * FROM run_performance_test();
预期输出示例：
第 8 步：生成存储分析报告
WITH
regular_stats AS (
SELECT COUNT(*) as records,
pg_total_relation_size('temperatures_regular') as bytes
FROM temperatures_regular
),
tsdb_chunks AS (
SELECT SUM(pg_total_relation_size('_timescaledb_internal.' || tablename)) as chunk_bytes
FROM pg_tables
WHERE schemaname = '_timescaledb_internal'
AND tablename LIKE 'compress_hyper_%'
),
tsdb_main AS (
SELECT COUNT(*) as records,
pg_total_relation_size('temperatures_hypertable') as main_bytes
FROM temperatures_hypertable
),
perf_results AS (
SELECT AVG(speedup_factor) as avg_speedup
FROM run_performance_test()
)
SELECT
'📈 最终测试报告' as "报告项目",
regular_stats.records || ' 条' as "数据总量",
pg_size_pretty(regular_stats.bytes) as "普通表存储",
pg_size_pretty(COALESCE(tsdb_chunks.chunk_bytes, 0) + tsdb_main.main_bytes) as "TimescaleDB存储",
ROUND(100.0 * (COALESCE(tsdb_chunks.chunk_bytes, 0) + tsdb_main.main_bytes) / NULLIF(regular_stats.bytes, 0), 2) || '%' as "存储占比",
ROUND(perf_results.avg_speedup, 2) || '倍' as "平均性能提升",
CASE
WHEN perf_results.avg_speedup > 5 THEN '⭐ 卓越'
WHEN perf_results.avg_speedup > 2 THEN '✅ 良好'
ELSE '👍 合格'
END as "评价"
FROM regular_stats, tsdb_chunks, tsdb_main, perf_results;
预期输出示例：
测试结果解读
存储对比
查询性能对比
| 10-50 倍 | |
| 15-25 倍 | |
| 25-35 倍 | |
| 25-30 倍 |
📌 性能提升主要来源：
1. 分区剪枝：查询自动跳过不相关的 Chunk 2. 列式压缩：减少 I/O，提升扫描效率 3. 连续聚合：预计算结果，无需实时聚合
进阶：使用连续聚合视图查询
-- 直接查询连续聚合视图（极快！）
SELECT
bucket,
sensor_id,
avg_temp,
max_temp,
min_temp,
readings_count
FROM temperatures_hourly_stats
WHERE bucket >= '2026-03-18 00:00:00'::TIMESTAMPTZ
AND bucket < '2026-03-19 00:00:00'::TIMESTAMPTZ
AND sensor_id = 5
ORDER BY bucket;
-- 对比：从原始表查询（较慢）
SELECT
time_bucket('1 hour', time) AS bucket,
sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp,
COUNT(*) as readings_count
FROM temperatures_hypertable
WHERE time >= '2026-03-18 00:00:00'::TIMESTAMPTZ
AND time < '2026-03-19 00:00:00'::TIMESTAMPTZ
AND sensor_id = 5
GROUP BY bucket, sensor_id
ORDER BY bucket;
清理测试环境
-- 删除连续聚合视图
DROP MATERIALIZED VIEW IF EXISTS temperatures_hourly_stats CASCADE;
-- 删除测试表
DROP TABLE IF EXISTS temperatures_regular CASCADE;
DROP TABLE IF EXISTS temperatures_hypertable CASCADE;
-- 删除函数
DROP FUNCTION IF EXISTS generate_sensor_data(TIMESTAMPTZ, TIMESTAMPTZ, INT, TEXT) CASCADE;
DROP FUNCTION IF EXISTS run_performance_test() CASCADE;
总结
通过这篇完整的性能测试代码，我们可以看到 TimescaleDB 在时序数据场景下的显著优势：
1. 存储节省：列式压缩可节省 65-75% 存储空间 2. 查询加速：分区剪枝 + 压缩 + 连续聚合，综合提升 15-35 倍 3. 使用简单：完全兼容 PostgreSQL SQL，无需改写应用代码 4. 运维友好：自动分区管理、自动压缩策略、自动聚合刷新
适用场景
• ✅ 物联网监控：百万级传感器数据实时采集 • ✅ 金融行情：高频交易数据存储与分析 • ✅ 应用性能监控：APM 指标数据聚合查询 • ✅ 能源管理：智能电表、环境监测数据
下次我们将分享如何在生产环境中部署 TimescaleDB，以及监控和调优的最佳实践。
感谢阅读！欢迎关注「PostgreSQL学习」合集，获取更多技术干货。如果觉得有帮助，欢迎一键三连支持！
参考资料
• TimescaleDB 官方文档[1] • TimescaleDB GitHub[2] • PostgreSQL 官方文档[3]
引用链接
[1]
TimescaleDB 官方文档: https://docs.timescaledb.cn[2]
TimescaleDB GitHub: https://github.com/timescale/timescaledb[3]
PostgreSQL 官方文档: https://www.postgresql.org/docs/
文章转载自绩隐金，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
