Support per-entity Top-k (Shape 3 & 4) #386

Background

Implements Shape 3 and Shape 4 of the universal Top-K query engine defined in #369:

• Shape 3 topk(Target | Filter) — top-k targets filtered by a specific entity, all time
• Shape 4 topk(Target | Filter, Window) — top-k targets filtered by entity within a sliding time window

The same machinery covers global, non per-entity rankings by declaring __GLOBAL__ as the entity value, so this is no longer strictly per-entity.

Metadata

Top-k is declared under aggregations on an edge group definition, and a group can declare more than one.

"aggregations": {
  "topk": [
    {
      "topk": "<topk_name>",
      "entity": "<entity_field>",
      "dimension": "<dimension_field>",
      "rank": "<database>.<table>",
      "ranges": "<range_expression>",
      "refreshAfterMillis": 3600000,
      "additionalProperties": ["<property>"]
    }
  ]
}

• entity — the field the ranking is scoped to. __GLOBAL__ selects a single global ranking instead.
• dimension — the field being ranked. Every other non-bucket field becomes part of the dimension key, via Group.dimensionFields.
• rank — fully qualified name of the rank table, split by parseFqn into database and table.
• ranges — range expression applied when scanning the source group. It is interpolated against the event before the scan, so it can reference event fields. Empty means no range, which is what covers Shape 3.
• refreshAfterMillis — how long until a sliding-window ranking is queued for refresh. -1 disables refresh, and refresh is skipped entirely for non-sliding declarations.
• additionalProperties — properties carried onto the rank row alongside the metric.

This replaces the earlier topk: { topk, window: { from, to } } shape. There is no window object; ranges plus refreshAfterMillis cover it.

System Tables

Rank rows live in the topk database. The keys and property names are fixed in AggregationConstants.Topk.

Rank table

• Row key: rankSource(database, table, topk, entity, dimensionValues), joined as database|table|topk|entity|dim1|dim2|...
• Properties: metric for the aggregated value, additionalProperties for the carried properties as one JSON string
• Index: metric_desc, so rows are already in top-K read order

Refresh queue table (refresh)

• Message key: refreshKey(database, table, topk, entity, topkDimensionValue, dimensionValues), joined as database|table|topk|entity|topkDimensionValue|dim1|...
• The queue derives its partition from this key, so it takes the raw composite rather than a pre-hashed value

This replaces the _score and _expire tables described earlier. There is no _expire table, no expiredAt index, and no sweeper scanning for expired entries. Refresh is queue-driven and only applies to sliding-window declarations.

Aggregator

1. Scan the source group table with ranges interpolated against the event
2. Sum the metric per (entity, dimension key) and write to the rank table
3. If the declaration is a sliding window and refreshAfterMillis > 0, enqueue a refresh message

Metadata API

GET /aggregations/v1/metadata?type={type} lists everything that declares an aggregation. type is optional and filters by AggregationType.

{
  "items": [
    { "type": "TOPK", "database": "...", "table": "...", "dedupeFields": [] }
  ]
}

Query API

POST /aggregations/v1/databases/{database}/tables/{table}/topks/{topk}

{
  "entity": "<entity_value>",
  "dimensionValues": { "<field>": "<value>" },
  "limit": 100,
  "offset": null
}

{
  "topks": [
    { "value": "<dimension_value>", "metric": 1234, "properties": { "<key>": "<value>" } }
  ],
  "count": 1
}

This is a POST rather than a GET. A ranking is picked by naming dimension values, and a query string cannot carry those names onto an immutable request. entity, limit and offset are all optional; limit falls back to ScanFilter.defaultLimit. Reads go through the metric_desc index for O(K) read latency.

This replaces the earlier GET /databases/{db}/tables/{table}/edges/topk/{topk}?entity={entity}&limit={k}, which had no way to name dimension values and no cursor.

Task

• Declare top-k inside edge group metadata, as `Group.aggregations
• List what declares an aggregation: GET /aggregations/v1/metadata
  • Declare a top-K aggregation on a group and list what declares one (Steps 1 and 2 of #386) #493
• Create the rank table in the topk database and the refresh queue table
  • Run a top-K aggregation and write the ranked rows (Step 3 of #386) #494
• Implement the aggregator: scan the source group, sum the metric per (entity, dimension key), write rank rows
  • Recompute a ranking from a refresh message (Step 4 of #386) #496
• Enqueue refresh for sliding-window declarations and consume the queue
  • Read a materialized top-K by naming its entity and dimension values (Step 5 of #386) #497
• Expose the query API
  • Read a materialized top-K as a step in a multi-hop query (Step 6 of #386) #498

Done When

• Shape 3 and Shape 4 queries return correct results with O(K) read latency
• __GLOBAL__ rankings work through the same path as per-entity ones
• The aggregator and the refresh consumer run without blocking the primary write path
• Unit/integration tests pass

Notes

• Depends on Support universal Top-K query engine #369: builds on the universal Top-K interface defined there; Shape 3 and 4 are the initial target implementations.
• Decoupled writes: the aggregator and the refresh consumer run as background jobs and must not block or bottleneck the primary write path.
• Pre-aggregated state: the rank table holds the pre-summed metric per (entity, dimension key) to guarantee O(K) read latency at query time.
• Empty ranges covers Shape 3 as a fallback of Shape 4, the same way the optional window did before