From f617dcc77e37c27b275a121f7d47fe1b9acde174 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 17:51:01 +0200 Subject: [PATCH] feat(sql): add applyMergeInsert and resolveMergeSource to DiskExecutor --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 154 ++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index f187f4e..51d0133 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -4148,3 +4148,157 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { return targetRow } + private fun applyMergeInsert( + schema: TableSchema, + table: DiskTable, + txn: Long, + clause: MergeNotMatchedInsert, + resolver: ColumnResolver, + subqueries: SubqueryRunner + ): Row { + val targetIndices = resolveInsertTargets(schema, clause.columns) + val evaluator = Evaluator(resolver, subqueries, this) + val values = MutableList(schema.columns.size) { Value.NullValue } + val touched = BooleanArray(schema.columns.size) + for ((position, expression) in clause.values.withIndex()) { + val columnIndex = targetIndices[position] + if (expression != null) { + rejectGeneratedAssignment(schema.columns[columnIndex], explicitValueGiven = true) + values[columnIndex] = coerceForColumn(evaluator.evaluate(expression), schema.columns[columnIndex]) + touched[columnIndex] = true + } + } + for ((index, column) in schema.columns.withIndex()) { + val default = column.default + if (!touched[index] && default != null) { + values[index] = coerceForColumn(evaluator.evaluate(default), column) + } + } + applyGeneratedColumns(schema, values, subqueries) + val row = Row(values) + 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) + fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Insert, listOf(null to row), txn) + return row + } + + /** A snapshot (schema, rows) for a MERGE `USING` source - ported 1:1 from [kestrel.engine.SelectRunner.resolveMergeSource], just via [resolveFromItem] rather than a separate in-memory `resolveSource`. */ + private fun resolveMergeSource(item: FromItem, txn: Long): Pair> { + val source = resolveFromItem(item, txn) + return source.schema to source.scan(txn).map { Row(it) } + } + + /** + * Resolves a MERGE `ON`/`WHEN MATCHED` column reference against whichever of [targetRow]/ + * [sourceRow] its qualifier names ([targetLabel]/[sourceLabel] - alias if given, bare name + * otherwise); an unqualified reference must exist on exactly one side. Ported 1:1 from + * [kestrel.engine.Executor.mergeResolver]. + */ + private fun mergeResolver( + targetSchema: TableSchema, + targetLabel: String, + targetRow: Row, + sourceSchema: TableSchema, + sourceLabel: String, + sourceRow: Row + ): ColumnResolver = object : ColumnResolver { + override fun resolve(expression: ColumnExpression): Value { + val qualifier = expression.qualifier + if (qualifier != null) { + return when { + qualifier.equals(sourceLabel, ignoreCase = true) -> + sourceRow.values[sourceSchema.columnIndex(expression.name) ?: throw DiskExecutorException("no such column: ${expression.name}")] + qualifier.equals(targetLabel, ignoreCase = true) -> + targetRow.values[targetSchema.columnIndex(expression.name) ?: throw DiskExecutorException("no such column: ${expression.name}")] + else -> throw DiskExecutorException("no such table or alias: $qualifier") + } + } + val targetIndex = targetSchema.columnIndex(expression.name) + val sourceIndex = sourceSchema.columnIndex(expression.name) + if (targetIndex != null && sourceIndex != null) { + throw DiskExecutorException("column reference '${expression.name}' is ambiguous between $targetLabel and $sourceLabel") + } + targetIndex?.let { return targetRow.values[it] } + sourceIndex?.let { return sourceRow.values[it] } + throw DiskExecutorException("no such column: ${expression.name}") + } + + override fun declaredCollation(expression: ColumnExpression): Collation? { + val qualifier = expression.qualifier + if (qualifier != null) { + return when { + qualifier.equals(sourceLabel, ignoreCase = true) -> sourceSchema.column(expression.name)?.collation + qualifier.equals(targetLabel, ignoreCase = true) -> targetSchema.column(expression.name)?.collation + else -> null + } + } + val targetColumn = targetSchema.column(expression.name) + val sourceColumn = sourceSchema.column(expression.name) + return if (targetColumn != null && sourceColumn == null) targetColumn.collation + else if (sourceColumn != null && targetColumn == null) sourceColumn.collation + else null + } + } + + /** A `WHEN NOT MATCHED` clause has no target row to read - a qualifier naming the target throws rather than silently resolving to nothing. Ported 1:1 from [kestrel.engine.Executor.notMatchedResolver]. */ + private fun notMatchedResolver( + sourceSchema: TableSchema, + sourceLabel: String, + sourceRow: Row, + targetLabel: String + ): ColumnResolver = object : ColumnResolver { + override fun resolve(expression: ColumnExpression): Value { + val qualifier = expression.qualifier + if (qualifier != null && qualifier.equals(targetLabel, ignoreCase = true)) { + throw DiskExecutorException("no target row is available in a WHEN NOT MATCHED clause: $targetLabel.${expression.name}") + } + if (qualifier != null && !qualifier.equals(sourceLabel, ignoreCase = true)) { + throw DiskExecutorException("no such table or alias: $qualifier") + } + val index = sourceSchema.columnIndex(expression.name) ?: throw DiskExecutorException("no such column: ${expression.name}") + return sourceRow.values[index] + } + + override fun declaredCollation(expression: ColumnExpression): Collation? = sourceSchema.column(expression.name)?.collation + } + + /** Upfront validation for every clause of a MERGE statement, run even when the source produces zero rows. Ported 1:1 from [kestrel.engine.Executor.checkMergeColumns]. */ + private fun checkMergeColumns( + statement: MergeStatement, + targetSchema: TableSchema, + targetLabel: String, + sourceSchema: TableSchema, + sourceLabel: String + ) { + checkMergeExpressionColumns(statement.onCondition, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) + for (clause in statement.whenClauses) { + when (clause) { + is MergeMatchedUpdate -> { + clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) } + for (assignment in clause.assignments) { + if (targetSchema.columnIndex(assignment.columnName) == null) { + throw DiskExecutorException("no such column: ${assignment.columnName}") + } + assignment.value?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) } + } + } + is MergeMatchedDelete -> + clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) } + is MergeNotMatchedInsert -> { + clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = false) } + val targetIndices = resolveInsertTargets(targetSchema, clause.columns) + if (clause.values.size != targetIndices.size) { + throw DiskExecutorException( + "table ${targetSchema.name} expects ${targetIndices.size} values but ${clause.values.size} were supplied" + ) + } + clause.values.forEach { value -> + value?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = false) } + } + } + } + } + } +