root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(monitoring/citations): consolidate domain mapping via CTE

Browse Source 
Replace the per-row tenant+global LATERAL joins with a single CTE that
materializes filtered citations and resolves domain mappings once per
distinct host/domain/site_key triple. Reduces redundant scans of
site_domain_mappings on large run batches.
This commit is contained in:
liangxu
2026-04-29 15:59:12 +08:00
parent 0ce20ac2fa
commit 2d66e32938
1 changed files with 125 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/monitoring_service.go
+125 -79
View File
@@ -2791,50 +2791,75 @@ func (s *MonitoringService) loadCitationsForRuns(ctx context.Context, tenantID i
}
rows, err := s.monitoringPool.Query(ctx, `
WITH filtered_citations AS (
SELECT
cf.id,
cf.run_id,
cf.cited_url,
cf.cited_title,
cf.host,
cf.registrable_domain,
cf.site_key,
cf.article_id,
cf.resolution_status,
cf.resolution_confidence
FROM monitoring_citation_facts cf
WHERE cf.tenant_id = $1
AND cf.run_id = ANY($2)
),
domain_keys AS (
SELECT DISTINCT host, registrable_domain, site_key
FROM filtered_citations
),
domain_mappings AS (
SELECT
dk.host,
dk.registrable_domain,
dk.site_key,
dm.registrable_domain AS mapped_domain,
dm.site_key AS mapped_site_key,
dm.site_name AS mapped_site_name
FROM domain_keys dk
LEFT JOIN LATERAL (
SELECT mapping.registrable_domain, mapping.site_key, mapping.site_name
FROM site_domain_mappings mapping
WHERE mapping.is_active = TRUE
AND mapping.registrable_domain = ANY(
ARRAY(
SELECT DISTINCT candidate
FROM (
VALUES (dk.registrable_domain), (dk.site_key), (dk.host)
UNION ALL
SELECT array_to_string(host_parts.value[idx.value:array_length(host_parts.value, 1)], '.')
FROM (SELECT string_to_array(dk.host, '.') AS value) host_parts
CROSS JOIN generate_subscripts(host_parts.value, 1) AS idx(value)
WHERE idx.value > 1
) candidates(candidate)
WHERE candidate IS NOT NULL AND candidate <> ''
)
)
ORDER BY LENGTH(mapping.registrable_domain) DESC, mapping.id DESC
LIMIT 1
) dm ON TRUE
)
SELECT
cf.run_id,
cf.cited_url,
cf.cited_title,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain) AS site_name,
COALESCE(tm.site_key, gm.site_key, cf.site_key) AS site_key,
cf.article_id,
fc.run_id,
fc.cited_url,
fc.cited_title,
COALESCE(dm.mapped_site_name, fc.site_key, fc.registrable_domain) AS site_name,
COALESCE(dm.mapped_site_key, fc.site_key) AS site_key,
fc.article_id,
alias.article_title_snapshot AS article_title,
cf.resolution_status,
cf.resolution_confidence
FROM monitoring_citation_facts cf
LEFT JOIN LATERAL (
SELECT site_key, site_name
FROM site_domain_mappings
WHERE tenant_id = cf.tenant_id
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) tm ON TRUE
LEFT JOIN LATERAL (
SELECT site_key, site_name
FROM site_domain_mappings
WHERE tenant_id IS NULL
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) gm ON tm.site_key IS NULL
fc.resolution_status,
fc.resolution_confidence
FROM filtered_citations fc
LEFT JOIN domain_mappings dm
ON dm.host = fc.host
AND dm.registrable_domain = fc.registrable_domain
AND dm.site_key = fc.site_key
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_id
WHERE cf.tenant_id = $1
AND cf.run_id = ANY($2)
ORDER BY cf.run_id ASC, cf.id ASC
ON alias.tenant_id = $1 AND alias.article_id = fc.article_id
ORDER BY fc.run_id ASC, fc.id ASC
`, tenantID, runIDs)
if err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to load citations")
@@ -2881,45 +2906,17 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
rows, err := s.monitoringPool.Query(ctx, `
WITH grouped AS (
WITH filtered_facts AS (
SELECT
r.ai_platform_id,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain) AS site_name,
COALESCE(tm.site_key, gm.site_key, cf.site_key) AS site_key,
COALESCE(tm.registrable_domain, gm.registrable_domain, cf.site_key, cf.registrable_domain) AS site_domain,
COUNT(cf.id) AS citation_count,
COUNT(*) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS content_citation_count
cf.id,
cf.host,
cf.registrable_domain,
cf.site_key,
cf.article_id
FROM question_monitor_runs r
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
LEFT JOIN LATERAL (
SELECT registrable_domain, site_key, site_name
FROM site_domain_mappings
WHERE tenant_id = cf.tenant_id
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) tm ON TRUE
LEFT JOIN LATERAL (
SELECT registrable_domain, site_key, site_name
FROM site_domain_mappings
WHERE tenant_id IS NULL
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) gm ON tm.site_key IS NULL
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.question_id = $3
@@ -2928,11 +2925,60 @@ func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, te
AND r.status = 'succeeded'
AND r.business_date BETWEEN $6::date AND $7::date
AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8))
),
domain_keys AS (
SELECT DISTINCT host, registrable_domain, site_key
FROM filtered_facts
),
domain_mappings AS (
SELECT
dk.host,
dk.registrable_domain,
dk.site_key,
dm.registrable_domain AS mapped_domain,
dm.site_key AS mapped_site_key,
dm.site_name AS mapped_site_name
FROM domain_keys dk
LEFT JOIN LATERAL (
SELECT mapping.registrable_domain, mapping.site_key, mapping.site_name
FROM site_domain_mappings mapping
WHERE mapping.is_active = TRUE
AND mapping.registrable_domain = ANY(
ARRAY(
SELECT DISTINCT candidate
FROM (
VALUES (dk.registrable_domain), (dk.site_key), (dk.host)
UNION ALL
SELECT array_to_string(host_parts.value[idx.value:array_length(host_parts.value, 1)], '.')
FROM (SELECT string_to_array(dk.host, '.') AS value) host_parts
CROSS JOIN generate_subscripts(host_parts.value, 1) AS idx(value)
WHERE idx.value > 1
) candidates(candidate)
WHERE candidate IS NOT NULL AND candidate <> ''
)
)
ORDER BY LENGTH(mapping.registrable_domain) DESC, mapping.id DESC
LIMIT 1
) dm ON TRUE
),
grouped AS (
SELECT
ff.ai_platform_id,
COALESCE(dm.mapped_site_name, ff.site_key, ff.registrable_domain) AS site_name,
COALESCE(dm.mapped_site_key, ff.site_key) AS site_key,
COALESCE(dm.mapped_domain, ff.site_key, ff.registrable_domain) AS site_domain,
COUNT(ff.id) AS citation_count,
COUNT(*) FILTER (WHERE ff.article_id IS NOT NULL AND ff.ai_platform_id <> 'qwen') AS content_citation_count
FROM filtered_facts ff
LEFT JOIN domain_mappings dm
ON dm.host = ff.host
AND dm.registrable_domain = ff.registrable_domain
AND dm.site_key = ff.site_key
GROUP BY
r.ai_platform_id,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain),
COALESCE(tm.site_key, gm.site_key, cf.site_key),
COALESCE(tm.registrable_domain, gm.registrable_domain, cf.site_key, cf.registrable_domain)
ff.ai_platform_id,
COALESCE(dm.mapped_site_name, ff.site_key, ff.registrable_domain),
COALESCE(dm.mapped_site_key, ff.site_key),
COALESCE(dm.mapped_domain, ff.site_key, ff.registrable_domain)
),
totals AS (
SELECT