feat(sql): add DirectoryEntryCodec and encode
This commit is contained in:
dualfroz committed
1 parent
b5ae9f27f4
commit
c1e885046f
1 file changed
+187
@@ -0,0 +1,187 @@
|
||||
package kestrel.store.sql
|
||||
|
||||
import kestrel.catalog.Column
|
||||
import kestrel.catalog.ForeignKeyAction
|
||||
import kestrel.catalog.ForeignKeyConstraint
|
||||
import kestrel.catalog.UniqueConstraint
|
||||
import kestrel.persistence.SqlWriter
|
||||
import kestrel.sql.ast.Expression
|
||||
import kestrel.sql.ast.ExpressionProjection
|
||||
import kestrel.sql.ast.SelectStatement
|
||||
import kestrel.sql.lexer.Lexer
|
||||
import kestrel.sql.parser.Parser
|
||||
import kestrel.store.codec.ByteReader
|
||||
import kestrel.store.codec.ByteWriter
|
||||
import kestrel.store.codec.CodecException
|
||||
import kestrel.storage.Collation
|
||||
import kestrel.storage.ColumnType
|
||||
|
||||
/**
|
||||
* Encodes a [TableDirectoryEntry] as a self-contained byte payload so it can
|
||||
* live as a single row in [DiskCatalog]'s directory heap. Unlike
|
||||
* [kestrel.store.codec.TupleCodec] this is not a generic row format: it is
|
||||
* shaped exactly around the fields a directory entry has, including nested
|
||||
* column and index-definition lists. A column's [Column.default] and
|
||||
* [Column.generated] are each stored as rendered SQL text and re-parsed through
|
||||
* a throwaway `SELECT (<expr>) FROM ...` (via the same [writeDefault]/[readDefault]
|
||||
* helpers - the wire shape is identical, an optional expression rendered as text)
|
||||
* the same way [DiskCatalog.parseCheckExpressionText] resolves a
|
||||
* [CheckConstraintEntry] - the alternative, a bespoke binary codec for
|
||||
* [Expression], would have to mirror the whole AST hierarchy for what is
|
||||
* always a constant-only (DEFAULT) or own-table-column-only (GENERATED)
|
||||
* expression in practice. [Column.collation] itself needs no such trick - a plain byte, its
|
||||
* [kestrel.storage.Collation] enum ordinal - since it is never anything but one of the three
|
||||
* built-in sequences.
|
||||
*/
|
||||
object DirectoryEntryCodec {
|
||||
private const val NO_PAGE = -1L
|
||||
private const val NO_PRECISION = -1L
|
||||
|
||||
fun encode(entry: TableDirectoryEntry): ByteArray {
|
||||
val writer = ByteWriter()
|
||||
writeString(writer, entry.tableName)
|
||||
writer.writeVarInt(entry.columns.size.toLong())
|
||||
for (column in entry.columns) {
|
||||
writeString(writer, column.name)
|
||||
writeString(writer, column.type.displayName)
|
||||
writer.writeByte(if (column.notNull) 1 else 0)
|
||||
writer.writeByte(if (column.primaryKey) 1 else 0)
|
||||
writer.writeZigZag(column.precision?.toLong() ?: NO_PRECISION)
|
||||
writer.writeZigZag(column.scale?.toLong() ?: NO_PRECISION)
|
||||
writeOptionalExpression(writer, column.default)
|
||||
writeOptionalExpression(writer, column.generated)
|
||||
writer.writeByte(column.collation.ordinal)
|
||||
}
|
||||
writer.writeFixedLong(entry.dataHeaderPageId)
|
||||
writer.writeFixedLong(entry.primaryKeyIndexRootPageId ?: NO_PAGE)
|
||||
writer.writeVarInt(entry.secondaryIndexes.size.toLong())
|
||||
for (index in entry.secondaryIndexes) {
|
||||
writeString(writer, index.name)
|
||||
writer.writeVarInt(index.columnNames.size.toLong())
|
||||
for (columnName in index.columnNames) writeString(writer, columnName)
|
||||
writer.writeFixedLong(index.rootPageId)
|
||||
writer.writeByte(if (index.unique) 1 else 0)
|
||||
}
|
||||
writer.writeVarInt(entry.compositePrimaryKey.size.toLong())
|
||||
for (columnName in entry.compositePrimaryKey) writeString(writer, columnName)
|
||||
writer.writeVarInt(entry.uniqueConstraints.size.toLong())
|
||||
for (unique in entry.uniqueConstraints) {
|
||||
writer.writeByte(if (unique.name != null) 1 else 0)
|
||||
unique.name?.let { writeString(writer, it) }
|
||||
writer.writeVarInt(unique.columns.size.toLong())
|
||||
for (columnName in unique.columns) writeString(writer, columnName)
|
||||
}
|
||||
writer.writeVarInt(entry.checks.size.toLong())
|
||||
for (check in entry.checks) {
|
||||
writer.writeByte(if (check.name != null) 1 else 0)
|
||||
check.name?.let { writeString(writer, it) }
|
||||
writeString(writer, check.expressionText)
|
||||
}
|
||||
writer.writeVarInt(entry.foreignKeys.size.toLong())
|
||||
for (fk in entry.foreignKeys) {
|
||||
writer.writeByte(if (fk.name != null) 1 else 0)
|
||||
fk.name?.let { writeString(writer, it) }
|
||||
writer.writeVarInt(fk.columns.size.toLong())
|
||||
for (columnName in fk.columns) writeString(writer, columnName)
|
||||
writeString(writer, fk.referencedTable)
|
||||
writer.writeVarInt(fk.referencedColumns.size.toLong())
|
||||
for (columnName in fk.referencedColumns) writeString(writer, columnName)
|
||||
writer.writeByte(fk.onDelete.ordinal)
|
||||
writer.writeByte(fk.onUpdate.ordinal)
|
||||
}
|
||||
return writer.toByteArray()
|
||||
}
|
||||
|
||||
fun decode(bytes: ByteArray): TableDirectoryEntry {
|
||||
val reader = ByteReader(bytes)
|
||||
val tableName = readString(reader)
|
||||
val columnCount = reader.readVarInt().toInt()
|
||||
val columns = ArrayList<Column>(columnCount)
|
||||
repeat(columnCount) {
|
||||
val name = readString(reader)
|
||||
val type = ColumnType.fromKeyword(readString(reader))
|
||||
?: throw CodecException("unknown column type in directory entry for table $tableName")
|
||||
val notNull = reader.readByte() != 0
|
||||
val primaryKey = reader.readByte() != 0
|
||||
val precision = reader.readZigZag().takeIf { it != NO_PRECISION }?.toInt()
|
||||
val scale = reader.readZigZag().takeIf { it != NO_PRECISION }?.toInt()
|
||||
val default = readOptionalExpression(reader, name, "DEFAULT")
|
||||
val generated = readOptionalExpression(reader, name, "GENERATED")
|
||||
val collation = Collation.entries[reader.readByte()]
|
||||
columns.add(Column(name, type, notNull, primaryKey, precision, scale, default, generated, collation))
|
||||
}
|
||||
val dataHeaderPageId = reader.readFixedLong()
|
||||
val primaryKeyIndexRootPageId = reader.readFixedLong().takeIf { it != NO_PAGE }
|
||||
val indexCount = reader.readVarInt().toInt()
|
||||
val secondaryIndexes = ArrayList<SecondaryIndexEntry>(indexCount)
|
||||
repeat(indexCount) {
|
||||
val name = readString(reader)
|
||||
val columnNameCount = reader.readVarInt().toInt()
|
||||
val columnNames = ArrayList<String>(columnNameCount)
|
||||
repeat(columnNameCount) { columnNames.add(readString(reader)) }
|
||||
val rootPageId = reader.readFixedLong()
|
||||
val unique = reader.readByte() != 0
|
||||
secondaryIndexes.add(SecondaryIndexEntry(name, columnNames, rootPageId, unique))
|
||||
}
|
||||
val compositePrimaryKeyCount = reader.readVarInt().toInt()
|
||||
val compositePrimaryKey = ArrayList<String>(compositePrimaryKeyCount)
|
||||
repeat(compositePrimaryKeyCount) { compositePrimaryKey.add(readString(reader)) }
|
||||
val uniqueConstraintCount = reader.readVarInt().toInt()
|
||||
val uniqueConstraints = ArrayList<UniqueConstraint>(uniqueConstraintCount)
|
||||
repeat(uniqueConstraintCount) {
|
||||
val name = if (reader.readByte() != 0) readString(reader) else null
|
||||
val columnCount = reader.readVarInt().toInt()
|
||||
val columns = ArrayList<String>(columnCount)
|
||||
repeat(columnCount) { columns.add(readString(reader)) }
|
||||
uniqueConstraints.add(UniqueConstraint(name, columns))
|
||||
}
|
||||
val checkCount = reader.readVarInt().toInt()
|
||||
val checks = ArrayList<CheckConstraintEntry>(checkCount)
|
||||
repeat(checkCount) {
|
||||
val name = if (reader.readByte() != 0) readString(reader) else null
|
||||
val expressionText = readString(reader)
|
||||
checks.add(CheckConstraintEntry(name, expressionText))
|
||||
}
|
||||
val foreignKeyCount = reader.readVarInt().toInt()
|
||||
val foreignKeys = ArrayList<ForeignKeyConstraint>(foreignKeyCount)
|
||||
repeat(foreignKeyCount) {
|
||||
val name = if (reader.readByte() != 0) readString(reader) else null
|
||||
val columnCount = reader.readVarInt().toInt()
|
||||
val columns = ArrayList<String>(columnCount)
|
||||
repeat(columnCount) { columns.add(readString(reader)) }
|
||||
val referencedTable = readString(reader)
|
||||
val referencedColumnCount = reader.readVarInt().toInt()
|
||||
val referencedColumns = ArrayList<String>(referencedColumnCount)
|
||||
repeat(referencedColumnCount) { referencedColumns.add(readString(reader)) }
|
||||
val onDelete = ForeignKeyAction.values()[reader.readByte()]
|
||||
val onUpdate = ForeignKeyAction.values()[reader.readByte()]
|
||||
foreignKeys.add(ForeignKeyConstraint(columns, referencedTable, referencedColumns, onDelete, onUpdate, name))
|
||||
}
|
||||
return TableDirectoryEntry(
|
||||
tableName, columns, dataHeaderPageId, primaryKeyIndexRootPageId, secondaryIndexes,
|
||||
compositePrimaryKey, uniqueConstraints, checks, foreignKeys
|
||||
)
|
||||
}
|
||||
|
||||
private fun writeString(writer: ByteWriter, text: String) {
|
||||
writer.writeLengthPrefixed(text.toByteArray(Charsets.UTF_8))
|
||||
}
|
||||
|
||||
private fun readString(reader: ByteReader): String = String(reader.readLengthPrefixed(), Charsets.UTF_8)
|
||||
|
||||
private fun writeOptionalExpression(writer: ByteWriter, expression: Expression?) {
|
||||
writer.writeByte(if (expression != null) 1 else 0)
|
||||
expression?.let { writeString(writer, SqlWriter.expressionText(it)) }
|
||||
}
|
||||
|
||||
private fun readOptionalExpression(reader: ByteReader, columnName: String, label: String): Expression? {
|
||||
if (reader.readByte() == 0) return null
|
||||
val text = readString(reader)
|
||||
val statements = Parser(Lexer("SELECT ($text) FROM directory_entry_placeholder;").tokenize()).parseProgram()
|
||||
val statement = statements.singleOrNull() as? SelectStatement
|
||||
?: throw CodecException("corrupt $label expression for column $columnName: $text")
|
||||
val projection = statement.projection as? ExpressionProjection
|
||||
?: throw CodecException("corrupt $label expression for column $columnName: $text")
|
||||
return projection.items.single().expression
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user