test(sql): cover aRangeOnTheSecondColumnOfACompositeSecondaryIndexAft...

src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt

@@ -1270,3 +1270,165 @@ class DiskExecutorTest {
         assertEquals(scansBeforeSelect, table.scanCount)
     }
 
+    @Test
+    fun aRangeOnTheSecondColumnOfACompositeSecondaryIndexAfterAnEqualityPrefixUsesARangeScanNotAFullScan() {
+        val db = newDatabase()
+        db.run("CREATE TABLE order_lines (order_id INTEGER, line_no INTEGER, sku TEXT)")
+        db.run("INSERT INTO order_lines VALUES (1, 1, 'apple'), (1, 2, 'mango'), (2, 1, 'kiwi')")
+        db.run("CREATE INDEX idx_order_sku ON order_lines(order_id, sku)")
+        val table = db.catalog.table("order_lines")
+        val scansBeforeSelect = table.scanCount // backfilling the new index already scanned once
+        val result = db.select("SELECT line_no FROM order_lines WHERE order_id = 1 AND sku > 'apple'")
+        assertEquals(listOf(listOf(int(2))), result.rows)
+        assertEquals(1, table.rangeScanCount)
+        assertEquals(0, table.secondaryLookupCount)
+        assertEquals(scansBeforeSelect, table.scanCount)
+    }
+
+    @Test
+    fun aRangeOnTheSecondColumnOfACompositeSecondaryIndexWithoutAnEqualityPrefixFallsBackToAFullScan() {
+        // Mirrors equalityOnOnlyOneColumnOfACompositeSecondaryIndexFallsBackToAFullScan - the index
+        // is sorted by order_id first, so a bare range on sku cannot narrow the scan without an
+        // equality prefix pinning order_id down first.
+        val db = newDatabase()
+        db.run("CREATE TABLE order_lines (order_id INTEGER, line_no INTEGER, sku TEXT)")
+        db.run("INSERT INTO order_lines VALUES (1, 1, 'apple'), (1, 2, 'mango'), (2, 1, 'kiwi')")
+        db.run("CREATE INDEX idx_order_sku ON order_lines(order_id, sku)")
+        val table = db.catalog.table("order_lines")
+        val scansBeforeSelect = table.scanCount
+        val result = db.select("SELECT line_no FROM order_lines WHERE sku > 'apple'")
+        assertEquals(2, result.rows.size)
+        assertEquals(0, table.rangeScanCount)
+        assertEquals(scansBeforeSelect + 1, table.scanCount)
+    }
+
+    @Test
+    fun equalityOnASecondaryIndexIsPreferredOverARangeOnTheSameColumnAtRuntime() {
+        val db = newDatabase()
+        db.seedUsers()
+        db.run("CREATE INDEX idx_age ON users(age)")
+        val table = db.catalog.table("users")
+        val result = db.select("SELECT name FROM users WHERE age = 54 AND age > 0")
+        assertEquals(listOf(listOf(text("linus"))), result.rows)
+        assertEquals(1, table.secondaryLookupCount)
+        assertEquals(0, table.rangeScanCount)
+    }
+
+    @Test
+    fun aPrimaryKeyEqualityIsPreferredOverASecondaryIndexInTheSamePredicate() {
+        val db = newDatabase()
+        db.seedUsers()
+        db.run("CREATE INDEX idx_age ON users(age)")
+        val table = db.catalog.table("users")
+        // age = 999 matches nobody, but a point lookup on id must still run (and the extra
+        // conjunct is applied afterwards by the ordinary full-predicate filter), proving the
+        // planner picked the primary key rather than the secondary index for this predicate.
+        val result = db.select("SELECT name FROM users WHERE id = 2 AND age = 999")
+        assertEquals(emptyList<List<Value>>(), result.rows)
+        assertEquals(1, table.pointLookupCount)
+        assertEquals(0, table.secondaryLookupCount)
+    }
+
+    @Test
+    fun updateWithAPrimaryKeyEqualityWhereUsesAPointLookupNotAFullScan() {
+        val db = newDatabase()
+        db.seedUsers()
+        val table = db.catalog.table("users")
+        val updated = db.run("UPDATE users SET age = 99 WHERE id = 1").single()
+        assertEquals(UpdateResult(1), updated)
+        assertEquals(1, table.pointLookupCount)
+        assertEquals(0, table.scanCount)
+        assertEquals(int(99), db.select("SELECT age FROM users WHERE id = 1").rows.single().single())
+    }
+
+    @Test
+    fun deleteWithAPrimaryKeyEqualityWhereUsesAPointLookupNotAFullScan() {
+        val db = newDatabase()
+        db.seedUsers()
+        val table = db.catalog.table("users")
+        val deleted = db.run("DELETE FROM users WHERE id = 3").single()
+        assertEquals(DeleteResult(1), deleted)
+        assertEquals(1, table.pointLookupCount)
+        assertEquals(0, table.scanCount)
+        assertEquals(2, db.select("SELECT * FROM users").rows.size)
+    }
+
+    // --- table-level composite PRIMARY KEY (a, b): PRIMARY KEY (a, b) instead of an inline
+    // `col TYPE PRIMARY KEY`, mirroring kestrel.engine.Executor's own handling of the same clause.
+
+    private fun Database.seedOrderLines() {
+        run("CREATE TABLE order_lines (order_id INTEGER, line_no INTEGER, sku TEXT, PRIMARY KEY (order_id, line_no))")
+        run("INSERT INTO order_lines VALUES (1, 1, 'widget'), (1, 2, 'gadget'), (2, 1, 'widget')")
+    }
+
+    @Test
+    fun compositePrimaryKeyColumnsAreImplicitlyNotNullLikeAnInlinePrimaryKey() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        assertFailsWith<DiskTableException> { db.run("INSERT INTO order_lines (line_no, sku) VALUES (1, 'x')") }
+    }
+
+    @Test
+    fun compositePrimaryKeyRejectsADuplicateColumnCombinationButAllowsSharedPartialValues() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        assertFailsWith<DiskTableException> { db.run("INSERT INTO order_lines VALUES (1, 1, 'duplicate')") }
+        assertEquals(3, db.select("SELECT * FROM order_lines").rows.size)
+    }
+
+    @Test
+    fun compositePrimaryKeyEqualityOnAllColumnsUsesAPointLookupNotAFullScan() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        val table = db.catalog.table("order_lines")
+        val result = db.select("SELECT sku FROM order_lines WHERE order_id = 1 AND line_no = 2")
+        assertEquals(listOf(listOf(text("gadget"))), result.rows)
+        assertEquals(1, table.pointLookupCount)
+        assertEquals(0, table.scanCount)
+    }
+
+    @Test
+    fun compositePrimaryKeyEqualityOnOnlyOneColumnFallsBackToAFullScan() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        val table = db.catalog.table("order_lines")
+        val result = db.select("SELECT sku FROM order_lines WHERE order_id = 1")
+        assertEquals(setOf("widget", "gadget"), result.rows.map { (it.single() as Value.TextValue).value }.toSet())
+        assertEquals(1, table.scanCount)
+        assertEquals(0, table.pointLookupCount)
+    }
+
+    @Test
+    fun aRangeOnTheLastCompositePrimaryKeyColumnAfterAnEqualityPrefixUsesARangeScanNotAFullScan() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        val table = db.catalog.table("order_lines")
+        val result = db.select("SELECT sku FROM order_lines WHERE order_id = 1 AND line_no > 1")
+        assertEquals(listOf(listOf(text("gadget"))), result.rows)
+        assertEquals(1, table.rangeScanCount)
+        assertEquals(0, table.scanCount)
+    }
+
+    @Test
+    fun aTwoSidedRangeOnTheLastCompositePrimaryKeyColumnAfterAnEqualityPrefixUsesARangeScan() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        db.run("INSERT INTO order_lines VALUES (1, 3, 'sprocket')")
+        val table = db.catalog.table("order_lines")
+        val result = db.select("SELECT sku FROM order_lines WHERE order_id = 1 AND line_no >= 1 AND line_no <= 2")
+        assertEquals(setOf("widget", "gadget"), result.rows.map { (it.single() as Value.TextValue).value }.toSet())
+        assertEquals(1, table.rangeScanCount)
+        assertEquals(0, table.scanCount)
+    }
+
+    @Test
+    fun aRangeOnAnyOtherCompositePrimaryKeyColumnWithoutTheFullEqualityPrefixFallsBackToAFullScan() {
+        val db = newDatabase()
+        db.seedOrderLines()
+        val table = db.catalog.table("order_lines")
+        val result = db.select("SELECT sku FROM order_lines WHERE order_id > 1")
+        assertEquals(listOf(listOf(text("widget"))), result.rows)
+        assertEquals(1, table.scanCount)
+        assertEquals(0, table.rangeScanCount)
+    }
+