feat(sql): add createTemporaryTable and temporaryTableOrNull

src/main/kotlin/kestrel/store/sql/DiskCatalog.kt

@@ -329,3 +329,165 @@ class DiskCatalog private constructor(
         return table
     }
 
+    /**
+     * Registers a brand new `CREATE TEMP TABLE`/`CREATE TEMPORARY TABLE` - see [temporaryTables]'
+     * own KDoc for why this never touches [engine]/[directory] at all, unlike [createTable]. Name
+     * collisions are checked against every namespace a permanent table already checks against
+     * ([rowsByName]/[viewsByName]) plus every other temp table, deliberately NOT allowing a temp
+     * table to shadow a same-named permanent table/view the way SQLite does - see [DiskExecutor]'s
+     * own resolution order for the reasoning (a single, simpler, unambiguous namespace is a
+     * deliberate, narrower scope-cut for this executor).
+     */
+    fun createTemporaryTable(schema: TableSchema): Table {
+        val key = schema.name.lowercase()
+        if (rowsByName.containsKey(key)) throw DiskCatalogException("table '${schema.name}' already exists")
+        if (viewsByName.containsKey(key)) throw DiskCatalogException("a view named '${schema.name}' already exists")
+        if (temporaryTables.containsKey(key)) throw DiskCatalogException("table '${schema.name}' already exists")
+        val table = Table(schema)
+        temporaryTables[key] = table
+        return table
+    }
+
+    fun temporaryTableOrNull(name: String): Table? = temporaryTables[name.lowercase()]
+
+    fun dropTemporaryTable(name: String) {
+        temporaryTables.remove(name.lowercase()) ?: throw DiskCatalogException("no such temporary table: $name")
+    }
+
+    /**
+     * Every currently-open temp table, oldest first - used by [DiskExecutor]'s BEGIN/SAVEPOINT/
+     * ROLLBACK handling to snapshot/restore live [Table.rows] lists around a disk transaction. Since
+     * `CREATE`/`DROP TEMPORARY TABLE` are DDL and already rejected inside an explicit transaction
+     * (see [DiskExecutor.isDdl]), the *set* of temp tables never changes mid-transaction - only their
+     * row contents can - so returning the live objects (not a name list) is enough for that caller to
+     * snapshot/restore without a second lookup.
+     */
+    val temporaryTableValues: List<Table> get() = temporaryTables.values.toList()
+
+    /**
+     * Drops a table's directory entry so it is no longer visible, and frees every page its heap
+     * and indexes (primary and secondary) own back to the shared [kestrel.store.page.DiskManager]
+     * free list, via [DiskTable.destroyStorage] - so a later `CREATE TABLE`/`CREATE INDEX` reuses
+     * this space instead of only ever growing the data file. RESTRICT (default) is blocked by any view depending on it, directly or
+     * through another view, or by any other table's FOREIGN KEY referencing it (a self-referencing
+     * FK does not count - it disappears together with the table itself). [cascade] drops every
+     * dependent view (transitively, diamond-safe like [dropView]) and strips - never drops - every
+     * other table's FOREIGN KEY referencing this one, mirroring [kestrel.catalog.Catalog.dropTable]'s
+     * `dropDependentViews`/`dropDependentForeignKeys`: CASCADE never implies dropping a table that
+     * merely references the one being dropped, only the constraint pointing at it. Everything - the
+     * view deletions, the table's own row, and every stripped referencing schema - commits in one
+     * transaction so a mid-way failure leaves nothing half-dropped. A trigger defined on [name] is
+     * always dropped alongside it, CASCADE or not - mirrors [kestrel.catalog.Catalog.dropTable]: a
+     * trigger without its table makes no sense to keep around, unlike a FOREIGN KEY reference, so
+     * there is nothing to RESTRICT against.
+     */
+    fun dropTable(name: String, cascade: Boolean = false) {
+        val key = name.lowercase()
+        val row = rowsByName[key] ?: run {
+            if (viewsByName.containsKey(key)) throw DiskCatalogException("'$name' is a view, use DROP VIEW")
+            throw DiskCatalogException("no such table: $name")
+        }
+        val viewClosure = dependentViewsOrThrow(name, cascade)
+        val referencingTables = dependentForeignKeysOrThrow(name, cascade)
+        val droppedTriggerKeys = triggersByName.filterValues { it.definition.tableName.equals(name, ignoreCase = true) }.keys.toList()
+
+        val txn = engine.beginTransaction()
+        try {
+            for (viewKey in viewClosure) directory.delete(txn, viewsByName.getValue(viewKey).rid)
+            for (triggerKey in droppedTriggerKeys) directory.delete(txn, triggersByName.getValue(triggerKey).rid)
+            directory.delete(txn, row.rid)
+            row.table.destroyStorage(txn)
+            for ((otherKey, other) in referencingTables) {
+                val stripped = other.table.schema.foreignKeys.filterNot { it.referencedTable.equals(name, ignoreCase = true) }
+                other.table.replaceSchemaMetadata(withForeignKeys(other.table.schema, stripped))
+                persistEntry(otherKey, txn)
+            }
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+        for (viewKey in viewClosure) viewsByName.remove(viewKey)
+        for (triggerKey in droppedTriggerKeys) triggersByName.remove(triggerKey)
+        rowsByName.remove(key)
+    }
+
+    /**
+     * RESTRICT check against every view depending on [name] - returns the transitive closure to
+     * delete instead when [cascade] is set. See [dropTable]'s KDoc.
+     */
+    private fun dependentViewsOrThrow(name: String, cascade: Boolean): Set<String> {
+        val closure = transitiveViewDependents(name.lowercase())
+        if (closure.isEmpty() || cascade) return closure
+        val names = closure.mapNotNull { viewsByName[it]?.entry?.viewName }
+        throw DiskCatalogException("cannot drop '$name' because view(s) ${names.joinToString(", ")} depend on it")
+    }
+
+    /**
+     * RESTRICT check against every other table's FOREIGN KEY referencing [name] - returns those
+     * rows instead when [cascade] is set, so the caller can strip the constraint from each. See
+     * [dropTable]'s KDoc.
+     */
+    private fun dependentForeignKeysOrThrow(name: String, cascade: Boolean): List<Map.Entry<String, DirectoryRow>> {
+        val key = name.lowercase()
+        val referencing = rowsByName.entries.filter { (otherKey, other) ->
+            otherKey != key && other.table.schema.foreignKeys.any { fk -> fk.referencedTable.equals(name, ignoreCase = true) }
+        }
+        if (referencing.isEmpty() || cascade) return referencing
+        val names = referencing.map { it.value.table.schema.name }
+        throw DiskCatalogException(
+            "cannot drop '$name' because table(s) ${names.joinToString(", ")} have a FOREIGN KEY referencing it"
+        )
+    }
+
+    private fun rowOrThrow(tableName: String): DirectoryRow {
+        val key = tableName.lowercase()
+        rowsByName[key]?.let { return it }
+        if (viewsByName.containsKey(key)) throw DiskCatalogException("cannot modify view: $tableName")
+        throw DiskCatalogException("no such table: $tableName")
+    }
+
+    /**
+     * Mirrors [kestrel.catalog.Catalog.truncateTables]: wiping every row of [tableNames] makes every
+     * row of any OTHER table with a FOREIGN KEY into one of them dangling at once, so RESTRICT
+     * (default) blocks on that regardless of the FK's own `onDelete` action - there is no per-row
+     * delete for it to act on - and [cascade] means "also truncate the whole referencing table", a
+     * transitive closure over FOREIGN KEY edges computed exactly like [dependentForeignKeysOrThrow]'s
+     * single-hop version, never a per-row [DiskForeignKeyEnforcer.cascadeDelete]-style plan. Every
+     * table in the closure is emptied against the SAME [txn] the caller already opened (see
+     * [DiskExecutor.executeTruncate]), so it shares that statement's commit/rollback exactly like a
+     * WHERE-less DELETE does - unlike RESTART IDENTITY's sequence reset, which
+     * [DiskExecutor.executeTruncate] deliberately runs outside any transaction, mirroring [nextValue]'s
+     * own already-non-transactional sequence position.
+     */
+    fun truncateTables(txn: Long, tableNames: List<String>, cascade: Boolean): List<String> {
+        tableNames.forEach { rowOrThrow(it) }
+        val closure = truncateClosure(tableNames.map { it.lowercase() }.toCollection(linkedSetOf()), cascade, tableNames)
+        for (key in closure) {
+            val table = rowsByName.getValue(key).table
+            for ((rid, _) in table.scan(txn)) table.delete(txn, rid)
+        }
+        return closure.map { rowsByName.getValue(it).entry.tableName }
+    }
+
+    private fun truncateClosure(found: LinkedHashSet<String>, cascade: Boolean, requestedNames: List<String>): LinkedHashSet<String> {
+        var frontier: Set<String> = found
+        while (true) {
+            val referencing = rowsByName.values.filter { row ->
+                row.entry.tableName.lowercase() !in found &&
+                    row.table.schema.foreignKeys.any { fk -> fk.referencedTable.lowercase() in frontier }
+            }
+            if (referencing.isEmpty()) return found
+            if (!cascade) {
+                val names = referencing.map { it.entry.tableName }
+                throw DiskCatalogException(
+                    "cannot truncate ${requestedNames.joinToString(", ")} because table(s) " +
+                        "${names.joinToString(", ")} have a FOREIGN KEY referencing them; use CASCADE"
+                )
+            }
+            val next = referencing.map { it.entry.tableName.lowercase() }.toSet()
+            found.addAll(next)
+            frontier = next
+        }
+    }
+