feat(sql): add RangeBounds and rangeBounds to DiskIndexAccessPlanner

src/main/kotlin/kestrel/store/sql/DiskIndexAccessPlanner.kt

@@ -165,3 +165,74 @@ class DiskIndexAccessPlanner {
         return DiskAccessPlan.FullScan
     }
 
+    private class RangeBounds(val lower: Value?, val lowerInclusive: Boolean, val upper: Value?, val upperInclusive: Boolean)
+
+    private fun rangeBounds(predicates: List<Pair<BinaryOperator, Value>>): RangeBounds {
+        var lowerValue: Value? = null
+        var lowerInclusive = false
+        var upperValue: Value? = null
+        var upperInclusive = false
+        for ((operator, literal) in predicates) {
+            when (operator) {
+                BinaryOperator.Greater -> { lowerValue = literal; lowerInclusive = false }
+                BinaryOperator.GreaterEqual -> { lowerValue = literal; lowerInclusive = true }
+                BinaryOperator.Less -> { upperValue = literal; upperInclusive = false }
+                BinaryOperator.LessEqual -> { upperValue = literal; upperInclusive = true }
+                else -> Unit
+            }
+        }
+        return RangeBounds(lowerValue, lowerInclusive, upperValue, upperInclusive)
+    }
+
+    private fun rangePlan(predicates: List<Pair<BinaryOperator, Value>>): DiskAccessPlan {
+        val bounds = rangeBounds(predicates)
+        return DiskAccessPlan.PrimaryKeyRange(bounds.lower, bounds.lowerInclusive, bounds.upper, bounds.upperInclusive)
+    }
+
+    private fun asColumnPredicate(
+        expression: Expression,
+        alias: String?,
+        schema: TableSchema
+    ): Triple<String, BinaryOperator, Value>? {
+        if (expression !is BinaryExpression) return null
+        val operator = expression.operator
+        if (operator !in RANGE_OPERATORS) return null
+        val left = expression.left
+        val right = expression.right
+        fun matches(column: ColumnExpression): String? {
+            if (column.qualifier != null && alias != null && !column.qualifier.equals(alias, ignoreCase = true)) return null
+            return if (schema.columnIndex(column.name) != null) column.name else null
+        }
+        if (left is ColumnExpression && right is LiteralExpression && right.value !is Value.NullValue) {
+            val name = matches(left) ?: return null
+            return Triple(name, operator, right.value)
+        }
+        if (right is ColumnExpression && left is LiteralExpression && left.value !is Value.NullValue) {
+            val name = matches(right) ?: return null
+            return Triple(name, flip(operator), left.value)
+        }
+        return null
+    }
+
+    private fun conjuncts(expression: Expression): List<Expression> =
+        if (expression is BinaryExpression && expression.operator == BinaryOperator.And) {
+            conjuncts(expression.left) + conjuncts(expression.right)
+        } else {
+            listOf(expression)
+        }
+
+    private fun flip(operator: BinaryOperator): BinaryOperator = when (operator) {
+        BinaryOperator.Less -> BinaryOperator.Greater
+        BinaryOperator.LessEqual -> BinaryOperator.GreaterEqual
+        BinaryOperator.Greater -> BinaryOperator.Less
+        BinaryOperator.GreaterEqual -> BinaryOperator.LessEqual
+        else -> operator
+    }
+
+    private companion object {
+        val RANGE_OPERATORS = setOf(
+            BinaryOperator.Equal, BinaryOperator.Less, BinaryOperator.LessEqual,
+            BinaryOperator.Greater, BinaryOperator.GreaterEqual
+        )
+    }
+}