diff --git a/alembic/versions/20260927_0048_playback_sources_sort_date.py b/alembic/versions/20260927_0048_playback_sources_sort_date.py new file mode 100644 index 0000000..0a544e6 --- /dev/null +++ b/alembic/versions/20260927_0048_playback_sources_sort_date.py @@ -0,0 +1,86 @@ +"""Data sortowania sceny powielona w playback_sources (lista scen jednego serwisu). + +Ekran serwisu w apce (`/scenes?origin=tube:X&sort=release_date`) sortuje po dacie z +`scenes`, a filtruje po `playback_sources`. Dla serwisow, ktorych scen nie ma wsrod +najnowszych w katalogu (perverzija, hdporn.gg, tubepornclassic, yesporn), przejscie po +ix_scenes_effective_date_desc nie znajduje strony w 1 s, wiec API bierze wszystkie id +scen serwisu (20-115 tys.) i sortuje je po dacie. Na rozgrzanym cache 1-4 s, na zimnym +9-19 s (log API 2026-09-27, 39 wejsc w 3 h). Ten sam problem co przy tagach (0042). + +`sort_date` = to samo wyrazenie co w 0042 i w `_effective_date_desc()`. Indeks +(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL z migracji 0049 daje od +razu najnowsze sceny serwisu. + +Utrzymanie w bazie, bo do playback_sources pisze ingest, merge (przenosi zrodla miedzy +scenami) i skrypty: + - BEFORE INSERT / UPDATE OF scene_id na playback_sources: data z wiersza sceny, + - wyzwalacz scen z 0042 poprawia teraz tez zrodla tej sceny. + +Kolumna nullable i bez DEFAULT, wiec ALTER jest natychmiastowy. Istniejace 4,4 mln +wierszy wypelnia scripts/backfill_playback_sort_date.py, dopiero potem indeks (0049) +i przelaczenie zapytania. + +Revision ID: 0048_playback_sort_date +Revises: 0047_tag_vertical_counts +Create Date: 2026-09-27 +""" +from __future__ import annotations + +from alembic import op + +revision = "0048_playback_sort_date" +down_revision: str | None = "0047_tag_vertical_counts" +branch_labels = None +depends_on = None + +_SORT_DATE = "COALESCE({t}.release_date, ({t}.created_at AT TIME ZONE 'UTC')::date)" + + +def upgrade() -> None: + op.execute("SET LOCAL lock_timeout = '5s'") + op.execute("ALTER TABLE playback_sources ADD COLUMN IF NOT EXISTS sort_date date") + op.execute(f""" + CREATE OR REPLACE FUNCTION playback_sources_fill_sort_date() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + SELECT {_SORT_DATE.format(t='s')} INTO NEW.sort_date + FROM scenes s WHERE s.id = NEW.scene_id; + RETURN NEW; + END $$ + """) + op.execute(""" + CREATE TRIGGER trg_playback_sources_fill_sort_date + BEFORE INSERT OR UPDATE OF scene_id ON playback_sources + FOR EACH ROW EXECUTE FUNCTION playback_sources_fill_sort_date() + """) + # Ta sama funkcja co w 0042, z dopisanym playback_sources. Wyzwalacz na scenes + # zostaje bez zmian. + op.execute(f""" + CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger + LANGUAGE plpgsql AS $$ + DECLARE d date := {_SORT_DATE.format(t='NEW')}; + BEGIN + UPDATE scene_tags SET sort_date = d + WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d; + UPDATE playback_sources SET sort_date = d + WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d; + RETURN NULL; + END $$ + """) + + +def downgrade() -> None: + op.execute("SET LOCAL lock_timeout = '5s'") + op.execute(f""" + CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger + LANGUAGE plpgsql AS $$ + DECLARE d date := {_SORT_DATE.format(t='NEW')}; + BEGIN + UPDATE scene_tags SET sort_date = d + WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d; + RETURN NULL; + END $$ + """) + op.execute("DROP TRIGGER IF EXISTS trg_playback_sources_fill_sort_date ON playback_sources") + op.execute("DROP FUNCTION IF EXISTS playback_sources_fill_sort_date()") + op.execute("ALTER TABLE playback_sources DROP COLUMN IF EXISTS sort_date") diff --git a/alembic/versions/20260927_0049_playback_origin_sort_idx.py b/alembic/versions/20260927_0049_playback_origin_sort_idx.py new file mode 100644 index 0000000..c4c514a --- /dev/null +++ b/alembic/versions/20260927_0049_playback_origin_sort_idx.py @@ -0,0 +1,35 @@ +"""Indeks "najnowsze sceny serwisu" (kolumna z 0048). + +(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL: lista scen serwisu, +sort=release_date, idzie po tym indeksie od najnowszej i konczy po LIMIT, niezaleznie od +tego, gdzie w calym katalogu leza sceny serwisu. Uzasadnienie i pomiary w 0048. + +Zakladac DOPIERO po wypelnieniu `sort_date` (scripts/backfill_playback_sort_date.py), +z tego samego powodu co 0043. CONCURRENTLY w autocommit_block; nieudany zostawia indeks +INVALID, wtedy DROP INDEX i ponownie. + +Revision ID: 0049_playback_origin_sort_idx +Revises: 0048_playback_sort_date +Create Date: 2026-09-27 +""" +from __future__ import annotations + +from alembic import op + +revision = "0049_playback_origin_sort_idx" +down_revision: str | None = "0048_playback_sort_date" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_playback_sources_origin_sort " + "ON playback_sources (origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL" + ) + + +def downgrade() -> None: + with op.get_context().autocommit_block(): + op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_playback_sources_origin_sort") diff --git a/app/api/scenes.py b/app/api/scenes.py index 1578025..b74d88c 100644 --- a/app/api/scenes.py +++ b/app/api/scenes.py @@ -759,7 +759,14 @@ def list_scenes( # stwierdzic: 148 s na produkcji (pomiar 2026-09-16). Zwieramy od razu. base = base.where(false() if origin else ~live_playback_exists()) - if origin: + # Ekran serwisu od najnowszych: sceny serwisu bierzemy wprost z indeksu + # ix_playback_sources_origin_sort (data sortu powielona w zrodle, migracje 0048/0049) + # zamiast szukac ich w calym katalogu po dacie. Wtedy JOIN w galezi sortu zastepuje + # ponizszy EXISTS. Z fraza albo tagiem zostaja dotychczasowe sciezki. + origin_sort = bool( + sort == "release_date" and origin and ":" in origin and tag_sort_id is None and not q + ) + if origin and not origin_sort: # Pelny origin z ekranu strony ('tube:hdporngg') porownujemy wprost: rownosc idzie # po ix_playback_sources_origin_live, a ILIKE '%...%' przeszukiwal kazde zrodlo # kazdej kandydujacej sceny (wyszukiwanie w obrebie strony 4,6 s, Sentry Slow DB @@ -911,6 +918,21 @@ def list_scenes( ordered = base.join( st, and_(st.scene_id == Scene.id, st.tag_id == tag_sort_id) ).order_by(st.sort_date.desc(), st.scene_id.desc()) + elif origin_sort: + # Scena moze miec dwa zywe zrodla z tego samego serwisu (190 tys. par), a oba + # niosa te sama date, wiec DISTINCT ON zgodny z porzadkiem indeksu zbija je w + # jeden wiersz bez sortowania: skan dalej konczy sie po LIMIT. Pomiar i + # uzasadnienie w migracji 0048 (perverzija 13,7 s na zimnym cache). + ps_sort = ( + select(PlaybackSource.scene_id, PlaybackSource.sort_date) + .where(PlaybackSource.origin == origin, PlaybackSource.dead_at.is_(None)) + .distinct(PlaybackSource.sort_date, PlaybackSource.scene_id) + .order_by(PlaybackSource.sort_date.desc(), PlaybackSource.scene_id.desc()) + .subquery("origin_sort") + ) + ordered = base.join(ps_sort, ps_sort.c.scene_id == Scene.id).order_by( + ps_sort.c.sort_date.desc(), ps_sort.c.scene_id.desc() + ) else: ordered = base.order_by( _effective_date_desc(), Scene.created_at.desc(), Scene.id.desc() @@ -1054,7 +1076,7 @@ def list_scenes( Scene.id == any_(bindparam("origin_scene_ids", origin_ids, type_=ARRAY(PG_UUID(as_uuid=True)))) ) - if rows is None and origin and not q and sort in ("created_at", "release_date"): + if rows is None and origin and not q and not origin_sort and sort in ("created_at", "release_date"): walked = _search_walk_ids(session, ordered, _off, _want, _ORIGIN_LIST_WALK_BUDGET_MS) if walked is not None: rows = _scenes_in_order(session, walked) diff --git a/app/models/playback_source.py b/app/models/playback_source.py index 01ff7bc..c8c5f4b 100644 --- a/app/models/playback_source.py +++ b/app/models/playback_source.py @@ -8,9 +8,10 @@ nie tworzył duplikatów. from __future__ import annotations import uuid -from datetime import datetime +from datetime import date, datetime from sqlalchemy import ( + Date, DateTime, ForeignKey, Integer, @@ -66,3 +67,8 @@ class PlaybackSource(UUIDPKMixin, TimestampMixin, Base): API filtruje takie sources z listy. None = aktywny.""" dead_reason: Mapped[str | None] = mapped_column(String(512), nullable=True) + + # Data sortowania sceny (premiera albo dzien dodania), powielona tu pod indeks + # "najnowsze sceny serwisu" (migracje 0048 i 0049). Pisze ja WYZWALACZ w bazie, + # nie aplikacja: wartosc podana przy INSERT i tak zostanie nadpisana. + sort_date: Mapped[date | None] = mapped_column(Date) diff --git a/scripts/backfill_playback_sort_date.py b/scripts/backfill_playback_sort_date.py new file mode 100644 index 0000000..8bff458 --- /dev/null +++ b/scripts/backfill_playback_sort_date.py @@ -0,0 +1,76 @@ +"""Wypelnia playback_sources.sort_date w istniejacych wierszach (migracja 0048). + +Kopia scripts/backfill_scene_tags_sort_date.py dla playback_sources: partiami po zakresach +stron heapu, commit po kazdej partii, VACUUM co kilka partii, pauza oddaje dysk apce. +Nowe wiersze w trakcie wypelnia juz wyzwalacz z 0048. Wznawialny: rusza tylko wiersze z +sort_date IS NULL, na koncu dobiera resztki. + + docker compose run -d --name backfill_ps_sort worker \ + python3 -u /srv/scripts/backfill_playback_sort_date.py [stron_na_partie] [pauza_s] +Po nim: alembic upgrade (0049, indeks) i dopiero wtedy kod, ktory z indeksu korzysta. +""" +from __future__ import annotations + +import sys +import time + +from sqlalchemy import text + +from app.db import engine + +PAGES = int(sys.argv[1]) if len(sys.argv) > 1 else 1000 +PAUSE = float(sys.argv[2]) if len(sys.argv) > 2 else 10.0 +VACUUM_EVERY = 8 +SORT_DATE = "COALESCE(s.release_date, (s.created_at AT TIME ZONE 'UTC')::date)" + + +def rel_pages() -> int: + with engine.connect() as c: + return int(c.execute(text( + "SELECT pg_relation_size('playback_sources') / current_setting('block_size')::int" + )).scalar()) + + +def vacuum(analyze: bool = False) -> None: + with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as c: + c.execute(text("VACUUM (ANALYZE) playback_sources" if analyze else "VACUUM playback_sources")) + + +t0 = time.time() +total = 0 +page = 0 +batch = 0 +pages = rel_pages() +print(f"playback_sources: {pages} stron, partia {PAGES} stron, pauza {PAUSE} s", flush=True) +while page < pages: + tb = time.time() + with engine.begin() as c: + c.execute(text("SET LOCAL statement_timeout = '600s'")) + n = c.execute(text(f""" + UPDATE playback_sources ps SET sort_date = {SORT_DATE} + FROM scenes s + WHERE s.id = ps.scene_id AND ps.sort_date IS NULL + AND ps.ctid >= format('(%s,0)', CAST(:p0 AS int))::tid + AND ps.ctid < format('(%s,0)', CAST(:p1 AS int))::tid + """), {"p0": page, "p1": page + PAGES}).rowcount + total += n + page += PAGES + batch += 1 + if batch % VACUUM_EVERY == 0: + vacuum() + pages = rel_pages() # tabela moze urosnac, zanim VACUUM zwolni miejsce + print(f"partia {batch}: strony do {page}/{pages}, {n} wierszy w {time.time() - tb:.1f} s, " + f"razem {total}, {time.time() - t0:.0f} s", flush=True) + if n: + time.sleep(PAUSE) + +with engine.begin() as c: + rest = c.execute(text(f""" + UPDATE playback_sources ps SET sort_date = {SORT_DATE} + FROM scenes s WHERE s.id = ps.scene_id AND ps.sort_date IS NULL + """)).rowcount +print(f"resztki: {rest}", flush=True) +vacuum(analyze=True) +with engine.connect() as c: + left = c.execute(text("SELECT count(*) FROM playback_sources WHERE sort_date IS NULL")).scalar() +print(f"KONIEC: {total + rest} wierszy, puste po wszystkim: {left}, {time.time() - t0:.0f} s", flush=True)