feat(engine): extend Executor with Executor and nextValue
This commit is contained in:
dualfroz committed
1 parent
644c83f436
commit
95120b236f
1 file changed
+201
-3
@@ -3,52 +3,215 @@ package kestrel.engine
|
||||
import kestrel.KestrelException
|
||||
import kestrel.catalog.Catalog
|
||||
import kestrel.catalog.CatalogSnapshot
|
||||
import kestrel.catalog.CheckConstraint
|
||||
import kestrel.catalog.Column
|
||||
import kestrel.catalog.ForeignKeyConstraint
|
||||
import kestrel.catalog.IndexDefinition
|
||||
import kestrel.catalog.Sequence
|
||||
import kestrel.catalog.SequenceDefinition
|
||||
import kestrel.catalog.TableSchema
|
||||
import kestrel.catalog.TriggerDefinition
|
||||
import kestrel.catalog.UniqueConstraint
|
||||
import kestrel.sql.ast.AddColumn
|
||||
import kestrel.sql.ast.AddConstraint
|
||||
import kestrel.sql.ast.AddedCheck
|
||||
import kestrel.sql.ast.AddedConstraint
|
||||
import kestrel.sql.ast.AddedForeignKey
|
||||
import kestrel.sql.ast.AddedUnique
|
||||
import kestrel.sql.ast.AlterSequenceStatement
|
||||
import kestrel.sql.ast.AlterTableStatement
|
||||
import kestrel.sql.ast.BeginStatement
|
||||
import kestrel.sql.ast.BetweenExpression
|
||||
import kestrel.sql.ast.BinaryExpression
|
||||
import kestrel.sql.ast.CallExpression
|
||||
import kestrel.sql.ast.CaseExpression
|
||||
import kestrel.sql.ast.CastExpression
|
||||
import kestrel.sql.ast.ChangeColumnType
|
||||
import kestrel.sql.ast.CollateExpression
|
||||
import kestrel.sql.ast.ColumnExpression
|
||||
import kestrel.sql.ast.CommitStatement
|
||||
import kestrel.sql.ast.ConflictDoNothing
|
||||
import kestrel.sql.ast.ConflictDoUpdate
|
||||
import kestrel.sql.ast.CreateIndexStatement
|
||||
import kestrel.sql.ast.CreateSequenceStatement
|
||||
import kestrel.sql.ast.CreateTableStatement
|
||||
import kestrel.sql.ast.CreateTriggerStatement
|
||||
import kestrel.sql.ast.CreateViewStatement
|
||||
import kestrel.sql.ast.CurrentValueExpression
|
||||
import kestrel.sql.ast.DeallocateStatement
|
||||
import kestrel.sql.ast.DeleteStatement
|
||||
import kestrel.sql.ast.DropColumn
|
||||
import kestrel.sql.ast.DropConstraint
|
||||
import kestrel.sql.ast.DropIndexStatement
|
||||
import kestrel.sql.ast.DropMode
|
||||
import kestrel.sql.ast.DropSequenceStatement
|
||||
import kestrel.sql.ast.DropTableStatement
|
||||
import kestrel.sql.ast.DropTriggerStatement
|
||||
import kestrel.sql.ast.DropViewStatement
|
||||
import kestrel.sql.ast.ExecuteStatement
|
||||
import kestrel.sql.ast.ExistsExpression
|
||||
import kestrel.sql.ast.Expression
|
||||
import kestrel.sql.ast.ExplainStatement
|
||||
import kestrel.sql.ast.FunctionExpression
|
||||
import kestrel.sql.ast.InExpression
|
||||
import kestrel.sql.ast.InSubqueryExpression
|
||||
import kestrel.sql.ast.InsertStatement
|
||||
import kestrel.sql.ast.IsNullExpression
|
||||
import kestrel.sql.ast.LikeExpression
|
||||
import kestrel.sql.ast.LiteralExpression
|
||||
import kestrel.sql.ast.MergeActionExpression
|
||||
import kestrel.sql.ast.MergeMatchedDelete
|
||||
import kestrel.sql.ast.MergeMatchedUpdate
|
||||
import kestrel.sql.ast.MergeNotMatchedInsert
|
||||
import kestrel.sql.ast.MergeStatement
|
||||
import kestrel.sql.ast.NextValueExpression
|
||||
import kestrel.sql.ast.ExpressionProjection
|
||||
import kestrel.sql.ast.OnConflictClause
|
||||
import kestrel.sql.ast.ParameterExpression
|
||||
import kestrel.sql.ast.PrepareStatement
|
||||
import kestrel.sql.ast.PreparableStatement
|
||||
import kestrel.sql.ast.Projection
|
||||
import kestrel.sql.ast.QueryStatement
|
||||
import kestrel.sql.ast.ReleaseSavepointStatement
|
||||
import kestrel.sql.ast.RenameColumn
|
||||
import kestrel.sql.ast.RenameTable
|
||||
import kestrel.sql.ast.RollbackStatement
|
||||
import kestrel.sql.ast.SelectStatement
|
||||
import kestrel.sql.ast.RollbackToSavepointStatement
|
||||
import kestrel.sql.ast.SavepointStatement
|
||||
import kestrel.sql.ast.ScalarSubqueryExpression
|
||||
import kestrel.sql.ast.SelectItem
|
||||
import kestrel.sql.ast.StarProjection
|
||||
import kestrel.sql.ast.Statement
|
||||
import kestrel.sql.ast.TriggerEvent
|
||||
import kestrel.sql.ast.TriggerTiming
|
||||
import kestrel.sql.ast.TruncateTableStatement
|
||||
import kestrel.sql.ast.UnaryExpression
|
||||
import kestrel.sql.ast.UpdateStatement
|
||||
import kestrel.sql.ast.WindowExpression
|
||||
import java.math.BigDecimal
|
||||
import kestrel.storage.Collation
|
||||
import kestrel.storage.ColumnType
|
||||
import kestrel.storage.DecimalColumns
|
||||
import kestrel.storage.Row
|
||||
import kestrel.storage.Table
|
||||
import kestrel.storage.Value
|
||||
import kestrel.storage.displayWith
|
||||
import kestrel.storage.typeName
|
||||
|
||||
class Executor(private val catalog: Catalog) {
|
||||
class Executor(private val catalog: Catalog) : SequenceAccess {
|
||||
private val selectRunner = SelectRunner(catalog)
|
||||
private val subqueryRunner = SubqueryRunner { select, outer -> selectRunner.run(select, outer) }
|
||||
private var transactionSnapshot: CatalogSnapshot? = null
|
||||
private val savepoints = mutableListOf<Pair<String, CatalogSnapshot>>()
|
||||
|
||||
/** Guards against a trigger whose body (directly or transitively) fires itself forever. */
|
||||
private var triggerDepth = 0
|
||||
|
||||
/**
|
||||
* Postgres' CURRVAL is scoped to "the current session" - this engine's closest equivalent is
|
||||
* one [Executor] instance. Keyed by [Sequence] identity rather than by name so that ALTER
|
||||
* SEQUENCE (which always installs a new [Sequence] instance, see [Sequence.altered]) or a
|
||||
* DROP+CREATE of the same name invalidates any remembered value instead of silently handing
|
||||
* back a value that belonged to a different configuration of "the same" name - a deliberately
|
||||
* conservative choice over trying to decide case by case which ALTERs should or shouldn't
|
||||
* preserve it.
|
||||
*/
|
||||
private val lastSequenceValues = HashMap<Sequence, Long>()
|
||||
|
||||
/**
|
||||
* Session-scoped, deliberately outside [catalog] - see [PrepareStatement]'s own doc for why a
|
||||
* prepared statement is not part of the transactional snapshot/restore every other catalog
|
||||
* object participates in.
|
||||
*/
|
||||
private val preparedStatements = linkedMapOf<String, PreparableStatement>()
|
||||
|
||||
val preparedStatementNames: List<String> get() = preparedStatements.keys.toList()
|
||||
|
||||
override fun nextValue(sequenceName: String): Long {
|
||||
val sequence = catalog.sequence(sequenceName)
|
||||
return sequence.next().also { lastSequenceValues[sequence] = it }
|
||||
}
|
||||
|
||||
override fun currentValue(sequenceName: String): Long {
|
||||
val sequence = catalog.sequence(sequenceName)
|
||||
return lastSequenceValues[sequence]
|
||||
?: throw KestrelException("currval of sequence '$sequenceName' is not yet defined in this session - call NEXTVAL first")
|
||||
}
|
||||
|
||||
val inTransaction: Boolean get() = transactionSnapshot != null
|
||||
|
||||
fun execute(statement: Statement): StatementResult = when (statement) {
|
||||
is CreateTableStatement -> executeCreateTable(statement)
|
||||
is DropTableStatement -> executeDropTable(statement)
|
||||
is TruncateTableStatement -> executeTruncate(statement)
|
||||
is AlterTableStatement -> executeAlterTable(statement)
|
||||
is CreateIndexStatement -> executeCreateIndex(statement)
|
||||
is DropIndexStatement -> executeDropIndex(statement)
|
||||
is CreateViewStatement -> executeCreateView(statement)
|
||||
is DropViewStatement -> executeDropView(statement)
|
||||
is CreateTriggerStatement -> executeCreateTrigger(statement)
|
||||
is DropTriggerStatement -> executeDropTrigger(statement)
|
||||
is CreateSequenceStatement -> executeCreateSequence(statement)
|
||||
is DropSequenceStatement -> executeDropSequence(statement)
|
||||
is AlterSequenceStatement -> executeAlterSequence(statement)
|
||||
is InsertStatement -> executeInsert(statement)
|
||||
is SelectStatement -> selectRunner.run(statement)
|
||||
is QueryStatement -> selectRunner.runQuery(statement)
|
||||
is DeleteStatement -> executeDelete(statement)
|
||||
is UpdateStatement -> executeUpdate(statement)
|
||||
is MergeStatement -> executeMerge(statement)
|
||||
is ExplainStatement -> ExplainResult(selectRunner.explain(statement.select))
|
||||
is BeginStatement -> executeBegin()
|
||||
is CommitStatement -> executeCommit()
|
||||
is RollbackStatement -> executeRollback()
|
||||
is SavepointStatement -> executeSavepoint(statement)
|
||||
is ReleaseSavepointStatement -> executeReleaseSavepoint(statement)
|
||||
is RollbackToSavepointStatement -> executeRollbackToSavepoint(statement)
|
||||
is PrepareStatement -> executePrepare(statement)
|
||||
is ExecuteStatement -> execute(resolveExecute(statement))
|
||||
is DeallocateStatement -> executeDeallocate(statement)
|
||||
}
|
||||
|
||||
private fun executePrepare(statement: PrepareStatement): StatementResult {
|
||||
val key = statement.name.lowercase()
|
||||
if (preparedStatements.containsKey(key)) {
|
||||
throw KestrelException("prepared statement '${statement.name}' already exists")
|
||||
}
|
||||
preparedStatements[key] = statement.target
|
||||
return PrepareResult(statement.name)
|
||||
}
|
||||
|
||||
/**
|
||||
* Looks [statement]'s name up, evaluates every argument expression as a CONSTANT (a bare
|
||||
* `Evaluator()` with no [ColumnResolver]/[SubqueryRunner]/[SequenceAccess] - a column reference,
|
||||
* subquery, or NEXTVAL/CURRVAL in an EXECUTE argument throws exactly like one would in a DEFAULT
|
||||
* expression, for the same reason: there is no row to resolve any of them against), and
|
||||
* substitutes the results for every `$1`, `$2`, ... in the prepared statement. The result is
|
||||
* always a concrete [PreparableStatement] - never [ExecuteStatement] itself - ready to run
|
||||
* through [execute] like any statement typed directly.
|
||||
*/
|
||||
fun resolveExecute(statement: ExecuteStatement): Statement {
|
||||
val target = preparedStatements[statement.name.lowercase()]
|
||||
?: throw KestrelException("no such prepared statement: ${statement.name}")
|
||||
val required = ParameterSubstitution.parameterCount(target)
|
||||
if (statement.arguments.size != required) {
|
||||
throw KestrelException(
|
||||
"prepared statement '${statement.name}' expects $required parameter(s) but got ${statement.arguments.size}"
|
||||
)
|
||||
}
|
||||
val values = statement.arguments.map { Evaluator().evaluate(it) }
|
||||
return ParameterSubstitution.substitute(target, values)
|
||||
}
|
||||
|
||||
private fun executeDeallocate(statement: DeallocateStatement): StatementResult {
|
||||
if (statement.all) {
|
||||
preparedStatements.clear()
|
||||
return DeallocateResult(null, all = true)
|
||||
}
|
||||
val name = statement.name!!
|
||||
if (preparedStatements.remove(name.lowercase()) == null) {
|
||||
throw KestrelException("no such prepared statement: $name")
|
||||
}
|
||||
return DeallocateResult(name, all = false)
|
||||
}
|
||||
|
||||
private fun executeBegin(): StatementResult {
|
||||
@@ -60,6 +223,7 @@ class Executor(private val catalog: Catalog) {
|
||||
private fun executeCommit(): StatementResult {
|
||||
if (!inTransaction) throw KestrelException("no transaction is in progress")
|
||||
transactionSnapshot = null
|
||||
savepoints.clear()
|
||||
return CommitResult
|
||||
}
|
||||
|
||||
@@ -67,9 +231,43 @@ class Executor(private val catalog: Catalog) {
|
||||
val snapshot = transactionSnapshot ?: throw KestrelException("no transaction is in progress")
|
||||
catalog.restore(snapshot)
|
||||
transactionSnapshot = null
|
||||
savepoints.clear()
|
||||
return RollbackResult
|
||||
}
|
||||
|
||||
/**
|
||||
* A savepoint that reuses an already-established name is intentionally allowed (matches
|
||||
* Postgres): it shadows the earlier one, and RELEASE/ROLLBACK TO always target the most
|
||||
* recently established savepoint of that name via [indexOfLastSavepoint], leaving the older,
|
||||
* shadowed one addressable again once the newer one is released or rolled back past.
|
||||
*/
|
||||
private fun executeSavepoint(statement: SavepointStatement): StatementResult {
|
||||
if (!inTransaction) throw KestrelException("SAVEPOINT can only be used inside a transaction")
|
||||
savepoints.add(statement.name to catalog.snapshot())
|
||||
return SavepointResult(statement.name)
|
||||
}
|
||||
|
||||
private fun executeReleaseSavepoint(statement: ReleaseSavepointStatement): StatementResult {
|
||||
if (!inTransaction) throw KestrelException("RELEASE SAVEPOINT can only be used inside a transaction")
|
||||
val index = indexOfLastSavepoint(statement.name)
|
||||
while (savepoints.size > index) savepoints.removeAt(savepoints.size - 1)
|
||||
return ReleaseSavepointResult(statement.name)
|
||||
}
|
||||
|
||||
private fun executeRollbackToSavepoint(statement: RollbackToSavepointStatement): StatementResult {
|
||||
if (!inTransaction) throw KestrelException("ROLLBACK TO SAVEPOINT can only be used inside a transaction")
|
||||
val index = indexOfLastSavepoint(statement.name)
|
||||
catalog.restore(savepoints[index].second)
|
||||
while (savepoints.size > index + 1) savepoints.removeAt(savepoints.size - 1)
|
||||
return RollbackToSavepointResult(statement.name)
|
||||
}
|
||||
|
||||
private fun indexOfLastSavepoint(name: String): Int {
|
||||
val index = savepoints.indexOfLast { it.first.equals(name, ignoreCase = true) }
|
||||
if (index < 0) throw KestrelException("no such savepoint: $name")
|
||||
return index
|
||||
}
|
||||
|
||||
private fun executeCreateTable(statement: CreateTableStatement): StatementResult {
|
||||
if (statement.columns.count { it.primaryKey } > 1) {
|
||||
throw KestrelException("table '${statement.tableName}' declares more than one primary key")
|
||||
|
||||
Reference in new issue
Block a user