From 5f65d514066780231654464a8e240418c3c4692a Mon Sep 17 00:00:00 2001 From: Dokyung Lee <112409928+zipdoki@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:13:42 +0900 Subject: [PATCH 1/3] Recompute a ranking from a refresh message --- .../payload/AggregationExpireItemRequest.kt | 11 ++ .../edge/payload/AggregationExpireResult.kt | 9 + .../edge/payload/AggregationSweepRequest.kt | 40 ++++ .../edge/payload/AggregationSweepResult.kt | 10 + .../payload/AggregationsExpireResponse.kt | 29 +++ .../edge/payload/AggregationsSweepResponse.kt | 31 +++ .../engine/service/AggregationService.kt | 10 + .../service/aggregation/AggregationHandler.kt | 6 + .../aggregation/TopkAggregationHandler.kt | 57 +++++- .../engine/service/AggregationServiceTest.kt | 36 +++- .../aggregation/TopkAggregationHandlerTest.kt | 121 ++++++++++++ .../v3/metadata/MetadataAggController.kt | 10 + .../metadata/MetadataAggControllerE2ETest.kt | 176 ++++++++++++++++++ .../filter/ReadOnlyRequestFilterTest.kt | 1 + 14 files changed, 545 insertions(+), 2 deletions(-) create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepRequest.kt create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepResult.kt create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt create mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsSweepResponse.kt diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt new file mode 100644 index 000000000..a71ff6ea4 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt @@ -0,0 +1,11 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationExpireItemRequest( + val items: List, +) + +data class AggregationExpireItemPayload( + val database: String, + val table: String, + val edge: EdgePayload, +) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt new file mode 100644 index 000000000..129ddb9a0 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt @@ -0,0 +1,9 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationExpireResult( + val database: String, + val table: String, + val source: String, + val status: String, + val error: String?, +) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepRequest.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepRequest.kt new file mode 100644 index 000000000..102c20816 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepRequest.kt @@ -0,0 +1,40 @@ +package com.kakao.actionbase.core.edge.payload + +import com.kakao.actionbase.core.metadata.common.AggregationType + +import com.fasterxml.jackson.annotation.JsonSubTypes +import com.fasterxml.jackson.annotation.JsonTypeInfo + +data class AggregationSweepRequest( + val items: List, +) + +data class SweepItem( + val type: AggregationType, + @field:JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.EXTERNAL_PROPERTY, + property = "type", + ) + @field:JsonSubTypes( + JsonSubTypes.Type(value = TopkSweepItem::class, name = "TOPK"), + ) + val item: SweepItemPayload, +) + +sealed interface SweepItemPayload + +data class TopkSweepItem( + val database: String, + val table: String, + val topk: String, + val source: String, + val target: String, + val direction: String, + val ranges: String = "", + val entity: String, + val topkDimensionValue: String, + val dimensionValues: String = "", + val properties: Map = emptyMap(), + val refreshAt: Long = -1, +) : SweepItemPayload diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepResult.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepResult.kt new file mode 100644 index 000000000..3f3d1fa2c --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationSweepResult.kt @@ -0,0 +1,10 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationSweepResult( + val database: String, + val table: String, + val topk: String, + val entity: String, + val status: String, + val error: String?, +) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt new file mode 100644 index 000000000..0fbbb73f9 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt @@ -0,0 +1,29 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationsExpireResponse( + val items: List, +) { + data class Item( + val database: String, + val table: String, + val source: String, + val status: String, + val error: String?, + ) + + companion object { + fun from(expireResults: List): AggregationsExpireResponse = + AggregationsExpireResponse( + items = + expireResults.map { aggregationResult -> + Item( + database = aggregationResult.database, + table = aggregationResult.table, + source = aggregationResult.source, + status = aggregationResult.status, + error = aggregationResult.error, + ) + }, + ) + } +} diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsSweepResponse.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsSweepResponse.kt new file mode 100644 index 000000000..c37467aab --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsSweepResponse.kt @@ -0,0 +1,31 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationsSweepResponse( + val items: List, +) { + data class Item( + val database: String, + val table: String, + val topk: String, + val entity: String, + val status: String, + val error: String?, + ) + + companion object { + fun from(sweepResults: List): AggregationsSweepResponse = + AggregationsSweepResponse( + items = + sweepResults.map { result -> + Item( + database = result.database, + table = result.table, + topk = result.topk, + entity = result.entity, + status = result.status, + error = result.error, + ) + }, + ) + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt index 662c85c01..94bfbf83b 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt @@ -2,6 +2,8 @@ package com.kakao.actionbase.engine.service import com.kakao.actionbase.core.edge.payload.AggregationItemPayload import com.kakao.actionbase.core.edge.payload.AggregationResult +import com.kakao.actionbase.core.edge.payload.AggregationSweepResult +import com.kakao.actionbase.core.edge.payload.SweepItem import com.kakao.actionbase.core.metadata.QualifiedAggregations import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.engine.AggregationEngine @@ -23,4 +25,12 @@ class AggregationService( .fromIterable(items) .flatMapSequential { item -> Flux.mergeSequential(handlersByType.values.map { it.aggregate(item) }) } .collectList() + + fun sweep(items: List): Mono> = + Flux + .fromIterable(items) + .flatMap { item -> handler(item.type).sweep(item.item) } + .collectList() + + private fun handler(type: AggregationType): AggregationHandler = handlersByType[type] ?: error("No aggregation handler for type $type") } diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/AggregationHandler.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/AggregationHandler.kt index 036ab6675..f21039355 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/AggregationHandler.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/AggregationHandler.kt @@ -2,9 +2,12 @@ package com.kakao.actionbase.engine.service.aggregation import com.kakao.actionbase.core.edge.payload.AggregationItemPayload import com.kakao.actionbase.core.edge.payload.AggregationResult +import com.kakao.actionbase.core.edge.payload.AggregationSweepResult +import com.kakao.actionbase.core.edge.payload.SweepItemPayload import com.kakao.actionbase.core.metadata.common.AggregationType import reactor.core.publisher.Flux +import reactor.core.publisher.Mono /** * One aggregation kind's write path. [AggregationService] dispatches to the handler @@ -16,4 +19,7 @@ interface AggregationHandler { /** Aggregates a single edge event and writes its result rows. */ fun aggregate(item: AggregationItemPayload): Flux + + /** Recomputes a single refreshed event and re-writes its result row. */ + fun sweep(item: SweepItemPayload): Mono } diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandler.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandler.kt index f0a94ffea..c9780fc30 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandler.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandler.kt @@ -5,8 +5,11 @@ import com.kakao.actionbase.v2.core.metadata.Direction as V2Direction import com.kakao.actionbase.core.edge.Edge import com.kakao.actionbase.core.edge.payload.AggregationItemPayload import com.kakao.actionbase.core.edge.payload.AggregationResult +import com.kakao.actionbase.core.edge.payload.AggregationSweepResult import com.kakao.actionbase.core.edge.payload.EdgeBulkMutationRequest.MutationItem import com.kakao.actionbase.core.edge.payload.MutationResult +import com.kakao.actionbase.core.edge.payload.SweepItemPayload +import com.kakao.actionbase.core.edge.payload.TopkSweepItem import com.kakao.actionbase.core.metadata.common.AggregationConstants import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.metadata.common.Aggregations @@ -54,6 +57,58 @@ class TopkAggregationHandler( }, ).flatMapSequential { (event, direction, topk) -> processTopk(event, direction, topk) } + override fun sweep(item: SweepItemPayload): Mono { + require(item is TopkSweepItem) { "TopkAggregationHandler handles TopkSweepItem, got ${item::class.simpleName}" } + + fun result( + status: String, + error: String? = null, + ) = AggregationSweepResult( + database = item.database, + table = item.table, + topk = item.topk, + entity = item.entity, + status = status, + error = error, + ) + + val tb = engine.getTableBinding(database = item.database, alias = item.table) + val group = + tb.schema.groups + .firstOrNull { g -> g.aggregations.topk.any { it.topk == item.topk } } + ?: return Mono.just(result(SKIPPED)) + + val topk = group.aggregations.topk.first { it.topk == item.topk } + val direction = Direction.valueOf(item.direction) + val directedSource = if (direction == Direction.IN) item.target else item.source + val dimensionValues = AggregationConstants.Topk.splitValues(item.dimensionValues) + val (rankDatabase, rankTable) = parseFqn(topk.rank) + + return writeRank( + sourceDatabase = item.database, + sourceTable = tb.table, + group = group.group, + start = directedSource, + direction = direction, + ranges = item.ranges.takeIf { it.isNotEmpty() }, + ranking = + Ranking( + database = rankDatabase, + table = rankTable, + topk = item.topk, + entity = item.entity, + topkDimensionValue = item.topkDimensionValue, + dimensionValues = dimensionValues, + properties = item.properties, + ), + ).map { results -> + result(status = if (results.any { it.status == ERROR }) ERROR else SUCCESS) + }.onErrorResume { err -> + logger.error("topk sweep failed for {}.{} topk={}", item.database, item.table, item.topk, err) + Mono.just(result(status = ERROR, error = err.message)) + } + } + private fun createTopkEvent(item: AggregationItemPayload): List { val tb = engine.getTableBinding(database = item.database, alias = item.table) @@ -128,7 +183,7 @@ class TopkAggregationHandler( /** * Runs the aggregation query for one ranking and writes its rank row. - * Used by the aggregate flow, which then enqueues a refresh message + * Shared by the aggregate flow (which then enqueues a refresh message) and the sweep flow * (which recomputes from an already-resolved [Ranking]). */ private fun writeRank( diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt index 29cf06fd6..30ed7aac0 100644 --- a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt @@ -2,7 +2,11 @@ package com.kakao.actionbase.engine.service import com.kakao.actionbase.core.edge.payload.AggregationItemPayload import com.kakao.actionbase.core.edge.payload.AggregationResult +import com.kakao.actionbase.core.edge.payload.AggregationSweepResult import com.kakao.actionbase.core.edge.payload.EdgePayload +import com.kakao.actionbase.core.edge.payload.SweepItem +import com.kakao.actionbase.core.edge.payload.SweepItemPayload +import com.kakao.actionbase.core.edge.payload.TopkSweepItem import com.kakao.actionbase.core.metadata.QualifiedAggregations import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.engine.AggregationEngine @@ -23,7 +27,7 @@ import reactor.test.StepVerifier /** * `AggregationService` is a thin dispatcher: it forwards metadata lookups to the engine and routes - * each item to the handler registered for its type. Per-type behavior (top-K ranking) lives + * each item to the handler registered for its type. Per-type behavior (top-K ranking, refresh) lives * in the handlers, and dispatch through a real handler is exercised end-to-end by * `MetadataAggQueryControllerE2ETest` — so here we only pin what the dispatcher itself owns. */ @@ -108,6 +112,18 @@ class AggregationServiceTest { .verifyComplete() } } + + @Nested + inner class Sweep { + @Test + fun `errors when no handler is registered for the type`() { + val service = AggregationService(engine, emptyList()) + + StepVerifier + .create(service.sweep(listOf(sweepItem()))) + .verifyError(IllegalStateException::class.java) + } + } } // region test fixtures @@ -126,6 +142,22 @@ private fun item(source: String): AggregationItemPayload = ), ) +private fun sweepItem(): SweepItem = + SweepItem( + type = AggregationType.TOPK, + item = + TopkSweepItem( + database = "commerce", + table = "orders", + topk = "top_purchased", + source = "user1", + target = "item1", + direction = "OUT", + entity = "user1", + topkDimensionValue = "item1", + ), + ) + /** * Answers after a delay the test fixes per item, standing in for a real handler whose storage calls come * back out of order. Without it every inner publisher completes on the subscribing thread and the @@ -154,6 +186,8 @@ private class DelayedHandler( ).delayElement(delays[source] ?: Duration.ZERO) .flux() } + + override fun sweep(item: SweepItemPayload): Mono = error("this fake stands in on the aggregate path only") } // endregion diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandlerTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandlerTest.kt index f54d4fc49..b3133a883 100644 --- a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandlerTest.kt +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandlerTest.kt @@ -6,6 +6,7 @@ import com.kakao.actionbase.core.edge.payload.DataFrameEdgeAggPayload import com.kakao.actionbase.core.edge.payload.EdgeBulkMutationRequest.MutationItem import com.kakao.actionbase.core.edge.payload.EdgePayload import com.kakao.actionbase.core.edge.payload.MutationResult +import com.kakao.actionbase.core.edge.payload.TopkSweepItem import com.kakao.actionbase.core.metadata.common.AggregationConstants import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.metadata.common.Aggregations @@ -500,6 +501,98 @@ class TopkAggregationHandlerTest { }.verifyComplete() } } + + @Nested + inner class Sweep { + @Test + fun `recomputes the ranking and rewrites the rank row`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "top_purchased")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 9)) + + val mutations = slot>() + every { + mutationService.mutate("commerce", "orders__topk", capture(mutations), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.sweep(sweepTarget(topk = "top_purchased"))) + .assertNext { result -> + result.status shouldBe "SUCCESS" + result.topk shouldBe "top_purchased" + result.entity shouldBe "user1" + }.verifyComplete() + + val edge = mutations.captured.single().edge + edge.source shouldBe "commerce|orders|top_purchased|user1" + edge.target shouldBe "item1" + } + + @Test + fun `rewrites the carried properties onto the rank row`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "top_purchased")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 9)) + + val mutations = slot>() + every { + mutationService.mutate("commerce", "orders__topk", capture(mutations), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.sweep(sweepTarget(topk = "top_purchased", properties = mapOf("category" to "fruit")))) + .assertNext { it.status shouldBe "SUCCESS" } + .verifyComplete() + + val edge = mutations.captured.single().edge + edge.properties.containsKey("metric") shouldBe true + edge.properties["additionalProperties"] shouldBe """{"category":"fruit"}""" + } + + @Test + fun `is skipped when the topk is not defined on the table`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "top_purchased")) + + StepVerifier + .create(handler.sweep(sweepTarget(topk = "unknown_topk"))) + .assertNext { it.status shouldBe "SKIPPED" } + .verifyComplete() + + verify(exactly = 0) { mutationService.mutate(any(), any(), any(), any(), any(), any(), any()) } + } + + @Test + fun `IN aggregates from the target endpoint`() { + stubTopkBinding( + engine = engine, + topk = topkConfig(name = "top_purchased_by", entity = "target", dimension = "source"), + directionType = DirectionType.IN, + ) + + val starts = slot>() + every { + queryService.agg(any(), any(), any(), capture(starts), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 4)) + + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create( + handler.sweep( + sweepTarget(topk = "top_purchased_by", direction = "IN", entity = "item1", topkDimensionValue = "user1"), + ), + ).assertNext { it.status shouldBe "SUCCESS" } + .verifyComplete() + + starts.captured shouldBe listOf("item1") + } + } } // region test fixtures @@ -540,6 +633,34 @@ private fun topkConfig( ranges = ranges, ) +private fun sweepTarget( + topk: String, + database: String = "commerce", + table: String = "orders", + source: String = "user1", + target: String = "item1", + direction: String = "OUT", + ranges: String = "_target:eq:item1", + entity: String = "user1", + topkDimensionValue: String = "item1", + dimensionValues: String = "", + properties: Map = emptyMap(), +): TopkSweepItem = + TopkSweepItem( + database = database, + table = table, + topk = topk, + source = source, + target = target, + direction = direction, + ranges = ranges, + entity = entity, + topkDimensionValue = topkDimensionValue, + dimensionValues = dimensionValues, + properties = properties, + refreshAt = 123L, + ) + private fun stubTopkBinding( engine: AggregationEngine, topk: Topk, diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggController.kt b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggController.kt index 6ad9a11bc..a26dcc77c 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggController.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggController.kt @@ -1,7 +1,9 @@ package com.kakao.actionbase.server.api.graph.v3.metadata import com.kakao.actionbase.core.edge.payload.AggregationItemRequest +import com.kakao.actionbase.core.edge.payload.AggregationSweepRequest import com.kakao.actionbase.core.edge.payload.AggregationsItemResponse +import com.kakao.actionbase.core.edge.payload.AggregationsSweepResponse import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.metadata.payload.AggregationsListResponse import com.kakao.actionbase.engine.service.AggregationService @@ -35,4 +37,12 @@ class MetadataAggController( aggregationService .aggregate(items = request.items) .map { results -> ResponseEntity.ok(AggregationsItemResponse.from(results)) } + + @PostMapping("/aggregations/v1/sweep") + fun sweep( + @RequestBody request: AggregationSweepRequest, + ): Mono> = + aggregationService + .sweep(items = request.items) + .map { results -> ResponseEntity.ok(AggregationsSweepResponse.from(results)) } } diff --git a/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggControllerE2ETest.kt b/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggControllerE2ETest.kt index dcfcfa57b..d8eefebb8 100644 --- a/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggControllerE2ETest.kt +++ b/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggControllerE2ETest.kt @@ -1,5 +1,7 @@ package com.kakao.actionbase.server.api.graph.v3.metadata +import com.kakao.actionbase.core.edge.payload.AggregationsSweepResponse +import com.kakao.actionbase.core.edge.payload.DataFrameEdgePayload import com.kakao.actionbase.core.metadata.common.AggregationConstants import com.kakao.actionbase.core.metadata.common.AggregationConstants.Topk.DATABASE import com.kakao.actionbase.core.metadata.common.AggregationConstants.Topk.REFRESH_QUEUE @@ -156,6 +158,180 @@ class MetadataAggControllerE2ETest : E2ETestBase() { assertEquals(listOf("source", "_target"), source?.dedupeFields) } + /** + * `POST /aggregations/v1/sweep` recomputes a ranking on demand from an already-resolved + * refresh target and re-writes its rank row. + * + * 1. Create a rank table and a source table under `commerce_sweep`. + * 2. Record one edge user1 -> item1. + * 3. Sweep the target. The response reports the recomputed ranking: + * + * commerce_sweep.orders__topk (rank) + * | row key | target | metric | + * |-----------------------|--------|--------| + * | top_purchased | user1 | item1 | 1 | + * + * 4. Scan `metric_desc` from the entity prefix -> the top row's target is `item1`. + */ + @Test + fun `sweep recomputes the ranking and re-writes the rank row`() { + val sweepDb = "commerce_sweep" + val sweepTable = "orders" + val sweepRank = "orders__topk" + val sweepRankFqn = "$sweepDb.$sweepRank" + val topkName = "top_purchased" + + client + .post() + .uri("/graph/v3/databases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"database": "$sweepDb", "comment": "test"}""") + .exchange() + + client + .post() + .uri("/graph/v3/databases/$sweepDb/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$sweepRank", + "schema": { + "type": "EDGE", + "source": {"type": "string", "comment": "topk|entity|dimensionValues"}, + "target": {"type": "string", "comment": "topkDimensionValue"}, + "properties": [ + {"name": "metric", "type": "long", "comment": "metric", "nullable": false} + ], + "direction": "OUT", + "indexes": [ + {"index": "metric_desc", "fields": [{"field": "metric", "order": "DESC"}]} + ], + "groups": [], + "caches": [] + }, + "storage": "datastore://test_namespace/$sweepRank", + "mode": "SYNC", + "comment": "topk rank rows" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/graph/v3/databases/$sweepDb/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$sweepTable", + "schema": { + "type": "MULTI_EDGE", + "id": {"type": "long", "comment": "order id"}, + "source": {"type": "string", "comment": "user"}, + "target": {"type": "string", "comment": "item"}, + "properties": [], + "direction": "OUT", + "indexes": [], + "groups": [{ + "group": "purchased_count", + "type": "COUNT", + "fields": [{"name": "_target"}], + "directionType": "OUT", + "aggregations": { + "topk": [{ + "topk": "$topkName", + "entity": "source", + "ranges": "_target:eq:{_target}", + "dimension": "target", + "rank": "$sweepRankFqn" + }] + } + }], + "caches": [] + }, + "storage": "datastore://test_namespace/$sweepTable", + "mode": "SYNC", + "comment": "orders" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/graph/v3/databases/$sweepDb/tables/$sweepTable/multi-edges") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "id": 1, "source": "user1", "target": "item1", "properties": {}}} + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + val response = + client + .post() + .uri("/aggregations/v1/sweep") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "items": [ + { + "type": "TOPK", + "item": {"database": "$sweepDb", "table": "$sweepTable", "topk": "$topkName", + "source": "user1", "target": "item1", "direction": "OUT", "ranges": "_target:eq:item1", + "entity": "user1", "topkDimensionValue": "item1", "dimensionValues": "", "refreshAt": 123} + } + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + .expectBody() + .returnResult() + .responseBody!! + + assertEquals(1, response.items.size) + assertEquals(sweepDb, response.items[0].database) + assertEquals(sweepTable, response.items[0].table) + assertEquals(topkName, response.items[0].topk) + assertEquals("user1", response.items[0].entity) + assertEquals("SUCCESS", response.items[0].status) + + client + .get() + .uri( + "/graph/v3/databases/$sweepDb/tables/$sweepRank/edges/scan/metric_desc" + + "?start=$sweepDb|$sweepTable|$topkName|user1&direction=OUT", + ).exchange() + .expectStatus() + .isOk + .expectBody() + .returnResult() + .responseBody!! + .let { payload -> + assertEquals(1, payload.count) + assertEquals( + "item1", + payload.edges + .single() + .target + .toString(), + ) + } + } + /** * The `commerce.purchases` top-K enables `refreshAfterMillis`, so an aggregation also enqueues a * refresh message. diff --git a/server/src/test/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilterTest.kt b/server/src/test/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilterTest.kt index eca7b3473..c2798620c 100644 --- a/server/src/test/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilterTest.kt +++ b/server/src/test/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilterTest.kt @@ -229,6 +229,7 @@ class ReadOnlyRequestFilterTest { "DELETE /graph/v3/databases/{database}/tables/{table}", "DELETE /graph/v3/databases/{database}/tables/{table}/edges/scan/{index}", "POST /aggregations/v1/aggregate", + "POST /aggregations/v1/sweep", "POST /graph/v3/databases", "POST /graph/v3/databases/{database}/aliases", "POST /graph/v3/databases/{database}/tables", From e2c2b0570ec2626731249e0e98772fb9b93cca5d Mon Sep 17 00:00:00 2001 From: Dokyung Lee <112409928+zipdoki@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:56:12 +0900 Subject: [PATCH 2/3] 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) --- .../payload/AggregationExpireItemRequest.kt | 11 ------- .../edge/payload/AggregationExpireResult.kt | 9 ------ .../payload/AggregationsExpireResponse.kt | 29 ------------------- 3 files changed, 49 deletions(-) delete mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt delete mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt delete mode 100644 core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt deleted file mode 100644 index a71ff6ea4..000000000 --- a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireItemRequest.kt +++ /dev/null @@ -1,11 +0,0 @@ -package com.kakao.actionbase.core.edge.payload - -data class AggregationExpireItemRequest( - val items: List, -) - -data class AggregationExpireItemPayload( - val database: String, - val table: String, - val edge: EdgePayload, -) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt deleted file mode 100644 index 129ddb9a0..000000000 --- a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationExpireResult.kt +++ /dev/null @@ -1,9 +0,0 @@ -package com.kakao.actionbase.core.edge.payload - -data class AggregationExpireResult( - val database: String, - val table: String, - val source: String, - val status: String, - val error: String?, -) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt deleted file mode 100644 index 0fbbb73f9..000000000 --- a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsExpireResponse.kt +++ /dev/null @@ -1,29 +0,0 @@ -package com.kakao.actionbase.core.edge.payload - -data class AggregationsExpireResponse( - val items: List, -) { - data class Item( - val database: String, - val table: String, - val source: String, - val status: String, - val error: String?, - ) - - companion object { - fun from(expireResults: List): AggregationsExpireResponse = - AggregationsExpireResponse( - items = - expireResults.map { aggregationResult -> - Item( - database = aggregationResult.database, - table = aggregationResult.table, - source = aggregationResult.source, - status = aggregationResult.status, - error = aggregationResult.error, - ) - }, - ) - } -} From 01284118e5fb4f52c49ca83d00a01241152ee57a Mon Sep 17 00:00:00 2001 From: Dokyung Lee <112409928+zipdoki@users.noreply.github.com> Date: Thu, 27 Aug 2026 18:46:37 +0900 Subject: [PATCH 3/3] Sweep results come back in request order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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`. --- .../engine/service/AggregationService.kt | 2 +- .../engine/service/AggregationServiceTest.kt | 48 +++++++++++++++++-- 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt index 94bfbf83b..fe232468e 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationService.kt @@ -29,7 +29,7 @@ class AggregationService( fun sweep(items: List): Mono> = Flux .fromIterable(items) - .flatMap { item -> handler(item.type).sweep(item.item) } + .flatMapSequential { item -> handler(item.type).sweep(item.item) } .collectList() private fun handler(type: AggregationType): AggregationHandler = handlersByType[type] ?: error("No aggregation handler for type $type") diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt index 30ed7aac0..777d1fcfd 100644 --- a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt @@ -115,6 +115,32 @@ class AggregationServiceTest { @Nested inner class Sweep { + /** `AggregationSweepResult` pairs back to its request by position too, same as the aggregate path. */ + @Test + fun `returns results in the order the items were given`() { + val service = + AggregationService( + engine, + listOf( + DelayedHandler( + delays = + mapOf( + "user1" to Duration.ofMillis(150), + "user2" to Duration.ofMillis(80), + "user3" to Duration.ofMillis(10), + ), + ), + ), + ) + + StepVerifier + .withVirtualTime { service.sweep(listOf(sweepItem("user1"), sweepItem("user2"), sweepItem("user3"))) } + .expectSubscription() + .thenAwait(Duration.ofMinutes(1)) + .assertNext { results -> results.map { it.entity } shouldContainExactly listOf("user1", "user2", "user3") } + .verifyComplete() + } + @Test fun `errors when no handler is registered for the type`() { val service = AggregationService(engine, emptyList()) @@ -142,7 +168,7 @@ private fun item(source: String): AggregationItemPayload = ), ) -private fun sweepItem(): SweepItem = +private fun sweepItem(entity: String = "user1"): SweepItem = SweepItem( type = AggregationType.TOPK, item = @@ -150,10 +176,10 @@ private fun sweepItem(): SweepItem = database = "commerce", table = "orders", topk = "top_purchased", - source = "user1", + source = entity, target = "item1", direction = "OUT", - entity = "user1", + entity = entity, topkDimensionValue = "item1", ), ) @@ -187,7 +213,21 @@ private class DelayedHandler( .flux() } - override fun sweep(item: SweepItemPayload): Mono = error("this fake stands in on the aggregate path only") + override fun sweep(item: SweepItemPayload): Mono { + val topk = item as TopkSweepItem + + return Mono + .just( + AggregationSweepResult( + database = topk.database, + table = tag, + topk = topk.topk, + entity = topk.entity, + status = "SUCCESS", + error = null, + ), + ).delayElement(delays[topk.entity] ?: Duration.ZERO) + } } // endregion