feat(sql): add structuralSourceColumnNames and runSelect to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
cf9a017913
commit
b2e64dc4f5
1 file changed
+153
@@ -2735,3 +2735,156 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun structuralSourceColumnNames(item: FromItem, localCtes: Map<String, List<String>>): List<String> =
|
||||||
|
when (item) {
|
||||||
|
is TableSource -> {
|
||||||
|
val key = item.name.lowercase()
|
||||||
|
localCtes[key]
|
||||||
|
?: activeCtes.lastOrNull { it.name.equals(item.name, ignoreCase = true) }?.schema?.columnNames
|
||||||
|
?: catalog.temporaryTableOrNull(item.name)?.schema?.columnNames
|
||||||
|
?: catalog.tableOrNull(item.name)?.schema?.columnNames
|
||||||
|
?: catalog.viewOrNull(item.name)?.let { structuralColumnNames(it, localCtes) }
|
||||||
|
?: throw DiskExecutorException("no such table: ${item.name}")
|
||||||
|
}
|
||||||
|
is SubquerySource -> structuralColumnNames(item.select, localCtes)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Runs [statement] against the disk backend inside the already-open [txn]. [outer] is
|
||||||
|
* non-null exactly when this call is itself a scalar/EXISTS/IN subquery nested inside another
|
||||||
|
* statement's expression, reached through [Evaluator]'s [SubqueryRunner] hook - it lets a
|
||||||
|
* column [statement]'s own FROM/JOIN sources cannot resolve fall back to the row of whichever
|
||||||
|
* statement contains it, mirroring [kestrel.engine.SelectRunner.run]'s own `outer` parameter.
|
||||||
|
* A subquery therefore runs inside the same disk transaction as its containing statement
|
||||||
|
* rather than opening (and committing/aborting) one of its own.
|
||||||
|
*/
|
||||||
|
private fun runSelect(statement: SelectStatement, txn: Long, outer: ColumnResolver?): SelectResult {
|
||||||
|
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 })
|
||||||
|
val subqueries = subqueryRunnerFor(txn)
|
||||||
|
var combined = initialRows(statement, txn, statement.from, fromSources.first())
|
||||||
|
statement.joins.forEachIndexed { position, join ->
|
||||||
|
join.on?.let { rejectWindowFunction(it, "an ON condition") }
|
||||||
|
val rightSource = sources[position + 1]
|
||||||
|
val lateralSelect = lateralSelectOf(join.from)
|
||||||
|
combined = if (lateralSelect != null) {
|
||||||
|
executeLateralJoin(join, combined, lateralSelect, sources.subList(0, position + 1), rightSource, outer, subqueries, txn)
|
||||||
|
} else {
|
||||||
|
val rightRows = fromSources[position + 1].scan(txn)
|
||||||
|
applyJoin(join, combined, rightRows, sources.subList(0, position + 2), outer, subqueries, rightSource)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
statement.where?.let { where ->
|
||||||
|
rejectWindowFunction(where, "WHERE")
|
||||||
|
combined = combined.filter { row -> isTrue(evaluateCombinedRow(sources, outer, subqueries, where, row)) }
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
return GroupAggregation(
|
||||||
|
{ expression -> resolveIndex(sources, expression) },
|
||||||
|
{ expression, row -> evaluateCombinedRow(sources, outer, subqueries, expression, row) },
|
||||||
|
::columnLabel,
|
||||||
|
subqueries,
|
||||||
|
{ expression -> declaredCollationOrNull(sources, expression) }
|
||||||
|
).aggregate(statement, orderKeys, combined)
|
||||||
|
}
|
||||||
|
val baseColumnCount = sources.sumOf { it.schema.columns.size }
|
||||||
|
if (windowExpressions.isNotEmpty()) {
|
||||||
|
val evaluateRow: (Expression, List<Value>) -> Value = { expression, row -> evaluateCombinedRow(sources, outer, subqueries, expression, row) }
|
||||||
|
val perExpression = windowExpressions.map { window ->
|
||||||
|
computeWindowValues(window, combined, evaluateRow) { expression -> declaredCollationOrNull(sources, expression) }
|
||||||
|
}
|
||||||
|
combined = combined.mapIndexed { rowIndex, row -> row + windowExpressions.indices.map { perExpression[it][rowIndex] } }
|
||||||
|
}
|
||||||
|
if (orderKeys.isNotEmpty()) {
|
||||||
|
combined = sortCombinedRows(sources, outer, subqueries, combined, orderKeys, windowExpressions, baseColumnCount)
|
||||||
|
}
|
||||||
|
if (statement.distinctOn.isNotEmpty()) {
|
||||||
|
val kept = LinkedHashMap<List<Value>, List<Value>>()
|
||||||
|
for (row in combined) {
|
||||||
|
val key = statement.distinctOn.map { evaluateWindowAware(sources, outer, subqueries, windowExpressions, baseColumnCount, it, row) }
|
||||||
|
val bucketKey = collationFoldedKey(key, statement.distinctOn) { expression -> declaredCollationOrNull(sources, expression) }
|
||||||
|
kept.putIfAbsent(bucketKey, row)
|
||||||
|
}
|
||||||
|
combined = kept.values.toList()
|
||||||
|
}
|
||||||
|
statement.offset?.let { if (it < 0) throw DiskExecutorException("OFFSET must not be negative") }
|
||||||
|
statement.limit?.let { if (it < 0) throw DiskExecutorException("LIMIT must not be negative") }
|
||||||
|
if (!statement.distinct) {
|
||||||
|
if (statement.offset != null || statement.limit != null) {
|
||||||
|
combined = combined.drop(statement.offset ?: 0)
|
||||||
|
statement.limit?.let { combined = combined.take(it) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val (columns, projected) = projectCombined(statement.projection, sources, outer, subqueries, combined, windowExpressions, baseColumnCount)
|
||||||
|
var resultRows = projected
|
||||||
|
if (statement.distinct) {
|
||||||
|
if (statement.distinctOn.isEmpty()) {
|
||||||
|
val itemExpressions = when (statement.projection) {
|
||||||
|
is ExpressionProjection -> statement.projection.items.map { it.expression }
|
||||||
|
is StarProjection -> columns.map { ColumnExpression(it) }
|
||||||
|
}
|
||||||
|
resultRows = resultRows.distinctBy { row -> collationFoldedKey(row, itemExpressions) { expression -> declaredCollationOrNull(sources, expression) } }
|
||||||
|
}
|
||||||
|
if (statement.offset != null || statement.limit != null) {
|
||||||
|
resultRows = resultRows.drop(statement.offset ?: 0)
|
||||||
|
statement.limit?.let { resultRows = resultRows.take(it) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return SelectResult(columns, resultRows)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* `ORDER BY` may name a `SELECT`-list alias instead of repeating its expression (e.g.
|
||||||
|
* `SELECT SUM(amount) AS total ... ORDER BY total`) - resolved here, once, before either
|
||||||
|
* [GroupAggregation] or [sortCombinedRows] ever sees the order keys, mirroring
|
||||||
|
* [kestrel.engine.SelectRunner]'s own `resolveOrderAlias`. A qualified column
|
||||||
|
* (`ORDER BY t.total`) can never be an alias reference, so it is left untouched.
|
||||||
|
*/
|
||||||
|
private fun resolvedOrderKeys(statement: SelectStatement): List<OrderKey> =
|
||||||
|
statement.orderBy?.keys.orEmpty().map { key -> OrderKey(resolveOrderAlias(statement.projection, key.expression), key.descending, key.nullsFirst) }
|
||||||
|
|
||||||
|
private fun resolveOrderAlias(projection: Projection, expression: Expression): Expression {
|
||||||
|
if (expression is ColumnExpression && expression.qualifier == null && projection is ExpressionProjection) {
|
||||||
|
val aliased = projection.items.firstOrNull { it.alias?.equals(expression.name, ignoreCase = true) == true }
|
||||||
|
if (aliased != null) return aliased.expression
|
||||||
|
}
|
||||||
|
return expression
|
||||||
|
}
|
||||||
|
|
||||||
|
/** One joined table's schema plus the offset of its first column within a combined `List<Value>` row. */
|
||||||
|
private class JoinSource(val alias: String, val schema: TableSchema, val offset: Int)
|
||||||
|
|
||||||
|
private fun buildJoinSources(fromItems: List<FromItem>, schemas: List<TableSchema>): List<JoinSource> {
|
||||||
|
val sources = mutableListOf<JoinSource>()
|
||||||
|
var offset = 0
|
||||||
|
for ((item, schema) in fromItems.zip(schemas)) {
|
||||||
|
val alias = sourceLabel(item)
|
||||||
|
if (sources.any { it.alias.equals(alias, ignoreCase = true) }) {
|
||||||
|
throw DiskExecutorException("duplicate table alias: $alias")
|
||||||
|
}
|
||||||
|
sources.add(JoinSource(alias, schema, offset))
|
||||||
|
offset += schema.columns.size
|
||||||
|
}
|
||||||
|
return sources
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun sourceLabel(item: FromItem): String = when (item) {
|
||||||
|
is TableSource -> item.alias ?: item.name
|
||||||
|
is SubquerySource -> item.alias
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user