feat(sql): add sequenceOrNull and createSequence to DiskCatalog

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

@@ -1142,3 +1142,131 @@ class DiskCatalog private constructor(
 
     fun sequence(name: String): Sequence = sequenceOrNull(name) ?: throw DiskCatalogException("no such sequence: $name")
 
+    fun sequenceOrNull(name: String): Sequence? = sequencesByName[name.lowercase()]?.sequence
+
+    /**
+     * Sequence names deliberately live in their own namespace here, same as
+     * [kestrel.catalog.Catalog.createSequence] - no collision check against [rowsByName]/[viewsByName].
+     */
+    fun createSequence(definition: SequenceDefinition): Sequence {
+        val key = definition.name.lowercase()
+        if (sequencesByName.containsKey(key)) throw DiskCatalogException("sequence '${definition.name}' already exists")
+        val positionPageId = engine.allocateRawPage()
+        writeSequencePosition(positionPageId, definition.startValue to false)
+        val entry = SequenceDirectoryEntry(
+            definition.name, definition.incrementBy, definition.minValue, definition.maxValue,
+            definition.startValue, definition.cycle, positionPageId
+        )
+        val txn = engine.beginTransaction()
+        val rid = directory.insert(txn, taggedSequence(entry))
+        engine.commit(txn)
+        val sequence = Sequence(definition)
+        sequencesByName[key] = SequenceRow(rid, entry, sequence)
+        return sequence
+    }
+
+    /**
+     * Frees the dropped sequence's [SequenceDirectoryEntry.positionPageId] back to the shared page
+     * free list via [StorageEngine.freeRawPage], outside the directory-row transaction - like every
+     * other read/write of that page (see [writeSequencePosition]), freeing it is deliberately not
+     * part of the transactional/WAL-logged path, so an abort of surrounding SQL never resurrects a
+     * page already handed back to the free list.
+     */
+    fun dropSequence(name: String) {
+        val key = name.lowercase()
+        val row = sequencesByName[key] ?: throw DiskCatalogException("no such sequence: $name")
+        val txn = engine.beginTransaction()
+        directory.delete(txn, row.rid)
+        engine.commit(txn)
+        engine.freeRawPage(row.entry.positionPageId)
+        sequencesByName.remove(key)
+    }
+
+    /**
+     * Mirrors [kestrel.catalog.Catalog.alterSequence]: [Sequence.altered] computes the position ALTER
+     * SEQUENCE leaves the counter at (carried over unchanged, or reset by `RESTART`, see [Sequence]'s
+     * own doc) purely in memory first. Only once the transactional definition update in [directory]
+     * has actually committed is that new position written to the raw counter page - so a failure
+     * building or persisting the new [SequenceDirectoryEntry] (e.g. the sequence vanished from under
+     * a concurrent caller) can never leave the on-disk counter reflecting an ALTER whose definition
+     * change did not itself survive.
+     */
+    fun alterSequence(name: String, newDefinition: SequenceDefinition, restartAt: Long?) {
+        val key = name.lowercase()
+        val row = sequencesByName[key] ?: throw DiskCatalogException("no such sequence: $name")
+        val altered = row.sequence.altered(newDefinition, restartAt)
+        val newEntry = SequenceDirectoryEntry(
+            newDefinition.name, newDefinition.incrementBy, newDefinition.minValue, newDefinition.maxValue,
+            newDefinition.startValue, newDefinition.cycle, row.entry.positionPageId
+        )
+        val txn = engine.beginTransaction()
+        val rid = directory.update(txn, row.rid, taggedSequence(newEntry))
+        engine.commit(txn)
+        writeSequencePosition(row.entry.positionPageId, altered.snapshotPosition())
+        row.rid = rid
+        row.entry = newEntry
+        row.sequence = altered
+    }
+
+    /**
+     * Advances [name] and durably persists the new position immediately via
+     * [StorageEngine.writeRawPage][kestrel.store.engine.StorageEngine.writeRawPage] - deliberately
+     * outside of any [beginTransaction]/[commit]/[abort], so the advance survives even a ROLLBACK of
+     * whichever statement's transaction called it (see [SequenceDirectoryEntry]'s own doc). Guarded
+     * by a lock on this catalog instance rather than [kestrel.store.txn.LockManager] (which the raw
+     * counter page never goes through at all) - the one piece of this catalog actually exercised by
+     * more than one thread at a time in this codebase (see the concurrent-session BEGIN/COMMIT test),
+     * so two overlapping NEXTVAL calls on the same sequence must not race on [Sequence.next]'s plain,
+     * unsynchronized `var`s.
+     */
+    fun nextValue(name: String): Long = synchronized(this) {
+        val row = sequencesByName[name.lowercase()] ?: throw DiskCatalogException("no such sequence: $name")
+        val value = row.sequence.next()
+        writeSequencePosition(row.entry.positionPageId, row.sequence.snapshotPosition())
+        value
+    }
+
+    // ---- TRIGGER ----
+
+    val triggerNames: List<String> get() = triggersByName.values.map { it.entry.triggerName }
+
+    /** Every trigger definition, in creation order - mirrors [kestrel.catalog.Catalog.triggerDefinitions]. */
+    fun triggerDefinitions(): List<TriggerDefinition> = triggersByName.values.map { it.definition }
+
+    /** Whether any trigger (any timing/event) is defined ON [tableName] - mirrors [kestrel.catalog.Catalog.hasTriggers]. */
+    fun hasTriggers(tableName: String): Boolean = triggersByName.values.any { it.definition.tableName.equals(tableName, ignoreCase = true) }
+
+    /** Every trigger registered for ([tableName], [timing], [event]), in creation order - mirrors [kestrel.catalog.Catalog.triggersOn]. */
+    fun triggersOn(tableName: String, timing: TriggerTiming, event: TriggerEvent): List<TriggerDefinition> =
+        triggersByName.values.filter {
+            it.definition.tableName.equals(tableName, ignoreCase = true) && it.definition.timing == timing && it.definition.event == event
+        }.map { it.definition }
+
+    /**
+     * Persists [definition] as rendered `CREATE TRIGGER ...;` text (see [TriggerDirectoryEntry]) -
+     * mirrors [kestrel.catalog.Catalog.createTrigger]'s name-collision rule. Every other eager check
+     * (the trigger's own table and every body statement's target table exist, the WHEN clause shape)
+     * is [DiskExecutor]'s responsibility, same split as [createView] versus [DiskExecutor.
+     * executeCreateView].
+     */
+    fun createTrigger(definition: TriggerDefinition) {
+        val key = definition.name.lowercase()
+        if (triggersByName.containsKey(key)) throw DiskCatalogException("trigger '${definition.name}' already exists")
+        val entry = TriggerDirectoryEntry(definition.name, SqlWriter.triggerText(definition))
+        val txn = engine.beginTransaction()
+        val rid = directory.insert(txn, taggedTrigger(entry))
+        engine.commit(txn)
+        triggersByName[key] = TriggerRow(rid, entry, definition)
+    }
+
+    fun dropTrigger(name: String) {
+        val key = name.lowercase()
+        val row = triggersByName[key] ?: throw DiskCatalogException("no such trigger: $name")
+        val txn = engine.beginTransaction()
+        directory.delete(txn, row.rid)
+        engine.commit(txn)
+        triggersByName.remove(key)
+    }
+
+    override fun close() = engine.close()
+}