feat(sql): add DiskCatalogException and DiskCatalog
This commit is contained in:
dualfroz committed
1 parent
c1e885046f
commit
46914bb67c
1 file changed
+170
@@ -0,0 +1,170 @@
|
||||
package kestrel.store.sql
|
||||
|
||||
import kestrel.KestrelException
|
||||
import kestrel.catalog.CheckConstraint
|
||||
import kestrel.catalog.CheckExpressionColumns
|
||||
import kestrel.catalog.Column
|
||||
import kestrel.catalog.DependencyAnalysis
|
||||
import kestrel.catalog.ForeignKeyConstraint
|
||||
import kestrel.catalog.RelationRename
|
||||
import kestrel.catalog.Sequence
|
||||
import kestrel.catalog.SequenceDefinition
|
||||
import kestrel.catalog.TableSchema
|
||||
import kestrel.catalog.TriggerDefinition
|
||||
import kestrel.catalog.UniqueConstraint
|
||||
import kestrel.persistence.SqlWriter
|
||||
import kestrel.sql.ast.CreateTriggerStatement
|
||||
import kestrel.sql.ast.CreateViewStatement
|
||||
import kestrel.sql.ast.Expression
|
||||
import kestrel.sql.ast.ExpressionProjection
|
||||
import kestrel.sql.ast.QueryStatement
|
||||
import kestrel.sql.ast.SelectStatement
|
||||
import kestrel.sql.ast.TriggerEvent
|
||||
import kestrel.sql.ast.TriggerTiming
|
||||
import kestrel.sql.lexer.Lexer
|
||||
import kestrel.sql.parser.Parser
|
||||
import kestrel.store.codec.ByteReader
|
||||
import kestrel.store.codec.ByteWriter
|
||||
import kestrel.store.engine.StorageEngine
|
||||
import kestrel.store.engine.TransactionalHeap
|
||||
import kestrel.store.heap.RecordId
|
||||
import kestrel.storage.Row
|
||||
import kestrel.storage.Table
|
||||
import kestrel.storage.Value
|
||||
import java.io.File
|
||||
|
||||
class DiskCatalogException(message: String) : KestrelException(message)
|
||||
|
||||
/**
|
||||
* A disk-backed system catalog: one [DiskTable] per user table, described by
|
||||
* a [TableDirectoryEntry] row in a dedicated directory heap. The directory
|
||||
* heap is always the very first heap allocated in a fresh database file, so
|
||||
* it always lands on page 0 (see [DiskManager][kestrel.store.page.DiskManager]:
|
||||
* page ids are handed out from 0 upward and nothing else runs before it) -
|
||||
* that fixed, well-known id is what lets a fresh process find the directory
|
||||
* again without another bootstrap record pointing at it.
|
||||
*
|
||||
* Every [DiskTable] this catalog opens is created with a callback that
|
||||
* rewrites its directory row, in the same transaction as the write that
|
||||
* triggered it, whenever a primary or secondary index's root page changes
|
||||
* (see [DiskTable]'s documentation on why that matters). That keeps the
|
||||
* directory heap's on-disk contents always in sync with the last committed
|
||||
* root, so [open] can reconstruct every table and index purely from what is
|
||||
* on disk after a full process restart - not just within one process's
|
||||
* lifetime the way holding a live [kestrel.store.engine.TransactionalIndex]
|
||||
* object would.
|
||||
*
|
||||
* The same directory heap also holds one row per view, tagged (see
|
||||
* `ENTRY_TAG_TABLE`/`ENTRY_TAG_VIEW`) and encoded by [ViewDirectoryEntryCodec] instead of
|
||||
* [DirectoryEntryCodec] - a view has no heap/index pages of its own to describe, only its
|
||||
* `CREATE VIEW ... AS ...;` text (see [ViewDirectoryEntry]), so it rides along in the one heap
|
||||
* that is already guaranteed to exist at a well-known page id, rather than needing a second
|
||||
* fixed-page-id heap of its own.
|
||||
*
|
||||
* A third tag, `ENTRY_TAG_SEQUENCE`, holds one row per SEQUENCE, encoded by
|
||||
* [SequenceDirectoryEntryCodec] - see [SequenceDirectoryEntry]'s own doc for why its live counter
|
||||
* (unlike everything else in this heap) is deliberately NOT part of that transactional row.
|
||||
*
|
||||
* A fourth tag, `ENTRY_TAG_TRIGGER`, holds one row per TRIGGER, encoded by
|
||||
* [TriggerDirectoryEntryCodec] - the same "store rendered SQL text, re-parse on open" trick as
|
||||
* [ViewDirectoryEntry], since a trigger's WHEN clause and body are arbitrarily deep AST too.
|
||||
*/
|
||||
class DiskCatalog private constructor(
|
||||
private val engine: StorageEngine,
|
||||
private val directory: TransactionalHeap
|
||||
) : AutoCloseable {
|
||||
companion object {
|
||||
private const val DIRECTORY_HEADER_PAGE_ID = 0L
|
||||
|
||||
/**
|
||||
* Leading byte on every row stored in the directory heap, distinguishing a
|
||||
* [TableDirectoryEntry] from a [ViewDirectoryEntry] - both share the same heap rather than
|
||||
* each getting a dedicated one, since a fresh heap's header page id is only deterministic
|
||||
* (page 0) for the very first heap [open] allocates; a second fixed-page-id heap would need
|
||||
* its own bootstrap record pointing at it, which this tag byte avoids entirely.
|
||||
*/
|
||||
private const val ENTRY_TAG_TABLE: Int = 0
|
||||
private const val ENTRY_TAG_VIEW: Int = 1
|
||||
private const val ENTRY_TAG_SEQUENCE: Int = 2
|
||||
private const val ENTRY_TAG_TRIGGER: Int = 3
|
||||
|
||||
fun open(dataFile: File, walFile: File, poolCapacity: Int = 64): DiskCatalog {
|
||||
val isFresh = !dataFile.exists() || dataFile.length() == 0L
|
||||
val engine = StorageEngine.open(dataFile, walFile, poolCapacity)
|
||||
val directory = if (isFresh) {
|
||||
val heap = engine.createHeap()
|
||||
check(heap.headerPageId == DIRECTORY_HEADER_PAGE_ID) {
|
||||
"expected the catalog directory to be the first page allocated in a fresh file, " +
|
||||
"got page ${heap.headerPageId}"
|
||||
}
|
||||
heap
|
||||
} else {
|
||||
engine.openHeap(DIRECTORY_HEADER_PAGE_ID)
|
||||
}
|
||||
val catalog = DiskCatalog(engine, directory)
|
||||
catalog.loadDirectory()
|
||||
return catalog
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A mutable box for the [rowsByName] key a [DiskTable]'s `onRootsChanged` callback should
|
||||
* persist under - captured by reference rather than by value so `ALTER TABLE ... RENAME TO`
|
||||
* (see [renameTable]) can repoint an already-constructed [DiskTable]'s callback at its new key
|
||||
* without having to reconstruct the table (which would mean reopening its heap/index pages).
|
||||
*/
|
||||
private class TableKeyBox(var value: String)
|
||||
|
||||
private class DirectoryRow(var rid: RecordId, var entry: TableDirectoryEntry, val table: DiskTable, val keyBox: TableKeyBox)
|
||||
private class ViewRow(var rid: RecordId, val entry: ViewDirectoryEntry, val query: QueryStatement)
|
||||
private class SequenceRow(var rid: RecordId, var entry: SequenceDirectoryEntry, var sequence: Sequence)
|
||||
private class TriggerRow(var rid: RecordId, var entry: TriggerDirectoryEntry, var definition: TriggerDefinition)
|
||||
|
||||
private val rowsByName = LinkedHashMap<String, DirectoryRow>()
|
||||
private val viewsByName = LinkedHashMap<String, ViewRow>()
|
||||
private val sequencesByName = LinkedHashMap<String, SequenceRow>()
|
||||
private val triggersByName = LinkedHashMap<String, TriggerRow>()
|
||||
|
||||
/**
|
||||
* `CREATE TEMP TABLE`/`CREATE TEMPORARY TABLE`: a plain in-memory [Table] - the exact same
|
||||
* class [kestrel.catalog.Catalog] uses for every table on the in-memory engine - rather than a
|
||||
* [DiskTable]. It never allocates a heap or index page and never gains a directory entry, so
|
||||
* [loadDirectory] can never discover it again after a process restart: there is no "don't
|
||||
* persist this one" flag to thread through the pager/WAL at all, unlike
|
||||
* [kestrel.catalog.TableSchema.temporary] on the in-memory engine (which shares its `tables` map
|
||||
* with every ordinary table and instead relies on [kestrel.persistence.SqlWriter] skipping it).
|
||||
* See [DiskExecutor]'s temporary-table INSERT/UPDATE/DELETE/SELECT path for how it is read and
|
||||
* mutated, and its BEGIN/SAVEPOINT/ROLLBACK handling for why a live [Table] reference (not just
|
||||
* its rows) needs to be reachable from outside this class.
|
||||
*/
|
||||
private val temporaryTables = LinkedHashMap<String, Table>()
|
||||
|
||||
private fun loadDirectory() {
|
||||
val txn = engine.beginTransaction()
|
||||
for ((rid, bytes) in directory.scan(txn)) {
|
||||
val payload = bytes.copyOfRange(1, bytes.size)
|
||||
when (val tag = bytes[0].toInt()) {
|
||||
ENTRY_TAG_TABLE -> {
|
||||
val entry = DirectoryEntryCodec.decode(payload)
|
||||
val keyBox = TableKeyBox(entry.tableName.lowercase())
|
||||
val table = openTableFromEntry(entry, keyBox)
|
||||
rowsByName[keyBox.value] = DirectoryRow(rid, entry, table, keyBox)
|
||||
}
|
||||
ENTRY_TAG_VIEW -> {
|
||||
val entry = ViewDirectoryEntryCodec.decode(payload)
|
||||
viewsByName[entry.viewName.lowercase()] = ViewRow(rid, entry, parseViewQuery(entry))
|
||||
}
|
||||
ENTRY_TAG_SEQUENCE -> {
|
||||
val entry = SequenceDirectoryEntryCodec.decode(payload)
|
||||
sequencesByName[entry.name.lowercase()] = SequenceRow(rid, entry, sequenceFromEntry(entry))
|
||||
}
|
||||
ENTRY_TAG_TRIGGER -> {
|
||||
val entry = TriggerDirectoryEntryCodec.decode(payload)
|
||||
triggersByName[entry.triggerName.lowercase()] = TriggerRow(rid, entry, parseTriggerDefinition(entry))
|
||||
}
|
||||
else -> throw DiskCatalogException("unknown directory entry tag: $tag")
|
||||
}
|
||||
}
|
||||
engine.commit(txn)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user