feat(sql): add executeInsertWithConflictHandling
This commit is contained in:
dualfroz committed
1 parent
ab24de18a4
commit
0eff2e8e8b
1 file changed
+162
@@ -1917,3 +1917,165 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Ported 1:1 from [kestrel.engine.Executor.executeInsertWithConflictHandling]: each candidate
|
||||
* row is processed as its own micro-insert against the table's current state (not a
|
||||
* batch-validated snapshot), so a later row in the same multi-row INSERT correctly sees an
|
||||
* earlier row's insert or update. [triggerContext] can be non-null here for two independent
|
||||
* reasons: [schema] itself may have triggers of its own (fired per row below, exactly like
|
||||
* [kestrel.engine.Executor] does), and/or this INSERT ... ON CONFLICT may itself be a trigger
|
||||
* body statement targeting a *different* table, so `NEW`/`OLD` from the outer firing trigger
|
||||
* must still resolve in its DO UPDATE assignment values - both compose without conflict since
|
||||
* [fireRowTriggers] below builds its own fresh [TriggerRowContext] from [schema]'s own triggers,
|
||||
* independent of [triggerContext]. Unlike the batch INSERT/UPDATE paths, no separate reindex step
|
||||
* is needed around trigger firing here: [DiskTable]'s secondary indexes are B+Trees maintained
|
||||
* incrementally by [table.insert]/[table.update] themselves, not rebuilt in a batch pass like
|
||||
* [kestrel.catalog.Catalog.reindexTable], so a trigger body's own query against this table always
|
||||
* sees a consistent index the instant this row's own mutation returns.
|
||||
*/
|
||||
private fun executeInsertWithConflictHandling(
|
||||
schema: TableSchema,
|
||||
table: DiskTable,
|
||||
txn: Long,
|
||||
candidateRows: List<Row>,
|
||||
onConflict: OnConflictClause,
|
||||
subqueries: SubqueryRunner,
|
||||
returning: Projection? = null,
|
||||
triggerContext: TriggerRowContext? = null
|
||||
): StatementResult {
|
||||
val keys = diskUniqueKeysOf(schema, table)
|
||||
val targetKeys = resolveConflictTargetKeys(schema, onConflict.targetColumns, keys)
|
||||
var affected = 0
|
||||
// DO NOTHING contributes no row, and a DO UPDATE row whose WHERE guard fails `continue`s
|
||||
// before reaching the append below - both absent from RETURNING, matching Postgres, same
|
||||
// as the in-memory engine.
|
||||
val returnedRows = mutableListOf<Row>()
|
||||
for (row in candidateRows) {
|
||||
val conflict = findConflictingRow(txn, table, row.values, targetKeys)
|
||||
if (conflict == null) {
|
||||
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Insert, listOf(null to row), txn)
|
||||
DiskForeignKeyEnforcer.validateRow(catalog, txn, schema, row.values, uncommittedSiblings = emptyList())
|
||||
enforceChecks(schema, row, subqueries)
|
||||
table.insert(txn, row)
|
||||
affected++
|
||||
returnedRows.add(row)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Insert, listOf(null to row), txn)
|
||||
continue
|
||||
}
|
||||
val (conflictingRid, conflictingRow) = conflict
|
||||
when (val action = onConflict.action) {
|
||||
ConflictDoNothing -> Unit
|
||||
is ConflictDoUpdate -> {
|
||||
val resolver = composeResolver(excludedAwareResolver(schema, conflictingRow, row), triggerContext)
|
||||
// NEXTVAL/CURRVAL are deliberately usable in the assignment values below but not
|
||||
// in this WHERE-shaped guard, matching the in-memory engine's own two-Evaluator split.
|
||||
val conditionEvaluator = Evaluator(resolver, subqueries)
|
||||
if (action.where != null && !isTrue(conditionEvaluator.evaluate(action.where))) continue
|
||||
val assignmentEvaluator = Evaluator(resolver, subqueries, this)
|
||||
val values = conflictingRow.values.toMutableList()
|
||||
for (assignment in action.assignments) {
|
||||
val index = schema.columnIndex(assignment.columnName)
|
||||
?: throw DiskExecutorException("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, subqueries)
|
||||
val updated = Row(values)
|
||||
// Fires before any constraint below is checked, same as executeUpdateInTxn's own
|
||||
// BEFORE trigger - a BEFORE UPDATE trigger's body can reject this row's update
|
||||
// before this executor's own validation ever runs.
|
||||
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, listOf(conflictingRow to updated), txn)
|
||||
DiskForeignKeyEnforcer.validateRow(catalog, txn, schema, updated.values, uncommittedSiblings = emptyList())
|
||||
enforceChecks(schema, updated, subqueries)
|
||||
// Same as executeUpdate: any table with a FK pointing at one of this table's
|
||||
// enforced-unique key groups needs a cascade/RESTRICT pass when that group's
|
||||
// value actually changes, not just the PRIMARY KEY.
|
||||
for (group in schema.enforcedUniqueKeyGroups()) {
|
||||
val columnIndices = group.map { schema.columnIndex(it)!! }
|
||||
val oldKey = columnIndices.map { conflictingRow.values[it] }
|
||||
val newKey = columnIndices.map { updated.values[it] }
|
||||
DiskForeignKeyEnforcer.cascadeUpdate(catalog, txn, schema.name, group, listOf(oldKey to newKey))
|
||||
}
|
||||
table.update(txn, conflictingRid, updated)
|
||||
affected++
|
||||
returnedRows.add(updated)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, listOf(conflictingRow to updated), txn)
|
||||
}
|
||||
}
|
||||
}
|
||||
return InsertResult(affected, returning?.let { evaluateReturning(schema, returnedRows, it, subqueries) })
|
||||
}
|
||||
|
||||
/**
|
||||
* Ported 1:1 from [kestrel.engine.Executor.excludedAwareResolver]: bare column names resolve
|
||||
* against [target] (the existing conflicting row, matching plain UPDATE semantics), while
|
||||
* `EXCLUDED.column` resolves against [excluded] (the row proposed by this INSERT that failed
|
||||
* to land) - the same pseudo-table Postgres exposes in DO UPDATE.
|
||||
*/
|
||||
private fun excludedAwareResolver(schema: TableSchema, target: Row, excluded: Row): ColumnResolver = object : ColumnResolver {
|
||||
override fun resolve(expression: ColumnExpression): Value {
|
||||
val qualifier = expression.qualifier
|
||||
val useExcluded = qualifier != null && qualifier.equals("excluded", ignoreCase = true)
|
||||
if (qualifier != null && !useExcluded && !qualifier.equals(schema.name, ignoreCase = true)) {
|
||||
throw DiskExecutorException("no such table or alias: $qualifier")
|
||||
}
|
||||
val index = schema.columnIndex(expression.name)
|
||||
?: throw DiskExecutorException("no such column: ${expression.name}")
|
||||
return if (useExcluded) excluded.values[index] else target.values[index]
|
||||
}
|
||||
|
||||
override fun declaredCollation(expression: ColumnExpression): Collation? = schema.column(expression.name)?.collation
|
||||
}
|
||||
|
||||
private fun resolveInsertTargets(schema: TableSchema, columns: List<String>?): List<Int> {
|
||||
if (columns == null) return schema.columns.indices.toList()
|
||||
return columns.map { columnName ->
|
||||
schema.columnIndex(columnName)
|
||||
?: throw DiskExecutorException("no such column: $columnName in table ${schema.name}")
|
||||
}
|
||||
}
|
||||
|
||||
private fun executeQuery(statement: QueryStatement): StatementResult =
|
||||
withStatementTransaction { txn -> runQuery(statement, txn, outer = null) }
|
||||
|
||||
/**
|
||||
* Closes the "this executor has no `explain`" scope cut documented on [executeCreateView]:
|
||||
* describes the plan [runSelect]/[runQuery] would use for [statement.select] without running
|
||||
* it, mirroring [kestrel.engine.SelectRunner.explain]'s shape (Seq Scan/Index Scan header per
|
||||
* driving source, Nested Loop per JOIN, GroupAggregate/Aggregate, Sort, Limit, CTE/UNION/WITH
|
||||
* RECURSIVE nesting) with two differences forced by this executor's disk-backed row supply
|
||||
* rather than the in-memory engine's already-resident rows:
|
||||
*
|
||||
* - A materialized CTE, a VIEW, or a non-LATERAL derived table in FROM still runs its defining
|
||||
* query for real via [resolveFromItem]/[buildCteScope] (exactly what [runSelect] itself would
|
||||
* do, and exactly what [kestrel.engine.SelectRunner.explain] does too through `resolveSource`)
|
||||
* since only running it produces a real schema/row set to plan further nesting against. The
|
||||
* query's own driving table(s) are never scanned this way, though: [resolveFromItem] only
|
||||
* wraps a real [DiskTable] in a [DiskTableSource] (schema/index metadata, no I/O) - the
|
||||
* [DiskTableSource.scan] a real [runSelect] would call to actually read rows is deliberately
|
||||
* never invoked here, so `EXPLAIN SELECT * FROM huge_table` costs nothing proportional to
|
||||
* `huge_table`'s size the way `SELECT * FROM huge_table` would.
|
||||
* - [applyJoin] has no hash-join optimization at all (see its own doc), so "Nested Loop" is the
|
||||
* only join header this ever prints - "Hash Join" (which the in-memory engine's EXPLAIN can
|
||||
* print for an equality ON condition) never appears here.
|
||||
*
|
||||
* Deliberately narrower than [kestrel.engine.SelectRunner.explain] in one respect: it reuses
|
||||
* every validation [runSelect] performs *structurally*, independent of row data - duplicate
|
||||
* alias detection ([buildJoinSources]), window-function placement ([rejectWindowFunction]/
|
||||
* [validateWindows]), and GROUP BY/aggregate correctness ([validateAggregation], itself
|
||||
* structural since [resolveIndex] only consults [JoinSource.schema], never row data) - but does
|
||||
* NOT eagerly resolve WHERE/ON column references the way [kestrel.engine.SelectRunner.explain]'s
|
||||
* `Plan.validate()` does. [runSelect] itself only resolves those against real row data inside
|
||||
* `combined.filter { ... }`/[applyJoin], so on an empty table a bad WHERE/ON column reference
|
||||
* never surfaces there either (a pre-existing, documented gap versus the in-memory engine, not
|
||||
* a new one introduced here) - EXPLAIN staying exactly as permissive keeps its guarantees no
|
||||
* stronger than what actually running the statement would give you.
|
||||
*/
|
||||
private fun executeExplain(statement: ExplainStatement): StatementResult =
|
||||
withStatementTransaction { txn -> ExplainResult(explainQuery(statement.select, txn)) }
|
||||
|
||||
Reference in new issue
Block a user