Read a materialized top-K as a step in a multi-hop query (Step 6 of #386)

engine/src/main/kotlin/com/kakao/actionbase/engine/binding/TableBinding.kt

@@ -16,6 +16,15 @@ interface TableBinding {
     val schema: ModelSchema
     val mutationMode: MutationMode
 
+    /**
+     * An empty result that still carries this table's columns.
+     *
+     * A read that found nothing has no rows, but the table it read from still has the columns it always
+     * had. Dropping them turns "no rows" into "no such column" for anything that resolves names against
+     * the frame — a SQL transform planned against a schema-less frame cannot resolve a single one.
+     */
+    val emptyFrame: DataFrame get() = DataFrame.empty
+
     // -- mutation
 
     fun <T> withLock(

engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQuery.kt

@@ -13,7 +13,7 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
 
 data class ActionbaseQuery(
-    val query: List<Item>,
+    val fetch: List<Item>,
     val stats: Set<StatKey> = emptySet(),
 ) {
     @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
@@ -23,6 +23,7 @@ data class ActionbaseQuery(
         JsonSubTypes.Type(value = Item.Count::class, name = "COUNT"),
         JsonSubTypes.Type(value = Item.Scan::class, name = "SCAN"),
         JsonSubTypes.Type(value = Item.Seek::class, name = "CACHE"),
+        JsonSubTypes.Type(value = Item.Topk::class, name = "TOPK"),
     )
     sealed class Item {
         abstract val name: String
@@ -96,6 +97,21 @@ data class ActionbaseQuery(
             override val post: List<PostProcessor> = emptyList(),
             override val aggregators: List<Aggregator> = emptyList(),
         ) : Item()
+
+        data class Topk(
+            override val name: String = UUID.randomUUID().toString(),
+            val database: String,
+            val table: String,
+            val topk: String,
+            val entity: Vertex? = null,
+            val dimensionValues: Map<String, String> = emptyMap(),
+            val limit: Int = DEFAULT_SCAN_LIMIT,
+            val offset: String? = null,
+            override val include: Boolean = false,
+            override val memoize: Boolean = false,
+            override val post: List<PostProcessor> = emptyList(),
+            override val aggregators: List<Aggregator> = emptyList(),
+        ) : Item()
     }
 
     @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")

engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQueryExecutor.kt

@@ -5,6 +5,8 @@ import com.kakao.actionbase.core.metadata.common.StructField
 import com.kakao.actionbase.core.metadata.common.StructType
 import com.kakao.actionbase.core.types.PrimitiveType
 import com.kakao.actionbase.engine.QueryEngine
+import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.service.RankScan
 import com.kakao.actionbase.engine.sql.DataFrame
 import com.kakao.actionbase.engine.sql.Row
 
@@ -51,16 +53,15 @@ class ActionbaseQueryExecutor(
 
     fun query(actionBaseQuery: ActionbaseQuery): Mono<Map<String, DataFrame>> {
         val computed: Mono<Map<String, DataFrame>> =
-            actionBaseQuery.query.fold(Mono.just(emptyMap())) { acc, queryItem ->
+            actionBaseQuery.fetch.fold(Mono.just(emptyMap())) { acc, queryItem ->
                 acc.flatMap { context ->
                     processQuery(queryItem, context, actionBaseQuery)
                         .map { context + (queryItem.name to it) }
                 }
             }
-        val returnNames = actionBaseQuery.query.filter { it.include }.map { it.name }
-        return computed.map { context ->
-            returnNames.associateWith { context.getValue(it) }
-        }
+        val returnNames = actionBaseQuery.fetch.filter { it.include }.map { it.name }
+
+        return computed.map { context -> returnNames.associateWith { context.getValue(it) } }
     }
 
     private fun processQuery(
@@ -84,6 +85,7 @@ class ActionbaseQueryExecutor(
             is ActionbaseQuery.Item.Count -> processCount(queryItem, context, actionBaseQuery)
             is ActionbaseQuery.Item.Scan -> processScan(queryItem, context, actionBaseQuery)
             is ActionbaseQuery.Item.Seek -> processSeek(queryItem, context, actionBaseQuery)
+            is ActionbaseQuery.Item.Topk -> processTopk(queryItem, context, actionBaseQuery)
         }
 
     private fun applyPostProcessors(
@@ -168,15 +170,16 @@ class ActionbaseQueryExecutor(
         actionBaseQuery: ActionbaseQuery,
     ): Mono<DataFrame> =
         resolveVertex(queryItem.source, context, actionBaseQuery).flatMap { source ->
+            val binding = engine.getTableBinding(database = queryItem.database, alias = queryItem.table)
+
             Flux
                 .fromIterable(source)
                 .flatMap { start ->
-                    engine
-                        .getTableBinding(database = queryItem.database, alias = queryItem.table)
+                    binding
                         .scan(queryItem.index, start, queryItem.direction, queryItem.limit, queryItem.offset, ranges = queryItem.ranges, filters = null, features = emptyList())
                 }.reduce { a, b ->
                     DataFrame(rows = a.rows + b.rows, schema = a.schema, total = a.total + b.total)
-                }.switchIfEmpty(Mono.just(DataFrame.empty))
+                }.switchIfEmpty(Mono.just(binding.emptyFrame))
         }
 
     private fun processSeek(
@@ -185,12 +188,68 @@ class ActionbaseQueryExecutor(
         actionBaseQuery: ActionbaseQuery,
     ): Mono<DataFrame> =
         resolveVertex(queryItem.source, context, actionBaseQuery).flatMap { source ->
-            engine
-                .getTableBinding(database = queryItem.database, alias = queryItem.table)
+            val binding = engine.getTableBinding(database = queryItem.database, alias = queryItem.table)
+
+            binding
                 .seek(queryItem.cache, source.toList(), queryItem.direction, queryItem.limit, offset = null, ranges = queryItem.ranges, filters = null, features = emptyList())
-                .switchIfEmpty(Mono.just(DataFrame.empty))
+                .switchIfEmpty(Mono.just(binding.emptyFrame))
         }
 
+    private fun processTopk(
+        queryItem: ActionbaseQuery.Item.Topk,
+        context: Map<String, DataFrame>,
+        actionBaseQuery: ActionbaseQuery,
+    ): Mono<DataFrame> {
+        val binding = engine.getTableBinding(database = queryItem.database, alias = queryItem.table)
+        val entity = queryItem.entity ?: return scanRanking(queryItem, binding, entity = null)
+
+        return resolveVertex(entity, context, actionBaseQuery).flatMap { entities ->
+            Flux
+                .fromIterable(entities)
+                .flatMap { scanRanking(queryItem, binding, it.toString()) }
+                .reduce { a, b -> DataFrame(rows = a.rows + b.rows, schema = a.schema, total = a.total + b.total) }
+                .switchIfEmpty(Mono.just(rankBindingOf(queryItem, binding).emptyFrame))
+        }
+    }
+
+    private fun scanRanking(
+        queryItem: ActionbaseQuery.Item.Topk,
+        binding: TableBinding,
+        entity: String?,
+    ): Mono<DataFrame> {
+        val rank = rankScanOf(queryItem, binding, entity)
+
+        return engine
+            .getTableBinding(database = rank.database, alias = rank.table)
+            .scan(rank.index, rank.start, rank.direction, queryItem.limit, queryItem.offset, ranges = null, filters = null, features = emptyList())
+    }
+
+    /**
+     * The step names the source table, but the rows come from the ranking table it points at, so an empty
+     * ranking has to answer with the ranking table's columns rather than the source table's.
+     */
+    private fun rankBindingOf(
+        queryItem: ActionbaseQuery.Item.Topk,
+        binding: TableBinding,
+    ): TableBinding =
+        rankScanOf(queryItem, binding, entity = null).let { rank ->
+            engine.getTableBinding(database = rank.database, alias = rank.table)
+        }
+
+    private fun rankScanOf(
+        queryItem: ActionbaseQuery.Item.Topk,
+        binding: TableBinding,
+        entity: String?,
+    ): RankScan =
+        RankScan.from(
+            schema = binding.schema,
+            database = queryItem.database,
+            table = binding.table,
+            topk = queryItem.topk,
+            entity = entity,
+            dimensionValues = queryItem.dimensionValues,
+        )
+
     // region Aggregators
 
     internal fun aggregateCount(

engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBinding.kt

@@ -57,6 +57,8 @@ class V2BackedTableBinding(
     override val schema: ModelSchema = descriptor.schema
     override val mutationMode: MutationMode = MutationMode.valueOf(label.entity.mode.name)
 
+    override val emptyFrame: DataFrame by lazy { V2DataFrame.empty(label.entity.schema.allStructType).toV3(total = 0L) }
+
     private val groupRecordMapper = mapper.group
     private val cacheRecordMapper = mapper.cache
 

engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/MultiHopQuerySpec.kt

@@ -148,7 +148,7 @@ class MultiHopQuerySpec :
             // 2-hop query: user1's friends' wishlists
             val query =
                 ActionbaseQuery(
-                    query =
+                    fetch =
                         listOf(
                             ActionbaseQuery.Item.Scan(
                                 name = "hop1",
@@ -336,7 +336,7 @@ class MultiHopQuerySpec :
             // 3-hop query: user1's friends' wishlists' reviews
             val query =
                 ActionbaseQuery(
-                    query =
+                    fetch =
                         listOf(
                             ActionbaseQuery.Item.Scan(
                                 name = "hop1",