perf(db): each feed sort rides an index, and an ICU upgrade rebuilds the title ones
The views, duration and title sorts had no index: every page read the whole videos heap and kept a top-N (prod 350 ms a page, 550-780 for title). Migration 0096 adds one per direction and, for title, per reader collation, in the exact ORDER BY feedquery builds. Localdev prod copy: plain sorts 180-310 ms -> ~1 ms, page 20 3-5 ms, narrow filters unchanged. view_count/duration are only written by the enrich, which already rewrites indexed columns, so no HOT update is lost. The title indexes are the first on an ICU collation, and postgres:16-alpine floats: collation.heal_drift rebuilds them (CONCURRENTLY) at startup when Postgres reports the ICU version moved. test_feed_sort_indexes holds every videos sort to an index and checks the rebuild.
This commit is contained in:
@@ -0,0 +1,59 @@
|
||||
"""videos: an index per feed sort, so a sorted page stops after its rows (E31 S4e)
|
||||
|
||||
The feed's views, duration and title sorts had no index. Every page (the first and each next cursor)
|
||||
read the whole videos heap, filtered it and kept a top-N: on prod ~350 ms a page for views/duration,
|
||||
550–780 ms for title (the ICU comparison on every row), 2.3 s for page 2 when the copies of one request
|
||||
queued. With an index in the exact ORDER BY (`feedquery.order_by`: the key, NULLS LAST, then id) the
|
||||
planner walks it and stops after the page. On a prod-copy localdev every plain sort fell from 180–310 ms
|
||||
to ~1 ms, deep pages (20) to 3–5 ms, and narrow filters (a tag, a channel, a year, nothing found) stayed
|
||||
at 0.3–25 ms.
|
||||
|
||||
One index per direction: a backward scan of `DESC NULLS LAST` gives `ASC NULLS FIRST`, not the
|
||||
`ASC NULLS LAST` the asc sorts ask for. The title pair is per reader language (`app.collation.ICU`):
|
||||
an index only serves an ORDER BY in its own collation.
|
||||
|
||||
Cost: ~130 MB (localdev: 10 + 10 + 8 + 8 + 4 × 23). Writes stay cheap: view_count and duration are
|
||||
written by the enrich, which rewrites indexed columns already (the revalidation job only touches
|
||||
last_checked_at), so no HOT update is lost. A search combined with one of these sorts can now walk the
|
||||
index too: a term matching ~2 % of the videos took 130–200 ms instead of 7 (still under the plain sort it
|
||||
replaces); rarer and broader terms stay fast. Fencing the index off under a search made broad terms
|
||||
(`m`, `the`) take 2 s, so it is left to the planner.
|
||||
|
||||
The title indexes are the schema's first ones on an ICU collation: an ICU upgrade in the Postgres image
|
||||
reorders them, and `app.collation.heal_drift` rebuilds them at startup when it sees one.
|
||||
tests/test_feed_sort_indexes.py holds each sort and its index together.
|
||||
|
||||
CONCURRENTLY (autocommit_block) builds them without blocking a live prod; IF (NOT) EXISTS keeps the drift
|
||||
test's downgrade → re-upgrade round trip idempotent.
|
||||
"""
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0096_videos_sort_indexes"
|
||||
down_revision: str | None = "0095_videos_channel_cover_index"
|
||||
branch_labels: str | None = None
|
||||
depends_on: str | None = None
|
||||
|
||||
# name → the indexed ORDER BY (feedquery.sort_keys + order_by: key NULLS LAST, then id).
|
||||
INDEXES = {
|
||||
"ix_videos_views_desc": "view_count DESC NULLS LAST, id",
|
||||
"ix_videos_views_asc": "view_count ASC NULLS LAST, id",
|
||||
"ix_videos_duration_desc": "duration_seconds DESC NULLS LAST, id",
|
||||
"ix_videos_duration_asc": "duration_seconds ASC NULLS LAST, id",
|
||||
"ix_videos_title_en": '(lower(title) COLLATE "en-x-icu") ASC NULLS LAST, id',
|
||||
"ix_videos_title_en_desc": '(lower(title) COLLATE "en-x-icu") DESC NULLS LAST, id',
|
||||
"ix_videos_title_hu": '(lower(title) COLLATE "hu-x-icu") ASC NULLS LAST, id',
|
||||
"ix_videos_title_hu_desc": '(lower(title) COLLATE "hu-x-icu") DESC NULLS LAST, id',
|
||||
}
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.get_context().autocommit_block():
|
||||
for name, columns in INDEXES.items():
|
||||
op.execute(f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} ON videos ({columns})")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.get_context().autocommit_block():
|
||||
for name in INDEXES:
|
||||
op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {name}")
|
||||
@@ -7,17 +7,25 @@ sort names one: Hungarian puts "Ö" after "O" and "Cs" after "Cu", English folds
|
||||
|
||||
`collated` wraps an ORDER BY expression — and because a keyset cursor compares with the SAME expression
|
||||
(`feedquery.sort_keys`), the cursor pages in the same order it sorted in. `sorted_words` orders a Python
|
||||
list the same way through the database, so a list built in Python matches one built in SQL exactly."""
|
||||
list the same way through the database, so a list built in Python matches one built in SQL exactly.
|
||||
|
||||
The feed's title sorts ride indexes built in these collations (migration 0096), and an index keeps the
|
||||
order of the ICU that built it. `postgres:16-alpine` is a floating tag: an image update can bring a newer
|
||||
ICU, after which those indexes disagree with the comparisons a query makes (a keyset page skips or
|
||||
repeats a title). `heal_drift` rebuilds them at startup once Postgres reports the version moved."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import SQLColumnExpression, column, func, select, values
|
||||
from sqlalchemy import Engine, SQLColumnExpression, column, func, select, text, values
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.sql.elements import ColumnElement
|
||||
from sqlalchemy.types import String
|
||||
|
||||
from app import logs
|
||||
from app.mailtext import normalize
|
||||
|
||||
log = logs.get_logger(__name__)
|
||||
|
||||
ICU: dict[str, str] = {"en": "en-x-icu", "hu": "hu-x-icu"}
|
||||
|
||||
|
||||
@@ -37,3 +45,39 @@ def sorted_words(db: Session, words: list[str], lang: str | None) -> list[str]:
|
||||
return []
|
||||
w = values(column("w", String), name="words").data([(x,) for x in words])
|
||||
return list(db.scalars(select(w.c.w).order_by(collated(w.c.w, lang), w.c.w)))
|
||||
|
||||
|
||||
def heal_drift(engine: Engine) -> list[str]:
|
||||
"""Rebuild every index on an ICU collation whose library version moved, then record the new version.
|
||||
Returns the rebuilt indexes. CONCURRENTLY: the feed keeps reading while it runs."""
|
||||
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
|
||||
drifted = conn.execute(
|
||||
text(
|
||||
"SELECT oid, collname FROM pg_collation WHERE collname = ANY(:names) "
|
||||
"AND collversion IS DISTINCT FROM pg_collation_actual_version(oid)"
|
||||
),
|
||||
{"names": list(ICU.values())},
|
||||
).all()
|
||||
if not drifted:
|
||||
return []
|
||||
indexes = list(
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT DISTINCT d.objid::regclass::text FROM pg_depend d "
|
||||
"JOIN pg_class c ON c.oid = d.objid AND c.relkind = 'i' "
|
||||
"WHERE d.classid = 'pg_class'::regclass "
|
||||
"AND d.refclassid = 'pg_collation'::regclass AND d.refobjid = ANY(:oids)"
|
||||
),
|
||||
{"oids": [row.oid for row in drifted]},
|
||||
).scalars()
|
||||
)
|
||||
log.warning(
|
||||
"ICU collation version changed (%s): rebuilding %d indexes",
|
||||
", ".join(row.collname for row in drifted),
|
||||
len(indexes),
|
||||
)
|
||||
for name in indexes:
|
||||
conn.execute(text(f"REINDEX INDEX CONCURRENTLY {name}"))
|
||||
for row in drifted:
|
||||
conn.execute(text(f'ALTER COLLATION "{row.collname}" REFRESH VERSION'))
|
||||
return indexes
|
||||
|
||||
+18
-2
@@ -24,11 +24,12 @@ from sqlalchemy.exc import DataError
|
||||
from starlette.middleware.sessions import SessionMiddleware
|
||||
|
||||
from app import bodylimit
|
||||
from app import collation
|
||||
from app import logsink
|
||||
from app import state
|
||||
from app import sysconfig
|
||||
from app.config import settings
|
||||
from app.db import SessionLocal
|
||||
from app.db import SessionLocal, engine
|
||||
from app.plex import stream as plex_stream
|
||||
from app.plex import subtitles as plex_subtitles
|
||||
from app.routes import (
|
||||
@@ -97,6 +98,9 @@ async def lifespan(app: FastAPI):
|
||||
# owns, so racing it against early playback is safe.
|
||||
sweep = asyncio.create_task(_sweep_hls_orphans())
|
||||
reaper = asyncio.create_task(_reap_hls_sessions())
|
||||
# Fire and forget as well: a rebuild after an ICU upgrade takes seconds, and the title sorts serve
|
||||
# meanwhile (CONCURRENTLY) in the old index's order, as they did before the restart.
|
||||
heal = asyncio.create_task(_heal_collation_drift())
|
||||
start_scheduler()
|
||||
try:
|
||||
yield
|
||||
@@ -106,7 +110,7 @@ async def lifespan(app: FastAPI):
|
||||
# pending ("Task was destroyed but it is pending!" — noise that reads like a bug during
|
||||
# triage). Cancelling does NOT interrupt the rmtree thread itself; this only makes our own
|
||||
# bookkeeping honest.
|
||||
for task in (sweep, reaper):
|
||||
for task in (sweep, reaper, heal):
|
||||
task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await task
|
||||
@@ -139,6 +143,18 @@ async def _reap_hls_sessions() -> None:
|
||||
log.exception("HLS session reap failed")
|
||||
|
||||
|
||||
async def _heal_collation_drift() -> None:
|
||||
"""Background startup check of the ICU collation the title-sort indexes are built in (see
|
||||
`collation.heal_drift`). A failure is logged, never fatal: the sorts still answer, only possibly in
|
||||
the old order."""
|
||||
try:
|
||||
await asyncio.to_thread(collation.heal_drift, engine)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception:
|
||||
log.exception("ICU collation drift check failed")
|
||||
|
||||
|
||||
async def _sweep_hls_orphans() -> None:
|
||||
"""Background half of the startup HLS sweep (see lifespan). Owns its own errors: a failure here
|
||||
must never take the app down, and a cancelled shutdown is not one."""
|
||||
|
||||
@@ -349,6 +349,9 @@ class Video(Base, TimestampMixin):
|
||||
DateTime(timezone=True), index=True, nullable=False, server_default=func.now()
|
||||
)
|
||||
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
|
||||
# The feed's views / duration / title sorts each ride a migration-defined index in their exact
|
||||
# ORDER BY, one per direction (and per ICU collation for title): migration 0096, held to
|
||||
# `feedquery.order_by` by test_feed_sort_indexes.
|
||||
duration_seconds: Mapped[int | None] = mapped_column(Integer)
|
||||
view_count: Mapped[int | None] = mapped_column(BigInteger)
|
||||
like_count: Mapped[int | None] = mapped_column(BigInteger)
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Every feed sort on a videos column rides an index, and an ICU upgrade rebuilds the title ones (E31 S4e).
|
||||
|
||||
Without an index in the exact ORDER BY, each feed page read the whole videos heap and kept a top-N
|
||||
(prod: 350 ms a page for views/duration, 550–780 ms for title). Migration 0096 adds one per direction
|
||||
and per reader language. The planner is told it may not sort or scan the table, so a plan that still
|
||||
sorts it whole means no index matches the ORDER BY `feedquery.order_by` builds: a key, direction, NULLS
|
||||
rule or collation changed on one side only. An Incremental Sort is fine: the index gives the key's order
|
||||
and only its ties are sorted by id (`newest`/`oldest` ride the published_at indexes that way). DB lane.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
from app import collation
|
||||
from app.db import engine
|
||||
from app.feedquery import order_by, sort_keys
|
||||
from app.models import Video
|
||||
|
||||
VIDEO_SORTS = [
|
||||
"newest",
|
||||
"oldest",
|
||||
"views",
|
||||
"views_asc",
|
||||
"duration_desc",
|
||||
"duration_asc",
|
||||
"title",
|
||||
"title_desc",
|
||||
]
|
||||
|
||||
|
||||
def _plan(db, sort: str, lang: str) -> str:
|
||||
stmt = select(Video.id).order_by(*order_by(sort_keys(sort, 0, None, lang))).limit(60)
|
||||
sql = str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
|
||||
db.execute(text("SET LOCAL enable_seqscan = off"))
|
||||
db.execute(text("SET LOCAL enable_sort = off"))
|
||||
db.execute(text("SET LOCAL enable_bitmapscan = off"))
|
||||
return "\n".join(db.execute(text(f"EXPLAIN {sql}")).scalars())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lang", sorted(collation.ICU))
|
||||
@pytest.mark.parametrize("sort", VIDEO_SORTS)
|
||||
def test_the_sort_walks_an_index(db, sort, lang):
|
||||
plan = _plan(db, sort, lang)
|
||||
db.rollback()
|
||||
full_sort = re.search(r"(?<!Incremental )Sort \(", plan)
|
||||
assert not full_sort, f"{sort} ({lang}) sorts the table:\n{plan}"
|
||||
assert "Index Scan" in plan or "Index Only Scan" in plan, plan
|
||||
|
||||
|
||||
def _drifted(db) -> list[str]:
|
||||
return list(
|
||||
db.execute(
|
||||
text(
|
||||
"SELECT collname FROM pg_collation WHERE collname = ANY(:names) "
|
||||
"AND collversion IS DISTINCT FROM pg_collation_actual_version(oid)"
|
||||
),
|
||||
{"names": list(collation.ICU.values())},
|
||||
).scalars()
|
||||
)
|
||||
|
||||
|
||||
def _files(db) -> dict[str, int]:
|
||||
return dict(
|
||||
db.execute(
|
||||
text("SELECT relname, relfilenode FROM pg_class WHERE relname LIKE 'ix_videos_title_%'")
|
||||
).all()
|
||||
)
|
||||
|
||||
|
||||
def test_an_icu_upgrade_rebuilds_the_title_indexes(db):
|
||||
# What an image update with a newer ICU looks like from inside: the recorded version no longer
|
||||
# matches the library's (the test database runs as the bootstrap superuser, as the app's does).
|
||||
db.execute(text("UPDATE pg_collation SET collversion = '0.0' WHERE collname = 'hu-x-icu'"))
|
||||
db.commit()
|
||||
try:
|
||||
assert _drifted(db) == ["hu-x-icu"]
|
||||
before = _files(db)
|
||||
# A concurrent rebuild waits for every open transaction, this session's too.
|
||||
db.rollback()
|
||||
rebuilt = collation.heal_drift(engine)
|
||||
assert sorted(rebuilt) == ["ix_videos_title_hu", "ix_videos_title_hu_desc"]
|
||||
after = _files(db)
|
||||
# Rebuilt = written to new files; the English pair, on an unchanged ICU, is left alone.
|
||||
assert {n for n in before if before[n] != after[n]} == set(rebuilt)
|
||||
assert _drifted(db) == []
|
||||
finally:
|
||||
db.execute(
|
||||
text(
|
||||
"UPDATE pg_collation SET collversion = pg_collation_actual_version(oid) "
|
||||
"WHERE collname = 'hu-x-icu'"
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
|
||||
|
||||
def test_no_drift_rebuilds_nothing(db):
|
||||
assert collation.heal_drift(engine) == []
|
||||
Reference in New Issue
Block a user