diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationItemRequest.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationItemRequest.kt new file mode 100644 index 000000000..e7290c4c8 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationItemRequest.kt @@ -0,0 +1,11 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationItemRequest( + val items: List, +) + +data class AggregationItemPayload( + val database: String, + val table: String, + val edge: EdgePayload, +) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationResult.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationResult.kt new file mode 100644 index 000000000..a9adf42f9 --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationResult.kt @@ -0,0 +1,10 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationResult( + val database: String, + val table: String, + val source: String, + val target: String, + val status: String, + val error: String?, +) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsItemResponse.kt b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsItemResponse.kt new file mode 100644 index 000000000..e78b45b2b --- /dev/null +++ b/core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsItemResponse.kt @@ -0,0 +1,31 @@ +package com.kakao.actionbase.core.edge.payload + +data class AggregationsItemResponse( + val items: List, +) { + data class Item( + val database: String, + val table: String, + val source: String, + val target: String, + val status: String, + val error: String?, + ) + + companion object { + fun from(aggregationResults: List): AggregationsItemResponse = + AggregationsItemResponse( + items = + aggregationResults.map { aggregationResult -> + Item( + database = aggregationResult.database, + table = aggregationResult.table, + source = aggregationResult.source, + target = aggregationResult.target, + status = aggregationResult.status, + error = aggregationResult.error, + ) + }, + ) + } +} diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstants.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstants.kt index e84cb085c..8894649e7 100644 --- a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstants.kt +++ b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstants.kt @@ -22,7 +22,7 @@ object AggregationConstants { topk: String, entity: String, dimensionValues: List, - ): String = (listOf(database, table, topk, entity) + dimensionValues).joinToString("|") + ): String = joinValues(listOf(database, table, topk, entity) + dimensionValues) // refresh queue message key: database | table | topk | entity | topkDimensionValue | dimensionValue1 | ... // the queue derives the partition from this key, so pass the raw composite (not a pre-hashed value). @@ -33,6 +33,44 @@ object AggregationConstants { entity: String, topkDimensionValue: String, dimensionValues: List, - ): String = (listOf(database, table, topk, entity, topkDimensionValue) + dimensionValues).joinToString("|") + ): String = joinValues(listOf(database, table, topk, entity, topkDimensionValue) + dimensionValues) + + /** An entity id or a dimension value can hold the separator itself (`kakao|12345`), which would otherwise let two rankings share one key. */ + fun joinValues(values: List): String = values.joinToString(SEPARATOR.toString()) { escape(it) } + + /** An empty string reads back as no values: a ranking with none joins to the same string as one whose single value is empty, and none is the common case. */ + fun splitValues(joined: String): List { + if (joined.isEmpty()) return emptyList() + + val values = mutableListOf() + val value = StringBuilder() + var i = 0 + + while (i < joined.length) { + val char = joined[i] + when { + char == ESCAPE && i + 1 < joined.length -> value.append(joined[i + 1]).also { i += 2 } + char == SEPARATOR -> values.add(value.toString()).also { value.clear() }.also { i++ } + else -> value.append(char).also { i++ } + } + } + + return values + value.toString() + } + + private fun escape(value: String): String = + if (value.none { it == SEPARATOR || it == ESCAPE }) { + value + } else { + buildString { + value.forEach { char -> + if (char == SEPARATOR || char == ESCAPE) append(ESCAPE) + append(char) + } + } + } + + private const val SEPARATOR = '|' + private const val ESCAPE = '\\' } } diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Bucket.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Bucket.kt index c5ceaf390..86ee1f127 100644 --- a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Bucket.kt +++ b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Bucket.kt @@ -6,6 +6,7 @@ import java.time.Duration import java.time.Instant import java.time.ZoneId import java.time.format.DateTimeFormatter +import java.time.temporal.ChronoUnit import java.util.regex.Pattern import com.fasterxml.jackson.annotation.JsonIgnore @@ -46,22 +47,65 @@ sealed class Bucket { if (value == null) return null return try { - val longValue = PrimitiveType.LONG.cast(value) as Long - - val instant = - when (unit) { - ValueUnit.NANOSECOND -> Instant.ofEpochSecond(0, longValue) - ValueUnit.MICROSECOND -> Instant.ofEpochSecond(longValue / 1_000_000, (longValue % 1_000_000) * 1000) - ValueUnit.MILLISECOND -> Instant.ofEpochMilli(longValue) - ValueUnit.SECOND -> Instant.ofEpochSecond(longValue) - } - - instant.atZone(zoneId).format(formatter) + toInstant(value).atZone(zoneId).format(formatter) } catch (_: Exception) { null } } + /** The instant this value's bucket begins at — the same truncation [apply] performs, as a time. */ + fun startOf(value: Any?): Instant? = + if (value == null) { + null + } else { + try { + floorToFormatPrecision(toInstant(value)) + } catch (_: Exception) { + null + } + } + + /** How long one bucket spans, which is the precision the format writes down. */ + fun interval(): Duration = granularity().duration + + /** Whether a range bound moves with the clock (`now`, `now-365d`) rather than naming a fixed point. */ + fun isRelative(value: Any): Boolean { + val input = value.toString().trim() + return input == "now" || pattern.matcher(input).matches() + } + + private fun toInstant(value: Any): Instant { + val longValue = PrimitiveType.LONG.cast(value) as Long + + return when (unit) { + ValueUnit.NANOSECOND -> Instant.ofEpochSecond(0, longValue) + ValueUnit.MICROSECOND -> Instant.ofEpochSecond(longValue / 1_000_000, (longValue % 1_000_000) * 1000) + ValueUnit.MILLISECOND -> Instant.ofEpochMilli(longValue) + ValueUnit.SECOND -> Instant.ofEpochSecond(longValue) + } + } + + private fun floorToFormatPrecision(instant: Instant): Instant = instant.atZone(zoneId).truncatedTo(granularity()).toInstant() + + /** The unit the format keeps, and so the unit the bucket and every bound built from it move in. */ + private fun granularity(): ChronoUnit = + when { + // Includes nanoseconds/microseconds (S, SSS, SSSSSS, SSSSSSSSS, etc.) + format.contains('S') -> throw IllegalArgumentException("Units below milliseconds are not supported: $format") + + // Up to seconds only (includes ss, no S) + format.contains("ss") -> throw IllegalArgumentException("Second units are not supported: $format") + + // Up to minutes only (includes mm, no ss) + format.contains("mm") -> ChronoUnit.MINUTES + + // Up to hours only (includes HH, no mm) + format.contains("HH") || format.contains("H") -> ChronoUnit.HOURS + + // Day units only (date only) + else -> ChronoUnit.DAYS + } + override fun handleQueryValue( value: Any, ceil: Boolean, @@ -106,46 +150,9 @@ sealed class Bucket { private fun ceilToFormatPrecision(instant: Instant): Instant { val zoned = instant.atZone(zoneId) + val truncated = zoned.truncatedTo(granularity()) - return when { - // Includes nanoseconds/microseconds (S, SSS, SSSSSS, SSSSSSSSS, etc.) - format.contains('S') -> - throw IllegalArgumentException("Units below milliseconds are not supported: $format") - - // Up to seconds only (includes ss, no S) - format.contains("ss") -> - throw IllegalArgumentException("Second units are not supported: $format") - - // Up to minutes only (includes mm, no ss) - format.contains("mm") -> { - val truncated = zoned.truncatedTo(java.time.temporal.ChronoUnit.MINUTES) - if (truncated == zoned) { - instant - } else { - truncated.plusMinutes(1).toInstant() - } - } - - // Up to hours only (includes HH, no mm) - format.contains("HH") || format.contains("H") -> { - val truncated = zoned.truncatedTo(java.time.temporal.ChronoUnit.HOURS) - if (truncated == zoned) { - instant - } else { - truncated.plusHours(1).toInstant() - } - } - - // Day units only (date only) - else -> { - val truncated = zoned.truncatedTo(java.time.temporal.ChronoUnit.DAYS) - if (truncated == zoned) { - instant - } else { - truncated.plusDays(1).toInstant() - } - } - } + return if (truncated == zoned) instant else truncated.plus(1, granularity()).toInstant() } companion object { diff --git a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstantsTest.kt b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstantsTest.kt new file mode 100644 index 000000000..db7fc6103 --- /dev/null +++ b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstantsTest.kt @@ -0,0 +1,61 @@ +package com.kakao.actionbase.core.metadata.common + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotEquals +import org.junit.jupiter.api.Test + +/** + * A rank key is built from data — an entity id, a dimension value — so it has to survive a value that + * holds the separator itself. + */ +class AggregationConstantsTest { + @Test + fun `two rankings that differ only in where the separator falls get different keys`() { + val entityHoldsIt = AggregationConstants.Topk.rankSource("shop", "purchased", "popular", "kakao|1", listOf("fruit")) + val dimensionHoldsIt = AggregationConstants.Topk.rankSource("shop", "purchased", "popular", "kakao", listOf("1|fruit")) + + assertNotEquals(entityHoldsIt, dimensionHoldsIt) + } + + @Test + fun `a refresh key separates the same way`() { + val entityHoldsIt = AggregationConstants.Topk.refreshKey("shop", "purchased", "popular", "kakao|1", "apple", listOf("fruit")) + val dimensionHoldsIt = AggregationConstants.Topk.refreshKey("shop", "purchased", "popular", "kakao", "apple", listOf("1|fruit")) + + assertNotEquals(entityHoldsIt, dimensionHoldsIt) + } + + /** Keys already written by the unescaped version have to keep reading back the same row. */ + @Test + fun `a value without the separator is joined as it was before`() { + assertEquals( + "shop|purchased|popular|user1|fruit", + AggregationConstants.Topk.rankSource("shop", "purchased", "popular", "user1", listOf("fruit")), + ) + } + + @Test + fun `a joined value reads back as it was written`() { + val values = listOf("kakao|1", "fruit", "a\\b", "|", "\\", "", "plain") + + assertEquals(values, AggregationConstants.Topk.splitValues(AggregationConstants.Topk.joinValues(values))) + } + + @Test + fun `an escaped separator stays inside its value`() { + assertEquals("kakao\\|1|fruit", AggregationConstants.Topk.joinValues(listOf("kakao|1", "fruit"))) + assertEquals(listOf("kakao|1", "fruit"), AggregationConstants.Topk.splitValues("kakao\\|1|fruit")) + } + + @Test + fun `no dimension values joins and reads back as none`() { + assertEquals("", AggregationConstants.Topk.joinValues(emptyList())) + assertEquals(emptyList(), AggregationConstants.Topk.splitValues("")) + } + + @Test + fun `an empty value keeps its place next to another`() { + assertEquals(listOf("", "fruit"), AggregationConstants.Topk.splitValues(AggregationConstants.Topk.joinValues(listOf("", "fruit")))) + assertEquals(listOf("fruit", ""), AggregationConstants.Topk.splitValues(AggregationConstants.Topk.joinValues(listOf("fruit", "")))) + } +} diff --git a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/BucketTest.kt b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/BucketTest.kt new file mode 100644 index 000000000..359882a90 --- /dev/null +++ b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/BucketTest.kt @@ -0,0 +1,83 @@ +package com.kakao.actionbase.core.metadata.common + +import java.time.Instant +import java.time.LocalDate +import java.time.LocalDateTime +import java.time.ZoneId +import java.time.format.DateTimeFormatter + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows + +/** + * Pins what a date bucket writes down and how a range bound is read back, so that the arithmetic behind + * both stays observable while it is refactored. + */ +class BucketTest { + private val utc = ZoneId.of("UTC") + + private val day = Bucket.Date(name = "purchasedAt", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd") + + private val hour = Bucket.Date(name = "purchasedAt", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd HH") + + private val second = Bucket.Date(name = "purchasedAt", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd HH:mm:ss") + + @Test + fun `a value is written down at the format's precision`() { + val purchasedAt = Instant.parse("2026-01-01T14:32:07Z").toEpochMilli() + + assertEquals("2026-01-01", day.apply(purchasedAt)) + assertEquals("2026-01-01 14", hour.apply(purchasedAt)) + } + + @Test + fun `a value that is not a time is dropped`() { + assertNull(day.apply("not a time")) + assertNull(day.apply(null)) + } + + @Test + fun `a bound that names a fixed point is returned as it was written`() { + assertEquals("2026-01-01", day.handleQueryValue("2026-01-01", ceil = true)) + assertEquals("2026-01-01", day.handleQueryValue("2026-01-01", ceil = false)) + } + + @Test + fun `a bound that is not text is returned as it was written`() { + assertEquals(42L, day.handleQueryValue(42L, ceil = true)) + } + + @Test + fun `now is the bucket the clock is in`() { + assertEquals(LocalDate.now(utc).toString(), day.handleQueryValue("now", ceil = false)) + assertEquals(LocalDate.now(utc).toString(), day.handleQueryValue("now", ceil = true)) + } + + @Test + fun `a relative bound counts from the clock`() { + assertEquals(LocalDate.now(utc).minusDays(1).toString(), day.handleQueryValue("now-1d", ceil = false)) + assertEquals(LocalDate.now(utc).plusDays(1).toString(), day.handleQueryValue("now+1d", ceil = false)) + } + + /** Raising the clock to the next bucket first is what keeps a truncated bound from widening the range. */ + @Test + fun `ceil raises the clock to the next bucket before counting`() { + assertEquals(LocalDate.now(utc).toString(), day.handleQueryValue("now-1d", ceil = true)) + + val currentHour = LocalDateTime.now(utc).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH")) + assertEquals(currentHour, hour.handleQueryValue("now-1h", ceil = true)) + } + + @Test + fun `a bound is left alone when its unit is not supported`() { + assertEquals("now-1y", day.handleQueryValue("now-1y", ceil = false)) + assertEquals("yesterday", day.handleQueryValue("yesterday", ceil = false)) + } + + @Test + fun `a format finer than a bucket is refused when the clock has to be raised`() { + assertThrows { second.handleQueryValue("now-1h", ceil = true) } + } +} 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 24a569f67..662c85c01 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 @@ -1,11 +1,26 @@ 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.metadata.QualifiedAggregations import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.engine.AggregationEngine +import com.kakao.actionbase.engine.service.aggregation.AggregationHandler + +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono class AggregationService( private val engine: AggregationEngine, + handlers: List, ) { + private val handlersByType: Map = handlers.associateBy { it.type } + fun getAggregations(type: AggregationType? = null): List = engine.getListWithAggregations(type) + + fun aggregate(items: List): Mono> = + Flux + .fromIterable(items) + .flatMapSequential { item -> Flux.mergeSequential(handlersByType.values.map { it.aggregate(item) }) } + .collectList() } 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 new file mode 100644 index 000000000..036ab6675 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/AggregationHandler.kt @@ -0,0 +1,19 @@ +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.metadata.common.AggregationType + +import reactor.core.publisher.Flux + +/** + * One aggregation kind's write path. [AggregationService] dispatches to the handler + * whose [type] matches the request, so a new [AggregationType] is added by providing a + * new handler bean — no changes to the dispatcher. + */ +interface AggregationHandler { + val type: AggregationType + + /** Aggregates a single edge event and writes its result rows. */ + fun aggregate(item: AggregationItemPayload): Flux +} 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 new file mode 100644 index 000000000..f0a94ffea --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandler.kt @@ -0,0 +1,442 @@ +package com.kakao.actionbase.engine.service.aggregation + +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.EdgeBulkMutationRequest.MutationItem +import com.kakao.actionbase.core.edge.payload.MutationResult +import com.kakao.actionbase.core.metadata.common.AggregationConstants +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.core.metadata.common.Aggregations +import com.kakao.actionbase.core.metadata.common.Bucket +import com.kakao.actionbase.core.metadata.common.Direction +import com.kakao.actionbase.core.metadata.common.DirectionType +import com.kakao.actionbase.core.metadata.common.Group +import com.kakao.actionbase.core.metadata.common.ModelSchema +import com.kakao.actionbase.core.metadata.common.Topk +import com.kakao.actionbase.core.state.EventType +import com.kakao.actionbase.engine.AggregationEngine +import com.kakao.actionbase.engine.queue.EnqueueMessage +import com.kakao.actionbase.engine.queue.EnqueueRequest +import com.kakao.actionbase.engine.queue.EnqueueResponse +import com.kakao.actionbase.engine.queue.QueueService +import com.kakao.actionbase.engine.service.MutationService +import com.kakao.actionbase.engine.service.QueryService +import com.kakao.actionbase.v2.engine.sql.WherePredicate +import com.kakao.actionbase.v2.engine.util.getLogger + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +class TopkAggregationHandler( + private val queryService: QueryService, + private val mutationService: MutationService, + private val queueService: QueueService, + private val engine: AggregationEngine, +) : AggregationHandler { + private val logger = getLogger() + + override val type: AggregationType = AggregationType.TOPK + + // Flatten every (event, direction, topk) into one fan-out. Per-item cardinality is schema-bounded + // (a handful of topks × directions), so concurrency is capped once at the service, not here. + override fun aggregate(item: AggregationItemPayload): Flux = + Flux + .fromIterable( + createTopkEvent(item).flatMap { event -> + event.aggregations.topk.flatMap { topk -> + event.direction.directions().map { direction -> Triple(event, direction, topk) } + } + }, + ).flatMapSequential { (event, direction, topk) -> processTopk(event, direction, topk) } + + private fun createTopkEvent(item: AggregationItemPayload): List { + val tb = engine.getTableBinding(database = item.database, alias = item.table) + + require(tb.schema is ModelSchema.Edge || tb.schema is ModelSchema.MultiEdge) { + "Aggregation is only supported for Edge and MultiEdge tables." + } + + return tb.schema.groups + .filter { it.aggregations.topk.isNotEmpty() } + .map { group -> + EdgeAggregationEvent.of(type = AggregationType.TOPK, database = item.database, table = tb.table, item, group) + } + } + + private fun processTopk( + event: EdgeAggregationEvent, + direction: Direction, + topk: Topk, + ): Mono { + fun result( + status: String, + error: String? = null, + ) = AggregationResult( + database = event.database, + table = event.table, + source = event.source, + target = event.target, + status = status, + error = error, + ) + + val inputs = RankingInputs.from(event, direction, topk) + val (rankDatabase, rankTable) = parseFqn(topk.rank) + val ranking = + Ranking( + database = rankDatabase, + table = rankTable, + topk = topk.topk, + entity = inputs.entity, + topkDimensionValue = inputs.topkDimensionValue, + dimensionValues = inputs.dimensionValues, + properties = inputs.properties, + ) + + return writeRank( + sourceDatabase = event.database, + sourceTable = event.table, + group = event.group.group, + start = inputs.directedSource, + direction = direction, + ranges = inputs.ranges, + ranking = ranking, + ).flatMap { results -> + val status = if (results.any { it.status == ERROR }) ERROR else SUCCESS + if (status == ERROR || topk.refreshAfterMillis <= 0 || !isSlidingWindow(event, inputs.ranges)) { + return@flatMap Mono.just(result(status = status)) + } + + writeRefresh( + event, + ranking, + direction, + inputs.ranges, + refreshAfterMillis = topk.refreshAfterMillis, + refreshQueue = topk.refreshQueue, + ).map { result(status = if (it.results.any { r -> r.status == ERROR }) ERROR else SUCCESS) } + }.onErrorResume { err -> + logger.error("topk aggregate failed for {}.{} topk={}", event.database, event.table, topk.topk, err) + Mono.just(result(ERROR, err.message)) + } + } + + /** + * Runs the aggregation query for one ranking and writes its rank row. + * Used by the aggregate flow, which then enqueues a refresh message + * (which recomputes from an already-resolved [Ranking]). + */ + private fun writeRank( + sourceDatabase: String, + sourceTable: String, + group: String, + start: String, + direction: Direction, + ranges: String?, + ranking: Ranking, + ): Mono> = + queryService + .agg( + database = sourceDatabase, + table = sourceTable, + group = group, + start = listOf(start), + direction = V2Direction.valueOf(direction.name), + ranges = ranges, + ).flatMap { response -> + val metric = response.groups.firstOrNull()?.value ?: 0L + + mutationService + .mutate( + database = ranking.database, + alias = ranking.table, + unresolvedEvents = + listOf( + MutationItem( + type = EventType.INSERT, + edge = + Edge( + version = System.currentTimeMillis(), + source = + AggregationConstants.Topk.rankSource( + database = sourceDatabase, + table = sourceTable, + topk = ranking.topk, + entity = ranking.entity, + dimensionValues = ranking.dimensionValues, + ), + target = ranking.topkDimensionValue, + properties = + buildMap { + put(AggregationConstants.Topk.METRIC, metric) + if (ranking.properties.isNotEmpty()) { + put( + AggregationConstants.Topk.ADDITIONAL_PROPERTIES, + MAPPER.writeValueAsString(ranking.properties), + ) + } + }, + ), + ), + ), + ) + } + + private fun writeRefresh( + event: EdgeAggregationEvent, + ranking: Ranking, + direction: Direction, + ranges: String?, + refreshAfterMillis: Long, + refreshQueue: String, + ): Mono { + val refreshAt = refreshAt(event, refreshAfterMillis) + val message = + EnqueueMessage( + key = + AggregationConstants.Topk.refreshKey( + database = event.database, + table = event.table, + topk = ranking.topk, + entity = ranking.entity, + topkDimensionValue = ranking.topkDimensionValue, + dimensionValues = ranking.dimensionValues, + ), + seq = refreshAt, + value = + TopkRefreshMessage.of( + type = AggregationType.TOPK, + database = event.database, + table = event.table, + topk = ranking.topk, + source = event.source, + target = event.target, + direction = direction.name, + ranges = ranges ?: "", + entity = ranking.entity, + topkDimensionValue = ranking.topkDimensionValue, + dimensionValues = AggregationConstants.Topk.joinValues(ranking.dimensionValues), + properties = ranking.properties, + refreshAt = refreshAt, + ), + ) + + return queueService.enqueue( + namespace = AggregationConstants.Topk.DATABASE, + queue = refreshQueue, + request = EnqueueRequest(messages = listOf(message)), + ) + } + + /** + * The instant this event leaves the window. `now + window` misses it: the event is stored under a bucket, + * the window's bounds are read at the same precision, so the window only moves at bucket boundaries — a + * refresh scheduled off the raw event time fires while the event is still inside, changes nothing, and is + * consumed. Counting from the event's bucket drops the same remainder the write dropped, and the extra + * bucket accounts for the lower bound being truncated rather than raised. + */ + private fun refreshAt( + event: EdgeAggregationEvent, + refreshAfterMillis: Long, + ): Long { + val bucketed = event.dateBucket() + val start = bucketed?.let { (field, bucket) -> bucket.startOf(event.properties[field.name]) } + + return if (start == null) { + System.currentTimeMillis() + refreshAfterMillis + } else { + start.toEpochMilli() + refreshAfterMillis + bucketed.second.interval().toMillis() + } + } + + private fun isSlidingWindow( + event: EdgeAggregationEvent, + ranges: String?, + ): Boolean { + val bucket = event.dateBucket()?.second ?: return false + val bound = + ranges + ?.let { WherePredicate.parse(it) } + ?.filterIsInstance() + ?.firstOrNull { it.key == bucket.name } + ?.from + ?: return false + + return bucket.isRelative(bound) + } + + private fun EdgeAggregationEvent.dateBucket(): Pair? = group.fields.firstNotNullOfOrNull { field -> (field.bucket as? Bucket.Date)?.let { field to it } } + + private fun parseFqn(fqn: String): Pair { + val dot = fqn.indexOf('.') + require(dot > 0 && dot < fqn.lastIndex) { + "rank table must be a fully-qualified `database.table`, got: $fqn" + } + return fqn.substring(0, dot) to fqn.substring(dot + 1) + } + + private companion object { + private val MAPPER = jacksonObjectMapper() + + const val SUCCESS = "SUCCESS" + const val ERROR = "ERROR" + const val SKIPPED = "SKIPPED" + } +} + +private data class Ranking( + val database: String, + val table: String, + val topk: String, + val entity: String, + val topkDimensionValue: String, + val dimensionValues: List, + val properties: Map = emptyMap(), +) + +internal data class RankingInputs( + val directedSource: String, + val entity: String, + val topkDimensionValue: String, + val dimensionValues: List, + val properties: Map, + val ranges: String?, +) { + companion object { + private val PLACEHOLDER = Regex("""\{([a-zA-Z_][a-zA-Z0-9_]*)}""") + + fun from( + event: EdgeAggregationEvent, + direction: Direction, + topk: Topk, + ): RankingInputs { + val directedSource = if (direction == Direction.IN) event.target else event.source + return RankingInputs( + directedSource = directedSource, + entity = if (topk.entity == AggregationConstants.Topk.GLOBAL_ENTITY) AggregationConstants.Topk.GLOBAL_ENTITY else directedSource, + topkDimensionValue = fieldValue(topk.dimension, event), + dimensionValues = event.group.dimensionFields(topk).map { fieldValue(it.name, event) }, + properties = topk.additionalProperties.associateWith { fieldValue(it, event) }, + ranges = topk.ranges.takeIf { it.isNotEmpty() }?.let { interpolate(it, event) }, + ) + } + + private fun fieldValue( + name: String, + event: EdgeAggregationEvent, + ): String = + when (name) { + "source", "_source" -> event.source + "target", "_target" -> event.target + else -> event.properties[name]?.toString().orEmpty() + } + + private fun interpolate( + template: String, + event: EdgeAggregationEvent, + ): String = + PLACEHOLDER.replace(template) { match -> + val value = + when (val key = match.groupValues[1]) { + "source", "_source" -> event.source + "target", "_target" -> event.target + else -> event.properties[key]?.toString() + } + value?.let { Regex.escapeReplacement(it) } ?: Regex.escapeReplacement(match.value) + } + } +} + +data class EdgeAggregationEvent( + val type: AggregationType, + val database: String, + val table: String, + val source: String, + val target: String, + val properties: Map, + val direction: DirectionType, + val group: Group, + val aggregations: Aggregations, +) { + companion object { + fun of( + type: AggregationType, + database: String, + table: String, + item: AggregationItemPayload, + group: Group, + ): EdgeAggregationEvent = + EdgeAggregationEvent( + type = type, + database = database, + table = table, + source = item.edge.source.toString(), + target = item.edge.target.toString(), + properties = item.edge.properties, + direction = group.directionType, + group = group, + aggregations = group.aggregations, + ) + } +} + +data class TopkRefreshMessage( + val type: AggregationType, + val item: RefreshItem, +) { + data class RefreshItem( + 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, + val refreshAt: Long, + ) + + companion object { + fun of( + type: AggregationType, + database: String, + table: String, + topk: String, + source: String, + target: String, + direction: String, + ranges: String, + entity: String, + topkDimensionValue: String, + dimensionValues: String, + properties: Map, + refreshAt: Long, + ): TopkRefreshMessage = + TopkRefreshMessage( + type, + item = + RefreshItem( + database = database, + table = table, + topk = topk, + source = source, + target = target, + direction = direction, + ranges = ranges, + entity = entity, + topkDimensionValue = topkDimensionValue, + dimensionValues = dimensionValues, + properties = properties, + refreshAt = refreshAt, + ), + ) + } +} 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 new file mode 100644 index 000000000..29cf06fd6 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/AggregationServiceTest.kt @@ -0,0 +1,159 @@ +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.EdgePayload +import com.kakao.actionbase.core.metadata.QualifiedAggregations +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.engine.AggregationEngine +import com.kakao.actionbase.engine.service.aggregation.AggregationHandler + +import java.time.Duration + +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test + +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder +import io.mockk.every +import io.mockk.mockk +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +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 + * 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. + */ +class AggregationServiceTest { + private val engine = mockk() + + @Nested + inner class GetAggregations { + private val service = AggregationService(engine, emptyList()) + private val entry = QualifiedAggregations(type = AggregationType.TOPK, database = "commerce", table = "orders") + + @Test + fun `forwards results from the engine`() { + every { engine.getListWithAggregations(null) } returns listOf(entry) + + service.getAggregations() shouldContainExactlyInAnyOrder listOf(entry) + } + + @Test + fun `forwards the type filter to the engine`() { + every { engine.getListWithAggregations(AggregationType.TOPK) } returns listOf(entry) + + service.getAggregations(AggregationType.TOPK) shouldContainExactlyInAnyOrder listOf(entry) + } + } + + @Nested + inner class Aggregate { + @Test + fun `completes with an empty list when given no items`() { + val service = AggregationService(engine, listOf(DelayedHandler())) + + StepVerifier + .create(service.aggregate(items = emptyList())) + .assertNext { results -> results shouldContainExactly emptyList() } + .verifyComplete() + } + + /** + * `AggregationResult` carries no field naming the item it came from, so a caller can only pair a + * result back to its request by position. That holds only if the dispatcher emits in request order — + * otherwise the same batch returns differently shaped responses depending on how fast each handler + * happened to answer. The delays here stand in for that variance: the slowest item is requested first. + */ + @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.aggregate(items = listOf(item("user1"), item("user2"), item("user3"))) } + .expectSubscription() + .thenAwait(Duration.ofMinutes(1)) + .assertNext { results -> results.map { it.source } shouldContainExactly listOf("user1", "user2", "user3") } + .verifyComplete() + } + + /** + * Handlers arrive as an injected `List` (`GraphConfiguration`), and the dispatcher + * keys them by type. Two beans claiming one type is a wiring mistake, and the map resolves it by + * silently keeping the last — pinned here so that changing how it resolves stays a deliberate choice. + */ + @Test + fun `keeps only the last handler registered for a type`() { + val service = AggregationService(engine, listOf(DelayedHandler(tag = "first"), DelayedHandler(tag = "second"))) + + StepVerifier + .create(service.aggregate(items = listOf(item("user1")))) + .assertNext { results -> results.map { it.table } shouldContainExactly listOf("second") } + .verifyComplete() + } + } +} + +// region test fixtures + +private fun item(source: String): AggregationItemPayload = + AggregationItemPayload( + database = "commerce", + table = "orders", + edge = + EdgePayload( + version = 1L, + source = source, + target = "item1", + properties = emptyMap(), + context = emptyMap(), + ), + ) + +/** + * 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 + * dispatcher's interleaving never becomes observable. The ordering tests drive this under + * `withVirtualTime`, so the delays cost no wall-clock time. + */ +private class DelayedHandler( + private val tag: String = "orders", + private val delays: Map = emptyMap(), +) : AggregationHandler { + override val type: AggregationType = AggregationType.TOPK + + override fun aggregate(item: AggregationItemPayload): Flux { + val source = item.edge.source.toString() + + return Mono + .just( + AggregationResult( + database = item.database, + table = tag, + source = source, + target = item.edge.target.toString(), + status = "SUCCESS", + error = null, + ), + ).delayElement(delays[source] ?: Duration.ZERO) + .flux() + } +} + +// endregion diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/RankingInputsTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/RankingInputsTest.kt new file mode 100644 index 000000000..c66dfb484 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/RankingInputsTest.kt @@ -0,0 +1,148 @@ +package com.kakao.actionbase.engine.service.aggregation + +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.core.metadata.common.Aggregations +import com.kakao.actionbase.core.metadata.common.Bucket +import com.kakao.actionbase.core.metadata.common.Direction +import com.kakao.actionbase.core.metadata.common.DirectionType +import com.kakao.actionbase.core.metadata.common.Group +import com.kakao.actionbase.core.metadata.common.GroupType +import com.kakao.actionbase.core.metadata.common.Topk + +import org.junit.jupiter.api.Test + +import io.kotest.matchers.collections.shouldBeEmpty +import io.kotest.matchers.shouldBe + +class RankingInputsTest { + @Test + fun `OUT ranks the source, with the dimension value as the target`() { + val inputs = RankingInputs.from(event = event(source = "user1", target = "item1"), direction = Direction.OUT, topk = topk(entity = "source", dimension = "target")) + + inputs.entity shouldBe "user1" + inputs.topkDimensionValue shouldBe "item1" + inputs.dimensionValues.shouldBeEmpty() + } + + @Test + fun `IN ranks the target, with the source as the target`() { + val inputs = RankingInputs.from(event = event(source = "user1", target = "item1"), direction = Direction.IN, topk = topk(entity = "target", dimension = "source")) + + inputs.entity shouldBe "item1" + inputs.topkDimensionValue shouldBe "user1" + inputs.dimensionValues.shouldBeEmpty() + } + + @Test + fun `the global sentinel entity is kept as-is`() { + val inputs = RankingInputs.from(event = event(source = "user1"), direction = Direction.OUT, topk = topk(entity = "__GLOBAL__")) + + inputs.entity shouldBe "__GLOBAL__" + } + + @Test + fun `bucket fields are excluded from the dimension values`() { + val inputs = + RankingInputs.from( + event = + event( + properties = mapOf("category" to "fruit", "purchasedAt" to 1_700_000_000_000L), + fields = listOf(Group.Field(name = "category"), Group.Field(name = "purchasedAt", bucket = day())), + ), + direction = Direction.OUT, + topk = topk(dimension = "target"), + ) + + inputs.dimensionValues shouldBe listOf("fruit") + } + + @Test + fun `a property-backed dimension becomes the target and drops out of the dimension values`() { + val inputs = + RankingInputs.from( + event = + event( + properties = mapOf("category" to "fruit", "purchasedAt" to 1_700_000_000_000L), + fields = listOf(Group.Field(name = "category"), Group.Field(name = "purchasedAt", bucket = day())), + ), + direction = Direction.OUT, + topk = topk(dimension = "category"), + ) + + inputs.topkDimensionValue shouldBe "fruit" + inputs.dimensionValues.shouldBeEmpty() + } + + @Test + fun `multiple non-bucket fields join the dimension values in order`() { + val inputs = + RankingInputs.from( + event = + event( + properties = mapOf("category" to "fruit", "region" to "seoul", "purchasedAt" to 1_700_000_000_000L), + fields = + listOf( + Group.Field(name = "_target"), + Group.Field(name = "category"), + Group.Field(name = "region"), + Group.Field(name = "purchasedAt", bucket = day()), + ), + ), + direction = Direction.OUT, + topk = topk(dimension = "target"), + ) + + inputs.dimensionValues shouldBe listOf("fruit", "seoul") + } + + @Test + fun `declared properties resolve from endpoints and edge properties`() { + val inputs = + RankingInputs.from( + event = event(target = "item1", properties = mapOf("category" to "fruit")), + direction = Direction.OUT, + topk = topk(additionalProperties = listOf("category", "target")), + ) + + inputs.properties shouldBe mapOf("category" to "fruit", "target" to "item1") + } + + @Test + fun `ranges placeholders are interpolated from the edge`() { + val inputs = RankingInputs.from(event = event(target = "item1"), direction = Direction.OUT, topk = topk(ranges = "_target:eq:{_target}")) + + inputs.ranges shouldBe "_target:eq:item1" + } +} + +// region test fixtures + +private fun event( + source: String = "user1", + target: String = "item1", + properties: Map = emptyMap(), + fields: List = emptyList(), +): EdgeAggregationEvent = + EdgeAggregationEvent( + type = AggregationType.TOPK, + database = "commerce", + table = "orders", + source = source, + target = target, + properties = properties, + direction = DirectionType.BOTH, + group = Group(group = "purchased_count", type = GroupType.SUM, fields = fields, directionType = DirectionType.BOTH, aggregations = Aggregations(topk = emptyList())), + aggregations = Aggregations(topk = emptyList()), + ) + +// `refreshAfterMillis` and `rank` do not affect resolution, so they stay defaulted +private fun topk( + entity: String = "source", + dimension: String = "target", + ranges: String = "", + additionalProperties: List = emptyList(), +): Topk = Topk(topk = "top_purchased", entity = entity, dimension = dimension, ranges = ranges, rank = "commerce.orders__topk", additionalProperties = additionalProperties) + +private fun day(): Bucket.Date = Bucket.Date(name = "day", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd") + +// 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 new file mode 100644 index 000000000..f54d4fc49 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/service/aggregation/TopkAggregationHandlerTest.kt @@ -0,0 +1,596 @@ +package com.kakao.actionbase.engine.service.aggregation + +import com.kakao.actionbase.core.edge.MutationKey +import com.kakao.actionbase.core.edge.payload.AggregationItemPayload +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.metadata.common.AggregationConstants +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.core.metadata.common.Aggregations +import com.kakao.actionbase.core.metadata.common.Bucket +import com.kakao.actionbase.core.metadata.common.DirectionType +import com.kakao.actionbase.core.metadata.common.Field +import com.kakao.actionbase.core.metadata.common.Group +import com.kakao.actionbase.core.metadata.common.GroupType +import com.kakao.actionbase.core.metadata.common.ModelSchema +import com.kakao.actionbase.core.metadata.common.Topk +import com.kakao.actionbase.core.types.PrimitiveType +import com.kakao.actionbase.engine.AggregationEngine +import com.kakao.actionbase.engine.binding.TableBinding +import com.kakao.actionbase.engine.queue.EnqueueRequest +import com.kakao.actionbase.engine.queue.EnqueueResponse +import com.kakao.actionbase.engine.queue.EnqueueResult +import com.kakao.actionbase.engine.queue.QueueService +import com.kakao.actionbase.engine.service.MutationService +import com.kakao.actionbase.engine.service.QueryService + +import java.time.Instant + +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test + +import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder +import io.kotest.matchers.collections.shouldHaveSize +import io.kotest.matchers.longs.shouldBeGreaterThanOrEqual +import io.kotest.matchers.longs.shouldBeLessThanOrEqual +import io.kotest.matchers.shouldBe +import io.mockk.every +import io.mockk.mockk +import io.mockk.slot +import io.mockk.verify +import reactor.core.publisher.Mono +import reactor.test.StepVerifier + +class TopkAggregationHandlerTest { + private val queryService = mockk() + private val mutationService = mockk() + private val queueService = mockk() + private val engine = mockk() + private val handler = TopkAggregationHandler(queryService, mutationService, queueService, engine) + + @Nested + inner class Aggregate { + @Test + fun `returns SUCCESS when mutate succeeds`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 42)) + + every { + mutationService.mutate(any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload(source = "user1", target = "item1")).collectList()) + .assertNext { results -> + results shouldHaveSize 1 + results[0].status shouldBe "SUCCESS" + results[0].error shouldBe null + }.verifyComplete() + } + + @Test + fun `returns ERROR when the mutate fails`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + + every { + mutationService.mutate(any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "ERROR"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { results -> + results shouldHaveSize 1 + results[0].status shouldBe "ERROR" + results[0].error shouldBe null + }.verifyComplete() + } + + @Test + fun `BOTH fans out into an OUT and an IN mutation`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "top_both"), directionType = DirectionType.BOTH) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 5)) + + val mutations = mutableListOf>() + every { + mutationService.mutate(any(), any(), capture(mutations), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload(source = "user1", target = "item1")).collectList()) + .assertNext { results -> + results shouldHaveSize 2 + }.verifyComplete() + + val edges = mutations.map { it.single().edge } + edges.map { it.source to it.target } shouldContainExactlyInAnyOrder + listOf( + "commerce|orders|top_both|user1" to "item1", + "commerce|orders|top_both|item1" to "item1", + ) + } + + @Test + fun `carries the declared properties onto the rank row`() { + stubTopkBinding( + engine = engine, + topk = topkConfig(name = "top_purchased", additionalProperties = listOf("category", "target")), + ) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 4)) + + 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 + .aggregate(aggregationItemPayload(source = "user1", target = "item1", properties = mapOf("category" to "fruit"))) + .collectList(), + ).assertNext { it.single().status shouldBe "SUCCESS" } + .verifyComplete() + + val edge = mutations.captured.single().edge + edge.properties.containsKey("metric") shouldBe true + edge.properties["additionalProperties"] shouldBe """{"category":"fruit","target":"item1"}""" + } + + @Test + fun `enqueues a refresh after the rank write`() { + val refreshAfter = 60_000L + stubTopkBinding(engine = engine, topk = topkConfig(name = "top_purchased", refreshAfterMillis = refreshAfter, ranges = SLIDING_WINDOW), bucketed = true) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 4)) + + val rankMutations = slot>() + every { + mutationService.mutate("commerce", "orders__topk", capture(rankMutations), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + val refreshRequest = slot() + every { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + capture(refreshRequest), + ) + } returns Mono.just(enqueueResponse(status = "CREATED")) + + val before = System.currentTimeMillis() + StepVerifier + .create(handler.aggregate(aggregationItemPayload(source = "user1", target = "item1")).collectList()) + .assertNext { results -> + results shouldHaveSize 1 + results[0].status shouldBe "SUCCESS" + }.verifyComplete() + val after = System.currentTimeMillis() + + rankMutations.captured + .single() + .edge.source shouldBe "commerce|orders|top_purchased|user1" + rankMutations.captured + .single() + .edge.target shouldBe "item1" + + val message = refreshRequest.captured.messages.single() + message.key shouldBe + AggregationConstants.Topk.refreshKey( + database = "commerce", + table = "orders", + topk = "top_purchased", + entity = "user1", + topkDimensionValue = "item1", + dimensionValues = emptyList(), + ) + val refreshAt = message.seq + refreshAt shouldBeGreaterThanOrEqual (before + refreshAfter) + refreshAt shouldBeLessThanOrEqual (after + refreshAfter) + + val refresh = message.value as TopkRefreshMessage + refresh.type shouldBe AggregationType.TOPK + refresh.item.database shouldBe "commerce" + refresh.item.table shouldBe "orders" + refresh.item.topk shouldBe "top_purchased" + refresh.item.source shouldBe "user1" + refresh.item.target shouldBe "item1" + refresh.item.direction shouldBe "OUT" + refresh.item.entity shouldBe "user1" + refresh.item.topkDimensionValue shouldBe "item1" + refresh.item.dimensionValues shouldBe "" + refresh.item.refreshAt shouldBe refreshAt + } + + /** + * The queue decides which sweeper picks the refresh up, so a ranking that names one has to be + * enqueued there and nowhere else. Every other test here names none and asserts the shared queue, + * which is the other half of the contract. + */ + @Test + fun `a refresh goes onto the queue the declaration names`() { + stubTopkBinding( + engine = engine, + topk = + topkConfig( + name = "top_purchased", + refreshAfterMillis = 60_000L, + refreshQueue = "purchase_refresh", + ranges = SLIDING_WINDOW, + ), + bucketed = true, + ) + + every { + queryService.agg(any(), any(), any(), any(), 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"))) + + val refreshRequest = slot() + every { + queueService.enqueue(AggregationConstants.Topk.DATABASE, "purchase_refresh", capture(refreshRequest)) + } returns Mono.just(enqueueResponse(status = "CREATED")) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload(source = "user1", target = "item1")).collectList()) + .assertNext { results -> results.single().status shouldBe "SUCCESS" } + .verifyComplete() + + refreshRequest.captured.messages + .single() + .key shouldBe + AggregationConstants.Topk.refreshKey( + database = "commerce", + table = "orders", + topk = "top_purchased", + entity = "user1", + topkDimensionValue = "item1", + dimensionValues = emptyList(), + ) + + verify(exactly = 0) { + queueService.enqueue(any(), AggregationConstants.Topk.REFRESH_QUEUE, any()) + } + } + + @Test + fun `carries the declared properties into the refresh message`() { + stubTopkBinding( + engine = engine, + topk = topkConfig(name = "top_purchased", refreshAfterMillis = 60_000L, additionalProperties = listOf("category"), ranges = SLIDING_WINDOW), + bucketed = true, + ) + + every { + queryService.agg(any(), any(), any(), any(), 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"))) + + val refreshRequest = slot() + every { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + capture(refreshRequest), + ) + } returns Mono.just(enqueueResponse(status = "CREATED")) + + StepVerifier + .create( + handler + .aggregate(aggregationItemPayload(source = "user1", target = "item1", properties = mapOf("category" to "fruit"))) + .collectList(), + ).assertNext { it.single().status shouldBe "SUCCESS" } + .verifyComplete() + + val refresh = + refreshRequest.captured.messages + .single() + .value as TopkRefreshMessage + refresh.item.properties shouldBe mapOf("category" to "fruit") + } + + @Test + fun `skips the refresh when refreshAfterMillis is not positive`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { it.single().status shouldBe "SUCCESS" } + .verifyComplete() + + verify(exactly = 0) { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + any(), + ) + } + } + + @Test + fun `skips the refresh when the window does not slide`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk", refreshAfterMillis = 60_000L, ranges = FIXED_WINDOW), bucketed = true) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { it.single().status shouldBe "SUCCESS" } + .verifyComplete() + + verify(exactly = 0) { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + any(), + ) + } + } + + @Test + fun `skips the refresh when the rank write fails`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk", refreshAfterMillis = 60_000L, ranges = SLIDING_WINDOW), bucketed = true) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "ERROR"))) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { it.single().status shouldBe "ERROR" } + .verifyComplete() + + verify(exactly = 0) { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + any(), + ) + } + } + + @Test + fun `schedules the refresh for the instant the event leaves the window`() { + val refreshAfter = 365L * 24 * 60 * 60 * 1000 + // 14:32 on a day bucket is stored as that day, so the remainder must not carry into the due time. + val purchasedAt = Instant.parse("2026-01-01T14:32:00Z").toEpochMilli() + + stubTopkBinding( + engine = engine, + topk = topkConfig(name = "top_purchased", refreshAfterMillis = refreshAfter, ranges = SLIDING_WINDOW), + bucketed = true, + ) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + val refreshRequest = slot() + every { + queueService.enqueue(AggregationConstants.Topk.DATABASE, AggregationConstants.Topk.REFRESH_QUEUE, capture(refreshRequest)) + } returns Mono.just(enqueueResponse(status = "CREATED")) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload(properties = mapOf(PURCHASED_AT to purchasedAt))).collectList()) + .assertNext { results -> results.single().status shouldBe "SUCCESS" } + .verifyComplete() + + // The purchase's day stays inside a `now-365d` window until the day after it turns 365 days old. + refreshRequest.captured.messages + .single() + .seq shouldBe Instant.parse("2027-01-02T00:00:00Z").toEpochMilli() + } + + @Test + fun `schedules from the field's value when the bucket carries a name of its own`() { + val refreshAfter = 365L * 24 * 60 * 60 * 1000 + val purchasedAt = Instant.parse("2026-01-01T14:32:00Z").toEpochMilli() + + stubTopkBinding( + engine = engine, + topk = topkConfig(name = "top_purchased", refreshAfterMillis = refreshAfter, ranges = ALIASED_SLIDING_WINDOW), + bucketed = true, + bucket = ALIASED_DAY_BUCKET, + ) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + val refreshRequest = slot() + every { + queueService.enqueue(AggregationConstants.Topk.DATABASE, AggregationConstants.Topk.REFRESH_QUEUE, capture(refreshRequest)) + } returns Mono.just(enqueueResponse(status = "CREATED")) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload(properties = mapOf(PURCHASED_AT to purchasedAt))).collectList()) + .assertNext { results -> results.single().status shouldBe "SUCCESS" } + .verifyComplete() + + refreshRequest.captured.messages + .single() + .seq shouldBe Instant.parse("2027-01-02T00:00:00Z").toEpochMilli() + } + + @Test + fun `reports ERROR when the refresh enqueue fails`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk", refreshAfterMillis = 60_000L, ranges = SLIDING_WINDOW), bucketed = true) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.just(aggPayload(count = 1)) + + every { + mutationService.mutate("commerce", "orders__topk", any(), any(), any(), any(), any()) + } returns Mono.just(listOf(mutationResult(status = "CREATED"))) + + every { + queueService.enqueue( + AggregationConstants.Topk.DATABASE, + AggregationConstants.Topk.REFRESH_QUEUE, + any(), + ) + } returns Mono.just(enqueueResponse(status = "ERROR")) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { it.single().status shouldBe "ERROR" } + .verifyComplete() + } + + @Test + fun `maps a thrown error to ERROR with its message`() { + stubTopkBinding(engine = engine, topk = topkConfig(name = "topk")) + + every { + queryService.agg(any(), any(), any(), any(), any(), any(), any(), any()) + } returns Mono.error(RuntimeException("agg boom")) + + StepVerifier + .create(handler.aggregate(aggregationItemPayload()).collectList()) + .assertNext { results -> + results shouldHaveSize 1 + results[0].status shouldBe "ERROR" + results[0].error shouldBe "agg boom" + }.verifyComplete() + } + } +} + +// region test fixtures + +private const val PURCHASED_AT = "purchasedAt" + +private val DAY_BUCKET = Bucket.Date(name = PURCHASED_AT, unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd") + +private const val SLIDING_WINDOW = "$PURCHASED_AT:bt:now-365d,now" + +/** The same field bounded by fixed points rather than by the clock, so the window never moves. */ +private const val FIXED_WINDOW = "$PURCHASED_AT:bt:2026-01-01,2026-12-31" + +/** The same bucket, named for the value it yields rather than for the field it reads. */ +private val ALIASED_DAY_BUCKET = Bucket.Date(name = "time", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd") + +private const val ALIASED_SLIDING_WINDOW = "time:bt:now-365d,now" + +// rank table naming rule: the source table name with the `__topk` suffix +private fun topkConfig( + name: String, + entity: String = "source", + dimension: String = "target", + rank: String = "commerce.orders__topk", + refreshAfterMillis: Long = 0, + refreshQueue: String = AggregationConstants.Topk.REFRESH_QUEUE, + additionalProperties: List = emptyList(), + ranges: String = "", +): Topk = + Topk( + topk = name, + entity = entity, + dimension = dimension, + rank = rank, + refreshAfterMillis = refreshAfterMillis, + refreshQueue = refreshQueue, + additionalProperties = additionalProperties, + ranges = ranges, + ) + +private fun stubTopkBinding( + engine: AggregationEngine, + topk: Topk, + database: String = "commerce", + table: String = "orders", + directionType: DirectionType = DirectionType.OUT, + bucketed: Boolean = false, + bucket: Bucket.Date = DAY_BUCKET, +) { + val group = + Group( + group = "purchased_count", + type = GroupType.SUM, + fields = if (bucketed) listOf(Group.Field(PURCHASED_AT, bucket = bucket)) else emptyList(), + directionType = directionType, + aggregations = Aggregations(topk = listOf(topk)), + ) + val binding = mockk() + every { binding.schema } returns ModelSchema.Edge(source = Field(type = PrimitiveType.STRING, comment = ""), target = Field(type = PrimitiveType.STRING, comment = ""), direction = DirectionType.BOTH, groups = listOf(group)) + every { binding.table } returns table + every { engine.getTableBinding(database = database, alias = table) } returns binding +} + +private fun aggregationItemPayload( + database: String = "commerce", + table: String = "orders", + source: String = "user1", + target: String = "item1", + properties: Map = emptyMap(), +): AggregationItemPayload = + AggregationItemPayload( + database = database, + table = table, + edge = + EdgePayload( + version = 1L, + source = source, + target = target, + properties = properties, + context = emptyMap(), + ), + ) + +private fun aggPayload(count: Int): DataFrameEdgeAggPayload = DataFrameEdgeAggPayload(groups = emptyList(), count = count, context = emptyMap()) + +private fun mutationResult(status: String): MutationResult = MutationResult.of(key = MutationKey.SourceTarget("s", "t"), count = 1, status = status) + +private fun enqueueResponse(status: String): EnqueueResponse = + EnqueueResponse( + accepted = if (status == "CREATED") 1 else 0, + results = listOf(EnqueueResult(partition = 0, id = "01", status = status)), + ) + +// endregion 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 055b3e70c..6ad9a11bc 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,14 +1,20 @@ package com.kakao.actionbase.server.api.graph.v3.metadata +import com.kakao.actionbase.core.edge.payload.AggregationItemRequest +import com.kakao.actionbase.core.edge.payload.AggregationsItemResponse import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.metadata.payload.AggregationsListResponse import com.kakao.actionbase.engine.service.AggregationService import org.springframework.http.ResponseEntity import org.springframework.web.bind.annotation.GetMapping +import org.springframework.web.bind.annotation.PostMapping +import org.springframework.web.bind.annotation.RequestBody import org.springframework.web.bind.annotation.RequestParam import org.springframework.web.bind.annotation.RestController +import reactor.core.publisher.Mono + @RestController class MetadataAggController( private val aggregationService: AggregationService, @@ -21,4 +27,12 @@ class MetadataAggController( return ResponseEntity.ok(AggregationsListResponse.of(aggregations)) } + + @PostMapping("/aggregations/v1/aggregate") + fun aggregate( + @RequestBody request: AggregationItemRequest, + ): Mono> = + aggregationService + .aggregate(items = request.items) + .map { results -> ResponseEntity.ok(AggregationsItemResponse.from(results)) } } diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/GraphConfiguration.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/GraphConfiguration.kt index edd71244b..23fb8afdf 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/GraphConfiguration.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/GraphConfiguration.kt @@ -5,6 +5,8 @@ import com.kakao.actionbase.engine.queue.QueueService import com.kakao.actionbase.engine.service.AggregationService import com.kakao.actionbase.engine.service.MutationService import com.kakao.actionbase.engine.service.QueryService +import com.kakao.actionbase.engine.service.aggregation.AggregationHandler +import com.kakao.actionbase.engine.service.aggregation.TopkAggregationHandler import com.kakao.actionbase.server.client.kafka.SpringKafkaClientFactory import com.kakao.actionbase.server.client.web.SpringWebClientFactory import com.kakao.actionbase.v2.engine.Graph @@ -145,7 +147,18 @@ class GraphConfiguration { fun provideQueryService(engine: V2BackedEngine): QueryService = QueryService(engine) @Bean - fun provideAggregationService(engine: V2BackedEngine): AggregationService = AggregationService(engine) + fun provideTopkAggregationHandler( + queryService: QueryService, + mutationService: MutationService, + queueService: QueueService, + engine: V2BackedEngine, + ): AggregationHandler = TopkAggregationHandler(queryService, mutationService, queueService, engine) + + @Bean + fun provideAggregationService( + engine: V2BackedEngine, + handlers: List, + ): AggregationService = AggregationService(engine, handlers) @Bean fun provideMutationService( 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 new file mode 100644 index 000000000..dcfcfa57b --- /dev/null +++ b/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggControllerE2ETest.kt @@ -0,0 +1,289 @@ +package com.kakao.actionbase.server.api.graph.v3.metadata + +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 +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.core.metadata.payload.AggregationsListResponse +import com.kakao.actionbase.engine.queue.PartitionHasher +import com.kakao.actionbase.engine.queue.PollResponse +import com.kakao.actionbase.server.test.E2ETestBase + +import java.time.Duration +import java.time.Instant +import java.time.temporal.ChronoUnit + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.TestInstance +import org.springframework.http.MediaType +import org.springframework.test.web.reactive.server.expectBody + +/** + * E2E for `POST /aggregations/v1/aggregate`, which materializes per-entity top-K rankings from an edge + * event into two tables. + * + * Rank table `.__topk` — a plain edge table read as a sorted index via `metric_desc`: + * + * | row key (source) | target | value | + * |-------------------------------------|--------------------|--------| + * | topk | entity | dimensionValues... | topkDimensionValue | metric | + * + * Refresh queue `topk`/`refresh` — a queue/v1 table the aggregate flow enqueues onto when a top-K + * declares `refreshAfterMillis`: + * + * | partition | key | seq | value | + * |---------------|-------------------------------------|---------------|--------------------| + * | hash(key) % N | db|table|topk|entity|dimVal|dims... | refreshAt(ms) | {type, item:{ … }} | + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class MetadataAggControllerE2ETest : E2ETestBase() { + private val db = "commerce" + private val table = "purchases" + private val rank = "${table}__topk" + private val rankFqn = "$db.$rank" + private val refreshPartitions = 4 + private val refreshAfter = 3_600_000L + + @BeforeAll + fun setup() { + client + .post() + .uri("/graph/v3/databases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"database": "$db", "comment": "test"}""") + .exchange() + .expectStatus() + .isOk + + createRankTable(rank) + + client + .post() + .uri("/graph/v3/databases/$db/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$table", + "schema": { + "type": "MULTI_EDGE", + "id": {"type": "long", "comment": "purchase id"}, + "source": {"type": "string", "comment": "user"}, + "target": {"type": "string", "comment": "item"}, + "properties": [ + {"name": "purchasedAt", "type": "long", "comment": "purchase time ms", "nullable": false} + ], + "direction": "BOTH", + "indexes": [], + "groups": [{ + "group": "purchased_count", + "type": "COUNT", + "fields": [ + {"name": "_target"}, + {"name": "purchasedAt", "bucket": {"type": "date", "name": "purchasedAt", "unit": "MILLISECOND", "timezone": "UTC", "format": "yyyy-MM-dd"}} + ], + "directionType": "OUT", + "aggregations": { + "topk": [{ + "topk": "top_purchased", + "entity": "source", + "ranges": "_target:eq:{_target};purchasedAt:bt:now-365d,now", + "dimension": "target", + "refreshAfterMillis": $refreshAfter, + "rank": "$rankFqn" + }] + } + }], + "caches": [] + }, + "storage": "datastore://test_namespace/$table", + "mode": "SYNC", + "comment": "user-to-item purchase edges" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/graph/v3/databases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"database": "$DATABASE", "comment": "test"}""") + .exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/queue/v1/namespaces/$DATABASE/queues") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """{"queue": "$REFRESH_QUEUE", "storage": "datastore://test_namespace/$REFRESH_QUEUE", "partitions": $refreshPartitions}""", + ).exchange() + .expectStatus() + .isOk + } + + /** + * The listing endpoint surfaces every table that declares a top-K aggregation, keyed by + * (database, table). + */ + @Test + fun `listing aggregations returns the tables that declare a topk aggregation`() { + val response = + client + .get() + .uri("/aggregations/v1/metadata") + .exchange() + .expectStatus() + .isOk + .expectBody() + .returnResult() + .responseBody!! + + val byLocation = response.items.associateBy { it.database to it.table } + + val source = byLocation[db to table] + assertNotNull(source) + assertEquals(AggregationType.TOPK, source?.type) + assertEquals(db, source?.database) + assertEquals(table, source?.table) + + assertEquals(listOf("source", "_target"), source?.dedupeFields) + } + + /** + * The `commerce.purchases` top-K enables `refreshAfterMillis`, so an aggregation also enqueues a + * refresh message. + * + * 1. Record one edge user1 -> item1 on `commerce.purchases`. + * 2. Run the aggregation. Besides the rank row it enqueues one message onto the refresh queue, + * keyed by the same composite the rank row uses and ordered by `seq` = refreshAt: + * + * topk/refresh (queue; poll partition = hash(key) % 4) + * | seq (refreshAt) | value.type | value.item | + * |--------------------------|------------|---------------------------------------------| + * | now + refreshAfterMillis | TOPK | {database, table, topk, source, target, … } | + * + * 3. Poll that partition -> exactly one message carrying the full recompute payload. + */ + @Test + fun `running a topk aggregation enqueues a refresh message when refreshAfterMillis is set`() { + val purchasedAt = System.currentTimeMillis() + + client + .post() + .uri("/graph/v3/databases/$db/tables/$table/multi-edges") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "id": 1, "source": "user1", "target": "item1", "properties": {"purchasedAt": $purchasedAt}}} + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/aggregations/v1/aggregate") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "items": [ + {"database": "$db", "table": "$table", + "edge": {"version": 1, "source": "user1", "target": "item1", "properties": {"purchasedAt": $purchasedAt}, "context": {}}} + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + val key = + AggregationConstants.Topk.refreshKey( + database = db, + table = table, + topk = "top_purchased", + entity = "user1", + topkDimensionValue = "item1", + dimensionValues = emptyList(), + ) + val partition = PartitionHasher.partition(key, refreshPartitions) + + val page = + client + .get() + .uri("/queue/v1/namespaces/$DATABASE/queues/$REFRESH_QUEUE/partitions/$partition/poll?limit=10") + .exchange() + .expectStatus() + .isOk + .expectBody() + .returnResult() + .responseBody!! + + // The refresh is due when the purchase leaves the window, not a window's length after the purchase + // itself: the window moves in whole days, so the day it landed in stays inside for one more. + val bucketStart = Instant.ofEpochMilli(purchasedAt).truncatedTo(ChronoUnit.DAYS).toEpochMilli() + + val message = page.messages.single() + assertEquals(bucketStart + refreshAfter + Duration.ofDays(1).toMillis(), message.seq) + + val value = message.value as Map<*, *> + assertEquals("TOPK", value["type"]) + val refreshItem = value["item"] as Map<*, *> + assertEquals(db, refreshItem["database"]) + assertEquals(table, refreshItem["table"]) + assertEquals("top_purchased", refreshItem["topk"]) + assertEquals("user1", refreshItem["source"]) + assertEquals("item1", refreshItem["target"]) + assertEquals("OUT", refreshItem["direction"]) + assertEquals("user1", refreshItem["entity"]) + assertEquals("item1", refreshItem["topkDimensionValue"]) + assertEquals("", refreshItem["dimensionValues"]) + assertEquals(message.seq, (refreshItem["refreshAt"] as Number).toLong()) + } + + /** A rank table is a plain edge table read as a sorted index: `source` is the composite key, + * `target` is the ranked dimension value, and the `metric_desc` index orders by `metric`. */ + private fun createRankTable(rankTable: String) { + client + .post() + .uri("/graph/v3/databases/$db/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$rankTable", + "schema": { + "type": "EDGE", + "source": {"type": "string", "comment": "topk|entity|dimensionValues"}, + "target": {"type": "string", "comment": "topkDimensionValue"}, + "properties": [ + {"name": "metric", "type": "long", "comment": "aggregated metric", "nullable": false} + ], + "direction": "OUT", + "indexes": [ + {"index": "metric_desc", "fields": [{"field": "metric", "order": "DESC"}]} + ], + "groups": [], + "caches": [] + }, + "storage": "datastore://test_namespace/$rankTable", + "mode": "SYNC", + "comment": "topk rank rows" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + } +} 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 0fd527aaa..eca7b3473 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 @@ -228,6 +228,7 @@ class ReadOnlyRequestFilterTest { "DELETE /graph/v3/databases/{database}/aliases/{alias}", "DELETE /graph/v3/databases/{database}/tables/{table}", "DELETE /graph/v3/databases/{database}/tables/{table}/edges/scan/{index}", + "POST /aggregations/v1/aggregate", "POST /graph/v3/databases", "POST /graph/v3/databases/{database}/aliases", "POST /graph/v3/databases/{database}/tables",