Read a materialized top-K by naming its entity and dimension values (Step 5 of #386)

core/src/main/kotlin/com/kakao/actionbase/core/edge/payload/AggregationsTopkResponse.kt

@@ -0,0 +1,36 @@
+package com.kakao.actionbase.core.edge.payload
+
+import com.kakao.actionbase.core.metadata.common.AggregationConstants
+
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.fasterxml.jackson.module.kotlin.readValue
+
+data class AggregationsTopkResponse(
+    val topks: List<TopkItem>,
+    val count: Int,
+) {
+    data class TopkItem(
+        val value: String,
+        val metric: Long,
+        val properties: Map<String, String>,
+    )
+
+    companion object {
+        private val MAPPER = jacksonObjectMapper()
+
+        fun from(payload: DataFrameEdgePayload): AggregationsTopkResponse {
+            val topkItems =
+                payload.edges.map { edge ->
+                    TopkItem(
+                        value = edge.target.toString(),
+                        metric = (edge.properties[AggregationConstants.Topk.METRIC] as? Number)?.toLong() ?: 0L,
+                        properties =
+                            (edge.properties[AggregationConstants.Topk.ADDITIONAL_PROPERTIES] as? String)
+                                ?.let { MAPPER.readValue<Map<String, String>>(it) }
+                                ?: emptyMap(),
+                    )
+                }
+            return AggregationsTopkResponse(topks = topkItems, count = payload.count)
+        }
+    }
+}

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)
+}

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

@@ -0,0 +1,43 @@
+package com.kakao.actionbase.engine.service
+
+import com.kakao.actionbase.core.edge.payload.DataFrameEdgePayload
+import com.kakao.actionbase.engine.AggregationEngine
+import com.kakao.actionbase.v2.engine.sql.ScanFilter
+
+import reactor.core.publisher.Mono
+
+class AggregationQueryService(
+    private val queryService: QueryService,
+    private val engine: AggregationEngine,
+) {
+    fun topk(
+        database: String,
+        table: String,
+        topk: String,
+        entity: String? = null,
+        dimensionValues: Map<String, String> = emptyMap(),
+        limit: Int = ScanFilter.defaultLimit,
+        offset: String? = null,
+    ): Mono<DataFrameEdgePayload> {
+        val tb = engine.getTableBinding(database = database, alias = table)
+        val rank =
+            RankScan.from(
+                schema = tb.schema,
+                database = database,
+                table = tb.table,
+                topk = topk,
+                entity = entity,
+                dimensionValues = dimensionValues,
+            )
+
+        return queryService.scan(
+            database = rank.database,
+            table = rank.table,
+            index = rank.index,
+            start = rank.start,
+            direction = rank.direction,
+            limit = limit,
+            offset = offset,
+        )
+    }
+}

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

@@ -0,0 +1,47 @@
+package com.kakao.actionbase.engine.service
+
+import com.kakao.actionbase.core.metadata.common.AggregationConstants
+import com.kakao.actionbase.core.metadata.common.ModelSchema
+import com.kakao.actionbase.v2.core.metadata.Direction
+
+internal data class RankScan(
+    val database: String,
+    val table: String,
+    val index: String,
+    val start: String,
+    val direction: Direction,
+) {
+    companion object {
+        fun from(
+            schema: ModelSchema,
+            database: String,
+            table: String,
+            topk: String,
+            entity: String?,
+            dimensionValues: Map<String, String>,
+        ): RankScan {
+            val config =
+                schema.topkByName[topk]
+                    ?: throw IllegalArgumentException("Unknown topk `$topk` for $database.$table.")
+            val group =
+                schema.groupByTopkName[topk]
+                    ?: throw IllegalArgumentException("Topk `$topk` of $database.$table is not declared on any group.")
+            val (rankDatabase, rankTable) = parseFqn(config.rank)
+
+            return RankScan(
+                database = rankDatabase,
+                table = rankTable,
+                index = AggregationConstants.Topk.RANK_INDEX,
+                start =
+                    AggregationConstants.Topk.rankSource(
+                        database = database,
+                        table = table,
+                        topk = topk,
+                        entity = if (config.entity == AggregationConstants.Topk.GLOBAL_ENTITY) AggregationConstants.Topk.GLOBAL_ENTITY else entity.orEmpty(),
+                        dimensionValues = group.dimensionFields(config).map { dimensionValues[it.name].orEmpty() },
+                    ),
+                direction = Direction.OUT,
+            )
+        }
+    }
+}

engine/src/test/kotlin/com/kakao/actionbase/engine/service/RankScanTest.kt

@@ -0,0 +1,119 @@
+package com.kakao.actionbase.engine.service
+
+import com.kakao.actionbase.core.metadata.common.Aggregations
+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.v2.core.metadata.Direction
+
+import org.junit.jupiter.api.Test
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+
+class RankScanTest {
+    @Test
+    fun `reads the rank table declared by the topk, in rank order`() {
+        val rank = rankScan()
+
+        rank.database shouldBe "commerce"
+        rank.table shouldBe "orders_table__topk"
+        rank.index shouldBe "metric_desc"
+        rank.direction shouldBe Direction.OUT
+    }
+
+    @Test
+    fun `keys the scan by the source table and the entity`() {
+        rankScan().start shouldBe "commerce|orders_table|top_purchased|user1"
+    }
+
+    @Test
+    fun `orders the named values the way the group declares them, not the way they were passed`() {
+        val rank =
+            rankScan(
+                schema = schema(dimensionFields = listOf("category", "region")),
+                dimensionValues = mapOf("region" to "seoul", "category" to "fruit"),
+            )
+
+        rank.start shouldBe "commerce|orders_table|top_purchased|user1|fruit|seoul"
+    }
+
+    @Test
+    fun `a value left out reads as empty`() {
+        val rank =
+            rankScan(
+                schema = schema(dimensionFields = listOf("category", "region")),
+                dimensionValues = mapOf("region" to "seoul"),
+            )
+
+        rank.start shouldBe "commerce|orders_table|top_purchased|user1||seoul"
+    }
+
+    @Test
+    fun `a value the group does not declare is ignored`() {
+        val rank =
+            rankScan(
+                schema = schema(dimensionFields = listOf("category")),
+                dimensionValues = mapOf("category" to "fruit", "shoeSize" to "270"),
+            )
+
+        rank.start shouldBe "commerce|orders_table|top_purchased|user1|fruit"
+    }
+
+    @Test
+    fun `a global topk keys by the sentinel and ignores the entity`() {
+        val rank = rankScan(schema = schema(entity = "__GLOBAL__"), entity = "user1")
+
+        rank.start shouldBe "commerce|orders_table|top_purchased|__GLOBAL__"
+    }
+
+    @Test
+    fun `rejects a topk the table does not declare`() {
+        shouldThrow<IllegalArgumentException> { rankScan(topk = "top_viewed") }
+    }
+}
+
+// region test fixtures
+
+private fun rankScan(
+    schema: ModelSchema = schema(),
+    topk: String = "top_purchased",
+    entity: String? = "user1",
+    dimensionValues: Map<String, String> = emptyMap(),
+): RankScan =
+    RankScan.from(
+        schema = schema,
+        database = "commerce",
+        table = "orders_table",
+        topk = topk,
+        entity = entity,
+        dimensionValues = dimensionValues,
+    )
+
+private fun schema(
+    entity: String = "source",
+    dimensionFields: List<String> = emptyList(),
+): ModelSchema =
+    ModelSchema.Edge(
+        source = Field(type = PrimitiveType.STRING, comment = "user"),
+        target = Field(type = PrimitiveType.STRING, comment = "item"),
+        direction = DirectionType.OUT,
+        groups =
+            listOf(
+                Group(
+                    group = "purchased_count",
+                    type = GroupType.COUNT,
+                    fields = listOf(Group.Field(name = "_target")) + dimensionFields.map { Group.Field(name = it) },
+                    aggregations =
+                        Aggregations(
+                            topk = listOf(Topk(topk = "top_purchased", entity = entity, dimension = "target", rank = "commerce.orders_table__topk")),
+                        ),
+                ),
+            ),
+    )
+
+// endregion

server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/metadata/MetadataAggQueryController.kt

@@ -0,0 +1,36 @@
+package com.kakao.actionbase.server.api.graph.v3.metadata
+
+import com.kakao.actionbase.core.edge.payload.AggregationsTopkResponse
+import com.kakao.actionbase.engine.service.AggregationQueryService
+import com.kakao.actionbase.server.payload.AggregationsTopkRequest
+import com.kakao.actionbase.server.util.mapToResponseEntity
+import com.kakao.actionbase.v2.engine.sql.ScanFilter
+
+import org.springframework.http.ResponseEntity
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
+import org.springframework.web.bind.annotation.RestController
+
+import reactor.core.publisher.Mono
+
+@RestController
+class MetadataAggQueryController(
+    private val aggregationQueryService: AggregationQueryService,
+) {
+    /**
+     * Reads a ranking over the body. A ranking is picked by naming values, and a query string cannot
+     * carry those names onto an immutable request, so the read is not offered over `GET`.
+     */
+    @PostMapping("/aggregations/v1/databases/{database}/tables/{table}/topks/{topk}")
+    fun topk(
+        @PathVariable database: String,
+        @PathVariable table: String,
+        @PathVariable topk: String,
+        @RequestBody request: AggregationsTopkRequest,
+    ): Mono<ResponseEntity<AggregationsTopkResponse>> =
+        aggregationQueryService
+            .topk(database, table, topk, request.entity, request.dimensionValues, request.limit ?: ScanFilter.defaultLimit, request.offset)
+            .map(AggregationsTopkResponse::from)
+            .mapToResponseEntity()
+}

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.AggregationQueryService
 import com.kakao.actionbase.engine.service.AggregationService
 import com.kakao.actionbase.engine.service.MutationService
 import com.kakao.actionbase.engine.service.QueryService
@@ -160,6 +161,12 @@ class GraphConfiguration {
         handlers: List<AggregationHandler>,
     ): AggregationService = AggregationService(engine, handlers)
 
+    @Bean
+    fun provideAggregationQueryService(
+        queryService: QueryService,
+        engine: V2BackedEngine,
+    ): AggregationQueryService = AggregationQueryService(queryService, engine)
+
     @Bean
     fun provideMutationService(
         engine: V2BackedEngine,

server/src/main/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilter.kt

@@ -11,7 +11,8 @@ import org.springframework.web.server.WebFilterChain
 import reactor.core.publisher.Mono
 
 // Rejects non-GET methods on graph path prefixes with 403.
-// Exceptions: read-only POST endpoints matched by readSuffixes.
+// Exceptions: read-only POST endpoints, matched by readSuffixes, or by readSegments when the path ends in
+// a variable and so has no fixed suffix to match on.
 class ReadOnlyRequestFilter : WebFilter {
     private val log = LoggerFactory.getLogger(ReadOnlyRequestFilter::class.java)
 
@@ -23,6 +24,7 @@ class ReadOnlyRequestFilter : WebFilter {
             "/multi-edges/ids",
             "/query",
         )
+    private val readSegments = setOf("/topks/")
 
     init {
         log.info("ReadOnlyRequestFilter is active. Write operations on {} will be rejected.", paths)
@@ -50,5 +52,5 @@ class ReadOnlyRequestFilter : WebFilter {
         return exchange.response.writeWith(Mono.just(messageBuffer))
     }
 
-    private fun isRead(path: String): Boolean = readSuffixes.any { path.endsWith(it) }
+    private fun isRead(path: String): Boolean = readSuffixes.any { path.endsWith(it) } || readSegments.any { path.contains(it) }
 }

server/src/main/kotlin/com/kakao/actionbase/server/payload/AggregationsTopkRequest.kt

@@ -0,0 +1,8 @@
+package com.kakao.actionbase.server.payload
+
+data class AggregationsTopkRequest(
+    val entity: String? = null,
+    val dimensionValues: Map<String, String> = emptyMap(),
+    val limit: Int? = null,
+    val offset: String? = null,
+)