feat(sql): add runTriggerBody and buildAssignments to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
dff226d846
commit
daf6ce803f
1 file changed
+157
@@ -642,3 +642,160 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Runs a fired trigger's body statements in order, directly against [txn] - NOT through
|
||||||
|
* [executeInsert]/[executeUpdate]/[executeDelete] (which would re-select a transaction via
|
||||||
|
* [withStatementTransaction] and re-take a trigger-atomicity savepoint via
|
||||||
|
* [withTriggerAtomicity]) - so the whole cascade shares exactly one disk transaction and exactly
|
||||||
|
* one savepoint, taken once by the outermost caller. Mirrors [kestrel.engine.Executor.
|
||||||
|
* runTriggerBody], including [MAX_TRIGGER_DEPTH].
|
||||||
|
*/
|
||||||
|
private fun runTriggerBody(trigger: TriggerDefinition, context: TriggerRowContext, txn: Long) {
|
||||||
|
if (triggerDepth >= MAX_TRIGGER_DEPTH) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"trigger recursion limit ($MAX_TRIGGER_DEPTH) exceeded while firing '${trigger.name}' - " +
|
||||||
|
"check for a trigger whose body ends up re-triggering itself"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
triggerDepth++
|
||||||
|
try {
|
||||||
|
for (bodyStatement in trigger.body) {
|
||||||
|
when (bodyStatement) {
|
||||||
|
is InsertStatement -> {
|
||||||
|
val table = catalog.table(bodyStatement.tableName)
|
||||||
|
executeInsertInTxn(bodyStatement, table, table.schema, txn, context)
|
||||||
|
}
|
||||||
|
is UpdateStatement -> {
|
||||||
|
val table = catalog.table(bodyStatement.tableName)
|
||||||
|
executeUpdateInTxn(bodyStatement, buildAssignments(table.schema, bodyStatement), table, table.schema, txn, context)
|
||||||
|
}
|
||||||
|
is DeleteStatement -> {
|
||||||
|
val table = catalog.table(bodyStatement.tableName)
|
||||||
|
executeDeleteInTxn(bodyStatement, table, table.schema, txn, context)
|
||||||
|
}
|
||||||
|
else -> error("unreachable: trigger bodies only ever contain INSERT/UPDATE/DELETE")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
triggerDepth--
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun buildAssignments(schema: TableSchema, statement: UpdateStatement): List<Pair<Int, Expression?>> =
|
||||||
|
statement.assignments.map { assignment ->
|
||||||
|
val index = schema.columnIndex(assignment.columnName)
|
||||||
|
?: throw DiskExecutorException("no such column: ${assignment.columnName} in table ${schema.name}")
|
||||||
|
rejectGeneratedAssignment(schema.columns[index], explicitValueGiven = assignment.value != null)
|
||||||
|
index to assignment.value
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun executeCreateIndex(statement: CreateIndexStatement): StatementResult {
|
||||||
|
catalog.createIndex(statement.tableName, statement.indexName, statement.columnNames, statement.unique)
|
||||||
|
return CreateIndexResult(statement.indexName)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun executeDropIndex(statement: DropIndexStatement): StatementResult {
|
||||||
|
catalog.dropIndex(statement.indexName)
|
||||||
|
return DropIndexResult(statement.indexName)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Ported 1:1 from [kestrel.engine.Executor.executeCreateSequence] - see that method's own doc
|
||||||
|
* for the ascending/descending default resolution and the one thing [SequenceDefinition]'s own
|
||||||
|
* `init` cannot check itself (START WITH falling inside [MINVALUE, MAXVALUE]).
|
||||||
|
*/
|
||||||
|
private fun executeCreateSequence(statement: CreateSequenceStatement): StatementResult {
|
||||||
|
val incrementBy = statement.incrementBy ?: 1L
|
||||||
|
if (incrementBy == 0L) throw DiskExecutorException("sequence '${statement.sequenceName}' INCREMENT BY cannot be zero")
|
||||||
|
val ascending = incrementBy > 0
|
||||||
|
val minValue = statement.minValue ?: if (ascending) 1L else Long.MIN_VALUE
|
||||||
|
val maxValue = statement.maxValue ?: if (ascending) Long.MAX_VALUE else -1L
|
||||||
|
if (minValue > maxValue) {
|
||||||
|
throw DiskExecutorException("sequence '${statement.sequenceName}' MINVALUE ($minValue) must not exceed MAXVALUE ($maxValue)")
|
||||||
|
}
|
||||||
|
val startValue = statement.startValue ?: if (ascending) minValue else maxValue
|
||||||
|
if (startValue < minValue || startValue > maxValue) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"START WITH ($startValue) must be between MINVALUE ($minValue) and MAXVALUE ($maxValue)"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
catalog.createSequence(SequenceDefinition(statement.sequenceName, incrementBy, minValue, maxValue, startValue, statement.cycle))
|
||||||
|
return CreateSequenceResult(statement.sequenceName)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun executeDropSequence(statement: DropSequenceStatement): StatementResult {
|
||||||
|
catalog.dropSequence(statement.sequenceName)
|
||||||
|
return DropSequenceResult(statement.sequenceName)
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Ported 1:1 from [kestrel.engine.Executor.executeAlterSequence] - see that method's own doc. */
|
||||||
|
private fun executeAlterSequence(statement: AlterSequenceStatement): StatementResult {
|
||||||
|
val existing = catalog.sequence(statement.sequenceName).definition
|
||||||
|
val incrementBy = statement.incrementBy ?: existing.incrementBy
|
||||||
|
if (incrementBy == 0L) throw DiskExecutorException("sequence '${statement.sequenceName}' INCREMENT BY cannot be zero")
|
||||||
|
val ascending = incrementBy > 0
|
||||||
|
val minValue = when {
|
||||||
|
statement.clearMinValue -> if (ascending) 1L else Long.MIN_VALUE
|
||||||
|
statement.minValue != null -> statement.minValue
|
||||||
|
else -> existing.minValue
|
||||||
|
}
|
||||||
|
val maxValue = when {
|
||||||
|
statement.clearMaxValue -> if (ascending) Long.MAX_VALUE else -1L
|
||||||
|
statement.maxValue != null -> statement.maxValue
|
||||||
|
else -> existing.maxValue
|
||||||
|
}
|
||||||
|
if (minValue > maxValue) {
|
||||||
|
throw DiskExecutorException("sequence '${statement.sequenceName}' MINVALUE ($minValue) must not exceed MAXVALUE ($maxValue)")
|
||||||
|
}
|
||||||
|
val cycle = statement.cycle ?: existing.cycle
|
||||||
|
val newDefinition = SequenceDefinition(statement.sequenceName, incrementBy, minValue, maxValue, existing.startValue, cycle)
|
||||||
|
val restartAt = if (statement.restart) (statement.restartWith ?: existing.startValue) else null
|
||||||
|
if (restartAt != null && (restartAt < minValue || restartAt > maxValue)) {
|
||||||
|
throw DiskExecutorException("RESTART value ($restartAt) must be between MINVALUE ($minValue) and MAXVALUE ($maxValue)")
|
||||||
|
}
|
||||||
|
catalog.alterSequence(statement.sequenceName, newDefinition, restartAt)
|
||||||
|
return AlterSequenceResult(statement.sequenceName)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Runs [CreateViewStatement.query] once, discarding the result, purely to fail fast on a bad
|
||||||
|
* column/table reference before anything is persisted - mirrors [kestrel.engine.Executor.
|
||||||
|
* executeCreateView] calling `selectRunner.explain` for the same reason, just via a real
|
||||||
|
* execution rather than a plan-only walk (this executor has no `explain`).
|
||||||
|
*/
|
||||||
|
private fun executeCreateView(statement: CreateViewStatement): StatementResult {
|
||||||
|
val txn = catalog.beginTransaction()
|
||||||
|
runInTransaction(txn) { runQuery(statement.query, txn, outer = null) }
|
||||||
|
catalog.createView(statement.viewName, statement.query, statement.orReplace)
|
||||||
|
return CreateViewResult(statement.viewName)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun executeDropView(statement: DropViewStatement): StatementResult {
|
||||||
|
catalog.dropView(statement.viewName, cascade = statement.mode == DropMode.Cascade)
|
||||||
|
return DropViewResult(statement.viewName)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Ported 1:1 from [kestrel.engine.Executor.executeCreateTrigger]: body statements are
|
||||||
|
* deliberately NOT deep-validated here (no eager check that every column referenced in a body
|
||||||
|
* statement's WHERE/SET/VALUES actually exists) - a trigger body can target a different table per
|
||||||
|
* statement, and this executor already defers exactly this kind of check to the moment the
|
||||||
|
* trigger actually fires, same as SQLite. What IS checked eagerly: the trigger's own table and
|
||||||
|
* every body statement's target table both exist right now, and the WHEN clause - evaluated
|
||||||
|
* directly by this executor rather than through a nested statement run, so nothing else will ever
|
||||||
|
* validate it - contains no subquery/aggregate and only ever references NEW/OLD columns that both
|
||||||
|
* exist on this table and are actually available for this trigger's event.
|
||||||
|
*/
|
||||||
|
private fun executeCreateTrigger(statement: CreateTriggerStatement): StatementResult {
|
||||||
|
val schema = catalog.table(statement.tableName).schema
|
||||||
|
statement.whenClause?.let { whenClause ->
|
||||||
|
requireNoSubquery(whenClause)
|
||||||
|
validateTriggerRowReferences(whenClause, statement.event, schema)
|
||||||
|
}
|
||||||
|
for (bodyStatement in statement.body) validateTriggerBodyStatementTarget(bodyStatement)
|
||||||
|
catalog.createTrigger(
|
||||||
|
TriggerDefinition(statement.triggerName, statement.timing, statement.event, statement.tableName, statement.whenClause, statement.body)
|
||||||
|
)
|
||||||
|
return CreateTriggerResult(statement.triggerName)
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user