feat(external): 宽表强制一条码一行保护(DISTINCT ON barcode_business_key)

根因:external.product_catalog_wide_v1 的 FROM sku_source 行粒度是 SKU,
barcode 只是字段。当前 43261 行 barcode 全唯一是数据巧合(formal 0 行 +
intake ready 每条码唯一),视图定义没强制一条码一行。一旦出现一条码多 SKU
(正式 ProductSKU 落库 + intake ready 同条码,或 intake 内部同条码多行),
宽表就会重复,污染外部系统 barcode 主同步键。

修复:复用项目已有 DISTINCT ON(barcode_business_key) 模式(latest_quality /
limit_by_barcode CTE),把原主查询整体包成 catalog_by_sku CTE,外层加
DISTINCT ON (barcode_business_key)。优先级:FORMAL > INTAKE_READY,
updated_at 最新 > created_at 最新。

末尾追加 catalog_source 字段输出(数据来源标记,对外部系统有用)。
用 CREATE OR REPLACE VIEW 保留 external_reader 已有 GRANT 权限。

生产验证:行数 43261 不变,0 重复 barcode 组,字段完整。
This commit is contained in:
WuYongJie 2026-06-29 19:40:23 +08:00
parent 9c04e39aaf
commit e0bc1db6bf

View File

@ -0,0 +1,445 @@
-- 把宽表 product_catalog_wide_v1 的口径从"一 SKU 一行"明确保护成"一条码一行"。
-- 原视图 FROM sku_source 的行粒度是 SKU,barcode 只是字段;当前 43261 行 barcode 全唯一是数据巧合,
-- 一旦出现"一条码多 SKU"(正式 ProductSKU 落库 + intake ready 同条码 / intake 内部同条码多行),
-- 宽表就会重复,污染外部系统的 barcode 主同步键。
-- 修复:复用项目已有的 DISTINCT ON(barcode_business_key) 模式(latest_quality / limit_by_barcode CTE),
-- 把原主查询整体包成 catalog_by_sku CTE,外层加 DISTINCT ON (barcode_business_key)。
-- 优先级:FORMAL(正式商品)> INTAKE_READY(开发阶段),同源内 updated_at 最新 > created_at 最新。
-- 用 CREATE OR REPLACE VIEW(不用 DROP),保留 external_reader 已有的 GRANT SELECT 权限。
-- Expand the external product catalog wide view so external systems can
-- consume ready non-integrated SKU rows before formal Product / ProductSKU
-- records are materialized.
CREATE OR REPLACE VIEW external.product_catalog_wide_v1 AS
WITH formal_sku_source AS (
SELECT
sku.product_id,
sku.product_model_number,
sku.product_name,
sku.product_description,
sku.product_category_id,
sku.product_category_name,
sku.product_thumbnail_url,
sku.product_model_url,
sku.sku_id,
sku.sku_code,
sku.sku_name,
sku.price,
sku.cost_price,
sku.stock,
sku.is_active,
sku.record_status,
sku.model_url,
sku.image_url,
sku.image_urls,
sku.source_pre_sku_id,
sku.source_asset_variant_id,
sku.source_asset_variant_ids,
sku.asset_family_code,
sku.asset_family_name,
sku.spec_code,
sku.spec_name,
sku.barcode,
sku.size_json,
sku.color,
sku.material,
sku.source_status,
sku.synced_at,
sku.warehouse_location_id,
sku.warehouse_location_code,
sku.warehouse_location_name,
sku.created_at,
sku.updated_at,
sku.product_business_key,
sku.sku_business_key,
sku.barcode_business_key,
'FORMAL'::text AS catalog_source
FROM external.product_skus_v1 sku
),
formal_identity_map AS (
SELECT
barcode_business_key,
CASE
WHEN NULLIF(btrim(COALESCE(sku_code, '')), '') IS NOT NULL
THEN upper(btrim(sku_code))
ELSE NULL
END AS normalized_sku_code
FROM formal_sku_source
),
intake_ready_sku_source AS (
SELECT
product.intake_product_id AS product_id,
NULLIF(btrim(COALESCE(product.spu, '')), '') AS product_model_number,
NULLIF(btrim(COALESCE(product.product_name, '')), '') AS product_name,
NULL::text AS product_description,
NULL::text AS product_category_id,
NULLIF(btrim(COALESCE(product.product_category, '')), '') AS product_category_name,
COALESCE(
NULLIF(btrim(COALESCE(product.product_image_url, '')), ''),
NULLIF(btrim(COALESCE(sku.image_url, '')), '')
) AS product_thumbnail_url,
NULL::text AS product_model_url,
sku.intake_sku_item_id AS sku_id,
NULLIF(btrim(COALESCE(sku.sku_code, '')), '') AS sku_code,
COALESCE(
NULLIF(btrim(COALESCE(sku.spec_name, '')), ''),
NULLIF(btrim(COALESCE(sku.sku_code, '')), '')
) AS sku_name,
NULL::numeric(10,2) AS price,
NULL::numeric(10,2) AS cost_price,
0::integer AS stock,
true AS is_active,
'ACTIVE'::text AS record_status,
NULLIF(btrim(COALESCE(sku.model_3d_url, '')), '') AS model_url,
COALESCE(
NULLIF(btrim(COALESCE(sku.image_url, '')), ''),
NULLIF(btrim(COALESCE(product.product_image_url, '')), '')
) AS image_url,
CASE
WHEN COALESCE(
NULLIF(btrim(COALESCE(sku.image_url, '')), ''),
NULLIF(btrim(COALESCE(product.product_image_url, '')), '')
) IS NOT NULL
THEN ARRAY[
COALESCE(
NULLIF(btrim(COALESCE(sku.image_url, '')), ''),
NULLIF(btrim(COALESCE(product.product_image_url, '')), '')
)
]
ELSE ARRAY[]::text[]
END AS image_urls,
NULL::text AS source_pre_sku_id,
NULL::text AS source_asset_variant_id,
ARRAY[]::text[] AS source_asset_variant_ids,
NULLIF(btrim(COALESCE(sku.model, '')), '') AS asset_family_code,
NULL::text AS asset_family_name,
NULLIF(btrim(COALESCE(sku.attribute_code, '')), '') AS spec_code,
NULLIF(btrim(COALESCE(sku.spec_name, '')), '') AS spec_name,
NULLIF(btrim(COALESCE(sku.barcode, '')), '') AS barcode,
CASE
WHEN NULLIF(btrim(COALESCE(sku.size, '')), '') IS NOT NULL
THEN jsonb_build_object('label', btrim(sku.size))
ELSE NULL::jsonb
END AS size_json,
NULLIF(btrim(COALESCE(sku.color, '')), '') AS color,
NULLIF(btrim(COALESCE(sku.material, '')), '') AS material,
NULL::text AS source_status,
NULL::text AS synced_at,
NULL::text AS warehouse_location_id,
NULL::text AS warehouse_location_code,
NULL::text AS warehouse_location_name,
product.created_at AS created_at,
product.updated_at AS updated_at,
CASE
WHEN NULLIF(btrim(COALESCE(product.spu, '')), '') IS NOT NULL
THEN upper(btrim(product.spu))
ELSE NULL
END AS product_business_key,
CASE
WHEN NULLIF(btrim(COALESCE(sku.sku_code, '')), '') IS NOT NULL
THEN upper(btrim(sku.sku_code))
ELSE NULL
END AS sku_business_key,
CASE
WHEN NULLIF(btrim(COALESCE(sku.barcode, '')), '') IS NOT NULL
THEN upper(btrim(sku.barcode))
ELSE NULL
END AS barcode_business_key,
'INTAKE_READY'::text AS catalog_source
FROM external.intake_sku_items_ready_v1 sku
JOIN external.intake_products_ready_v1 product
ON product.intake_product_id = sku.intake_product_id
WHERE NOT EXISTS (
SELECT 1
FROM formal_identity_map formal
WHERE (
NULLIF(btrim(COALESCE(sku.barcode, '')), '') IS NOT NULL
AND formal.barcode_business_key = upper(btrim(sku.barcode))
)
OR (
NULLIF(btrim(COALESCE(sku.barcode, '')), '') IS NULL
AND NULLIF(btrim(COALESCE(sku.sku_code, '')), '') IS NOT NULL
AND formal.normalized_sku_code = upper(btrim(sku.sku_code))
)
)
),
sku_source AS (
SELECT * FROM formal_sku_source
UNION ALL
SELECT * FROM intake_ready_sku_source
),
asset_summary AS (
SELECT
product_business_key,
sku_business_key,
COUNT(*) FILTER (WHERE asset_type = 'SKU_IMAGE')::integer AS image_asset_count,
COUNT(*) FILTER (WHERE asset_type = 'SKU_MODEL')::integer AS model_asset_count,
COUNT(*) FILTER (WHERE asset_type = 'SKU_THUMBNAIL')::integer AS thumbnail_asset_count,
MIN(asset_url) FILTER (WHERE asset_type = 'SKU_MODEL') AS first_model_asset_url,
MIN(asset_url) FILTER (WHERE asset_type = 'SKU_THUMBNAIL') AS first_thumbnail_asset_url
FROM external.product_assets_v1
GROUP BY product_business_key, sku_business_key
),
packaged_summary AS (
SELECT
barcode_business_key,
COUNT(*)::integer AS packaged_item_count,
COALESCE(SUM(stock_quantity), 0)::integer AS packaged_stock_quantity
FROM external.packaged_items_v1
WHERE NULLIF(barcode_business_key, '') IS NOT NULL
GROUP BY barcode_business_key
),
quality_summary AS (
SELECT
barcode_business_key,
COUNT(*)::integer AS quality_record_count,
COUNT(*) FILTER (WHERE source_type = 'WAREHOUSE')::integer AS warehouse_quality_record_count,
COUNT(*) FILTER (WHERE source_type = 'FIRST_BATCH')::integer AS first_batch_quality_record_count
FROM external.sku_quality_inspections_v1
WHERE NULLIF(barcode_business_key, '') IS NOT NULL
GROUP BY barcode_business_key
),
latest_quality AS (
SELECT DISTINCT ON (barcode_business_key)
barcode_business_key,
record_key AS latest_quality_record_key,
source_type AS latest_quality_source_type,
batch_no AS latest_quality_batch_no,
inspected_at AS latest_quality_inspected_at,
conclusion AS latest_quality_conclusion,
status AS latest_quality_status
FROM external.sku_quality_inspections_v1
WHERE NULLIF(barcode_business_key, '') IS NOT NULL
ORDER BY barcode_business_key, COALESCE(inspected_at, updated_at, created_at) DESC, created_at DESC
),
limit_by_barcode AS (
SELECT DISTINCT ON (identity_value)
identity_value,
product_business_key AS limit_source_product_business_key,
sku_business_key AS limit_source_sku_business_key,
limit_price_applicable,
limit_price_confirmed,
limit_price_status,
expected_daily_limit_price,
expected_limit_gross_margin_rate,
non_spu_owner_limit_price,
limit_confirmed_at
FROM (
SELECT
upper(btrim(barcode)) AS identity_value,
product_business_key,
sku_business_key,
limit_price_applicable,
limit_price_confirmed,
limit_price_status,
expected_daily_limit_price,
expected_limit_gross_margin_rate,
non_spu_owner_limit_price,
limit_confirmed_at,
updated_at
FROM external.intake_product_limits_v1
WHERE NULLIF(btrim(COALESCE(barcode, '')), '') IS NOT NULL
) limit_rows
WHERE NULLIF(identity_value, '') IS NOT NULL
ORDER BY
identity_value,
limit_price_applicable DESC,
limit_price_confirmed DESC,
limit_confirmed_at DESC NULLS LAST,
updated_at DESC
),
limit_by_sku AS (
SELECT DISTINCT ON (identity_value)
identity_value,
product_business_key AS limit_source_product_business_key,
sku_business_key AS limit_source_sku_business_key,
limit_price_applicable,
limit_price_confirmed,
limit_price_status,
expected_daily_limit_price,
expected_limit_gross_margin_rate,
non_spu_owner_limit_price,
limit_confirmed_at
FROM (
SELECT
upper(btrim(sku_code)) AS identity_value,
product_business_key,
sku_business_key,
limit_price_applicable,
limit_price_confirmed,
limit_price_status,
expected_daily_limit_price,
expected_limit_gross_margin_rate,
non_spu_owner_limit_price,
limit_confirmed_at,
updated_at
FROM external.intake_product_limits_v1
WHERE NULLIF(btrim(COALESCE(sku_code, '')), '') IS NOT NULL
) limit_rows
WHERE NULLIF(identity_value, '') IS NOT NULL
ORDER BY
identity_value,
limit_price_applicable DESC,
limit_price_confirmed DESC,
limit_confirmed_at DESC NULLS LAST,
updated_at DESC
),
catalog_by_sku AS (
SELECT
sku.product_id,
sku.product_model_number,
sku.product_business_key,
sku.product_name,
sku.product_description,
sku.product_category_id,
sku.product_category_name,
sku.product_thumbnail_url,
sku.product_model_url,
sku.sku_id,
sku.sku_code,
sku.sku_business_key,
sku.sku_name,
sku.price,
sku.cost_price,
sku.stock,
sku.is_active,
sku.record_status,
sku.model_url,
sku.image_url,
sku.image_urls,
sku.source_pre_sku_id,
sku.source_asset_variant_id,
sku.source_asset_variant_ids,
sku.asset_family_code,
sku.asset_family_name,
sku.spec_code,
sku.spec_name,
sku.barcode,
sku.barcode_business_key,
sku.size_json,
sku.color,
sku.material,
sku.source_status,
sku.synced_at,
sku.warehouse_location_id,
sku.warehouse_location_code,
sku.warehouse_location_name,
stock.warehouse_zone,
stock.warehouse_aisle,
stock.warehouse_shelf,
stock.warehouse_bin,
stock.warehouse_location_is_active,
stock.warehouse_location_status,
COALESCE(limit_by_barcode.limit_price_applicable, limit_by_sku.limit_price_applicable, false) AS limit_price_applicable,
COALESCE(limit_by_barcode.limit_price_confirmed, limit_by_sku.limit_price_confirmed, false) AS limit_price_confirmed,
COALESCE(limit_by_barcode.limit_price_status, limit_by_sku.limit_price_status, 'NOT_APPLICABLE') AS limit_price_status,
COALESCE(limit_by_barcode.expected_daily_limit_price, limit_by_sku.expected_daily_limit_price) AS expected_daily_limit_price,
COALESCE(limit_by_barcode.expected_limit_gross_margin_rate, limit_by_sku.expected_limit_gross_margin_rate) AS expected_limit_gross_margin_rate,
COALESCE(limit_by_barcode.non_spu_owner_limit_price, limit_by_sku.non_spu_owner_limit_price) AS non_spu_owner_limit_price,
COALESCE(limit_by_barcode.limit_confirmed_at, limit_by_sku.limit_confirmed_at) AS limit_confirmed_at,
COALESCE(limit_by_barcode.limit_source_product_business_key, limit_by_sku.limit_source_product_business_key) AS limit_source_product_business_key,
COALESCE(limit_by_barcode.limit_source_sku_business_key, limit_by_sku.limit_source_sku_business_key) AS limit_source_sku_business_key,
CASE
WHEN sku.catalog_source = 'INTAKE_READY'
THEN CASE WHEN NULLIF(btrim(COALESCE(sku.image_url, '')), '') IS NOT NULL THEN 1 ELSE 0 END
ELSE COALESCE(asset_summary.image_asset_count, 0)
END AS image_asset_count,
CASE
WHEN sku.catalog_source = 'INTAKE_READY'
THEN CASE WHEN NULLIF(btrim(COALESCE(sku.model_url, '')), '') IS NOT NULL THEN 1 ELSE 0 END
ELSE COALESCE(asset_summary.model_asset_count, 0)
END AS model_asset_count,
CASE
WHEN sku.catalog_source = 'INTAKE_READY'
THEN CASE
WHEN COALESCE(
NULLIF(btrim(COALESCE(sku.product_thumbnail_url, '')), ''),
NULLIF(btrim(COALESCE(sku.image_url, '')), '')
) IS NOT NULL
THEN 1
ELSE 0
END
ELSE COALESCE(asset_summary.thumbnail_asset_count, 0)
END AS thumbnail_asset_count,
COALESCE(
asset_summary.first_model_asset_url,
CASE
WHEN sku.catalog_source = 'INTAKE_READY'
THEN NULLIF(btrim(COALESCE(sku.model_url, '')), '')
ELSE NULL
END
) AS first_model_asset_url,
COALESCE(
asset_summary.first_thumbnail_asset_url,
CASE
WHEN sku.catalog_source = 'INTAKE_READY'
THEN COALESCE(
NULLIF(btrim(COALESCE(sku.product_thumbnail_url, '')), ''),
NULLIF(btrim(COALESCE(sku.image_url, '')), '')
)
ELSE NULL
END
) AS first_thumbnail_asset_url,
COALESCE(bom_summary.bom_item_count, 0) AS bom_item_count,
COALESCE(bom_summary.panel_item_count, 0) AS panel_item_count,
COALESCE(bom_summary.hardware_item_count, 0) AS hardware_item_count,
COALESCE(bom_summary.material_codes, ARRAY[]::text[]) AS material_codes,
COALESCE(bom_summary.material_names, ARRAY[]::text[]) AS material_names,
COALESCE(packaged_summary.packaged_item_count, 0) AS packaged_item_count,
COALESCE(packaged_summary.packaged_stock_quantity, 0) AS packaged_stock_quantity,
COALESCE(quality_summary.quality_record_count, 0) AS quality_record_count,
COALESCE(quality_summary.warehouse_quality_record_count, 0) AS warehouse_quality_record_count,
COALESCE(quality_summary.first_batch_quality_record_count, 0) AS first_batch_quality_record_count,
latest_quality.latest_quality_record_key,
latest_quality.latest_quality_source_type,
latest_quality.latest_quality_batch_no,
latest_quality.latest_quality_inspected_at,
latest_quality.latest_quality_conclusion,
latest_quality.latest_quality_status,
sku.created_at,
sku.updated_at,
sku.catalog_source
FROM sku_source sku
LEFT JOIN external.sku_warehouse_stock_v1 stock
ON stock.barcode_business_key = sku.barcode_business_key
LEFT JOIN limit_by_sku
ON limit_by_sku.identity_value = upper(btrim(COALESCE(sku.sku_code, '')))
LEFT JOIN limit_by_barcode
ON limit_by_barcode.identity_value = sku.barcode_business_key
LEFT JOIN asset_summary
ON asset_summary.product_business_key = sku.product_business_key
AND asset_summary.sku_business_key = sku.sku_business_key
LEFT JOIN LATERAL (
SELECT
COUNT(*)::integer AS bom_item_count,
COUNT(*) FILTER (WHERE item_type = 'PANEL')::integer AS panel_item_count,
COUNT(*) FILTER (WHERE item_type = 'HARDWARE')::integer AS hardware_item_count,
array_remove(array_agg(DISTINCT material_code ORDER BY material_code), NULL) AS material_codes,
array_remove(array_agg(DISTINCT material_name ORDER BY material_name), NULL) AS material_names
FROM external.product_bom_v1 bom
WHERE bom.product_business_key = sku.product_business_key
AND (
bom.sku_business_key = sku.sku_business_key
OR bom.sku_business_key IS NULL
)
) bom_summary ON true
LEFT JOIN packaged_summary
ON packaged_summary.barcode_business_key = sku.barcode_business_key
LEFT JOIN quality_summary
ON quality_summary.barcode_business_key = sku.barcode_business_key
LEFT JOIN latest_quality
ON latest_quality.barcode_business_key = sku.barcode_business_key
)
SELECT DISTINCT ON (catalog_by_sku.barcode_business_key)
catalog_by_sku.*
FROM catalog_by_sku
WHERE catalog_by_sku.barcode_business_key IS NOT NULL
AND btrim(catalog_by_sku.barcode_business_key) <> ''
ORDER BY
catalog_by_sku.barcode_business_key,
CASE WHEN catalog_by_sku.catalog_source IS NULL THEN 1
WHEN catalog_by_sku.catalog_source = 'FORMAL' THEN 0
ELSE 1 END,
catalog_by_sku.updated_at DESC NULLS LAST,
catalog_by_sku.created_at DESC;