From 28783ebb75c5ff4d9e3d5261737d4ddbf2631b0b Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 6 Sep 2026 23:13:10 +0200 Subject: [PATCH] feat(sql): add explainQuery and explainOperatorLabel to DiskExecutor --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 159 ++++++++++++++++++ 1 file changed, 159 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index 8f73245..08cc6f4 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -2079,3 +2079,162 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { private fun executeExplain(statement: ExplainStatement): StatementResult = withStatementTransaction { txn -> ExplainResult(explainQuery(statement.select, txn)) } + private fun explainQuery(query: QueryStatement, txn: Long): List = when (query) { + is SelectStatement -> explainSelect(query, txn) + is SetOperationStatement -> explainSetOperation(query, txn) + is WithQuery -> explainWith(query, txn) + } + + /** Title-case labels for an EXPLAIN header - deliberately separate from [setOperatorKeyword]'s ALL-CAPS SQL-keyword text used in error messages. */ + private fun explainOperatorLabel(operator: SetOperator): String = when (operator) { + SetOperator.Union -> "Union" + SetOperator.UnionAll -> "Union All" + SetOperator.Intersect -> "Intersect" + SetOperator.IntersectAll -> "Intersect All" + SetOperator.Except -> "Except" + SetOperator.ExceptAll -> "Except All" + } + + private fun explainSetOperation(statement: SetOperationStatement, txn: Long): List { + val header = explainOperatorLabel(statement.operator) + val left = explainQuery(statement.left, txn).map { " -> $it" } + val right = explainQuery(statement.right, txn).map { " -> $it" } + var lines = listOf(header) + left + right + statement.orderBy?.let { lines = listOf("Sort") + lines.map { line -> " $line" } } + statement.offset?.let { lines = listOf("Offset ${statement.offset}") + lines.map { line -> " $line" } } + statement.limit?.let { lines = listOf("Limit ${statement.limit}") + lines.map { line -> " $line" } } + return lines + } + + /** + * Materializes every CTE for real via [buildCteScope] (needed so a query nested under this WITH + * can itself be explained against a real schema/row set), then re-walks each CTE's own query a + * second time purely to render its plan text - the same double pass [kestrel.engine.SelectRunner. + * explainWith] already accepts for the in-memory engine, including the same cost of re-running a + * *nested* recursive CTE's fixed point twice over if one WITH RECURSIVE is nested inside another. + */ + private fun explainWith(query: WithQuery, txn: Long): List { + var pushed = 0 + try { + for (cte in query.ctes) { + activeCtes.add(buildCteScope(query.recursive, cte, txn)) + pushed++ + } + val cteLines = query.ctes.flatMap { cte -> + listOf("CTE ${cte.name}") + explainQuery(cte.query, txn).map { " -> $it" } + } + return cteLines + explainQuery(query.body, txn) + } finally { + repeat(pushed) { activeCtes.removeAt(activeCtes.size - 1) } + } + } + + private fun explainSelect(statement: SelectStatement, txn: Long): List { + val fromItems = listOf(statement.from) + statement.joins.map { it.from } + val fromSources = fromItems.map { item -> resolveFromItem(item, txn) } + val sources = buildJoinSources(fromItems, fromSources.map { it.schema }) + statement.joins.forEach { join -> join.on?.let { rejectWindowFunction(it, "an ON condition") } } + var body = explainJoinLines(statement, sources, fromSources) + statement.where?.let { rejectWindowFunction(it, "WHERE") } + statement.groupBy.forEach { rejectWindowFunction(it, "GROUP BY") } + statement.having?.let { rejectWindowFunction(it, "HAVING") } + statement.distinctOn.forEach { rejectWindowFunction(it, "DISTINCT ON") } + val orderKeys = resolvedOrderKeys(statement) + val aggregated = statement.groupBy.isNotEmpty() || statement.groupingSpec != null || + containsAggregate(statement.projection) || + (statement.having != null && containsAggregate(statement.having)) + if (statement.distinctOn.isNotEmpty()) validateDistinctOn(statement, orderKeys, aggregated) + val windowExpressions = collectWindowExpressions(statement.projection, orderKeys) + if (windowExpressions.isNotEmpty()) validateWindows(windowExpressions, aggregated) + if (aggregated) { + if (statement.projection is StarProjection) { + throw DiskExecutorException("cannot use * with GROUP BY or aggregate functions") + } + validateAggregation(statement, orderKeys, sources) + val keys = statement.groupBy.map { describeExpression(it) } + val header = if (keys.isEmpty()) { + "Aggregate" + } else { + val setCount = GroupingSetExpansion.setsFor(statement.groupingSpec, statement.groupBy).size + val suffix = if (setCount > 1) " ($setCount grouping sets)" else "" + "GroupAggregate keys: ${keys.joinToString(", ")}$suffix" + } + body = listOf(header) + body.map { " $it" } + } + statement.orderBy?.let { body = listOf("Sort") + body.map { " $it" } } + if (statement.distinct) { + val label = if (statement.distinctOn.isNotEmpty()) "Unique (DISTINCT ON)" else "Unique" + body = listOf(label) + body.map { " $it" } + } + statement.offset?.let { body = listOf("Offset ${statement.offset}") + body.map { " $it" } } + statement.limit?.let { body = listOf("Limit ${statement.limit}") + body.map { " $it" } } + return body + } + + private fun explainJoinLines(statement: SelectStatement, sources: List, fromSources: List): List { + var lines = listOf(explainScanLine(statement, fromSources.first())) + statement.joins.forEachIndexed { position, join -> + val right = sources[position + 1] + val lateral = lateralSelectOf(join.from) != null + val label = joinLabel(join.type) + if (lateral) " Lateral" else "" + val header = "Nested Loop $label" + val rightScan = if (lateral) "Lateral Subquery Scan on ${right.schema.name}" else "Seq Scan on ${right.schema.name}" + lines = listOf(header) + lines.map { " -> $it" } + listOf(" -> $rightScan") + } + return lines + } + + /** The driving source's scan line: an indexed [DiskAccessPlan] when there is no JOIN and it is a real disk table, a full scan otherwise - mirrors [initialRows]'s own choice of when to push down. */ + private fun explainScanLine(statement: SelectStatement, fromSource: DiskFromSource): String { + if (statement.joins.isEmpty() && fromSource is DiskTableSource) { + val alias = (statement.from as? TableSource)?.alias + val plan = accessPlanner.plan(fromSource.table.schema, fromSource.table.secondaryIndexes, statement.where, alias) + return describeAccessPlan(plan, fromSource.table.schema, fromSource.table.secondaryIndexes) + } + return "Seq Scan on ${fromSource.schema.name}" + } + + private fun describeAccessPlan(plan: DiskAccessPlan, schema: TableSchema, secondaryIndexes: List): String { + val tableName = schema.name + return when (plan) { + DiskAccessPlan.FullScan -> "Seq Scan on $tableName" + is DiskAccessPlan.PrimaryKeyPoint -> { + val pkColumn = schema.columns[schema.primaryKeyIndex!!].name + "Primary Key Scan on $tableName ($pkColumn = ${literalText(plan.key)})" + } + is DiskAccessPlan.PrimaryKeyRange -> { + val pkColumn = schema.columns[schema.primaryKeyIndex!!].name + "Primary Key Range Scan on $tableName (${rangeDescription(pkColumn, plan.lower, plan.lowerInclusive, plan.upper, plan.upperInclusive)})" + } + is DiskAccessPlan.CompositePrimaryKeyPoint -> { + val parts = schema.primaryKeyColumns.zip(plan.keys) { name, value -> "$name = ${literalText(value)}" } + "Primary Key Scan on $tableName (${parts.joinToString(" AND ")})" + } + is DiskAccessPlan.CompositePrimaryKeyRange -> { + val columnName = schema.primaryKeyColumns[plan.prefix.size] + val prefixParts = schema.primaryKeyColumns.zip(plan.prefix) { name, value -> "$name = ${literalText(value)}" } + val rangePart = rangeDescription(columnName, plan.lower, plan.lowerInclusive, plan.upper, plan.upperInclusive) + "Primary Key Range Scan on $tableName (${(prefixParts + rangePart).joinToString(" AND ")})" + } + is DiskAccessPlan.Secondary -> { + val handle = secondaryIndexes.first { it.name == plan.indexName } + val parts = handle.columnNames.zip(plan.keys) { name, value -> "$name = ${literalText(value)}" } + "Index Scan using ${plan.indexName} on $tableName (${parts.joinToString(" AND ")})" + } + is DiskAccessPlan.SecondaryRange -> { + val handle = secondaryIndexes.first { it.name == plan.indexName } + val columnName = handle.columnNames[plan.prefix.size] + val prefixParts = handle.columnNames.zip(plan.prefix) { name, value -> "$name = ${literalText(value)}" } + val rangePart = rangeDescription(columnName, plan.lower, plan.lowerInclusive, plan.upper, plan.upperInclusive) + "Index Range Scan using ${plan.indexName} on $tableName (${(prefixParts + rangePart).joinToString(" AND ")})" + } + } + } + + private fun rangeDescription(columnName: String, lower: Value?, lowerInclusive: Boolean, upper: Value?, upperInclusive: Boolean): String { + val parts = mutableListOf() + lower?.let { parts.add("$columnName ${if (lowerInclusive) ">=" else ">"} ${literalText(it)}") } + upper?.let { parts.add("$columnName ${if (upperInclusive) "<=" else "<"} ${literalText(it)}") } + return parts.joinToString(" AND ") + } +