Private PostgreSQL procedural dedup from first read through result snapshot.
This is deliberately separate from the public audited gate. Both graph table
locks span planning, application, and the final read on one bound connection.
run
async
run(policy: SemanticMerge, scope: TenantScope, backend: PostgresGraphBackend) -> tuple[int, list[MemoryNode]]
Return detached exact-owner survivors from one owned transaction.
Source code in src/symfonic/memory/backends/postgres_bulk_full_pass.py
| async def run(
policy: SemanticMerge, scope: TenantScope, backend: PostgresGraphBackend,
) -> tuple[int, list[MemoryNode]]:
"""Return detached exact-owner survivors from one owned transaction."""
from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
from symfonic.memory.janitor_bulk_plan import plan_bulk_dedup
if type(backend) is not PostgresGraphBackend:
raise TypeError("Private full pass requires exact PostgresGraphBackend")
if type(scope) is not TenantScope:
raise TypeError("Private full pass requires exact TenantScope")
if bound_connection(backend.pool) is not None:
raise RuntimeError("Private PostgreSQL full pass requires its own transaction")
async with backend.pool.transaction():
conn = await backend.pool.acquire()
await lock_graph_tables(conn)
rows = [
row for row in await backend.query_nodes(
scope, {"layer": MemoryLayer.PROCEDURAL}, None,
) if not is_retracted(row.properties)
]
plan = await plan_bulk_dedup(policy, scope, rows)
count = await apply_locked(conn, scope, plan)
rows = [
row for row in await backend.query_nodes(
scope, {"layer": MemoryLayer.PROCEDURAL}, None,
) if not is_retracted(row.properties)
]
remaining = [
row.model_copy(deep=True) for row in rows
if policy._is_exact_owner(scope, row)
]
return count, remaining
|