feat(sql): add dropIndex and persistEntry to DiskCatalog

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

@@ -984,3 +984,161 @@ class DiskCatalog private constructor(
                 ?: throw DiskCatalogException("no such column: $columnName in table $tableName")
         }
 
+        val index = engine.createIndex()
+        val handle = SecondaryIndexHandle(indexName, columnNames, columnIndices, index, unique)
+        val backfill = engine.beginTransaction()
+        try {
+            row.table.backfillSecondaryIndex(backfill, handle)
+        } catch (failure: Exception) {
+            engine.abort(backfill)
+            throw failure
+        }
+        engine.commit(backfill)
+
+        row.table.attachSecondaryIndex(handle)
+        val persistTxn = engine.beginTransaction()
+        persistEntry(key, persistTxn)
+        engine.commit(persistTxn)
+        return handle
+    }
+
+    /**
+     * Drops a secondary index by name, searched across every table since `DROP INDEX` names only
+     * the index, not its owning table - mirrors [kestrel.catalog.Catalog.dropIndex]'s flat,
+     * table-less namespace. Unlike [dropColumn]'s CASCADE-guarded removal, this never needs a
+     * dependency check: nothing in this schema (FOREIGN KEY/CHECK/UNIQUE constraint) ever
+     * references an index by name, only by the columns it happens to cover, so
+     * [kestrel.catalog.Catalog.dropIndex] never needed one either. Frees the dropped B+Tree's own
+     * pages back to the shared free list via [DiskTable.dropSecondaryIndexByName], the same
+     * reclamation [dropTable] does for a whole table's storage.
+     */
+    fun dropIndex(indexName: String) {
+        val (key, row) = rowsByName.entries.firstOrNull { (_, row) ->
+            row.table.secondaryIndexes.any { it.name.equals(indexName, ignoreCase = true) }
+        }?.let { it.key to it.value } ?: throw DiskCatalogException("no such index: $indexName")
+        val txn = engine.beginTransaction()
+        try {
+            row.table.dropSecondaryIndexByName(indexName, txn)
+            persistEntry(key, txn)
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+
+    private fun persistEntry(tableKey: String, txnId: Long) {
+        val row = rowsByName.getValue(tableKey)
+        val updated = row.table.snapshotDirectoryEntry()
+        row.rid = directory.update(txnId, row.rid, taggedTable(updated))
+        row.entry = updated
+    }
+
+    private fun taggedTable(entry: TableDirectoryEntry): ByteArray =
+        byteArrayOf(ENTRY_TAG_TABLE.toByte()) + DirectoryEntryCodec.encode(entry)
+
+    private fun taggedView(entry: ViewDirectoryEntry): ByteArray =
+        byteArrayOf(ENTRY_TAG_VIEW.toByte()) + ViewDirectoryEntryCodec.encode(entry)
+
+    private fun taggedTrigger(entry: TriggerDirectoryEntry): ByteArray =
+        byteArrayOf(ENTRY_TAG_TRIGGER.toByte()) + TriggerDirectoryEntryCodec.encode(entry)
+
+    val viewNames: List<String> get() = viewsByName.values.map { it.entry.viewName }
+
+    fun viewOrNull(name: String): QueryStatement? = viewsByName[name.lowercase()]?.query
+
+    /**
+     * Persists a view's defining [query] as rendered SQL text (see [ViewDirectoryEntry]) - mirrors
+     * [kestrel.catalog.Catalog.createView]'s name-collision rules (a view and a table can never
+     * share a name, `orReplace` only waives the view-already-exists check).
+     */
+    fun createView(name: String, query: QueryStatement, orReplace: Boolean) {
+        val key = name.lowercase()
+        if (rowsByName.containsKey(key)) throw DiskCatalogException("a table named '$name' already exists")
+        if (!orReplace && viewsByName.containsKey(key)) throw DiskCatalogException("view '$name' already exists")
+        val entry = ViewDirectoryEntry(name, SqlWriter.write(CreateViewStatement(name, query, orReplace = false)))
+        val bytes = taggedView(entry)
+        val txn = engine.beginTransaction()
+        val existing = viewsByName[key]
+        val rid = if (existing != null) directory.update(txn, existing.rid, bytes) else directory.insert(txn, bytes)
+        engine.commit(txn)
+        viewsByName[key] = ViewRow(rid, entry, query)
+    }
+
+    /**
+     * Drops a view, blocked (RESTRICT, the default) by any other view depending on it - directly or
+     * transitively through a chain of views - unless [cascade] is set, in which case the whole
+     * transitive closure of dependents is dropped alongside it. Mirrors
+     * [kestrel.catalog.Catalog.dropView]/`dropDependentViews`'s "diamond-safe" ordering: the closure
+     * is computed once, up front, before anything is removed.
+     */
+    fun dropView(name: String, cascade: Boolean = false) {
+        val key = name.lowercase()
+        if (!viewsByName.containsKey(key)) {
+            if (rowsByName.containsKey(key)) throw DiskCatalogException("'$name' is a table, use DROP TABLE")
+            throw DiskCatalogException("no such view: $name")
+        }
+        val closure = transitiveViewDependents(key)
+        if (closure.isNotEmpty() && !cascade) {
+            val names = closure.mapNotNull { viewsByName[it]?.entry?.viewName }
+            throw DiskCatalogException(
+                "cannot drop '$name' because view(s) ${names.joinToString(", ")} depend on it; use CASCADE"
+            )
+        }
+        val txn = engine.beginTransaction()
+        for (depKey in closure) directory.delete(txn, viewsByName.getValue(depKey).rid)
+        directory.delete(txn, viewsByName.getValue(key).rid)
+        engine.commit(txn)
+        for (depKey in closure) viewsByName.remove(depKey)
+        viewsByName.remove(key)
+    }
+
+    /**
+     * Every view depending on [rootKey] (already lowercase), directly or through another view -
+     * [rootKey] itself may name a table or a view, [DependencyAnalysis] does not care which.
+     * Identical fixed-point walk to [kestrel.catalog.Catalog.transitiveViewDependents].
+     */
+    private fun transitiveViewDependents(rootKey: String): Set<String> {
+        val found = linkedSetOf<String>()
+        var frontier = setOf(rootKey)
+        while (true) {
+            val next = viewsByName.entries
+                .filter { (key, _) -> key != rootKey && key !in found }
+                .filter { (_, row) -> frontier.any { DependencyAnalysis.referencesRelation(row.query, it) } }
+                .map { it.key }
+                .toSet()
+            if (next.isEmpty()) return found
+            found.addAll(next)
+            frontier = next
+        }
+    }
+
+    // ---- SEQUENCE ----
+
+    /**
+     * Rebuilds a live [Sequence] from its persisted definition plus whatever [SequenceDirectoryEntry.
+     * positionPageId] currently holds - the exact `(value, called)` pair [Sequence.snapshotPosition]
+     * last wrote there, so reload reproduces the live instance's internal state precisely rather than
+     * approximating it (see [Sequence.snapshotPosition]'s own doc for why [Sequence.peekNext] is the
+     * wrong thing to persist this way).
+     */
+    private fun sequenceFromEntry(entry: SequenceDirectoryEntry): Sequence {
+        val definition = SequenceDefinition(entry.name, entry.incrementBy, entry.minValue, entry.maxValue, entry.startValue, entry.cycle)
+        val reader = ByteReader(engine.readRawPage(entry.positionPageId))
+        val value = reader.readFixedLong()
+        val called = reader.readByte() != 0
+        return Sequence(definition, initialValue = value, initialCalled = called)
+    }
+
+    private fun writeSequencePosition(pageId: Long, position: Pair<Long, Boolean>) {
+        val (value, called) = position
+        engine.writeRawPage(pageId, ByteWriter(9).writeFixedLong(value).writeByte(if (called) 1 else 0).toByteArray())
+    }
+
+    private fun taggedSequence(entry: SequenceDirectoryEntry): ByteArray =
+        byteArrayOf(ENTRY_TAG_SEQUENCE.toByte()) + SequenceDirectoryEntryCodec.encode(entry)
+
+    val sequenceNames: List<String> get() = sequencesByName.values.map { it.entry.name }
+
+    fun sequence(name: String): Sequence = sequenceOrNull(name) ?: throw DiskCatalogException("no such sequence: $name")
+