fix: keep bulk purge reference lookups on indexes
This commit is contained in:
parent
a1ffe3b106
commit
5b07f5ca54
@ -6,6 +6,8 @@
|
||||
|
||||
索引仅返回候选。应用仍用完整存储类型、桶及路径判定共享引用；哈希冲突只会增加候选。回收站素材、软删除 SKU、历史对象地址、白底派生文件和活动上传/审批引用继续受到保护。URL 点段和控制字符的少量特殊引用保留为索引候选，再交由原生 URL 规则核对；Windows Unicode 路径保守匹配。SQL 函数明确禁用并行执行，避免错误恢复子事务在并行查询内报错。
|
||||
|
||||
线上只读验收发现，较大的检索键数组在 PostgreSQL 的 custom plan 中被误估为命中大量数据，前五次查询仍选择全表扫描；之后 generic plan 才使用索引。查询改用 MATERIALIZED 参数和 LATERAL 边界，在首次及后续调用都通过参数化索引查找；不会改数据库全局设置。
|
||||
|
||||
提交后立即唤醒已经启动的清理 worker。空队列检查与入队交错时保留唤醒信号，每轮满 20 条立即接着处理下一轮；5 秒定时器只用于恢复和到期重试。没有把排队或未完成的物理清理显示成成功。
|
||||
|
||||
前端只在任务进行中且进度接口正常时每 500ms 更新状态，完成后及接口报错时回到 3 秒间隔；页面隐藏时停止请求，已有请求未完成时不重叠请求。完成提示仍由服务端真实完成状态驱动，避免后台已经完成但界面固定多等 3 秒。
|
||||
@ -13,12 +15,15 @@
|
||||
## 验证
|
||||
|
||||
- 23 个相关 Jest 测试，包括入队与空队列检查交错时不会丢失唤醒和进行中状态的 500ms 刷新。
|
||||
- 14 个真实 PostgreSQL/文件系统/MinIO 集成验证组；包括幂等、事务回滚、租约接管、真实删除、共享保护、直接 SQL 更新、队列状态变化、特殊地址和失败续跑。
|
||||
- 独立数据库 400,000 条额外资产：引用检查 67ms，提交 31ms，4 个文件从提交到实际删除及审计完成 141ms；资产查询计划实际使用 `material_assets_purge_references_idx`。
|
||||
- 已启动 worker 收到提交后 92ms 完成，不等待 5 秒轮询。
|
||||
- 15 个真实 PostgreSQL/文件系统/MinIO 集成验证组；包括幂等、事务回滚、租约接管、真实删除、共享保护、直接 SQL 更新、队列状态变化、特殊地址和失败续跑。
|
||||
- 独立数据库 400,000 条额外资产：引用检查 56ms，提交 32ms，4 个文件从提交到实际删除及审计完成 136ms；资产查询计划实际使用 `material_assets_purge_references_idx`。
|
||||
- 107 个文件在强制 custom/generic 两种计划下均通过索引查找并保护共享文件，耗时 699ms / 702ms；验收保留刚批量写入的大量索引待合并项。
|
||||
- 已启动 worker 收到提交后 93ms 完成，不等待 5 秒轮询。
|
||||
- 白底派生地址中的原始查询、片段和点段与原生 URL 解析一致；真实 MinIO 集成验证不会漏掉共享文件。
|
||||
- 类型检查、存储边界检查和 14 个安全迁移测试通过。迁移审计核对九个索引的表达式、状态谓词、有效性及三个函数的完整定义。
|
||||
|
||||
生产验收只读核对原先 18 个任务的 1,276 个文件和现存引用样本，不由自动化提交真实素材删除。隔离测试耗时不等于生产完整物理清理耗时；生产查询耗时和发布凭据另存于发布回执。
|
||||
|
||||
最终查询上线前用生产数据只读复核：全量扫描 18,324ms；18 个任务、1,276 个文件逐任务索引检查合计 396ms，每任务 7–57ms，引用结果零差异，9 个现存共享引用样本保留。最大 106 个文件任务的实际执行计划使用素材与资产引用索引，SQL 执行 14.572ms。
|
||||
|
||||
镜像构建将提交 SHA 参数放在固定系统依赖层之后，避免每次提交使编译器安装缓存失效；镜像标签、提交文件和发布时的精确 SHA 校验保持有效。
|
||||
|
||||
@ -37,24 +37,32 @@ export function purgeReferenceLookupKeys(files: readonly PurgeFile[]): string[]
|
||||
|
||||
export async function findIndexedPurgeReferenceSources(db: Prisma.TransactionClient, files: readonly PurgeFile[]): Promise<ReferenceRow[]> {
|
||||
const keys = purgeReferenceLookupKeys(files)
|
||||
// MATERIALIZED 与 OFFSET 0 保留参数化查找边界，避免 custom plan
|
||||
// 把大量检索键误估为命中全库；同一批键仍只访问每个引用索引一次。
|
||||
return db.$queryRaw<ReferenceRow[]>`
|
||||
SELECT ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages" AS sources, false AS derived
|
||||
FROM materials WHERE purge_reference_keys(ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages", false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_url], true FROM material_assets WHERE purge_reference_keys(ARRAY[file_url], true) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_path, preview_url], false FROM material_skus WHERE purge_reference_keys(ARRAY[file_path, preview_url], false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_path, preview_image_path] || preview_image_slices, false FROM brand_knowledge WHERE purge_reference_keys(ARRAY[file_path, preview_image_path] || preview_image_slices, false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT ARRAY["headerLogoUrl", "loginLogoUrl"], false FROM system_settings WHERE purge_reference_keys(ARRAY["headerLogoUrl", "loginLogoUrl"], false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(image_urls), false FROM performance_survey_work_items WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(image_urls), false FROM work_review_versions WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(assets), false FROM product_package_upload_jobs WHERE status IN ('queued', 'processing') AND purge_reference_keys(purge_json_reference_strings(assets), false) && ${keys}::text[]
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(operations), false FROM task_detail_change_requests WHERE status NOT IN ('applied', 'rejected', 'expired', 'cancelled') AND purge_reference_keys(purge_json_reference_strings(operations), false) && ${keys}::text[]
|
||||
WITH lookup AS MATERIALIZED (SELECT ${keys}::text[] AS keys)
|
||||
SELECT hits.sources, hits.derived
|
||||
FROM lookup
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages" AS sources, false AS derived
|
||||
FROM materials WHERE purge_reference_keys(ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages", false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_url], true FROM material_assets WHERE purge_reference_keys(ARRAY[file_url], true) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_path, preview_url], false FROM material_skus WHERE purge_reference_keys(ARRAY[file_path, preview_url], false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT ARRAY[file_path, preview_image_path] || preview_image_slices, false FROM brand_knowledge WHERE purge_reference_keys(ARRAY[file_path, preview_image_path] || preview_image_slices, false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT ARRAY["headerLogoUrl", "loginLogoUrl"], false FROM system_settings WHERE purge_reference_keys(ARRAY["headerLogoUrl", "loginLogoUrl"], false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(image_urls), false FROM performance_survey_work_items WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(image_urls), false FROM work_review_versions WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(assets), false FROM product_package_upload_jobs WHERE status IN ('queued', 'processing') AND purge_reference_keys(purge_json_reference_strings(assets), false) && lookup.keys
|
||||
UNION ALL
|
||||
SELECT purge_json_reference_strings(operations), false FROM task_detail_change_requests WHERE status NOT IN ('applied', 'rejected', 'expired', 'cancelled') AND purge_reference_keys(purge_json_reference_strings(operations), false) && lookup.keys
|
||||
OFFSET 0
|
||||
) AS hits
|
||||
`
|
||||
}
|
||||
|
||||
@ -4,6 +4,8 @@ import { randomUUID } from 'node:crypto'
|
||||
import { promises as fs } from 'node:fs'
|
||||
import os from 'node:os'
|
||||
import path from 'node:path'
|
||||
import type { Prisma } from '@prisma/client'
|
||||
import type { PurgeFile } from '../lib/material-purge-files'
|
||||
|
||||
async function main() {
|
||||
const url = new URL(process.env.DATABASE_URL || '')
|
||||
@ -42,7 +44,18 @@ async function main() {
|
||||
const { prisma } = await import('../lib/prisma')
|
||||
const { enqueueMaterialPurge, processNextMaterialPurge, retryMaterialPurge, ensureMaterialPurgeWorkerStarted } = await import('../lib/material-purge')
|
||||
const { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity, identifyPurgeFile } = await import('../lib/material-purge-files')
|
||||
const { purgeReferenceLookupKeys } = await import('../lib/material-purge-reference-query')
|
||||
const { findIndexedPurgeReferenceSources } = await import('../lib/material-purge-reference-query')
|
||||
async function explainReferences(db: Prisma.TransactionClient, files: PurgeFile[]) {
|
||||
let query: { strings: TemplateStringsArray; values: unknown[] } | undefined
|
||||
const wrapped = { $queryRaw: (strings: TemplateStringsArray, ...values: unknown[]) => {
|
||||
query = { strings, values }
|
||||
return db.$queryRaw(strings, ...values)
|
||||
} } as Prisma.TransactionClient
|
||||
await findIndexedPurgeReferenceSources(wrapped, files)
|
||||
assert(query)
|
||||
const sql = query.strings.reduce((value, part, index) => value + (index ? '$' + index : '') + part, '')
|
||||
return db.$queryRawUnsafe<any[]>('EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ' + sql, ...query.values)
|
||||
}
|
||||
const run = randomUUID()
|
||||
const passed: string[] = []
|
||||
passed.push('真实升级 SQL 创建清理表、重试索引与审计表，并解除历史日志的素材级联删除')
|
||||
@ -279,9 +292,7 @@ async function main() {
|
||||
const scaleSources = await Promise.all(Array.from({ length: 4 }, (_, i) => file(`scale-delete-${i}.bin`)))
|
||||
const scaleMaterial = await make('scale-delete', { zipFilePath: scaleSources[0], displayImages: scaleSources.slice(1) })
|
||||
const scaleFiles = await collectPurgeFiles({ ...scaleMaterial, assets: [], skus: [] })
|
||||
const keys = purgeReferenceLookupKeys(scaleFiles)
|
||||
const plan = await prisma.$queryRaw<any[]>`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
|
||||
SELECT file_url FROM material_assets WHERE purge_reference_keys(ARRAY[file_url], true) && ${keys}::text[]`
|
||||
const plan = await explainReferences(prisma, scaleFiles)
|
||||
assert(JSON.stringify(plan).includes('material_assets_purge_references_idx'), '40万行必须实际使用引用索引')
|
||||
const referenceStarted = Date.now()
|
||||
await findPurgeFileReferences(prisma, scaleFiles)
|
||||
@ -295,9 +306,30 @@ async function main() {
|
||||
assert.equal(result.status, 'completed')
|
||||
assert((await Promise.all(scaleSources.map((_, i) => exists(`scale-delete-${i}.bin`)))).every(value => !value))
|
||||
assert(referenceMs < 500 && completedMs < 1500, `索引检查 ${referenceMs}ms / 实际删除 ${completedMs}ms 超出验收线`)
|
||||
const bulkFiles: PurgeFile[] = Array.from({ length: 106 }, (_, index) => ({
|
||||
source: `https://isolated.test/project/bulk/${index}.bin`, storage: 's3',
|
||||
bucket: process.env.S3_BUCKET || 'qa-bucket', key: `project/bulk/${index}.bin`, outcome: 'pending',
|
||||
}))
|
||||
bulkFiles.push(...specialFiles)
|
||||
const bulkPlans = []
|
||||
for (const mode of ['force_custom_plan', 'force_generic_plan']) {
|
||||
const measurement = await prisma.$transaction(async tx => {
|
||||
await tx.$executeRawUnsafe('SET LOCAL plan_cache_mode = ' + mode)
|
||||
const started = Date.now()
|
||||
const refs = await findPurgeFileReferences(tx, bulkFiles)
|
||||
const ms = Date.now() - started
|
||||
assert(refs.has(purgeFileIdentity(specialFiles[0])), '大批量查询仍须保留共享文件')
|
||||
const executionPlan = await explainReferences(tx, bulkFiles)
|
||||
assert(JSON.stringify(executionPlan).includes('material_assets_purge_references_idx'), `${mode} 大批量必须使用索引`)
|
||||
assert(ms < 1500, `${mode} 大批量引用查询 ${ms}ms 超出验收线`)
|
||||
return { mode, files: bulkFiles.length, ms, plan: executionPlan }
|
||||
}, { timeout: 30_000 })
|
||||
bulkPlans.push(measurement)
|
||||
}
|
||||
await fs.mkdir('tmp/material-purge-index-qa', { recursive: true })
|
||||
await fs.writeFile('tmp/material-purge-index-qa/scale.json', JSON.stringify({ assets: await prisma.materialAsset.count(), referenceMs, acceptedMs, completedMs, plan }, null, 2))
|
||||
await fs.writeFile('tmp/material-purge-index-qa/scale.json', JSON.stringify({ assets: await prisma.materialAsset.count(), referenceMs, acceptedMs, completedMs, plan, bulkPlans }, null, 2))
|
||||
passed.push(`40万资产下只查目标引用 ${referenceMs}ms；提交 ${acceptedMs}ms、4个文件真实删除及审计完成 ${completedMs}ms，执行计划使用 GIN 索引`)
|
||||
passed.push(`107个文件在 custom/generic 两种执行计划下均使用索引并保护共享引用：${bulkPlans.map(p => p.ms + 'ms').join(' / ')}`)
|
||||
}
|
||||
|
||||
delete process.env.NEXT_BUILD
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user