"""Per-field indexes. A datasource or form field marked "Indexed" (MasterFieldDef.indexed / FormFieldDef.indexed) gets its own Postgres index on its value: an expression index on the partition of master_records / form_submissions that holds the source, partial to that one source, so it holds only that source's rows and other sources' writes rarely meet it. The indexed expressions are exactly the ones app/crud/record_sql.py writes into queries — that's what lets the planner use them: * numeric field types: record_sql.numeric_key — equality, ranges, IN; * everything else: record_sql.text_key, the value's text in "C" order — equality, ranges, IN, date ranges on ISO date strings. Indexes are built with CREATE INDEX CONCURRENTLY, which never blocks writes to the table but can't run inside a transaction, and waits for every transaction already open — including the request that asked for it. So a save checks whether anything changes (a cheap catalog read, inside the request) and if so queues a job (app/core/jobs.py) holding the indexes the source wants; the job works out what to build and drop when it runs, on its own autocommit connection — so the latest save always wins. A build that fails (a lock, out of disk) leaves an invalid index behind; the job drops it and is retried. """ import hashlib import logging import uuid from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.core import jobs from app.core.database import engine from app.crud import record_sql from app.crud.record_sql import NUMERIC_FIELD_TYPES log = logging.getLogger(__name__) # kind -> (table, source column, index name tag) _TABLES = {"master": ("master_records", "master_id", "m"), "form": ("form_submissions", "form_id", "f")} # Values these hold are lists, documents or files — nothing an equality or # range lookup on the whole value would serve. UNINDEXABLE_TYPES = { "json", "jsonb", "dynamic_grid", "subform", "embedded_form", "multiselect", "treeselect", "tags", "file", "file_upload", "image_upload", "attachment", "multi_attachment", "signature", "camera", "address", "matrix", "ranking", "rich_text", "markdown", "html", "code", "button", } JOB_KIND = "field_indexes.sync" def _quote(value: str) -> str: return "'" + value.replace("'", "''") + "'" def _prefix(kind: str, source_id: uuid.UUID) -> str: return f"ix_ff_{_TABLES[kind][2]}_{uuid.UUID(str(source_id)).hex}_" def _expression(field: dict) -> tuple[str, str]: """(kind tag, index expression) — kept identical to record_sql's numeric_key / text_key so queries match the index.""" key = _quote(field["name"]) if field.get("type") in NUMERIC_FIELD_TYPES: return "n", ( f"CASE WHEN jsonb_typeof(field_values -> {key}) = 'number' THEN CAST(field_values -> {key} AS NUMERIC) " f"WHEN jsonb_typeof(field_values -> {key}) = 'boolean' " f"THEN CASE WHEN (field_values -> {key}) = 'true'::jsonb THEN 1 ELSE 0 END END" ) return "t", f'(field_values ->> {key}) COLLATE "C"' async def record_partition(db, kind: str, source_id: uuid.UUID) -> str: """The partition of its records table that holds a source's records.""" return ( await db.execute(text("SELECT record_partition(CAST(:t AS regclass), CAST(:s AS uuid))::text"), {"t": _TABLES[kind][0], "s": str(source_id)}) ).scalar_one() def wanted_indexes(kind: str, source_id: uuid.UUID, field_defs: list[dict], partition: str) -> dict[str, str]: """{index name: CREATE statement} for every indexed field. The name hashes the field's name and index kind, so retyping a field from text to number swaps its index rather than keeping a useless one.""" _, source_col, _ = _TABLES[kind] out: dict[str, str] = {} if any(field.get("searchable") for field in field_defs): name = _prefix(kind, source_id) + hashlib.sha1(b"search").hexdigest()[:10] out[name] = ( f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} ON {partition} USING gin ({record_sql.SEARCH_DOCUMENT} gin_trgm_ops) " f"WHERE {source_col} = '{uuid.UUID(str(source_id))}'::uuid" ) for field in field_defs: if not field.get("indexed") or field.get("type") in UNINDEXABLE_TYPES or not field.get("name"): continue tag, expr = _expression(field) digest = hashlib.sha1(f"{field['name']}|{tag}".encode()).hexdigest()[:10] name = _prefix(kind, source_id) + digest out[name] = ( f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} ON {partition} (({expr})) " f"WHERE {source_col} = '{uuid.UUID(str(source_id))}'::uuid" ) return out async def existing_indexes(db, kind: str, source_id: uuid.UUID) -> dict[str, bool]: """{index name: is valid} for this source's field indexes.""" rows = await db.execute( text( "SELECT c.relname, i.indisvalid FROM pg_class c JOIN pg_index i ON i.indexrelid = c.oid " "WHERE starts_with(c.relname, :prefix)" ), {"prefix": _prefix(kind, source_id)}, ) return {name: valid for name, valid in rows.all()} async def sync_field_indexes(db: AsyncSession, kind: str, source_id: uuid.UUID, field_defs: list[dict]) -> None: """Brings this source's field indexes in line with its fields — pass [] when the source is deleted. Queues the work, which runs once the calling request has committed (see the module docstring).""" wanted = wanted_indexes(kind, source_id, field_defs, await record_partition(db, kind, source_id)) drop, create = _diff(wanted, await existing_indexes(db, kind, source_id)) if not drop and not create: return tenant = (await db.execute(text("SELECT nullif(current_setting('app.current_tenant_id', true), '')"))).scalar() await jobs.enqueue( db, JOB_KIND, {"kind": kind, "source_id": str(source_id), "wanted": wanted}, tenant_id=uuid.UUID(tenant) if tenant else None, dedupe_key=f"{JOB_KIND}:{kind}:{source_id}", ) async def drop_tenant_indexes(db: AsyncSession, tenant_id: uuid.UUID) -> None: """Queues the removal of every field index of a tenant that's being deleted. Call it while the tenant's sources still exist. Left behind, an index slows every write to its records table for good. The jobs belong to no tenant, so they outlive this one.""" sources = await db.execute( text("SELECT 'master', id FROM masters WHERE tenant_id = :t UNION SELECT DISTINCT 'form', form_id FROM forms WHERE tenant_id = :t"), {"t": str(tenant_id)}, ) for kind, source_id in sources.all(): if await existing_indexes(db, kind, source_id): await jobs.enqueue(db, JOB_KIND, {"kind": kind, "source_id": str(source_id), "wanted": {}}, dedupe_key=f"{JOB_KIND}:{kind}:{source_id}") def _diff(wanted: dict[str, str], existing: dict[str, bool]) -> tuple[list[str], dict[str, str]]: drop = [name for name, valid in existing.items() if name not in wanted or not valid] create = {name: sql for name, sql in wanted.items() if not existing.get(name)} return drop, create @jobs.handler(JOB_KIND, max_attempts=5) async def _sync_job(job: jobs.Job) -> str: kind, source_id, wanted = job.payload["kind"], uuid.UUID(job.payload["source_id"]), job.payload["wanted"] async with engine.connect() as conn: conn = await conn.execution_options(isolation_level="AUTOCOMMIT") drop, create = _diff(wanted, await existing_indexes(conn, kind, source_id)) failed = [] for name in drop: try: await conn.exec_driver_sql(f"DROP INDEX CONCURRENTLY IF EXISTS {name}") except Exception as exc: # noqa: BLE001 - the rest still run; the job is retried below log.warning("Dropping field index %s failed: %s", name, exc) failed.append(name) for name, sql in create.items(): try: await conn.exec_driver_sql(sql) log.info("Built field index %s", name) except Exception as exc: # noqa: BLE001 - an invalid leftover is dropped on the retry log.warning("Building field index %s failed: %s", name, exc) failed.append(name) if failed: raise RuntimeError(f"{len(failed)} of {len(drop) + len(create)} index change(s) failed: {', '.join(failed)}") return f"{len(create)} built, {len(drop)} dropped" async def wait_for_pending() -> None: """Runs the queued index jobs now, in this process (tests, scripts).""" await jobs.run_due([JOB_KIND])