新增 Kafka 消息中 electricity[] 和 air_conditioner[] 数组字段支持,用于存储电力与空调设备明细数据。数据库表新增对应数组列并创建 GIN 索引优化查询性能,processor 实现数组字段校验与聚合转换逻辑。 主要变更: - Kafka 消息规范新增 electricity 和 air_conditioner 数组字段定义 - 数据库 heartbeat_events 表新增 14 个数组列并创建 4 个 GIN 索引 - processor 实现数组字段解析、校验及聚合转换逻辑 - 更新相关文档与测试用例,确保端到端功能完整
130 lines
3.2 KiB
JavaScript
130 lines
3.2 KiB
JavaScript
import { Client } from 'pg';
|
|
import config from '../../src/config/config.js';
|
|
|
|
async function main() {
|
|
const client = new Client({
|
|
host: config.db.host,
|
|
port: config.db.port,
|
|
user: config.db.user,
|
|
password: config.db.password,
|
|
database: config.db.database,
|
|
});
|
|
|
|
await client.connect();
|
|
|
|
// 预创建今日分区,避免“无分区时 INSERT 直接失败”
|
|
await client.query('SELECT heartbeat.ensure_partitions(current_date, current_date)');
|
|
|
|
const ipType = await client.query(
|
|
`
|
|
SELECT format_type(a.atttypid, a.atttypmod) AS type
|
|
FROM pg_attribute a
|
|
JOIN pg_class c ON c.oid = a.attrelid
|
|
JOIN pg_namespace n ON n.oid = c.relnamespace
|
|
WHERE n.nspname = 'heartbeat'
|
|
AND c.relname = 'heartbeat_events'
|
|
AND a.attname = 'ip'
|
|
AND a.attnum > 0
|
|
AND NOT a.attisdropped
|
|
`
|
|
);
|
|
|
|
const type = String(ipType?.rows?.[0]?.type ?? '').toLowerCase();
|
|
if (type.startsWith('inet')) {
|
|
await client.query(
|
|
`ALTER TABLE heartbeat.heartbeat_events
|
|
ALTER COLUMN ip TYPE varchar(21)
|
|
USING ip::text`
|
|
);
|
|
}
|
|
|
|
const ts = Date.now();
|
|
await client.query(
|
|
`INSERT INTO heartbeat.heartbeat_events (
|
|
ts_ms, hotel_id, room_id, device_id, ip,
|
|
power_state, guest_type, cardless_state, service_mask,
|
|
pms_state, carbon_state, device_count, comm_seq,
|
|
elec_address, voltage, ampere, power, phase, energy, sum_energy,
|
|
air_address, state, model, speed, set_temp, now_temp, solenoid_valve,
|
|
extra
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28)`,
|
|
[
|
|
ts,
|
|
1,
|
|
101,
|
|
'dev-1',
|
|
'192.168.0.1:12345',
|
|
1,
|
|
0,
|
|
0,
|
|
5,
|
|
0,
|
|
0,
|
|
1,
|
|
1,
|
|
['add11', 'add12'],
|
|
[3.2, 3.3],
|
|
[1.1, 2.2],
|
|
[704.3, 705.3],
|
|
['A', 'B'],
|
|
[10.5, 11.5],
|
|
[100.5, 101.5],
|
|
['ac1'],
|
|
[1],
|
|
[2],
|
|
[3],
|
|
[26],
|
|
[25],
|
|
[1],
|
|
{ source: 'smoke-test' },
|
|
]
|
|
);
|
|
|
|
const partitions = await client.query(
|
|
`SELECT c.relname AS partition
|
|
FROM pg_inherits i
|
|
JOIN pg_class c ON c.oid = i.inhrelid
|
|
JOIN pg_class p ON p.oid = i.inhparent
|
|
JOIN pg_namespace n ON n.oid = p.relnamespace
|
|
WHERE n.nspname = 'heartbeat'
|
|
AND p.relname = 'heartbeat_events'
|
|
ORDER BY c.relname`
|
|
);
|
|
|
|
const parent = await client.query(
|
|
`
|
|
SELECT c.relkind AS kind
|
|
FROM pg_class c
|
|
JOIN pg_namespace n ON n.oid = c.relnamespace
|
|
WHERE n.nspname = 'heartbeat'
|
|
AND c.relname = 'heartbeat_events'
|
|
`
|
|
);
|
|
|
|
const parentIndexes = await client.query(
|
|
`
|
|
SELECT indexname
|
|
FROM pg_indexes
|
|
WHERE schemaname = 'heartbeat'
|
|
AND tablename = 'heartbeat_events'
|
|
ORDER BY indexname
|
|
`
|
|
);
|
|
|
|
const cnt = await client.query(
|
|
'SELECT count(*)::int AS n FROM heartbeat.heartbeat_events'
|
|
);
|
|
|
|
console.log('parentKind:', parent.rows?.[0]?.kind);
|
|
console.log('partitions:', partitions.rows.map((r) => r.partition));
|
|
console.log('parentIndexes:', parentIndexes.rows.map((r) => r.indexname));
|
|
console.log('rows:', cnt.rows[0].n);
|
|
|
|
await client.end();
|
|
}
|
|
|
|
main().catch((err) => {
|
|
console.error('smoke test failed:', err);
|
|
process.exit(1);
|
|
});
|