diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQueryExecutor.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQueryExecutor.kt index 8097e9151..d096827a3 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQueryExecutor.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/query/ActionbaseQueryExecutor.kt @@ -56,6 +56,14 @@ class ActionbaseQueryExecutor( fun query(actionBaseQuery: ActionbaseQuery): Mono> = query(actionBaseQuery, emptyList()) + fun query( + prepared: PreparedQuery, + arguments: Map, + ): Mono> { + val bound = prepared.bind(arguments) + return query(bound, bound.transform.map { transform -> prepared.transformArguments(transform, arguments) }) + } + private fun query( actionBaseQuery: ActionbaseQuery, transformArguments: List>, diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/query/PreparedQuery.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/query/PreparedQuery.kt new file mode 100644 index 000000000..31272be10 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/query/PreparedQuery.kt @@ -0,0 +1,191 @@ +package com.kakao.actionbase.engine.query + +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.core.types.PrimitiveType +import com.kakao.actionbase.v2.engine.sql.StatKey + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ArrayNode +import com.fasterxml.jackson.databind.node.ObjectNode +import com.fasterxml.jackson.module.kotlin.convertValue +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +/** + * A query whose shape is fixed and whose values are not: anywhere a whole value would go, the body may + * hold `{name}` instead. + * + * Binding happens on the JSON, before it is read into an [ActionbaseQuery]: `"limit": "{limit}"` cannot + * be the `Int` that slot wants until the value is in, so substitution has to come first. + * + * A placeholder inside SQL becomes a `?` bound at execution instead, which keeps the SQL text — and so + * the compiled plan — the same between calls. + */ +class PreparedQuery private constructor( + private val template: ObjectNode, + val parameters: Set, + /** Empty for a query nobody registered: with no declaration, a value goes in as the JSON it came as. */ + private val types: Map, +) { + fun bind(values: Map): ActionbaseQuery { + val missing = parameters - values.keys + require(missing.isEmpty()) { "This query takes ${parameters.joinToString()}; the call is missing ${missing.joinToString()}." } + + val unknown = values.keys - parameters + require(unknown.isEmpty()) { "This query has no argument named ${unknown.joinToString()}." } + + val cast = parameters.associateWith { name -> cast(name, values.getValue(name)) } + return objectMapper.convertValue(template.deepCopy().substitute(cast)) + } + + private fun cast( + name: String, + value: Any, + ): Any = types[name]?.cast(value) ?: value + + fun transformArguments( + transform: ActionbaseQuery.Transform, + values: Map, + ): List = + when (transform) { + is ActionbaseQuery.Transform.Sql -> + transform.arguments.map { name -> + require(name in parameters) { "`${transform.name}` binds `$name`, which this query has no argument for." } + cast(name, values.getValue(name)) + } + } + + private fun JsonNode.substitute(values: Map): JsonNode { + when (this) { + is ObjectNode -> fieldNames().asSequence().toList().forEach { field -> replace(field, get(field).substituted(values)) } + is ArrayNode -> (0 until size()).forEach { index -> set(index, get(index).substituted(values)) } + else -> Unit + } + return this + } + + private fun JsonNode.substituted(values: Map): JsonNode = + parameterOf(this)?.let { name -> objectMapper.valueToTree(values.getValue(name)) } + ?: substitute(values) + + companion object { + private val objectMapper = jacksonObjectMapper() + + private const val SQL_FIELD = "sql" + + private const val ARGUMENTS_FIELD = "arguments" + + /** Whole values only, so that an argument can be a number: `"user-{entity}"` is a string. */ + private val PARAMETER = Regex("^\\{([A-Za-z_][A-Za-z0-9_]*)}$") + + private val IN_SQL = Regex("\\{([A-Za-z_][A-Za-z0-9_]*)}") + + /** A registered query, which has to use exactly the names it declares. */ + fun of( + fetch: JsonNode, + transform: JsonNode, + stats: Set, + arguments: List, + ): PreparedQuery { + val template = template(fetch, transform, stats) + val used = template.prepare() + val declared = arguments.map { it.name }.toSet() + + require(used == declared) { + "This query uses ${used.joinToString().ifEmpty { "no arguments" }} " + + "but declares ${declared.joinToString().ifEmpty { "none" }}." + } + + return PreparedQuery(template, declared, arguments.associate { it.name to it.type }) + } + + fun of(query: ActionbaseQuery): PreparedQuery { + // The whole query, not its parts: a step's `type` is written from the property's declared type, + // which a bare list of items no longer carries. + val template = objectMapper.valueToTree(query) + return PreparedQuery(template, template.prepare(), emptyMap()) + } + + /** A query nobody registered: the names are the ones the body uses, and nothing declares a type. */ + fun adhoc( + fetch: JsonNode, + transform: JsonNode, + stats: Set, + ): PreparedQuery { + val template = template(fetch, transform, stats) + return PreparedQuery(template, template.prepare(), emptyMap()) + } + + private fun template( + fetch: JsonNode, + transform: JsonNode, + stats: Set, + ): ObjectNode = + objectMapper.createObjectNode().apply { + set("fetch", fetch.deepCopy()) + set("transform", transform.deepCopy()) + set("stats", objectMapper.valueToTree(stats)) + } + + /** + * Rewrites every SQL body to the `?` form once, and answers with the names the template asks for. + * Doing it here rather than per call is what keeps the SQL text, and so the compiled plan, the same + * between calls. + * + * The SQL itself is not read here. Calcite first sees it on the call that runs it, because a plan is + * compiled against the columns of the frames it reads and those are only known once the fetch steps + * have run — so a statement Calcite would reject registers cleanly and fails on its first call. + */ + private fun JsonNode.prepare(): Set = + when (this) { + is ObjectNode -> + fieldNames() + .asSequence() + .toList() + .flatMap { field -> + val child = get(field) + if (field == SQL_FIELD && child.isTextual) { + val (sql, bound) = rewrite(child.asText()) + put(SQL_FIELD, sql) + set(ARGUMENTS_FIELD, objectMapper.valueToTree(bound)) + bound + } else { + child.prepare() + } + }.toSet() + is ArrayNode -> (0 until size()).flatMap { get(it).prepare() }.toSet() + else -> parameterOf(this)?.let { setOf(it) } ?: emptySet() + } + + /** Braces inside a string literal are left alone: `WHERE tag = \'{limit}\'` is a literal. */ + private fun rewrite(sql: String): Pair> { + val rewritten = StringBuilder() + val bound = mutableListOf() + var index = 0 + var quoted = false + + while (index < sql.length) { + val char = sql[index] + if (char == '\'') { + quoted = !quoted + rewritten.append(char) + index++ + continue + } + + val match = if (!quoted && char == '{') IN_SQL.matchAt(sql, index) else null + if (match == null) { + rewritten.append(char) + index++ + } else { + rewritten.append('?') + bound += match.groupValues[1] + index += match.value.length + } + } + + return rewritten.toString() to bound + } + + private fun parameterOf(node: JsonNode): String? = node.takeIf { it.isTextual }?.asText()?.let { PARAMETER.find(it)?.groupValues?.get(1) } + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/PreparedQueryService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/PreparedQueryService.kt new file mode 100644 index 000000000..6b2bf66eb --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/PreparedQueryService.kt @@ -0,0 +1,282 @@ +package com.kakao.actionbase.engine.service + +import com.kakao.actionbase.core.Constants +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.engine.query.PreparedQuery +import com.kakao.actionbase.engine.sql.DataFrame +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.entity.QueryEntity +import com.kakao.actionbase.v2.engine.service.ddl.QueryAliasCreateRequest +import com.kakao.actionbase.v2.engine.service.ddl.QueryAliasDeleteRequest +import com.kakao.actionbase.v2.engine.service.ddl.QueryAliasUpdateRequest +import com.kakao.actionbase.v2.engine.service.ddl.QueryCreateRequest +import com.kakao.actionbase.v2.engine.service.ddl.QueryDeleteRequest +import com.kakao.actionbase.v2.engine.service.ddl.QueryUpdateRequest +import com.kakao.actionbase.v2.engine.sql.StatKey + +import java.util.UUID + +import com.fasterxml.jackson.databind.JsonNode +import com.github.benmanes.caffeine.cache.Caffeine + +import reactor.core.publisher.Mono + +class PreparedQueryService( + private val graph: Graph, + private val queries: QueryService, +) { + private val byId = + Caffeine + .newBuilder() + .also { builder -> graph.metastoreReloadInterval?.let { builder.expireAfterWrite(it) } } + .maximumSize(MAXIMUM_CACHED_QUERIES) + .build() + + private val idOfAlias = + Caffeine + .newBuilder() + .also { builder -> graph.metastoreReloadInterval?.let { builder.expireAfterWrite(it) } } + .maximumSize(MAXIMUM_CACHED_QUERIES) + .build() + + fun register( + desc: String, + arguments: List, + fetch: JsonNode, + transform: JsonNode, + stats: Set = emptySet(), + ): Mono { + validate(arguments, fetch, transform, stats) + + val id = UUID.randomUUID().toString() + return graph.queryDdl + .create(EntityName.fromOrigin(id), QueryCreateRequest(desc, arguments, fetch, transform, stats)) + .then(Mono.defer { get(id) }) + } + + fun amend( + id: String, + desc: String? = null, + arguments: List? = null, + fetch: JsonNode? = null, + transform: JsonNode? = null, + stats: Set? = null, + ): Mono = + readQuery(id).flatMap { existing -> + validate( + arguments ?: existing.arguments, + fetch ?: existing.fetch, + transform ?: existing.transform, + stats ?: existing.stats, + ) + + graph.queryDdl + .update( + EntityName.fromOrigin(id), + QueryUpdateRequest(desc = desc, arguments = arguments, fetch = fetch, transform = transform, stats = stats), + ).then(Mono.fromRunnable { byId.invalidate(id) }) + .then(Mono.defer { get(id) }) + } + + fun get(id: String): Mono = registered(id).map { it.descriptor } + + fun list(status: MetadataStatus = MetadataStatus.ACTIVE): Mono> = + graph.queryDdl + .getAll(EntityName.origin) + .map { page -> page.content.filter { status.matches(it.active) }.map { it.toDescriptor() } } + + fun delete(id: String): Mono = + namesOf(id).flatMap { named -> + require(named.isEmpty()) { "`$id` is still named by ${named.joinToString()}; drop those first." } + + val name = EntityName.fromOrigin(id) + graph.queryDdl + .update(name, QueryUpdateRequest(active = false)) + .then(graph.queryDdl.delete(name, QueryDeleteRequest())) + .then(Mono.fromRunnable { byId.invalidate(id) }) + } + + fun createAlias( + alias: String, + desc: String, + target: String, + ): Mono { + require(alias.matches(ALIAS)) { "Invalid alias: ${Constants.Name.MESSAGE}" } + require(alias != ALIASES_PATH) { "`$ALIASES_PATH` is reserved: it is the path the alias list is read from." } + + val name = EntityName.fromOrigin(alias) + return readAlias(name) + .flatMap { Mono.error { IllegalArgumentException("alias name already exists : $alias") } } + .switchIfEmpty( + Mono.defer { + requireQuery(target) + .then(graph.queryAliasDdl.create(name, QueryAliasCreateRequest(desc, target))) + .then(Mono.defer { alias(alias) }) + }, + ) + } + + fun updateAlias( + alias: String, + desc: String? = null, + target: String? = null, + active: Boolean? = null, + ): Mono { + val name = EntityName.fromOrigin(alias) + return readAlias(name) + .switchIfEmpty(Mono.error { NoSuchPreparedQueryException(alias) }) + .flatMap { target?.let { requireQuery(it).thenReturn(Unit) } ?: Mono.just(Unit) } + .flatMap { graph.queryAliasDdl.update(name, QueryAliasUpdateRequest(active = active, desc = desc, target = target)) } + .then(Mono.fromRunnable { idOfAlias.invalidate(alias) }) + .then(Mono.defer { alias(alias) }) + } + + fun alias(alias: String): Mono = + readAlias(EntityName.fromOrigin(alias)) + .map { PreparedQueryAliasDescriptor(it.alias, it.target, it.active, it.desc) } + .switchIfEmpty(Mono.error { NoSuchPreparedQueryException(alias) }) + + fun aliases(status: MetadataStatus = MetadataStatus.ACTIVE): Mono> = + graph.queryAliasDdl + .getAll(EntityName.origin) + .map { page -> + page.content + .filter { status.matches(it.active) } + .map { PreparedQueryAliasDescriptor(it.alias, it.target, it.active, it.desc) } + } + + fun deleteAlias(alias: String): Mono { + val name = EntityName.fromOrigin(alias) + return readAlias(name) + .switchIfEmpty(Mono.error { NoSuchPreparedQueryException(alias) }) + .flatMap { + graph.queryAliasDdl + .update(name, QueryAliasUpdateRequest(active = false)) + .then(graph.queryAliasDdl.delete(name, QueryAliasDeleteRequest())) + }.then(Mono.fromRunnable { idOfAlias.invalidate(alias) }) + } + + fun query( + id: String, + arguments: Map, + ): Mono> = registered(id).flatMap { queries.query(it.prepared, arguments) } + + private fun registered(id: String): Mono = + if (id.matches(ALIAS)) { + Mono + .justOrEmpty(idOfAlias.getIfPresent(id)) + .switchIfEmpty( + readAlias(EntityName.fromOrigin(id)) + .map { it.target } + .doOnNext { idOfAlias.put(id, it) }, + ).switchIfEmpty(Mono.error { NoSuchPreparedQueryException(id) }) + .flatMap { target -> loaded(target) } + } else { + loaded(id) + } + + private fun loaded(id: String): Mono = + Mono + .justOrEmpty(byId.getIfPresent(id)) + .switchIfEmpty( + Mono.defer { + readQuery(id) + .map { entity -> Registered(entity.toDescriptor(), entity.toPrepared()) } + .doOnNext { byId.put(id, it) } + }, + ).switchIfEmpty(Mono.error { NoSuchPreparedQueryException(id) }) + + private fun readQuery(id: String): Mono = graph.queryDdl.getSingle(EntityName.fromOrigin(id)).filter { it.active } + + private fun requireQuery(id: String): Mono = readQuery(id).switchIfEmpty(Mono.error { NoSuchPreparedQueryException(id) }) + + private fun readAlias(name: EntityName) = graph.queryAliasDdl.getSingle(name).filter { it.active } + + /** + * A metadata scan stops at `metadataFetchLimit` without saying so, and a page filled to it cannot prove + * that no name points at the query. Refuse rather than report an absence we can't back — the caller is + * about to drop the query. `DatastoreTableReferences` reads its own scans under the same rule. + */ + private fun namesOf(id: String): Mono> = + graph.queryAliasDdl.getAll(EntityName.origin).map { page -> + check(page.content.size < graph.metadataFetchLimit) { + "Cannot tell whether `$id` is still named: the alias scan hit the metadataFetchLimit of " + + "${graph.metadataFetchLimit}, so the result may be incomplete" + } + page.content.filter { it.target == id }.map { it.alias } + } + + private fun validate( + arguments: List, + fetch: JsonNode, + transform: JsonNode, + stats: Set, + ) { + val prepared = PreparedQuery.of(fetch, transform, stats, arguments) + prepared.bind(arguments.associate { it.name to sample(it) }) + } + + private fun sample(argument: StructField): Any = argument.type.cast(SAMPLE_TEXT) + + private fun QueryEntity.toPrepared(): PreparedQuery = PreparedQuery.of(fetch, transform, stats, arguments) + + private fun QueryEntity.toDescriptor(): PreparedQueryDescriptor = + PreparedQueryDescriptor( + id = id, + arguments = arguments, + fetch = fetch, + transform = transform, + stats = stats, + active = active, + desc = desc, + ) + + private data class Registered( + val descriptor: PreparedQueryDescriptor, + val prepared: PreparedQuery, + ) + + private companion object { + val ALIAS = Regex(Constants.Name.PATTERN) + const val ALIASES_PATH = "aliases" + const val MAXIMUM_CACHED_QUERIES = 1_000L + + const val SAMPLE_TEXT = "1" + } +} + +class NoSuchPreparedQueryException( + id: String, +) : NoSuchElementException("No prepared query `$id`.") + +enum class MetadataStatus { + ACTIVE, + INACTIVE, + ALL, + ; + + fun matches(active: Boolean): Boolean = + when (this) { + ACTIVE -> active + INACTIVE -> !active + ALL -> true + } +} + +data class PreparedQueryDescriptor( + val id: String, + val arguments: List, + val fetch: JsonNode, + val transform: JsonNode, + val stats: Set, + val active: Boolean, + val desc: String, +) + +data class PreparedQueryAliasDescriptor( + val alias: String, + val target: String, + val active: Boolean, + val desc: String, +) diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt index f255fe08b..f46bb163a 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt @@ -12,6 +12,7 @@ import com.kakao.actionbase.core.edge.payload.EdgePayload import com.kakao.actionbase.engine.QueryEngine import com.kakao.actionbase.engine.query.ActionbaseQuery import com.kakao.actionbase.engine.query.ActionbaseQueryExecutor +import com.kakao.actionbase.engine.query.PreparedQuery import com.kakao.actionbase.engine.sql.DataFrame import com.kakao.actionbase.v2.core.metadata.Direction import com.kakao.actionbase.v2.engine.sql.ScanFilter @@ -183,6 +184,11 @@ class QueryService( fun query(request: ActionbaseQuery): Mono> = queryExecutor.query(request) + fun query( + prepared: PreparedQuery, + arguments: Map, + ): Mono> = queryExecutor.query(prepared, arguments) + private fun DataFrame.toEdgePayload(flip: Boolean = false): DataFrameEdgePayload { val edges = rows.map { row -> 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 f50bf3986..ac396787c 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 @@ -38,6 +38,8 @@ import com.kakao.actionbase.v2.engine.entity.AliasEntity import com.kakao.actionbase.v2.engine.entity.EdgeEntity import com.kakao.actionbase.v2.engine.entity.EntityName import com.kakao.actionbase.v2.engine.entity.LabelEntity +import com.kakao.actionbase.v2.engine.entity.QueryAliasEntity +import com.kakao.actionbase.v2.engine.entity.QueryEntity import com.kakao.actionbase.v2.engine.entity.ServiceEntity import com.kakao.actionbase.v2.engine.entity.StorageEntity import com.kakao.actionbase.v2.engine.entity.hasAggregation @@ -58,6 +60,8 @@ import com.kakao.actionbase.v2.engine.metadata.sync.MetadataType import com.kakao.actionbase.v2.engine.migration.Migration import com.kakao.actionbase.v2.engine.service.ddl.AliasDdlService import com.kakao.actionbase.v2.engine.service.ddl.LabelDdlService +import com.kakao.actionbase.v2.engine.service.ddl.QueryAliasDdlService +import com.kakao.actionbase.v2.engine.service.ddl.QueryDdlService import com.kakao.actionbase.v2.engine.service.ddl.ServiceDdlService import com.kakao.actionbase.v2.engine.service.ddl.StorageDdlService import com.kakao.actionbase.v2.engine.sql.DataFrame @@ -110,6 +114,8 @@ class Graph( labelLabel: Label, infoLabel: Label, aliasLabel: Label, + queryLabel: Label, + queryAliasLabel: Label, onlineMetadataLabel: Label, private val nilLabel: Label, ) : GraphDefaults, @@ -143,6 +149,8 @@ class Graph( labelLabel.name to labelLabel, infoLabel.name to infoLabel, aliasLabel.name to aliasLabel, + queryLabel.name to queryLabel, + queryAliasLabel.name to queryAliasLabel, onlineMetadataLabel.name to onlineMetadataLabel, nilLabel.name to nilLabel, ) @@ -165,12 +173,18 @@ class Graph( val aliasDdl = AliasDdlService(this, aliasLabel, AliasEntity) + val queryDdl = QueryDdlService(this, queryLabel, QueryEntity) + + val queryAliasDdl = QueryAliasDdlService(this, queryAliasLabel, QueryAliasEntity) + fun isReady(): Boolean = metadataInitialized val encoderPoolSize = config.encoderPoolSize val metadataFetchLimit = config.metadataFetchLimit + val metastoreReloadInterval: Duration? = config.metastoreReloadInterval + val systemMutationMode = config.systemMutationMode init { @@ -263,11 +277,6 @@ 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 { @@ -386,6 +395,11 @@ class Graph( .subscribeOn(Schedulers.boundedElastic()) } + fun listWithAggregations(type: AggregationType? = null): List = + labels.values + .filter { it.entity.hasAggregation(type) } + .flatMap { it.entity.toQualifiedAggregations(type) } + /** * For `system=ASYNC + label=SYNC` (no force), return the SYNC-shaped status derived from the * request's [EdgeOperation] so clients keep their contract; mutations are still queued via the @@ -1037,6 +1051,10 @@ class Graph( val aliasLabel = Metadata.aliasLabelEntity.materialize(defaults) + val queryLabel = Metadata.queryLabelEntity.materialize(defaults) + + val queryAliasLabel = Metadata.queryAliasLabelEntity.materialize(defaults) + val onlineMetadataLabel = onlineMetadataLabelEntity.materialize(defaults) val nilLabel = @@ -1051,6 +1069,8 @@ class Graph( .then(mutate(Metadata.labelLabelEntity.toEdge(), EdgeOperation.INSERT)) .then(mutate(Metadata.infoLabelEntity.toEdge(), EdgeOperation.INSERT)) .then(mutate(Metadata.aliasLabelEntity.toEdge(), EdgeOperation.INSERT)) + .then(mutate(Metadata.queryLabelEntity.toEdge(), EdgeOperation.INSERT)) + .then(mutate(Metadata.queryAliasLabelEntity.toEdge(), EdgeOperation.INSERT)) .then(mutate(onlineMetadataLabelEntity.toEdge(), EdgeOperation.INSERT)) .then(mutate(Metadata.sysNilLabelEntity.toEdge(), EdgeOperation.INSERT)) .block() @@ -1073,6 +1093,8 @@ class Graph( labelLabel, infoLabel, aliasLabel, + queryLabel, + queryAliasLabel, onlineMetadataLabel, nilLabel, ) diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryAliasEntity.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryAliasEntity.kt new file mode 100644 index 000000000..3bf313c0d --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryAliasEntity.kt @@ -0,0 +1,43 @@ +package com.kakao.actionbase.v2.engine.entity + +import com.kakao.actionbase.v2.core.edge.TraceEdge +import com.kakao.actionbase.v2.engine.edge.HashEdge +import com.kakao.actionbase.v2.engine.sql.RowWithSchema + +data class QueryAliasEntity( + override val active: Boolean, + override val name: EntityName, + val desc: String, + val target: String, +) : EdgeEntity { + val alias: String + get() = name.nameNotNull + + override fun toEdge(): TraceEdge = + name.toTraceEdge( + props = + mapOf( + "props_active" to active, + "desc" to desc, + "target" to target, + ), + ) + + companion object : EntityFactory { + override fun toEntity(edge: HashEdge): QueryAliasEntity = + QueryAliasEntity( + active = (edge.props.getOrDefault("props_active", null) ?: true).toString().toBoolean(), + name = EntityName.withPhase(edge.src.toString(), edge.tgt.toString()), + desc = edge.props["desc"].toString(), + target = edge.props["target"].toString(), + ) + + override fun toEntity(row: RowWithSchema): QueryAliasEntity = + QueryAliasEntity( + active = (row.getOrNull("props_active") ?: true).toString().toBoolean(), + name = EntityName.withPhase(row.getString("src"), row.getString("tgt")), + desc = row.getString("desc"), + target = row.getString("target"), + ) + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryEntity.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryEntity.kt new file mode 100644 index 000000000..ac7696dd3 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/QueryEntity.kt @@ -0,0 +1,75 @@ +package com.kakao.actionbase.v2.engine.entity + +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.v2.core.edge.TraceEdge +import com.kakao.actionbase.v2.engine.edge.HashEdge +import com.kakao.actionbase.v2.engine.sql.RowWithSchema +import com.kakao.actionbase.v2.engine.sql.StatKey + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.module.kotlin.readValue + +data class QueryEntity( + override val active: Boolean, + override val name: EntityName, + val desc: String, + val arguments: List = emptyList(), + val fetch: JsonNode, + val transform: JsonNode = objectMapper.createArrayNode(), + val stats: Set = emptySet(), +) : EdgeEntity { + val id: String + get() = name.nameNotNull + + override fun toEdge(): TraceEdge = + name.toTraceEdge( + props = + mapOf( + "props_active" to active, + "desc" to desc, + "arguments" to objectMapper.writeValueAsString(arguments), + "fetch" to objectMapper.writeValueAsString(fetch), + "transform" to objectMapper.writeValueAsString(transform), + "stats" to stats.joinToString(",") { it.name }, + ), + ) + + companion object : EntityFactory { + internal val objectMapper = jacksonObjectMapper() + + override fun toEntity(edge: HashEdge): QueryEntity = + QueryEntity( + active = (edge.props.getOrDefault("props_active", null) ?: true).toString().toBoolean(), + name = EntityName.withPhase(edge.src.toString(), edge.tgt.toString()), + desc = edge.props["desc"].toString(), + arguments = toArguments(edge.props["arguments"]?.toString()), + fetch = toTree(edge.props["fetch"].toString()), + transform = toTree(edge.props["transform"]?.toString()), + stats = toStats(edge.props["stats"]?.toString()), + ) + + override fun toEntity(row: RowWithSchema): QueryEntity = + QueryEntity( + active = (row.getOrNull("props_active") ?: true).toString().toBoolean(), + name = EntityName.withPhase(row.getString("src"), row.getString("tgt")), + desc = row.getString("desc"), + arguments = toArguments(row.getOrNull("arguments")?.toString()), + fetch = toTree(row.getString("fetch")), + transform = toTree(row.getOrNull("transform")?.toString()), + stats = toStats(row.getOrNull("stats")?.toString()), + ) + + private fun toArguments(text: String?): List = objectMapper.readValue(text?.takeIf { it.isNotBlank() } ?: "[]") + + private fun toTree(text: String?): JsonNode = objectMapper.readTree(text?.takeIf { it.isNotBlank() } ?: "[]") + + private fun toStats(text: String?): Set = + text + ?.split(",") + ?.filter { it.isNotBlank() } + ?.map { StatKey.valueOf(it) } + ?.toSet() + ?: emptySet() + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/Metadata.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/Metadata.kt index 91c387d21..22a8d4d23 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/Metadata.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/Metadata.kt @@ -38,6 +38,10 @@ object Metadata { const val heartBeatLabelName = "heartbeat" + const val sysQueryLabelName = "query" + + const val sysQueryAliasLabelName = "query_alias" + const val sysOnlineMetadataLabelV2Name = "online_metadata_v2" const val sysNilLabelName = "nil" @@ -159,6 +163,49 @@ object Metadata { storage = EngineConstants.METASTORE_URI, ) + val queryLabelEntity = + LabelEntity( + active = true, + name = EntityName(sysServiceName, sysQueryLabelName), + desc = "System query label", + type = LabelType.HASH, + schema = + EdgeSchema( + VertexField(VertexType.STRING, "origin"), + VertexField(VertexType.STRING, "{{queryId}}"), + listOf( + Field("props_active", DataType.BOOLEAN, true), + Field("desc", DataType.STRING, false), + Field("arguments", DataType.STRING, true), + Field("fetch", DataType.STRING, false), + Field("transform", DataType.STRING, true), + Field("stats", DataType.STRING, true), + ), + ), + dirType = DirectionType.OUT, + storage = EngineConstants.METASTORE_URI, + ) + + val queryAliasLabelEntity = + LabelEntity( + active = true, + name = EntityName(sysServiceName, sysQueryAliasLabelName), + desc = "System query alias label", + type = LabelType.HASH, + schema = + EdgeSchema( + VertexField(VertexType.STRING, "origin"), + VertexField(VertexType.STRING, "{{alias}}"), + listOf( + Field("props_active", DataType.BOOLEAN, true), + Field("desc", DataType.STRING, false), + Field("target", DataType.STRING, false), + ), + ), + dirType = DirectionType.OUT, + storage = EngineConstants.METASTORE_URI, + ) + val onlineMetadataLabelV2Entity = LabelEntity( active = true, diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryAliasDdlService.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryAliasDdlService.kt new file mode 100644 index 000000000..831bb1890 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryAliasDdlService.kt @@ -0,0 +1,60 @@ +package com.kakao.actionbase.v2.engine.service.ddl + +import com.kakao.actionbase.v2.core.edge.TraceEdge +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.audit.Audit +import com.kakao.actionbase.v2.engine.edge.HashEdge +import com.kakao.actionbase.v2.engine.entity.EntityFactory +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.entity.QueryAliasEntity +import com.kakao.actionbase.v2.engine.label.Label + +import reactor.core.publisher.Mono + +class QueryAliasDdlService( + graph: Graph, + label: Label, + factory: EntityFactory, +) : DdlService(graph, label, factory) { + override fun canDeactivate(name: EntityName): Mono = Mono.just(true) + + override fun toEntity(edge: HashEdge): QueryAliasEntity = QueryAliasEntity.toEntity(edge) + + override fun sync(): Mono = Mono.empty() +} + +data class QueryAliasCreateRequest( + val desc: String, + val target: String, + override val audit: Audit = Audit.default, +) : DdlRequest { + override fun toEdge(name: EntityName): TraceEdge = + QueryAliasEntity( + active = true, + name = name, + desc = desc, + target = target, + ).toEdge() +} + +data class QueryAliasUpdateRequest( + val active: Boolean? = null, + val desc: String? = null, + val target: String? = null, + override val audit: Audit = Audit.default, +) : DdlRequest { + private fun toNotNullMap(): Map = + buildMap { + active?.let { put("props_active", it) } + desc?.let { put("desc", it) } + target?.let { put("target", it) } + } + + override fun toEdge(name: EntityName): TraceEdge = name.toTraceEdge(props = toNotNullMap()) +} + +data class QueryAliasDeleteRequest( + override val audit: Audit = Audit.default, +) : DdlRequest { + override fun toEdge(name: EntityName): TraceEdge = name.toTraceEdge() +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryDdlService.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryDdlService.kt new file mode 100644 index 000000000..397c52e6a --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/service/ddl/QueryDdlService.kt @@ -0,0 +1,81 @@ +package com.kakao.actionbase.v2.engine.service.ddl + +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.v2.core.edge.TraceEdge +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.audit.Audit +import com.kakao.actionbase.v2.engine.edge.HashEdge +import com.kakao.actionbase.v2.engine.entity.EntityFactory +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.entity.QueryEntity +import com.kakao.actionbase.v2.engine.label.Label +import com.kakao.actionbase.v2.engine.sql.StatKey + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +import reactor.core.publisher.Mono + +class QueryDdlService( + graph: Graph, + label: Label, + factory: EntityFactory, +) : DdlService(graph, label, factory) { + override fun canDeactivate(name: EntityName): Mono = Mono.just(true) + + override fun toEntity(edge: HashEdge): QueryEntity = QueryEntity.toEntity(edge) + + override fun sync(): Mono = Mono.empty() +} + +data class QueryCreateRequest( + val desc: String, + val arguments: List, + val fetch: JsonNode, + val transform: JsonNode, + val stats: Set = emptySet(), + override val audit: Audit = Audit.default, +) : DdlRequest { + override fun toEdge(name: EntityName): TraceEdge = + QueryEntity( + active = true, + name = name, + desc = desc, + arguments = arguments, + fetch = fetch, + transform = transform, + stats = stats, + ).toEdge() +} + +data class QueryUpdateRequest( + val active: Boolean? = null, + val desc: String? = null, + val arguments: List? = null, + val fetch: JsonNode? = null, + val transform: JsonNode? = null, + val stats: Set? = null, + override val audit: Audit = Audit.default, +) : DdlRequest { + private fun toNotNullMap(): Map = + buildMap { + active?.let { put("props_active", it) } + desc?.let { put("desc", it) } + arguments?.let { put("arguments", objectMapper.writeValueAsString(it)) } + fetch?.let { put("fetch", objectMapper.writeValueAsString(it)) } + transform?.let { put("transform", objectMapper.writeValueAsString(it)) } + stats?.let { put("stats", it.joinToString(",") { key -> key.name }) } + } + + override fun toEdge(name: EntityName): TraceEdge = name.toTraceEdge(props = toNotNullMap()) + + private companion object { + val objectMapper = jacksonObjectMapper() + } +} + +data class QueryDeleteRequest( + override val audit: Audit = Audit.default, +) : DdlRequest { + override fun toEdge(name: EntityName): TraceEdge = name.toTraceEdge() +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/query/PreparedQueryTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/query/PreparedQueryTest.kt new file mode 100644 index 000000000..f77c248d0 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/query/PreparedQueryTest.kt @@ -0,0 +1,220 @@ +package com.kakao.actionbase.engine.query + +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.sql.DataFrame +import com.kakao.actionbase.engine.sql.Row + +import org.junit.jupiter.api.Test + +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.matchers.shouldBe +import io.mockk.every +import io.mockk.mockk +import reactor.core.publisher.Mono + +class PreparedQueryTest { + @Test + fun `a parameter is found wherever a VALUE vertex holds it`() { + val prepared = PreparedQuery.of(query(source = value("{entity}"))) + + prepared.parameters shouldBe setOf("entity") + } + + @Test + fun `a value that only contains braces is not a parameter`() { + PreparedQuery.of(query(source = value("user-{entity}"))).parameters shouldBe emptySet() + } + + @Test + fun `binding replaces the placeholder and leaves the shape alone`() { + val template = query(source = value("{entity}")) + val prepared = PreparedQuery.of(template) + + val bound = prepared.bind(mapOf("entity" to "U1")) + + (bound.fetch.single() as ActionbaseQuery.Item.Self).source shouldBe ActionbaseQuery.Vertex.Value(listOf("U1")) + // The template is reusable: binding it again with something else has to work. + (prepared.bind(mapOf("entity" to "U2")).fetch.single() as ActionbaseQuery.Item.Self).source shouldBe + ActionbaseQuery.Vertex.Value(listOf("U2")) + template.fetch.single() shouldBe query(source = value("{entity}")).fetch.single() + } + + @Test + fun `an argument keeps its own type instead of becoming a string`() { + val bound = PreparedQuery.of(query(source = value("{entity}"))).bind(mapOf("entity" to 42L)) + + (bound.fetch.single() as ActionbaseQuery.Item.Self).source shouldBe ActionbaseQuery.Vertex.Value(listOf(42L)) + } + + @Test + fun `binding without an argument says which parameter is missing`() { + val prepared = PreparedQuery.of(query(source = value("{entity}"))) + + shouldThrow { prepared.bind(emptyMap()) } + .message shouldBe "This query takes entity; the call is missing entity." + } + + @Test + fun `a transform reads a step the caller does not get back`() { + val executor = ActionbaseQueryExecutor(engineReturning("hop1" to ranked("PG1" to 30L, "PG2" to 20L))) + val prepared = + PreparedQuery.of( + ActionbaseQuery( + fetch = listOf(self(name = "hop1", source = value("{entity}"), include = false)), + transform = + listOf( + ActionbaseQuery.Transform.Sql( + name = "result", + sql = "SELECT hop1.target AS productGroupId FROM hop1 ORDER BY hop1.metric DESC", + ), + ), + ), + ) + + val returned = executor.query(prepared, mapOf("entity" to "U1")).block()!! + + returned.keys shouldBe setOf("result") + returned.getValue("result").getColumn("productGroupId") shouldBe listOf("PG1", "PG2") + } + + @Test + fun `a transform binds its own placeholders from the same arguments`() { + val executor = ActionbaseQueryExecutor(engineReturning("hop1" to ranked("PG1" to 30L, "PG2" to 20L, "PG3" to 10L))) + val prepared = + PreparedQuery.of( + ActionbaseQuery( + fetch = listOf(self(name = "hop1", source = value("{entity}"))), + transform = + listOf( + // The name is written where the value goes; it becomes a `?` on the way out. + ActionbaseQuery.Transform.Sql( + name = "result", + sql = "SELECT hop1.target AS productGroupId FROM hop1 WHERE hop1.metric >= {floor} ORDER BY hop1.metric DESC", + ), + ), + ), + ) + + prepared.parameters shouldBe setOf("entity", "floor") + executor + .query(prepared, mapOf("entity" to "U1", "floor" to 20L)) + .block()!! + .getValue("result") + .getColumn("productGroupId") shouldBe listOf("PG1", "PG2") + executor + .query(prepared, mapOf("entity" to "U1", "floor" to 30L)) + .block()!! + .getValue("result") + .getColumn("productGroupId") shouldBe listOf("PG1") + } + + @Test + fun `braces inside a string literal are not a placeholder`() { + val prepared = + PreparedQuery.of( + ActionbaseQuery( + fetch = listOf(self(name = "hop1", source = value("{entity}"))), + transform = + listOf( + ActionbaseQuery.Transform.Sql( + name = "result", + sql = "SELECT hop1.target AS t FROM hop1 WHERE hop1.target <> '{floor}'", + ), + ), + ), + ) + + prepared.parameters shouldBe setOf("entity") + } + + /** + * The rewriter tracks `'` and nothing else, so a comment is ordinary text to it and its braces become a + * `?` Calcite never sees. The count is what catches it, on the first call rather than at registration, + * and the message says nothing about the comment. Pinned as it stands: the SQL is refused, which is the + * part that matters. + */ + @Test + fun `braces inside a comment are rewritten, and the call fails on the count`() { + val executor = ActionbaseQueryExecutor(engineReturning("hop1" to ranked("PG1" to 30L))) + val prepared = + PreparedQuery.of( + ActionbaseQuery( + fetch = listOf(self(name = "hop1", source = value("{entity}"))), + transform = + listOf( + ActionbaseQuery.Transform.Sql( + name = "result", + sql = "SELECT hop1.target AS t FROM hop1 -- keep {floor}\n", + ), + ), + ), + ) + + prepared.parameters shouldBe setOf("entity", "floor") + + shouldThrow { + executor.query(prepared, mapOf("entity" to "U1", "floor" to 20L)).block() + }.message shouldBe "This transform takes 0 arguments, got 1." + } + + @Test + fun `an included step comes back alongside the transform`() { + val executor = ActionbaseQueryExecutor(engineReturning("hop1" to ranked("PG1" to 30L))) + val prepared = + PreparedQuery.of( + ActionbaseQuery( + fetch = listOf(self(name = "hop1", source = value("{entity}"), include = true)), + transform = listOf(ActionbaseQuery.Transform.Sql(name = "result", sql = "SELECT hop1.target AS t FROM hop1")), + ), + ) + + executor.query(prepared, mapOf("entity" to "U1")).block()!!.keys shouldBe setOf("hop1", "result") + } + + private companion object { + val rankedSchema = + StructType( + listOf( + StructField("source", PrimitiveType.STRING, "", false), + StructField("target", PrimitiveType.STRING, "", false), + StructField("metric", PrimitiveType.LONG, "", false), + ), + ) + + fun ranked(vararg entries: Pair): DataFrame = + DataFrame( + entries.map { (target, metric) -> Row(mapOf("source" to "U1", "target" to target, "metric" to metric), rankedSchema) }, + rankedSchema, + total = entries.size.toLong(), + ) + + fun value(vararg values: Any): ActionbaseQuery.Vertex = ActionbaseQuery.Vertex.Value(values.toList()) + + fun self( + name: String, + source: ActionbaseQuery.Vertex, + include: Boolean = false, + ): ActionbaseQuery.Item.Self = ActionbaseQuery.Item.Self(name = name, database = "market", table = "orders", source = source, include = include) + + fun query(source: ActionbaseQuery.Vertex): ActionbaseQuery = ActionbaseQuery(fetch = listOf(self(name = "hop1", source = source))) + + /** A stub engine whose every step returns [frame], so the test is about transforms, not storage. */ + fun engineReturning(frame: Pair): QueryEngine { + val binding = mockk() + every { binding.gets(any(), any()) } returns Mono.just(frame.second) + // The interface has a default, but a mock answers nothing it was not told to answer. + every { binding.emptyFrame } returns DataFrame(emptyList(), frame.second.schema, total = 0L) + + return object : QueryEngine { + override fun getTableBinding( + database: String, + alias: String, + ): TableBinding = binding + } + } + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/query/TransformRequestTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/query/TransformRequestTest.kt new file mode 100644 index 000000000..955d4f079 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/query/TransformRequestTest.kt @@ -0,0 +1,169 @@ +package com.kakao.actionbase.engine.query + +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.sql.DataFrame +import com.kakao.actionbase.engine.sql.Row +import com.kakao.actionbase.engine.sql.calcite.SqlTransform + +import org.junit.jupiter.api.Test + +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.matchers.shouldBe +import io.mockk.every +import io.mockk.mockk +import reactor.core.publisher.Mono + +/** + * The request shape as it is actually sent: `fetch` steps, a `transform` holding SQL, and `{entity}` + * filled in per call. + */ +class TransformRequestTest { + @Test + fun `the request parses`() { + val query = ActionbaseQuery.from(REQUEST) + + query.fetch.map { it.name } shouldBe listOf("hop1", "hop2") + query.transform.map { it.name } shouldBe listOf("result") + PreparedQuery.of(query).parameters shouldBe setOf("entity") + } + + @Test + fun `a raw newline inside a json string is rejected by json itself`() { + val broken = "{\"fetch\": [], \"transform\": [{\"type\":\"SQL\",\"name\":\"r\",\"sql\":\"SELECT 1\nFROM x\"}]}" + + shouldThrow { ActionbaseQuery.from(broken) } + } + + @Test + fun `the transform in the request runs over the frames the fetch steps returned`() { + // `TOPK` resolves its ranking through table metadata, which a stub has no business faking, so the + // ranking step is spelled as the `SCAN` it becomes underneath. + val query = ActionbaseQuery.from(REQUEST.replace(TOPK_STEP, SCAN_STEP)) + val executor = ActionbaseQueryExecutor(engineReturning(scanned = ranked, fetched = paid), SqlTransform()) + + val result = executor.query(PreparedQuery.of(query), mapOf("entity" to "U1")).block()!! + + result.keys shouldBe setOf("result") + result.getValue("result").rows.map { listOf(it.data["productGroupId"], it.data["metric"], it.data["paidAt"]) } shouldBe + listOf( + listOf("PG1", 30L, 1_700_000_000L), + listOf("PG2", 20L, -1L), + listOf("PG3", 10L, 1_600_000_000L), + ) + } + + private companion object { + /** `SCAN` answers the ranking step and `gets` answers the `GET` step, so the join has two sides. */ + fun engineReturning( + scanned: DataFrame, + fetched: DataFrame, + ): QueryEngine { + val binding = mockk() + every { binding.scan(any(), any(), any(), any(), any(), any(), any(), any()) } returns Mono.just(scanned) + every { binding.gets(any(), any()) } returns Mono.just(fetched) + // The interface has a default, but a mock answers nothing it was not told to answer. + every { binding.emptyFrame } returns DataFrame(emptyList(), scanned.schema, total = 0L) + + return object : QueryEngine { + override fun getTableBinding( + database: String, + alias: String, + ): TableBinding = binding + } + } + + val REQUEST = + """ + { + "fetch": [ + { + "type": "TOPK", + "name": "hop1", + "database": "market", + "table": "market_order_collection_v1", + "topk": "top_product_groups_1y", + "entity": { "type": "VALUE", "value": ["{entity}"] }, + "limit": 100 + }, + { + "type": "GET", + "name": "hop2", + "database": "market", + "table": "market_order_product_group_v1", + "source": { "type": "VALUE", "value": ["{entity}"] }, + "target": { "type": "REF", "ref": "hop1", "field": "target" } + } + ], + "transform": [ + { + "type": "SQL", + "name": "result", + "sql": "SELECT hop1.target AS productGroupId, hop1.metric AS metric, IFNULL(hop2.paidAt, -1) AS paidAt FROM hop1 LEFT JOIN hop2 ON hop1.source = hop2.source AND hop1.target = hop2.target ORDER BY metric DESC, paidAt DESC" + } + ] + } + """.trimIndent() + + val TOPK_STEP = + """ + "type": "TOPK", + "name": "hop1", + "database": "market", + "table": "market_order_collection_v1", + "topk": "top_product_groups_1y", + "entity": { "type": "VALUE", "value": ["{entity}"] }, + """.trimIndent().trim() + + val SCAN_STEP = + """ + "type": "SCAN", + "name": "hop1", + "database": "market", + "table": "market_order_collection_v1", + "index": "top_product_groups_1y", + "direction": "OUT", + "source": { "type": "VALUE", "value": ["{entity}"] }, + """.trimIndent().trim() + + val rankedSchema = + StructType( + listOf( + StructField("source", PrimitiveType.STRING, "", false), + StructField("target", PrimitiveType.STRING, "", false), + StructField("metric", PrimitiveType.LONG, "", false), + ), + ) + + val paidSchema = + StructType( + listOf( + StructField("source", PrimitiveType.STRING, "", false), + StructField("target", PrimitiveType.STRING, "", false), + StructField("paidAt", PrimitiveType.LONG, "", false), + ), + ) + + val ranked = + DataFrame( + listOf("PG1" to 30L, "PG2" to 20L, "PG3" to 10L).map { (target, metric) -> + Row(mapOf("source" to "U1", "target" to target, "metric" to metric), rankedSchema) + }, + rankedSchema, + total = 3L, + ) + + // PG2 is missing: a product group can rank high and still have no recent order. + val paid = + DataFrame( + listOf("PG1" to 1_700_000_000L, "PG3" to 1_600_000_000L).map { (target, paidAt) -> + Row(mapOf("source" to "U1", "target" to target, "paidAt" to paidAt), paidSchema) + }, + paidSchema, + total = 2L, + ) + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt index 9191a77a9..866450e9b 100644 --- a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt @@ -267,6 +267,8 @@ object GraphFixtures { Metadata.labelLabelEntity.name, Metadata.infoLabelEntity.name, Metadata.aliasLabelEntity.name, + Metadata.queryLabelEntity.name, + Metadata.queryAliasLabelEntity.name, Metadata.onlineMetadataLabelV2Entity.name, Metadata.sysNilLabelEntity.name, ) @@ -301,6 +303,16 @@ object GraphFixtures { Metadata.labelLabelEntity.id, null, ), + listOf( + EntityName.withPhase(Metadata.sysServiceName, Metadata.sysQueryLabelName), + Metadata.labelLabelEntity.id, + null, + ), + listOf( + EntityName.withPhase(Metadata.sysServiceName, Metadata.sysQueryAliasLabelName), + Metadata.labelLabelEntity.id, + null, + ), listOf( EntityName.withPhase(Metadata.sysServiceName, Metadata.sysOnlineMetadataLabelV2Name), Metadata.labelLabelEntity.id, diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/PreparedQuerySpec.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/PreparedQuerySpec.kt new file mode 100644 index 000000000..8d623f009 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/PreparedQuerySpec.kt @@ -0,0 +1,373 @@ +package com.kakao.actionbase.v2.engine.v3 + +import com.kakao.actionbase.core.edge.payload.EdgeBulkMutationRequest +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.core.types.PrimitiveType +import com.kakao.actionbase.engine.query.PreparedQuery +import com.kakao.actionbase.engine.service.MetadataStatus +import com.kakao.actionbase.engine.service.MutationService +import com.kakao.actionbase.engine.service.PreparedQueryService +import com.kakao.actionbase.engine.service.QueryService +import com.kakao.actionbase.v2.core.code.Index +import com.kakao.actionbase.v2.core.code.hbase.Order +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 com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.GraphConfig +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.service.ddl.LabelCreateRequest +import com.kakao.actionbase.v2.engine.test.GraphFixtures + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.module.kotlin.readValue + +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.core.spec.style.StringSpec +import io.kotest.matchers.shouldBe +import reactor.kotlin.test.test + +class PreparedQuerySpec : + StringSpec( + { + val database = GraphFixtures.serviceName + val viewTable = "views" + val mapper = jacksonObjectMapper() + + lateinit var graph: Graph + lateinit var mutations: MutationService + lateinit var queries: PreparedQueryService + + beforeTest { + graph = GraphFixtures.create() + val engine = V2BackedEngine(graph) + mutations = MutationService(engine) + queries = PreparedQueryService(graph, QueryService(engine)) + } + + afterTest { + graph.close() + } + + fun createViewTable() { + graph.labelDdl + .create( + EntityName(database, viewTable), + LabelCreateRequest( + desc = "views", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.STRING), + VertexField(VertexType.STRING), + listOf(Field("viewedAt", DataType.LONG, false)), + ), + dirType = DirectionType.OUT, + storage = GraphFixtures.datastoreStorage, + indices = listOf(Index("viewed_desc", listOf(Index.Field("viewedAt", Order.DESC)))), + ), + ).test() + .assertNext { it.status.name shouldBe "CREATED" } + .verifyComplete() + } + + fun seed() { + val request = + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "source": "user1", "target": "item1", "properties": {"viewedAt": 300}}}, + {"type": "INSERT", "edge": {"version": 1, "source": "user1", "target": "item2", "properties": {"viewedAt": 200}}}, + {"type": "INSERT", "edge": {"version": 1, "source": "user2", "target": "item3", "properties": {"viewedAt": 100}}} + ] + } + """.trimIndent() + mutations + .mutate(database, viewTable, mapper.readValue(request).mutations) + .test() + .assertNext { } + .verifyComplete() + } + + /** `limit` sits where an `Int` goes and `minViewedAt` sits inside SQL, so both routes are covered. */ + fun fetch(): JsonNode = + mapper.readTree( + """ + [ + { + "type": "SCAN", + "name": "viewed", + "database": "$database", + "table": "$viewTable", + "index": "viewed_desc", + "direction": "OUT", + "source": {"type": "VALUE", "value": ["{entity}"]}, + "limit": "{limit}" + } + ] + """.trimIndent(), + ) + + fun transform(sql: String): JsonNode = + mapper.readTree( + """ + [ + {"type": "SQL", "name": "result", "sql": "$sql"} + ] + """.trimIndent(), + ) + + val sql = "SELECT target AS itemId, viewedAt FROM viewed WHERE viewedAt >= {minViewedAt} ORDER BY viewedAt DESC" + + fun arguments(): List = + listOf( + StructField("entity", PrimitiveType.STRING, "조회 대상", false), + StructField("limit", PrimitiveType.INT, "읽어올 행 수", false), + StructField("minViewedAt", PrimitiveType.LONG, "이 값 미만은 버림", false), + ) + + fun register() = queries.register("recent views", arguments(), fetch(), transform(sql)).block()!! + + "a registered query runs by id, with a value bound where an Int goes and inside SQL" { + createViewTable() + seed() + + val registered = register() + registered.arguments.map { it.name } shouldBe listOf("entity", "limit", "minViewedAt") + registered.active shouldBe true + + queries + .query(registered.id, mapOf("entity" to "user1", "limit" to 10, "minViewedAt" to 0)) + .test() + .assertNext { result -> + result.keys shouldBe setOf("result") + result.getValue("result").rows.map { listOf(it.data["itemId"], it.data["viewedAt"]) } shouldBe + listOf(listOf("item1", 300L), listOf("item2", 200L)) + }.verifyComplete() + + // The same registration, a different bound value: the SQL text never changed. + queries + .query(registered.id, mapOf("entity" to "user1", "limit" to 10, "minViewedAt" to 250)) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data["itemId"] } shouldBe listOf("item1") + }.verifyComplete() + } + + "a value of the wrong type is read as the type the registration declared" { + createViewTable() + seed() + + val registered = register() + + queries + .query(registered.id, mapOf("entity" to "user1", "limit" to "10", "minViewedAt" to "250")) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data["itemId"] } shouldBe listOf("item1") + }.verifyComplete() + } + + "a body that uses a name it does not declare is refused" { + createViewTable() + + shouldThrow { + queries + .register( + "missing declarations", + listOf(StructField("entity", PrimitiveType.STRING, "", false)), + fetch(), + transform(sql), + ).block() + } + } + + /** + * A type that reads as the slot's own is accepted, even when it is not the same type: Jackson + * coerces `"1"` into the `Int` that `limit` wants, so declaring `STRING` there is not an error + * and does not have to be treated as one. + */ + "a declared type only has to read back as the slot's own" { + createViewTable() + seed() + + val registered = + queries + .register( + "limit declared as text", + listOf( + StructField("entity", PrimitiveType.STRING, "", false), + StructField("limit", PrimitiveType.STRING, "", false), + StructField("minViewedAt", PrimitiveType.LONG, "", false), + ), + fetch(), + transform(sql), + ).block()!! + + queries + .query(registered.id, mapOf("entity" to "user1", "limit" to 10, "minViewedAt" to 250)) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data["itemId"] } shouldBe listOf("item1") + }.verifyComplete() + } + + "a name resolves to the query it points at, and the query is readable either way" { + createViewTable() + seed() + + val registered = register() + + queries + .createAlias("recent_views", "홈 화면", registered.id) + .test() + .assertNext { it.target shouldBe registered.id } + .verifyComplete() + + queries + .get("recent_views") + .test() + .assertNext { it.id shouldBe registered.id } + .verifyComplete() + + queries + .query("recent_views", mapOf("entity" to "user2", "limit" to 10, "minViewedAt" to 0)) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data["itemId"] } shouldBe listOf("item3") + }.verifyComplete() + } + + "a name can be moved to another query and back" { + createViewTable() + seed() + + val first = register() + val second = queries.register("only the id", arguments(), fetch(), transform(sql)).block()!! + + queries.createAlias("recent_views", "홈 화면", first.id).block() + + queries + .updateAlias("recent_views", target = second.id) + .test() + .assertNext { it.target shouldBe second.id } + .verifyComplete() + + queries + .updateAlias("recent_views", target = first.id) + .test() + .assertNext { it.target shouldBe first.id } + .verifyComplete() + } + + "a query a name still points at cannot be dropped" { + createViewTable() + + val registered = register() + queries.createAlias("recent_views", "홈 화면", registered.id).block() + + shouldThrow { queries.delete(registered.id).block() } + + queries.deleteAlias("recent_views").test().verifyComplete() + queries.delete(registered.id).test().verifyComplete() + + shouldThrow { queries.get(registered.id).block() } + queries + .list() + .test() + .assertNext { it shouldBe emptyList() } + .verifyComplete() + } + + /** + * The alias scan stops at `metadataFetchLimit` without saying so, so a page filled to it cannot + * show that nothing names the query. Here the one alias the scan returns names a different query, + * which read literally would clear the delete — the names it could not reach are the point. + */ + "a query is not dropped on an alias scan that may be truncated" { + val truncating = GraphFixtures.create(GraphConfig.Builder().withMetadataFetchLimit(1), withTestData = false) + try { + val service = PreparedQueryService(truncating, QueryService(V2BackedEngine(truncating))) + val named = service.register("named", arguments(), fetch(), transform(sql)).block()!! + val other = service.register("other", arguments(), fetch(), transform(sql)).block()!! + service.createAlias("recent_views", "홈 화면", named.id).block() + + shouldThrow { service.delete(other.id).block() } + } finally { + truncating.close() + } + } + + "replacing the transform leaves the fetch steps as they were" { + createViewTable() + seed() + + val registered = register() + val amended = "SELECT target AS itemId FROM viewed WHERE viewedAt >= {minViewedAt}" + + queries + .amend(registered.id, transform = transform(amended)) + .test() + .assertNext { it.fetch shouldBe registered.fetch } + .verifyComplete() + + queries + .query(registered.id, mapOf("entity" to "user1", "limit" to 10, "minViewedAt" to 0)) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data.keys.toList() } shouldBe listOf(listOf("itemId"), listOf("itemId")) + }.verifyComplete() + } + + "listing answers with what the status asks for" { + createViewTable() + + val registered = register() + + queries + .list() + .test() + .assertNext { it.map { query -> query.id } shouldBe listOf(registered.id) } + .verifyComplete() + queries + .list(MetadataStatus.INACTIVE) + .test() + .assertNext { it shouldBe emptyList() } + .verifyComplete() + queries + .aliases() + .test() + .assertNext { it shouldBe emptyList() } + .verifyComplete() + } + + "calling a query nobody registered says so" { + createViewTable() + + shouldThrow { + queries.query("no_such_query", mapOf("entity" to "user1")).block() + } + } + + /** The same body run without being registered: values come in the call, with no declared types. */ + "a query sent whole runs with the values sent alongside it" { + createViewTable() + seed() + + val prepared = PreparedQuery.adhoc(fetch(), transform(sql), emptySet()) + prepared.parameters shouldBe setOf("entity", "limit", "minViewedAt") + + QueryService(V2BackedEngine(graph)) + .query(prepared, mapOf("entity" to "user1", "limit" to 10, "minViewedAt" to 250)) + .test() + .assertNext { result -> + result.getValue("result").rows.map { it.data["itemId"] } shouldBe listOf("item1") + }.verifyComplete() + } + }, + ) diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/TransformQuerySpec.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/TransformQuerySpec.kt new file mode 100644 index 000000000..298209b18 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/TransformQuerySpec.kt @@ -0,0 +1,292 @@ +package com.kakao.actionbase.v2.engine.v3 + +import com.kakao.actionbase.core.metadata.common.DirectionType as V3DirectionType + +import com.kakao.actionbase.core.edge.payload.EdgeBulkMutationRequest +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.engine.query.ActionbaseQuery +import com.kakao.actionbase.engine.query.ActionbaseQueryExecutor +import com.kakao.actionbase.engine.query.PreparedQuery +import com.kakao.actionbase.engine.service.MutationService +import com.kakao.actionbase.v2.core.code.Index +import com.kakao.actionbase.v2.core.code.hbase.Order +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 com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.service.ddl.LabelCreateRequest +import com.kakao.actionbase.v2.engine.test.GraphFixtures + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.module.kotlin.readValue + +import io.kotest.core.spec.style.StringSpec +import io.kotest.matchers.shouldBe +import reactor.kotlin.test.test + +/** + * The whole request over real tables: a `TOPK` step reads a materialized ranking, a `GET` step reads + * when each ranked target was last paid for, and a `SQL` transform joins the two. + * + * Why the join is an outer one shows up here — a product group can rank high and still have no recent + * order, and it has to survive with a fallback rather than disappear. + */ +class TransformQuerySpec : + StringSpec( + { + val database = GraphFixtures.serviceName + val orderTable = "orders" + val rankTable = "orders__topk" + val paidTable = "paid" + val topkName = "top_purchased" + val mapper = jacksonObjectMapper() + + lateinit var graph: Graph + lateinit var mutationService: MutationService + lateinit var queryExecutor: ActionbaseQueryExecutor + + beforeTest { + graph = GraphFixtures.create() + val engine = V2BackedEngine(graph) + mutationService = MutationService(engine) + queryExecutor = ActionbaseQueryExecutor(engine) + } + + afterTest { + graph.close() + } + + fun createRankTable() { + graph.labelDdl + .create( + EntityName(database, rankTable), + LabelCreateRequest( + desc = "rank rows", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.STRING), + VertexField(VertexType.STRING), + listOf(Field("metric", DataType.LONG, false)), + ), + dirType = DirectionType.OUT, + storage = GraphFixtures.datastoreStorage, + indices = listOf(Index("metric_desc", listOf(Index.Field("metric", Order.DESC)))), + ), + ).test() + .assertNext { it.status.name shouldBe "CREATED" } + .verifyComplete() + } + + fun createOrderTable() { + graph.labelDdl + .create( + EntityName(database, orderTable), + LabelCreateRequest( + desc = "purchases", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.STRING), + VertexField(VertexType.STRING), + listOf(Field("purchasedAt", DataType.LONG, false)), + ), + dirType = DirectionType.BOTH, + storage = GraphFixtures.datastoreStorage, + groups = + listOf( + Group( + group = "purchased_count", + type = GroupType.COUNT, + fields = + listOf( + Group.Field( + "purchasedAt", + bucket = + Bucket.Date( + name = "day", + unit = Bucket.ValueUnit.MILLISECOND, + timezone = "UTC", + format = "yyyy-MM-dd", + ), + ), + ), + directionType = V3DirectionType.OUT, + aggregations = + Aggregations( + topk = listOf(Topk(topk = topkName, entity = "source", dimension = "target", rank = "$database.$rankTable")), + ), + ), + ), + ), + ).test() + .assertNext { it.status.name shouldBe "CREATED" } + .verifyComplete() + } + + fun createPaidTable() { + graph.labelDdl + .create( + EntityName(database, paidTable), + LabelCreateRequest( + desc = "last payment per item", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.STRING), + VertexField(VertexType.STRING), + listOf(Field("paidAt", DataType.LONG, false)), + ), + dirType = DirectionType.OUT, + storage = GraphFixtures.datastoreStorage, + ), + ).test() + .assertNext { it.status.name shouldBe "CREATED" } + .verifyComplete() + } + + fun createTables() { + createRankTable() + createOrderTable() + createPaidTable() + } + + fun seed( + table: String, + mutations: String, + ) { + mutationService + .mutate(database, table, mapper.readValue(mutations).mutations) + .test() + .assertNext { } + .verifyComplete() + } + + /** + * The ranking's `source` is the composed rank key, not the entity, so the join matches on `target` + * alone. Both sides are already limited to one entity by the steps that read them. + */ + fun preparedQuery(): PreparedQuery = + PreparedQuery.of( + ActionbaseQuery( + fetch = + listOf( + ActionbaseQuery.Item.Topk( + name = "hop1", + database = database, + table = orderTable, + topk = topkName, + entity = ActionbaseQuery.Vertex.Value(listOf("{entity}")), + limit = 100, + ), + ActionbaseQuery.Item.Get( + name = "hop2", + database = database, + table = paidTable, + source = ActionbaseQuery.Vertex.Value(listOf("{entity}")), + target = ActionbaseQuery.Vertex.Ref(ref = "hop1", field = "target"), + ), + ), + transform = + listOf( + ActionbaseQuery.Transform.Sql( + name = "result", + sql = + """ + SELECT hop1.target AS productGroupId + , hop1.metric AS metric + , IFNULL(hop2.paidAt, -1) AS paidAt + FROM hop1 + LEFT JOIN hop2 ON hop1.target = hop2.target + ORDER BY metric DESC, paidAt DESC + """.trimIndent(), + ), + ), + ), + ) + + /** + * test.orders__topk (ranking for user1) test.paid + * | target | metric | | target | paidAt | + * |--------|--------| |--------|------------| + * | item1 | 30 | | item1 | 1700000000 | + * | item2 | 20 | | item3 | 1600000000 | + * | item3 | 10 | + * + * `item2` is ranked but never paid for, so the transform has to keep it at `-1`. + */ + "a SQL transform joins a materialized ranking with a GET over real tables" { + createTables() + seed( + rankTable, + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "source": "test|orders|top_purchased|user1", "target": "item1", "properties": {"metric": 30}}}, + {"type": "INSERT", "edge": {"version": 1, "source": "test|orders|top_purchased|user1", "target": "item2", "properties": {"metric": 20}}}, + {"type": "INSERT", "edge": {"version": 1, "source": "test|orders|top_purchased|user1", "target": "item3", "properties": {"metric": 10}}} + ] + } + """.trimIndent(), + ) + seed( + paidTable, + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "source": "user1", "target": "item1", "properties": {"paidAt": 1700000000}}}, + {"type": "INSERT", "edge": {"version": 1, "source": "user1", "target": "item3", "properties": {"paidAt": 1600000000}}} + ] + } + """.trimIndent(), + ) + + val prepared = preparedQuery() + prepared.parameters shouldBe setOf("entity") + + queryExecutor + .query(prepared, mapOf("entity" to "user1")) + .test() + .assertNext { result -> + // Only the transform comes back: neither fetch step asked to be included. + result.keys shouldBe setOf("result") + result + .getValue("result") + .rows + .map { listOf(it.data["productGroupId"], it.data["metric"], it.data["paidAt"]) } shouldBe + listOf( + listOf("item1", 30L, 1_700_000_000L), + listOf("item2", 20L, -1L), + listOf("item3", 10L, 1_600_000_000L), + ) + }.verifyComplete() + } + + /** + * An entity nobody ranked is an ordinary request, not an error. The step finds no rows, and a + * frame with no rows still has to say what its columns are — a transform planned against a + * schema-less frame cannot resolve a single column name. + */ + "an entity with no ranking returns no rows, under the columns it would have had" { + createTables() + + queryExecutor + .query(preparedQuery(), mapOf("entity" to "nobody")) + .test() + .assertNext { result -> + val transformed = result.getValue("result") + transformed.rows shouldBe emptyList() + transformed.schema.fields.map { it.name } shouldBe listOf("productGroupId", "metric", "paidAt") + }.verifyComplete() + } + }, + ) diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/api/exception/ExceptionController.kt b/server/src/main/kotlin/com/kakao/actionbase/server/api/exception/ExceptionController.kt index 4748ee2dc..564c5ed06 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/api/exception/ExceptionController.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/exception/ExceptionController.kt @@ -1,5 +1,7 @@ package com.kakao.actionbase.server.api.exception +import com.kakao.actionbase.engine.service.NoSuchPreparedQueryException + import org.slf4j.LoggerFactory import org.springframework.http.HttpStatus import org.springframework.http.ResponseEntity @@ -45,4 +47,7 @@ class ExceptionController { IllegalArgumentException::class, ) fun handleClientError(e: Exception): Mono> = createErrorResponse(e, HttpStatus.BAD_REQUEST) { logger.error("Bad request occurred: ${e.message}", e) } + + @ExceptionHandler(NoSuchPreparedQueryException::class) + fun handleNotFound(e: Exception): Mono> = createErrorResponse(e, HttpStatus.NOT_FOUND) { logger.info("Not found: ${e.message}") } } diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/QueryController.kt b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/QueryController.kt index 84bba1c12..8e3a4ce0b 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/QueryController.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/QueryController.kt @@ -1,12 +1,16 @@ package com.kakao.actionbase.server.api.graph.v3 -import com.kakao.actionbase.engine.query.ActionbaseQuery +import com.kakao.actionbase.engine.query.PreparedQuery +import com.kakao.actionbase.engine.service.PreparedQueryService import com.kakao.actionbase.engine.service.QueryService import com.kakao.actionbase.engine.sql.DataFrame +import com.kakao.actionbase.server.api.graph.v3.query.PreparedQueryExecuteRequest +import com.kakao.actionbase.server.api.graph.v3.query.QueryExecuteRequest import com.kakao.actionbase.server.util.mapToResponseEntity import com.kakao.actionbase.v2.engine.sql.QueryResult 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 @@ -16,18 +20,29 @@ import reactor.core.publisher.Mono @RestController class QueryController( private val queryService: QueryService, + private val preparedQueryService: PreparedQueryService, ) { @PostMapping("/graph/v3/query") fun query( - @RequestBody actionBaseQuery: ActionbaseQuery, + @RequestBody request: QueryExecuteRequest, ): Mono> = queryService - .query(actionBaseQuery) + .query(PreparedQuery.adhoc(request.fetch, request.transform, request.stats), request.arguments) .map { val items = it.map { entry -> entry.value.toNamedJsonFormat(entry.key) } NamedQueryResult(items) }.mapToResponseEntity() + @PostMapping("/graph/v3/query/{id}") + fun queryById( + @PathVariable id: String, + @RequestBody request: PreparedQueryExecuteRequest, + ): Mono> = + preparedQueryService + .query(id, request.arguments) + .map { NamedQueryResult(it.map { entry -> entry.value.toNamedJsonFormat(entry.key) }) } + .mapToResponseEntity() + private fun DataFrame.toNamedJsonFormat(name: String): QueryResult.NamedJsonFormat { val meta = schema.fields.map { QueryResult.Meta(it.name, it.type.name) } val data = rows.map { it.data } diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryController.kt b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryController.kt new file mode 100644 index 000000000..7415af2b8 --- /dev/null +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryController.kt @@ -0,0 +1,152 @@ +package com.kakao.actionbase.server.api.graph.v3.query + +import com.kakao.actionbase.engine.service.MetadataStatus +import com.kakao.actionbase.engine.service.PreparedQueryAliasDescriptor +import com.kakao.actionbase.engine.service.PreparedQueryDescriptor +import com.kakao.actionbase.engine.service.PreparedQueryService +import com.kakao.actionbase.v2.engine.entity.EntityName + +import org.springframework.http.ResponseEntity +import org.springframework.web.bind.annotation.DeleteMapping +import org.springframework.web.bind.annotation.GetMapping +import org.springframework.web.bind.annotation.PathVariable +import org.springframework.web.bind.annotation.PostMapping +import org.springframework.web.bind.annotation.PutMapping +import org.springframework.web.bind.annotation.RequestBody +import org.springframework.web.bind.annotation.RequestParam +import org.springframework.web.bind.annotation.RestController + +import jakarta.validation.Valid +import reactor.core.publisher.Mono + +/** + * Running a query lives on the query path (`POST /graph/v3/query/{id}`), so that reading one and running + * it never look alike. These paths sit under `/graph/v3` rather than a database, because a query names its + * own databases per step. + */ +@RestController +class PreparedQueryController( + private val preparedQueryService: PreparedQueryService, +) { + private val tenant: String + get() = EntityName.tenant + + @GetMapping("/graph/v3/prepared-queries") + fun listQueries( + @RequestParam(required = false, defaultValue = "ACTIVE") status: MetadataStatus, + ): Mono>> = + preparedQueryService + .list(status) + .map { queries -> ResponseEntity.ok(queries.map { it.toResponse() }) } + + /** [id] takes either form: the id a registration was given, or a name pointing at one. */ + @GetMapping("/graph/v3/prepared-queries/{id}") + fun getQuery( + @PathVariable id: String, + ): Mono> = + preparedQueryService + .get(id) + .map { ResponseEntity.ok(it.toResponse()) } + .defaultIfEmpty(ResponseEntity.notFound().build()) + + @PostMapping("/graph/v3/prepared-queries") + fun createQuery( + @Valid @RequestBody request: PreparedQueryCreateRequest, + ): Mono> = + preparedQueryService + .register( + desc = request.comment, + arguments = request.arguments.map { it.toStructField() }, + fetch = request.fetch, + transform = request.transform, + stats = request.stats, + ).map { ResponseEntity.ok(it.toResponse()) } + + @PutMapping("/graph/v3/prepared-queries/{id}") + fun updateQuery( + @PathVariable id: String, + @Valid @RequestBody request: PreparedQueryUpdateRequest, + ): Mono> = + preparedQueryService + .amend( + id = id, + desc = request.comment, + arguments = request.arguments?.map { it.toStructField() }, + fetch = request.fetch, + transform = request.transform, + stats = request.stats, + ).map { ResponseEntity.ok(it.toResponse()) } + .defaultIfEmpty(ResponseEntity.notFound().build()) + + @DeleteMapping("/graph/v3/prepared-queries/{id}") + fun deleteQuery( + @PathVariable id: String, + ): Mono> = + preparedQueryService + .delete(id) + .then(Mono.just(ResponseEntity.noContent().build())) + + @GetMapping("/graph/v3/prepared-queries/aliases") + fun listAliases( + @RequestParam(required = false, defaultValue = "ACTIVE") status: MetadataStatus, + ): Mono>> = + preparedQueryService + .aliases(status) + .map { aliases -> ResponseEntity.ok(aliases.map { it.toResponse() }) } + + @GetMapping("/graph/v3/prepared-queries/aliases/{alias}") + fun getAlias( + @PathVariable alias: String, + ): Mono> = + preparedQueryService + .alias(alias) + .map { ResponseEntity.ok(it.toResponse()) } + .defaultIfEmpty(ResponseEntity.notFound().build()) + + @PostMapping("/graph/v3/prepared-queries/aliases") + fun createAlias( + @Valid @RequestBody request: PreparedQueryAliasCreateRequest, + ): Mono> = + preparedQueryService + .createAlias(request.alias, request.comment, request.target) + .map { ResponseEntity.ok(it.toResponse()) } + + @PutMapping("/graph/v3/prepared-queries/aliases/{alias}") + fun updateAlias( + @PathVariable alias: String, + @Valid @RequestBody request: PreparedQueryAliasUpdateRequest, + ): Mono> = + preparedQueryService + .updateAlias(alias, request.comment, request.target, request.active) + .map { ResponseEntity.ok(it.toResponse()) } + .defaultIfEmpty(ResponseEntity.notFound().build()) + + @DeleteMapping("/graph/v3/prepared-queries/aliases/{alias}") + fun deleteAlias( + @PathVariable alias: String, + ): Mono> = + preparedQueryService + .deleteAlias(alias) + .then(Mono.just(ResponseEntity.noContent().build())) + + private fun PreparedQueryDescriptor.toResponse(): PreparedQueryResponse = + PreparedQueryResponse( + tenant = tenant, + id = id, + arguments = arguments, + fetch = fetch, + transform = transform, + stats = stats, + active = active, + comment = desc, + ) + + private fun PreparedQueryAliasDescriptor.toResponse(): PreparedQueryAliasResponse = + PreparedQueryAliasResponse( + tenant = tenant, + alias = alias, + target = target, + active = active, + comment = desc, + ) +} diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryPayload.kt b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryPayload.kt new file mode 100644 index 000000000..65f5bc1d5 --- /dev/null +++ b/server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryPayload.kt @@ -0,0 +1,100 @@ +package com.kakao.actionbase.server.api.graph.v3.query + +import com.kakao.actionbase.core.Constants +import com.kakao.actionbase.core.metadata.common.StructField +import com.kakao.actionbase.core.types.PrimitiveType +import com.kakao.actionbase.v2.engine.sql.StatKey + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.JsonNodeFactory + +import jakarta.validation.constraints.NotBlank +import jakarta.validation.constraints.Pattern +import jakarta.validation.constraints.Size + +data class QueryArgumentRequest( + @field:NotBlank(message = "argument name is required") + val name: String, + val type: PrimitiveType, + @field:Size(max = Constants.Name.COMMENT_MAX_LENGTH, message = Constants.Name.COMMENT_SIZE_MESSAGE) + val comment: String = "", +) { + fun toStructField(): StructField = StructField(name = name, type = type, comment = comment, nullable = false) +} + +data class PreparedQueryCreateRequest( + @field:Size(max = Constants.Name.COMMENT_MAX_LENGTH, message = Constants.Name.COMMENT_SIZE_MESSAGE) + val comment: String = "", + val arguments: List = emptyList(), + val fetch: JsonNode = JsonNodeFactory.instance.arrayNode(), + val transform: JsonNode = JsonNodeFactory.instance.arrayNode(), + val stats: Set = emptySet(), +) + +data class PreparedQueryUpdateRequest( + val active: Boolean? = null, + @field:Size(max = Constants.Name.COMMENT_MAX_LENGTH, message = Constants.Name.COMMENT_SIZE_MESSAGE) + val comment: String? = null, + val arguments: List? = null, + val fetch: JsonNode? = null, + val transform: JsonNode? = null, + val stats: Set? = null, +) + +data class PreparedQueryAliasCreateRequest( + @field:NotBlank(message = "alias is required") + @field:Pattern(regexp = Constants.Name.PATTERN, message = Constants.Name.MESSAGE) + val alias: String, + @field:NotBlank(message = "target is required") + val target: String, + @field:Size(max = Constants.Name.COMMENT_MAX_LENGTH, message = Constants.Name.COMMENT_SIZE_MESSAGE) + val comment: String = "", +) + +data class PreparedQueryAliasUpdateRequest( + val active: Boolean? = null, + @field:Size(max = Constants.Name.COMMENT_MAX_LENGTH, message = Constants.Name.COMMENT_SIZE_MESSAGE) + val comment: String? = null, + val target: String? = null, +) + +data class PreparedQueryExecuteRequest( + val arguments: Map = emptyMap(), +) + +/** `arguments` carries values rather than a declaration, so a placeholder takes the type it was sent as. */ +data class QueryExecuteRequest( + val arguments: Map = emptyMap(), + val fetch: JsonNode = JsonNodeFactory.instance.arrayNode(), + val transform: JsonNode = JsonNodeFactory.instance.arrayNode(), + val stats: Set = emptySet(), +) + +data class PreparedQueryResponse( + val tenant: String, + val id: String, + val arguments: List, + val fetch: JsonNode, + val transform: JsonNode, + val stats: Set, + val active: Boolean = true, + val comment: String = Constants.DEFAULT_COMMENT, + val revision: Long = Constants.DEFAULT_REVISION, + val createdAt: Long = Constants.DEFAULT_CREATED_AT, + val createdBy: String = Constants.DEFAULT_CREATED_BY, + val updatedAt: Long = Constants.DEFAULT_UPDATED_AT, + val updatedBy: String = Constants.DEFAULT_UPDATED_BY, +) + +data class PreparedQueryAliasResponse( + val tenant: String, + val alias: String, + val target: String, + val active: Boolean = true, + val comment: String = Constants.DEFAULT_COMMENT, + val revision: Long = Constants.DEFAULT_REVISION, + val createdAt: Long = Constants.DEFAULT_CREATED_AT, + val createdBy: String = Constants.DEFAULT_CREATED_BY, + val updatedAt: Long = Constants.DEFAULT_UPDATED_AT, + val updatedBy: String = Constants.DEFAULT_UPDATED_BY, +) 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 bc2c19ce5..69aaf0a36 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 @@ -5,6 +5,7 @@ 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.PreparedQueryService import com.kakao.actionbase.engine.service.QueryService import com.kakao.actionbase.engine.service.aggregation.AggregationHandler import com.kakao.actionbase.engine.service.aggregation.TopkAggregationHandler @@ -147,6 +148,25 @@ class GraphConfiguration { @Bean fun provideQueryService(engine: V2BackedEngine): QueryService = QueryService(engine) + @Bean + fun providePreparedQueryService( + graph: Graph, + queryService: QueryService, + ): PreparedQueryService = PreparedQueryService(graph, queryService) + + @Bean + fun provideMutationService( + engine: V2BackedEngine, + graph: Graph, + ): MutationService = MutationService(engine, graph.featureFlags) + + @Bean + fun provideQueueService( + graph: Graph, + mutationService: MutationService, + queryService: QueryService, + ): QueueService = QueueService(graph, mutationService, queryService) + @Bean fun provideTopkAggregationHandler( queryService: QueryService, @@ -166,17 +186,4 @@ class GraphConfiguration { queryService: QueryService, engine: V2BackedEngine, ): AggregationQueryService = AggregationQueryService(queryService, engine) - - @Bean - fun provideMutationService( - engine: V2BackedEngine, - graph: Graph, - ): MutationService = MutationService(engine, graph.featureFlags) - - @Bean - fun provideQueueService( - graph: Graph, - mutationService: MutationService, - queryService: QueryService, - ): QueueService = QueueService(graph, mutationService, queryService) } diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilter.kt b/server/src/main/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilter.kt index 4afa72519..f7b3ea85f 100644 --- a/server/src/main/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilter.kt +++ b/server/src/main/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilter.kt @@ -11,7 +11,7 @@ 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, or by readSegments when the path ends in +// Exceptions: read-only POST endpoints, matched by readSuffixes, or by readPaths 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) @@ -24,7 +24,15 @@ class ReadOnlyRequestFilter : WebFilter { "/multi-edges/ids", "/query", ) - private val readSegments = setOf("/topks/") + + // Read-only POSTs whose path ends in a variable, so `readSuffixes` has no fixed tail to match on. + // Anchored on the whole path rather than searched for inside it: `contains("/query/")` would also + // let through every write under a database named `query`, and the names here are ordinary ones. + private val readPaths = + listOf( + Regex("^/graph/v3/query/[^/]+$"), + Regex("^/aggregations/v1/databases/[^/]+/tables/[^/]+/topks/[^/]+$"), + ) init { log.info("ReadOnlyRequestFilter is active. Write operations on {} will be rejected.", paths) @@ -52,5 +60,5 @@ class ReadOnlyRequestFilter : WebFilter { return exchange.response.writeWith(Mono.just(messageBuffer)) } - private fun isRead(path: String): Boolean = readSuffixes.any { path.endsWith(it) } || readSegments.any { path.contains(it) } + private fun isRead(path: String): Boolean = readSuffixes.any { path.endsWith(it) } || readPaths.any { it.matches(path) } } diff --git a/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryE2ETest.kt b/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryE2ETest.kt new file mode 100644 index 000000000..1474c961f --- /dev/null +++ b/server/src/test/kotlin/com/kakao/actionbase/server/api/graph/v3/query/PreparedQueryE2ETest.kt @@ -0,0 +1,416 @@ +package com.kakao.actionbase.server.api.graph.v3.query + +import com.kakao.actionbase.server.test.E2ETestBase + +import java.time.Duration + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.TestInstance +import org.springframework.http.MediaType + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class PreparedQueryE2ETest : E2ETestBase() { + private val db = "shop" + private val table = "purchases_table" + private val rank = "${table}__topk" + private val rankFqn = "$db.$rank" + private val topkName = "top_purchased" + private val mapper = jacksonObjectMapper() + + @BeforeAll + fun setup() { + client + .post() + .uri("/graph/v3/databases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"database": "$db", "comment": "test"}""") + .exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/graph/v3/databases/$db/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$rank", + "schema": { + "type": "EDGE", + "source": {"type": "string", "comment": "database|table|topk|entity|dimensionValues"}, + "target": {"type": "string", "comment": "topkDimensionValue"}, + "properties": [ + {"name": "metric", "type": "long", "comment": "aggregated metric", "nullable": false}, + {"name": "additionalProperties", "type": "string", "comment": "carried properties as JSON", "nullable": true} + ], + "direction": "OUT", + "indexes": [ + {"index": "metric_desc", "fields": [{"field": "metric", "order": "DESC"}]} + ], + "groups": [], + "caches": [] + }, + "storage": "datastore://test_namespace/$rank", + "mode": "SYNC", + "comment": "topk rank rows" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/graph/v3/databases/$db/tables") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "table": "$table", + "schema": { + "type": "MULTI_EDGE", + "id": {"type": "long", "comment": "order id"}, + "source": {"type": "string", "comment": "user"}, + "target": {"type": "string", "comment": "item"}, + "properties": [ + {"name": "category", "type": "string", "comment": "item category", "nullable": false}, + {"name": "purchasedAt", "type": "long", "comment": "purchase time ms", "nullable": false} + ], + "direction": "OUT", + "indexes": [], + "groups": [{ + "group": "purchased_count", + "type": "COUNT", + "fields": [ + {"name": "_target"}, + {"name": "category"}, + {"name": "purchasedAt", "bucket": {"type": "date", "name": "day", "unit": "MILLISECOND", "timezone": "UTC", "format": "yyyy-MM-dd"}} + ], + "directionType": "OUT", + "aggregations": { + "topk": [{ + "topk": "$topkName", + "entity": "source", + "ranges": "_target:eq:{_target};category:eq:{category};day:bt:2023-11-14,2024-11-13", + "dimension": "target", + "rank": "$rankFqn", + "additionalProperties": ["category"] + }] + } + }], + "caches": [] + }, + "storage": "datastore://test_namespace/$table", + "mode": "SYNC", + "comment": "user-to-item purchase edges" + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + seedRanking() + } + + /** Two purchases of `item1` and one of `item2`, then the aggregate call that writes the rank rows. */ + private fun seedRanking() { + client + .post() + .uri("/graph/v3/databases/$db/tables/$table/multi-edges") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "mutations": [ + {"type": "INSERT", "edge": {"version": 1, "id": 1, "source": "user1", "target": "item1", "properties": {"category": "grocery", "purchasedAt": 1710000000000}}}, + {"type": "INSERT", "edge": {"version": 1, "id": 2, "source": "user1", "target": "item1", "properties": {"category": "grocery", "purchasedAt": 1710000000000}}}, + {"type": "INSERT", "edge": {"version": 1, "id": 3, "source": "user1", "target": "item2", "properties": {"category": "grocery", "purchasedAt": 1710000000000}}} + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + client + .post() + .uri("/aggregations/v1/aggregate") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "items": [ + {"database": "$db", "table": "$table", + "edge": {"version": 1, "source": "user1", "target": "item1", "properties": {"category": "grocery", "purchasedAt": 1710000000000}, "context": {}}}, + {"database": "$db", "table": "$table", + "edge": {"version": 1, "source": "user1", "target": "item2", "properties": {"category": "grocery", "purchasedAt": 1710000000000}, "context": {}}} + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + + warmUpTheTransform() + } + + /** + * Every test here registers the same SQL, so the first one to run it parses, plans and opens the + * statement's sessions while the rest read a cached plan — `CalciteLatencyProbe` puts that at several + * milliseconds warm and far more cold, and a registered query does not pay it at registration yet. + * Spend it once here, so the default five second response timeout keeps guarding the path every test + * actually takes. This one call gets longer: on a loaded build it has been seen past five seconds, and + * the wait is the cost being measured rather than a symptom of anything the tests are checking. + */ + private fun warmUpTheTransform() { + client + .mutate() + .responseTimeout(Duration.ofSeconds(60)) + .build() + .post() + .uri("/graph/v3/query/${register()}") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"arguments": {"entity": "user1", "limit": 100}}""") + .exchange() + .expectStatus() + .isOk + } + + /** `limit` is declared an `INT` and written as a placeholder, so the cast has to happen for it to bind. */ + private fun register(): String { + val body = + client + .post() + .uri("/graph/v3/prepared-queries") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "comment": "top purchased grocery items for one user", + "arguments": [ + {"name": "entity", "type": "string", "comment": "the user asked about"}, + {"name": "limit", "type": "int", "comment": "how many ranked rows to read"} + ], + "fetch": [ + { + "type": "TOPK", + "name": "ranked", + "database": "$db", + "table": "$table", + "topk": "$topkName", + "entity": {"type": "VALUE", "value": ["{entity}"]}, + "dimensionValues": {"category": "grocery"}, + "limit": "{limit}" + } + ], + "transform": [ + { + "type": "SQL", + "name": "result", + "sql": "SELECT target AS itemId, metric FROM ranked ORDER BY metric DESC" + } + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + .expectBody(String::class.java) + .returnResult() + .responseBody!! + + val registered = mapper.readTree(body) + assertEquals("top purchased grocery items for one user", registered["comment"].asText()) + assertEquals(listOf("entity", "limit"), registered["arguments"].map { it["name"].asText() }) + // The body comes back as it was registered, placeholder and all. + assertEquals("{limit}", registered["fetch"][0]["limit"].asText()) + return registered["id"].asText() + } + + private fun name( + alias: String, + target: String, + ) { + client + .post() + .uri("/graph/v3/prepared-queries/aliases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"alias": "$alias", "target": "$target", "comment": "홈 화면 추천"}""") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.target") + .isEqualTo(target) + } + + @Test + fun `a registered query runs by the id it was given`() { + val id = register() + + client + .post() + .uri("/graph/v3/query/$id") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"arguments": {"entity": "user1", "limit": 100}}""") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.items.length()") + .isEqualTo(1) + .jsonPath("$.items[0].name") + .isEqualTo("result") + .jsonPath("$.items[0].data.length()") + .isEqualTo(2) + .jsonPath("$.items[0].data[0].itemId") + .isEqualTo("item1") + .jsonPath("$.items[0].data[0].metric") + .isEqualTo(2) + .jsonPath("$.items[0].data[1].itemId") + .isEqualTo("item2") + } + + /** `limit` arrives as text, and the declared `INT` is what makes it usable where a number goes. */ + @Test + fun `a value sent as text is read as the type the registration declared`() { + val id = register() + + client + .post() + .uri("/graph/v3/query/$id") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"arguments": {"entity": "user1", "limit": "1"}}""") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.items[0].data.length()") + .isEqualTo(1) + .jsonPath("$.items[0].data[0].itemId") + .isEqualTo("item1") + } + + @Test + fun `a named query runs and reads back under its name`() { + val id = register() + name("top_grocery", id) + + client + .get() + .uri("/graph/v3/prepared-queries/top_grocery") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.id") + .isEqualTo(id) + .jsonPath("$.active") + .isEqualTo(true) + + client + .post() + .uri("/graph/v3/query/top_grocery") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"arguments": {"entity": "user1", "limit": 100}}""") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.items[0].data[0].itemId") + .isEqualTo("item1") + } + + @Test + fun `queries and names are listed on their own paths`() { + val id = register() + name("listed_grocery", id) + + client + .get() + .uri("/graph/v3/prepared-queries") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$[?(@.id == '$id')].active") + .isEqualTo(true) + + client + .get() + .uri("/graph/v3/prepared-queries/aliases") + .exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$[?(@.alias == 'listed_grocery')].target") + .isEqualTo(id) + } + + /** A body sent whole, with its values alongside it: the same query, never registered. */ + @Test + fun `a query sent whole runs with the values sent alongside it`() { + client + .post() + .uri("/graph/v3/query") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue( + """ + { + "arguments": {"entity": "user1", "limit": 100}, + "fetch": [ + { + "type": "TOPK", + "name": "ranked", + "database": "$db", + "table": "$table", + "topk": "$topkName", + "entity": {"type": "VALUE", "value": ["{entity}"]}, + "dimensionValues": {"category": "grocery"}, + "limit": "{limit}", + "include": true + } + ] + } + """.trimIndent(), + ).exchange() + .expectStatus() + .isOk + .expectBody() + .jsonPath("$.items[0].name") + .isEqualTo("ranked") + .jsonPath("$.items[0].data.length()") + .isEqualTo(2) + } + + @Test + fun `calling a query nobody registered is a 404`() { + client + .post() + .uri("/graph/v3/query/no_such_query") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"arguments": {"entity": "user1", "limit": 100}}""") + .exchange() + .expectStatus() + .isNotFound + } + + @Test + fun `a name shaped like an id is refused`() { + val id = register() + + client + .post() + .uri("/graph/v3/prepared-queries/aliases") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue("""{"alias": "$id", "target": "$id", "comment": "id shaped"}""") + .exchange() + .expectStatus() + .isBadRequest + } +} 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 fbb662a11..5d3e89ff7 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 @@ -70,6 +70,20 @@ class ReadOnlyRequestFilterTest { assertAllowed(method, path) } + /** + * `query` and `topks` are names a caller may pick for a database or a table, so a read-only endpoint + * recognised by the segment appearing anywhere in the path would hand every write under such a name + * through. The endpoints these stand for sit at a fixed depth, so they are matched whole. + */ + @Test + fun `a database named after a read-only endpoint does not turn its writes into reads`() { + assertBlocked("POST", "/graph/v3/databases/query/tables") + assertBlocked("POST", "/graph/v3/databases/query/tables/orders/edges") + assertBlocked("DELETE", "/graph/v3/databases/query/tables/orders") + assertBlocked("POST", "/graph/v3/databases/topks/tables") + assertBlocked("POST", "/graph/v3/databases/db/tables/topks/edges") + } + // No /control write exists yet, so nothing else would notice the prefix leaving the filter. @Test fun `should block writes on the control prefix`() { @@ -161,6 +175,7 @@ class ReadOnlyRequestFilterTest { "GET /graph/v2/storage/{storage}", // v3 GET "GET /graph/v3", + "GET /aggregations/v1/metadata", "GET /graph/v3/databases", "GET /graph/v3/databases/{database}", "GET /graph/v3/databases/{database}/aliases", @@ -175,7 +190,10 @@ 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/prepared-queries", + "GET /graph/v3/prepared-queries/{id}", + "GET /graph/v3/prepared-queries/aliases", + "GET /graph/v3/prepared-queries/aliases/{alias}", "GET /graph/v3/datastore", "GET /graph/v3/datastore/hbase/namespaces", "GET /graph/v3/datastore/hbase/references", @@ -185,9 +203,10 @@ class ReadOnlyRequestFilterTest { "GET /graph/v3/datastore/hbase/tables/{tableName}/references", // read-only POST "POST /graph/v3/query", - "POST /aggregations/v1/databases/{database}/tables/{table}/topks/{topk}", + "POST /graph/v3/query/{id}", "POST /graph/v3/databases/{database}/tables/{table}/edges/get", "POST /graph/v3/databases/{database}/tables/{table}/multi-edges/ids", + "POST /aggregations/v1/databases/{database}/tables/{table}/topks/{topk}", // queue/v1 GET "GET /queue/v1/namespaces/{namespace}/queues/{queue}", "GET /queue/v1/namespaces/{namespace}/queues/{queue}/partitions", @@ -225,6 +244,12 @@ class ReadOnlyRequestFilterTest { "PUT /graph/v2/service/{service}/label/{label}/edge/sync", "PUT /graph/v2/storage/{storage}", // v3 mutation + "POST /graph/v3/prepared-queries", + "PUT /graph/v3/prepared-queries/{id}", + "DELETE /graph/v3/prepared-queries/{id}", + "POST /graph/v3/prepared-queries/aliases", + "PUT /graph/v3/prepared-queries/aliases/{alias}", + "DELETE /graph/v3/prepared-queries/aliases/{alias}", "DELETE /graph/v3/databases/{database}", "DELETE /graph/v3/databases/{database}/aliases/{alias}", "DELETE /graph/v3/databases/{database}/tables/{table}", @@ -324,14 +349,15 @@ class ReadOnlyRequestFilterTest { "name" to "n", "namespace" to "n", "partition" to "0", - "topk" to "tk", "queue" to "q", "service" to "s", "storage" to "st", "table" to "t", "tableFullName" to "t", "tableName" to "t", + "id" to "q", "tenant" to "alpha", + "topk" to "tk", ) private fun resolvePath(template: String): String =