m038
d9ba194b30
Consolidate the scattered run instructions and pass/fail criteria into one place: - design spec gains 'Running the spike' (venv bootstrap + flags + DB prereqs) and 'Definition of done' (Req 0-4 are the bar; coverage does not gate done-ness; the only follow-up at ~100% CLIP coverage is one no-flag re-run to snapshot the final coverage into the contract doc) sections, and the status line now reads 'passed 2026-06-27'. - script docstring Usage now includes the missing 'python3 -m venv .venv' bootstrap a fresh checkout needs, and points to the spec's definition of done.
584 lines
26 KiB
Python
584 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""pgvector embedding feasibility spike — disposable read-only probe.
|
||
|
||
Answers the M1.5 prerequisite questions against the *live* Immich Postgres and
|
||
records them in a findings doc. See the design spec:
|
||
docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md
|
||
|
||
Questions (pass/fail):
|
||
1. Can we read the CLIP embeddings at all?
|
||
2. Can each embedding join back to a photo we track (embedding key -> assets.id)?
|
||
3. Exact shapes: table, embedding column, vector dimension, distance operator.
|
||
4. Coverage over the *embeddable* (IMAGE) population — raw and image-only ratios.
|
||
|
||
This is read-only and idempotent. The DB session is opened read-only at the
|
||
session layer (not merely by which statements we issue), so even Immich's
|
||
write-capable postgres user cannot mutate the source-of-truth DB while we probe.
|
||
|
||
Usage (from the repo root; a fresh checkout / new worktree has no .venv):
|
||
python3 -m venv .venv
|
||
.venv/bin/python -m pip install -r scripts/requirements-spike.txt
|
||
.venv/bin/python scripts/pgvector_spike.py # read-only, idempotent, safe to re-run
|
||
|
||
IMMICH_DB_URL is read from the environment, else from the nearest .env searched upward from the
|
||
working directory (postgresql://USER:PASS@HOST:PORT/DBNAME). Or pass it inline:
|
||
IMMICH_DB_URL=postgresql://user:pass@host:5432/immich .venv/bin/python scripts/pgvector_spike.py
|
||
|
||
Flags: --sqlite PATH (optional SQLite cross-check), --findings PATH, --no-write (report only).
|
||
See docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md for the definition of done.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import textwrap
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import psycopg
|
||
from psycopg import sql
|
||
|
||
# pgvector type candidates to probe first (verified against the catalog, not trusted).
|
||
KNOWN_TABLE = "smart_search"
|
||
KNOWN_COLUMN = "embedding"
|
||
# pgvector opclass -> (distance operator, human name).
|
||
OPCLASS_OPERATORS = {
|
||
"vector_cosine_ops": ("<=>", "cosine"),
|
||
"vector_l2_ops": ("<->", "L2 / euclidean"),
|
||
"vector_ip_ops": ("<#>", "negative inner product"),
|
||
}
|
||
DEFAULT_FINDINGS = "docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md"
|
||
|
||
|
||
# --------------------------------------------------------------------------- env
|
||
def find_dsn() -> Optional[str]:
|
||
"""IMMICH_DB_URL from the environment, else from the nearest .env walking up."""
|
||
dsn = (os.environ.get("IMMICH_DB_URL") or "").strip()
|
||
if dsn:
|
||
return dsn
|
||
here = Path.cwd().resolve()
|
||
for base in [here, *here.parents, Path(__file__).resolve().parent.parent]:
|
||
env_file = base / ".env"
|
||
if not env_file.is_file():
|
||
continue
|
||
for line in env_file.read_text().splitlines():
|
||
line = line.strip()
|
||
if line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, val = line.partition("=")
|
||
if key.strip() == "IMMICH_DB_URL":
|
||
val = val.strip().strip('"').strip("'")
|
||
if val:
|
||
return val
|
||
return None
|
||
|
||
|
||
def connect_readonly(dsn: str) -> psycopg.Connection:
|
||
"""Open a connection and force the whole session read-only.
|
||
|
||
Belt-and-suspenders: psycopg's ``read_only`` wraps each transaction READ ONLY,
|
||
and ``default_transaction_read_only`` covers anything that slips outside it.
|
||
"""
|
||
conn = psycopg.connect(dsn, autocommit=True)
|
||
conn.execute("SET default_transaction_read_only = on")
|
||
conn.read_only = True
|
||
return conn
|
||
|
||
|
||
# ----------------------------------------------------------------------- discovery
|
||
def discover_vector_columns(conn: psycopg.Connection) -> list[tuple[str, str, str]]:
|
||
"""All (schema, table, column) holding a pgvector ``vector`` column.
|
||
|
||
Catalog-driven so it survives Immich's cross-version schema drift. The known
|
||
``smart_search.embedding`` pair is sorted first when present.
|
||
"""
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT n.nspname, c.relname, a.attname
|
||
FROM pg_attribute a
|
||
JOIN pg_class c ON c.oid = a.attrelid
|
||
JOIN pg_namespace n ON n.oid = c.relnamespace
|
||
JOIN pg_type t ON t.oid = a.atttypid
|
||
WHERE t.typname = 'vector'
|
||
AND a.attnum > 0 AND NOT a.attisdropped
|
||
AND c.relkind IN ('r', 'p') -- ordinary + partitioned tables
|
||
AND n.nspname NOT IN ('pg_catalog', 'information_schema')
|
||
ORDER BY (c.relname = %s AND a.attname = %s) DESC, n.nspname, c.relname
|
||
""",
|
||
(KNOWN_TABLE, KNOWN_COLUMN),
|
||
).fetchall()
|
||
return [(r[0], r[1], r[2]) for r in rows]
|
||
|
||
|
||
def vector_dimension(conn, schema: str, table: str, column: str) -> tuple[Optional[int], str]:
|
||
"""Vector dimension, preferring a live sample over the declared typmod."""
|
||
# atttypmod holds the declared dimension for pgvector (>0 when fixed).
|
||
typmod = conn.execute(
|
||
"""
|
||
SELECT a.atttypmod
|
||
FROM pg_attribute a
|
||
JOIN pg_class c ON c.oid = a.attrelid
|
||
JOIN pg_namespace n ON n.oid = c.relnamespace
|
||
WHERE n.nspname = %s AND c.relname = %s AND a.attname = %s
|
||
""",
|
||
(schema, table, column),
|
||
).fetchone()
|
||
declared = typmod[0] if typmod and typmod[0] and typmod[0] > 0 else None
|
||
|
||
sampled = None
|
||
try:
|
||
ident = sql.Identifier(schema, table)
|
||
col = sql.Identifier(column)
|
||
row = conn.execute(
|
||
sql.SQL("SELECT vector_dims({col}) FROM {tbl} WHERE {col} IS NOT NULL LIMIT 1")
|
||
.format(col=col, tbl=ident)
|
||
).fetchone()
|
||
sampled = row[0] if row else None
|
||
except Exception as exc: # noqa: BLE001 — record, don't abort the probe
|
||
return declared, f"declared typmod={declared}; live sample failed: {exc}"
|
||
|
||
if sampled is not None:
|
||
note = f"live sample={sampled}" + (f", declared typmod={declared}" if declared else "")
|
||
return sampled, note
|
||
return declared, f"declared typmod={declared}; table empty (no live vector to sample)"
|
||
|
||
|
||
def distance_operator(conn, schema: str, table: str, column: str) -> tuple[str, str]:
|
||
"""The distance operator Immich's index was built for, read from its opclass."""
|
||
row = conn.execute(
|
||
"""
|
||
SELECT opc.opcname
|
||
FROM pg_index i
|
||
JOIN pg_class ic ON ic.oid = i.indexrelid
|
||
JOIN pg_class tc ON tc.oid = i.indrelid
|
||
JOIN pg_namespace n ON n.oid = tc.relnamespace
|
||
JOIN pg_opclass opc ON opc.oid = ANY (i.indclass)
|
||
JOIN pg_attribute at ON at.attrelid = i.indrelid AND at.attnum = ANY (i.indkey)
|
||
WHERE n.nspname = %s AND tc.relname = %s AND at.attname = %s
|
||
AND opc.opcname LIKE 'vector_%%'
|
||
LIMIT 1
|
||
""",
|
||
(schema, table, column),
|
||
).fetchone()
|
||
if row and row[0] in OPCLASS_OPERATORS:
|
||
op, name = OPCLASS_OPERATORS[row[0]]
|
||
return op, f"{op} ({name}) — from index opclass {row[0]}"
|
||
return "<=>", "<=> (cosine) — assumed; no vector index opclass found on this column"
|
||
|
||
|
||
def self_similarity_ok(conn, schema, table, column, key_col, op) -> str:
|
||
"""Sanity-check the operator: a vector must be its own nearest neighbour (dist ~0).
|
||
|
||
Param-free (seed picked inside a CTE) so it works regardless of how psycopg
|
||
binds vector literals.
|
||
"""
|
||
try:
|
||
tbl, col, key = sql.Identifier(schema, table), sql.Identifier(column), sql.Identifier(key_col)
|
||
rows = conn.execute(
|
||
sql.SQL(
|
||
"WITH seed AS (SELECT {key} AS k, {col} AS v FROM {tbl} "
|
||
" WHERE {col} IS NOT NULL LIMIT 1) "
|
||
"SELECT (e.{key} = seed.k) AS is_seed, (e.{col} {op} seed.v) AS dist, "
|
||
" (seed.v {op} seed.v) AS self_dist "
|
||
"FROM {tbl} e, seed WHERE e.{col} IS NOT NULL ORDER BY dist ASC LIMIT 5"
|
||
).format(key=key, col=col, tbl=tbl, op=sql.SQL(op))
|
||
).fetchall()
|
||
if not rows:
|
||
return "skipped — no embeddings present"
|
||
is_seed, top_dist, self_dist = rows[0]
|
||
# Operator works iff an identical vector sorts first at the operator's own
|
||
# self-distance. That value is operator-dependent (0 for cosine <=> / L2 <->,
|
||
# ~-1 for inner product <#>), so compare the nearest distance to the seed's
|
||
# self-distance rather than to a hardcoded 0. The nearest row may be a duplicate
|
||
# photo rather than the seed itself — still a pass.
|
||
good = abs(float(top_dist) - float(self_dist)) < 1e-6
|
||
who = "the seed itself" if is_seed else "a duplicate (identical-vector) asset"
|
||
return (
|
||
f"{'OK' if good else 'UNEXPECTED'} — nearest neighbour ({who}) "
|
||
f"at distance {float(top_dist):.6g} (seed self-distance {float(self_dist):.6g})"
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
return f"failed: {exc}"
|
||
|
||
|
||
def detect_asset_key(conn, schema, table) -> tuple[Optional[str], Optional[str], Optional[str], str]:
|
||
"""Find the column + table joining the embedding row to the asset record.
|
||
|
||
Returns ``(key_col, asset_table, asset_pk, note)``. The asset table name is
|
||
*discovered from the FK*, never assumed — Immich renamed ``assets`` -> ``asset``
|
||
across versions, exactly the drift the spec warns about.
|
||
"""
|
||
fks = conn.execute(
|
||
"""
|
||
SELECT kcu.column_name, ccu.table_name, ccu.column_name
|
||
FROM information_schema.table_constraints tc
|
||
JOIN information_schema.key_column_usage kcu
|
||
ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema
|
||
JOIN information_schema.constraint_column_usage ccu
|
||
ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema
|
||
WHERE tc.constraint_type = 'FOREIGN KEY'
|
||
AND tc.table_schema = %s AND tc.table_name = %s
|
||
""",
|
||
(schema, table),
|
||
).fetchall()
|
||
# Prefer the FK whose target table is the asset record (singular/plural tolerated).
|
||
for col, ftable, fcol in fks:
|
||
if ftable in ("asset", "assets"):
|
||
return col, ftable, fcol, f"FK {table}.{col} -> {ftable}.{fcol}"
|
||
|
||
cols = {
|
||
r[0]
|
||
for r in conn.execute(
|
||
"SELECT column_name FROM information_schema.columns "
|
||
"WHERE table_schema = %s AND table_name = %s",
|
||
(schema, table),
|
||
).fetchall()
|
||
}
|
||
existing_tables = {
|
||
r[0]
|
||
for r in conn.execute(
|
||
"SELECT table_name FROM information_schema.tables "
|
||
"WHERE table_schema = %s AND table_name IN ('asset', 'assets')",
|
||
(schema,),
|
||
).fetchall()
|
||
}
|
||
asset_table = "asset" if "asset" in existing_tables else ("assets" if "assets" in existing_tables else None)
|
||
for cand in ("assetId", "asset_id", "assetsId"):
|
||
if cand in cols and asset_table:
|
||
return cand, asset_table, "id", f"no FK; matched column {table}.{cand} -> {asset_table}.id (assumed)"
|
||
return None, None, None, "no asset-key column found (FK or assetId/asset_id)"
|
||
|
||
|
||
def join_and_coverage(conn, schema, table, key_col, asset_table, assets_pk):
|
||
"""Join validity + coverage (raw and image-only) inside Postgres.
|
||
|
||
Returns a dict (the contract its three callers read by key):
|
||
n_embeddings_distinct, n_joined_to_assets, n_orphan, n_assets_total: int
|
||
n_assets_image, n_embedded_image: int | None (None when no asset 'type' column)
|
||
raw_ratio, image_ratio: float | None (None when the denominator is 0/None)
|
||
"""
|
||
tbl, key = sql.Identifier(schema, table), sql.Identifier(key_col)
|
||
atbl, apk = sql.Identifier(schema, asset_table), sql.Identifier(assets_pk)
|
||
|
||
n_emb = conn.execute(
|
||
sql.SQL("SELECT count(DISTINCT {key}) FROM {tbl} WHERE {key} IS NOT NULL")
|
||
.format(key=key, tbl=tbl)
|
||
).fetchone()[0]
|
||
n_joined = conn.execute(
|
||
sql.SQL(
|
||
"SELECT count(DISTINCT e.{key}) FROM {tbl} e "
|
||
"JOIN {atbl} a ON a.{apk} = e.{key}"
|
||
).format(key=key, tbl=tbl, atbl=atbl, apk=apk)
|
||
).fetchone()[0]
|
||
n_assets_total = conn.execute(
|
||
sql.SQL("SELECT count(*) FROM {atbl}").format(atbl=atbl)
|
||
).fetchone()[0]
|
||
|
||
# Image-only population — Immich never embeds video/audio/other with CLIP.
|
||
has_type = bool(
|
||
conn.execute(
|
||
"SELECT 1 FROM information_schema.columns "
|
||
"WHERE table_schema = %s AND table_name = %s AND column_name = 'type' LIMIT 1",
|
||
(schema, asset_table),
|
||
).fetchone()
|
||
)
|
||
n_assets_image = n_emb_image = None
|
||
if has_type:
|
||
n_assets_image = conn.execute(
|
||
sql.SQL("SELECT count(*) FROM {atbl} WHERE type = 'IMAGE'").format(atbl=atbl)
|
||
).fetchone()[0]
|
||
n_emb_image = conn.execute(
|
||
sql.SQL(
|
||
"SELECT count(DISTINCT a.{apk}) FROM {atbl} a "
|
||
"JOIN {tbl} e ON e.{key} = a.{apk} WHERE a.type = 'IMAGE'"
|
||
).format(apk=apk, atbl=atbl, tbl=tbl, key=key)
|
||
).fetchone()[0]
|
||
|
||
return {
|
||
"n_embeddings_distinct": n_emb,
|
||
"n_joined_to_assets": n_joined,
|
||
"n_orphan": n_emb - n_joined,
|
||
"n_assets_total": n_assets_total,
|
||
"n_assets_image": n_assets_image,
|
||
"n_embedded_image": n_emb_image,
|
||
"raw_ratio": (n_joined / n_assets_total) if n_assets_total else None,
|
||
"image_ratio": (n_emb_image / n_assets_image)
|
||
if (n_assets_image not in (None, 0)) else None,
|
||
}
|
||
|
||
|
||
def sqlite_crosscheck(conn, schema, table, key_col, sqlite_path: Path) -> str:
|
||
"""Optional confirmation that embedding asset IDs match our SQLite store.
|
||
|
||
Degrades gracefully: a fresh checkout has no `ingest` run, so an empty/absent
|
||
store is not a failure of the join requirement.
|
||
|
||
NOTE: this opens SQLite directly, a deliberate exception for this throwaway probe.
|
||
Production M1.5 code must NOT — it must go through shared/photoflow/core/store.py,
|
||
the only SQLite owner (CLAUDE.md). This raw read-only sample check stays in the spike.
|
||
"""
|
||
if not sqlite_path.is_file():
|
||
return f"skipped — no SQLite store at {sqlite_path} (fresh checkout; not a failure)"
|
||
try:
|
||
sample = [
|
||
str(r[0])
|
||
for r in conn.execute(
|
||
sql.SQL("SELECT {key} FROM {tbl} WHERE {key} IS NOT NULL LIMIT 50")
|
||
.format(key=sql.Identifier(key_col), tbl=sql.Identifier(schema, table))
|
||
).fetchall()
|
||
]
|
||
if not sample:
|
||
return "skipped — no embeddings to sample"
|
||
sconn = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
|
||
try:
|
||
tbl_exists = sconn.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='assets'"
|
||
).fetchone()
|
||
if not tbl_exists:
|
||
return "skipped — SQLite store has no assets table (unpopulated)"
|
||
total = sconn.execute("SELECT count(*) FROM assets").fetchone()[0]
|
||
if total == 0:
|
||
return "skipped — SQLite assets table empty (no ingest run; not a failure)"
|
||
placeholders = ",".join("?" * len(sample))
|
||
hits = sconn.execute(
|
||
f"SELECT count(*) FROM assets WHERE immich_id IN ({placeholders})", sample
|
||
).fetchone()[0]
|
||
finally:
|
||
sconn.close()
|
||
return f"{hits}/{len(sample)} sampled embedding asset IDs found in SQLite (store has {total} assets)"
|
||
except Exception as exc: # noqa: BLE001
|
||
return f"failed (non-fatal): {exc}"
|
||
|
||
|
||
def environment_meta(conn) -> dict:
|
||
"""Best-effort version context. Immich's *server* version is not in the DB."""
|
||
out = {}
|
||
try:
|
||
out["postgres_version"] = conn.execute("SHOW server_version").fetchone()[0]
|
||
except Exception as exc: # noqa: BLE001
|
||
out["postgres_version"] = f"unknown: {exc}"
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT extversion FROM pg_extension WHERE extname IN ('vector', 'vectors') LIMIT 1"
|
||
).fetchone()
|
||
out["pgvector_extension"] = row[0] if row else "not found"
|
||
except Exception as exc: # noqa: BLE001
|
||
out["pgvector_extension"] = f"unknown: {exc}"
|
||
# Newest applied migration as a proxy for schema age (table name varies by version).
|
||
out["latest_migration"] = "unknown"
|
||
for mig_table in ("migrations", "kysely_migration", "typeorm_metadata"):
|
||
try:
|
||
row = conn.execute(
|
||
sql.SQL("SELECT name FROM {} ORDER BY 1 DESC LIMIT 1")
|
||
.format(sql.Identifier(mig_table))
|
||
).fetchone()
|
||
if row:
|
||
out["latest_migration"] = f"{mig_table}: {row[0]}"
|
||
break
|
||
except Exception: # noqa: BLE001 — table absent in this version
|
||
continue
|
||
return out
|
||
|
||
|
||
# -------------------------------------------------------------------------- output
|
||
def _pct(ratio: Optional[float]) -> str:
|
||
"""Format a coverage ratio as a percentage, or 'n/a' when it could not be computed."""
|
||
return "n/a" if ratio is None else f"{ratio * 100:.1f}%"
|
||
|
||
|
||
def render_findings(facts: dict) -> str:
|
||
cov = facts["coverage"]
|
||
raw_pct = _pct(cov["raw_ratio"])
|
||
img_pct = _pct(cov["image_ratio"])
|
||
meta = facts["meta"]
|
||
return textwrap.dedent(
|
||
f"""\
|
||
# pgvector embedding feasibility spike — findings
|
||
|
||
**Generated:** {facts['generated']} by `scripts/pgvector_spike.py`
|
||
**Status:** machine-probed against the live Immich Postgres.
|
||
**This is M1.5's contract.** See the design spec:
|
||
`docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`.
|
||
|
||
> ⚠️ **Immich's undocumented, internal schema — no deprecation contract.**
|
||
> Everything below is valid **only** for the model + Immich version observed and
|
||
> can be renamed/restructured on any Immich upgrade. M1.5 must re-run this probe
|
||
> (or version-guard) on every Immich upgrade. The "contract" is version-pinned,
|
||
> not durable.
|
||
|
||
## Requirement 0 — does M1.5 need raw vectors (vs. a REST query)?
|
||
|
||
M1.5 clusters photos by *visual similarity* at trip level — it needs either the
|
||
raw CLIP embedding vectors or an arbitrary asset→asset nearest-neighbour query.
|
||
Immich's REST surface provides neither:
|
||
|
||
- `POST /api/search/smart` — **text→image** CLIP search: takes a text query,
|
||
returns assets. It never returns embedding vectors and cannot do asset→asset
|
||
similarity without a text prompt. Insufficient.
|
||
- `POST /api/search/metadata`, `/api/search/random` — metadata/random only; no
|
||
embeddings, no similarity.
|
||
- Duplicate detection (`/api/duplicates`) consumes embeddings *internally* but only
|
||
surfaces near-duplicate groups above Immich's own threshold — not a tunable
|
||
pairwise similarity usable for trip-level clustering. Insufficient.
|
||
- No documented endpoint returns raw CLIP vectors or arbitrary k-NN neighbours.
|
||
|
||
**Conclusion:** the load-bearing "REST can't expose embeddings" premise holds for
|
||
Immich's documented API → **read-only Postgres access (below) is the viable path.**
|
||
(Re-confirm against the OpenAPI of the running version on upgrade.)
|
||
|
||
## Requirements 1–4 — probed facts
|
||
|
||
| # | Question | Answer |
|
||
|---|----------|--------|
|
||
| 1 | Embeddings readable? | **{facts['readable']}** |
|
||
| 2 | Join to asset table? | **{facts['join_summary']}** |
|
||
| 3 | Table / column | `{facts['schema']}.{facts['table']}` / `{facts['column']}` |
|
||
| 3 | Vector dimension | **{facts['dimension']}** ({facts['dimension_note']}) |
|
||
| 3 | Distance operator | **{facts['operator']}** |
|
||
| 3 | Operator sanity check | {facts['self_similarity']} |
|
||
| 4 | Coverage (raw) | **{raw_pct}** ({cov['n_joined_to_assets']}/{cov['n_assets_total']} assets) |
|
||
| 4 | Coverage (image-only) | **{img_pct}** ({cov['n_embedded_image']}/{cov['n_assets_image']} IMAGE assets) |
|
||
|
||
- **Asset-key join:** {facts['asset_key_note']}
|
||
- **Orphan embeddings** (no matching asset): {cov['n_orphan']}
|
||
- **SQLite cross-check (optional):** {facts['sqlite']}
|
||
|
||
## The join M1.5 relies on
|
||
|
||
`{facts['table']}.{facts['asset_key']}` → `{facts['asset_table']}.{facts['assets_pk']}` →
|
||
SQLite `assets.immich_id` (`shared/photoflow/core`). The Immich asset UUID is the
|
||
same key our store uses, so embeddings index straight onto tracked photos.
|
||
|
||
> ⚠️ Immich's asset table is **`{facts['asset_table']}`** in the observed version
|
||
> (it was `assets` in older versions). The probe discovers this from the FK; M1.5's
|
||
> reader must not hardcode the name.
|
||
|
||
## Environment observed
|
||
|
||
- **Postgres:** {meta['postgres_version']}
|
||
- **pgvector extension:** {meta['pgvector_extension']}
|
||
- **Schema age proxy (latest migration):** {meta['latest_migration']}
|
||
- **Immich server version:** _record from the Immich UI / `GET /api/server/version`_
|
||
— not reliably available in the DB.
|
||
- **CLIP model:** _record the model from Immich's Machine-Learning settings_ — the
|
||
user is re-running CLIP with a stronger model, so dimension + coverage above are a
|
||
snapshot of whichever model was live at probe time.
|
||
|
||
## Read-only access (recommended hardening for M1.5)
|
||
|
||
The probe enforces read-only at the session layer
|
||
(`SET default_transaction_read_only = on` + psycopg `read_only`), which neutralises
|
||
write capability even with Immich's write/DDL-capable `postgres` user. For M1.5,
|
||
provision a dedicated least-privilege role instead:
|
||
|
||
```sql
|
||
CREATE ROLE photoflow_ro LOGIN PASSWORD '...';
|
||
GRANT CONNECT ON DATABASE immich TO photoflow_ro;
|
||
GRANT USAGE ON SCHEMA {facts['schema']} TO photoflow_ro;
|
||
GRANT SELECT ON {facts['schema']}.{facts['table']}, {facts['schema']}.{facts['asset_table']} TO photoflow_ro;
|
||
```
|
||
"""
|
||
)
|
||
|
||
|
||
def print_report(facts: dict) -> None:
|
||
cov = facts["coverage"]
|
||
print("\n=== pgvector spike — report ===")
|
||
print(f" table/column : {facts['schema']}.{facts['table']}.{facts['column']}")
|
||
print(f" dimension : {facts['dimension']} ({facts['dimension_note']})")
|
||
print(f" distance operator: {facts['operator']}")
|
||
print(f" operator sanity : {facts['self_similarity']}")
|
||
print(f" asset key : {facts['asset_key_note']}")
|
||
print(f" join : {facts['join_summary']} (orphans: {cov['n_orphan']})")
|
||
print(f" coverage (raw) : {_pct(cov['raw_ratio'])} ({cov['n_joined_to_assets']}/{cov['n_assets_total']})")
|
||
print(f" coverage (image) : {_pct(cov['image_ratio'])} ({cov['n_embedded_image']}/{cov['n_assets_image']})")
|
||
print(f" sqlite x-check : {facts['sqlite']}")
|
||
print(f" postgres : {facts['meta']['postgres_version']}")
|
||
print(f" pgvector ext : {facts['meta']['pgvector_extension']}")
|
||
print("===============================\n")
|
||
|
||
|
||
# ---------------------------------------------------------------------------- main
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="Read-only pgvector embedding feasibility probe.")
|
||
ap.add_argument("--sqlite", default="data/trip-cluster.db",
|
||
help="SQLite store path for the optional cross-check (default: data/trip-cluster.db)")
|
||
ap.add_argument("--findings", default=DEFAULT_FINDINGS, help="findings doc path to write")
|
||
ap.add_argument("--no-write", action="store_true", help="print the report but do not write the findings doc")
|
||
args = ap.parse_args()
|
||
|
||
dsn = find_dsn()
|
||
if not dsn:
|
||
print("ERROR: IMMICH_DB_URL not set (env or a .env up-tree). See .env.example.", file=sys.stderr)
|
||
return 2
|
||
|
||
try:
|
||
conn = connect_readonly(dsn)
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f"ERROR: could not connect read-only: {exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
with conn:
|
||
vec_cols = discover_vector_columns(conn)
|
||
if not vec_cols:
|
||
print("FAIL (Req 1): no pgvector `vector` column found in this database.", file=sys.stderr)
|
||
print(" Immich may not have run CLIP yet, or the schema changed beyond the probe's discovery.",
|
||
file=sys.stderr)
|
||
return 1
|
||
schema, table, column = vec_cols[0]
|
||
if len(vec_cols) > 1:
|
||
others = ", ".join(f"{s}.{t}.{c}" for s, t, c in vec_cols[1:])
|
||
print(f"NOTE: multiple vector columns found; probing {schema}.{table}.{column}. Others: {others}")
|
||
|
||
dim, dim_note = vector_dimension(conn, schema, table, column)
|
||
op, op_desc = distance_operator(conn, schema, table, column)
|
||
asset_key, asset_table, assets_pk, asset_key_note = detect_asset_key(conn, schema, table)
|
||
if not asset_key:
|
||
print(f"FAIL (Req 2): {asset_key_note}", file=sys.stderr)
|
||
return 1
|
||
|
||
self_sim = self_similarity_ok(conn, schema, table, column, asset_key, op)
|
||
cov = join_and_coverage(conn, schema, table, asset_key, asset_table, assets_pk)
|
||
sqlite_note = sqlite_crosscheck(conn, schema, table, asset_key, Path(args.sqlite))
|
||
meta = environment_meta(conn)
|
||
|
||
join_summary = (
|
||
f"{cov['n_joined_to_assets']}/{cov['n_embeddings_distinct']} embeddings join to "
|
||
f"{asset_table} via {asset_key}"
|
||
)
|
||
facts = {
|
||
"generated": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
|
||
"schema": schema, "table": table, "column": column,
|
||
"readable": f"yes — read {cov['n_embeddings_distinct']} distinct embedding(s)",
|
||
"dimension": dim if dim is not None else "unknown",
|
||
"dimension_note": dim_note,
|
||
"operator": op_desc,
|
||
"self_similarity": self_sim,
|
||
"asset_key": asset_key, "asset_table": asset_table, "assets_pk": assets_pk,
|
||
"asset_key_note": asset_key_note,
|
||
"join_summary": join_summary,
|
||
"coverage": cov, "sqlite": sqlite_note, "meta": meta,
|
||
}
|
||
|
||
print_report(facts)
|
||
|
||
join_clean = cov["n_orphan"] == 0 and cov["n_joined_to_assets"] > 0
|
||
if not args.no_write:
|
||
out = Path(args.findings)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
out.write_text(render_findings(facts))
|
||
print(f"Wrote findings -> {out}")
|
||
|
||
if cov["n_embeddings_distinct"] == 0:
|
||
print("WARN: zero embeddings — readable+well-shaped but coverage 0 (CLIP re-run unfinished?).")
|
||
if not join_clean:
|
||
print(f"WARN (Req 2): {cov['n_orphan']} orphan embedding(s) did not join to assets.")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|