From 364dd2e56460592ba494e83e50b4fc336988c5a7 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 2 Sep 2026 19:07:41 +0200 Subject: [PATCH] feat(persistence): extend SqlWriter with onConflictText --- .../kotlin/kestrel/persistence/SqlWriter.kt | 189 ++++++++++++++++-- 1 file changed, 172 insertions(+), 17 deletions(-) diff --git a/src/main/kotlin/kestrel/persistence/SqlWriter.kt b/src/main/kotlin/kestrel/persistence/SqlWriter.kt index 1d11e98..49867e1 100644 --- a/src/main/kotlin/kestrel/persistence/SqlWriter.kt +++ b/src/main/kotlin/kestrel/persistence/SqlWriter.kt @@ -234,19 +234,77 @@ object SqlWriter { fun dump(catalog: Catalog): String { val builder = StringBuilder() + // Sequences are dumped FIRST, strictly before every table - a table's own column DEFAULT + // can read `NEXTVAL('seq')` (see kestrel.engine.Executor.executeCreateTable's eager + // `catalog.sequence(...)` existence check), so replaying that CREATE TABLE on reload would + // fail with "no such sequence" if the sequence it names did not already exist yet. + for (definition in catalog.sequenceDefinitions()) { + builder.append("${sequenceDumpText(catalog.sequence(definition.name))}\n") + } for (tableName in catalog.tableNames) { val table = catalog.table(tableName) + // A TEMP/TEMPORARY table (see kestrel.catalog.TableSchema.temporary) is session-scoped by + // definition - dumping it (or its rows below) into a checkpoint would make it survive a + // reload, which is exactly what "temporary" promises it never does. + if (table.schema.temporary) continue builder.append("CREATE TABLE ${table.schema.name} (") - builder.append(table.schema.columns.joinToString(", ") { columnDefinition(it) }) + builder.append( + createTableBodyText( + table.schema.columns, + table.schema.foreignKeys, + table.schema.checks, + table.schema.uniqueConstraints, + table.schema.compositePrimaryKey + ) + ) builder.append(");\n") for (row in table.rows) { builder.append("INSERT INTO ${table.schema.name} VALUES (") - builder.append(row.values.joinToString(", ") { literal(it) }) + // A GENERATED column's position is dumped as the literal `DEFAULT` keyword rather + // than its already-computed value - the executor rejects an explicit value for a + // GENERATED column outright (see Executor.rejectGeneratedAssignment), and `DEFAULT` + // here triggers the exact same fresh-recompute path an omitted column would, which + // reproduces the same value on replay since the expression is a pure function of + // this row's other (literal, already-dumped) columns. + builder.append( + row.values.mapIndexed { index, value -> + if (table.schema.columns[index].generated != null) "DEFAULT" else literal(value) + }.joinToString(", ") + ) builder.append(");\n") } } for (definition in catalog.indexDefinitions()) { - builder.append("CREATE INDEX ${definition.name} ON ${definition.tableName} (${definition.columnName});\n") + // An index defined on a TEMP table never gets a CREATE TABLE to attach to above, so + // dumping it would replay as "no such table" - skip it right along with its table. + if (catalog.tableOrNull(definition.tableName)?.schema?.temporary == true) continue + builder.append( + "CREATE ${if (definition.unique) "UNIQUE " else ""}INDEX ${definition.name} ON " + + "${definition.tableName} (${definition.columnDisplayName});\n" + ) + } + // A VIEW whose query reads a TEMP table is NOT specially excluded here (unlike the index/ + // trigger cases above, its dependency can be buried arbitrarily deep in a join/subquery, not + // one flat tableName field) - dumping it and replaying it against a reload that never + // recreated that TEMP table fails loudly with "no such table", the same honest failure mode + // this engine already accepts for e.g. a prepared statement that does not survive reopen, + // rather than silently materializing an empty or wrong view. + for (viewName in catalog.viewNames) { + val view = catalog.viewOrNull(viewName)!! + builder.append("CREATE VIEW ${view.name} AS ${queryText(view.query)};\n") + } + // Triggers are dumped LAST, strictly after every row above has already been inserted as + // plain data - a trigger created before its table's INSERTs would re-fire (duplicating an + // AFTER trigger's side effects, or re-running a BEFORE trigger's validation) the moment + // this very dump is replayed on reload, since replay re-executes each of these statements + // for real. Creating the trigger only once the table is already fully populated means + // those historical rows never pass through it, while every INSERT/UPDATE/DELETE a user + // issues after reload still does. + for (definition in catalog.triggerDefinitions()) { + // Same reasoning as the index skip above: a trigger ON a TEMP table has no table left to + // attach to after reload. + if (catalog.tableOrNull(definition.tableName)?.schema?.temporary == true) continue + builder.append("${triggerText(definition)}\n") } return builder.toString() } @@ -254,9 +312,23 @@ object SqlWriter { private fun writeInsert(statement: InsertStatement): String { val columns = statement.columns?.let { " (${it.joinToString(", ")})" } ?: "" val rows = statement.rows.joinToString(", ") { row -> - "(" + row.joinToString(", ") { expression(it) } + ")" + "(" + row.joinToString(", ") { it?.let { expression(it) } ?: "DEFAULT" } + ")" } - return "INSERT INTO ${statement.tableName}$columns VALUES $rows;" + val onConflict = statement.onConflict?.let { " ${onConflictText(it)}" } ?: "" + return "INSERT INTO ${statement.tableName}$columns VALUES $rows$onConflict;" + } + + private fun onConflictText(clause: OnConflictClause): String { + val target = if (clause.targetColumns.isEmpty()) "" else " (${clause.targetColumns.joinToString(", ")})" + val action = when (val action = clause.action) { + ConflictDoNothing -> "DO NOTHING" + is ConflictDoUpdate -> { + val assignments = action.assignments.joinToString(", ") { assignment(it) } + val where = action.where?.let { " WHERE ${expression(it)}" } ?: "" + "DO UPDATE SET $assignments$where" + } + } + return "ON CONFLICT$target $action" } private fun writeUpdate(statement: UpdateStatement): String { @@ -265,28 +337,111 @@ object SqlWriter { return "UPDATE ${statement.tableName} SET $assignments$where;" } + private fun writeTruncate(statement: TruncateTableStatement): String { + val identity = if (statement.restartIdentity) " RESTART IDENTITY" else "" + return "TRUNCATE TABLE ${statement.tableNames.joinToString(", ")}$identity${dropModeText(statement.mode)};" + } + private fun writeDelete(statement: DeleteStatement): String { val where = statement.where?.let { " WHERE ${expression(it)}" } ?: "" return "DELETE FROM ${statement.tableName}$where;" } - private fun assignment(assignment: Assignment): String = - "${assignment.columnName} = ${expression(assignment.value)}" - - private fun columnDefinition(column: Column): String { - val constraint = when { - column.primaryKey -> " PRIMARY KEY" - column.notNull -> " NOT NULL" - else -> "" - } - return "${column.name} ${column.type.displayName}$constraint" + /** + * Persisted literally like every other DML statement (see [writeInsert]/[writeUpdate]): by the + * time this redo-log entry replays, every earlier statement - including whatever populated a + * table-shaped `USING` source - has already replayed too, so re-running the exact same MERGE + * text against the freshly rebuilt catalog reproduces the same result deterministically. + */ + private fun writeMerge(statement: MergeStatement): String { + val targetAlias = statement.targetAlias?.let { " AS $it" } ?: "" + val whenClauses = statement.whenClauses.joinToString(" ") { mergeWhenClauseText(it) } + return "MERGE INTO ${statement.targetTable}$targetAlias USING ${fromItemText(statement.source)} " + + "ON ${expression(statement.onCondition)} $whenClauses;" } + private fun mergeWhenClauseText(clause: MergeWhenClause): String { + val condition = clause.condition?.let { " AND ${expression(it)}" } ?: "" + return when (clause) { + is MergeMatchedUpdate -> + "WHEN MATCHED$condition THEN UPDATE SET ${clause.assignments.joinToString(", ") { assignment(it) }}" + is MergeMatchedDelete -> "WHEN MATCHED$condition THEN DELETE" + is MergeNotMatchedInsert -> { + val columns = clause.columns?.let { " (${it.joinToString(", ")})" } ?: "" + val values = clause.values.joinToString(", ") { it?.let { value -> expression(value) } ?: "DEFAULT" } + "WHEN NOT MATCHED$condition THEN INSERT$columns VALUES ($values)" + } + } + } + + private fun assignment(assignment: Assignment): String = + "${assignment.columnName} = ${assignment.value?.let { expression(it) } ?: "DEFAULT"}" + + private fun columnDefinition(column: Column): String = buildString { + append(column.name) + append(" ") + append(column.type.displayWith(column.precision, column.scale)) + when { + column.primaryKey -> append(" PRIMARY KEY") + column.notNull -> append(" NOT NULL") + } + if (column.collation != Collation.Binary) append(" COLLATE ${column.collation.sqlName}") + column.default?.let { append(" DEFAULT ${expression(it)}") } + column.generated?.let { append(" GENERATED ALWAYS AS (${expression(it)}) STORED") } + } + + private fun alterTableText(statement: AlterTableStatement): String { + val action = when (val action = statement.action) { + is AddColumn -> "ADD COLUMN ${columnDefinition(action.column)}" + is DropColumn -> "DROP COLUMN ${action.columnName}${dropModeText(action.mode)}" + is RenameColumn -> "RENAME COLUMN ${action.oldName} TO ${action.newName}" + is RenameTable -> "RENAME TO ${action.newName}" + is AddConstraint -> "ADD ${addedConstraintText(action.constraint)}" + is DropConstraint -> "DROP CONSTRAINT ${action.name}${dropModeText(action.mode)}" + is ChangeColumnType -> { + val using = action.using?.let { " USING ${expression(it)}" } ?: "" + "ALTER COLUMN ${action.columnName} TYPE ${action.newType.displayWith(action.precision, action.scale)}$using" + } + } + return "ALTER TABLE ${statement.tableName} $action;" + } + + private fun addedConstraintText(constraint: AddedConstraint): String = when (constraint) { + is AddedCheck -> constraintText(constraint.check.name, "CHECK (${expression(constraint.check.expression)})") + is AddedUnique -> constraintText(constraint.unique.name, "UNIQUE (${constraint.unique.columns.joinToString(", ")})") + is AddedForeignKey -> constraintText(constraint.foreignKey.name, foreignKeyText(constraint.foreignKey)) + } + + private fun constraintText(name: String?, body: String): String = + if (name != null) "CONSTRAINT $name $body" else body + + private fun createTableBodyText( + columns: List, + foreignKeys: List, + checks: List, + uniqueConstraints: List, + compositePrimaryKey: List = emptyList() + ): String { + val parts = columns.map { columnDefinition(it) } + + (if (compositePrimaryKey.isNotEmpty()) listOf("PRIMARY KEY (${compositePrimaryKey.joinToString(", ")})") else emptyList()) + + foreignKeys.map { constraintText(it.name, foreignKeyText(it)) } + + checks.map { constraintText(it.name, "CHECK (${expression(it.expression)})") } + + uniqueConstraints.map { constraintText(it.name, "UNIQUE (${it.columns.joinToString(", ")})") } + return parts.joinToString(", ") + } + + private fun foreignKeyText(fk: ForeignKeyConstraint): String = buildString { + append("FOREIGN KEY (${fk.columns.joinToString(", ")}) REFERENCES ${fk.referencedTable} (${fk.referencedColumns.joinToString(", ")})") + if (fk.onDelete != ForeignKeyAction.NoAction) append(" ON DELETE ${fk.onDelete.sqlKeyword}") + if (fk.onUpdate != ForeignKeyAction.NoAction) append(" ON UPDATE ${fk.onUpdate.sqlKeyword}") + } + + private fun dropModeText(mode: DropMode): String = if (mode == DropMode.Cascade) " CASCADE" else "" + private fun expression(expression: Expression): String = when (expression) { is LiteralExpression -> literal(expression.value) is ColumnExpression -> expression.qualifier?.let { "$it.${expression.name}" } ?: expression.name - is FunctionExpression -> - "${expression.name}(${if (expression.star) "*" else expression.argument?.let { expression(it) } ?: ""})" + is FunctionExpression -> "${expression.name}(${functionArgumentsText(expression)})${filterClauseText(expression)}" is BinaryExpression -> "(${expression(expression.left)} ${expression.operator.symbol} ${expression(expression.right)})" is UnaryExpression -> "(${expression.operator.symbol} ${expression(expression.operand)})"