diff --git a/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt b/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
index b1ccfdd..179e288 100644
--- a/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
+++ b/src/test/kotlin/kestrel/store/sql/DiskExecutorTest.kt
@@ -635,3 +635,163 @@ class DiskExecutorTest {
         )
     }
 
+    @Test
+    fun leftJoinLateralOverAnEmptyBaseTableProducesNoRowsWithoutRunningTheSubquery() {
+        val db = newDatabase()
+        db.run("CREATE TABLE departments (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
+        db.run("CREATE TABLE employees (id INTEGER PRIMARY KEY, deptId INTEGER, name TEXT)")
+        val result = db.select(
+            "SELECT d.name, top.name FROM departments d " +
+                "LEFT JOIN LATERAL (SELECT e.name FROM employees e WHERE e.deptId = d.id) AS top ON TRUE"
+        )
+        assertEquals(emptyList(), result.rows)
+    }
+
+    @Test
+    fun chainedLateralJoinsSeeEarlierLateralAliases() {
+        val db = newDatabase()
+        db.seedDepartmentsAndSalaries()
+        db.run("CREATE TABLE dual (x INTEGER)")
+        db.run("INSERT INTO dual VALUES (1)")
+        val result = db.select(
+            "SELECT d.name, bonus.amount FROM departments d " +
+                "JOIN LATERAL (SELECT e.salary FROM employees e WHERE e.deptId = d.id " +
+                "ORDER BY e.salary DESC LIMIT 1) AS top ON TRUE " +
+                "JOIN LATERAL (SELECT top.salary / 10 AS amount FROM dual) AS bonus ON TRUE " +
+                "ORDER BY d.name ASC"
+        )
+        assertEquals(listOf(listOf(text("eng"), int(40)), listOf(text("sales"), int(20))), result.rows)
+    }
+
+    @Test
+    fun lateralSubqueryCannotReferenceATableJoinedLater() {
+        val db = newDatabase()
+        db.seedDepartmentsAndSalaries()
+        val failure = assertFailsWith<DiskExecutorException> {
+            db.select(
+                "SELECT * FROM departments d " +
+                    "JOIN LATERAL (SELECT e.name FROM employees e WHERE e.salary = later.salary) AS top ON TRUE " +
+                    "JOIN employees later ON later.deptId = d.id"
+            )
+        }
+        assertTrue(failure.message!!.contains("not joined yet"))
+    }
+
+    @Test
+    fun lateralSubqueryWithStarProjectionResolvesColumnsStructurallyFromItsBaseTable() {
+        val db = newDatabase()
+        db.seedDepartmentsAndSalaries()
+        val result = db.select(
+            "SELECT d.name, top.name, top.salary FROM departments d " +
+                "JOIN LATERAL (SELECT * FROM employees e WHERE e.deptId = d.id " +
+                "ORDER BY e.salary DESC LIMIT 1) AS top ON TRUE " +
+                "ORDER BY d.name ASC"
+        )
+        assertEquals(
+            listOf(listOf(text("eng"), text("grace"), int(400)), listOf(text("sales"), text("kay"), int(200))),
+            result.rows
+        )
+    }
+
+    @Test
+    fun groupByComputesCountSumAvgMinMaxPerGroup() {
+        val db = newDatabase()
+        db.seedOrders()
+        val result = db.select(
+            "SELECT customer_id, COUNT(*), SUM(amount), AVG(amount), MIN(amount), MAX(amount) " +
+                "FROM orders GROUP BY customer_id ORDER BY customer_id"
+        )
+        assertEquals(
+            listOf(
+                listOf(int(1), int(2), int(150), Value.RealValue(75.0), int(50), int(100)),
+                listOf(int(2), int(1), int(200), Value.RealValue(200.0), int(200), int(200))
+            ),
+            result.rows
+        )
+    }
+
+    /**
+     * STRING_AGG's implementation lives entirely in the shared [kestrel.engine.GroupAggregation]/
+     * [kestrel.engine.AggregateFunctions] classes that this disk-backed executor already reuses for
+     * every other aggregate - this proves that reuse actually extends to STRING_AGG too, not just that
+     * the in-memory engine (AggregateTest) got it right.
+     */
+    @Test
+    fun stringAggGroupsAndOrdersPerGroupOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE tags (item TEXT, label TEXT)")
+        db.run("INSERT INTO tags VALUES ('widget', 'blue'), ('widget', 'small'), ('gadget', 'red')")
+        val result = db.select(
+            "SELECT item, STRING_AGG(label, ', ' ORDER BY label) FROM tags GROUP BY item ORDER BY item"
+        )
+        assertEquals(
+            listOf(listOf(text("gadget"), text("red")), listOf(text("widget"), text("blue, small"))),
+            result.rows
+        )
+    }
+
+    /**
+     * `COUNT/SUM/AVG/MIN/MAX(DISTINCT ...)` dedup lives in the same shared [kestrel.engine.
+     * AggregateFunctions.compute] this executor already reuses for every other aggregate - this
+     * proves the DISTINCT dedup path itself, not just plain aggregation, actually runs on the
+     * disk-backed engine too.
+     */
+    @Test
+    fun countAndSumDistinctDedupOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE t (dept TEXT, amount INTEGER)")
+        db.run("INSERT INTO t VALUES ('eng', 100), ('eng', 100), ('eng', 200), ('ops', 100), ('ops', NULL)")
+        assertEquals(listOf(listOf(int(4))), db.select("SELECT COUNT(amount) FROM t").rows)
+        assertEquals(listOf(listOf(int(2))), db.select("SELECT COUNT(DISTINCT amount) FROM t").rows)
+        assertEquals(listOf(listOf(int(500))), db.select("SELECT SUM(amount) FROM t").rows)
+        assertEquals(listOf(listOf(int(300))), db.select("SELECT SUM(DISTINCT amount) FROM t").rows)
+    }
+
+    @Test
+    fun countDistinctPerGroupOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE t (dept TEXT, amount INTEGER)")
+        db.run("INSERT INTO t VALUES ('eng', 100), ('eng', 100), ('eng', 200), ('ops', 100), ('ops', NULL)")
+        val result = db.select("SELECT dept, COUNT(DISTINCT amount) FROM t GROUP BY dept ORDER BY dept ASC")
+        assertEquals(listOf(listOf(text("eng"), int(2)), listOf(text("ops"), int(1))), result.rows)
+    }
+
+    /**
+     * `FILTER (WHERE ...)` narrows the rows an aggregate call itself sees before
+     * [kestrel.engine.AggregateFunctions.compute] runs - lives in the same shared
+     * [kestrel.engine.GroupAggregation.computeAggregate] this executor already reuses for every other
+     * aggregate, so this proves the filtering itself, not just plain/DISTINCT aggregation, runs here too.
+     */
+    @Test
+    fun filterRestrictsWhichRowsEachAggregateSeesOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE t (dept TEXT, amount INTEGER)")
+        db.run("INSERT INTO t VALUES ('eng', 100), ('eng', 100), ('eng', 200), ('ops', 100), ('ops', NULL)")
+        assertEquals(listOf(listOf(int(1))), db.select("SELECT COUNT(*) FILTER (WHERE amount > 100) FROM t").rows)
+        assertEquals(listOf(listOf(int(400))), db.select("SELECT SUM(amount) FILTER (WHERE dept = 'eng') FROM t").rows)
+        assertEquals(
+            listOf(listOf(int(0), Value.NullValue)),
+            db.select("SELECT COUNT(*) FILTER (WHERE dept = 'nope') AS c, SUM(amount) FILTER (WHERE dept = 'nope') AS s FROM t").rows
+        )
+    }
+
+    @Test
+    fun filterCombinesWithDistinctAndGroupByOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE t (dept TEXT, amount INTEGER)")
+        db.run("INSERT INTO t VALUES ('eng', 100), ('eng', 100), ('eng', 200), ('ops', 100), ('ops', NULL)")
+        val result = db.select(
+            "SELECT dept, COUNT(DISTINCT amount) FILTER (WHERE amount > 50) AS matching FROM t GROUP BY dept ORDER BY dept ASC"
+        )
+        assertEquals(listOf(listOf(text("eng"), int(2)), listOf(text("ops"), int(1))), result.rows)
+    }
+
+    @Test
+    fun offsetSkipsGroupsAfterOrderingAggregatedResultsOnTheDiskBackedEngine() {
+        val db = newDatabase()
+        db.run("CREATE TABLE t (dept TEXT, amount INTEGER)")
+        db.run("INSERT INTO t VALUES ('eng', 100), ('eng', 100), ('eng', 200), ('ops', 100), ('ops', NULL)")
+        val result = db.select("SELECT dept, COUNT(DISTINCT amount) FROM t GROUP BY dept ORDER BY dept ASC LIMIT 1 OFFSET 1")
+        assertEquals(listOf(listOf(text("ops"), int(1))), result.rows)
+    }
+