Skip to content

Commit 6663b90

Browse files
ewlarsonewlarson
authored
Merge pull request #16 from OpenGeoMetadata/feature/dashboard-query-performance
Fix slow repository dashboard with grouped counts
2 parents 3d0f5ff + 7db8c92 commit 6663b90

2 files changed

Lines changed: 137 additions & 44 deletions

File tree

‎backend/app/services/ogm_harvest/repository.py‎

Lines changed: 54 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
from datetime import datetime
44
from typing import Any, Dict, List, Optional
55

6-
from sqlalchemy import String, cast, func, literal, select, update
6+
from sqlalchemy import String, cast, func, literal, select, true, update
77
from sqlalchemy.dialects.postgresql import insert as pg_insert
88

99
from db.database import database
@@ -40,12 +40,14 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]:
4040
.group_by(ogm_harvest_runs.c.ogm_repo_name)
4141
.subquery()
4242
)
43-
harvested_record_count = (
44-
select(func.count())
45-
.select_from(ogm_resource_state)
46-
.where(ogm_resource_state.c.ogm_repo_name == ogm_repos.c.ogm_repo_name)
43+
harvested_counts = (
44+
select(
45+
ogm_resource_state.c.ogm_repo_name,
46+
func.count().label("harvested_record_count"),
47+
)
4748
.where(ogm_resource_state.c.ogm_missing_since.is_(None))
48-
.scalar_subquery()
49+
.group_by(ogm_resource_state.c.ogm_repo_name)
50+
.subquery("harvested_counts")
4951
)
5052
empty_string = cast(literal(""), String())
5153
published_state = cast(literal("published"), String())
@@ -57,41 +59,31 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]:
5759
published_state,
5860
)
5961
)
60-
published_available_record_count = (
61-
select(func.count())
62-
.select_from(resources)
63-
.where(resources.c.b1g_adminTags_sm.is_not(None))
64-
.where(
65-
resources.c.b1g_adminTags_sm.any(
66-
func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name)
67-
)
68-
)
69-
.where(effective_publication_state == published_state)
70-
.scalar_subquery()
62+
# Expand each resource's tags once, then calculate all counts together.
63+
# DISTINCT preserves ANY's one-match-per-resource behavior for duplicate tags.
64+
repo_tags = (
65+
select(func.unnest(resources.c.b1g_adminTags_sm).label("repo_tag"))
66+
.distinct()
67+
.correlate(resources)
68+
.lateral("repo_tags")
7169
)
72-
suppressed_record_count = (
73-
select(func.count())
74-
.select_from(resources)
75-
.where(resources.c.b1g_adminTags_sm.is_not(None))
76-
.where(
77-
resources.c.b1g_adminTags_sm.any(
78-
func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name)
79-
)
80-
)
81-
.where(func.coalesce(resources.c.gbl_suppressed_b, False).is_(True))
82-
.scalar_subquery()
83-
)
84-
unpublished_record_count = (
85-
select(func.count())
86-
.select_from(resources)
87-
.where(resources.c.b1g_adminTags_sm.is_not(None))
88-
.where(
89-
resources.c.b1g_adminTags_sm.any(
90-
func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name)
91-
)
70+
resource_counts = (
71+
select(
72+
repo_tags.c.repo_tag,
73+
func.count()
74+
.filter(effective_publication_state == published_state)
75+
.label("available_record_count"),
76+
func.count()
77+
.filter(resources.c.gbl_suppressed_b.is_(True))
78+
.label("suppressed_record_count"),
79+
func.count()
80+
.filter(effective_publication_state != published_state)
81+
.label("unpublished_record_count"),
9282
)
93-
.where(effective_publication_state != published_state)
94-
.scalar_subquery()
83+
.select_from(resources.join(repo_tags, true()))
84+
.where(func.starts_with(repo_tags.c.repo_tag, ogm_repo_tag_prefix))
85+
.group_by(repo_tags.c.repo_tag)
86+
.subquery("resource_counts")
9587
)
9688

9789
q = (
@@ -109,17 +101,35 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]:
109101
ogm_harvest_runs.c.ogm_completed_at.label("last_run_completed_at"),
110102
ogm_harvest_runs.c.ogm_status.label("last_run_status"),
111103
ogm_harvest_runs.c.ogm_stats_json.label("last_run_stats_json"),
112-
harvested_record_count.label("harvested_record_count"),
113-
published_available_record_count.label("available_record_count"),
114-
suppressed_record_count.label("suppressed_record_count"),
115-
unpublished_record_count.label("unpublished_record_count"),
104+
func.coalesce(harvested_counts.c.harvested_record_count, 0).label(
105+
"harvested_record_count"
106+
),
107+
func.coalesce(resource_counts.c.available_record_count, 0).label(
108+
"available_record_count"
109+
),
110+
func.coalesce(resource_counts.c.suppressed_record_count, 0).label(
111+
"suppressed_record_count"
112+
),
113+
func.coalesce(resource_counts.c.unpublished_record_count, 0).label(
114+
"unpublished_record_count"
115+
),
116116
)
117117
.select_from(
118118
ogm_repos.outerjoin(
119119
latest_run_ids, ogm_repos.c.ogm_repo_name == latest_run_ids.c.repo_name
120-
).outerjoin(
120+
)
121+
.outerjoin(
121122
ogm_harvest_runs, ogm_harvest_runs.c.ogm_id == latest_run_ids.c.latest_ogm_id
122123
)
124+
.outerjoin(
125+
harvested_counts,
126+
harvested_counts.c.ogm_repo_name == ogm_repos.c.ogm_repo_name,
127+
)
128+
.outerjoin(
129+
resource_counts,
130+
resource_counts.c.repo_tag
131+
== func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name),
132+
)
123133
)
124134
.order_by(ogm_repos.c.ogm_repo_name)
125135
)

‎backend/tests/services/test_ogm_harvest_repository.py‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,3 +33,86 @@ async def fake_fetch_all(query):
3333
assert "nullif(resources.b1g_publication_state_s, CAST(" in sql
3434
assert "nullif(resources.publication_state, CAST(" in sql
3535
assert ") = CAST(" in sql
36+
37+
38+
@pytest.mark.integration
39+
@pytest.mark.asyncio
40+
async def test_public_repo_counts_preserve_tag_and_publication_semantics(
41+
monkeypatch, db_connection
42+
):
43+
"""Exercise real PostgreSQL array/count behavior using connection-local tables."""
44+
import os
45+
46+
import asyncpg
47+
from sqlalchemy.schema import CreateTable
48+
49+
from db.models import ogm_harvest_runs, ogm_repos, ogm_resource_state
50+
51+
url = os.getenv("OGM_QUERY_TEST_DATABASE_URL")
52+
if not url and db_connection is not None:
53+
url = str(db_connection.url).replace("postgresql+asyncpg://", "postgresql://")
54+
if not url:
55+
pytest.skip("Set OGM_QUERY_TEST_DATABASE_URL to run isolated PostgreSQL query tests")
56+
connection = await asyncpg.connect(url)
57+
try:
58+
for table in (ogm_repos, ogm_harvest_runs, ogm_resource_state):
59+
ddl = str(CreateTable(table).compile(dialect=database._backend._dialect))
60+
await connection.execute(ddl.replace("CREATE TABLE", "CREATE TEMP TABLE", 1))
61+
await connection.execute("""
62+
CREATE TEMP TABLE resources (
63+
id text PRIMARY KEY, "b1g_adminTags_sm" varchar[],
64+
b1g_publication_state_s varchar, publication_state varchar,
65+
gbl_suppressed_b boolean
66+
);
67+
INSERT INTO ogm_repos (ogm_repo_name, ogm_enabled) VALUES
68+
('alpha', true), ('beta', false), ('empty', true);
69+
INSERT INTO ogm_resource_state
70+
(ogm_repo_name, ogm_resource_id, ogm_missing_since) VALUES
71+
('alpha', 'active', NULL), ('alpha', 'missing', now()),
72+
('beta', 'active', NULL);
73+
INSERT INTO ogm_harvest_runs
74+
(ogm_id, ogm_repo_name, ogm_trigger, ogm_status) VALUES
75+
(1, 'alpha', 'manual', 'failed'), (2, 'alpha', 'manual', 'success');
76+
""")
77+
await connection.executemany(
78+
"""INSERT INTO resources VALUES ($1, $2, $3, $4, $5)""",
79+
[
80+
("duplicate", ["ogm_repo:alpha", "ogm_repo:alpha"], None, None, None),
81+
("shared", ["ogm_repo:alpha", "ogm_repo:beta"], "PUBLISHED", "draft", True),
82+
("fallback", ["ogm_repo:alpha"], "", "Published", False),
83+
("unpublished", ["ogm_repo:alpha"], "Draft", "published", True),
84+
("legacy", ["ogm_repo:alpha"], None, "draft", None),
85+
("blank", ["ogm_repo:beta"], "", "", False),
86+
("null-tags", None, None, None, True),
87+
("empty-tags", [], None, None, True),
88+
("other-tags", [None, "ogmXrepo:alpha", "ogm_repo:unknown"], None, None, True),
89+
],
90+
)
91+
92+
async def fetch_all(query):
93+
compiled = query.compile(
94+
dialect=database._backend._dialect,
95+
compile_kwargs={"render_postcompile": True},
96+
)
97+
parameters = [value for _, value in sorted(compiled.params.items())]
98+
return await connection.fetch(_compile_like_databases(query), *parameters)
99+
100+
monkeypatch.setattr(ogm_repository.database, "fetch_all", fetch_all)
101+
summaries = await OGMHarvestRepository().list_public_repo_summaries()
102+
assert [row["ogm_repo_name"] for row in summaries] == ["alpha", "beta", "empty"]
103+
counts = {
104+
row["ogm_repo_name"]: (
105+
row["harvested_record_count"],
106+
row["available_record_count"],
107+
row["suppressed_record_count"],
108+
row["unpublished_record_count"],
109+
)
110+
for row in summaries
111+
}
112+
assert counts == {"alpha": (1, 3, 2, 2), "beta": (1, 2, 1, 0), "empty": (0, 0, 0, 0)}
113+
assert summaries[0]["last_run_id"] == 2
114+
assert summaries[0]["last_crawl_status"] == "success"
115+
assert summaries[1]["ogm_enabled"] is False
116+
assert summaries[2]["last_run_id"] is None
117+
finally:
118+
await connection.close()

0 commit comments

Comments
 (0)