feat: 更新文档和数据库管理逻辑,添加运行模式说明,修复分区表空间设置
This commit is contained in:
@@ -27,7 +27,7 @@ class DatabaseManager {
|
||||
});
|
||||
|
||||
// 监听连接池错误,防止后端断开导致进程崩溃
|
||||
this.pool.on('error', (err, client) => {
|
||||
this.pool.on('error', (err) => {
|
||||
console.error('[db] 发生未捕获的连接池错误:', err);
|
||||
// 不抛出,让应用层通过心跳检测发现问题
|
||||
});
|
||||
@@ -213,7 +213,7 @@ class DatabaseManager {
|
||||
part_name := heartbeat.partition_name_for_day(p_day);
|
||||
|
||||
EXECUTE format(
|
||||
'CREATE TABLE IF NOT EXISTS heartbeat.%I PARTITION OF heartbeat.heartbeat_events FOR VALUES FROM (%s) TO (%s)',
|
||||
'CREATE TABLE IF NOT EXISTS heartbeat.%I PARTITION OF heartbeat.heartbeat_events FOR VALUES FROM (%s) TO (%s) TABLESPACE ts_hot',
|
||||
part_name, start_ms, end_ms
|
||||
);
|
||||
END;
|
||||
@@ -316,10 +316,21 @@ class DatabaseManager {
|
||||
|
||||
getPartitionConfig() {
|
||||
const cfg = this.config.partitionMaintenance ?? {};
|
||||
const futureDaysRaw = Number(cfg.futureDays);
|
||||
const intervalHoursRaw = Number(cfg.intervalHours);
|
||||
|
||||
const futureDays = Number.isFinite(futureDaysRaw) && futureDaysRaw > 0
|
||||
? Math.floor(futureDaysRaw)
|
||||
: 30;
|
||||
|
||||
const intervalHours = Number.isFinite(intervalHoursRaw) && intervalHoursRaw > 0
|
||||
? intervalHoursRaw
|
||||
: 6;
|
||||
|
||||
return {
|
||||
enabled: cfg.enabled !== false,
|
||||
futureDays: Number.isFinite(cfg.futureDays) ? cfg.futureDays : 30,
|
||||
intervalHours: Number.isFinite(cfg.intervalHours) ? cfg.intervalHours : 6,
|
||||
futureDays,
|
||||
intervalHours,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -490,24 +501,21 @@ class DatabaseManager {
|
||||
await this.ensurePartitionsForTsRange(Math.min(...tsValues), Math.max(...tsValues));
|
||||
}
|
||||
|
||||
const generateRows = async function* () {
|
||||
for (const e of events) {
|
||||
const rowValues = toRowValues(e);
|
||||
const line = rowValues.map(formatPgCol).join('\t') + '\n';
|
||||
yield line;
|
||||
}
|
||||
};
|
||||
|
||||
const client = await this.pool.connect();
|
||||
try {
|
||||
const copySql = `COPY heartbeat.heartbeat_events (${columns.join(', ')}) FROM STDIN WITH (FORMAT text, DELIMITER E'\\t', NULL '\\N')`;
|
||||
const stream = client.query(copyFrom(copySql));
|
||||
|
||||
// Use a generator to stream rows directly
|
||||
async function* generateRows() {
|
||||
for (const e of events) {
|
||||
const rowValues = toRowValues(e);
|
||||
const line = rowValues.map(formatPgCol).join('\t') + '\n';
|
||||
yield line;
|
||||
}
|
||||
}
|
||||
|
||||
await pipeline(Readable.from(generateRows()), stream);
|
||||
return { insertedCount: events.length };
|
||||
} catch (error) {
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user