From 91db4cd952c12f99b3977ff4bb6eb93fa8e833f9 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 18:11:56 +0200 Subject: [PATCH] feat(sql): add DiskForeignKeyEnforcer and validateRow --- .../store/sql/DiskForeignKeyEnforcer.kt | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 src/main/kotlin/kestrel/store/sql/DiskForeignKeyEnforcer.kt diff --git a/src/main/kotlin/kestrel/store/sql/DiskForeignKeyEnforcer.kt b/src/main/kotlin/kestrel/store/sql/DiskForeignKeyEnforcer.kt new file mode 100644 index 0000000..4b1224e --- /dev/null +++ b/src/main/kotlin/kestrel/store/sql/DiskForeignKeyEnforcer.kt @@ -0,0 +1,201 @@ +package kestrel.store.sql + +import kestrel.catalog.ForeignKeyAction +import kestrel.catalog.TableSchema +import kestrel.store.heap.RecordId +import kestrel.storage.Row +import kestrel.storage.Value + +/** + * Disk-backed counterpart of [kestrel.engine.ForeignKeyEnforcer]: existence checks on INSERT/UPDATE + * of the referencing (child) table, and ON DELETE/ON UPDATE actions on the referenced (parent) + * table, operating on [DiskCatalog]/[DiskTable] within an active disk transaction instead of the + * in-memory engine's `Table.rows` ArrayList. Deliberately duplicated rather than shared - same + * reasoning as every other validation [DiskExecutor] duplicates (see + * [DiskExecutor.validateUniqueDefinitions]). + * + * [validateRow]'s existence check is a full [DiskTable.scan] of the referenced table, exactly + * mirroring the in-memory engine's own full scan of `Table.rows` - a documented performance + * compromise (not correctness), same category as CTE materialization or the disk executor's UNIQUE + * constraint scan: a real index-driven point lookup here would need to first prove the referenced + * columns line up with the PRIMARY KEY or a secondary index before choosing a plan, which is scope + * this pass does not take on. + */ +internal object DiskForeignKeyEnforcer { + + /** + * [uncommittedSiblings] covers a self-referencing FK (`referencedTable == schema.name`, e.g. an + * employee/manager hierarchy) inside a multi-row INSERT/UPDATE statement: those sibling rows are + * being validated as part of the very same statement and have not been written to [txn] yet, so + * without this a later row could not reference an earlier row from the same batch (or vice + * versa - order within the batch must not matter). + * + * Follows MATCH SIMPLE (the SQL default): if *any* column of the FK is NULL in this row, the + * whole constraint is skipped for that row - not just the null column. + */ + fun validateRow( + catalog: DiskCatalog, + txn: Long, + schema: TableSchema, + values: List, + uncommittedSiblings: List = emptyList() + ) { + for (fk in schema.foreignKeys) { + val columnIndices = fk.columns.map { schema.columnIndex(it)!! } + val key = columnIndices.map { values[it] } + if (key.any { it is Value.NullValue }) continue + val referenced = catalog.tableOrNull(fk.referencedTable) + ?: throw DiskExecutorException( + "FOREIGN KEY constraint failed: table '${fk.referencedTable}' referenced by " + + "${schema.name}(${fk.columns.joinToString(", ")}) does not exist" + ) + val refIndices = fk.referencedColumns.map { referenced.schema.columnIndex(it)!! } + val siblings = if (fk.referencedTable.equals(schema.name, ignoreCase = true)) uncommittedSiblings else emptyList() + val exists = referenced.scan(txn).any { (_, row) -> refIndices.map { idx -> row.values[idx] } == key } || + siblings.any { row -> refIndices.map { idx -> row.values[idx] } == key } + if (!exists) { + throw DiskExecutorException( + "FOREIGN KEY constraint failed: ${schema.name}(${fk.columns.joinToString(", ")}) has no " + + "matching row in ${fk.referencedTable}(${fk.referencedColumns.joinToString(", ")})" + ) + } + } + } + + private sealed interface CascadeStep { + data class Delete(val table: DiskTable, val rows: List>, val next: List) : CascadeStep + data class SetToNull(val table: DiskTable, val columnIndices: List, val rows: List>) : CascadeStep + data class SetTo(val table: DiskTable, val columnIndices: List, val rows: List>, val values: List) : CascadeStep + } + + fun cascadeDelete(catalog: DiskCatalog, txn: Long, parentTableName: String, deletedRows: List) { + applySteps(txn, planDelete(catalog, txn, parentTableName, deletedRows)) + } + + /** + * [referencedColumns] is the full column group whose values changed together (a PRIMARY KEY or + * UNIQUE key group) and [changedPairs] holds the old/new key tuple for every row of the parent + * table, positionally aligned with [referencedColumns] - a matching FK may declare its own + * `referencedColumns` in a different order, so [planUpdate] re-projects each tuple by name. + */ + fun cascadeUpdate( + catalog: DiskCatalog, + txn: Long, + parentTableName: String, + referencedColumns: List, + changedPairs: List, List>> + ) { + applySteps(txn, planUpdate(catalog, txn, parentTableName, referencedColumns, changedPairs)) + } + + private fun planDelete(catalog: DiskCatalog, txn: Long, parentTableName: String, deletedRows: List): List { + if (deletedRows.isEmpty()) return emptyList() + val parent = catalog.table(parentTableName) + val steps = mutableListOf() + for (childName in catalog.tableNames) { + val child = catalog.table(childName) + for (fk in child.schema.foreignKeys) { + if (!fk.referencedTable.equals(parentTableName, ignoreCase = true)) continue + val refIndices = fk.referencedColumns.map { parent.schema.columnIndex(it)!! } + val deletedKeys = deletedRows.map { row -> refIndices.map { row.values[it] } }.toHashSet() + val childColumnIndices = fk.columns.map { child.schema.columnIndex(it)!! } + val matches = child.scan(txn).filter { (_, row) -> childColumnIndices.map { row.values[it] } in deletedKeys } + if (matches.isEmpty()) continue + when (fk.onDelete) { + ForeignKeyAction.Cascade -> + steps.add(CascadeStep.Delete(child, matches, planDelete(catalog, txn, child.schema.name, matches.map { it.second }))) + ForeignKeyAction.SetNull -> { + requireNullable(child.schema, childColumnIndices, "ON DELETE SET NULL") + steps.add(CascadeStep.SetToNull(child, childColumnIndices, matches)) + } + ForeignKeyAction.Restrict, ForeignKeyAction.NoAction -> throw DiskExecutorException( + "FOREIGN KEY constraint failed: ${matches.size} row(s) in ${child.schema.name} still " + + "reference $parentTableName (use ON DELETE CASCADE or remove them first)" + ) + } + } + } + return steps + } + + private fun planUpdate( + catalog: DiskCatalog, + txn: Long, + parentTableName: String, + referencedColumns: List, + changedPairs: List, List>> + ): List { + val changes = changedPairs.filter { (old, new) -> old != new } + if (changes.isEmpty()) return emptyList() + val steps = mutableListOf() + for (childName in catalog.tableNames) { + val child = catalog.table(childName) + for (fk in child.schema.foreignKeys) { + if (!fk.referencedTable.equals(parentTableName, ignoreCase = true)) continue + if (fk.referencedColumns.size != referencedColumns.size) continue + // Re-project the tuple by name: this FK's referencedColumns may list the same key + // group in a different order than the one changedPairs was built from. + val positions = fk.referencedColumns.map { refCol -> + referencedColumns.indexOfFirst { it.equals(refCol, ignoreCase = true) } + } + if (positions.any { it == -1 }) continue + val childColumnIndices = fk.columns.map { child.schema.columnIndex(it)!! } + for ((oldTuple, newTuple) in changes) { + val oldKey = positions.map { oldTuple[it] } + val newKey = positions.map { newTuple[it] } + if (oldKey == newKey) continue + val matches = child.scan(txn).filter { (_, row) -> childColumnIndices.map { row.values[it] } == oldKey } + if (matches.isEmpty()) continue + when (fk.onUpdate) { + ForeignKeyAction.Cascade -> + steps.add(CascadeStep.SetTo(child, childColumnIndices, matches, newKey)) + ForeignKeyAction.SetNull -> { + requireNullable(child.schema, childColumnIndices, "ON UPDATE SET NULL") + steps.add(CascadeStep.SetToNull(child, childColumnIndices, matches)) + } + ForeignKeyAction.Restrict, ForeignKeyAction.NoAction -> throw DiskExecutorException( + "FOREIGN KEY constraint failed: ${matches.size} row(s) in ${child.schema.name} still " + + "reference $parentTableName (use ON UPDATE CASCADE)" + ) + } + } + } + } + return steps + } + + private fun applySteps(txn: Long, steps: List) { + for (step in steps) { + when (step) { + is CascadeStep.Delete -> { + for ((rid, _) in step.rows) step.table.delete(txn, rid) + applySteps(txn, step.next) + } + is CascadeStep.SetToNull -> { + val nulls = step.columnIndices.map { Value.NullValue } + for ((rid, row) in step.rows) step.table.update(txn, rid, withColumns(row, step.columnIndices, nulls)) + } + is CascadeStep.SetTo -> { + for ((rid, row) in step.rows) step.table.update(txn, rid, withColumns(row, step.columnIndices, step.values)) + } + } + } + } + + private fun requireNullable(schema: TableSchema, columnIndices: List, action: String) { + for (columnIndex in columnIndices) { + val column = schema.columns[columnIndex] + if (column.notNull) { + throw DiskExecutorException( + "FOREIGN KEY constraint failed: cannot $action on NOT NULL column ${schema.name}.${column.name}" + ) + } + } + } + + private fun withColumns(row: Row, columnIndices: List, values: List): Row { + val result = row.values.toMutableList() + columnIndices.forEachIndexed { position, columnIndex -> result[columnIndex] = values[position] } + return Row(result) + } +}