From 65211a3f2f30fd5fe171e612dd5baf2a70977e88 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 11 Jun 2026 09:18:25 -0400 Subject: [PATCH] fix(migration): make 0045 DDL-only; backfill image_prediction via batched task (#768) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The inline INSERT…SELECT backfill in migration 0045 wrapped the table creation and a ~100 GB pass over image_record.tagger_predictions in one transaction: nothing committed until the end, it was unmonitorable, and an earlier MATERIALIZED-CTE form spilled the full 100 GB to temp on NFS. A deploy got stuck on it for ~2h with image_prediction never appearing. Split the concerns: - 0045 now creates ONLY the table + indexes (instant DDL → web boots). - New backend.app.tasks.admin.backfill_image_predictions_task copies the >= store-floor predictions from the JSON into image_prediction, batched by id window and committed per chunk: live progress, resumable (re-enqueues from the last committed id), idempotent (ON CONFLICT DO NOTHING). json_each stays in the DB executor streaming each window — no Python-side 100 GB load, no materialization. - POST /api/admin/maintenance/backfill-predictions + a Maintenance-tab card to trigger the one-time run after upgrading. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../versions/0045_image_prediction_table.py | 63 ++++------ backend/app/api/admin.py | 13 ++ backend/app/tasks/admin.py | 113 ++++++++++++++++++ .../settings/BackfillPredictionsCard.vue | 50 ++++++++ .../components/settings/MaintenancePanel.vue | 2 + tests/test_tasks_admin.py | 7 ++ 6 files changed, 206 insertions(+), 42 deletions(-) create mode 100644 frontend/src/components/settings/BackfillPredictionsCard.vue diff --git a/alembic/versions/0045_image_prediction_table.py b/alembic/versions/0045_image_prediction_table.py index ebfa6c2..df11de2 100644 --- a/alembic/versions/0045_image_prediction_table.py +++ b/alembic/versions/0045_image_prediction_table.py @@ -1,13 +1,22 @@ -"""image_prediction table + backfill from image_record.tagger_predictions +"""image_prediction table (DDL only — backfill runs as a background task) Normalizes the per-image tagger predictions out of the JSON blob into a -queryable table (#768). Backfills from the existing JSON in one set-based -INSERT…SELECT over json_each — fast because the #764 prune already shrank -each row to its >=0.70 entries. The old image_record.tagger_predictions -column is left in place here (vestigial) and dropped in a follow-up once the -code cutover is verified — dropping it needs an ACCESS EXCLUSIVE lock on the -hot image_record table (the 0044 lock class), so it's deferred to a -quiesced-worker window. +queryable table (#768). This migration creates ONLY the table + indexes — it +is pure DDL and commits instantly, so web boots immediately. + +The data backfill from the existing image_record.tagger_predictions JSON is +deliberately NOT done here. Doing it inline made the whole migration one +transaction over the ~100 GB TOAST: nothing committed until the very end, it +was invisible/unmonitorable mid-run, and an early MATERIALIZED-CTE form spilled +the full 100 GB to temp. Instead the backfill is the +backend.app.tasks.admin.backfill_image_predictions_task — batched by id window, +committed per chunk (visible progress + resumable), idempotent +(ON CONFLICT DO NOTHING). Trigger it from Settings → Maintenance once web is up. + +The old image_record.tagger_predictions column is left in place (vestigial) and +dropped in a follow-up once the backfill + code cutover are verified — dropping +it needs an ACCESS EXCLUSIVE lock on the hot image_record table (the 0044 lock +class), so it's deferred to a quiesced-worker window. Revision ID: 0045 Revises: 0044 @@ -48,40 +57,10 @@ def upgrade() -> None: "ix_image_prediction_name_score", "image_prediction", ["raw_name", "score"], ) - # Backfill from the JSON blob. json_each expands {name: {category, - # confidence}} into one row per prediction. category defaults to 'general' - # to mirror the suggestion read path; rows with no confidence are skipped. - # Filter to >= the store floor (ml_settings.tagger_store_floor, default - # 0.70) right here so this is self-sufficient — it does NOT depend on the - # #764 prune having run, and extracting only the >=floor tail keeps - # image_prediction small (~tens of rows/image) even from the full JSON. - # Guard json_each against non-object rows (some tagger_predictions are JSON - # scalars/null → "cannot deconstruct a scalar"). The inline CASE passes an - # empty object for those, so json_each yields nothing — a single STREAMING - # pass with NO materialization/temp spill (an earlier MATERIALIZED-CTE guard - # forced ~100 GB to temp on NFS and was pathologically slow). - op.execute( - """ - INSERT INTO image_prediction (image_record_id, raw_name, category, score) - SELECT ir.id, - je.key, - COALESCE(je.value ->> 'category', 'general'), - (je.value ->> 'confidence')::double precision - FROM image_record ir, - json_each( - CASE WHEN json_typeof(ir.tagger_predictions) = 'object' - THEN ir.tagger_predictions - ELSE '{}'::json END - ) je - WHERE je.value ->> 'confidence' IS NOT NULL - AND (je.value ->> 'confidence')::double precision - >= COALESCE( - (SELECT tagger_store_floor FROM ml_settings WHERE id = 1), - 0.70 - ) - ON CONFLICT (image_record_id, raw_name) DO NOTHING - """ - ) + # No data backfill here — see the module docstring. The one-time copy from + # image_record.tagger_predictions runs as backfill_image_predictions_task + # (batched, resumable, idempotent), kept out of this transaction so web boots + # without waiting on a ~100 GB pass. def downgrade() -> None: diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py index 3f77d9a..fed2c77 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -360,3 +360,16 @@ async def trigger_prune_predictions(): async_result = prune_low_confidence_predictions_task.delay() return jsonify({"task_id": async_result.id, "status": "queued"}), 202 + + +@admin_bp.route("/maintenance/backfill-predictions", methods=["POST"]) +async def trigger_backfill_predictions(): + """Operator-triggered #768 backfill: copy stored tagger predictions from the + image_record.tagger_predictions JSON into the normalized image_prediction + table. Batched + resumable + idempotent; runs on the maintenance_long lane. + Run this once after deploying migration 0045 (which creates the empty table) + to populate predictions for the existing library.""" + from ..tasks.admin import backfill_image_predictions_task + + async_result = backfill_image_predictions_task.delay() + return jsonify({"task_id": async_result.id, "status": "queued"}), 202 diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index 7123a71..4ca9faf 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -303,3 +303,116 @@ def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict: "floor": floor, "scanned": scanned, "pruned": pruned, "allowlist_clamped": clamped, "last_id": last_id, } + + +# Backfill image_prediction from image_record.tagger_predictions (#768). +# Deliberately NOT done in migration 0045: a single INSERT…SELECT over the +# ~100 GB TOAST is one transaction — invisible until commit, unmonitorable, and +# the MATERIALIZED-CTE form spilled the whole 100 GB to temp on NFS. Instead we +# walk image_record in id WINDOWS, running a bounded INSERT…SELECT over each +# window and committing per chunk: progress is visible (image_prediction grows +# live), it's resumable (re-enqueues from the last committed id), and json_each +# stays in the DB executor streaming each window (no Python-side 100 GB load, no +# materialization). Idempotent via ON CONFLICT DO NOTHING. +_BACKFILL_PRED_CHUNK_SECONDS = 600 # re-enqueue boundary, like normalize_tags +_BACKFILL_PRED_ID_WINDOW = 2000 # image_record ids per committed batch + + +@celery.task( + name="backend.app.tasks.admin.backfill_image_predictions_task", + bind=True, + autoretry_for=(OperationalError, DBAPIError), + retry_backoff=15, retry_backoff_max=180, max_retries=1, + soft_time_limit=1800, time_limit=2400, # 30 min / 40 min +) +def backfill_image_predictions_task(self, after_id: int = 0) -> dict: + """One-time #768 backfill: copy each image_record's stored tagger + predictions (the >= store-floor entries) from the tagger_predictions JSON + into the normalized image_prediction table. + + Batched by id window + committed per chunk so it's monitorable and + resumable; idempotent (ON CONFLICT DO NOTHING) so re-running is safe. + Filters to >= ml_settings.tagger_store_floor (default 0.70) so the table + stays small even from the full pre-prune JSON tail. Guards json_each against + non-object rows (scalar/null tagger_predictions → "cannot deconstruct a + scalar") via an inline CASE. Self-resumes on the soft time limit.""" + import time + + from celery.exceptions import SoftTimeLimitExceeded + from sqlalchemy import func, select, text + + from ..models import ImageRecord, MLSettings + + _INSERT_WINDOW = text( + """ + INSERT INTO image_prediction (image_record_id, raw_name, category, score) + SELECT ir.id, + je.key, + COALESCE(je.value ->> 'category', 'general'), + (je.value ->> 'confidence')::double precision + FROM image_record ir, + json_each( + CASE WHEN json_typeof(ir.tagger_predictions) = 'object' + THEN ir.tagger_predictions + ELSE '{}'::json END + ) je + WHERE ir.id > :lo AND ir.id <= :hi + AND je.value ->> 'confidence' IS NOT NULL + AND (je.value ->> 'confidence')::double precision >= :floor + ON CONFLICT (image_record_id, raw_name) DO NOTHING + """ + ) + + SessionLocal = _sync_session_factory() + started = time.monotonic() + last_id = after_id + inserted = 0 + windows = 0 + with SessionLocal() as session: + floor = session.execute( + select(MLSettings.tagger_store_floor).where(MLSettings.id == 1) + ).scalar_one() + max_id = session.execute( + select(func.max(ImageRecord.id)) + ).scalar() or 0 + + try: + while last_id < max_id: + hi = last_id + _BACKFILL_PRED_ID_WINDOW + res = session.execute( + _INSERT_WINDOW, {"lo": last_id, "hi": hi, "floor": floor} + ) + session.commit() + inserted += res.rowcount or 0 + windows += 1 + last_id = hi # advance only after commit, for resume + if time.monotonic() - started > _BACKFILL_PRED_CHUNK_SECONDS: + log.info( + "backfill_image_predictions chunk done (windows=%d " + "inserted=%d up to id=%d/%d) — re-enqueuing", + windows, inserted, min(last_id, max_id), max_id, + ) + backfill_image_predictions_task.delay(last_id) + return { + "partial": True, "last_id": last_id, "max_id": max_id, + "inserted": inserted, "windows": windows, + } + except SoftTimeLimitExceeded: + log.warning( + "backfill_image_predictions soft-limited at id=%d " + "(inserted=%d) — re-enqueuing", last_id, inserted, + ) + backfill_image_predictions_task.delay(last_id) + return { + "partial": True, "last_id": last_id, "max_id": max_id, + "inserted": inserted, "windows": windows, + } + + log.info( + "backfill_image_predictions complete: floor=%s inserted=%d windows=%d " + "max_id=%d", floor, inserted, windows, max_id, + ) + return { + "floor": floor, "inserted": inserted, "windows": windows, + "max_id": max_id, "last_id": max_id, + } diff --git a/frontend/src/components/settings/BackfillPredictionsCard.vue b/frontend/src/components/settings/BackfillPredictionsCard.vue new file mode 100644 index 0000000..f1d6aaa --- /dev/null +++ b/frontend/src/components/settings/BackfillPredictionsCard.vue @@ -0,0 +1,50 @@ + + + diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 69761d1..3baf30e 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -12,6 +12,7 @@ + @@ -32,6 +33,7 @@ import MLBackfillCard from './MLBackfillCard.vue' import CentroidRecomputeCard from './CentroidRecomputeCard.vue' import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue' import MLThresholdSliders from './MLThresholdSliders.vue' +import BackfillPredictionsCard from './BackfillPredictionsCard.vue' import PrunePredictionsCard from './PrunePredictionsCard.vue' import AllowlistTable from './AllowlistTable.vue' import AliasTable from './AliasTable.vue' diff --git a/tests/test_tasks_admin.py b/tests/test_tasks_admin.py index d975259..80bf4e1 100644 --- a/tests/test_tasks_admin.py +++ b/tests/test_tasks_admin.py @@ -49,6 +49,13 @@ def test_prune_low_confidence_predictions_task_registered(): ) +def test_backfill_image_predictions_task_registered(): + assert ( + "backend.app.tasks.admin.backfill_image_predictions_task" + in celery.tasks + ) + + @pytest.mark.asyncio async def test_prune_low_confidence_predictions(db_sync, tmp_path): # #764: drop stored tagger predictions below the store floor (default -- 2.52.0