diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index cdc7a6c..da057d6 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -2573,3 +2573,165 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { return scope } + private data class RecursiveCtePlan(val base: QueryStatement, val operator: SetOperator, val recursiveTerm: QueryStatement) + + /** + * Recognizes `base UNION [ALL] recursiveTerm` where exactly one side names [cte] in its own + * FROM/JOIN - structurally, on the AST, without running anything - mirroring + * [kestrel.engine.SelectRunner.splitRecursiveCte] 1:1. Both sides referencing it, or neither, + * are rejected/treated as a plain (non-recursive) compound respectively, exactly like the + * in-memory engine. + */ + private fun splitRecursiveCte(cte: CommonTableExpression): RecursiveCtePlan? { + val query = cte.query + if (query !is SetOperationStatement) return null + if (query.operator != SetOperator.Union && query.operator != SetOperator.UnionAll) return null + val leftRefs = referencesName(query.left, cte.name) + val rightRefs = referencesName(query.right, cte.name) + return when { + leftRefs && rightRefs -> + throw DiskExecutorException( + "recursive common table expression '${cte.name}' must not reference itself on both sides of ${query.operator}" + ) + rightRefs -> RecursiveCtePlan(query.left, query.operator, query.right) + leftRefs -> RecursiveCtePlan(query.right, query.operator, query.left) + else -> null + } + } + + private fun referencesName(query: QueryStatement, name: String): Boolean = when (query) { + is SelectStatement -> referencesName(query.from, name) || query.joins.any { referencesName(it.from, name) } + is SetOperationStatement -> referencesName(query.left, name) || referencesName(query.right, name) + is WithQuery -> query.ctes.none { it.name.equals(name, ignoreCase = true) } && referencesName(query.body, name) + } + + private fun referencesName(item: FromItem, name: String): Boolean = when (item) { + is TableSource -> item.name.equals(name, ignoreCase = true) + is SubquerySource -> referencesName(item.select, name) + } + + private fun cteSchema(cte: CommonTableExpression, result: SelectResult): TableSchema { + val labels = cte.columns ?: result.columns + if (labels.size != result.columns.size) { + throw DiskExecutorException( + "common table expression '${cte.name}' expects ${labels.size} columns but its query produces ${result.columns.size}" + ) + } + return derivedSchema(cte.name, result, labels) + } + + /** + * A synthetic schema for a CTE's materialized rows: same idea as + * [kestrel.engine.SelectRunner.derivedSchema] - each column's type is inferred from the first + * non-NULL value seen in it (defaulting to TEXT when every row is NULL there), since a CTE has + * no backing [TableSchema] of its own to borrow one from. + */ + private fun derivedSchema(name: String, result: SelectResult, labels: List = result.columns): TableSchema { + val columns = labels.mapIndexed { index, columnName -> + val type = result.rows.asSequence().map { it[index] }.firstOrNull { !it.isNull }?.type ?: ColumnType.Text + Column(columnName, type, notNull = false, primaryKey = false) + } + return TableSchema(name, columns) + } + + private class CteScope(val name: String, val schema: TableSchema, var rows: List>) + + /** One FROM/JOIN source's row supply: either a real disk table or a CTE's already-materialized rows. */ + private sealed interface DiskFromSource { + val schema: TableSchema + fun scan(txn: Long): List> + } + + private class DiskTableSource(val table: DiskTable) : DiskFromSource { + override val schema get() = table.schema + override fun scan(txn: Long): List> = table.scan(txn).map { it.second.values } + } + + private class CteRowSource(override val schema: TableSchema, val rows: List>) : DiskFromSource { + override fun scan(txn: Long): List> = rows + } + + /** + * Resolves one FROM/JOIN [name] to its row supply: an active CTE shadows a same-named real + * table or view (checked innermost-first via [MutableList.lastOrNull], exactly like + * [kestrel.engine.SelectRunner.resolveNamedSource]'s own CTE lookup), then a temp table, then a + * real disk table, and only then a view - materialized fresh by running its defining query + * inside the same [txn], with [expandingViews] guarding against a view (directly or + * transitively) referencing itself, mirroring [kestrel.engine.SelectRunner.resolveNamedSource] + * exactly. A temp table's current rows are captured into a [CteRowSource] snapshot exactly like + * a CTE's or view's - a plain SELECT never mutates its own FROM source mid-query, so this is + * exactly as fresh as scanning a live [DiskTable] would be, and it is what gives a temp table + * full JOIN/subquery/aggregate participation for free instead of needing its own parallel query + * pipeline. A name that is neither falls through to [DiskCatalog.table], solely for its + * "no such table" [DiskCatalogException] - already confirmed absent from + * [DiskCatalog.temporaryTableOrNull]/[DiskCatalog.tableOrNull]/[DiskCatalog.viewOrNull], so this + * always throws, preserving the exact exception type callers relied on before views existed. + */ + private fun resolveFromSource(name: String, txn: Long): DiskFromSource { + activeCtes.lastOrNull { it.name.equals(name, ignoreCase = true) }?.let { return CteRowSource(it.schema, it.rows) } + catalog.temporaryTableOrNull(name)?.let { return CteRowSource(it.schema, it.rows.map { row -> row.values }) } + catalog.tableOrNull(name)?.let { return DiskTableSource(it) } + val view = catalog.viewOrNull(name) ?: return DiskTableSource(catalog.table(name)) + val key = name.lowercase() + if (!expandingViews.add(key)) { + throw DiskExecutorException("recursive view definition: $name") + } + try { + val result = runQuery(view, txn, outer = null) + return CteRowSource(derivedSchema(name, result), result.rows) + } finally { + expandingViews.remove(key) + } + } + + /** + * Resolves one FROM/JOIN [item] to its row supply, mirroring + * [kestrel.engine.SelectRunner.resolveSource]/`buildSources`: a [TableSource] falls through to + * [resolveFromSource] by name (CTE/table/view, unchanged); a non-LATERAL [SubquerySource] (a + * derived table) runs its query eagerly, right here, inside the same [txn] - it cannot correlate + * to anything, so its rows are fixed for the whole statement, exactly like a materialized view. A + * LATERAL [SubquerySource] is different: its rows depend on whichever row of an earlier FROM/JOIN + * source it correlates with, which does not exist yet at this point - only its column *names* are + * needed here (via [lateralSchema], computed structurally from the AST without running anything) + * for offset bookkeeping; its real rows are produced later, per left row, in [executeLateralJoin]. + */ + private fun resolveFromItem(item: FromItem, txn: Long): DiskFromSource = when (item) { + is TableSource -> resolveFromSource(item.name, txn) + is SubquerySource -> + if (item.lateral) { + CteRowSource(lateralSchema(item), emptyList()) + } else { + val result = runQuery(item.select, txn, outer = null) + CteRowSource(derivedSchema(item.alias, result), result.rows) + } + } + + private fun lateralSchema(item: SubquerySource): TableSchema { + val labels = structuralColumnNames(item.select) + val columns = labels.map { Column(it, ColumnType.Text, notNull = false, primaryKey = false) } + return TableSchema(item.alias, columns) + } + + /** + * The output column names of [query], derived purely from its AST (no execution, no catalog row + * data) - needed to give a LATERAL derived table a stable schema before any row of the source it + * correlates with is known. Mirrors [kestrel.engine.SelectRunner.structuralColumnNames] 1:1. + */ + private fun structuralColumnNames(query: QueryStatement, localCtes: Map> = emptyMap()): List = + when (query) { + is SelectStatement -> when (val projection = query.projection) { + is StarProjection -> + structuralSourceColumnNames(query.from, localCtes) + + query.joins.flatMap { structuralSourceColumnNames(it.from, localCtes) } + is ExpressionProjection -> projection.items.mapIndexed { index, item -> columnLabel(item, index) } + } + is SetOperationStatement -> structuralColumnNames(query.left, localCtes) + is WithQuery -> { + val extended = localCtes.toMutableMap() + for (cte in query.ctes) { + extended[cte.name.lowercase()] = cte.columns ?: structuralColumnNames(cte.query, extended) + } + structuralColumnNames(query.body, extended) + } + } +