diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/QualifiedAggregations.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/QualifiedAggregations.kt new file mode 100644 index 000000000..2edbca330 --- /dev/null +++ b/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 = emptyList(), +) 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 new file mode 100644 index 000000000..e84cb085c --- /dev/null +++ b/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 = (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 = (listOf(database, table, topk, entity, topkDimensionValue) + dimensionValues).joinToString("|") + } +} diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Group.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Group.kt index 8a3b090be..ed41e382d 100644 --- a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/Group.kt +++ b/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 = 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 = 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 = emptyList(), +) + +enum class AggregationType { + TOPK { + override fun has(aggregations: Aggregations): Boolean = aggregations.topk.isNotEmpty() + }, ; + + abstract fun has(aggregations: Aggregations): Boolean +} diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/ModelSchema.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/ModelSchema.kt index c4dc50c20..8f8f7054b 100644 --- a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/ModelSchema.kt +++ b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/common/ModelSchema.kt @@ -22,9 +22,25 @@ import com.fasterxml.jackson.annotation.JsonTypeName sealed class ModelSchema : AbstractSchema { abstract val properties: List + open val groups: List = emptyList() + @get:JsonIgnore val propertiesByName: Map by lazy { properties.associateBy { it.name } } + @get:JsonIgnore + val topkByName: Map by lazy { + groups + .flatMap { it.aggregations.topk } + .associateBy { it.topk } + } + + @get:JsonIgnore + val groupByTopkName: Map by lazy { + groups + .flatMap { group -> group.aggregations.topk.map { it.topk to group } } + .toMap() + } + @JsonTypeName("edge") data class Edge( val source: Field, @@ -32,7 +48,7 @@ sealed class ModelSchema : AbstractSchema { override val properties: List = emptyList(), val direction: DirectionType, val indexes: List = emptyList(), - val groups: List = emptyList(), + override val groups: List = emptyList(), val caches: List = emptyList(), ) : ModelSchema(), AbstractSchema by Schema(properties.associate { it.name to it.nullable }) @@ -44,7 +60,7 @@ sealed class ModelSchema : AbstractSchema { override val properties: List = emptyList(), val direction: DirectionType, val indexes: List = emptyList(), - val groups: List = emptyList(), + override val groups: List = 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 = emptyList(), val direction: DirectionType, val indexes: List = emptyList(), - val groups: List = emptyList(), + override val groups: List = emptyList(), val caches: List = emptyList(), ) : ModelSchema(), AbstractSchema by Schema(properties.associate { it.name to it.nullable } + listOf("_source" to false, "_target" to false)) diff --git a/core/src/main/kotlin/com/kakao/actionbase/core/metadata/payload/AggregationsListResponse.kt b/core/src/main/kotlin/com/kakao/actionbase/core/metadata/payload/AggregationsListResponse.kt new file mode 100644 index 000000000..77bb0910c --- /dev/null +++ b/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, +) { + data class Item( + val type: AggregationType, + val database: String, + val table: String, + val dedupeFields: List, + ) + + companion object { + fun of(aggregations: List): AggregationsListResponse = + AggregationsListResponse( + items = + aggregations.map { + Item(type = it.type, database = it.database, table = it.table, dedupeFields = it.dedupeFields) + }, + ) + } +} diff --git a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/TableSerializationTest.kt b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/TableSerializationTest.kt index 10e4b3184..d089a281c 100644 --- a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/TableSerializationTest.kt +++ b/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": [] diff --git a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationsTest.kt b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/AggregationsTest.kt new file mode 100644 index 000000000..9dca33a15 --- /dev/null +++ b/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") + } +} diff --git a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/EdgeSchemaSerializationTest.kt b/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/EdgeSchemaSerializationTest.kt index b54a216a3..859f74521 100644 --- a/core/src/test/kotlin/com/kakao/actionbase/core/metadata/common/EdgeSchemaSerializationTest.kt +++ b/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(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": "top_purchased", "entity": "source", "dimension": "target", + "rank": "commerce.order_product__topk", "refreshQueue": "purchase_refresh"} + """.trimIndent(), + ) + assertEquals("purchase_refresh", named.refreshQueue) + assertEquals(named, objectMapper.readValue(objectMapper.writeValueAsString(named))) + + val unnamed = + objectMapper.readValue( + """ + {"topk": "top_purchased", "entity": "source", "dimension": "target", + "rank": "commerce.order_product__topk"} + """.trimIndent(), + ) + assertEquals(AggregationConstants.Topk.REFRESH_QUEUE, unnamed.refreshQueue) + } + @ObjectSourceParameterizedTest @ObjectSource( """ diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/AggregationEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/AggregationEngine.kt new file mode 100644 index 000000000..895e7e94b --- /dev/null +++ b/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 +} 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 new file mode 100644 index 000000000..24a569f67 --- /dev/null +++ b/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 = engine.getListWithAggregations(type) +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt index a0771fcd0..f50bf3986 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt @@ -9,6 +9,8 @@ import com.kakao.actionbase.core.edge.mapper.EdgeIndexRecordMapper import com.kakao.actionbase.core.edge.mapper.EdgeLockRecordMapper import com.kakao.actionbase.core.edge.mapper.EdgeRecordMapper import com.kakao.actionbase.core.edge.mapper.EdgeStateRecordMapper +import com.kakao.actionbase.core.metadata.QualifiedAggregations +import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.metadata.features.FeatureFlags import com.kakao.actionbase.engine.query.LabelProvider import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory @@ -38,6 +40,8 @@ import com.kakao.actionbase.v2.engine.entity.EntityName import com.kakao.actionbase.v2.engine.entity.LabelEntity import com.kakao.actionbase.v2.engine.entity.ServiceEntity import com.kakao.actionbase.v2.engine.entity.StorageEntity +import com.kakao.actionbase.v2.engine.entity.hasAggregation +import com.kakao.actionbase.v2.engine.entity.toQualifiedAggregations import com.kakao.actionbase.v2.engine.exception.MutationError import com.kakao.actionbase.v2.engine.fake.fakeEdges import com.kakao.actionbase.v2.engine.label.DeleteEdgeRequest @@ -259,6 +263,11 @@ class Graph( .defaultIfEmpty(false) } + fun listWithAggregations(type: AggregationType? = null): List = + labels.values + .filter { it.entity.hasAggregation(type) } + .flatMap { it.entity.toQualifiedAggregations(type) } + fun getAllLabels(): Map = ( labels.map { diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctions.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctions.kt new file mode 100644 index 000000000..9671de878 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctions.kt @@ -0,0 +1,37 @@ +package com.kakao.actionbase.v2.engine.entity + +import com.kakao.actionbase.core.metadata.QualifiedAggregations +import com.kakao.actionbase.core.metadata.common.AggregationType +import com.kakao.actionbase.core.metadata.common.DirectionType +import com.kakao.actionbase.core.metadata.common.Group + +fun LabelEntity.hasAggregation(type: AggregationType? = null): Boolean = + groups.any { g -> + if (type == null) { + !g.aggregations.isEmpty() + } else { + g.aggregations.supports(type) + } + } + +fun LabelEntity.toQualifiedAggregations(type: AggregationType? = null): List { + val candidateTypes = if (type != null) listOf(type) else AggregationType.entries + return candidateTypes + .filter { t -> groups.any { g -> t.has(g.aggregations) } } + .map { t -> + QualifiedAggregations( + type = t, + database = name.service, + table = name.nameNotNull, + dedupeFields = groups.filter { g -> t.has(g.aggregations) }.flatMap { it.dedupeFields() }.distinct(), + ) + } +} + +// The endpoint is derived from the group direction rather than declared, but it is part of the rank row's +// identity either way, so it is always keyed on. Buckets are not identity. BOTH would produce two +// endpoints; it keys on source until something rejects it. +private fun Group.dedupeFields(): List { + val endpoint = if (directionType == DirectionType.IN) "target" else "source" + return listOf(endpoint) + fields.filter { it.bucket == null }.map { it.name } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt index 359933694..919164f93 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt @@ -3,7 +3,10 @@ package com.kakao.actionbase.v2.engine.v3 import com.kakao.actionbase.v2.core.metadata.MutationMode as V2MutationMode import com.kakao.actionbase.core.edge.MutationEvent +import com.kakao.actionbase.core.metadata.QualifiedAggregations +import com.kakao.actionbase.core.metadata.common.AggregationType import com.kakao.actionbase.core.state.State +import com.kakao.actionbase.engine.AggregationEngine import com.kakao.actionbase.engine.MutationContext import com.kakao.actionbase.engine.MutationEngine import com.kakao.actionbase.engine.QueryEngine @@ -23,7 +26,8 @@ import reactor.core.publisher.Mono class V2BackedEngine( private val graph: Graph, ) : MutationEngine, - QueryEngine { + QueryEngine, + AggregationEngine { override fun getTableBinding( database: String, alias: String, @@ -41,6 +45,8 @@ class V2BackedEngine( return label.tableBinding } + override fun getListWithAggregations(type: AggregationType?): List = graph.listWithAggregations(type) + private val messaging = V2BackedMessageBinding(wal = graph.wal, cdc = graph.cdc) override fun writeWal( diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctionsTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctionsTest.kt new file mode 100644 index 000000000..5900fdb2e --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntityFunctionsTest.kt @@ -0,0 +1,181 @@ +package com.kakao.actionbase.v2.engine.entity + +import com.kakao.actionbase.core.metadata.common.DirectionType as GroupDirectionType + +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.Group +import com.kakao.actionbase.core.metadata.common.GroupType +import com.kakao.actionbase.core.metadata.common.Topk +import com.kakao.actionbase.v2.core.metadata.DirectionType +import com.kakao.actionbase.v2.core.metadata.LabelType +import com.kakao.actionbase.v2.core.types.DataType +import com.kakao.actionbase.v2.core.types.EdgeSchema +import com.kakao.actionbase.v2.core.types.Field +import com.kakao.actionbase.v2.core.types.VertexField +import com.kakao.actionbase.v2.core.types.VertexType + +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test + +import io.kotest.matchers.booleans.shouldBeFalse +import io.kotest.matchers.booleans.shouldBeTrue +import io.kotest.matchers.collections.shouldBeEmpty +import io.kotest.matchers.collections.shouldHaveSize +import io.kotest.matchers.shouldBe + +class LabelEntityFunctionsTest { + @Nested + inner class DetectingAggregations { + @Test + fun `returns true when a group defines topk`() { + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = listOf(topk(name = "top_purchased"))))) + .hasAggregation() + .shouldBeTrue() + } + + @Test + fun `returns false when no group defines any aggregation`() { + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = emptyList()))) + .hasAggregation() + .shouldBeFalse() + } + + @Test + fun `with the TOPK filter, returns true when a group defines topk`() { + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = listOf(topk(name = "top_purchased"))))) + .hasAggregation(AggregationType.TOPK) + .shouldBeTrue() + } + + @Test + fun `with the TOPK filter, returns false when the label defines no topk`() { + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = emptyList()))) + .hasAggregation(AggregationType.TOPK) + .shouldBeFalse() + } + } + + @Nested + inner class BuildingQualifiedAggregations { + @Test + fun `emits one entry per kind on a user table`() { + val label = + label( + groups = + listOf( + group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = listOf(topk(name = "top_purchased"))), + group(name = "viewed_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = emptyList()), + group(name = "liked_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = listOf(topk(name = "top_liked"))), + ), + ) + + val result = label.toQualifiedAggregations() + + result shouldHaveSize 1 + result.single().type shouldBe AggregationType.TOPK + result.single().database shouldBe "commerce" + result.single().table shouldBe "orders" + } + + @Test + fun `returns empty for a user table with no topk`() { + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = emptyList()))) + .toQualifiedAggregations() + .shouldBeEmpty() + } + + @Test + fun `with the TOPK filter, keeps a label that defines topk`() { + val result = + label(groups = listOf(group(name = "purchased_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = listOf(topk(name = "top_purchased"))))) + .toQualifiedAggregations(AggregationType.TOPK) + + result shouldHaveSize 1 + result.single().type shouldBe AggregationType.TOPK + } + + @Test + fun `unions every ranking's dedupe fields into one key`() { + val label = + label( + groups = + listOf( + group(name = "purchased_count", direction = GroupDirectionType.OUT, fields = listOf(Group.Field("category")), topks = listOf(topk(name = "top_purchased", entity = "source"))), + group(name = "viewed_count", direction = GroupDirectionType.BOTH, fields = emptyList(), topks = emptyList()), + group(name = "liked_count", direction = GroupDirectionType.OUT, fields = emptyList(), topks = listOf(topk(name = "top_liked", entity = "source"))), + group(name = "shared_count", direction = GroupDirectionType.OUT, fields = listOf(Group.Field("channel")), topks = listOf(topk(name = "top_shared", entity = "source"))), + group(name = "purchased_by_count", direction = GroupDirectionType.IN, fields = emptyList(), topks = listOf(topk(name = "top_purchasers"))), + ), + ) + + label.toQualifiedAggregations().single().dedupeFields shouldBe + listOf("source", "category", "channel", "target") + } + + @Test + fun `drops bucket fields from the dedupe key`() { + val label = + label( + groups = + listOf( + group( + name = "purchased_count", + direction = GroupDirectionType.OUT, + fields = + listOf( + Group.Field("target"), + Group.Field("day", bucket = Bucket.Date(name = "day", unit = Bucket.ValueUnit.MILLISECOND, timezone = "UTC", format = "yyyy-MM-dd")), + ), + topks = listOf(topk(name = "top_purchased", entity = "source")), + ), + ), + ) + + label.toQualifiedAggregations().single().dedupeFields shouldBe listOf("source", "target") + } + } +} + +// region test fixtures + +// builds the commerce.orders label; only `groups` varies, the rest is required-but-irrelevant setup +private fun label(groups: List): LabelEntity = + LabelEntity( + active = true, + name = EntityName("commerce", "orders"), + desc = "", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.LONG, "src"), + VertexField(VertexType.LONG, "tgt"), + listOf(Field("payload", DataType.STRING, true, "")), + ), + dirType = DirectionType.BOTH, + storage = "test", + groups = groups, + ) + +private fun group( + name: String, + direction: GroupDirectionType, + fields: List, + topks: List, +): Group = + Group( + group = name, + type = GroupType.SUM, + fields = fields, + directionType = direction, + aggregations = Aggregations(topk = topks), + ) + +// `dimension` and `rank` do not affect these functions, so they stay defaulted +private fun topk( + name: String, + entity: String = "__GLOBAL__", +): Topk = Topk(topk = name, entity = entity, dimension = "target", rank = "${name}__topk") + +// 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 new file mode 100644 index 000000000..055b3e70c --- /dev/null +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggController.kt @@ -0,0 +1,24 @@ +package com.kakao.actionbase.server.api.graph.v3.metadata + +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.RequestParam +import org.springframework.web.bind.annotation.RestController + +@RestController +class MetadataAggController( + private val aggregationService: AggregationService, +) { + @GetMapping("/aggregations/v1/metadata") + fun getAggregations( + @RequestParam(required = false) type: AggregationType?, + ): ResponseEntity { + val aggregations = aggregationService.getAggregations(type) + + return ResponseEntity.ok(AggregationsListResponse.of(aggregations)) + } +} 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 6f4a53a5e..edd71244b 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 @@ -2,6 +2,7 @@ package com.kakao.actionbase.server.configuration import com.kakao.actionbase.core.metadata.features.FeatureFlags 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.server.client.kafka.SpringKafkaClientFactory @@ -143,6 +144,9 @@ class GraphConfiguration { @Bean fun provideQueryService(engine: V2BackedEngine): QueryService = QueryService(engine) + @Bean + fun provideAggregationService(engine: V2BackedEngine): AggregationService = AggregationService(engine) + @Bean fun provideMutationService( engine: V2BackedEngine, diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/filter/PathPrefixes.kt b/server/src/main/kotlin/com/kakao/actionbase/server/filter/PathPrefixes.kt index 1ce488d1f..e9a2f8c20 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/filter/PathPrefixes.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/filter/PathPrefixes.kt @@ -6,9 +6,11 @@ object PathPrefixes { val QUEUE = setOf("/queue/v1") + val AGGREGATIONS = setOf("/aggregations/v1") + val CONTROL = setOf("/control") - val DATA = GRAPH + QUEUE + val DATA = GRAPH + QUEUE + AGGREGATIONS val FILTERED = DATA + CONTROL } 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 37ffb5e8c..0fd527aaa 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 @@ -175,6 +175,7 @@ class ReadOnlyRequestFilterTest { "GET /graph/v3/databases/{database}/tables/{table}/edges/scan/{index}", "GET /graph/v3/databases/{database}/tables/{table}/multi-edges/ids", "GET /graph/v3/databases/{database}/tables/{table}/vertices/get", + "GET /aggregations/v1/metadata", "GET /graph/v3/datastore", "GET /graph/v3/datastore/hbase/namespaces", "GET /graph/v3/datastore/hbase/references",