diff --git a/src/test/kotlin/kestrel/store/sql/DiskTableTest.kt b/src/test/kotlin/kestrel/store/sql/DiskTableTest.kt
index 16c71b1..8cc0cfd 100644
--- a/src/test/kotlin/kestrel/store/sql/DiskTableTest.kt
+++ b/src/test/kotlin/kestrel/store/sql/DiskTableTest.kt
@@ -155,3 +155,159 @@ class DiskTableTest {
         val txn = catalog.beginTransaction()
         val rid = table.insert(txn, row(1, "a", 1))
 
+        val bigName = "x".repeat(3000)
+        val newRid = table.update(txn, rid, row(1, bigName, 1))
+
+        assertTrue(newRid != rid, "growing the row past its slot capacity should relocate it")
+        assertEquals(bigName, table.findByPrimaryKey(txn, Value.IntegerValue(1))!!.second.values[1].let { (it as Value.TextValue).value })
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun abortingATransactionUndoesInsertsAndIndexEntries() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("users", usersColumns)
+        val committed = catalog.beginTransaction()
+        table.insert(committed, row(1, "alice", 30))
+        catalog.commit(committed)
+
+        val doomed = catalog.beginTransaction()
+        table.insert(doomed, row(2, "bob", 40))
+        catalog.abort(doomed)
+
+        val reader = catalog.beginTransaction()
+        assertEquals(listOf(1L), table.scan(reader).map { (it.second.values[0] as Value.IntegerValue).value })
+        assertNull(table.findByPrimaryKey(reader, Value.IntegerValue(2)))
+        catalog.commit(reader)
+    }
+
+    @Test
+    fun notNullConstraintIsEnforcedAtInsert() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("users", usersColumns)
+        val txn = catalog.beginTransaction()
+        assertFailsWith<DiskTableException> {
+            table.insert(txn, Row(listOf(Value.IntegerValue(1), Value.NullValue, Value.NullValue)))
+        }
+        catalog.abort(txn)
+    }
+
+    private val orderLineColumns = listOf(
+        Column("order_id", ColumnType.Integer, notNull = true, primaryKey = false),
+        Column("line_no", ColumnType.Integer, notNull = true, primaryKey = false),
+        Column("sku", ColumnType.Text, notNull = false, primaryKey = false)
+    )
+
+    private fun orderLine(orderId: Long, lineNo: Long, sku: String?) =
+        Row(listOf(Value.IntegerValue(orderId), Value.IntegerValue(lineNo), sku?.let { Value.TextValue(it) } ?: Value.NullValue))
+
+    @Test
+    fun compositePrimaryKeyAcceptsDistinctCombinationsAndRejectsDuplicates() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("order_lines", orderLineColumns, compositePrimaryKey = listOf("order_id", "line_no"))
+        val txn = catalog.beginTransaction()
+        table.insert(txn, orderLine(1, 1, "widget"))
+        table.insert(txn, orderLine(1, 2, "gadget"))
+        table.insert(txn, orderLine(2, 1, "widget"))
+
+        assertFailsWith<DiskTableException> { table.insert(txn, orderLine(1, 1, "duplicate")) }
+
+        assertEquals(3, table.scan(txn).size)
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun compositePrimaryKeyPointLookupFindsTheExactRowByAllColumns() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("order_lines", orderLineColumns, compositePrimaryKey = listOf("order_id", "line_no"))
+        val txn = catalog.beginTransaction()
+        table.insert(txn, orderLine(1, 1, "widget"))
+        table.insert(txn, orderLine(1, 2, "gadget"))
+
+        val found = table.findByPrimaryKey(txn, listOf(Value.IntegerValue(1), Value.IntegerValue(2)))!!
+        assertEquals("gadget", (found.second.values[2] as Value.TextValue).value)
+        assertEquals(null, table.findByPrimaryKey(txn, listOf(Value.IntegerValue(1), Value.IntegerValue(99))))
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun compositePrimaryKeyUpdateAndDeleteMoveOrRemoveTheIndexEntry() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("order_lines", orderLineColumns, compositePrimaryKey = listOf("order_id", "line_no"))
+        val txn = catalog.beginTransaction()
+        val rid = table.insert(txn, orderLine(1, 1, "widget"))
+
+        val newRid = table.update(txn, rid, orderLine(1, 2, "widget"))
+        assertNull(table.findByPrimaryKey(txn, listOf(Value.IntegerValue(1), Value.IntegerValue(1))))
+        assertEquals("widget", table.findByPrimaryKey(txn, listOf(Value.IntegerValue(1), Value.IntegerValue(2)))!!.second.values[2].let { (it as Value.TextValue).value })
+
+        table.delete(txn, newRid)
+        assertNull(table.findByPrimaryKey(txn, listOf(Value.IntegerValue(1), Value.IntegerValue(2))))
+        assertTrue(table.scan(txn).isEmpty())
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun compositePrimaryKeyUpdateThatCollidesWithAnotherRowIsRejected() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("order_lines", orderLineColumns, compositePrimaryKey = listOf("order_id", "line_no"))
+        val txn = catalog.beginTransaction()
+        val rid = table.insert(txn, orderLine(1, 1, "widget"))
+        table.insert(txn, orderLine(1, 2, "gadget"))
+
+        assertFailsWith<DiskTableException> { table.update(txn, rid, orderLine(1, 2, "widget")) }
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun decimalAndDateColumnsRoundTripThroughTheFullDiskBackedStack() {
+        val catalog = newCatalog()
+        val columns = listOf(
+            Column("id", ColumnType.Integer, notNull = true, primaryKey = true),
+            Column("amount", ColumnType.Decimal, notNull = false, primaryKey = false, precision = 12, scale = 4),
+            Column("placed", ColumnType.Date, notNull = false, primaryKey = false)
+        )
+        val table = catalog.createTable("orders", columns)
+        val txn = catalog.beginTransaction()
+        val amount = BigDecimal("123456.7890")
+        val placed = LocalDate.of(2024, 1, 15)
+        val inserted = Row(listOf(Value.IntegerValue(1), Value.DecimalValue(amount), Value.DateValue(placed)))
+        val rid = table.insert(txn, inserted)
+        assertEquals(inserted, table.get(txn, rid))
+        catalog.commit(txn)
+
+        val reader = catalog.beginTransaction()
+        assertEquals(inserted, table.get(reader, rid))
+        catalog.commit(reader)
+    }
+
+    // --- UNIQUE secondary index (CREATE UNIQUE INDEX): backed by the same B+Tree secondary
+    // index shape as a non-unique one, but insert/update reject a value collision.
+
+    @Test
+    fun uniqueSecondaryIndexRejectsAnInsertSharingAnExistingNonNullValue() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("users", usersColumns)
+        catalog.createIndex("users", "idx_name", "name", unique = true)
+        val txn = catalog.beginTransaction()
+        table.insert(txn, row(1, "alice", 30))
+
+        assertFailsWith<DiskTableException> { table.insert(txn, row(2, "alice", 40)) }
+
+        assertEquals(1, table.scan(txn).size)
+        catalog.commit(txn)
+    }
+
+    @Test
+    fun uniqueSecondaryIndexAllowsMultipleNullValues() {
+        val catalog = newCatalog()
+        val table = catalog.createTable("users", usersColumns)
+        catalog.createIndex("users", "idx_age", "age", unique = true)
+        val txn = catalog.beginTransaction()
+        table.insert(txn, row(1, "alice", null))
+        table.insert(txn, row(2, "bob", null))
+
+        assertEquals(2, table.scan(txn).size)
+        catalog.commit(txn)
+    }
+