feat(engine): extend SelectRunner with appendWindowColumns
This commit is contained in:
dualfroz committed
1 parent
e3811b3be2
commit
16f7b062e1
1 file changed
+173
-1
@@ -790,7 +790,179 @@ class SelectRunner(private val catalog: Catalog) {
|
|||||||
statement.where?.let { predicate ->
|
statement.where?.let { predicate ->
|
||||||
combined = combined.filter { isTrue(evaluateRow(predicate, it)) }
|
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<Value>>): List<List<Value>> {
|
||||||
|
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<Value>>): List<Value> {
|
||||||
|
val values = arrayOfNulls<Value>(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<Int>, combined: List<List<Value>>, out: Array<Value?>) {
|
||||||
|
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<Value>? = 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<Int>, combined: List<List<Value>>, out: Array<Value?>) {
|
||||||
|
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<Int>, combined: List<List<Value>>): List<Int> {
|
||||||
|
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<Int>, combined: List<List<Value>>): 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<Int>,
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun primaryRows(): List<Row> {
|
private fun primaryRows(): List<Row> {
|
||||||
|
|||||||
Reference in new issue
Block a user