Commit 1fa47b9d by 久林

fix: 修复 timeseries 合并损坏导致服务端无法启动

Co-authored-by: Cursor <cursoragent@cursor.com>
parent e3b335fb
...@@ -362,7 +362,7 @@ export function resolveTsTimeRange(opts: { ...@@ -362,7 +362,7 @@ export function resolveTsTimeRange(opts: {
} }
if (toMs - fromMs > maxMs) { if (toMs - fromMs > maxMs) {
throw new Error(查询跨度不能超过 天) throw new Error(`查询跨度不能超过 ${maxHistorySpanDays()} 天`)
} }
if (toMs - fromMs < 60_000) { if (toMs - fromMs < 60_000) {
throw new Error('查询跨度至少 1 分钟') throw new Error('查询跨度至少 1 分钟')
...@@ -436,7 +436,7 @@ export async function queryTsSeries(opts: { ...@@ -436,7 +436,7 @@ export async function queryTsSeries(opts: {
let deviceFilter = '' let deviceFilter = ''
if (opts.deviceIds?.length) { if (opts.deviceIds?.length) {
params.push(opts.deviceIds) params.push(opts.deviceIds)
deviceFilter = AND device_id = ANY({params.length}::text[]) deviceFilter = ` AND device_id = ANY($${params.length}::text[])`
} }
let { rows: rollupRows } = await query<{ let { rows: rollupRows } = await query<{
...@@ -444,12 +444,12 @@ export async function queryTsSeries(opts: { ...@@ -444,12 +444,12 @@ export async function queryTsSeries(opts: {
bucket: Date bucket: Date
avg_eng: string | number avg_eng: string | number
}>( }>(
SELECT device_id, bucket, avg_eng `SELECT device_id, bucket, avg_eng
FROM ts_rollups FROM ts_rollups
WHERE source = AND metric = AND bucket_minutes = WHERE source = $1 AND metric = $2 AND bucket_minutes = $5
AND bucket >= AND bucket <= AND bucket >= $3 AND bucket <= $4
${deviceFilter}
ORDER BY device_id, bucket, ORDER BY device_id, bucket`,
params, params,
) )
...@@ -460,19 +460,19 @@ export async function queryTsSeries(opts: { ...@@ -460,19 +460,19 @@ export async function queryTsSeries(opts: {
let fbFilter = '' let fbFilter = ''
if (opts.deviceIds?.length) { if (opts.deviceIds?.length) {
fbParams.push(opts.deviceIds) fbParams.push(opts.deviceIds)
fbFilter = AND device_id = ANY({fbParams.length}::text[]) fbFilter = ` AND device_id = ANY($${fbParams.length}::text[])`
} }
const fallback = await query<{ const fallback = await query<{
device_id: string device_id: string
bucket: Date bucket: Date
avg_eng: string | number avg_eng: string | number
}>( }>(
SELECT device_id, bucket, avg_eng `SELECT device_id, bucket, avg_eng
FROM ts_rollups FROM ts_rollups
WHERE source = AND metric = AND bucket_minutes = WHERE source = $1 AND metric = $2 AND bucket_minutes = $5
AND bucket >= AND bucket <= AND bucket >= $3 AND bucket <= $4
${fbFilter}
ORDER BY device_id, bucket, ORDER BY device_id, bucket`,
fbParams, fbParams,
) )
if (fallback.rows.length) { if (fallback.rows.length) {
...@@ -488,12 +488,12 @@ export async function queryTsSeries(opts: { ...@@ -488,12 +488,12 @@ export async function queryTsSeries(opts: {
bucket: Date bucket: Date
avg_eng: string | number avg_eng: string | number
}>( }>(
SELECT device_id, bucket, avg_eng `SELECT device_id, bucket, avg_eng
FROM ts_rollups FROM ts_rollups
WHERE source = AND metric = AND bucket_minutes = 15 WHERE source = $1 AND metric = $2 AND bucket_minutes = 15
AND bucket >= AND bucket <= AND bucket >= $3 AND bucket <= $4
${deviceFilter}
ORDER BY device_id, bucket, ORDER BY device_id, bucket`,
params, params,
) )
rollupRows = fallback.rows rollupRows = fallback.rows
...@@ -510,16 +510,16 @@ export async function queryTsSeries(opts: { ...@@ -510,16 +510,16 @@ export async function queryTsSeries(opts: {
bucket: Date bucket: Date
avg_eng: string | number avg_eng: string | number
}>( }>(
SELECT `SELECT
device_id, device_id,
to_timestamp(floor(extract(epoch FROM ts) / ( * 60)) * ( * 60)) AS bucket, to_timestamp(floor(extract(epoch FROM ts) / ($5 * 60)) * ($5 * 60)) AS bucket,
AVG(eng) AS avg_eng AVG(eng) AS avg_eng
FROM ts_samples FROM ts_samples
WHERE source = AND metric = WHERE source = $1 AND metric = $2
AND ts >= AND ts <= AND ts >= $3 AND ts <= $4
${deviceFilter}
GROUP BY device_id, bucket GROUP BY device_id, bucket
ORDER BY device_id, bucket, ORDER BY device_id, bucket`,
params, params,
) )
...@@ -557,7 +557,7 @@ export async function queryTsExportRows(opts: { ...@@ -557,7 +557,7 @@ export async function queryTsExportRows(opts: {
let deviceFilter = '' let deviceFilter = ''
if (opts.deviceIds?.length) { if (opts.deviceIds?.length) {
params.push(opts.deviceIds) params.push(opts.deviceIds)
deviceFilter = AND device_id = ANY({params.length}::text[]) deviceFilter = ` AND device_id = ANY($${params.length}::text[])`
} }
const { rows: raw } = await query<{ const { rows: raw } = await query<{
...@@ -568,13 +568,13 @@ export async function queryTsExportRows(opts: { ...@@ -568,13 +568,13 @@ export async function queryTsExportRows(opts: {
eng: string | number eng: string | number
raw: string | number | null raw: string | number | null
}>( }>(
SELECT ts, device_id, source, metric, eng, raw `SELECT ts, device_id, source, metric, eng, raw
FROM ts_samples FROM ts_samples
WHERE source = AND metric = WHERE source = $1 AND metric = $2
AND ts >= AND ts <= AND ts >= $3 AND ts <= $4
${deviceFilter}
ORDER BY device_id, ts ORDER BY device_id, ts
LIMIT , LIMIT $5`,
params, params,
) )
...@@ -602,7 +602,7 @@ export async function queryTsExportRows(opts: { ...@@ -602,7 +602,7 @@ export async function queryTsExportRows(opts: {
let rollDeviceFilter = '' let rollDeviceFilter = ''
if (opts.deviceIds?.length) { if (opts.deviceIds?.length) {
rollParams.push(opts.deviceIds) rollParams.push(opts.deviceIds)
rollDeviceFilter = AND device_id = ANY({rollParams.length}::text[]) rollDeviceFilter = ` AND device_id = ANY($${rollParams.length}::text[])`
} }
rollParams.push(EXPORT_ROW_LIMIT + 1) rollParams.push(EXPORT_ROW_LIMIT + 1)
const limitIdx = rollParams.length const limitIdx = rollParams.length
...@@ -613,13 +613,13 @@ export async function queryTsExportRows(opts: { ...@@ -613,13 +613,13 @@ export async function queryTsExportRows(opts: {
metric: string metric: string
avg_eng: string | number avg_eng: string | number
}>( }>(
SELECT bucket, device_id, source, metric, avg_eng `SELECT bucket, device_id, source, metric, avg_eng
FROM ts_rollups FROM ts_rollups
WHERE source = AND metric = AND bucket_minutes = WHERE source = $1 AND metric = $2 AND bucket_minutes = $5
AND bucket >= AND bucket <= AND bucket >= $3 AND bucket <= $4
${rollDeviceFilter}
ORDER BY device_id, bucket ORDER BY device_id, bucket
LIMIT {limitIdx}, LIMIT $${limitIdx}`,
rollParams, rollParams,
) )
if (!roll.length) continue if (!roll.length) continue
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment