Skip to content

symfonic.memory.backends.postgres_candidates

postgres_candidates

Native eligible top-k selection; no unbounded row materialization in Python.

statement 

statement(scope, request, expiry='TRUE', indexed=False, jobs=None)

Return parameterized SQL; all user cue/scope/session values are bound.

indexed reads native rows through the eligible-candidate index seeks. jobs names the erase job table: owners an erasing job hides leave the owner array before any row is read (CORE-199).

Source code in src/symfonic/memory/backends/postgres_candidates.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def statement(scope, request, expiry='TRUE', indexed=False, jobs=None):
    """Return parameterized SQL; all user cue/scope/session values are bound.

    ``indexed`` reads native rows through the eligible-candidate index seeks.
    ``jobs`` names the erase job table: owners an erasing job hides leave the
    owner array before any row is read (CORE-199).
    """
    params = [scope.tenant_id, list(request.layers), request.session_id,
              sorted(set(request.cue.lower().split())), request.limit, request.profile_slots]
    prefix, values = sql_prefix_condition(scope, "scope_path", 7)
    native = _is_native_scope(scope)
    if native:
        params.extend(values)
        if request.allowed_scope_paths is not None:
            index = len(params) + 1
            params.append(list(request.allowed_scope_paths))
            prefix += f' AND scope_path COLLATE "C"=ANY(${index}::text[])'
        if request.exact_scope:
            index = len(params) + 1
            params.append(materialise_scope_path(scope))
            prefix += f' AND scope_path COLLATE "C"=${index}'
        owner = "scope_path"
    elif request.allowed_scope_paths is not None:
        params.extend([list(request.allowed_scope_paths), values[1]])
        prefix = 'COALESCE(scope_path,$8) COLLATE "C"=ANY($7::text[])'
    elif request.exact_scope:
        params.extend([materialise_scope_path(scope), values[1]])
        prefix = 'COALESCE(scope_path,$8) COLLATE "C"=$7'
    else:
        params.extend(values)
    if not native:
        owner = "COALESCE(scope_path,$8)"
    params.append(scope.ancestor_prefix_paths())
    ancestors_index = len(params)
    source, table, final = "", "memory_nodes", """
SELECT * FROM ranked
ORDER BY CASE WHEN profile AND ordinal <= $6 THEN 0 ELSE 1 END,cue_score DESC,id COLLATE "C"
LIMIT $5
"""
    if native:
        params.append(budget.visible_paths(scope, request))
        live = None
        if jobs is not None:
            from symfonic.memory.backends.postgres_erase_jobs import live_paths

            live = live_paths(jobs, "$1", f"${len(params)}")
        source, final = budget.bounded_source(len(params), live), budget.BOUNDED_SELECT
        table = "visible WHERE NOT (SELECT exceeded FROM budget) AND"
        if indexed:
            paths = len(params)
            params.append(sorted(set(request.layers)))
            params.append(budget.seek_cap(scope, request, params[3]))
            source = budget.eligible_source(
                eligible_index.seek_source(1, paths, paths + 1, paths + 2, expiry,
                                           budget.CANDIDATE_SCAN_BUDGET + 1, live))
    else:
        table += " WHERE"
    flags = rows.lifecycle_flags()
    event = rows.identity("past_event")
    profile = rows.profile()
    session = rows.session()
    sql = f"""
WITH {source}
eligible AS {"MATERIALIZED " if native else ""}(
 SELECT *, {profile} AS profile,
 (SELECT count(*) FROM unnest($4::text[]) token
  WHERE token=ANY(regexp_split_to_array(
    lower(label || ' ' || COALESCE(properties->>'content','')), '[[:space:]]+'))) AS hits
 FROM {table} tenant_id COLLATE "C"=$1 AND layer=ANY($2::text[]) AND {prefix}
 AND {expiry}
 AND {' AND '.join(flags)}
 AND (layer NOT IN ('working','episodic')
      OR (layer='episodic' AND {event})
      OR {session}=to_jsonb($3::text))
), scored AS (
 SELECT *, {rows.RANK} *
 (0.25 + 0.75 * CASE WHEN cardinality($4::text[])=0 THEN 1.0
                   ELSE hits::float/cardinality($4::text[]) END) *
 power(0.5,COALESCE(cardinality(${ancestors_index}::text[])-array_position(
   ${ancestors_index}::text[],{owner}),1))
 AS cue_score FROM eligible
), ranked AS (
 SELECT *, row_number() OVER (
 PARTITION BY profile ORDER BY cue_score DESC,id COLLATE "C") AS ordinal,
 count(*) OVER () AS eligible_count FROM scored
){final}"""
    return sql, params