Declare a top-K aggregation on a group and list what declares one (Steps 1 and 2 of #386)

core/src/main/kotlin/com/kakao/actionbase/core/metadata/QualifiedAggregations.kt

@@ -0,0 +1,10 @@
+package com.kakao.actionbase.core.metadata
+
+import com.kakao.actionbase.core.metadata.common.AggregationType
+
+data class QualifiedAggregations(
+    val type: AggregationType,
+    val database: String,
+    val table: String,
+    val dedupeFields: List<String> = emptyList(),
+)

core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/AggregationConstants.kt

@@ -0,0 +1,38 @@
+package com.kakao.actionbase.core.metadata.common
+
+object AggregationConstants {
+    object Topk {
+        const val DATABASE = "topk"
+        const val REFRESH_QUEUE = "refresh"
+
+        // entity sentinel for global (non per-entity) rankings
+        const val GLOBAL_ENTITY = "__GLOBAL__"
+
+        // rank table index: rows sorted by `metric` descending, i.e. top-K read order
+        const val RANK_INDEX = "metric_desc"
+
+        // rank row properties: the aggregated metric, plus the carried properties as one JSON string
+        const val METRIC = "metric"
+        const val ADDITIONAL_PROPERTIES = "additionalProperties"
+
+        // rank table src key: database | table | topk | entity | dimensionValue1 | dimensionValue2 | ...
+        fun rankSource(
+            database: String,
+            table: String,
+            topk: String,
+            entity: String,
+            dimensionValues: List<String>,
+        ): String = (listOf(database, table, topk, entity) + dimensionValues).joinToString("|")
+
+        // 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).
+        fun refreshKey(
+            database: String,
+            table: String,
+            topk: String,
+            entity: String,
+            topkDimensionValue: String,
+            dimensionValues: List<String>,
+        ): String = (listOf(database, table, topk, entity, topkDimensionValue) + dimensionValues).joinToString("|")
+    }
+}

core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Group.kt

@@ -13,17 +13,56 @@ data class Group(
     val comment: String = Constants.DEFAULT_COMMENT,
     val directionType: DirectionType = DirectionType.BOTH,
     val ttl: Long = Constants.Group.DEFAULT_TTL,
+    val aggregations: Aggregations = Aggregations.EMPTY,
 ) {
     @JsonIgnore
     val code = XXHash32Wrapper.default.stringHash(group)
 
+    @JsonIgnore
+    fun dimensionFields(topk: Topk): List<Field> = fields.filter { it.bucket == null && !it.matchesDimension(topk.dimension) }
+
     data class Field(
         val name: String,
         val bucket: Bucket? = null,
     ) {
+        fun matchesDimension(dimension: String): Boolean = name.removePrefix("_") == dimension.removePrefix("_")
+
         fun bucketOrGet(
             value: Any,
             ceil: Boolean,
         ): Any = bucket?.handleQueryValue(value, ceil)?.toString() ?: value
     }
 }
+
+data class Aggregations(
+    val topk: List<Topk> = emptyList(),
+) {
+    @JsonIgnore
+    fun isEmpty(): Boolean = AggregationType.entries.none { it.has(this) }
+
+    @JsonIgnore
+    fun supports(type: AggregationType): Boolean = type.has(this)
+
+    companion object {
+        val EMPTY = Aggregations()
+    }
+}
+
+data class Topk(
+    val topk: String,
+    val entity: String,
+    val ranges: String = "",
+    val dimension: String,
+    val refreshAfterMillis: Long = -1,
+    val refreshQueue: String = AggregationConstants.Topk.REFRESH_QUEUE,
+    val rank: String,
+    val additionalProperties: List<String> = emptyList(),
+)
+
+enum class AggregationType {
+    TOPK {
+        override fun has(aggregations: Aggregations): Boolean = aggregations.topk.isNotEmpty()
+    }, ;
+
+    abstract fun has(aggregations: Aggregations): Boolean
+}

core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/ModelSchema.kt

@@ -22,17 +22,33 @@ import com.fasterxml.jackson.annotation.JsonTypeName
 sealed class ModelSchema : AbstractSchema {
     abstract val properties: List<StructField>
 
+    open val groups: List<Group> = emptyList()
+
     @get:JsonIgnore
     val propertiesByName: Map<String, StructField> by lazy { properties.associateBy { it.name } }
 
+    @get:JsonIgnore
+    val topkByName: Map<String, Topk> by lazy {
+        groups
+            .flatMap { it.aggregations.topk }
+            .associateBy { it.topk }
+    }
+
+    @get:JsonIgnore
+    val groupByTopkName: Map<String, Group> by lazy {
+        groups
+            .flatMap { group -> group.aggregations.topk.map { it.topk to group } }
+            .toMap()
+    }
+
     @JsonTypeName("edge")
     data class Edge(
         val source: Field,
         val target: Field,
         override val properties: List<StructField> = emptyList(),
         val direction: DirectionType,
         val indexes: List<Index> = emptyList(),
-        val groups: List<Group> = emptyList(),
+        override val groups: List<Group> = emptyList(),
         val caches: List<Cache> = emptyList(),
     ) : ModelSchema(),
         AbstractSchema by Schema(properties.associate { it.name to it.nullable })
@@ -44,7 +60,7 @@ sealed class ModelSchema : AbstractSchema {
         override val properties: List<StructField> = emptyList(),
         val direction: DirectionType,
         val indexes: List<Index> = emptyList(),
-        val groups: List<Group> = emptyList(),
+        override val groups: List<Group> = emptyList(),
     ) : ModelSchema(),
         AbstractSchema by Schema(properties.associate { it.name to it.nullable }) {
         init {
@@ -65,7 +81,7 @@ sealed class ModelSchema : AbstractSchema {
         override val properties: List<StructField> = emptyList(),
         val direction: DirectionType,
         val indexes: List<Index> = emptyList(),
-        val groups: List<Group> = emptyList(),
+        override val groups: List<Group> = emptyList(),
         val caches: List<Cache> = emptyList(),
     ) : ModelSchema(),
         AbstractSchema by Schema(properties.associate { it.name to it.nullable } + listOf("_source" to false, "_target" to false))

core/src/main/kotlin/com/kakao/actionbase/core/metadata/payload/AggregationsListResponse.kt

@@ -0,0 +1,25 @@
+package com.kakao.actionbase.core.metadata.payload
+
+import com.kakao.actionbase.core.metadata.QualifiedAggregations
+import com.kakao.actionbase.core.metadata.common.AggregationType
+
+data class AggregationsListResponse(
+    val items: List<Item>,
+) {
+    data class Item(
+        val type: AggregationType,
+        val database: String,
+        val table: String,
+        val dedupeFields: List<String>,
+    )
+
+    companion object {
+        fun of(aggregations: List<QualifiedAggregations>): AggregationsListResponse =
+            AggregationsListResponse(
+                items =
+                    aggregations.map {
+                        Item(type = it.type, database = it.database, table = it.table, dedupeFields = it.dedupeFields)
+                    },
+            )
+    }
+}

core/src/test/kotlin/com/kakao/actionbase/core/metadata/TableSerializationTest.kt

@@ -113,7 +113,8 @@ class TableSerializationTest {
                     "valueField": "-",
                     "comment": "group by day",
                     "directionType": "BOTH",
-                    "ttl": 691200000
+                    "ttl": 691200000,
+                    "aggregations": {"topk": []}
                   }
                 ],
                 "caches": []

core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationsTest.kt

@@ -0,0 +1,32 @@
+package com.kakao.actionbase.core.metadata.common
+
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class AggregationsTest {
+    @Test
+    fun `supports TOPK when topk is present`() {
+        assertTrue(Aggregations(topk = listOf(topk)).supports(AggregationType.TOPK))
+    }
+
+    @Test
+    fun `does not support TOPK when topk is empty`() {
+        assertFalse(Aggregations().supports(AggregationType.TOPK))
+    }
+
+    @Test
+    fun `isEmpty is true when no aggregation is defined`() {
+        assertTrue(Aggregations().isEmpty())
+    }
+
+    @Test
+    fun `isEmpty is false when topk is present`() {
+        assertFalse(Aggregations(topk = listOf(topk)).isEmpty())
+    }
+
+    companion object {
+        private val topk =
+            Topk(topk = "topk", entity = "__GLOBAL__", dimension = "target", rank = "db.table__topk")
+    }
+}

core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/EdgeSchemaSerializationTest.kt

@@ -165,6 +165,106 @@ class EdgeSchemaSerializationTest {
                 }
               ]
             }
+        - name: with aggregations (per-entity with window)
+          input: |-
+            {
+              "type": "MULTI_EDGE",
+              "id": {"type": "long", "comment": ""},
+              "source": {"type": "long", "comment": "user id"},
+              "target": {"type": "long", "comment": "item id"},
+              "properties": [],
+              "direction": "BOTH",
+              "groups": [
+                {
+                  "group": "purchased_count_1y",
+                  "type": "COUNT",
+                  "fields": [
+                    {"name": "_target"},
+                    {"name": "day", "bucket": {"type": "date", "name": "time", "unit": "MILLISECOND", "timezone": "+09:00", "format": "yyyy-MM-dd"}}
+                  ],
+                  "directionType": "OUT",
+                  "aggregations": {
+                    "topk": [
+                      {
+                        "topk": "top_purchased_1y",
+                        "entity": "source",
+                        "ranges": "time:bt:now-365d,now",
+                        "dimension": "target",
+                        "refreshAfterMillis": 31536000000,
+                        "rank": "commerce.order_product__topk"
+                      }
+                    ]
+                  }
+                }
+              ]
+            }
+          expected: {
+              "type": "multiEdge",
+              "id": {"type": "long", "comment": ""},
+              "source": {"type": "long", "comment": "user id"},
+              "target": {"type": "long", "comment": "item id"},
+              "properties": [],
+              "direction": "BOTH",
+              "groups": [
+                {
+                  "group": "purchased_count_1y",
+                  "type": "COUNT",
+                  "fields": [
+                    {"name": "_target"},
+                    {"name": "day", "bucket": {"type": "date", "name": "time", "unit": "MILLISECOND", "timezone": "+09:00", "format": "yyyy-MM-dd"}}
+                  ],
+                  "directionType": "OUT",
+                  "aggregations": {
+                    "topk": [
+                      {
+                        "topk": "top_purchased_1y",
+                        "entity": "source",
+                        "ranges": "time:bt:now-365d,now",
+                        "dimension": "target",
+                        "refreshAfterMillis": 31536000000,
+                        "rank": "commerce.order_product__topk"
+                      }
+                    ]
+                  }
+                }
+              ]
+            }
+        - name: with aggregations (no ranges)
+          input: |-
+            {
+              "type": "EDGE",
+              "source": {"type": "long", "comment": ""},
+              "target": {"type": "long", "comment": ""},
+              "properties": [],
+              "direction": "OUT",
+              "groups": [
+                {
+                  "group": "purchased_count",
+                  "type": "COUNT",
+                  "fields": [{"name": "_target"}],
+                  "aggregations": {
+                    "topk": [{"topk": "top_purchased", "entity": "__GLOBAL__", "dimension": "target", "rank": "commerce.order_product__topk"}]
+                  }
+                }
+              ]
+            }
+          expected: {
+              "type": "edge",
+              "source": {"type": "long", "comment": ""},
+              "target": {"type": "long", "comment": ""},
+              "properties": [],
+              "direction": "OUT",
+              "groups": [
+                {
+                  "group": "purchased_count",
+                  "type": "COUNT",
+                  "fields": [{"name": "_target"}],
+                  "aggregations": {
+                    "topk": [{"topk": "top_purchased", "entity": "__GLOBAL__", "dimension": "target", "rank": "commerce.order_product__topk"}]
+                  }
+                }
+              ]
+            }
         """,
     )
     fun `deserializes edge schema from JSON`(
@@ -175,6 +275,28 @@ class EdgeSchemaSerializationTest {
         assertEquals(expected, objectMapper.readValue<ModelSchema>(input))
     }
 
+    @Test
+    fun `a top-K refreshes onto the queue it names, and onto the shared one when it names none`() {
+        val named =
+            objectMapper.readValue<Topk>(
+                """
+                {"topk": "top_purchased", "entity": "source", "dimension": "target",
+                 "rank": "commerce.order_product__topk", "refreshQueue": "purchase_refresh"}
+                """.trimIndent(),
+            )
+        assertEquals("purchase_refresh", named.refreshQueue)
+        assertEquals(named, objectMapper.readValue<Topk>(objectMapper.writeValueAsString(named)))
+
+        val unnamed =
+            objectMapper.readValue<Topk>(
+                """
+                {"topk": "top_purchased", "entity": "source", "dimension": "target",
+                 "rank": "commerce.order_product__topk"}
+                """.trimIndent(),
+            )
+        assertEquals(AggregationConstants.Topk.REFRESH_QUEUE, unnamed.refreshQueue)
+    }
+
     @ObjectSourceParameterizedTest
     @ObjectSource(
         """

engine/src/main/kotlin/com/kakao/actionbase/engine/AggregationEngine.kt

@@ -0,0 +1,14 @@
+package com.kakao.actionbase.engine
+
+import com.kakao.actionbase.core.metadata.QualifiedAggregations
+import com.kakao.actionbase.core.metadata.common.AggregationType
+import com.kakao.actionbase.engine.binding.TableBinding
+
+interface AggregationEngine {
+    fun getTableBinding(
+        database: String,
+        alias: String,
+    ): TableBinding
+
+    fun getListWithAggregations(type: AggregationType? = null): List<QualifiedAggregations>
+}

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

@@ -0,0 +1,11 @@
+package com.kakao.actionbase.engine.service
+
+import com.kakao.actionbase.core.metadata.QualifiedAggregations
+import com.kakao.actionbase.core.metadata.common.AggregationType
+import com.kakao.actionbase.engine.AggregationEngine
+
+class AggregationService(
+    private val engine: AggregationEngine,
+) {
+    fun getAggregations(type: AggregationType? = null): List<QualifiedAggregations> = engine.getListWithAggregations(type)
+}