nearle_daily/catalogue_backend
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

image vector dimensionality reduction

Browse Source
This commit is contained in:
sriram
2026-09-17 14:21:48 +05:30
parent deae694a1f
commit afa0bfa743
11 changed files with 907 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
scripts/backfill_image_vectors.py
View File

@@ -4,11 +4,12 @@ Fill `img_vector` for the rows that already have a primary image.
WHY THIS EXISTS
---------------
New writes get their pixel vector from the worker `upsert_brand_products`
New writes get their image embedding from the worker `upsert_brand_products`
schedules (app/services/image_vector.py). Rows written before the column
existed never pass through that hook, and rows whose download failed at the
time (a CDN that 403'd, a timeout) are left NULL on purpose. This script is
the second chance for both.
the second chance for both - and, with `--force`, the way to recompute every
row after the model or its preprocessing changes.
WHAT IT DOES, PER TABLE
-----------------------
@@ -19,8 +20,10 @@ WHAT IT DOES, PER TABLE
primary. `image_vector.primary_image_url` decides what "primary" means,
the same way the product card does.
3. Downloads each image with a small thread pool, no more than one request
in flight per host and a pause between them, decodes it to 32x32 RGB and
writes the 3072 values with a targeted UPDATE. `updated_at` is untouched.
in flight per host and a pause between them, runs it through the
MobileNetV3 embedder (inference is serialised on the embedder's lock; the
downloads are what run in parallel) and writes the 1024 L2-normalised
values with a targeted UPDATE. `updated_at` is untouched.
USAGE
-----
@@ -28,10 +31,12 @@ USAGE
python -m scripts.backfill_image_vectors --brands Amul --apply
python -m scripts.backfill_image_vectors --all --apply
python -m scripts.backfill_image_vectors --all --recompute-stale --apply
python -m scripts.backfill_image_vectors --all --force --apply # model changed
A dry run downloads nothing; it reports how many rows would be attempted and
shows the first few URLs. No backup is written: nothing else writes this
column, and `--recompute-stale` repairs any row.
A dry run downloads nothing and needs no model; it reports how many rows would
be attempted and shows the first few URLs. `--apply` refuses to start when the
embedder cannot load. No backup is written: nothing else writes this column,
and `--force` rebuilds any row.
"""
from __future__ import annotations
@@ -53,6 +58,7 @@ from app.infrastructure.settings import ( # noqa: E402
DB_NAME,
IMAGE_VECTOR_HOST_PAUSE_SECONDS,
)
from app.services import image_embedder # noqa: E402
from app.services import image_vector as iv # noqa: E402
from app.services.vector_store import _connect # noqa: E402
from scripts.repair_brand_images import _brand_tables # noqa: E402
@@ -76,7 +82,7 @@ class _HostPacer:
with self._lock:
return self._busy.setdefault(host, threading.Lock())
def fetch(self, url: str, timeout: Optional[float]) -> Tuple[Optional[List[int]], str]:
def fetch(self, url: str, timeout: Optional[float]) -> Tuple[Optional[List[float]], str]:
host = urlparse(url).netloc.lower()
with self._host_lock(host):
with self._lock:
@@ -97,6 +103,7 @@ def backfill_table(
*,
apply: bool,
recompute_stale: bool,
force: bool,
limit: Optional[int],
workers: int,
timeout: Optional[float],
@@ -110,14 +117,15 @@ def backfill_table(
entry["skipped"] = True
return entry
rows = iv.rows_needing_vectors(cur, table, recompute_stale=recompute_stale, limit=limit)
rows = iv.rows_needing_vectors(cur, table, recompute_stale=recompute_stale, limit=limit, force=force)
entry["candidates"] = len(rows)
if not rows:
logger.info("%s: nothing to do", table)
return entry
stale = sum(1 for r in rows if not r.get("missing"))
logger.info("%s: %d row(s) to compute (%d missing, %d stale)", table, len(rows), len(rows) - stale, stale)
logger.info("%s: %d row(s) to compute (%d missing, %d %s)", table, len(rows), len(rows) - stale, stale,
"forced" if force else "stale")
if not apply:
for r in rows[:5]:
logger.info(" %-42s %s", (r.get("product_name") or "")[:42], r["primary_url"][:90])
@@ -125,7 +133,7 @@ def backfill_table(
logger.info(" ... and %d more", len(rows) - 5)
return entry
results: Dict[str, Tuple[Optional[List[int]], str]] = {}
results: Dict[str, Tuple[Optional[List[float]], str]] = {}
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {
pool.submit(pacer.fetch, r["primary_url"], timeout): r["image_id"] for r in rows
@@ -161,6 +169,8 @@ def main() -> int:
parser.add_argument("--limit", type=int, default=None, help="at most N rows per table")
parser.add_argument("--recompute-stale", action="store_true",
help="also redo rows whose vector came from a URL that is no longer the primary")
parser.add_argument("--force", action="store_true",
help="redo EVERY row with a primary image - after a model or preprocessing change")
parser.add_argument("--workers", type=int, default=3, help=f"download threads (1-{MAX_WORKERS}, default 3)")
parser.add_argument("--timeout", type=float, default=None, help="per-request timeout in seconds")
parser.add_argument("--pause", type=float, default=IMAGE_VECTOR_HOST_PAUSE_SECONDS,
@@ -172,6 +182,13 @@ def main() -> int:
logger.info("Target database: %s / %s", DB_HOST, DB_NAME)
logger.info("Mode: %s", "APPLY - this writes" if args.apply else "DRY RUN - nothing is downloaded or written")
if args.apply:
# Fail before the first download, not after: a missing model would
# otherwise cost every image fetch and write nothing.
logger.info("Embedder: %s", image_embedder.describe())
if not image_embedder.available():
logger.error("The embedder cannot run - fix the above before --apply.")
return 2
conn = _connect()
if conn is None:
@@ -186,7 +203,7 @@ def main() -> int:
for _suffix, table in _brand_tables(cur, brands):
report.append(backfill_table(
cur, conn, table,
apply=args.apply, recompute_stale=args.recompute_stale,
apply=args.apply, recompute_stale=args.recompute_stale, force=args.force,
limit=args.limit, workers=workers, timeout=args.timeout, pacer=pacer,
))
finally:

52
scripts/migrate_brand_schema.py
View File

@@ -33,6 +33,13 @@ WHAT IT WILL NOT DO
`barcode_last_updated` and REAL for the tax figures rather than the
types those values look like they want. Type drift is REPORTED here,
never silently "fixed".
THE ONE EXCEPTION is `img_vector`. A column of any dimension other than
the embedder's (vector_store.IMG_VECTOR_DIMS) is DROPPED and re-added
empty, `img_vector_src` is cleared, and the column gets its hnsw index -
see `_ensure_img_vector_type`. That discards every stored vector on the
table, on purpose: a vector of the wrong width is not data, and the
backfill script rebuilds them. Those statements are shown as `~` lines.
* It will not create a brand table that does not exist.
* It will not touch `nutrition_facts` or any non-brand table.
@@ -108,13 +115,13 @@ _TYPE_ALIASES = {
"JSONB": {"jsonb"},
"DOUBLE PRECISION": {"double precision"},
"vector(384)": {"USER-DEFINED"},
"vector(3072)": {"USER-DEFINED"},
"vector(1024)": {"USER-DEFINED"},
}
# vector(3072) needs the 16000-dimension ceiling pgvector raised to in 0.4.0.
# Below that, `_ensure_columns` logs and skips the column rather than failing
# the write, and this script says so up front.
_MIN_PGVECTOR_FOR_IMG_VECTOR = (0, 4, 0)
# The hnsw index on img_vector needs pgvector 0.5.0. Below that,
# `_ensure_img_vector_type` logs and skips the index (an exact scan still
# works), and this script says so up front.
_MIN_PGVECTOR_FOR_IMG_VECTOR = (0, 5, 0)
def _pgvector_version(cur) -> Optional[str]:
@@ -184,8 +191,8 @@ def main() -> int:
pgv = _pgvector_version(cur)
logger.info("pgvector : %s", pgv or "unknown (extension not found?)")
if pgv and _version_tuple(pgv) < _MIN_PGVECTOR_FOR_IMG_VECTOR:
logger.warning("pgvector %s is older than %s: img_vector vector(3072) will be "
"skipped on this server (every other column still applies).",
logger.warning("pgvector %s is older than %s: the hnsw index on img_vector will be "
"skipped on this server (the column itself still applies).",
pgv, ".".join(map(str, _MIN_PGVECTOR_FOR_IMG_VECTOR)))
logger.info("")
@@ -198,6 +205,7 @@ def main() -> int:
report: List[Dict[str, Any]] = []
total_missing = 0
total_drift = 0
total_vec = 0
try:
with conn.cursor() as cur:
@@ -213,6 +221,21 @@ def main() -> int:
_ensure_columns(recorder, table)
adds = [s for s in recorder.statements if "ADD COLUMN" in s]
# The img_vector retype/index (see the docstring's one
# exception). A DROP means the ADD that follows is a re-add,
# not a missing column - keep it out of that list.
vec_ops = [
s for s in recorder.statements
if "img_vector" in s and (
("DROP COLUMN" in s) or s.startswith("UPDATE") or s.startswith("CREATE INDEX")
)
]
if any("DROP COLUMN img_vector" in s for s in vec_ops):
readd = [s for s in adds if "IF NOT EXISTS img_vector " in s]
adds = [s for s in adds if s not in readd]
vec_ops = [s for s in vec_ops if "DROP COLUMN" in s] + readd + \
[s for s in vec_ops if "DROP COLUMN" not in s]
# Type drift: a column that exists but whose type is not what
# col_defs would have created. Reported, never altered.
drift = []
@@ -228,12 +251,15 @@ def main() -> int:
"table": table,
"missing_columns": [s.split("ADD COLUMN IF NOT EXISTS ")[1] for s in adds],
"type_drift": drift,
"img_vector": vec_ops,
}
report.append(entry)
total_missing += len(adds)
total_drift += len(drift)
if vec_ops:
total_vec += 1
if apply and adds:
if apply and (adds or vec_ops):
_ensure_columns(cur, table)
if apply:
@@ -246,12 +272,14 @@ def main() -> int:
return 0
width = max(len(e["table"]) for e in report)
changed = [e for e in report if e["missing_columns"] or e["type_drift"]]
changed = [e for e in report if e["missing_columns"] or e["type_drift"] or e["img_vector"]]
for entry in sorted(changed, key=lambda e: -len(e["missing_columns"])):
logger.info("%-*s %d column(s) missing", width, entry["table"],
len(entry["missing_columns"]))
for col in entry["missing_columns"]:
logger.info("%-*s + %s", width, "", col)
for stmt in entry["img_vector"]:
logger.info("%-*s ~ %s", width, "", stmt)
for d in entry["type_drift"]:
logger.info("%-*s ! %s is %s, col_defs declares %s (NOT changed)",
width, "", d["column"], d["actual"], d["declared"])
@@ -261,7 +289,11 @@ def main() -> int:
len(report), len(report) - len(changed))
logger.info("%d column(s) %s, %d type mismatch(es) reported",
total_missing, "added" if apply else "would be added", total_drift)
if not apply and total_missing:
if total_vec:
logger.info("%d table(s) %s img_vector retype/index (stored vectors on a retyped "
"table are discarded - run scripts/backfill_image_vectors afterwards)",
total_vec, "had" if apply else "need")
if not apply and (total_missing or total_vec):
logger.info("")
logger.info("Re-run with --apply to write these changes.")
return 0