diff --git a/src/main/kotlin/kestrel/engine/SelectRunner.kt b/src/main/kotlin/kestrel/engine/SelectRunner.kt index 671d562..fed7c44 100644 --- a/src/main/kotlin/kestrel/engine/SelectRunner.kt +++ b/src/main/kotlin/kestrel/engine/SelectRunner.kt @@ -790,7 +790,179 @@ class SelectRunner(private val catalog: Catalog) { statement.where?.let { predicate -> combined = combined.filter { isTrue(evaluateRow(predicate, it)) } } - return combined + return appendWindowColumns(combined) + } + + /** + * Runs after FROM/JOIN/WHERE and (per [validateWindows]) never alongside GROUP BY, matching + * standard SQL's logical order of window-function evaluation. Each entry of + * [windowExpressions] contributes one trailing value per row, computed once over the whole + * (already filtered) row set - [evaluateRow]/[substituteWindowValues] read them back out by + * position for the final projection and this query's own ORDER BY. + */ + private fun appendWindowColumns(combined: List>): List> { + if (windowExpressions.isEmpty()) return combined + val perExpression = windowExpressions.map { computeWindowValues(it, combined) } + return combined.mapIndexed { rowIndex, row -> row + windowExpressions.indices.map { perExpression[it][rowIndex] } } + } + + private fun computeWindowValues(window: WindowExpression, combined: List>): List { + val values = arrayOfNulls(combined.size) + val partitions = combined.indices.groupBy { index -> window.partitionBy.map { evaluateRow(it, combined[index]) } } + for (memberIndices in partitions.values) { + val ordered = if (window.orderBy.isEmpty()) { + memberIndices + } else { + memberIndices.sortedWith(Comparator { first, second -> + var result = 0 + for (key in window.orderBy) { + val comparison = compareOrderKeyValues( + evaluateRow(key.expression, combined[first]), + evaluateRow(key.expression, combined[second]), + key, + ::declaredCollationOrNull + ) + if (comparison != 0) { + result = comparison + break + } + } + result + }) + } + assignWindowValues(window, ordered, combined, values) + } + return values.map { it ?: error("window value not computed for every row") } + } + + private fun assignWindowValues(window: WindowExpression, ordered: List, combined: List>, out: Array) { + when (window.kind) { + WindowFunctionKind.RowNumber -> + ordered.forEachIndexed { position, index -> out[index] = Value.IntegerValue((position + 1).toLong()) } + WindowFunctionKind.Rank, WindowFunctionKind.DenseRank -> { + var rank = 0 + var denseRank = 0 + var previousKey: List? = null + ordered.forEachIndexed { position, index -> + val key = window.orderBy.map { evaluateRow(it.expression, combined[index]) } + if (previousKey == null || key != previousKey) { + rank = position + 1 + denseRank++ + previousKey = key + } + out[index] = Value.IntegerValue((if (window.kind == WindowFunctionKind.Rank) rank else denseRank).toLong()) + } + } + WindowFunctionKind.Aggregate -> assignAggregateWindowValues(window, ordered, combined, out) + WindowFunctionKind.Lag -> assignOffsetWindowValues(window, ordered, combined, out, direction = -1) + WindowFunctionKind.Lead -> assignOffsetWindowValues(window, ordered, combined, out, direction = 1) + WindowFunctionKind.FirstValue -> { + val frame = window.frame + if (frame == null) { + val firstValue = evaluateRow(window.argument!!, combined[ordered.first()]) + ordered.forEach { index -> out[index] = firstValue } + } else { + ordered.indices.forEach { position -> + val members = frameMembers(window, position, ordered, combined) + out[ordered[position]] = + if (members.isEmpty()) Value.NullValue else evaluateRow(window.argument!!, combined[ordered[members.first()]]) + } + } + } + WindowFunctionKind.LastValue -> { + val frame = window.frame + if (frame == null) { + for (range in peerGroupRanges(window.orderBy, ordered, combined)) { + val value = evaluateRow(window.argument!!, combined[ordered[range.last]]) + for (position in range) out[ordered[position]] = value + } + } else { + ordered.indices.forEach { position -> + val members = frameMembers(window, position, ordered, combined) + out[ordered[position]] = + if (members.isEmpty()) Value.NullValue else evaluateRow(window.argument!!, combined[ordered[members.last()]]) + } + } + } + WindowFunctionKind.Ntile -> assignNtileWindowValues(window, ordered, combined, out) + } + } + + /** + * With no ORDER BY in OVER, the whole partition is one frame (every row gets the same + * value). With an ORDER BY, the default SQL frame is `RANGE BETWEEN UNBOUNDED PRECEDING AND + * CURRENT ROW`: a running aggregate where rows tied on the order key (a "peer group") share + * one value - the aggregate through the end of their peer group, not through their own row + * alone. A single-column running total (`ORDER BY` with no ties) falls out of this as the + * special case where every peer group has size 1. + */ + private fun assignAggregateWindowValues(window: WindowExpression, ordered: List, combined: List>, out: Array) { + val frame = window.frame + if (frame == null) { + for (range in peerGroupRanges(window.orderBy, ordered, combined)) { + val value = aggregateWindowValue(window, (0..range.last).map { combined[ordered[it]] }) + for (position in range) out[ordered[position]] = value + } + } else { + ordered.indices.forEach { position -> + val members = frameMembers(window, position, ordered, combined) + out[ordered[position]] = aggregateWindowValue(window, members.map { combined[ordered[it]] }) + } + } + } + + /** + * The frame's base [frameRange], narrowed by its [FrameExclude] mode (default [FrameExclude.NoOthers], + * which leaves the range untouched). Excluding rows can split a contiguous range into two + * pieces (e.g. `EXCLUDE CURRENT ROW` on a frame that spans the current row), so this returns + * the actual member positions rather than an [IntRange]. + */ + private fun frameMembers(window: WindowExpression, position: Int, ordered: List, combined: List>): List { + val range = frameRange(window, position, ordered, combined) + if (range.isEmpty()) return emptyList() + return when (window.frame!!.exclude) { + FrameExclude.NoOthers -> range.toList() + FrameExclude.CurrentRow -> range.filter { it != position } + FrameExclude.Group -> { + val peer = peerBoundary(window.orderBy, position, ordered, combined, isStart = true).. + peerBoundary(window.orderBy, position, ordered, combined, isStart = false) + range.filter { it !in peer } + } + FrameExclude.Ties -> { + val peer = peerBoundary(window.orderBy, position, ordered, combined, isStart = true).. + peerBoundary(window.orderBy, position, ordered, combined, isStart = false) + range.filter { it == position || it !in peer } + } + } + } + + /** + * Resolves an explicit `ROWS|RANGE BETWEEN ... AND ...` [WindowFrame] to a range of positions + * into [ordered] for the row at [position] - clamped to the partition's edges rather than + * erroring, exactly like the implicit default frame already does. A frame that falls entirely + * off one edge (e.g. `2 FOLLOWING AND 3 FOLLOWING` on the partition's last row) yields an empty + * range, which [aggregateWindowValue]/FIRST_VALUE/LAST_VALUE each already know how to treat as + * "no rows" (NULL, or 0 for COUNT). + */ + private fun frameRange(window: WindowExpression, position: Int, ordered: List, combined: List>): IntRange { + val frame = window.frame!! + if (frame.unit == FrameUnit.Groups) { + return groupsFrameRange(frame, window.orderBy, position, ordered, combined) + } + val size = ordered.size + val start = resolveFrameBound(frame, frame.start, window, position, ordered, combined, isStart = true) + val end = resolveFrameBound(frame, frame.end, window, position, ordered, combined, isStart = false) + val clampedStart = start.coerceAtLeast(0) + val clampedEnd = end.coerceAtMost(size - 1) + return if (clampedStart > clampedEnd) IntRange.EMPTY else clampedStart..clampedEnd + } + + private fun resolveFrameBound( + frame: WindowFrame, + bound: FrameBound, + window: WindowExpression, + position: Int, + ordered: List, } private fun primaryRows(): List {