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

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<AggregationItemPayload>,
+)
+
+data class AggregationItemPayload(
+    val database: String,
+    val table: String,
+    val edge: EdgePayload,
+)

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?,
+)

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<Item>,
+) {
+    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<AggregationResult>): 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,
+                        )
+                    },
+            )
+    }
+}

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>,
-        ): 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>,
-        ): 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>): 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<String> {
+            if (joined.isEmpty()) return emptyList()
+
+            val values = mutableListOf<String>()
+            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 = '\\'
     }
 }

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 {

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<String>(), 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", ""))))
+    }
+}

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<IllegalArgumentException> { second.handleQueryValue("now-1h", ceil = true) }
+    }
+}

engine/src/main/kotlin/com/kakao/actionbase/engine/service/AggregationFunctions.kt

@@ -0,0 +1,9 @@
+package com.kakao.actionbase.engine.service
+
+internal fun parseFqn(fqn: String): Pair<String, String> {
+    val dot = fqn.indexOf('.')
+    require(dot > 0 && dot < fqn.lastIndex) {
+        "table must be a fully-qualified `database.table`, got: $fqn"
+    }
+    return fqn.substring(0, dot) to fqn.substring(dot + 1)
+}