Skip to content

Run a top-K aggregation and write the ranked rows (Step 3 of #386) - #494

Merged
zipdoki merged 11 commits into
mainfrom
feat/topk-aggregate
Aug 27, 2026
Merged

zipdoki merged 11 commits into
mainfrom
feat/topk-aggregate

Conversation

@zipdoki

@zipdoki zipdoki commented Aug 19, 2026

Copy link
Copy Markdown
Contributor

Summary

A group can declare a top-K, but nothing computes one. The declaration sits in metadata and no ranked row is ever written. This runs the aggregation.

TopkAggregationHandler scans the source group with the declaration's ranges interpolated against the event, sums metric per (entity, dimension key), and writes rank rows keyed by rankSource(...) under the metric_desc index. When a declaration is a sliding window and refreshAfterMillis > 0, it also schedules a refresh
for the instant the bucket leaves the window. Consuming that refresh message is the next PR in the stack.

POST /aggregations/v1/aggregate drives it, taking a batch of edges:

{
  "items": [
    {
      "database": "graph",
      "table": "likes",
      "edge": {
        "source": "...",
        "target": "...",
        "...": "..."
      }
    }
  ]
}
{
  "items": [
    {
      "database": "graph",
      "table": "likes",
      "source": "...",
      "target": "...",
      "status": "OK",
      "error": null
    }
  ]
}

Status and error are reported per item, so one bad edge does not sink the rest of the batch.

Two bucket changes ride along because the window math depends on them. Bucket now has a test pinning how a date bucket rounds both a value and a range bound, which the refresh schedule reads directly, and its format precision is read in one place instead of being re-derived at each call site. Separately, rank keys escape | so
two rankings can never collide on one row.

Stacked on #493 — that needs to land first.

Part of #386.

Test plan

  • ./gradlew :core:test --tests '*BucketTest*' — date bucket rounding for values and for range bounds
  • ./gradlew :core:test --tests '*AggregationConstantsTest*' — rank key composition and | escaping
  • ./gradlew :engine:test --tests '*RankingInputsTest*' — ranges interpolation, entity and dimension resolution
  • ./gradlew :engine:test --tests '*TopkAggregationHandlerTest*' — aggregation over the source group, rank row writes, refresh scheduling
  • ./gradlew :server:test --tests '*MetadataAggControllerE2ETest*' — POST /aggregations/v1/aggregate over HTTP
  • ./gradlew spotlessKotlinCheck build — formatting and full build

AI Assistance

  • This PR was written largely with AI assistance.
    • Tool / model: Claude Code (Opus 5)

@zipdoki zipdoki self-assigned this Aug 19, 2026
@zipdoki
zipdoki requested a review from em3s as a code owner August 19, 2026 04:31
@dosubot dosubot Bot added size:XL This PR changes 500-999 lines, ignoring generated files. enhancement New feature or request labels Aug 19, 2026
@zipdoki
zipdoki force-pushed the feat/topk-aggregate branch 2 times, most recently from 37e1c4b to 7af6396 Compare August 20, 2026 02:14
@zipdoki zipdoki mentioned this pull request Aug 20, 2026
6 tasks done
Base automatically changed from feat/topk-metadata to main August 27, 2026 08:08
zipdoki and others added 7 commits August 27, 2026 17:08
…t's name

A declaration may name a date bucket for the value it yields rather than for the field it reads, and a
`ranges` bound refers to it by that name. `refreshAt` looked the event's value up under the bucket's name,
found nothing whenever the two names differ, and fell back to counting from now — so an order that left
the window a year ago was scheduled to expire a year from ingest instead.

Nothing reported it. The aggregation answered SUCCESS and the message carried the wrong due time.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`refreshQueue` landed on the declaration in #493 with nothing reading it. The enqueue still went to the
constant, so naming a queue had no effect.

Existing tests here name no queue and assert the shared one, which pins the default; the new test names one
and also asserts nothing reaches the shared queue.
@zipdoki
zipdoki force-pushed the feat/topk-aggregate branch from 16c8202 to 51a1402 Compare August 27, 2026 08:08
`AggregationResult` carries no field naming the item it came from, so a caller pairs a result back to its request by position. The dispatcher merged handler output as it completed, which left the same batch coming back ordered by whichever writes finished first — a three-item batch returns fully reversed when the first item is the slowest.

`Graph.mutate` and `AbstractLabel.mutate` already made this choice on the mutation path. Concurrency is unchanged: `flatMapSequential` and `mergeSequential` subscribe eagerly under the same default `maxConcurrency`, so storage sees the same request pattern.

`AggregationService.aggregate` had no test of its own. The new ones pin fan-out, the empty batch, request ordering, and the fact that two handlers claiming one type collapse to the last — that last one is current behavior, pinned rather than fixed.
The branch was rebased onto the squash-merged #493, where the constant is `REFRESH_QUEUE`. The rebase carried that name into `AggregationConstants`, `Group` and `EdgeSchemaSerializationTest` but left `TopkAggregationHandlerTest` and `MetadataAggControllerE2ETest` on `REFRESH_TABLE`, so the branch did not compile. Point them at the merged name.
`AggregationFunctions.kt` and `TopkAggregationHandler` both declared `parseFqn`, added in the same commit. The handler's own private copy shadows the top-level one at the only call site, so nothing reached `AggregationFunctions` and the two spellings of its failure message could drift apart unnoticed. Extract a shared one again when a second handler needs it.
`processTopk` skips the refresh on three conditions and only two were covered. The third — a top-K that sets `refreshAfterMillis` but bounds its range with fixed dates — had no test, so nothing said what a non-sliding window is supposed to do. Drop the guard and this new test is the only one that fails, which is what it is for.

Also drops `returns a result for every item`, added earlier today: the ordering test asserts the same three items with `shouldContainExactly`, so it already covers membership.
@zipdoki

zipdoki commented Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Optimistic Merge

@zipdoki
zipdoki merged commit a10a15f into main Aug 27, 2026
3 checks passed
@zipdoki
zipdoki deleted the feat/topk-aggregate branch August 27, 2026 09:34
zipdoki added a commit that referenced this pull request Aug 27, 2026
`AggregationSweepResult` names no item it came from, so a caller pairs a result to its request by position — the same contract `aggregate` has. `sweep` merged handler output as it completed, so a three-item batch came back fully reversed when the first item was the slowest. Same fix as #494 made on the aggregate path, and concurrency is unchanged: `flatMapSequential` subscribes eagerly under the same default `maxConcurrency`.
zipdoki added a commit that referenced this pull request Aug 27, 2026
* Recompute a ranking from a refresh message

* Remove the expire payloads the refresh queue replaced

The sweep design moved from an _expire table scanned by expiredAt to a
refresh queue, and these three payloads never got wired to anything.
AggregationExpireItemRequest has no references at all, and
AggregationExpireResult is only read by AggregationsExpireResponse.from,
which nothing calls.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Sweep results come back in request order

`AggregationSweepResult` names no item it came from, so a caller pairs a result to its request by position — the same contract `aggregate` has. `sweep` merged handler output as it completed, so a three-item batch came back fully reversed when the first item was the slowest. Same fix as #494 made on the aggregate path, and concurrency is unchanged: `flatMapSequential` subscribes eagerly under the same default `maxConcurrency`.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
zipdoki added a commit that referenced this pull request Aug 31, 2026
Steps 3 and 4 landed on main as squash merges (#494, #496) carrying review
changes this branch never saw: `AggregationConstants.Topk.REFRESH_TABLE` is now
`REFRESH_QUEUE`, the aggregate and sweep dispatchers emit in request order via
`flatMapSequential`, and both test classes gained cases for that ordering and for
a window that never slides. Took main's side wherever the two overlapped, since
this branch's copies are the pre-review versions of the same work. Kept the
`AggregationQueryService` bean in `GraphConfiguration`, which is Step 5 and
exists only here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request size:XL This PR changes 500-999 lines, ignoring generated files.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant