bounded_node_rows(conn: Any, scope: NativeStorageScope, paths: list[str], filters: dict[str, Any], limit: int) -> list[Any]
Return the global top rows across one bounded page of canonical owners.
Each owner contributes at most limit rows, with at most 512 owners
and a 512-row limit. This private seam accepts no filters; selective SQL
filters, Python property filters, and unlimited reads need separate
evidence before this bound applies.
Source code in src/symfonic/memory/backends/postgres_native_subtree_rows.py
| async def bounded_node_rows(
conn: Any,
scope: NativeStorageScope,
paths: list[str],
filters: dict[str, Any],
limit: int,
) -> list[Any]:
"""Return the global top rows across one bounded page of canonical owners.
Each owner contributes at most ``limit`` rows, with at most 512 owners
and a 512-row limit. This private seam accepts no filters; selective SQL
filters, Python property filters, and unlimited reads need separate
evidence before this bound applies.
"""
if type(limit) is not int or limit < 1:
raise ValueError("limit must be a positive integer")
if limit > _MAX_LIMIT:
raise ValueError("limit must be at most 512")
if len(paths) > _MAX_OWNERS:
raise ValueError("too many owner paths")
if filters:
raise ValueError("filtered reads are not bounded by this selector")
root = scope.identity.levels
for path in paths:
if not isinstance(path, str):
raise ValueError("owner paths must be canonical strings")
try:
levels = StorageScopeKey.parse(path).levels
except MemoryContractError as exc:
raise ValueError("owner paths must be canonical strings") from exc
if len(levels) < len(root) or levels[:len(root)] != root:
raise ValueError("owner path is outside the requested subtree")
if not paths:
return []
await _require_order_index(conn)
from symfonic.memory.backends.postgres_working_expiry_reads import read_visibility, visible
expiry = visible(await read_visibility(conn, scope), 'n')
params: list[Any] = [scope.tenant_id, list(dict.fromkeys(paths)), limit]
sql = (
"SELECT picked.* FROM unnest($2::text[]) AS owner(path) "
"CROSS JOIN LATERAL (SELECT n.* FROM memory_nodes AS n WHERE "
'n.tenant_id=$1 AND n.tenant_id COLLATE "C"=$1 '
'AND n.scope_path COLLATE "C"=owner.path COLLATE "C" '
f'AND {expiry} '
"ORDER BY n.created_at DESC, n.id LIMIT $3) AS picked "
"ORDER BY picked.created_at DESC, picked.id LIMIT $3"
)
return await conn.fetch(sql, *params)
|