feat(ml): drop image_record.tagger_predictions — image_prediction is sole store (#768 step 3)
Read cutover verified in prod (suggestions + allowlist read image_prediction; backfill complete at 908k rows / 51k images). Removes the old JSON column and everything that fed it: - ImageRecord.tagger_predictions column removed; migration 0046 DROPs it. tagger_model_version kept as the "tagged / current?" signal the backfill sweep reads (needs-tagging check switched to tagger_model_version IS NULL). - tag_and_embed no longer dual-writes the JSON — image_prediction is the only write path. - importer re-import reset drops the JSON line (image_prediction rows are already deleted on re-import). - Retired the one-time #768 backfill task + the #764 prune task, their admin endpoints, and their Maintenance cards (Backfill/PrunePredictionsCard). - Tests seed/assert via image_prediction; stale column refs removed. Disk reclaim is NOT automatic: DROP COLUMN is a catalog change. Run `VACUUM FULL image_record` off-hours afterward to return the ~100 GB to the OS so DB backups go small (#739). image_prediction (~90 MB) stays in pg_dump — it's the source of truth now. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -207,212 +207,3 @@ def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
|
||||
)
|
||||
rescan_series_suggestions_task.delay(summary["resume_after_id"])
|
||||
return summary
|
||||
|
||||
|
||||
@celery.task(
|
||||
name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
|
||||
bind=True,
|
||||
autoretry_for=(OperationalError, DBAPIError),
|
||||
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
||||
soft_time_limit=3600, time_limit=4200, # 60 min / 70 min
|
||||
)
|
||||
def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
|
||||
"""One-time #764 backfill: drop tagger_predictions entries below the DB
|
||||
store floor (ml_settings.tagger_store_floor) from existing image_record
|
||||
rows, and clamp any allowlist min_confidence below the floor up to it.
|
||||
|
||||
The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
|
||||
near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
|
||||
each row to the new floor. Keyset by id ASC (restart-safe via after_id);
|
||||
idempotent — already-pruned rows rewrite to themselves and are skipped.
|
||||
Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
|
||||
image_record afterward to return the disk to the OS.
|
||||
|
||||
The keep predicate (confidence >= floor) mirrors Tagger.infer's store
|
||||
gate so backfilled rows match what new imports store. Self-resumes on the
|
||||
soft time limit (re-enqueues from the last committed id)."""
|
||||
from celery.exceptions import SoftTimeLimitExceeded
|
||||
from sqlalchemy import select, update
|
||||
|
||||
from ..models import ImageRecord, MLSettings, TagAllowlist
|
||||
|
||||
SessionLocal = _sync_session_factory()
|
||||
scanned = 0
|
||||
pruned = 0
|
||||
clamped = 0
|
||||
last_id = after_id
|
||||
try:
|
||||
with SessionLocal() as session:
|
||||
floor = session.execute(
|
||||
select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
|
||||
).scalar_one()
|
||||
# Clamp allowlist thresholds below the new floor once, on the
|
||||
# first pass (#764 consumer #4) — a sub-floor min_confidence can't
|
||||
# apply more permissively now that nothing below it is stored.
|
||||
if after_id == 0:
|
||||
clamped = session.execute(
|
||||
update(TagAllowlist)
|
||||
.where(TagAllowlist.min_confidence < floor)
|
||||
.values(min_confidence=floor)
|
||||
).rowcount or 0
|
||||
session.commit()
|
||||
|
||||
while True:
|
||||
rows = session.execute(
|
||||
select(ImageRecord.id, ImageRecord.tagger_predictions)
|
||||
.where(ImageRecord.id > last_id)
|
||||
.where(ImageRecord.tagger_predictions.is_not(None))
|
||||
.order_by(ImageRecord.id.asc())
|
||||
.limit(500)
|
||||
).all()
|
||||
if not rows:
|
||||
break
|
||||
for image_id, preds in rows:
|
||||
scanned += 1
|
||||
if not preds:
|
||||
continue
|
||||
kept = {
|
||||
name: p for name, p in preds.items()
|
||||
if float(p.get("confidence", 0.0)) >= floor
|
||||
}
|
||||
if len(kept) != len(preds):
|
||||
session.execute(
|
||||
update(ImageRecord)
|
||||
.where(ImageRecord.id == image_id)
|
||||
.values(tagger_predictions=kept)
|
||||
)
|
||||
pruned += 1
|
||||
session.commit()
|
||||
last_id = rows[-1].id # advance only after commit, for resume
|
||||
except SoftTimeLimitExceeded:
|
||||
log.warning(
|
||||
"prune_low_confidence_predictions soft-limited at id=%s "
|
||||
"(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
|
||||
)
|
||||
prune_low_confidence_predictions_task.delay(last_id)
|
||||
return {
|
||||
"partial": True, "last_id": last_id,
|
||||
"scanned": scanned, "pruned": pruned,
|
||||
}
|
||||
|
||||
log.info(
|
||||
"prune_low_confidence_predictions complete: floor=%s scanned=%d "
|
||||
"pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
|
||||
)
|
||||
return {
|
||||
"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,
|
||||
}
|
||||
|
||||
@@ -157,15 +157,15 @@ def tag_and_embed(self, image_id: int) -> dict:
|
||||
)
|
||||
|
||||
phase = "persist"
|
||||
record.tagger_predictions = preds
|
||||
record.tagger_model_version = settings.tagger_model_version
|
||||
record.siglip_embedding = embedding.tolist()
|
||||
record.siglip_model_version = settings.embedder_model_version
|
||||
session.add(record)
|
||||
# Write the normalized image_prediction rows (#768). Delete-then-
|
||||
# insert keeps a re-tag idempotent. tagger_store_floor was already
|
||||
# applied in tagger.infer, so preds is the >=floor set. (Transitional
|
||||
# dual-write alongside the JSON column until the read cutover lands.)
|
||||
# Write the normalized image_prediction rows (#768) — the sole home
|
||||
# for predictions now (image_record.tagger_predictions was dropped in
|
||||
# migration 0046). Delete-then-insert keeps a re-tag idempotent;
|
||||
# tagger_store_floor was already applied in tagger.infer, so preds is
|
||||
# the >=floor set.
|
||||
session.execute(
|
||||
delete(ImagePrediction).where(
|
||||
ImagePrediction.image_record_id == image_id
|
||||
@@ -282,7 +282,7 @@ def backfill(self) -> int:
|
||||
select(ImageRecord.id)
|
||||
.where(ImageRecord.id > last_id)
|
||||
.where(
|
||||
(ImageRecord.tagger_predictions.is_(None))
|
||||
(ImageRecord.tagger_model_version.is_(None))
|
||||
| (
|
||||
ImageRecord.tagger_model_version
|
||||
!= settings.tagger_model_version
|
||||
|
||||
Reference in New Issue
Block a user