Skip to content

symfonic.memory.backends.postgres_claimed_pending

postgres_claimed_pending

Claim-qualified native PG pending rows for the private v11 graph-only path.

page_sql 

page_sql() -> str

One page of owned claimed rows for $6 tokens; params as read.

The page is driven by the token primary key and reaches each node by its (id, tenant_id) primary key, so it costs the page, never the pending subtree. The owner range is checked on the claim, whose scope_path the join then requires the node to equal exactly.

Source code in src/symfonic/memory/backends/postgres_claimed_pending.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def page_sql() -> str:
    """One page of owned claimed rows for ``$6`` tokens; params as ``read``.

    The page is driven by the token primary key and reaches each node by its
    ``(id, tenant_id)`` primary key, so it costs the page, never the pending
    subtree. The owner range is checked on the claim, whose ``scope_path``
    the join then requires the node to equal exactly.
    """
    return (
        # OFFSET 0 keeps the owner range out of the claim lookup: pushed down,
        # a generic plan ANDs it with a scan of the owner's claim history.
        # The lateral node lookup fixes the join order: before statistics,
        # nodes drove the loop and re-read the whole token page per node.
        'SELECT n.* FROM (SELECT * FROM memory_pending_claims '
        'WHERE token=ANY($6::text[]) OFFSET 0) c CROSS JOIN LATERAL '
        '(SELECT * FROM memory_nodes WHERE id=c.record_id COLLATE "default" '
        'AND tenant_id=$1 OFFSET 0) n '
        'WHERE n.tenant_id COLLATE "C"=c.tenant_id '
        'AND n.id COLLATE "C"=c.record_id '
        'AND n.scope_path COLLATE "C"=c.scope_path '
        'AND n.properties->>$5=c.token '
        'AND c.tenant_id=$1 AND c.state=\'pending\' '
        'AND c.scope_path>=$2 AND c.scope_path<$3 '
        'AND ' + marked("n.properties") + ' AND '
        + pending_jsonb_condition("n.properties", 4) + ' '
        'ORDER BY n.tenant_id COLLATE "C", n.scope_path COLLATE "C" '
        f'LIMIT {NATIVE_SUBTREE_BUDGET + 1}'
    )

read async 

read(pool: Any, scope: Any, tokens: tuple[str, ...]) -> list[Any]

Refuse unclaimed rows, then select owned tokens in bounded token pages.

Each page reads at most one row per owned token, by index probes under any plan. The deadline, not a row count, bounds the whole selection, as it bounds the mutation after it.

Source code in src/symfonic/memory/backends/postgres_claimed_pending.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def read(pool: Any, scope: Any, tokens: tuple[str, ...]) -> list[Any]:
    """Refuse unclaimed rows, then select owned tokens in bounded token pages.

    Each page reads at most one row per owned token, by index probes under any
    plan. The deadline, not a row count, bounds the whole selection, as it
    bounds the mutation after it.
    """
    from symfonic.memory.backends.postgres_graph import row_to_node

    tokens = tuple(dict.fromkeys(tokens))
    rows: list[Any] = []
    conn = await pool.acquire()
    try:
        async with native_subtree_deadline(conn), index_probes(conn):
            await require_pending_index(conn)
            prefix = scope.identity.subtree_prefix
            params = (scope.tenant_id, prefix, prefix[:-1] + "0", PENDING_KEY,
                      PENDING_BATCH_KEY)
            unclaimed = await conn.fetch(unclaimed_sql(), *params)
            if len(unclaimed) > NATIVE_SUBTREE_BUDGET:
                raise MemoryContractError("native pending row has no durable claim")
            root = scope.identity.levels
            if any(_descends(row["scope_path"], root) for row in unclaimed):
                raise MemoryContractError("native pending row has no durable claim")
            for start in range(0, len(tokens), NATIVE_SUBTREE_BUDGET):
                page = await conn.fetch(
                    page_sql(), *params, list(tokens[start:start + NATIVE_SUBTREE_BUDGET]),
                )
                if len(page) > NATIVE_SUBTREE_BUDGET:
                    raise MemoryContractError("native subtree budget exceeded: pending rows")
                rows.extend(page)
    finally:
        await pool.release(conn)
    root = scope.identity.levels
    result = []
    for row in rows:
        try:
            levels = StorageScopeKey.parse(row["scope_path"]).levels
        except MemoryContractError:
            continue
        if len(levels) >= len(root) and levels[:len(root)] == root:
            node = row_to_node(row)
            if is_pending(node.properties):
                result.append(node)
    return result

unclaimed_sql 

unclaimed_sql() -> str

Owned pending rows without a pending claim; the limit is read at call time.

Source code in src/symfonic/memory/backends/postgres_claimed_pending.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def unclaimed_sql() -> str:
    """Owned pending rows without a pending claim; the limit is read at call time."""
    return (
        # An anti-join whose claim side is fenced to the token primary key:
        # unfenced, a plan without statistics probed the owner's claims by
        # scope once per node.
        'SELECT n.scope_path FROM memory_nodes n LEFT JOIN LATERAL '
        '(SELECT 1 AS owned FROM (SELECT * FROM memory_pending_claims '
        'WHERE token=n.properties->>$5 OFFSET 0) c '
        'WHERE c.tenant_id=n.tenant_id COLLATE "C" '
        'AND c.record_id=n.id COLLATE "C" '
        'AND c.scope_path=n.scope_path COLLATE "C" AND c.state=\'pending\') c ON true '
        'WHERE ' + _WHERE + ' AND c.owned IS NULL '
        f'LIMIT {NATIVE_SUBTREE_BUDGET + 1}'
    )