feat(engine): extend Executor with applyCompositePrimaryKey

src/main/kotlin/kestrel/engine/Executor.kt

@@ -269,7 +269,9 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
     }
 
     private fun executeCreateTable(statement: CreateTableStatement): StatementResult {
-        if (statement.columns.count { it.primaryKey } > 1) {
+        if (statement.columns.count { it.primaryKey } > 1 ||
+            (statement.compositePrimaryKey.isNotEmpty() && statement.columns.any { it.primaryKey })
+        ) {
             throw KestrelException("table '${statement.tableName}' declares more than one primary key")
         }
         val seen = mutableSetOf<String>()
@@ -278,10 +280,199 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
                 throw KestrelException("duplicate column name: ${column.name}")
             }
         }
-        catalog.createTable(TableSchema(statement.tableName, statement.columns))
+        val columns = applyCompositePrimaryKey(statement)
+        validateForeignKeyDefinitions(statement, columns)
+        validateCheckDefinitions(statement)
+        validateUniqueDefinitions(statement)
+        validateDefaultDefinitions(columns)
+        validateGeneratedDefinitions(columns)
+        catalog.createTable(
+            TableSchema(
+                statement.tableName,
+                columns,
+                statement.foreignKeys,
+                statement.checks,
+                statement.uniqueConstraints,
+                statement.compositePrimaryKey,
+                statement.temporary
+            )
+        )
         return CreateTableResult(statement.tableName)
     }
 
+    /**
+     * A table-level `PRIMARY KEY (a, b, ...)` implies NOT NULL on every listed column, exactly
+     * like the inline `col TYPE PRIMARY KEY` spelling already does - applied here (rather than in
+     * the parser) because the parser builds [Column]s one at a time and has no view of the whole
+     * table-level clause yet when it does.
+     */
+    private fun applyCompositePrimaryKey(statement: CreateTableStatement): List<Column> {
+        if (statement.compositePrimaryKey.isEmpty()) return statement.columns
+        val declared = statement.compositePrimaryKey.map { it.lowercase() }.toSet()
+        if (declared.size != statement.compositePrimaryKey.size) {
+            throw KestrelException("table '${statement.tableName}' declares a duplicate column in its PRIMARY KEY")
+        }
+        for (name in statement.compositePrimaryKey) {
+            if (statement.columns.none { it.name.equals(name, ignoreCase = true) }) {
+                throw KestrelException("PRIMARY KEY refers to unknown column: $name")
+            }
+        }
+        return statement.columns.map { column ->
+            if (column.name.lowercase() in declared) column.copy(notNull = true) else column
+        }
+    }
+
+    private fun validateUniqueDefinitions(statement: CreateTableStatement) {
+        for (unique in statement.uniqueConstraints) {
+            for (columnName in unique.columns) {
+                if (statement.columns.none { it.name.equals(columnName, ignoreCase = true) }) {
+                    throw KestrelException("UNIQUE constraint refers to unknown column: $columnName")
+                }
+            }
+        }
+    }
+
+    /**
+     * A self-reference (a FK on the table pointing back at itself, e.g. an employee/manager
+     * hierarchy) is checked against a [TableSchema] built from [columns] (already normalized by
+     * [applyCompositePrimaryKey]) rather than the catalog, because the table doesn't exist there
+     * yet while it's being created.
+     */
+    private fun validateForeignKeyDefinitions(statement: CreateTableStatement, columns: List<Column>) {
+        for (fk in statement.foreignKeys) {
+            if (fk.columns.size != fk.referencedColumns.size) {
+                throw KestrelException(
+                    "FOREIGN KEY (${fk.columns.joinToString(", ")}) column count does not match " +
+                        "REFERENCES ${fk.referencedTable}(${fk.referencedColumns.joinToString(", ")})"
+                )
+            }
+            for (columnName in fk.columns) {
+                if (columns.none { it.name.equals(columnName, ignoreCase = true) }) {
+                    throw KestrelException("FOREIGN KEY refers to unknown column: $columnName")
+                }
+            }
+            val selfReference = fk.referencedTable.equals(statement.tableName, ignoreCase = true)
+            val hasEnforcedUniqueness = if (selfReference) {
+                for (columnName in fk.referencedColumns) {
+                    if (columns.none { it.name.equals(columnName, ignoreCase = true) }) {
+                        throw KestrelException(
+                            "FOREIGN KEY references unknown column: ${fk.referencedTable}.$columnName"
+                        )
+                    }
+                }
+                val selfSchema = TableSchema(
+                    statement.tableName,
+                    columns,
+                    uniqueConstraints = statement.uniqueConstraints,
+                    compositePrimaryKey = statement.compositePrimaryKey
+                )
+                selfSchema.hasEnforcedUniqueness(fk.referencedColumns)
+            } else {
+                val referencedTable = catalog.tableOrNull(fk.referencedTable)
+                    ?: throw KestrelException("FOREIGN KEY references unknown table: ${fk.referencedTable}")
+                for (columnName in fk.referencedColumns) {
+                    if (referencedTable.schema.column(columnName) == null) {
+                        throw KestrelException(
+                            "FOREIGN KEY references unknown column: ${fk.referencedTable}.$columnName"
+                        )
+                    }
+                }
+                referencedTable.schema.hasEnforcedUniqueness(fk.referencedColumns)
+            }
+            if (!hasEnforcedUniqueness) {
+                throw KestrelException(
+                    "FOREIGN KEY must reference a PRIMARY KEY or UNIQUE column(s), but " +
+                        "${fk.referencedTable}(${fk.referencedColumns.joinToString(", ")}) is not one"
+                )
+            }
+        }
+    }
+
+    private fun validateCheckDefinitions(statement: CreateTableStatement) {
+        val schema = TableSchema(statement.tableName, statement.columns)
+        for (check in statement.checks) {
+            checkColumns(check.expression, schema)
+            requireNoSubquery(check.expression)
+        }
+    }
+
+    private fun validateDefaultDefinitions(columns: List<Column>) {
+        for (column in columns) {
+            val default = column.default ?: continue
+            validateConstantDefault(default)
+            if (default is NextValueExpression) {
+                validateNextValueDefault(default, column)
+                continue
+            }
+            coerceForColumn(Evaluator().evaluate(default), column)
+        }
+    }
+
+    /**
+     * A GENERATED column's expression is recomputed from this row's OTHER column values on every
+     * INSERT/UPDATE (see [applyGeneratedColumns]) - it must therefore be a pure function of that
+     * row's own columns: [checkColumns] rejects a reference to a nonexistent column plus any
+     * aggregate/window function, [requireNoSubquery] rejects a subquery and NEXTVAL/CURRVAL/
+     * MERGE_ACTION() on top of that (same two-function combination [validateCheckDefinitions] uses
+     * for CHECK), and [requireNoGeneratedColumnReference] additionally forbids referencing another
+     * GENERATED column - not because it's unsound in general, but because allowing it would need a
+     * dependency-ordered evaluation pass instead of the current single independent pass over every
+     * GENERATED column (see [applyGeneratedColumns]), matching Postgres's own restriction.
+     */
+    private fun validateGeneratedDefinitions(columns: List<Column>) {
+        val schema = TableSchema("", columns)
+        val generatedNames = columns.filter { it.generated != null }.map { it.name.lowercase() }.toSet()
+        for (column in columns) {
+            val generated = column.generated ?: continue
+            if (column.default != null) {
+                throw KestrelException("column ${column.name} cannot declare both DEFAULT and GENERATED ALWAYS AS")
+            }
+            checkColumns(generated, schema)
+            requireNoSubquery(generated)
+            requireNoGeneratedColumnReference(generated, generatedNames, column.name)
+        }
+    }
+
+    private fun requireNoGeneratedColumnReference(expression: Expression, generatedNames: Set<String>, ownerColumn: String) {
+        when (expression) {
+            is ColumnExpression -> if (expression.name.lowercase() in generatedNames) {
+                throw KestrelException(
+                    "GENERATED column $ownerColumn cannot reference another GENERATED column: ${expression.name}"
+                )
+            }
+            is LiteralExpression -> Unit
+            is BinaryExpression -> {
+                requireNoGeneratedColumnReference(expression.left, generatedNames, ownerColumn)
+                requireNoGeneratedColumnReference(expression.right, generatedNames, ownerColumn)
+            }
+            is UnaryExpression -> requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+            is CallExpression -> expression.arguments.forEach { requireNoGeneratedColumnReference(it, generatedNames, ownerColumn) }
+            is IsNullExpression -> requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+            is InExpression -> {
+                requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+                expression.values.forEach { requireNoGeneratedColumnReference(it, generatedNames, ownerColumn) }
+            }
+            is BetweenExpression -> {
+                requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+                requireNoGeneratedColumnReference(expression.low, generatedNames, ownerColumn)
+                requireNoGeneratedColumnReference(expression.high, generatedNames, ownerColumn)
+            }
+            is LikeExpression -> {
+                requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+                requireNoGeneratedColumnReference(expression.pattern, generatedNames, ownerColumn)
+            }
+            is CaseExpression -> {
+                expression.operand?.let { requireNoGeneratedColumnReference(it, generatedNames, ownerColumn) }
+                expression.branches.forEach {
+                    requireNoGeneratedColumnReference(it.condition, generatedNames, ownerColumn)
+                    requireNoGeneratedColumnReference(it.result, generatedNames, ownerColumn)
+                }
+                expression.elseExpression?.let { requireNoGeneratedColumnReference(it, generatedNames, ownerColumn) }
+            }
+            is CastExpression -> requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+            is CollateExpression -> requireNoGeneratedColumnReference(expression.operand, generatedNames, ownerColumn)
+            // Already rejected earlier in validateGeneratedDefinitions by checkColumns/requireNoSubquery -
+            // kept here only so this `when` stays exhaustive.
     private fun executeDropTable(statement: DropTableStatement): StatementResult {
         catalog.dropTable(statement.tableName)
         return DropTableResult(statement.tableName)