feat(engine): extend SelectRunner with resolveIntegerFrameOffset

src/main/kotlin/kestrel/engine/SelectRunner.kt

@@ -963,6 +963,179 @@ class SelectRunner(private val catalog: Catalog) {
             window: WindowExpression,
             position: Int,
             ordered: List<Int>,
+            combined: List<List<Value>>,
+            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<Value>, 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 <bound> AND <bound>` 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<OrderKey>,
+            position: Int,
+            ordered: List<Int>,
+            combined: List<List<Value>>
+        ): 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<Value>): 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<OrderKey>, position: Int, ordered: List<Int>, combined: List<List<Value>>, 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
+        }
+
+        /**
+         * `<value> PRECEDING`/`<value> 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<OrderKey>,
+            position: Int,
+            ordered: List<Int>,
+            combined: List<List<Value>>,
+            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<Int>, combined: List<List<Value>>, 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<Int>, combined: List<List<Value>>, 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<Value>): 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<Value>): 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<OrderKey>, ordered: List<Int>, combined: List<List<Value>>): List<IntRange> {
+            if (orderBy.isEmpty()) return listOf(ordered.indices)
+            val keys = ordered.map { index -> orderBy.map { key -> evaluateRow(key.expression, combined[index]) } }
+            val ranges = mutableListOf<IntRange>()
+            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<Row> {