feat: add durable persistence and SQL redo log writer
This commit is contained in:
1 parent
12a1f53c33
commit
ca5e3a67ff
2 files changed
+189
No files matched your search
@@ -0,0 +1,72 @@
|
||||
package kestrel.persistence
|
||||
|
||||
import java.io.File
|
||||
import java.io.FileOutputStream
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.StandardCopyOption
|
||||
|
||||
class Persistence(private val dataDir: File) {
|
||||
private var generation: Int
|
||||
|
||||
init {
|
||||
dataDir.mkdirs()
|
||||
generation = readManifest() ?: run { writeManifest(0); 0 }
|
||||
}
|
||||
|
||||
fun load(): String {
|
||||
val snapshot = snapshotFile(generation).let { if (it.exists()) it.readText() else "" }
|
||||
val wal = walFile(generation).let { if (it.exists()) it.readText() else "" }
|
||||
return buildString {
|
||||
append(snapshot)
|
||||
if (snapshot.isNotEmpty() && !snapshot.endsWith("\n")) append('\n')
|
||||
append(wal)
|
||||
}
|
||||
}
|
||||
|
||||
fun appendWal(statements: List<String>) {
|
||||
if (statements.isEmpty()) return
|
||||
FileOutputStream(walFile(generation), true).use { output ->
|
||||
output.write(statements.joinToString("\n", postfix = "\n").toByteArray())
|
||||
output.flush()
|
||||
output.fd.sync()
|
||||
}
|
||||
}
|
||||
|
||||
fun checkpoint(dumpSql: String) {
|
||||
val next = generation + 1
|
||||
writeSynced(snapshotFile(next), dumpSql)
|
||||
writeSynced(walFile(next), "")
|
||||
writeManifest(next)
|
||||
snapshotFile(generation).delete()
|
||||
walFile(generation).delete()
|
||||
generation = next
|
||||
}
|
||||
|
||||
private fun manifestFile() = File(dataDir, "manifest")
|
||||
|
||||
private fun snapshotFile(gen: Int) = File(dataDir, "snapshot.$gen.sql")
|
||||
|
||||
private fun walFile(gen: Int) = File(dataDir, "wal.$gen.log")
|
||||
|
||||
private fun readManifest(): Int? =
|
||||
manifestFile().takeIf { it.exists() }?.readText()?.trim()?.toIntOrNull()
|
||||
|
||||
private fun writeManifest(gen: Int) {
|
||||
val temp = File(dataDir, "manifest.tmp")
|
||||
writeSynced(temp, gen.toString())
|
||||
Files.move(
|
||||
temp.toPath(),
|
||||
manifestFile().toPath(),
|
||||
StandardCopyOption.REPLACE_EXISTING,
|
||||
StandardCopyOption.ATOMIC_MOVE
|
||||
)
|
||||
}
|
||||
|
||||
private fun writeSynced(file: File, text: String) {
|
||||
FileOutputStream(file).use { output ->
|
||||
output.write(text.toByteArray())
|
||||
output.flush()
|
||||
output.fd.sync()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
package kestrel.persistence
|
||||
|
||||
import java.math.BigDecimal
|
||||
import kestrel.catalog.Catalog
|
||||
import kestrel.catalog.Column
|
||||
import kestrel.sql.ast.Assignment
|
||||
import kestrel.sql.ast.BinaryExpression
|
||||
import kestrel.sql.ast.ColumnExpression
|
||||
import kestrel.sql.ast.CreateIndexStatement
|
||||
import kestrel.sql.ast.CreateTableStatement
|
||||
import kestrel.sql.ast.DeleteStatement
|
||||
import kestrel.sql.ast.DropIndexStatement
|
||||
import kestrel.sql.ast.DropTableStatement
|
||||
import kestrel.sql.ast.Expression
|
||||
import kestrel.sql.ast.FunctionExpression
|
||||
import kestrel.sql.ast.InsertStatement
|
||||
import kestrel.sql.ast.LiteralExpression
|
||||
import kestrel.sql.ast.Statement
|
||||
import kestrel.sql.ast.UnaryExpression
|
||||
import kestrel.sql.ast.UpdateStatement
|
||||
import kestrel.storage.Value
|
||||
|
||||
object SqlWriter {
|
||||
fun canPersist(statement: Statement): Boolean = statement is CreateTableStatement ||
|
||||
statement is DropTableStatement ||
|
||||
statement is CreateIndexStatement ||
|
||||
statement is DropIndexStatement ||
|
||||
statement is InsertStatement ||
|
||||
statement is UpdateStatement ||
|
||||
statement is DeleteStatement
|
||||
|
||||
fun write(statement: Statement): String = when (statement) {
|
||||
is CreateTableStatement -> "CREATE TABLE ${statement.tableName} (" +
|
||||
statement.columns.joinToString(", ") { columnDefinition(it) } + ");"
|
||||
is DropTableStatement -> "DROP TABLE ${statement.tableName};"
|
||||
is CreateIndexStatement ->
|
||||
"CREATE INDEX ${statement.indexName} ON ${statement.tableName} (${statement.columnName});"
|
||||
is DropIndexStatement -> "DROP INDEX ${statement.indexName};"
|
||||
is InsertStatement -> writeInsert(statement)
|
||||
is UpdateStatement -> writeUpdate(statement)
|
||||
is DeleteStatement -> writeDelete(statement)
|
||||
else -> throw IllegalArgumentException("statement is not persistable")
|
||||
}
|
||||
|
||||
fun dump(catalog: Catalog): String {
|
||||
val builder = StringBuilder()
|
||||
for (tableName in catalog.tableNames) {
|
||||
val table = catalog.table(tableName)
|
||||
builder.append("CREATE TABLE ${table.schema.name} (")
|
||||
builder.append(table.schema.columns.joinToString(", ") { columnDefinition(it) })
|
||||
builder.append(");\n")
|
||||
for (row in table.rows) {
|
||||
builder.append("INSERT INTO ${table.schema.name} VALUES (")
|
||||
builder.append(row.values.joinToString(", ") { literal(it) })
|
||||
builder.append(");\n")
|
||||
}
|
||||
}
|
||||
for (definition in catalog.indexDefinitions()) {
|
||||
builder.append("CREATE INDEX ${definition.name} ON ${definition.tableName} (${definition.columnName});\n")
|
||||
}
|
||||
return builder.toString()
|
||||
}
|
||||
|
||||
private fun writeInsert(statement: InsertStatement): String {
|
||||
val columns = statement.columns?.let { " (${it.joinToString(", ")})" } ?: ""
|
||||
val rows = statement.rows.joinToString(", ") { row ->
|
||||
"(" + row.joinToString(", ") { expression(it) } + ")"
|
||||
}
|
||||
return "INSERT INTO ${statement.tableName}$columns VALUES $rows;"
|
||||
}
|
||||
|
||||
private fun writeUpdate(statement: UpdateStatement): String {
|
||||
val assignments = statement.assignments.joinToString(", ") { assignment(it) }
|
||||
val where = statement.where?.let { " WHERE ${expression(it)}" } ?: ""
|
||||
return "UPDATE ${statement.tableName} SET $assignments$where;"
|
||||
}
|
||||
|
||||
private fun writeDelete(statement: DeleteStatement): String {
|
||||
val where = statement.where?.let { " WHERE ${expression(it)}" } ?: ""
|
||||
return "DELETE FROM ${statement.tableName}$where;"
|
||||
}
|
||||
|
||||
private fun assignment(assignment: Assignment): String =
|
||||
"${assignment.columnName} = ${expression(assignment.value)}"
|
||||
|
||||
private fun columnDefinition(column: Column): String {
|
||||
val constraint = when {
|
||||
column.primaryKey -> " PRIMARY KEY"
|
||||
column.notNull -> " NOT NULL"
|
||||
else -> ""
|
||||
}
|
||||
return "${column.name} ${column.type.displayName}$constraint"
|
||||
}
|
||||
|
||||
private fun expression(expression: Expression): String = when (expression) {
|
||||
is LiteralExpression -> literal(expression.value)
|
||||
is ColumnExpression -> expression.qualifier?.let { "$it.${expression.name}" } ?: expression.name
|
||||
is FunctionExpression ->
|
||||
"${expression.name}(${if (expression.star) "*" else expression.argument?.let { expression(it) } ?: ""})"
|
||||
is BinaryExpression ->
|
||||
"(${expression(expression.left)} ${expression.operator.symbol} ${expression(expression.right)})"
|
||||
is UnaryExpression -> "(${expression.operator.symbol} ${expression(expression.operand)})"
|
||||
}
|
||||
|
||||
private fun literal(value: Value): String = when (value) {
|
||||
is Value.IntegerValue -> value.value.toString()
|
||||
is Value.RealValue -> realText(value.value)
|
||||
is Value.TextValue -> "'" + value.value.replace("'", "''") + "'"
|
||||
is Value.BooleanValue -> value.value.toString()
|
||||
is Value.NullValue -> "NULL"
|
||||
}
|
||||
|
||||
private fun realText(value: Double): String {
|
||||
val plain = value.toString()
|
||||
return if (plain.contains('E') || plain.contains('e')) BigDecimal(value).toPlainString() else plain
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user