Skip to content

symfonic.memory.backends.postgres_native_edge_reads

postgres_native_edge_reads

Native edge admission before pagination, plus unchanged legacy read SQL.

This direct-backend prerequisite does not declare native persistence support. Exact canonical ancestor keys exclude NULL, v0 and malformed owners. Both live endpoints must be visible in the same statement snapshot as the edge.

neighbors_statement 

neighbors_statement(scope: Any, node_id: Any, relationship: str | None, expiry=None) -> tuple[str, list[Any]]

Use EXISTS to deduplicate neighbors; admit both anchor and destination.

Source code in src/symfonic/memory/backends/postgres_native_edge_reads.py 
 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
def neighbors_statement(scope: Any, node_id: Any,
                        relationship: str | None, expiry=None) -> tuple[str, list[Any]]:
    """Use EXISTS to deduplicate neighbors; admit both anchor and destination."""
    if not _is_native_scope(scope):
        return _legacy_neighbors(scope, node_id, relationship, expiry)
    params: list[Any] = [scope.tenant_id, str(node_id)]
    relation = ""
    if relationship:
        params.append(relationship)
        relation = (f" AND e.relationship=${len(params)}"
                    f' AND e.relationship COLLATE "C"=${len(params)}')
    admission = _admission(scope, params, expiry)
    admission += ' AND ' + visible(expiry, 'e', 'source') + ' AND ' + visible(expiry, 'e', 'target')
    sql = (
        "SELECT n.* FROM memory_nodes n "
        'WHERE n.tenant_id=$1 AND n.tenant_id COLLATE "C"=$1 '
        "AND EXISTS (SELECT 1 FROM memory_edges e "
        'WHERE e.tenant_id=$1 AND e.tenant_id COLLATE "C"=$1 AND '
        '((e.source=$2 AND e.target=n.id AND '
        'e.source COLLATE "C"=$2 AND e.target COLLATE "C"=n.id COLLATE "C") OR '
        '(e.target=$2 AND e.source=n.id AND '
        'e.target COLLATE "C"=$2 AND e.source COLLATE "C"=n.id COLLATE "C")) '
        f"AND {admission}{relation})"
    )
    return sql, params

query_statement 

query_statement(scope: Any, filters: dict[str, Any] | None, limit: int | None, offset: int, expiry=None) -> tuple[str, list[Any]] | None

Bind all caller values; owner/lifecycle predicates precede LIMIT/OFFSET.

Source code in src/symfonic/memory/backends/postgres_native_edge_reads.py 
41
42
43
44
45
46
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
75
76
77
78
79
def query_statement(scope: Any, filters: dict[str, Any] | None,
                    limit: int | None, offset: int, expiry=None) -> tuple[str, list[Any]] | None:
    """Bind all caller values; owner/lifecycle predicates precede LIMIT/OFFSET."""
    if not _is_native_scope(scope):
        return _legacy_query(scope, filters, limit, offset, expiry)
    params: list[Any] = [scope.tenant_id]
    conditions = ['e.tenant_id=$1', 'e.tenant_id COLLATE "C"=$1']
    for key, column in (("relationship", "relationship"), ("incident_node_id", None)):
        value = (filters or {}).get(key)
        if value is None:
            continue
        params.append(value if column else str(value))
        placeholder = f"${len(params)}"
        if column:
            conditions.extend((f"e.{column}={placeholder}",
                               f'e.{column} COLLATE "C"={placeholder}'))
        else:
            conditions.append(
                f'((e.source={placeholder} AND e.source COLLATE "C"={placeholder}) '
                f'OR (e.target={placeholder} AND e.target COLLATE "C"={placeholder}))'
            )
    induced = (filters or {}).get("induced_node_ids")
    if induced is not None:
        params.append(list(map(str, induced)))
        placeholder = f"${len(params)}::text[]"
        for column in ("source", "target"):
            conditions.extend((f"e.{column}=ANY({placeholder})",
                               f'e.{column} COLLATE "C"=ANY({placeholder})'))
    conditions.append(_admission(scope, params, expiry))
    conditions.extend((visible(expiry, 'e', 'source'), visible(expiry, 'e', 'target')))
    pagination = ""
    if limit is not None:
        params.append(limit)
        pagination += f" LIMIT ${len(params)}"
    if limit is not None or offset:
        params.append(offset)
        pagination += f" OFFSET ${len(params)}"
    return (f"SELECT e.* FROM memory_edges e WHERE {' AND '.join(conditions)}"
            f" ORDER BY e.created_at DESC{pagination}", params)