diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
index da057d6..fb7a2ba 100644
--- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
+++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
@@ -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
+    }
+