feat(engine): extend Executor with executeMerge and executeMergeInner
This commit is contained in:
dualfroz committed
1 parent
1542846ef3
commit
68f82b7815
1 file changed
+173
-1
@@ -1630,7 +1630,179 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
|
||||
table.rows.clear()
|
||||
table.rows.addAll(newRows)
|
||||
catalog.reindexTable(schema.name)
|
||||
return UpdateResult(updated)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, touchedRows)
|
||||
val returning = statement.returning?.let { evaluateReturning(schema, touchedRows.map { pair -> pair.second }, it) }
|
||||
return UpdateResult(updated, returning)
|
||||
}
|
||||
|
||||
private fun executeMerge(statement: MergeStatement): StatementResult =
|
||||
withStatementAtomicity(statement.targetTable, null) { executeMergeInner(statement) }
|
||||
|
||||
/**
|
||||
* The `ON` join is matched against [originalTargetRows] - a snapshot of the target table taken
|
||||
* once before any mutation - not the live, mutating `table.rows`: this is what makes "a target
|
||||
* row matched by two different source rows" a well-defined, detectable condition at all (see
|
||||
* below), and matches how a real MERGE's source/target join reads one consistent view of the
|
||||
* target rather than re-joining against its own in-flight writes. A row this MERGE inserts is
|
||||
* therefore never itself matched by a later source row (it never enters the snapshot), and a
|
||||
* row this MERGE updates/deletes keeps its ORIGINAL identity for join purposes even after the
|
||||
* live table's copy of it changes underneath - [applyMergeUpdate]/[applyMergeDelete] are handed
|
||||
* that same original [Row] object and use `===` against it to find their spot in the live table.
|
||||
*
|
||||
* For each source row, every target row (from the snapshot) satisfying [MergeStatement.onCondition]
|
||||
* is a MATCHED pair: the first [MergeMatchedUpdate]/[MergeMatchedDelete] clause (in
|
||||
* [MergeStatement.whenClauses] written order - MATCHED and NOT MATCHED clauses can be
|
||||
* interleaved, but a given pair only ever considers its own category) whose own `AND` condition
|
||||
* also holds runs; if none holds, that pair is left untouched, matching Postgres. A target row
|
||||
* already affected earlier in this same statement by a different matched pair throws rather than
|
||||
* silently applying a second action - the same "MERGE command cannot affect a row a second time"
|
||||
* guarantee Postgres gives, needed here because this engine has no snapshot isolation to make
|
||||
* that outcome well-defined otherwise. A source row matched by zero target rows is NOT MATCHED:
|
||||
* the first [MergeNotMatchedInsert] whose own condition holds (evaluated against the source row
|
||||
* alone - there is no target row to reference) inserts one new row; [checkColumns]-equivalent
|
||||
* validation for every clause happens up front via [checkMergeColumns], even when zero source
|
||||
* rows exist, matching this engine's usual eager-validation bar for every other schema-shaped
|
||||
* clause.
|
||||
*/
|
||||
private fun executeMergeInner(statement: MergeStatement): StatementResult {
|
||||
val table = catalog.table(statement.targetTable)
|
||||
val schema = table.schema
|
||||
val targetLabel = statement.targetAlias ?: statement.targetTable
|
||||
val (sourceSchema, sourceRows) = selectRunner.resolveMergeSource(statement.source)
|
||||
val sourceLabel = selectRunner.mergeSourceLabel(statement.source)
|
||||
if (targetLabel.equals(sourceLabel, ignoreCase = true)) {
|
||||
throw KestrelException("MERGE target and source cannot share the same name or alias: $targetLabel")
|
||||
}
|
||||
checkMergeColumns(statement, schema, targetLabel, sourceSchema, sourceLabel)
|
||||
|
||||
val originalTargetRows = table.rows.toList()
|
||||
val hasTriggers = catalog.hasTriggers(schema.name)
|
||||
val affectedTargets = mutableListOf<Row>()
|
||||
val returnedRows = mutableListOf<Pair<Row, String>>()
|
||||
var inserted = 0
|
||||
var updated = 0
|
||||
var deleted = 0
|
||||
|
||||
for (sourceRow in sourceRows) {
|
||||
val matchedTargets = originalTargetRows.filter { targetRow ->
|
||||
val resolver = mergeResolver(schema, targetLabel, targetRow, sourceSchema, sourceLabel, sourceRow)
|
||||
isTrue(Evaluator(resolver, subqueryRunner).evaluate(statement.onCondition))
|
||||
}
|
||||
if (matchedTargets.isNotEmpty()) {
|
||||
for (targetRow in matchedTargets) {
|
||||
val resolver = mergeResolver(schema, targetLabel, targetRow, sourceSchema, sourceLabel, sourceRow)
|
||||
val clause = statement.whenClauses.firstOrNull { candidate ->
|
||||
(candidate is MergeMatchedUpdate || candidate is MergeMatchedDelete) &&
|
||||
(candidate.condition == null || isTrue(Evaluator(resolver, subqueryRunner).evaluate(candidate.condition!!)))
|
||||
} ?: continue
|
||||
if (affectedTargets.any { it === targetRow }) {
|
||||
throw KestrelException("MERGE command cannot affect a target row a second time")
|
||||
}
|
||||
affectedTargets.add(targetRow)
|
||||
when (clause) {
|
||||
is MergeMatchedUpdate -> {
|
||||
val updatedRow = applyMergeUpdate(schema, table, targetRow, clause, resolver, hasTriggers)
|
||||
returnedRows.add(updatedRow to "UPDATE")
|
||||
updated++
|
||||
}
|
||||
is MergeMatchedDelete -> {
|
||||
val deletedRow = applyMergeDelete(schema, table, targetRow, hasTriggers)
|
||||
returnedRows.add(deletedRow to "DELETE")
|
||||
deleted++
|
||||
}
|
||||
else -> error("unreachable: filtered to MATCHED clauses above")
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
val resolver = notMatchedResolver(sourceSchema, sourceLabel, sourceRow, targetLabel)
|
||||
val clause = statement.whenClauses.firstOrNull { candidate ->
|
||||
candidate is MergeNotMatchedInsert &&
|
||||
(candidate.condition == null || isTrue(Evaluator(resolver, subqueryRunner).evaluate(candidate.condition!!)))
|
||||
} as? MergeNotMatchedInsert ?: continue
|
||||
val insertedRow = applyMergeInsert(schema, table, clause, resolver, hasTriggers)
|
||||
returnedRows.add(insertedRow to "INSERT")
|
||||
inserted++
|
||||
}
|
||||
if (!hasTriggers) catalog.reindexTable(schema.name)
|
||||
val returning = statement.returning?.let { evaluateMergeReturning(schema, returnedRows, it) }
|
||||
return MergeResult(inserted, updated, deleted, returning)
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluates a MERGE statement's RETURNING clause against every row an action of this MERGE
|
||||
* actually touched - the post-image for an inserted/updated row, the pre-deletion image for a
|
||||
* deleted one (same "nothing else to show" reasoning as [evaluateReturning]) - in the exact order
|
||||
* [executeMergeInner] applied those actions in (source-row order, and within a single source row's
|
||||
* multiple matched target rows, the order [executeMergeInner]'s own inner loop visits them).
|
||||
* [actionedRow]'s second element is `"INSERT"`/`"UPDATE"`/`"DELETE"`, bound to that row's
|
||||
* [Evaluator] as [MergeActionExpression]'s value so `MERGE_ACTION()` can appear in the projection -
|
||||
* the one context where it is actually legal. Column references resolve against [schema] (the
|
||||
* MERGE's target table) alone, same as [evaluateReturning] - RETURNING never sees the source row,
|
||||
* only what ended up true of the target.
|
||||
*/
|
||||
private fun evaluateMergeReturning(schema: TableSchema, actionedRows: List<Pair<Row, String>>, projection: Projection): SelectResult {
|
||||
val columns = when (projection) {
|
||||
is StarProjection -> schema.columns.map { it.name }
|
||||
is ExpressionProjection -> projection.items.mapIndexed { index, item ->
|
||||
checkColumns(item.expression, schema)
|
||||
returningColumnLabel(item, index)
|
||||
}
|
||||
}
|
||||
val resultRows = actionedRows.map { (row, action) ->
|
||||
when (projection) {
|
||||
is StarProjection -> row.values.toList()
|
||||
is ExpressionProjection -> {
|
||||
val evaluator = Evaluator(rowResolver(schema, row), subqueryRunner, mergeAction = action)
|
||||
projection.items.map { evaluator.evaluate(it.expression) }
|
||||
}
|
||||
}
|
||||
}
|
||||
return SelectResult(columns, resultRows)
|
||||
}
|
||||
|
||||
private fun applyMergeUpdate(
|
||||
schema: TableSchema,
|
||||
table: Table,
|
||||
targetRow: Row,
|
||||
clause: MergeMatchedUpdate,
|
||||
resolver: ColumnResolver,
|
||||
hasTriggers: Boolean
|
||||
): Row {
|
||||
val assignmentEvaluator = Evaluator(resolver, subqueryRunner, this)
|
||||
val values = targetRow.values.toMutableList()
|
||||
for (assignment in clause.assignments) {
|
||||
val index = schema.columnIndex(assignment.columnName)
|
||||
?: throw KestrelException("no such column: ${assignment.columnName}")
|
||||
val column = schema.columns[index]
|
||||
rejectGeneratedAssignment(column, explicitValueGiven = assignment.value != null)
|
||||
values[index] = if (assignment.value != null) {
|
||||
coerceForColumn(assignmentEvaluator.evaluate(assignment.value), column)
|
||||
} else {
|
||||
column.default?.let { coerceForColumn(assignmentEvaluator.evaluate(it), column) } ?: Value.NullValue
|
||||
}
|
||||
}
|
||||
applyGeneratedColumns(schema, values)
|
||||
validateNotNull(schema, values)
|
||||
val updatedRow = Row(values)
|
||||
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, listOf(targetRow to updatedRow))
|
||||
ForeignKeyEnforcer.validateRow(catalog, schema, updatedRow.values, uncommittedSiblings = emptyList())
|
||||
enforceChecks(schema, updatedRow)
|
||||
val others = table.rows.filter { it !== targetRow }
|
||||
enforcePrimaryKeyUniqueness(schema, others + updatedRow)
|
||||
enforceUniqueConstraints(schema, others + updatedRow)
|
||||
enforceUniqueIndexes(schema, others + updatedRow)
|
||||
for (group in schema.enforcedUniqueKeyGroups()) {
|
||||
val columnIndices = group.map { schema.columnIndex(it)!! }
|
||||
val oldKey = columnIndices.map { targetRow.values[it] }
|
||||
val newKey = columnIndices.map { updatedRow.values[it] }
|
||||
ForeignKeyEnforcer.cascadeUpdate(catalog, schema.name, group, listOf(oldKey to newKey))
|
||||
}
|
||||
val position = table.rows.indexOfFirst { it === targetRow }
|
||||
table.rows[position] = updatedRow
|
||||
if (hasTriggers) catalog.reindexTable(schema.name)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, listOf(targetRow to updatedRow))
|
||||
return updatedRow
|
||||
}
|
||||
|
||||
private fun coerceForColumn(value: Value, column: Column): Value {
|
||||
|
||||
Reference in new issue
Block a user