Skip to content

symfonic.memory.backends.postgres_auxiliary_schema

postgres_auxiliary_schema

Opt-in native QUICK auxiliary owner/fence migration; no QUICK admission.

Apply during an operator-maintained writer/DDL barrier. Positional, malformed, foreign-root and oversized owner keys must be remediated before installation. Existing invocation journal stores are outside this migration.

install async 

install(conn: Any, schema: str) -> None

Explicit operator migration, in its own transaction; never called by runtime.

Drain old writers/erasers first and reinstall exact eraser grants after. Backfill may scan during migration; retirement queries use owner indexes. A failed classification rolls the entire migration back.

Source code in src/symfonic/memory/backends/postgres_auxiliary_schema.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def install(conn: Any, schema: str) -> None:
    """Explicit operator migration, in its own transaction; never called by runtime.

    Drain old writers/erasers first and reinstall exact eraser grants after.
    Backfill may scan during migration; retirement queries use owner indexes.
    A failed classification rolls the entire migration back.
    """
    s = quote(schema)
    async with conn.transaction():
        await conn.execute(f"LOCK TABLE {s}.memory_nodes IN EXCLUSIVE MODE")
        if await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {s}.memory_erase_jobs WHERE permanent)"
        ):
            raise ValueError("CORE-460: drain permanent jobs before auxiliary migration")
        await conn.execute(
            f"CREATE OR REPLACE FUNCTION {s}.{OWNER}(row_data jsonb, tab text) RETURNS text "
            "LANGUAGE plpgsql IMMUTABLE SET search_path=pg_catalog AS $owner$"
            + OWNER_BODY + "$owner$"
        )
        await conn.execute(
            f"CREATE OR REPLACE FUNCTION {s}.{FENCE}() RETURNS trigger LANGUAGE plpgsql "
            "SET search_path=pg_catalog AS $fence$" + fence_body(schema) + "$fence$"
        )
        for name in await present(conn, schema):
            table = f"{s}.{quote(name)}"
            await conn.execute(f"LOCK TABLE {table} IN SHARE ROW EXCLUSIVE MODE")
            await conn.execute(f"ALTER TABLE {table} ADD COLUMN erase_scope text COLLATE \"C\"")
            await conn.execute(
                f"UPDATE {table} t SET erase_scope={s}.{OWNER}(to_jsonb(t), $1)", name
            )
            await conn.execute(f"ALTER TABLE {table} ALTER COLUMN erase_scope SET NOT NULL")
            await conn.execute(f"CREATE INDEX idx_core460_{TABLES[name]}_owner_c ON {table} "
                               '(erase_scope COLLATE "C")')
            await conn.execute(f"CREATE TRIGGER {FENCE} BEFORE INSERT OR UPDATE ON {table} "
                               f"FOR EACH ROW EXECUTE FUNCTION {s}.{FENCE}()")
            await conn.execute(f"ALTER TABLE {table} ENABLE ALWAYS TRIGGER {FENCE}")