feat(sql): add explainQuery and explainOperatorLabel to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
0eff2e8e8b
commit
28783ebb75
1 file changed
+159
@@ -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<String> = 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<String> {
|
||||
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<String> {
|
||||
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<String> {
|
||||
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<JoinSource>, fromSources: List<DiskFromSource>): List<String> {
|
||||
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<SecondaryIndexHandle>): 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<String>()
|
||||
lower?.let { parts.add("$columnName ${if (lowerInclusive) ">=" else ">"} ${literalText(it)}") }
|
||||
upper?.let { parts.add("$columnName ${if (upperInclusive) "<=" else "<"} ${literalText(it)}") }
|
||||
return parts.joinToString(" AND ")
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user