diff --git a/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt b/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
index 0483ce6..9923374 100644
--- a/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
+++ b/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
@@ -2215,3 +2215,161 @@ class DiskExecutorTest {
         )
     }
 
+    @Test
+    fun lagWithAnExplicitOffsetSkipsMultipleRows() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select("SELECT id, LAG(salary, 2) OVER (ORDER BY id) AS prev2 FROM employees ORDER BY id ASC")
+        assertEquals(
+            listOf(
+                listOf(int(1), Value.NullValue), listOf(int(2), Value.NullValue), listOf(int(3), int(400)),
+                listOf(int(4), int(400)), listOf(int(5), int(300)), listOf(int(6), int(200))
+            ),
+            result.rows
+        )
+    }
+
+    @Test
+    fun lagPastThePartitionEdgeFallsBackToTheDefaultExpressionInsteadOfNull() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, LAG(salary, 1, 0) OVER (PARTITION BY deptId ORDER BY id) AS prev FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(int(0), int(400), int(400), int(0), int(200), int(200)), result.rows.map { it[1] })
+    }
+
+    @Test
+    fun lagOffsetMustNotBeNegative() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val error = assertFailsWith<DiskExecutorException> {
+            db.run("SELECT LAG(salary, -1) OVER (ORDER BY id) FROM employees")
+        }
+        assertTrue(error.message!!.contains("negative"), error.message!!)
+    }
+
+    @Test
+    fun lagOffsetMustBeAnInteger() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val error = assertFailsWith<DiskExecutorException> {
+            db.run("SELECT LAG(salary, 1.5) OVER (ORDER BY id) FROM employees")
+        }
+        assertTrue(error.message!!.contains("integer"), error.message!!)
+    }
+
+    @Test
+    fun firstValueReturnsTheFirstRowsValueForEveryRowInThePartition() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, FIRST_VALUE(salary) OVER (PARTITION BY deptId ORDER BY id) AS f FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(400L, 400L, 400L, 200L, 200L, 200L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun lastValueSharesOneValuePerPeerGroupRatherThanTheWholePartition() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, LAST_VALUE(salary) OVER (PARTITION BY deptId ORDER BY salary DESC) AS l FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(400L, 400L, 300L, 200L, 200L, 100L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun lastValueWithoutOverOrderByUsesTheWholePartitionsLastRow() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, LAST_VALUE(salary) OVER (PARTITION BY deptId) AS l FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(300L, 300L, 300L, 100L, 100L, 100L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun ntileSplitsEachPartitionIntoAsEvenBucketsAsPossible() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, NTILE(2) OVER (PARTITION BY deptId ORDER BY salary DESC) AS bucket FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(1L, 1L, 2L, 1L, 1L, 2L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun ntileWithMoreBucketsThanRowsLeavesTheHighestBucketsUnused() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, NTILE(5) OVER (PARTITION BY deptId ORDER BY salary DESC) AS bucket FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(1L, 2L, 3L, 1L, 2L, 3L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun ntileBucketCountMustBePositive() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val error = assertFailsWith<DiskExecutorException> {
+            db.run("SELECT NTILE(0) OVER (ORDER BY id) FROM employees")
+        }
+        assertTrue(error.message!!.contains("positive"), error.message!!)
+    }
+
+    @Test
+    fun ntileBucketCountMustBeAnInteger() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val error = assertFailsWith<DiskExecutorException> {
+            db.run("SELECT NTILE(1.5) OVER (ORDER BY id) FROM employees")
+        }
+        assertTrue(error.message!!.contains("integer"), error.message!!)
+    }
+
+    @Test
+    fun windowFunctionNestedInsideLagsOffsetIsRejected() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val error = assertFailsWith<DiskExecutorException> {
+            db.run("SELECT LAG(salary, ROW_NUMBER() OVER (ORDER BY id)) OVER (ORDER BY id) FROM employees")
+        }
+        assertTrue(error.message!!.contains("nested"), error.message!!)
+    }
+
+    @Test
+    fun explicitRowsFrameProducesAMovingAverageInsteadOfARunningTotal() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, SUM(salary) OVER (PARTITION BY deptId ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS s " +
+                "FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(400L, 800L, 700L, 200L, 400L, 300L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun rowsBetweenCurrentRowAndFollowingLooksForward() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val result = db.select(
+            "SELECT id, SUM(salary) OVER (PARTITION BY deptId ORDER BY id ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS s " +
+                "FROM employees ORDER BY id ASC"
+        )
+        assertEquals(listOf(800L, 700L, 300L, 400L, 300L, 100L), result.rows.map { (it[1] as Value.IntegerValue).value })
+    }
+
+    @Test
+    fun rowsBetweenUnboundedPrecedingAndCurrentRowMatchesTheImplicitDefaultOnAUniqueOrderKey() {
+        val db = newDatabase()
+        db.seedWindowEmployees()
+        val explicit = db.select(
+            "SELECT id, SUM(salary) OVER (ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS s " +
+                "FROM employees ORDER BY id ASC"
+        )
+        val implicit = db.select("SELECT id, SUM(salary) OVER (ORDER BY id) AS s FROM employees ORDER BY id ASC")
+        assertEquals(implicit.rows, explicit.rows)
+    }
+