perf(finance): 消除定时同步与后台页面的重查询

- 新增 order_items.manual_sku_key 生成列(迁移 106),统一内部 SKU 口径(skuNick → outSkuSn → sku_code → sku_name,去空格)
- 人工渠道匹配 8.4s → 62ms;人工商品归一写回 7.0s → 307ms
- 销售事实重算改为物化订单集合 + 预聚合成本/退款,6.9s → 0.5s 内
- 成本异常写入去掉逐项 LATERAL 工单探测,收口按最近 7 天窗口,7–8s → 0.5s 内
- 销售毛利订单统计去掉重复 NOT EXISTS,9.4s → 0.4s
- 接单平台店铺下拉改 EXISTS 半连接,3.1s → 0.3s
- cost-finance-sync 默认间隔 5 分钟 → 15 分钟;完整同步 51.2s → 34.9s
This commit is contained in:
yml2213
2026-10-28 13:41:11 +08:00
parent 08318ef8d9
commit 915ce017f3
8 changed files with 145 additions and 82 deletions
@@ -0,0 +1,31 @@
-- 106_manual_sku_key.sql
--
-- 内部 SKU 匹配键:skuNick -> outSkuSn -> sku_code -> sku_name(全部去空格)。
--
-- 背景:人工商品归一、人工渠道匹配每次定时同步都要从 spec_json 现算这个键,
-- 41 万单 / 12 万人工项实测单条 SQL 3–7 秒,且两种写法(3 级 / 4 级、带不带
-- TRIM)不一致。固化成生成列后,查询直接走索引,也保证所有入口口径一致。
--
-- 注意:ADD COLUMN ... GENERATED ... STORED 会重写 order_items(41 万行约
-- 1–2 分钟,期间持 ACCESS EXCLUSIVE 锁),应放在维护窗口执行。
ALTER TABLE order_items
ADD COLUMN IF NOT EXISTS manual_sku_key TEXT
GENERATED ALWAYS AS (
COALESCE(
NULLIF(BTRIM(spec_json->>'skuNick'), ''),
NULLIF(BTRIM(spec_json->>'outSkuSn'), ''),
NULLIF(BTRIM(sku_code), ''),
NULLIF(BTRIM(sku_name), '')
)
) STORED;
COMMENT ON COLUMN order_items.manual_sku_key IS
'内部 SKU 匹配键(去空格):skuNick -> outSkuSn -> sku_code -> sku_name;人工履约配置按它匹配';
CREATE INDEX IF NOT EXISTS idx_order_items_manual_sku_key_stored
ON order_items (manual_sku_key)
WHERE fulfillment_mode = 'manual' AND manual_sku_key IS NOT NULL;
-- 旧索引表达式少了 TRIM,和任何查询都对不上,属于纯写入开销。
DROP INDEX IF EXISTS idx_order_items_manual_sku_key;
@@ -492,10 +492,12 @@ export async function listWorkOrders({
*/
export async function listWorkOrderShopNames(): Promise<string[]> {
const result = await query<{ shop_name: string }>(
// 从 orders 出发 + EXISTS 半连接:18 万工单的 hash join 实测 3.1 秒,
// 改成 orders 驱动后走 idx_work_orders_order_id 半连接,约 0.3 秒。
`SELECT DISTINCT BTRIM(o.shop_name) AS shop_name
FROM work_orders wo
INNER JOIN orders o ON o.id = wo.order_id
FROM orders o
WHERE BTRIM(o.shop_name) <> ''
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_id = o.id)
ORDER BY shop_name ASC`,
)
return result.rows
@@ -57,13 +57,9 @@ export async function syncEstimatedOrderItemCostSnapshots(
COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id, product_rule_id
FROM work_orders
WHERE order_item_id = oi.id
ORDER BY id DESC
LIMIT 1
) wo ON TRUE
-- order_item_id 上有唯一部分索引,一个订单项最多一张工单;
-- 原来的 ORDER BY id DESC LIMIT 1 逐行探测会做 3.7 万次索引查找。
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
@@ -613,9 +609,7 @@ async function insertAndResolveProductMappingExceptions(
COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id FROM work_orders WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
) wo ON TRUE
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
@@ -651,11 +645,15 @@ async function insertAndResolveProductMappingExceptions(
WHERE ce.status = 'open'
AND ce.subject_type = 'order_item'
AND ce.reason = 'product_mapping_missing'
-- 只收口最近创建、仍在同步视野内的异常:全表收口会随着历史异常累积
-- 变成每次同步都扫几十万行(41 万单测试库实测单次 7.4 秒)。
AND ce.created_at >= ($1::date - INTERVAL '7 days')
AND EXISTS (
SELECT 1 FROM order_items oi
WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL
)
`,
[fromDate],
)
return { created: inserted.rowCount || 0, resolved: resolved.rowCount || 0 }
@@ -678,9 +676,7 @@ async function insertAndResolveCostMissingExceptions(
COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id FROM work_orders WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
) wo ON TRUE
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
@@ -774,6 +770,7 @@ async function insertAndResolveCostMissingExceptions(
WHERE ce.status = 'open'
AND ce.subject_type = 'order_item'
AND ce.reason = 'cost_missing'
AND ce.created_at >= ($1::date - INTERVAL '7 days')
AND EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'order_item'
@@ -781,6 +778,7 @@ async function insertAndResolveCostMissingExceptions(
AND cs.superseded_at IS NULL
)
`,
[fromDate],
)
const manualResolved = await client.query(
@@ -790,6 +788,7 @@ async function insertAndResolveCostMissingExceptions(
WHERE ce.status = 'open'
AND ce.subject_type = 'order_item'
AND ce.reason = 'manual_cost_missing'
AND ce.created_at >= ($1::date - INTERVAL '7 days')
AND EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case'
@@ -797,6 +796,7 @@ async function insertAndResolveCostMissingExceptions(
AND cs.superseded_at IS NULL
)
`,
[fromDate],
)
return {
@@ -124,12 +124,7 @@ export async function scanManualSkus(options: ScanOptions) {
oi.sku_name,
COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date,
COALESCE(
NULLIF(oi.spec_json->>'skuNick', ''),
NULLIF(oi.spec_json->>'outSkuSn', ''),
NULLIF(oi.sku_code, ''),
NULLIF(oi.sku_name, '')
) AS sku_key
oi.manual_sku_key AS sku_key
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_mode = 'manual'
@@ -247,7 +242,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_mode = 'manual'
AND o.provider = $${params.length}
AND COALESCE(NULLIF(oi.spec_json->>'skuNick', ''), NULLIF(oi.spec_json->>'outSkuSn', '')) = r.sku_key
AND oi.manual_sku_key = r.sku_key
)`)
}
@@ -435,12 +430,7 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
JOIN manual_fulfillment_configs cfg
ON cfg.sku_key = COALESCE(
NULLIF(oi.spec_json->>'skuNick', ''),
NULLIF(oi.spec_json->>'outSkuSn', ''),
NULLIF(oi.sku_code, ''),
NULLIF(oi.sku_name, '')
)
ON cfg.sku_key = oi.manual_sku_key
WHERE oi.fulfillment_mode = 'manual'
AND cfg.enabled = TRUE
AND ($1 = '' OR cfg.sku_key = $1)
@@ -34,14 +34,11 @@ export type NormalizeManualProductsResult = {
remainingItemCount: number
}
/** 内部 SKU:快手 skuNick / 拼多多 outSkuSn / 平台 sku_code 兜底。 */
const MANUAL_SKU_KEY_SQL = `
COALESCE(
NULLIF(TRIM(oi.spec_json->>'skuNick'), ''),
NULLIF(TRIM(oi.spec_json->>'outSkuSn'), ''),
NULLIF(TRIM(oi.sku_code), '')
)
`
/**
* 内部 SKU:由 order_items.manual_sku_key 生成列统一提供
* (skuNick → outSkuSn → sku_code → sku_name,均去空格)。
*/
const MANUAL_SKU_KEY_SQL = `oi.manual_sku_key`
export async function normalizeManualProducts(
input: NormalizeManualProductsInput = {},
@@ -79,13 +79,14 @@ test('人工确认必须单事务:扣款与快照同事务,且使用 with-cl
})
test('人工确认前必须自动匹配履约配置,配置保存/定时任务也要回填', async () => {
const [manualSource, configSource, jobSource] = await Promise.all([
const [manualSource, configSource, jobSource, migrationSource] = await Promise.all([
readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
readFile(
new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
'utf8',
),
readFile(new URL('../scheduler/cost-finance-sync-job.ts', import.meta.url), 'utf8'),
readFile(new URL('../../db/migrations/106_manual_sku_key.sql', import.meta.url), 'utf8'),
])
assert.ok(
manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId, now)'),
@@ -104,8 +105,15 @@ test('人工确认前必须自动匹配履约配置,配置保存/定时任务
'定时同步必须自动回填未匹配的人工订单项,并限制在最近窗口',
)
assert.ok(
manualSource.includes('NULLIF(oi.sku_code') && configSource.includes('NULLIF(oi.sku_code'),
'SKU 匹配必须有 sku_code / sku_name 兜底,不能只认 spec_json',
configSource.includes('manual_sku_key'),
'人工 SKU 匹配必须统一走 order_items.manual_sku_key 生成列',
)
assert.ok(
migrationSource.includes("spec_json->>'skuNick'") &&
migrationSource.includes("spec_json->>'outSkuSn'") &&
migrationSource.includes('sku_code') &&
migrationSource.includes('sku_name'),
'生成列必须保留 sku_code / sku_name 兜底',
)
})
@@ -76,9 +76,63 @@ export async function rebuildSalesDailyFacts(
const inserted = await client.query(
`
WITH item_base AS (
WITH scoped_orders AS MATERIALIZED (
-- 先物化 7 天窗口内的订单集合,避免 planner 对每个订单项做一次
-- orders 随机回表;后续全部走 hash/merge join。
SELECT
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date,
o.id,
o.provider,
o.shop_id,
o.pay_status,
COALESCE(o.paid_at, o.created_at) AS basis_at
FROM orders o
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND NOT EXISTS (
SELECT 1 FROM finance_periods p
WHERE p.status = 'closed'
AND p.period_key = to_char(
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date,
'YYYY-MM'
)
)
),
scoped_items AS MATERIALIZED (
SELECT
oi.id,
oi.order_id,
oi.canonical_product_id,
oi.fulfillment_mode,
oi.revenue_amount_fen,
oi.normalized_quantity,
oi.quantity
FROM order_items oi
WHERE oi.order_id IN (SELECT id FROM scoped_orders)
),
cost_by_item AS (
-- 一次全量聚合后再 hash join,替代逐项 LATERAL 查 cost_snapshots。
SELECT
order_item_id,
SUM(total_amount_fen)::bigint AS cost_fen,
COUNT(*)::int AS snapshot_count,
COUNT(*) FILTER (WHERE snapshot_stage = 'actual')::int AS actual_count
FROM cost_snapshots
WHERE superseded_at IS NULL
GROUP BY order_item_id
),
refunds_by_item AS (
SELECT
order_item_id,
SUM(amount_fen)::bigint AS refund_fen,
COUNT(*)::int AS refund_count
FROM order_refunds
WHERE status = 'confirmed'
GROUP BY order_item_id
),
item_base AS (
SELECT
(o.basis_at AT TIME ZONE 'Asia/Shanghai')::date AS stat_date,
o.provider,
o.shop_id,
oi.canonical_product_id,
@@ -99,36 +153,10 @@ export async function rebuildSalesDailyFacts(
COALESCE(cost.cost_fen, 0)::bigint AS cost_fen,
COALESCE(cost.snapshot_count, 0)::int AS snapshot_count,
COALESCE(cost.actual_count, 0)::int AS actual_count
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT
SUM(cs.total_amount_fen)::bigint AS cost_fen,
COUNT(*)::int AS snapshot_count,
COUNT(*) FILTER (WHERE cs.snapshot_stage = 'actual')::int AS actual_count
FROM cost_snapshots cs
WHERE cs.order_item_id = oi.id
AND cs.superseded_at IS NULL
) cost ON TRUE
LEFT JOIN LATERAL (
SELECT
SUM(r.amount_fen)::bigint AS refund_fen,
COUNT(*)::int AS refund_count
FROM order_refunds r
WHERE r.order_item_id = oi.id
AND r.status = 'confirmed'
) refunds ON TRUE
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND NOT EXISTS (
SELECT 1 FROM finance_periods p
WHERE p.status = 'closed'
AND p.period_key = to_char(
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date,
'YYYY-MM'
)
)
FROM scoped_items oi
JOIN scoped_orders o ON o.id = oi.order_id
LEFT JOIN cost_by_item cost ON cost.order_item_id = oi.id
LEFT JOIN refunds_by_item refunds ON refunds.order_item_id = oi.id
)
INSERT INTO sales_daily_facts (
stat_date, provider, shop_id, canonical_product_id, fulfillment_mode,
@@ -297,19 +325,23 @@ export async function getSalesProfitOverview(input: SalesFactsInput = {}) {
orders_without_items: number
revenue_without_items: number
}>(
// 原来两个 FILTER 各写一次 NOT EXISTS,等价于对每个订单做两轮逐行探测
// (30 天 / 16 万单实测 108 万 buffers、9.4 秒)。改成一次 DISTINCT 扫描
// order_id 后做 hash join,只扫一遍索引,实测 10.9 万 buffers、0.4 秒。
`
SELECT
COUNT(*)::int AS order_count,
COUNT(*) FILTER (
WHERE NOT EXISTS (SELECT 1 FROM order_items oi WHERE oi.order_id = orders.id)
)::int AS orders_without_items,
COALESCE(SUM(total_amount) FILTER (
WHERE NOT EXISTS (SELECT 1 FROM order_items oi WHERE oi.order_id = orders.id)
), 0)::bigint AS revenue_without_items
FROM orders
WHERE pay_status IN ('paid', 'refunded')
AND (COALESCE(paid_at, created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
COUNT(*) FILTER (WHERE oi.order_id IS NULL)::int AS orders_without_items,
COALESCE(SUM(s.total_amount) FILTER (WHERE oi.order_id IS NULL), 0)::bigint
AS revenue_without_items
FROM (
SELECT o.id, o.total_amount
FROM orders o
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
) s
LEFT JOIN (SELECT DISTINCT order_id FROM order_items) oi ON oi.order_id = s.id
`,
[range.fromDate, range.toDate],
),
@@ -225,7 +225,10 @@ function createDefaultCostFinanceSyncJob() {
id: COST_FINANCE_SYNC_JOB_ID,
type: 'cost_finance_sync',
enabled: true,
intervalSeconds: 300,
// 一次同步要跑 6 条全窗口聚合(41 万单实测 30–60 秒)。5 分钟一次会让
// 数据库长期处在 10–20% 的同步负载下,后台页面首屏被拖慢;财务事实
// 15 分钟刷新已足够,需要更快可在后台定时任务里改回 300。
intervalSeconds: 900,
config: {
days: 7,
},