diff --git a/src/main/kotlin/kestrel/engine/SelectRunner.kt b/src/main/kotlin/kestrel/engine/SelectRunner.kt
index b29c66c..a80f9b9 100644
--- a/src/main/kotlin/kestrel/engine/SelectRunner.kt
+++ b/src/main/kotlin/kestrel/engine/SelectRunner.kt
@@ -1386,29 +1386,61 @@ class SelectRunner(private val catalog: Catalog) {
                     statement.limit?.let { rows = rows.take(it) }
                 }
             }
-            return when (val projection = statement.projection) {
-                is StarProjection -> SelectResult(columnNames, rows)
+            val (labels, projected) = when (val projection = statement.projection) {
+                // rows may carry trailing window-function values appended by appendWindowColumns
+                // (only possible here if ORDER BY names one directly, since `SELECT *` cannot spell
+                // an OVER clause itself) - trimmed back off so the row width matches columnNames.
+                is StarProjection -> columnNames to rows.map { it.subList(0, columnNames.size) }
                 is ExpressionProjection -> {
                     val labels = projection.items.mapIndexed { index, item -> columnLabel(item, index) }
-                    val projected = rows.map { row -> projection.items.map { evaluateRow(it.expression, row) } }
-                    SelectResult(labels, projected)
+                    labels to rows.map { row -> projection.items.map { evaluateRow(it.expression, row) } }
                 }
             }
+            var resultRows = projected
+            if (statement.distinct) {
+                // Plain DISTINCT dedups the *output* row (post-projection); DISTINCT ON already
+                // collapsed to one row per group above, on the pre-projection row, so re-deduping the
+                // projected columns here would incorrectly merge distinct groups that happen to
+                // project to the same visible values.
+                if (statement.distinctOn.isEmpty()) {
+                    val itemExpressions = when (val projection = statement.projection) {
+                        is ExpressionProjection -> projection.items.map { it.expression }
+                        is StarProjection -> columnNames.map { ColumnExpression(it) }
+                    }
+                    resultRows = resultRows.distinctBy { row -> collationFoldedKey(row, itemExpressions, ::declaredCollationOrNull) }
+                }
+                if (statement.offset != null || statement.limit != null) {
+                    resultRows = resultRows.drop(statement.offset ?: 0)
+                    statement.limit?.let { resultRows = resultRows.take(it) }
+                }
+            }
+            return SelectResult(labels, resultRows)
         }
 
         fun aggregate(combined: List<List<Value>>): SelectResult =
-            GroupAggregation(::resolveIndex, ::evaluateRow, this@SelectRunner::columnLabel)
-                .aggregate(statement, orderExpression, combined)
+            GroupAggregation(::resolveIndex, ::evaluateRow, this@SelectRunner::columnLabel, subqueryRunner, ::declaredCollationOrNull)
+                .aggregate(statement, orderKeys, combined)
 
         fun describe(): List<String> {
             val lines = mutableListOf<String>()
             var body = joinLines()
             if (aggregated) {
                 val keys = statement.groupBy.map { describeExpression(it) }
-                val header = if (keys.isEmpty()) "Aggregate" else "GroupAggregate keys: ${keys.joinToString(", ")}"
+                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" } }
             lines.addAll(body)
             return lines
@@ -1418,9 +1450,14 @@ class SelectRunner(private val catalog: Catalog) {
             var lines = listOf(scanLine(sources.first(), primaryAccess()))
             statement.joins.forEachIndexed { position, join ->
                 val right = sources[position + 1]
-                val hash = hashJoinKeys(join, position + 1) != null
-                val header = if (hash) "Hash Join ${describeExpression(join.on)}" else "Nested Loop ${joinLabel(join)}"
-                lines = listOf(header) + lines.map { "  -> $it" } + listOf("  -> Seq Scan on ${right.tableName}")
+                val lateral = lateralSelectOf(join.from) != null
+                val hash = !lateral && hashJoinKeys(join, position + 1) != null
+                val label = joinLabel(join) + if (lateral) " Lateral" else ""
+                // `hash` is only true when hashJoinKeys() matched a binary equality condition, which
+                // means join.on is guaranteed non-null here (CROSS JOIN's null `on` never hashes).
+                val header = if (hash) "Hash Join ${describeExpression(join.on!!)}" else "Nested Loop $label"
+                val rightScan = if (lateral) "Lateral Subquery Scan on ${right.tableName}" else "Seq Scan on ${right.tableName}"
+                lines = listOf(header) + lines.map { "  -> $it" } + listOf("  -> $rightScan")
             }
             return lines
         }
@@ -1431,14 +1468,21 @@ class SelectRunner(private val catalog: Catalog) {
                 "Index Scan using ${access.indexName} on ${source.tableName} (${access.description})"
         }
 
-        private fun joinLabel(join: JoinClause): String =
-            if (join.type == JoinType.Left) "Left" else "Inner"
+        private fun joinLabel(join: JoinClause): String = when (join.type) {
+            JoinType.Left -> "Left"
+            JoinType.Right -> "Right"
+            JoinType.Full -> "Full"
+            JoinType.Inner -> "Inner"
+            JoinType.Cross -> "Cross"
+        }
     }
 
     private fun columnLabel(item: SelectItem, index: Int): String =
         item.alias ?: when (val expression = item.expression) {
             is ColumnExpression -> expression.name
             is FunctionExpression -> expression.name.lowercase()
+            is CallExpression -> expression.name.lowercase()
+            is WindowExpression -> expression.name.lowercase()
             else -> "column${index + 1}"
         }
 
@@ -1446,8 +1490,34 @@ class SelectRunner(private val catalog: Catalog) {
         is ColumnExpression -> listOf(expression)
         is BinaryExpression -> collectColumns(expression.left) + collectColumns(expression.right)
         is UnaryExpression -> collectColumns(expression.operand)
-        is FunctionExpression -> expression.argument?.let { collectColumns(it) } ?: emptyList()
+        is FunctionExpression ->
+            (expression.argument?.let { collectColumns(it) } ?: emptyList()) +
+                (expression.separator?.let { collectColumns(it) } ?: emptyList()) +
+                expression.orderBy.flatMap { collectColumns(it.expression) } +
+                (expression.filter?.let { collectColumns(it) } ?: emptyList())
         is LiteralExpression -> emptyList()
+        is CallExpression -> expression.arguments.flatMap { collectColumns(it) }
+        is IsNullExpression -> collectColumns(expression.operand)
+        is InExpression -> collectColumns(expression.operand) + expression.values.flatMap { collectColumns(it) }
+        is BetweenExpression -> collectColumns(expression.operand) + collectColumns(expression.low) + collectColumns(expression.high)
+        is LikeExpression -> collectColumns(expression.operand) + collectColumns(expression.pattern)
+        is CaseExpression -> (expression.operand?.let { collectColumns(it) } ?: emptyList()) +
+            expression.branches.flatMap { collectColumns(it.condition) + collectColumns(it.result) } +
+            (expression.elseExpression?.let { collectColumns(it) } ?: emptyList())
+        is ScalarSubqueryExpression -> emptyList()
+        is ExistsExpression -> emptyList()
+        is InSubqueryExpression -> collectColumns(expression.operand)
+        is CastExpression -> collectColumns(expression.operand)
+        is CollateExpression -> collectColumns(expression.operand)
+        is WindowExpression ->
+            (expression.argument?.let { collectColumns(it) } ?: emptyList()) +
+                (expression.offset?.let { collectColumns(it) } ?: emptyList()) +
+                (expression.defaultValue?.let { collectColumns(it) } ?: emptyList()) +
+                (expression.frame?.start?.offset?.let { collectColumns(it) } ?: emptyList()) +
+                (expression.frame?.end?.offset?.let { collectColumns(it) } ?: emptyList()) +
+                expression.partitionBy.flatMap { collectColumns(it) } +
+                expression.orderBy.flatMap { collectColumns(it.expression) }
+        is NextValueExpression, is CurrentValueExpression, is MergeActionExpression, is ParameterExpression -> emptyList()
     }
 
     private fun containsAggregate(projection: kestrel.sql.ast.Projection): Boolean = when (projection) {
@@ -1459,9 +1529,66 @@ class SelectRunner(private val catalog: Catalog) {
         is FunctionExpression -> true
         is BinaryExpression -> containsAggregate(expression.left) || containsAggregate(expression.right)
         is UnaryExpression -> containsAggregate(expression.operand)
+        is CallExpression -> expression.arguments.any { containsAggregate(it) }
+        is IsNullExpression -> containsAggregate(expression.operand)
+        is InExpression -> containsAggregate(expression.operand) || expression.values.any { containsAggregate(it) }
+        is BetweenExpression ->
+            containsAggregate(expression.operand) || containsAggregate(expression.low) || containsAggregate(expression.high)
+        is LikeExpression -> containsAggregate(expression.operand) || containsAggregate(expression.pattern)
+        is CaseExpression ->
+            (expression.operand?.let { containsAggregate(it) } ?: false) ||
+                expression.branches.any { containsAggregate(it.condition) || containsAggregate(it.result) } ||
+                (expression.elseExpression?.let { containsAggregate(it) } ?: false)
+        is CastExpression -> containsAggregate(expression.operand)
+        is CollateExpression -> containsAggregate(expression.operand)
+        // A window function never triggers GROUP BY collapsing on its own - it is deliberately not
+        // recursed into here, so an aggregate nested inside its OVER clause (invalid SQL) is instead
+        // caught uniformly at evaluation time by the same guard that rejects any bare aggregate
+        // outside GROUP BY (see Evaluator.evaluate's FunctionExpression branch).
+        is WindowExpression -> false
         else -> false
     }
 
+    /** Window functions are only legal in the SELECT list and ORDER BY - never WHERE/GROUP BY/HAVING/ON. */
+    private fun containsWindowFunction(expression: Expression): Boolean = when (expression) {
+        is WindowExpression -> true
+        is BinaryExpression -> containsWindowFunction(expression.left) || containsWindowFunction(expression.right)
+        is UnaryExpression -> containsWindowFunction(expression.operand)
+        is CallExpression -> expression.arguments.any { containsWindowFunction(it) }
+        is IsNullExpression -> containsWindowFunction(expression.operand)
+        is InExpression -> containsWindowFunction(expression.operand) || expression.values.any { containsWindowFunction(it) }
+        is BetweenExpression ->
+            containsWindowFunction(expression.operand) || containsWindowFunction(expression.low) || containsWindowFunction(expression.high)
+        is LikeExpression -> containsWindowFunction(expression.operand) || containsWindowFunction(expression.pattern)
+        is CaseExpression ->
+            (expression.operand?.let { containsWindowFunction(it) } ?: false) ||
+                expression.branches.any { containsWindowFunction(it.condition) || containsWindowFunction(it.result) } ||
+                (expression.elseExpression?.let { containsWindowFunction(it) } ?: false)
+        is CastExpression -> containsWindowFunction(expression.operand)
+        is CollateExpression -> containsWindowFunction(expression.operand)
+        else -> false
+    }
+
+    /** Every distinct window-function call reachable from [expression] (nested inside arithmetic, CASE, ...). */
+    private fun collectWindowExpressions(expression: Expression): List<WindowExpression> = when (expression) {
+        is WindowExpression -> listOf(expression)
+        is BinaryExpression -> collectWindowExpressions(expression.left) + collectWindowExpressions(expression.right)
+        is UnaryExpression -> collectWindowExpressions(expression.operand)
+        is CallExpression -> expression.arguments.flatMap { collectWindowExpressions(it) }
+        is IsNullExpression -> collectWindowExpressions(expression.operand)
+        is InExpression -> collectWindowExpressions(expression.operand) + expression.values.flatMap { collectWindowExpressions(it) }
+        is BetweenExpression ->
+            collectWindowExpressions(expression.operand) + collectWindowExpressions(expression.low) + collectWindowExpressions(expression.high)
+        is LikeExpression -> collectWindowExpressions(expression.operand) + collectWindowExpressions(expression.pattern)
+        is CaseExpression ->
+            (expression.operand?.let { collectWindowExpressions(it) } ?: emptyList()) +
+                expression.branches.flatMap { collectWindowExpressions(it.condition) + collectWindowExpressions(it.result) } +
+                (expression.elseExpression?.let { collectWindowExpressions(it) } ?: emptyList())
+        is CastExpression -> collectWindowExpressions(expression.operand)
+        is CollateExpression -> collectWindowExpressions(expression.operand)
+        else -> emptyList()
+    }
+
     private fun isTrue(value: Value): Boolean = value is Value.BooleanValue && value.value
 
     private fun describe(expression: ColumnExpression): String =
@@ -1470,9 +1597,37 @@ class SelectRunner(private val catalog: Catalog) {
     private fun describeExpression(expression: Expression): String = when (expression) {
         is ColumnExpression -> describe(expression)
         is LiteralExpression -> literalText(expression.value)
-        is FunctionExpression -> "${expression.name}(${if (expression.star) "*" else expression.argument?.let { describeExpression(it) } ?: ""})"
+        is FunctionExpression -> "${expression.name}(${describeFunctionArguments(expression)})${describeFilterClause(expression)}"
         is BinaryExpression -> "${describeExpression(expression.left)} ${expression.operator.symbol} ${describeExpression(expression.right)}"
         is UnaryExpression -> "${expression.operator.symbol} ${describeExpression(expression.operand)}"
+        is CallExpression -> "${expression.name}(${expression.arguments.joinToString(", ") { describeExpression(it) }})"
+        is IsNullExpression -> "${describeExpression(expression.operand)} IS ${if (expression.negated) "NOT " else ""}NULL"
+        is InExpression ->
+            "${describeExpression(expression.operand)} ${if (expression.negated) "NOT " else ""}IN (${expression.values.joinToString(", ") { describeExpression(it) }})"
+        is BetweenExpression ->
+            "${describeExpression(expression.operand)} ${if (expression.negated) "NOT " else ""}BETWEEN ${describeExpression(expression.low)} AND ${describeExpression(expression.high)}"
+        is LikeExpression ->
+            "${describeExpression(expression.operand)} ${if (expression.negated) "NOT " else ""}LIKE ${describeExpression(expression.pattern)}"
+        is CaseExpression -> buildString {
+            append("CASE ")
+            expression.operand?.let { append(describeExpression(it)); append(' ') }
+            expression.branches.forEach { append("WHEN ${describeExpression(it.condition)} THEN ${describeExpression(it.result)} ") }
+            expression.elseExpression?.let { append("ELSE ${describeExpression(it)} ") }
+            append("END")
+        }
+        is ScalarSubqueryExpression -> "(subquery)"
+        is ExistsExpression -> "EXISTS (subquery)"
+        is InSubqueryExpression ->
+            "${describeExpression(expression.operand)} ${if (expression.negated) "NOT " else ""}IN (subquery)"
+        is CastExpression ->
+            "CAST(${describeExpression(expression.operand)} AS ${expression.targetType.displayWith(expression.precision, expression.scale)})"
+        is CollateExpression -> "${describeExpression(expression.operand)} COLLATE ${expression.collation.sqlName}"
+        is WindowExpression ->
+            "${expression.name}(${if (expression.star) "*" else expression.argument?.let { describeExpression(it) } ?: ""}) OVER (...)"
+        is NextValueExpression -> "NEXTVAL('${expression.sequenceName}')"
+        is CurrentValueExpression -> "CURRVAL('${expression.sequenceName}')"
+        is MergeActionExpression -> "MERGE_ACTION()"
+        is ParameterExpression -> "\$${expression.index}"
     }
 
     private fun literalText(value: Value): String = when (value) {