feat(persistence): extend SqlWriter with onConflictText

This commit is contained in:
dualfroz committed 2026-09-02 19:07:41 +02:00
1 parent 6ff6e4e91e
commit 364dd2e564
1 file changed
+172 -17
+172 -17
View File
@@ -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<Column>,
foreignKeys: List<ForeignKeyConstraint>,
checks: List<CheckConstraint>,
uniqueConstraints: List<UniqueConstraint>,
compositePrimaryKey: List<String> = 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)})"