SQL for the CORE-171 PostgreSQL positional migration; see the rows module.
Owners compare byte-exactly (COLLATE "C"). A candidate is a row whose
owner is NULL, does not start with the settled symfonic.memory. header
shared by native v1 and quarantine keys, or holds the v0 delimiter (chr(31))
no settled key contains: a v0 kind may spell the header. Every statement is
tenant-bound.
current_rows
async
current_rows(conn: Any, kind: str, tenant_id: str, ids: list[str]) -> list[str]
Which of ids still exist, whatever their owner now is.
Source code in src/symfonic/memory/backends/postgres_positional_sql.py
| async def current_rows(conn: Any, kind: str, tenant_id: str, ids: list[str]) -> list[str]:
"""Which of ``ids`` still exist, whatever their owner now is."""
rows = await conn.fetch(
f"SELECT {_ID[kind]} AS id FROM {_TABLE[kind]} "
f"WHERE tenant_id = $1 AND {_ID[kind]} = ANY($2::text[])", tenant_id, ids)
return [row["id"] for row in rows]
|
refuse_live_state
async
refuse_live_state(conn: Any, tenant_id: str) -> None
Refuse while an erase or a consolidation of this tenant is in flight.
Runs after the lease key and the memory_nodes lock, so a job the sweep
committed is visible here (the CORE-199 writer ordering), and no lease of
the tenant can be acquired or committed against until this transaction ends.
Source code in src/symfonic/memory/backends/postgres_positional_sql.py
| async def refuse_live_state(conn: Any, tenant_id: str) -> None:
"""Refuse while an erase or a consolidation of this tenant is in flight.
Runs after the lease key and the ``memory_nodes`` lock, so a job the sweep
committed is visible here (the CORE-199 writer ordering), and no lease of
the tenant can be acquired or committed against until this transaction ends.
"""
import json
from symfonic.memory.backends.postgres_erase_jobs import jobs_table
if await conn.fetchval("SHOW transaction_isolation") != "read committed":
raise MemoryContractError("positional migration requires READ COMMITTED isolation")
jobs = await jobs_table(conn)
if jobs is not None and await conn.fetchval(
f"SELECT EXISTS (SELECT 1 FROM {jobs} WHERE tenant_id = $1 AND state = 'erasing')",
tenant_id,
):
raise MemoryContractError(
"the tenant has an erase in progress; finish it before migrating")
if await conn.fetchval("SELECT to_regclass('memory_consolidation_leases') IS NOT NULL"):
leased = await conn.fetch(_TENANT_LEASES, tenant_id, json.dumps(tenant_id),
json.dumps(tenant_id, ensure_ascii=False))
if len(leased) > _LEASE_LIMIT or any(
_lease_of(row["scope_path"], tenant_id) for row in leased
):
raise MemoryContractError(LEASE_REFUSAL)
|
refuse_positional_vectors
async
refuse_positional_vectors(conn: Any, tenant_id: str) -> None
Planning refuses a tenant whose v0 rows have vector copies (VECTOR_GATE).
Source code in src/symfonic/memory/backends/postgres_positional_sql.py
| async def refuse_positional_vectors(conn: Any, tenant_id: str) -> None:
"""Planning refuses a tenant whose v0 rows have vector copies (VECTOR_GATE)."""
if await conn.fetchval("SELECT to_regclass('memory_vectors') IS NOT NULL") and (
await conn.fetchval(f"SELECT EXISTS (SELECT 1 FROM memory_vectors WHERE {_CANDIDATE})",
tenant_id)):
raise MemoryContractError(VECTOR_GATE)
|