symfonic.memory.backends.postgres_native_subtree_query

postgres_native_subtree_query

Canonical native owner selection for PostgreSQL descendant node reads.

The candidate prefix narrows a read; the parsed kind/ID levels decide whether each stored key belongs to the subtree. Invalid, NULL and v0 keys are excluded. This read does not claim public native backend admission.

CanonicalPathPage dataclass

CanonicalPathPage(paths: list[str], cursor: str | None)

Validated paths and the last raw key consumed, or None at exhaustion.

canonical_node_path_page async

canonical_node_path_page(conn: Any, scope: NativeStorageScope, node_id: object = None, *, cursor: str | None = None, page_size: int = 256) -> CanonicalPathPage

Read bounded SQL batches until a validated page fills.

The cursor advances over invalid keys too. A SQL LIMIT is never treated as the caller's result limit: malformed candidates may precede valid owners. This private selector does not authorize a public PostgreSQL native route.

Source code in src/symfonic/memory/backends/postgres_native_subtree_query.py

async def canonical_node_path_page(
    conn: Any,
    scope: NativeStorageScope,
    node_id: object = None,
    *,
    cursor: str | None = None,
    page_size: int = 256,
) -> CanonicalPathPage:
    """Read bounded SQL batches until a validated page fills.

    The cursor advances over invalid keys too. A SQL LIMIT is never treated as
    the caller's result limit: malformed candidates may precede valid owners.
    This private selector does not authorize a public PostgreSQL native route.
    """
    if type(page_size) is not int or not 1 <= page_size <= _MAX_PAGE_SIZE:
        raise ValueError(f"page_size must be in 1..{_MAX_PAGE_SIZE}")
    await _require_c_range_index(conn)
    from symfonic.memory.backends.postgres_working_expiry_reads import read_visibility, visible
    expiry = visible(await read_visibility(conn, scope))
    prefix = scope.identity.subtree_prefix
    # Complete v1 keys end in '/'. In C byte order, '/' is immediately before
    # '0', so this half-open range includes the root and every raw extension.
    upper = prefix[:-1] + "0"
    if cursor is not None and not isinstance(cursor, str):
        raise TypeError("cursor must be a storage key string")
    root = scope.identity.levels
    if node_id is not None:
        # The (id, tenant_id) primary key probes at most one collated identity.
        # Keep byte-exact guards for ICU databases; owner-range seeks could
        # otherwise walk every row sharing one path to find this ID.
        sql = (
            'SELECT scope_path FROM memory_nodes WHERE id=$1 AND tenant_id=$2 '
            'AND id COLLATE "C"=$1 AND tenant_id COLLATE "C"=$2 '
            f'AND scope_path COLLATE "C">=$3 AND scope_path COLLATE "C"<$4 AND {expiry}'
        )
        params: list[object] = [str(node_id), scope.tenant_id, prefix, upper]
        if cursor is not None:
            params.append(cursor)
            sql += ' AND scope_path COLLATE "C">$5'
        rows = await conn.fetch(sql, *params)
        paths = []
        for row in rows:
            path = row["scope_path"]
            try:
                levels = StorageScopeKey.parse(path).levels
            except MemoryContractError:
                continue
            if len(levels) >= len(root) and levels[:len(root)] == root:
                paths.append(path)
        return CanonicalPathPage(paths, None)
    paths: list[str] = []
    for _ in range(_MAX_BATCHES_PER_PAGE):
        # Seek beyond each distinct owner. DISTINCT/LIMIT scans all physical
        # rows of a heavily shared owner before returning one owner path.
        where = (
            'tenant_id=$1 AND tenant_id COLLATE "C"=$1 '
            f'AND scope_path COLLATE "C">=$2 AND scope_path COLLATE "C"<$3 AND {expiry}'
        )
        params: list[object] = [scope.tenant_id, prefix, upper]
        if cursor is not None:
            params.append(cursor)
            where += ' AND scope_path COLLATE "C">$' + str(len(params))
        params.append(page_size)
        sql = (
            'WITH RECURSIVE owners(scope_path, n) AS ('
            '(SELECT scope_path COLLATE "C", 1 FROM memory_nodes WHERE '
            + where + ' ORDER BY scope_path COLLATE "C" LIMIT 1) '
            'UNION ALL SELECT next_path.scope_path, owners.n + 1 FROM owners '
            'CROSS JOIN LATERAL (SELECT scope_path COLLATE "C" AS scope_path '
            'FROM memory_nodes WHERE ' + where
            + ' AND scope_path COLLATE "C">owners.scope_path '
            'ORDER BY scope_path COLLATE "C" LIMIT 1) next_path '
            'WHERE owners.n < $' + str(len(params)) + ') '
            'SELECT scope_path FROM owners ORDER BY n'
        )
        rows = await conn.fetch(sql, *params)
        if not rows:
            return CanonicalPathPage(paths, None)
        for row in rows:
            path = row["scope_path"]
            cursor = path
            try:
                levels = StorageScopeKey.parse(path).levels
            except MemoryContractError:
                continue
            if len(levels) >= len(root) and levels[:len(root)] == root:
                paths.append(path)
            if len(paths) == page_size:
                return CanonicalPathPage(paths, cursor)
        # Reached the end only if the fetched batch was wholly consumed.
        if len(rows) < page_size:
            return CanonicalPathPage(paths, None)
    # A sparse run of invalid keys can yield a partial or empty page. The
    # cursor lets the caller resume without treating SQL LIMIT as final.
    return CanonicalPathPage(paths, cursor)

native_subtree_deadline async

native_subtree_deadline(conn: Any)

Bound both native read phases; restore settings even inside a transaction.

The server cancels expensive scans and lock waits. The enclosing deadline also bounds the combined phases. Neither establishes a physical scan cap.

Source code in src/symfonic/memory/backends/postgres_native_subtree_query.py

@asynccontextmanager
async def native_subtree_deadline(conn: Any):
    """Bound both native read phases; restore settings even inside a transaction.

    The server cancels expensive scans and lock waits. The enclosing deadline
    also bounds the combined phases. Neither establishes a physical scan cap.
    """
    try:
        async with asyncio.timeout(NATIVE_SUBTREE_TIMEOUT_MS / 1000):
            async with conn.transaction():
                previous = await conn.fetchval("SHOW statement_timeout")
                await conn.fetchval(
                    "SELECT set_config('statement_timeout', "
                    "(LEAST(NULLIF(setting::int, 0), $1))::text || 'ms', true) "
                    "FROM pg_settings WHERE name='statement_timeout'",
                    NATIVE_SUBTREE_TIMEOUT_MS,
                )
                yield
                await conn.fetchval(
                    "SELECT set_config('statement_timeout', $1, true)", previous,
                )
    except Exception as exc:
        if isinstance(exc, TimeoutError) or getattr(exc, "sqlstate", None) == "57014":
            raise MemoryContractError("native subtree execution time budget exceeded") from exc
        raise