diff --git a/src/test/kotlin/kestrel/CompositeKeyTest.kt b/src/test/kotlin/kestrel/CompositeKeyTest.kt
new file mode 100644
index 0000000..f9f7eb0
--- /dev/null
+++ b/src/test/kotlin/kestrel/CompositeKeyTest.kt
@@ -0,0 +1,158 @@
+package kestrel
+
+import kestrel.catalog.Catalog
+import kestrel.engine.Executor
+import kestrel.engine.InsertResult
+import kestrel.engine.SelectResult
+import kestrel.engine.StatementResult
+import kestrel.persistence.SqlWriter
+import kestrel.sql.ast.CreateTableStatement
+import kestrel.sql.ast.Statement
+import kestrel.sql.lexer.Lexer
+import kestrel.sql.parser.Parser
+import kestrel.storage.Value
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+/**
+ * Multi-column PRIMARY KEY (`PRIMARY KEY (a, b, ...)`) and multi-column FOREIGN KEY
+ * (`FOREIGN KEY (a, b) REFERENCES parent(x, y)`), the last documented "not yet supported" gap
+ * left in TableSchema/ForeignKeyEnforcer after single-column UNIQUE landed.
+ */
+class CompositeKeyTest {
+    private val catalog = Catalog()
+    private val executor = Executor(catalog)
+
+    private fun run(sql: String): List<StatementResult> =
+        Parser(Lexer(sql).tokenize()).parseProgram().map { executor.execute(it) }
+
+    private fun exec(sql: String): StatementResult = run(sql).single()
+
+    private fun parseOne(sql: String): Statement = Parser(Lexer(sql).tokenize()).parseProgram().single()
+
+    private fun select(sql: String): SelectResult = exec(sql) as SelectResult
+
+    private fun int(value: Long) = Value.IntegerValue(value)
+
+    // ---- parsing ----
+
+    @Test
+    fun parsesTableLevelCompositePrimaryKey() {
+        val statement = parseOne(
+            "CREATE TABLE line_items (order_id INTEGER, sku TEXT, qty INTEGER, PRIMARY KEY (order_id, sku))"
+        ) as CreateTableStatement
+        assertEquals(listOf("order_id", "sku"), statement.compositePrimaryKey)
+        assertTrue(statement.columns.none { it.primaryKey })
+    }
+
+    @Test
+    fun parsesTableLevelCompositeForeignKey() {
+        val statement = parseOne(
+            "CREATE TABLE t (a INTEGER, b INTEGER, " +
+                "FOREIGN KEY (a, b) REFERENCES parent (x, y) ON DELETE CASCADE)"
+        ) as CreateTableStatement
+        val fk = statement.foreignKeys.single()
+        assertEquals(listOf("a", "b"), fk.columns)
+        assertEquals("parent", fk.referencedTable)
+        assertEquals(listOf("x", "y"), fk.referencedColumns)
+    }
+
+    @Test
+    fun mismatchedForeignKeyColumnCountIsRejectedAtParseTime() {
+        assertFailsWith<Exception> {
+            parseOne("CREATE TABLE t (a INTEGER, b INTEGER, FOREIGN KEY (a, b) REFERENCES parent (x))")
+        }
+    }
+
+    // ---- CREATE TABLE validation ----
+
+    @Test
+    fun declaringBothInlineAndTableLevelPrimaryKeyIsRejected() {
+        assertFailsWith<KestrelException> {
+            run("CREATE TABLE t (a INTEGER PRIMARY KEY, b INTEGER, PRIMARY KEY (a, b))")
+        }
+    }
+
+    @Test
+    fun duplicateColumnWithinCompositePrimaryKeyIsRejected() {
+        assertFailsWith<KestrelException> {
+            run("CREATE TABLE t (a INTEGER, b INTEGER, PRIMARY KEY (a, a))")
+        }
+    }
+
+    @Test
+    fun compositePrimaryKeyReferringToUnknownColumnIsRejected() {
+        assertFailsWith<KestrelException> {
+            run("CREATE TABLE t (a INTEGER, b INTEGER, PRIMARY KEY (a, missing))")
+        }
+    }
+
+    @Test
+    fun compositePrimaryKeyImpliesNotNullOnEveryListedColumn() {
+        run("CREATE TABLE t (a INTEGER, b INTEGER, PRIMARY KEY (a, b))")
+        assertFailsWith<KestrelException> { run("INSERT INTO t VALUES (1, NULL)") }
+        assertFailsWith<KestrelException> { run("INSERT INTO t VALUES (NULL, 1)") }
+    }
+
+    @Test
+    fun foreignKeyMustReferenceAnExactEnforcedUniqueGroupNotAnArbitraryPair() {
+        run("CREATE TABLE parent (a INTEGER, b INTEGER, c INTEGER, PRIMARY KEY (a, b))")
+        assertFailsWith<KestrelException> {
+            run("CREATE TABLE child (x INTEGER, y INTEGER, FOREIGN KEY (x, y) REFERENCES parent (b, c))")
+        }
+    }
+
+    @Test
+    fun foreignKeyMayReferenceACompositePrimaryKeyInAnyColumnOrder() {
+        run("CREATE TABLE parent (a INTEGER, b INTEGER, PRIMARY KEY (a, b))")
+        run("CREATE TABLE child (x INTEGER, y INTEGER, FOREIGN KEY (x, y) REFERENCES parent (b, a))")
+    }
+
+    // ---- composite PRIMARY KEY uniqueness ----
+
+    @Test
+    fun compositePrimaryKeyRejectsAnExactDuplicateCombination() {
+        run("CREATE TABLE line_items (order_id INTEGER, sku TEXT, qty INTEGER, PRIMARY KEY (order_id, sku))")
+        run("INSERT INTO line_items VALUES (1, 'A', 3)")
+        assertFailsWith<KestrelException> { run("INSERT INTO line_items VALUES (1, 'A', 5)") }
+    }
+
+    @Test
+    fun compositePrimaryKeyAllowsDifferentCombinationsSharingOneColumnValue() {
+        run("CREATE TABLE line_items (order_id INTEGER, sku TEXT, qty INTEGER, PRIMARY KEY (order_id, sku))")
+        run("INSERT INTO line_items VALUES (1, 'A', 3)")
+        val result = exec("INSERT INTO line_items VALUES (1, 'B', 3)")
+        assertEquals(InsertResult(1), result)
+    }
+
+    @Test
+    fun updateThatWouldCollideWithAnotherRowsCompositePrimaryKeyIsRejected() {
+        run("CREATE TABLE line_items (order_id INTEGER, sku TEXT, qty INTEGER, PRIMARY KEY (order_id, sku))")
+        run("INSERT INTO line_items VALUES (1, 'A', 3), (1, 'B', 5)")
+        assertFailsWith<KestrelException> { run("UPDATE line_items SET sku = 'A' WHERE sku = 'B'") }
+    }
+
+    // ---- composite FOREIGN KEY: existence checks ----
+
+    private fun seedOrdersAndLineItems() {
+        run("CREATE TABLE orders (region TEXT, order_no INTEGER, PRIMARY KEY (region, order_no))")
+        run(
+            "CREATE TABLE line_items (" +
+                "region TEXT, order_no INTEGER, sku TEXT, qty INTEGER, " +
+                "FOREIGN KEY (region, order_no) REFERENCES orders (region, order_no) " +
+                "ON DELETE CASCADE ON UPDATE CASCADE)"
+        )
+        run("INSERT INTO orders VALUES ('eu', 1), ('us', 1)")
+    }
+
+    @Test
+    fun compositeForeignKeyRejectsAPartialMatch() {
+        seedOrdersAndLineItems()
+        // ('eu', 1) and ('us', 1) both exist individually but not as the pair ('us', 2).
+        assertFailsWith<KestrelException> {
+            run("INSERT INTO line_items VALUES ('us', 2, 'SKU1', 3)")
+        }
+    }
+