From 98138f80e91abed94f64d8a0969531e8605e9e9a Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 2 Sep 2026 16:13:42 +0200 Subject: [PATCH] feat(engine): extend SelectRunner with resolveIntegerFrameOffset --- .../kotlin/kestrel/engine/SelectRunner.kt | 173 ++++++++++++++++++ 1 file changed, 173 insertions(+) diff --git a/src/main/kotlin/kestrel/engine/SelectRunner.kt b/src/main/kotlin/kestrel/engine/SelectRunner.kt index fed7c44..defb5bf 100644 --- a/src/main/kotlin/kestrel/engine/SelectRunner.kt +++ b/src/main/kotlin/kestrel/engine/SelectRunner.kt @@ -963,6 +963,179 @@ class SelectRunner(private val catalog: Catalog) { window: WindowExpression, position: Int, ordered: List, + combined: List>, + isStart: Boolean + ): Int = when (bound.type) { + FrameBoundType.UnboundedPreceding -> 0 + FrameBoundType.UnboundedFollowing -> ordered.size - 1 + FrameBoundType.CurrentRow -> + if (frame.unit == FrameUnit.Rows) position else peerBoundary(window.orderBy, position, ordered, combined, isStart) + FrameBoundType.Preceding -> + if (frame.unit == FrameUnit.Rows) position - resolveIntegerFrameOffset(bound, combined[ordered[position]], "ROWS") + else rangeBoundary(bound, window.orderBy, position, ordered, combined, isStart, subtract = true) + FrameBoundType.Following -> + if (frame.unit == FrameUnit.Rows) position + resolveIntegerFrameOffset(bound, combined[ordered[position]], "ROWS") + else rangeBoundary(bound, window.orderBy, position, ordered, combined, isStart, subtract = false) + } + + private fun resolveIntegerFrameOffset(bound: FrameBound, row: List, unit: String): Int { + val value = evaluateRow(bound.offset!!, row) + val amount = (value as? Value.IntegerValue)?.value + ?: throw KestrelException("$unit frame offset must be an integer") + if (amount < 0) throw KestrelException("$unit frame offset must not be negative") + return amount.toInt() + } + + /** + * `GROUPS BETWEEN AND ` counts whole peer groups (rows tied on the OVER's + * ORDER BY) rather than physical rows ([FrameUnit.Rows]) or distance in a value + * ([FrameUnit.Range]) - `1 PRECEDING` means "the peer group immediately before the current + * row's own peer group", however many rows that group has. Resolved entirely in group-index + * space (mirroring exactly how [frameRange] resolves ROWS/RANGE in row-index space, including + * the same one-sided-clamp trick for correctly detecting an out-of-partition frame as empty + * rather than accidentally clamping it back into range), then translated to a row [IntRange] + * only once both bounds land on a valid group. Unlike [rangeBoundary], this places no + * restriction on the OVER's ORDER BY - zero, one, or many columns of any type all define peer + * groups equally well, since [peerGroupRanges] only ever compares them for equality. + */ + private fun groupsFrameRange( + frame: WindowFrame, + orderBy: List, + position: Int, + ordered: List, + combined: List> + ): IntRange { + val groups = peerGroupRanges(orderBy, ordered, combined) + val currentGroup = groups.indexOfFirst { position in it } + val row = combined[ordered[position]] + val startGroup = resolveGroupBound(frame.start, currentGroup, groups.size, row).coerceAtLeast(0) + val endGroup = resolveGroupBound(frame.end, currentGroup, groups.size, row).coerceAtMost(groups.size - 1) + return if (startGroup > endGroup) IntRange.EMPTY else groups[startGroup].first..groups[endGroup].last + } + + private fun resolveGroupBound(bound: FrameBound, currentGroup: Int, groupCount: Int, row: List): Int = when (bound.type) { + FrameBoundType.UnboundedPreceding -> 0 + FrameBoundType.UnboundedFollowing -> groupCount - 1 + FrameBoundType.CurrentRow -> currentGroup + FrameBoundType.Preceding -> currentGroup - resolveIntegerFrameOffset(bound, row, "GROUPS") + FrameBoundType.Following -> currentGroup + resolveIntegerFrameOffset(bound, row, "GROUPS") + } + + /** + * `CURRENT ROW` in a RANGE frame does not mean "just this row" (that is what ROWS is for) - it + * means the boundary of the current row's peer group (every row tied with it on the OVER's + * ORDER BY, or the whole partition when there is no ORDER BY). [isStart] picks which edge of + * that peer group to return. + */ + private fun peerBoundary(orderBy: List, position: Int, ordered: List, combined: List>, isStart: Boolean): Int { + if (orderBy.isEmpty()) return if (isStart) 0 else ordered.size - 1 + fun keyAt(index: Int) = orderBy.map { evaluateRow(it.expression, combined[ordered[index]]) } + val key = keyAt(position) + var boundary = position + if (isStart) { + while (boundary > 0 && keyAt(boundary - 1) == key) boundary-- + } else { + while (boundary < ordered.size - 1 && keyAt(boundary + 1) == key) boundary++ + } + return boundary + } + + /** + * ` PRECEDING`/` FOLLOWING` in a RANGE frame measure distance in the value of a + * single numeric or DATE ORDER BY column rather than a row count, so the frame can span a + * different number of physical rows for every current row (e.g. `RANGE BETWEEN 5 PRECEDING AND + * 5 FOLLOWING` on `ORDER BY amount` widens or narrows with how densely `amount` is populated + * nearby). Binary search works because [ordered] is already sorted on this column - ascending + * in "logical" terms via [rangeLogicalValue], which folds DESC into a sign flip so the same + * `>=`/`<=` comparisons apply regardless of sort direction. + */ + private fun rangeBoundary( + bound: FrameBound, + orderBy: List, + position: Int, + ordered: List, + combined: List>, + isStart: Boolean, + subtract: Boolean + ): Int { + if (orderBy.size != 1) { + throw KestrelException("a RANGE frame with a PRECEDING/FOLLOWING bound requires exactly one ORDER BY column") + } + val key = orderBy.single() + val current = rangeLogicalValue(key, combined[ordered[position]]) + val offset = resolveRangeFrameOffset(bound, combined[ordered[position]]) + val threshold = if (subtract) current - offset else current + offset + return if (isStart) { + rangeLowerBound(ordered, combined, key, threshold) + } else { + rangeUpperBound(ordered, combined, key, threshold) + } + } + + private fun rangeLowerBound(ordered: List, combined: List>, key: OrderKey, threshold: BigDecimal): Int { + var low = 0 + var high = ordered.size + while (low < high) { + val mid = (low + high) / 2 + if (rangeLogicalValue(key, combined[ordered[mid]]) >= threshold) high = mid else low = mid + 1 + } + return low + } + + private fun rangeUpperBound(ordered: List, combined: List>, key: OrderKey, threshold: BigDecimal): Int { + var low = 0 + var high = ordered.size + while (low < high) { + val mid = (low + high) / 2 + if (rangeLogicalValue(key, combined[ordered[mid]]) <= threshold) low = mid + 1 else high = mid + } + return low - 1 + } + + private fun rangeLogicalValue(key: OrderKey, row: List): BigDecimal { + val magnitude = rangeMagnitude(evaluateRow(key.expression, row)) + return if (key.descending) magnitude.negate() else magnitude + } + + private fun rangeMagnitude(value: Value): BigDecimal = when (value) { + is Value.IntegerValue -> BigDecimal(value.value) + is Value.DecimalValue -> value.value + is Value.RealValue -> BigDecimal.valueOf(value.value) + is Value.DateValue -> BigDecimal(value.value.toEpochDay()) + else -> throw KestrelException("a RANGE frame requires the ORDER BY column to be numeric or DATE") + } + + private fun resolveRangeFrameOffset(bound: FrameBound, row: List): BigDecimal { + val offset = when (val value = evaluateRow(bound.offset!!, row)) { + is Value.IntegerValue -> BigDecimal(value.value) + is Value.DecimalValue -> value.value + is Value.RealValue -> BigDecimal.valueOf(value.value) + else -> throw KestrelException("RANGE frame offset must be numeric") + } + if (offset.signum() < 0) throw KestrelException("RANGE frame offset must not be negative") + return offset + } + + /** + * Splits [ordered] (positions into [combined], already sorted by the OVER's ORDER BY) into + * contiguous "peer group" ranges of rows tied on the order key - or one range spanning the + * whole partition when there is no ORDER BY, matching the default SQL frame. Shared by + * [assignAggregateWindowValues] (running aggregate through the end of a peer group) and + * [WindowFunctionKind.LastValue] (last row of a peer group). + */ + private fun peerGroupRanges(orderBy: List, ordered: List, combined: List>): List { + if (orderBy.isEmpty()) return listOf(ordered.indices) + val keys = ordered.map { index -> orderBy.map { key -> evaluateRow(key.expression, combined[index]) } } + val ranges = mutableListOf() + var i = 0 + while (i < ordered.size) { + var j = i + while (j < ordered.size && keys[j] == keys[i]) j++ + ranges.add(i until j) + i = j + } + return ranges + } } private fun primaryRows(): List {