feat(sql): add parseViewQuery and parseTriggerDefinition to DiskCatalog
This commit is contained in:
dualfroz committed
1 parent
46914bb67c
commit
249b3e03fd
1 file changed
+161
@@ -168,3 +168,164 @@ class DiskCatalog private constructor(
|
||||
engine.commit(txn)
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-derives a view's [QueryStatement] from its stored `CREATE VIEW ... AS ...;` text using the
|
||||
* same [Lexer]/[Parser] pair the REPL and every other SQL entry point already go through - no
|
||||
* bespoke AST codec, see [ViewDirectoryEntry]'s own KDoc for why.
|
||||
*/
|
||||
private fun parseViewQuery(entry: ViewDirectoryEntry): QueryStatement {
|
||||
val statements = Parser(Lexer(entry.statementText).tokenize()).parseProgram()
|
||||
val statement = statements.singleOrNull() as? CreateViewStatement
|
||||
?: throw DiskCatalogException("corrupt view directory entry for ${entry.viewName}")
|
||||
return statement.query
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-derives a trigger's [TriggerDefinition] from its stored `CREATE TRIGGER ...;` text - same
|
||||
* "reuse the real Lexer/Parser instead of a bespoke codec" trick [parseViewQuery] uses for a
|
||||
* whole query.
|
||||
*/
|
||||
private fun parseTriggerDefinition(entry: TriggerDirectoryEntry): TriggerDefinition {
|
||||
val statements = Parser(Lexer(entry.statementText).tokenize()).parseProgram()
|
||||
val statement = statements.singleOrNull() as? CreateTriggerStatement
|
||||
?: throw DiskCatalogException("corrupt trigger directory entry for ${entry.triggerName}")
|
||||
return TriggerDefinition(
|
||||
statement.triggerName, statement.timing, statement.event, statement.tableName, statement.whenClause, statement.body
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-derives a CHECK constraint's [Expression] from its stored text (see [CheckConstraintEntry])
|
||||
* by wrapping it back into a throwaway, never-executed `SELECT (<expr>) FROM ...` and pulling the
|
||||
* expression back out of the parsed projection - the same "reuse the real Lexer/Parser instead of
|
||||
* a bespoke codec" trick [parseViewQuery] uses for a whole query. The FROM target is never
|
||||
* resolved against this or any catalog, so its name is arbitrary.
|
||||
*/
|
||||
private fun parseCheckExpressionText(text: String): Expression {
|
||||
val statements = Parser(Lexer("SELECT ($text) FROM check_constraint_placeholder;").tokenize()).parseProgram()
|
||||
val statement = statements.singleOrNull() as? SelectStatement
|
||||
?: throw DiskCatalogException("corrupt CHECK constraint expression: $text")
|
||||
val projection = statement.projection as? ExpressionProjection
|
||||
?: throw DiskCatalogException("corrupt CHECK constraint expression: $text")
|
||||
return projection.items.single().expression
|
||||
}
|
||||
|
||||
private fun openTableFromEntry(entry: TableDirectoryEntry, keyBox: TableKeyBox): DiskTable {
|
||||
val heap = engine.openHeap(entry.dataHeaderPageId)
|
||||
val primaryIndex = entry.primaryKeyIndexRootPageId?.let { engine.openIndex(it) }
|
||||
val schema = TableSchema(
|
||||
entry.tableName, entry.columns,
|
||||
foreignKeys = entry.foreignKeys,
|
||||
uniqueConstraints = entry.uniqueConstraints,
|
||||
compositePrimaryKey = entry.compositePrimaryKey,
|
||||
checks = entry.checks.map { CheckConstraint(it.name, parseCheckExpressionText(it.expressionText)) }
|
||||
)
|
||||
val table = DiskTable(schema, heap, primaryIndex, mutableListOf()) { txnId ->
|
||||
persistEntry(keyBox.value, txnId)
|
||||
}
|
||||
for (secondary in entry.secondaryIndexes) {
|
||||
val columnIndices = secondary.columnNames.map { columnName ->
|
||||
schema.columnIndex(columnName)
|
||||
?: throw DiskCatalogException("directory entry for ${entry.tableName} references unknown column $columnName")
|
||||
}
|
||||
val index = engine.openIndex(secondary.rootPageId)
|
||||
table.attachSecondaryIndex(SecondaryIndexHandle(secondary.name, secondary.columnNames, columnIndices, index, secondary.unique))
|
||||
}
|
||||
return table
|
||||
}
|
||||
|
||||
fun beginTransaction(): Long = engine.beginTransaction()
|
||||
fun commit(txnId: Long) = engine.commit(txnId)
|
||||
fun abort(txnId: Long) = engine.abort(txnId)
|
||||
fun savepoint(txnId: Long): Long = engine.savepoint(txnId)
|
||||
fun rollbackToSavepoint(txnId: Long, savepointLsn: Long) = engine.rollbackToSavepoint(txnId, savepointLsn)
|
||||
fun discardSavepoint(txnId: Long, savepointLsn: Long) = engine.discardSavepoint(txnId, savepointLsn)
|
||||
|
||||
val tableNames: List<String> get() = rowsByName.values.map { it.entry.tableName } + temporaryTables.values.map { it.schema.name }
|
||||
|
||||
/**
|
||||
* Total pages ever allocated in the backing data file, including any now sitting on the free
|
||||
* list after a `DROP TABLE`/`DROP INDEX` (see [DiskTable.destroyStorage]/[DiskTable.dropSecondaryIndexByName]) -
|
||||
* exposed so a caller (chiefly a test) can confirm a later `CREATE TABLE`/`CREATE INDEX` reused
|
||||
* that reclaimed space instead of only ever growing the file, the same proof-by-counter
|
||||
* [DiskTable]'s own `scanCount`/`pointLookupCount`/etc. give for index pushdown.
|
||||
*/
|
||||
fun allocatedPageCount(): Long = engine.allocatedPageCount()
|
||||
|
||||
fun table(name: String): DiskTable {
|
||||
val key = name.lowercase()
|
||||
rowsByName[key]?.let { return it.table }
|
||||
if (viewsByName.containsKey(key)) throw DiskCatalogException("cannot modify view: $name")
|
||||
throw DiskCatalogException("no such table: $name")
|
||||
}
|
||||
|
||||
fun tableOrNull(name: String): DiskTable? = rowsByName[name.lowercase()]?.table
|
||||
|
||||
/**
|
||||
* `compositePrimaryKey` mirrors [kestrel.catalog.TableSchema.compositePrimaryKey]: a
|
||||
* table-level `PRIMARY KEY (a, b, ...)`, as an alternative to an inline `col TYPE PRIMARY KEY`
|
||||
* tracked by [Column.primaryKey]. Marking the listed columns NOT NULL (like the in-memory
|
||||
* [kestrel.engine.Executor] does for the same clause) is the caller's responsibility - this
|
||||
* catalog only records schema/storage, it does not rewrite the SQL-level column definitions.
|
||||
*/
|
||||
fun createTable(
|
||||
name: String,
|
||||
columns: List<Column>,
|
||||
compositePrimaryKey: List<String> = emptyList(),
|
||||
uniqueConstraints: List<UniqueConstraint> = emptyList(),
|
||||
checks: List<CheckConstraint> = emptyList(),
|
||||
foreignKeys: List<ForeignKeyConstraint> = emptyList()
|
||||
): DiskTable {
|
||||
val key = name.lowercase()
|
||||
if (rowsByName.containsKey(key)) throw DiskCatalogException("table '$name' already exists")
|
||||
if (viewsByName.containsKey(key)) throw DiskCatalogException("a view named '$name' already exists")
|
||||
if (temporaryTables.containsKey(key)) throw DiskCatalogException("a temporary table named '$name' already exists")
|
||||
if (columns.count { it.primaryKey } > 1 ||
|
||||
(compositePrimaryKey.isNotEmpty() && columns.any { it.primaryKey })
|
||||
) {
|
||||
throw DiskCatalogException("table '$name' declares more than one primary key")
|
||||
}
|
||||
if (compositePrimaryKey.isNotEmpty()) {
|
||||
val declared = compositePrimaryKey.map { it.lowercase() }.toSet()
|
||||
if (declared.size != compositePrimaryKey.size) {
|
||||
throw DiskCatalogException("table '$name' declares a duplicate column in its PRIMARY KEY")
|
||||
}
|
||||
for (columnName in compositePrimaryKey) {
|
||||
if (columns.none { it.name.equals(columnName, ignoreCase = true) }) {
|
||||
throw DiskCatalogException("PRIMARY KEY refers to unknown column: $columnName")
|
||||
}
|
||||
}
|
||||
}
|
||||
for (unique in uniqueConstraints) {
|
||||
for (columnName in unique.columns) {
|
||||
if (columns.none { it.name.equals(columnName, ignoreCase = true) }) {
|
||||
throw DiskCatalogException("UNIQUE constraint refers to unknown column: $columnName")
|
||||
}
|
||||
}
|
||||
}
|
||||
val hasPrimaryKey = columns.any { it.primaryKey } || compositePrimaryKey.isNotEmpty()
|
||||
|
||||
val heap = engine.createHeap()
|
||||
val primaryIndex = if (hasPrimaryKey) engine.createIndex() else null
|
||||
val entry = TableDirectoryEntry(
|
||||
name, columns, heap.headerPageId, primaryIndex?.rootPageId, emptyList(),
|
||||
compositePrimaryKey, uniqueConstraints, checks.map { CheckConstraintEntry.from(it) }, foreignKeys
|
||||
)
|
||||
|
||||
val txn = engine.beginTransaction()
|
||||
val rid = directory.insert(txn, taggedTable(entry))
|
||||
engine.commit(txn)
|
||||
|
||||
val schema = TableSchema(
|
||||
name, columns,
|
||||
foreignKeys = foreignKeys,
|
||||
uniqueConstraints = uniqueConstraints,
|
||||
compositePrimaryKey = compositePrimaryKey,
|
||||
checks = checks
|
||||
)
|
||||
val keyBox = TableKeyBox(key)
|
||||
val table = DiskTable(schema, heap, primaryIndex, mutableListOf()) { txnId -> persistEntry(keyBox.value, txnId) }
|
||||
rowsByName[key] = DirectoryRow(rid, entry, table, keyBox)
|
||||
return table
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user