add recursive CTEs, window functions, and unique upserts

Support the SQL surface Vikunja needs over the PizzaSQL engine:

- non-recursive CTEs are desugared into derived tables (column lists, chaining, UNION ALL)
- recursive CTEs are materialized with fixpoint iteration and a working set
- ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) window functions
- ORDER BY ... NULLS FIRST/LAST
- ON CONFLICT (cols) DO UPDATE/NOTHING on unique indexes, including excluded.col
- evaluate correlated EXISTS in a single-table WHERE outside the locked scan to avoid a deadlock
- expose CTE and derived-table columns to the analyzer, including wildcard expansion

pkg/analyzer/analyzer.go

@@ -107,6 +107,12 @@ func (a *Analyzer) GetCatalog() *Catalog {
 
 // analyzeSelect analyzes a SELECT statement.
 func (a *Analyzer) analyzeSelect(stmt *parser.SelectStmt) error {
+	// Register CTE names before resolving FROM so both the main query and the
+	// recursive legs can reference them.
+	for _, cte := range stmt.With {
+		a.scope.DefineTable(cteTableInfo(cte))
+	}
+
 	// First, resolve tables in FROM clause
 	if err := a.resolveFromClause(stmt.From); err != nil {
 		return err
@@ -261,69 +267,157 @@ func (a *Analyzer) analyzeSelect(stmt *parser.SelectStmt) error {
 // resolveFromClause adds tables from FROM clause to scope.
 func (a *Analyzer) resolveFromClause(tables []parser.TableRef) error {
 	for _, ref := range tables {
-		// Handle subquery (derived table)
-		if ref.Subquery != nil {
-			// Analyze the subquery
-			if err := a.analyzeSelect(ref.Subquery); err != nil {
+		if err := a.defineTableRef(ref); err != nil {
+			return err
+		}
+
+		// Handle JOINs
+		if ref.Join != nil {
+			if err := a.resolveJoin(ref.Join); err != nil {
 				return err
 			}
+		}
+	}
+	return nil
+}
 
-			// Create a table info from subquery columns
-			// For now, we'll use a simplified approach - just mark it as a derived table
-			tableInfo := &TableInfo{
-				Name:    ref.Alias, // Derived tables MUST have an alias
-				Columns: []ColumnInfo{},
-				Alias:   ref.Alias,
-			}
+// defineTableRef adds a single FROM/JOIN table to scope. The reference may be a
+// derived table (subquery), a CTE, or a real catalog table.
+func (a *Analyzer) defineTableRef(ref parser.TableRef) error {
+	if ref.Subquery != nil {
+		if err := a.analyzeSelect(ref.Subquery); err != nil {
+			return err
+		}
+		a.scope.DefineTable(a.derivedTableInfo(ref))
+		return nil
+	}
 
-			// Add columns from SELECT list
-			for _, col := range ref.Subquery.Columns {
-				colName := ""
-				if col.Alias != "" {
-					colName = col.Alias
-				} else if colRef, ok := col.Expr.(*parser.ColumnRef); ok {
-					colName = colRef.Column
-				} else {
-					// For expressions without alias, use a generated name
-					colName = fmt.Sprintf("col_%d", len(tableInfo.Columns))
-				}
+	if cte, ok := a.scope.LookupTable(ref.Name); ok {
+		a.scope.DefineTable(&TableInfo{
+			Name:    cte.Name,
+			Columns: cte.Columns,
+			Alias:   ref.Alias,
+			IsView:  cte.IsView,
+		})
+		return nil
+	}
 
-				tableInfo.Columns = append(tableInfo.Columns, ColumnInfo{
-					Name:      colName,
-					TableName: ref.Alias,
-					Type:      TypeAny, // We'd need type inference for proper typing
-				})
-			}
+	table, ok := a.catalog.GetTable(ref.Name)
+	if !ok {
+		return &AnalysisError{
+			Type:    ErrTableNotFound,
+			Message: fmt.Sprintf("table not found: %s", ref.Name),
+		}
+	}
+	a.scope.DefineTable(&TableInfo{
+		Name:    table.Name,
+		Columns: table.Columns,
+		Alias:   ref.Alias,
+		IsView:  table.IsView,
+	})
+	return nil
+}
 
-			a.scope.DefineTable(tableInfo)
-		} else {
-			// Regular table reference
-			table, ok := a.catalog.GetTable(ref.Name)
-			if !ok {
-				return &AnalysisError{
-					Type:    ErrTableNotFound,
-					Message: fmt.Sprintf("table not found: %s", ref.Name),
+// derivedTableInfo builds the scope entry for a derived table. A compound
+// subquery (UNION/…) keeps its projection on the first leg. A wildcard expands
+// the columns of the subquery's source tables so callers can reference them.
+func (a *Analyzer) derivedTableInfo(ref parser.TableRef) *TableInfo {
+	info := &TableInfo{Name: ref.Alias, Alias: ref.Alias}
+	leg := firstSelectLeg(ref.Subquery)
+	if leg == nil {
+		return info
+	}
+	for _, col := range leg.Columns {
+		if col.Star {
+			for _, tref := range collectAllTableRefs(leg.From) {
+				if t, ok := a.scope.LookupTable(tref.Name); ok {
+					for _, c := range t.Columns {
+						info.Columns = append(info.Columns, ColumnInfo{Name: c.Name, TableName: ref.Alias, Type: c.Type})
+					}
+				} else if t, ok := a.catalog.GetTable(tref.Name); ok {
+					for _, c := range t.Columns {
+						info.Columns = append(info.Columns, ColumnInfo{Name: c.Name, TableName: ref.Alias, Type: c.Type})
+					}
 				}
 			}
-
-			// Create a copy with alias if specified
-			tableInfo := &TableInfo{
-				Name:    table.Name,
-				Columns: table.Columns,
-				Alias:   ref.Alias,
-				IsView:  table.IsView,
+			continue
+		}
+		name := ""
+		switch {
+		case col.Alias != "":
+			name = col.Alias
+		case col.Expr != nil:
+			if colRef, ok := col.Expr.(*parser.ColumnRef); ok {
+				name = colRef.Column
 			}
-			a.scope.DefineTable(tableInfo)
 		}
+		if name == "" {
+			name = fmt.Sprintf("col_%d", len(info.Columns))
+		}
+		info.Columns = append(info.Columns, ColumnInfo{Name: name, TableName: ref.Alias, Type: TypeAny})
+	}
+	return info
+}
 
-		// Handle JOINs
-		if ref.Join != nil {
-			if err := a.resolveJoin(ref.Join); err != nil {
-				return err
+// collectAllTableRefs flattens a FROM list and its join chains.
+func collectAllTableRefs(from []parser.TableRef) []parser.TableRef {
+	var refs []parser.TableRef
+	var add func(parser.TableRef)
+	add = func(ref parser.TableRef) {
+		if ref.Subquery != nil {
+			return
+		}
+		refs = append(refs, ref)
+		if ref.Join != nil && ref.Join.Table != nil {
+			add(*ref.Join.Table)
+		}
+	}
+	for _, ref := range from {
+		add(ref)
+	}
+	return refs
+}
+
+// cteTableInfo describes the columns a CTE exposes. The declared column list
+// wins; otherwise the anchor/first-leg projection names are used.
+func cteTableInfo(cte *parser.CTE) *TableInfo {
+	leg := firstSelectLeg(cte.Query)
+	info := &TableInfo{Name: cte.Name}
+	if leg == nil {
+		return info
+	}
+	for i, col := range leg.Columns {
+		name := ""
+		switch {
+		case i < len(cte.Columns):
+			name = cte.Columns[i]
+		case col.Alias != "":
+			name = col.Alias
+		case col.Expr != nil:
+			if ref, ok := col.Expr.(*parser.ColumnRef); ok {
+				name = ref.Column
 			}
 		}
+		if name == "" {
+			name = fmt.Sprintf("col_%d", i)
+		}
+		info.Columns = append(info.Columns, ColumnInfo{
+			Name:      name,
+			TableName: cte.Name,
+			Type:      TypeAny,
+		})
 	}
-	return nil
+	return info
+}
+
+// firstSelectLeg returns the SELECT whose projection describes a (possibly
+// compound) subquery's output columns. A UNION/INTERSECT/EXCEPT keeps the
+// projection on its first leg.
+func firstSelectLeg(sel *parser.SelectStmt) *parser.SelectStmt {
+	for sel != nil && sel.Compound != nil {
+		sel = sel.Compound.Left
+	}
+	return sel
 }
 
 // validateTableStar checks that a qualified wildcard qualifier names a table or
@@ -345,21 +439,9 @@ func (a *Analyzer) resolveJoin(join *parser.JoinClause) error {
 		return nil
 	}
 
-	table, ok := a.catalog.GetTable(join.Table.Name)
-	if !ok {
-		return &AnalysisError{
-			Type:    ErrTableNotFound,
-			Message: fmt.Sprintf("table not found: %s", join.Table.Name),
-		}
-	}
-
-	tableInfo := &TableInfo{
-		Name:    table.Name,
-		Columns: table.Columns,
-		Alias:   join.Table.Alias,
-		IsView:  table.IsView,
+	if err := a.defineTableRef(*join.Table); err != nil {
+		return err
 	}
-	a.scope.DefineTable(tableInfo)
 
 	// Analyze ON condition
 	if join.Condition != nil {
@@ -658,11 +740,42 @@ func (a *Analyzer) analyzeExpr(expr parser.Expr) (*ExprInfo, error) {
 		return a.analyzeExistsExpr(e)
 	case *parser.SubqueryExpr:
 		return a.analyzeSubqueryExpr(e)
+	case *parser.WindowExpr:
+		return a.analyzeWindowExpr(e)
 	default:
 		return &ExprInfo{Type: TypeUnknown}, nil
 	}
 }
 
+// analyzeWindowExpr analyzes a window function. Window functions are not
+// aggregates for the purpose of GROUP BY handling.
+func (a *Analyzer) analyzeWindowExpr(e *parser.WindowExpr) (*ExprInfo, error) {
+	if e.Func != nil {
+		for _, arg := range e.Func.Args {
+			if _, err := a.analyzeExpr(arg); err != nil {
+				return nil, err
+			}
+		}
+	}
+	for _, p := range e.PartitionBy {
+		if _, err := a.analyzeExpr(p); err != nil {
+			return nil, err
+		}
+	}
+	for _, o := range e.OrderBy {
+		if _, err := a.analyzeExpr(o.Expr); err != nil {
+			return nil, err
+		}
+	}
+	ret := TypeInteger
+	if e.Func != nil {
+		if sig, ok := builtinFunctions[e.Func.Name]; ok {
+			ret = sig.ReturnType
+		}
+	}
+	return &ExprInfo{Type: ret}, nil
+}
+
 func (a *Analyzer) analyzeLiteral(e *parser.LiteralExpr) (*ExprInfo, error) {
 	info := &ExprInfo{IsConstant: true}
 
@@ -758,9 +871,10 @@ func (a *Analyzer) analyzeBinaryExpr(e *parser.BinaryExpr) (*ExprInfo, error) {
 
 	switch e.Op {
 	case lexer.TokenPlus, lexer.TokenMinus, lexer.TokenStar, lexer.TokenSlash, lexer.TokenPercent:
-		// Arithmetic operators
+		// Arithmetic operators. TypeAny comes from derived tables and CTEs whose
+		// column types are not tracked, so it is allowed rather than rejected.
 		info.Type = CommonType(left.Type, right.Type)
-		if !left.Type.IsNumeric() && left.Type != TypeNull && left.Type != TypeUnknown {
+		if !left.Type.IsNumeric() && left.Type != TypeNull && left.Type != TypeUnknown && left.Type != TypeAny {
 			return nil, &AnalysisError{
 				Type:    ErrTypeMismatch,
 				Message: fmt.Sprintf("arithmetic operator requires numeric type, got %s", left.Type),

pkg/executor/correlated_exists_test.go

@@ -0,0 +1,42 @@
+package executor
+
+import (
+	"testing"
+	"time"
+)
+
+// TestCorrelatedExistsInWhere reproduces Vikunja's task filter: a single-table
+// scan whose WHERE contains a correlated EXISTS. It must not deadlock.
+func TestCorrelatedExistsInWhere(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE tasks (id INTEGER PRIMARY KEY, done INTEGER, deleted_at TEXT)")
+	execMust(t, e, "CREATE TABLE task_assignees (task_id INTEGER, user_id INTEGER)")
+	execMust(t, e, "CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)")
+	execMust(t, e, "INSERT INTO tasks VALUES (1, 0, NULL)")
+	execMust(t, e, "INSERT INTO tasks VALUES (2, 0, NULL)")
+	execMust(t, e, "INSERT INTO task_assignees VALUES (1, 10)")
+	execMust(t, e, "INSERT INTO users VALUES (10, 'alice')")
+
+	done := make(chan *Result, 1)
+	go func() {
+		res, err := execSQL(e, "SELECT tasks.id FROM tasks WHERE tasks.done = 0 AND EXISTS (SELECT 1 FROM task_assignees INNER JOIN users ON users.id = user_id WHERE (tasks.id = task_id) AND username IN ('alice'))")
+		if err != nil {
+			done <- nil
+			return
+		}
+		done <- res
+	}()
+
+	select {
+	case res := <-done:
+		if res == nil {
+			t.Fatal("correlated EXISTS query failed")
+		}
+		if res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+			t.Fatalf("expected only task 1, got %v", res.Rows)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("correlated EXISTS query deadlocked")
+	}
+}

pkg/executor/cte.go

@@ -0,0 +1,198 @@
+package executor
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// maxRecursiveCTEIterations bounds fixpoint iteration so a recursive CTE with a
+// cycle cannot spin forever.
+const maxRecursiveCTEIterations = 1000
+
+// cteTable is a materialized common table expression.
+type cteTable struct {
+	columns []string
+	rows    []storage.Row
+}
+
+// cteTableFor returns the materialized CTE named name, if any.
+func (e *Executor) cteTableFor(name string) (*cteTable, bool) {
+	if e.cteTables == nil {
+		return nil, false
+	}
+	t, ok := e.cteTables[strings.ToLower(name)]
+	return t, ok
+}
+
+// executeWith materializes every CTE in order, then runs the main query with the
+// CTE tables available to table resolution. Nested WITH clauses keep the outer
+// tables visible.
+func (e *Executor) executeWith(stmt *parser.SelectStmt) (*Result, error) {
+	prev := e.cteTables
+	next := make(map[string]*cteTable, len(stmt.With)+len(prev))
+	for k, v := range prev {
+		next[k] = v
+	}
+	e.cteTables = next
+	defer func() { e.cteTables = prev }()
+
+	for _, cte := range stmt.With {
+		if err := e.materializeCTE(cte); err != nil {
+			return nil, err
+		}
+	}
+
+	main := *stmt
+	main.With = nil
+	return e.executeSelect(&main)
+}
+
+func (e *Executor) materializeCTE(cte *parser.CTE) error {
+	// WITH RECURSIVE marks the whole clause; only a compound CTE with a
+	// self-referencing leg is actually recursive. A plain SELECT is just a CTE.
+	compound := cte.Query.Compound
+	if !cte.Recursive || compound == nil {
+		res, err := e.executeSelect(cte.Query)
+		if err != nil {
+			return err
+		}
+		e.registerCTEResult(cte.Name, cte.Columns, res)
+		return nil
+	}
+
+	anchor, err := e.executeSelect(compound.Left)
+	if err != nil {
+		return err
+	}
+	cols := cte.Columns
+	if len(cols) == 0 {
+		cols = anchor.Columns
+	}
+
+	accumulated := resultToCTERows(anchor, cols)
+	working := accumulated
+	distinct := compound.Op != parser.SetOpUnionAll
+	seen := make(map[string]bool)
+	if distinct {
+		for _, row := range accumulated {
+			seen[cteRowKey(row, cols)] = true
+		}
+	}
+
+	for i := 0; i < maxRecursiveCTEIterations; i++ {
+		e.cteTables[strings.ToLower(cte.Name)] = &cteTable{columns: cols, rows: working}
+		recursive, err := e.executeSelect(compound.Right)
+		if err != nil {
+			return err
+		}
+		fresh := make([]storage.Row, 0)
+		for _, row := range resultToCTERows(recursive, cols) {
+			if distinct {
+				key := cteRowKey(row, cols)
+				if seen[key] {
+					continue
+				}
+				seen[key] = true
+			}
+			fresh = append(fresh, row)
+		}
+		if len(fresh) == 0 {
+			break
+		}
+		accumulated = append(accumulated, fresh...)
+		working = fresh
+	}
+
+	e.cteTables[strings.ToLower(cte.Name)] = &cteTable{columns: cols, rows: accumulated}
+	return nil
+}
+
+// registerCTEResult stores a query result as a CTE. Declared column names win;
+// otherwise the result's own column names are used.
+func (e *Executor) registerCTEResult(name string, declared []string, res *Result) {
+	cols := declared
+	if len(cols) == 0 {
+		cols = res.Columns
+	}
+	e.cteTables[strings.ToLower(name)] = &cteTable{columns: cols, rows: resultToCTERows(res, cols)}
+}
+
+// resultToCTERows maps each result row to a storage.Row keyed by cols
+// positionally, which renames a recursive term's columns to the anchor's names.
+func resultToCTERows(res *Result, cols []string) []storage.Row {
+	rows := make([]storage.Row, 0, len(res.Rows))
+	for _, values := range res.Rows {
+		row := make(storage.Row, len(cols))
+		for i, name := range cols {
+			if i < len(values) {
+				row[name] = values[i]
+			}
+		}
+		rows = append(rows, row)
+	}
+	return rows
+}
+
+// cteRowsToValues converts a materialized CTE to positional row values.
+func cteRowsToValues(cte *cteTable) [][]interface{} {
+	values := make([][]interface{}, len(cte.rows))
+	for i, row := range cte.rows {
+		vals := make([]interface{}, len(cte.columns))
+		for j, col := range cte.columns {
+			vals[j] = row[col]
+		}
+		values[i] = vals
+	}
+	return values
+}
+
+// cteTableExists reports whether name is a materialized CTE.
+func (e *Executor) cteTableExists(name string) bool {
+	_, ok := e.cteTableFor(name)
+	return ok
+}
+
+// materializeJoinSubquery runs a derived table used on the right side of a JOIN
+// and returns its rows and schema.
+func (e *Executor) materializeJoinSubquery(ref *parser.TableRef) ([]storage.Row, *storage.Schema, error) {
+	res, err := e.executeSelect(ref.Subquery)
+	if err != nil {
+		return nil, nil, err
+	}
+	rows := make([]storage.Row, 0, len(res.Rows))
+	for _, values := range res.Rows {
+		row := make(storage.Row, len(res.Columns))
+		for i, col := range res.Columns {
+			if i < len(values) {
+				row[col] = values[i]
+			}
+		}
+		rows = append(rows, row)
+	}
+	return rows, schemaFromColumns(res.Columns), nil
+}
+
+// cloneRows returns a shallow copy of each row so callers cannot mutate a
+// materialized CTE's stored rows.
+func cloneRows(rows []storage.Row) []storage.Row {
+	out := make([]storage.Row, len(rows))
+	for i, r := range rows {
+		c := make(storage.Row, len(r))
+		for k, v := range r {
+			c[k] = v
+		}
+		out[i] = c
+	}
+	return out
+}
+
+func cteRowKey(row storage.Row, cols []string) string {
+	var b strings.Builder
+	for _, c := range cols {
+		fmt.Fprintf(&b, "%v\x00", row[c])
+	}
+	return b.String()
+}

pkg/executor/cte_test.go

@@ -0,0 +1,87 @@
+package executor
+
+import "testing"
+
+// TestNonRecursiveCTE reproduces the shape of Vikunja's project-permission
+// query: a CTE with a column list over a UNION ALL subquery, joined back to a
+// real table and grouped.
+func TestNonRecursiveCTE(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+
+	execMust(t, e, "CREATE TABLE projects (id INTEGER PRIMARY KEY, owner_id INTEGER)")
+	execMust(t, e, "CREATE TABLE users_projects (project_id INTEGER, user_id INTEGER, permission INTEGER)")
+	execMust(t, e, "CREATE TABLE team_projects (project_id INTEGER, team_id INTEGER, permission INTEGER)")
+	execMust(t, e, "CREATE TABLE team_members (team_id INTEGER, user_id INTEGER)")
+	execMust(t, e, "CREATE TABLE project_ancestors (ancestor_id INTEGER, project_id INTEGER, depth INTEGER)")
+
+	execMust(t, e, "INSERT INTO projects VALUES (1, 3)")
+	execMust(t, e, "INSERT INTO projects VALUES (2, 9)")
+	execMust(t, e, "INSERT INTO projects VALUES (3, 9)")
+	execMust(t, e, "INSERT INTO users_projects VALUES (2, 3, 1)")
+	execMust(t, e, "INSERT INTO team_projects VALUES (3, 7, 2)")
+	execMust(t, e, "INSERT INTO team_members VALUES (7, 3)")
+	execMust(t, e, "INSERT INTO project_ancestors VALUES (1, 1, 0)")
+	execMust(t, e, "INSERT INTO project_ancestors VALUES (2, 2, 0)")
+	execMust(t, e, "INSERT INTO project_ancestors VALUES (3, 3, 0)")
+
+	res := execMust(t, e, `WITH grants (project_id, permission) AS (
+		SELECT project_id, MAX(permission) FROM (
+			SELECT id AS project_id, 2 AS permission FROM projects WHERE owner_id = 3
+			UNION ALL SELECT project_id, permission FROM users_projects WHERE user_id = 3
+			UNION ALL SELECT tp.project_id, tp.permission
+				FROM team_projects tp INNER JOIN team_members tm ON tm.team_id = tp.team_id
+				WHERE tm.user_id = 3
+		) direct_grants GROUP BY project_id
+	)
+	SELECT pa.project_id AS id, MAX(g.permission) AS permission
+	FROM grants g INNER JOIN project_ancestors pa ON pa.ancestor_id = g.project_id
+	GROUP BY pa.project_id`)
+
+	if res.RowCount != 3 {
+		t.Fatalf("expected 3 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+	got := map[int64]int64{}
+	for _, row := range res.Rows {
+		got[row[0].(int64)] = row[1].(int64)
+	}
+	want := map[int64]int64{1: 2, 2: 1, 3: 2}
+	for id, perm := range want {
+		if got[id] != perm {
+			t.Fatalf("permission for project %d = %v, want %v (all: %v)", id, got[id], perm, got)
+		}
+	}
+}
+
+// TestCTEChained verifies a CTE that references an earlier CTE.
+func TestCTEChained(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
+	execMust(t, e, "INSERT INTO t VALUES (1, 10)")
+	execMust(t, e, "INSERT INTO t VALUES (2, 20)")
+
+	res := execMust(t, e, `WITH base AS (SELECT id, v FROM t WHERE v > 5),
+		doubled AS (SELECT id, v * 2 AS d FROM base)
+		SELECT id, d FROM doubled ORDER BY id`)
+	if res.RowCount != 2 {
+		t.Fatalf("expected 2 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+	if res.Rows[0][1] != int64(20) || res.Rows[1][1] != int64(40) {
+		t.Fatalf("unexpected rows: %v", res.Rows)
+	}
+}
+
+// TestRecursiveCTENonCompound verifies a plain CTE under WITH RECURSIVE is
+// treated as non-recursive.
+func TestRecursiveCTENonCompound(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
+	execMust(t, e, "INSERT INTO t VALUES (1, 10)")
+	execMust(t, e, "INSERT INTO t VALUES (2, 20)")
+	res := execMust(t, e, "WITH RECURSIVE r AS (SELECT id, v FROM t) SELECT id, v FROM r ORDER BY id")
+	if res.RowCount != 2 {
+		t.Fatalf("expected 2 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+}

pkg/executor/executor.go

@@ -49,6 +49,11 @@ type Executor struct {
 	// In-memory view registry: view name (lowercase) → SELECT AST.
 	views map[string]*parser.SelectStmt
 
+	// Materialized common table expressions for the current query, keyed by
+	// lowercased CTE name. Recursive CTEs populate this during fixpoint
+	// iteration so their own legs can read the working set.
+	cteTables map[string]*cteTable
+
 	// Session-local SQLite compatibility state, tracked per connection so
 	// last_insert_rowid()/changes()/total_changes() reflect this session only.
 	lastInsertRowID int64 // rowid of the most recent successful INSERT
@@ -308,6 +313,9 @@ func isCountStarSingleTable(stmt *parser.SelectStmt) bool {
 
 // executeSelect executes a SELECT statement (or compound SELECT).
 func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
+	if len(stmt.With) > 0 {
+		return e.executeWith(stmt)
+	}
 	if stmt.Compound != nil {
 		return e.executeCompound(stmt.Compound)
 	}
@@ -323,6 +331,11 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 
 	tableName := stmt.From[0].Name
 
+	// A materialized CTE is served from memory, not from storage.
+	if cte, ok := e.cteTableFor(tableName); ok {
+		return e.executeSelectOnMaterialized(stmt, cte.columns, cteRowsToValues(cte))
+	}
+
 	// Transparently expand view references as derived-table subqueries.
 	if viewDef, ok := e.views[strings.ToLower(tableName)]; ok {
 		alias := stmt.From[0].Alias
@@ -420,6 +433,22 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		effectiveWhere = nil
 	}
 
+	// Pre-evaluate non-correlated subqueries in the WHERE before the scan takes
+	// the session lock.
+	e.primeSubqueries(effectiveWhere)
+
+	// A WHERE that contains a subquery is evaluated after the scan. Evaluating a
+	// correlated subquery inside the locked scan re-enters the session lock and
+	// deadlocks, so only a subquery-free predicate is pushed into the scan.
+	whereHasSubquery := false
+	for _, ref := range collectColumnRefs(effectiveWhere) {
+		if ref == "__subquery__" {
+			whereHasSubquery = true
+			break
+		}
+	}
+	appliedInScan := effectiveWhere != nil && !whereHasSubquery && stmt.From[0].Alias == "" && !isMultiTable && stmt.From[0].Join == nil
+
 	if effectiveWhere != nil && !isMultiTable {
 		// Check if we can use an index
 		colName, colValue, isEquality := e.extractIndexableCondition(stmt.Where)
@@ -459,7 +488,7 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 	if !usedIndex {
 		var filterErr error
 		var filter func(storage.Row) bool
-		if effectiveWhere != nil && stmt.From[0].Alias == "" && !isMultiTable && stmt.From[0].Join == nil {
+		if appliedInScan {
 			filter = func(row storage.Row) bool {
 				val, ferr := e.evalExpr(effectiveWhere, row)
 				if ferr != nil {
@@ -494,10 +523,11 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		}
 	}
 
-	// Apply WHERE for single-table with alias (after alias mapping so alias.col refs work).
-	// A query with a JOIN defers WHERE until after the join, because the WHERE may
+	// Apply WHERE for a single-table scan that did not push the predicate into
+	// the locked scan (aliased tables, or a WHERE containing a subquery). A
+	// query with a JOIN defers WHERE until after the join, because the WHERE may
 	// reference columns from the joined table.
-	if effectiveWhere != nil && stmt.From[0].Alias != "" && !isMultiTable && stmt.From[0].Join == nil {
+	if effectiveWhere != nil && !isMultiTable && stmt.From[0].Join == nil && !appliedInScan {
 		var filterErr error
 		var filtered []storage.Row
 		for _, row := range rows {
@@ -1165,8 +1195,15 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		result.ColumnTypes = selectColumnTypes(stmt, schema)
 	}
 
+	// Window functions in the projection are evaluated over the scanned rows
+	// before their per-row values are read.
+	windowValues, werr := e.computeWindowValues(stmt, rows)
+	if werr != nil {
+		return nil, werr
+	}
+
 	// Add rows - evaluate each select expression
-	for _, row := range rows {
+	for rowIdx, row := range rows {
 		values := make([]interface{}, 0)
 		for _, col := range stmt.Columns {
 			if col.Star {
@@ -1205,6 +1242,8 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 					}
 					values = append(values, val)
 				}
+			} else if we, ok := col.Expr.(*parser.WindowExpr); ok {
+				values = append(values, windowValues[we][rowIdx])
 			} else {
 				// Evaluate the expression
 				val, err := e.evalExpr(col.Expr, row)
@@ -1374,18 +1413,27 @@ func (e *Executor) executeSelectFromSubquery(stmt *parser.SelectStmt) (*Result,
 		return nil, fmt.Errorf("subquery error: %w", err)
 	}
 
-	// Convert subquery result to rows for further processing
-	derivedRows := make([]storage.Row, 0, subqueryResult.RowCount)
-	for _, rowValues := range subqueryResult.Rows {
-		row := make(storage.Row)
-		for i, col := range subqueryResult.Columns {
-			row[col] = rowValues[i]
+	return e.executeSelectOnMaterialized(stmt, subqueryResult.Columns, subqueryResult.Rows)
+}
+
+// executeSelectOnMaterialized runs a SELECT whose FROM[0] is already
+// materialized as (columns, rowValues). It is shared by derived tables and
+// materialized CTEs.
+func (e *Executor) executeSelectOnMaterialized(stmt *parser.SelectStmt, columns []string, rowValues [][]interface{}) (*Result, error) {
+	derivedRows := make([]storage.Row, 0, len(rowValues))
+	for _, values := range rowValues {
+		row := make(storage.Row, len(columns))
+		for i, col := range columns {
+			if i < len(values) {
+				row[col] = values[i]
+			}
 		}
 		derivedRows = append(derivedRows, row)
 	}
 
 	// Handle JOINs if present
 	if stmt.From[0].Join != nil {
+		var err error
 		derivedRows, err = e.executeJoin(stmt.From[0], derivedRows)
 		if err != nil {
 			return nil, err
@@ -1409,87 +1457,82 @@ func (e *Executor) executeSelectFromSubquery(stmt *parser.SelectStmt) (*Result,
 
 	// Handle GROUP BY
 	if len(stmt.GroupBy) > 0 {
-		// Create a temporary schema from subquery columns
-		tempSchema := &storage.Schema{
-			Name:    "derived",
-			Columns: make([]storage.Column, len(subqueryResult.Columns)),
-		}
-		for i, col := range subqueryResult.Columns {
-			tempSchema.Columns[i] = storage.Column{
-				Name: col,
-				Type: "ANY",
-			}
-		}
-		return e.executeGroupBy(stmt, derivedRows, tempSchema)
+		return e.executeGroupBy(stmt, derivedRows, schemaFromColumns(columns))
 	}
 
 	// Check for aggregate functions without GROUP BY
-	hasAggregate := e.hasAggregates(stmt.Columns)
-	if hasAggregate {
-		tempSchema := &storage.Schema{
-			Name:    "derived",
-			Columns: make([]storage.Column, len(subqueryResult.Columns)),
-		}
-		for i, col := range subqueryResult.Columns {
-			tempSchema.Columns[i] = storage.Column{
-				Name: col,
-				Type: "ANY",
-			}
-		}
-		return e.executeAggregateSelect(stmt, derivedRows, tempSchema)
+	if e.hasAggregates(stmt.Columns) {
+		return e.executeAggregateSelect(stmt, derivedRows, schemaFromColumns(columns))
 	}
 
 	// Apply ORDER BY, LIMIT, and OFFSET.
 	derivedRows = e.orderAndLimitRows(derivedRows, stmt.OrderBy, stmt.Limit, stmt.Offset, stmt.Columns)
 
-	// Build result
+	// Build result. A projection may mix * and expressions, so each column is
+	// expanded independently.
 	result := NewResult("SELECT")
-
-	// Determine output columns
-	if stmt.Columns[0].Star {
-		// SELECT * from derived table
-		for _, col := range subqueryResult.Columns {
-			result.AddColumn(col)
-		}
-	} else {
-		// Specific columns
-		for _, col := range stmt.Columns {
-			if col.Alias != "" {
-				result.AddColumn(col.Alias)
-			} else if colRef, ok := col.Expr.(*parser.ColumnRef); ok {
+	for _, col := range stmt.Columns {
+		switch {
+		case col.Star:
+			for _, c := range columns {
+				result.AddColumn(c)
+			}
+		case col.Alias != "":
+			result.AddColumn(col.Alias)
+		case col.Expr != nil:
+			if colRef, ok := col.Expr.(*parser.ColumnRef); ok {
 				result.AddColumn(colRef.Column)
 			} else {
 				result.AddColumn("column")
 			}
+		default:
+			result.AddColumn("column")
 		}
 	}
 
-	// Add rows
-	for _, row := range derivedRows {
-		if stmt.Columns[0].Star {
-			// SELECT * - use all columns
-			values := make([]interface{}, len(subqueryResult.Columns))
-			for i, col := range subqueryResult.Columns {
-				values[i] = row[col]
-			}
-			result.AddRow(values...)
-		} else {
-			// Specific columns - evaluate expressions
-			values := make([]interface{}, len(stmt.Columns))
-			for i, col := range stmt.Columns {
+	// Window functions in the projection are evaluated over the materialized
+	// rows before the values are read.
+	windowValues, err := e.computeWindowValues(stmt, derivedRows)
+	if err != nil {
+		return nil, err
+	}
+
+	for rowIdx, row := range derivedRows {
+		values := make([]interface{}, 0, len(stmt.Columns))
+		for _, col := range stmt.Columns {
+			switch {
+			case col.Star:
+				for _, c := range columns {
+					values = append(values, row[c])
+				}
+			case col.Expr != nil:
+				if we, ok := col.Expr.(*parser.WindowExpr); ok {
+					values = append(values, windowValues[we][rowIdx])
+					continue
+				}
 				val, err := e.evalExpr(col.Expr, row)
 				if err != nil {
 					return nil, err
 				}
-				values[i] = val
+				values = append(values, val)
 			}
-			result.AddRow(values...)
 		}
+		result.AddRow(values...)
 	}
 
 	return result, nil
 }
 
+// schemaFromColumns builds a schema with ANY-typed columns for materialized
+// derived tables and CTEs.
+func schemaFromColumns(columns []string) *storage.Schema {
+	schema := &storage.Schema{Name: "derived", Columns: make([]storage.Column, len(columns))}
+	for i, col := range columns {
+		schema.Columns[i] = storage.Column{Name: col, Type: "ANY"}
+	}
+	return schema
+}
+
 // executeAggregateSelect executes a SELECT with aggregate functions.
 func (e *Executor) executeAggregateSelect(stmt *parser.SelectStmt, rows []storage.Row, schema *storage.Schema) (*Result, error) {
 	result := NewResult("SELECT")
@@ -1860,36 +1903,50 @@ func (e *Executor) executeJoinsWithMode(tableRef parser.TableRef, leftRows []sto
 		Condition: tableRef.Join.Condition,
 	}
 	leftKey, rightKey, canHash := extractEqualityJoinKeys(tableRef.Join.Condition, syntheticLeft, syntheticJoin)
-	rightSchema, err := e.schema.GetSchema(rightTable)
-	if err != nil {
-		return nil, err
-	}
 	var rightRows []storage.Row
-	if canHash && strings.EqualFold(rightSchema.PrimaryKey, rightKey) && len(leftRows) <= 256 {
-		seen := make(map[string]bool, len(leftRows))
-		for _, left := range leftRows {
-			key := joinKeyString(left, leftKey)
-			if key == "\x00" || seen[key] {
-				continue
-			}
-			seen[key] = true
-			row, getErr := e.session.GetByPK(rightTable, key)
-			if getErr == storage.ErrKeyNotFound {
-				continue
-			}
-			if getErr != nil {
-				return nil, getErr
-			}
-			normalizeRowBySchema(row, rightSchema)
-			rightRows = append(rightRows, row)
+	var rightSchema *storage.Schema
+	var err error
+	switch {
+	case rightTableRef.Subquery != nil:
+		rightRows, rightSchema, err = e.materializeJoinSubquery(rightTableRef)
+		if err != nil {
+			return nil, err
 		}
-	} else {
-		rightRows, err = e.session.Select(rightTable, nil)
+	case e.cteTableExists(rightTable):
+		cte, _ := e.cteTableFor(rightTable)
+		rightRows = cloneRows(cte.rows)
+		rightSchema = schemaFromColumns(cte.columns)
+	default:
+		rightSchema, err = e.schema.GetSchema(rightTable)
 		if err != nil {
 			return nil, err
 		}
-		for _, row := range rightRows {
-			normalizeRowBySchema(row, rightSchema)
+		if canHash && strings.EqualFold(rightSchema.PrimaryKey, rightKey) && len(leftRows) <= 256 {
+			seen := make(map[string]bool, len(leftRows))
+			for _, left := range leftRows {
+				key := joinKeyString(left, leftKey)
+				if key == "\x00" || seen[key] {
+					continue
+				}
+				seen[key] = true
+				row, getErr := e.session.GetByPK(rightTable, key)
+				if getErr == storage.ErrKeyNotFound {
+					continue
+				}
+				if getErr != nil {
+					return nil, getErr
+				}
+				normalizeRowBySchema(row, rightSchema)
+				rightRows = append(rightRows, row)
+			}
+		} else {
+			rightRows, err = e.session.Select(rightTable, nil)
+			if err != nil {
+				return nil, err
+			}
+			for _, row := range rightRows {
+				normalizeRowBySchema(row, rightSchema)
+			}
 		}
 	}
 
@@ -1986,7 +2043,17 @@ func (e *Executor) executeJoin(tableRef parser.TableRef, leftRows []storage.Row)
 	}
 
 	rightTable := join.Table.Name
-	rightRows, err := e.session.Select(rightTable, nil)
+	var rightRows []storage.Row
+	var err error
+	switch {
+	case join.Table.Subquery != nil:
+		rightRows, _, err = e.materializeJoinSubquery(join.Table)
+	case e.cteTableExists(rightTable):
+		cte, _ := e.cteTableFor(rightTable)
+		rightRows = cloneRows(cte.rows)
+	default:
+		rightRows, err = e.session.Select(rightTable, nil)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -2266,6 +2333,66 @@ func (e *Executor) addTableAlias(row storage.Row, alias string) storage.Row {
 	return result
 }
 
+// isConflictError reports whether an insert failed because of a primary-key or
+// unique-index conflict.
+func isConflictError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	return strings.Contains(msg, "duplicate") || strings.Contains(msg, "UNIQUE constraint failed")
+}
+
+// findConflictRow returns the first durable row whose target columns match the
+// inserted row.
+func (e *Executor) findConflictRow(table string, row storage.Row, columns []string) (storage.Row, bool) {
+	rows, err := e.session.Select(table, func(existing storage.Row) bool {
+		for _, col := range columns {
+			if compare(existing[col], row[col]) != 0 {
+				return false
+			}
+		}
+		return true
+	})
+	if err != nil || len(rows) == 0 {
+		return nil, false
+	}
+	return rows[0], true
+}
+
+// applyUpsert implements ON CONFLICT (target) DO UPDATE SET ... for primary-key
+// and unique-index conflicts, resolving excluded.<col> to the new row.
+func (e *Executor) applyUpsert(table string, schema *storage.Schema, existing, row storage.Row, assignments []parser.Assignment) error {
+	context := make(storage.Row)
+	for k, v := range existing {
+		context[k] = v
+		context[table+"."+k] = v
+	}
+	for k, v := range row {
+		context["excluded."+k] = v
+	}
+
+	pk := fmt.Sprintf("%v", existing[schema.PrimaryKey])
+	_, updated, err := e.session.UpdateByPK(table, pk, func(storage.Row) (storage.Row, error) {
+		updates := make(storage.Row)
+		for _, assignment := range assignments {
+			value, evalErr := e.evalExpr(assignment.Value, context)
+			if evalErr != nil {
+				return nil, evalErr
+			}
+			updates[assignment.Column] = value
+		}
+		return updates, nil
+	})
+	if err != nil {
+		return err
+	}
+	if !updated {
+		return fmt.Errorf("ON CONFLICT row disappeared during update")
+	}
+	return nil
+}
+
 // executeInsert executes an INSERT statement.
 func (e *Executor) executeInsert(stmt *parser.InsertStmt) (*Result, error) {
 	tableName := stmt.Table.Name
@@ -2355,41 +2482,30 @@ func (e *Executor) executeInsert(stmt *parser.InsertStmt) (*Result, error) {
 				}
 			}
 
-			rowID, err := e.session.InsertWithRowID(tableName, row)
-			if err != nil {
-				if strings.Contains(err.Error(), "duplicate") && (stmt.ConflictDoNothing || len(stmt.ConflictUpdate) > 0) {
+			// ON CONFLICT is resolved before inserting. A unique-index conflict
+			// may only surface at transaction commit, so the insert error is not
+			// a reliable signal.
+			if stmt.ConflictDoNothing || len(stmt.ConflictUpdate) > 0 {
+				columns := stmt.ConflictTarget
+				if len(columns) == 0 {
+					columns = []string{schema.PrimaryKey}
+				}
+				if existing, found := e.findConflictRow(tableName, row, columns); found {
 					if stmt.ConflictDoNothing {
 						continue
 					}
-					if len(stmt.ConflictTarget) > 0 && !containsFold(stmt.ConflictTarget, schema.PrimaryKey) {
-						return fmt.Errorf("ON CONFLICT target must include primary key %s", schema.PrimaryKey)
-					}
-					pkValue := row[schema.PrimaryKey]
-					updated, updateErr := e.session.UpdateFunc(tableName, func(existing storage.Row) (storage.Row, error) {
-						context := e.addTableAlias(existing, tableName)
-						updates := make(storage.Row)
-						for _, assignment := range stmt.ConflictUpdate {
-							value, evalErr := e.evalExpr(assignment.Value, context)
-							if evalErr != nil {
-								return nil, evalErr
-							}
-							updates[assignment.Column] = value
-						}
-						return updates, nil
-					}, func(existing storage.Row) bool {
-						return fmt.Sprintf("%v", existing[schema.PrimaryKey]) == fmt.Sprintf("%v", pkValue)
-					})
-					if updateErr != nil {
-						return updateErr
-					}
-					if updated != 1 {
-						return fmt.Errorf("ON CONFLICT row disappeared during update")
+					if err := e.applyUpsert(tableName, schema, existing, row, stmt.ConflictUpdate); err != nil {
+						return err
 					}
 					count++
 					continue
 				}
+			}
+
+			rowID, err := e.session.InsertWithRowID(tableName, row)
+			if err != nil {
 				// Handle conflict based on OnConflict action
-				if strings.Contains(err.Error(), "duplicate") {
+				if isConflictError(err) {
 					switch stmt.OnConflict {
 					case parser.ConflictIgnore:
 						// Silently ignore the duplicate
@@ -5227,6 +5343,10 @@ func (e *Executor) hasAggregates(columns []parser.SelectColumn) bool {
 }
 
 func (e *Executor) isAggregate(expr parser.Expr) bool {
+	if _, ok := expr.(*parser.WindowExpr); ok {
+		// Window functions are computed after grouping; they are not aggregates.
+		return false
+	}
 	if fn, ok := expr.(*parser.FunctionCall); ok {
 		name := strings.ToUpper(fn.Name)
 		switch name {
@@ -5298,7 +5418,7 @@ func resolveOrderByPositions(orderBy []parser.OrderByItem, selectCols []parser.S
 			if pos, err := strconv.Atoi(lit.Value); err == nil && pos >= 1 && pos <= len(selectCols) {
 				col := selectCols[pos-1]
 				if col.Expr != nil {
-					result[i] = parser.OrderByItem{Expr: col.Expr, Desc: item.Desc}
+					result[i] = parser.OrderByItem{Expr: col.Expr, Desc: item.Desc, NullsOrder: item.NullsOrder}
 					continue
 				}
 			}
@@ -5312,7 +5432,26 @@ func resolveOrderByPositions(orderBy []parser.OrderByItem, selectCols []parser.S
 // Keys are precomputed per-row ORDER BY expression values, one per item.
 func orderByLess(a, b []interface{}, orderBy []parser.OrderByItem) bool {
 	for i, item := range orderBy {
-		cmp := compare(a[i], b[i])
+		av, bv := a[i], b[i]
+		if av == nil || bv == nil {
+			if av == nil && bv == nil {
+				continue
+			}
+			// SQLite treats NULL as smaller than any value, so the default puts
+			// NULLs first for ASC and last for DESC. NULLS FIRST/LAST overrides it.
+			nullsFirst := !item.Desc
+			switch item.NullsOrder {
+			case parser.NullsFirst:
+				nullsFirst = true
+			case parser.NullsLast:
+				nullsFirst = false
+			}
+			if av == nil {
+				return nullsFirst
+			}
+			return !nullsFirst
+		}
+		cmp := compare(av, bv)
 		if cmp != 0 {
 			if item.Desc {
 				return cmp > 0

pkg/executor/in_single_test.go

@@ -0,0 +1,14 @@
+package executor
+
+import "testing"
+
+func TestInSingleValueNoMatch(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE unsplash_photos (id INTEGER PRIMARY KEY, file_id INTEGER)")
+	execMust(t, e, "INSERT INTO unsplash_photos VALUES (1, 5)")
+	res := execMust(t, e, "SELECT id, file_id FROM unsplash_photos WHERE file_id IN (0)")
+	if res.RowCount != 0 {
+		t.Fatalf("expected 0 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+}

pkg/executor/orderby_nulls_test.go

@@ -0,0 +1,46 @@
+package executor
+
+import "testing"
+
+func TestOrderByNullsFirstLast(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, pos INTEGER)")
+	execMust(t, e, "INSERT INTO t VALUES (1, NULL)")
+	execMust(t, e, "INSERT INTO t VALUES (2, 5)")
+	execMust(t, e, "INSERT INTO t VALUES (3, 1)")
+	execMust(t, e, "INSERT INTO t VALUES (4, NULL)")
+
+	ids := func(res *Result) []int64 {
+		out := make([]int64, len(res.Rows))
+		for i, row := range res.Rows {
+			out[i] = row[0].(int64)
+		}
+		return out
+	}
+
+	got := ids(execMust(t, e, "SELECT id FROM t ORDER BY pos ASC NULLS LAST, id ASC"))
+	want := []int64{3, 2, 1, 4}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("NULLS LAST order = %v, want %v", got, want)
+		}
+	}
+
+	got = ids(execMust(t, e, "SELECT id FROM t ORDER BY pos ASC NULLS FIRST, id ASC"))
+	want = []int64{1, 4, 3, 2}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("NULLS FIRST order = %v, want %v", got, want)
+		}
+	}
+
+	// Default SQLite ordering: NULLs are smallest.
+	got = ids(execMust(t, e, "SELECT id FROM t ORDER BY pos ASC, id ASC"))
+	want = []int64{1, 4, 3, 2}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("default ASC order = %v, want %v", got, want)
+		}
+	}
+}

pkg/executor/recursive_cte_test.go

@@ -0,0 +1,107 @@
+package executor
+
+import "testing"
+
+func TestRecursiveCTE(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, parent_id INTEGER)")
+	execMust(t, e, "INSERT INTO t VALUES (1, NULL)")
+	execMust(t, e, "INSERT INTO t VALUES (2, 1)")
+	execMust(t, e, "INSERT INTO t VALUES (3, 2)")
+	execMust(t, e, "INSERT INTO t VALUES (4, NULL)")
+
+	res := execMust(t, e, `WITH RECURSIVE hier AS (
+		SELECT id, parent_id, 0 AS level, id AS root FROM t WHERE id = 3
+		UNION ALL
+		SELECT t.id, t.parent_id, h.level + 1, h.root FROM t INNER JOIN hier h ON t.id = h.parent_id
+	)
+	SELECT id, level, root FROM hier ORDER BY id`)
+
+	if res.RowCount != 3 {
+		t.Fatalf("expected 3 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+	want := map[int64]int64{3: 0, 2: 1, 1: 2}
+	for _, row := range res.Rows {
+		id := row[0].(int64)
+		if row[1].(int64) != want[id] {
+			t.Fatalf("id %d level = %v, want %v", id, row[1], want[id])
+		}
+		if row[2].(int64) != 3 {
+			t.Fatalf("id %d root = %v, want 3", id, row[2])
+		}
+	}
+}
+
+func TestRowNumberWindow(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE s (id INTEGER PRIMARY KEY, grp INTEGER, val INTEGER)")
+	execMust(t, e, "INSERT INTO s VALUES (1, 1, 10)")
+	execMust(t, e, "INSERT INTO s VALUES (2, 1, 5)")
+	execMust(t, e, "INSERT INTO s VALUES (3, 2, 7)")
+	execMust(t, e, "INSERT INTO s VALUES (4, 2, 7)")
+
+	res := execMust(t, e, "SELECT id, ROW_NUMBER() OVER (PARTITION BY grp ORDER BY val) AS rn FROM s ORDER BY id")
+	want := map[int64]int64{1: 2, 2: 1, 3: 1, 4: 2}
+	if res.RowCount != 4 {
+		t.Fatalf("expected 4 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+	for _, row := range res.Rows {
+		id := row[0].(int64)
+		if row[1].(int64) != want[id] {
+			t.Fatalf("id %d rn = %v, want %v (rows %v)", id, row[1], want[id], res.Rows)
+		}
+	}
+}
+
+// TestRecursiveCTEWithWindow reproduces the combined shape of Vikunja's
+// subscription query: a recursive CTE, a dependent CTE, and a windowed derived
+// table joined back to a real table.
+func TestRecursiveCTEWithWindow(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE projects (id INTEGER PRIMARY KEY, parent_project_id INTEGER)")
+	execMust(t, e, "CREATE TABLE subscriptions (id INTEGER PRIMARY KEY, entity_type INTEGER, entity_id INTEGER, user_id INTEGER, muted INTEGER)")
+	execMust(t, e, "CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)")
+	execMust(t, e, "INSERT INTO projects VALUES (1, NULL)")
+	execMust(t, e, "INSERT INTO projects VALUES (2, 1)")
+	execMust(t, e, "INSERT INTO projects VALUES (3, 2)")
+	execMust(t, e, "INSERT INTO subscriptions VALUES (1, 1, 1, 10, 0)")
+	execMust(t, e, "INSERT INTO subscriptions VALUES (2, 1, 3, 20, 0)")
+	execMust(t, e, "INSERT INTO users VALUES (10, 'alice')")
+	execMust(t, e, "INSERT INTO users VALUES (20, 'bob')")
+
+	res := execMust(t, e, `WITH RECURSIVE project_hierarchy AS (
+		SELECT id, parent_project_id, 0 AS level, id AS original_project_id FROM projects WHERE id IN (3)
+		UNION ALL
+		SELECT p.id, p.parent_project_id, ph.level + 1, ph.original_project_id
+		FROM projects p INNER JOIN project_hierarchy ph ON p.id = ph.parent_project_id
+	),
+	subscription_hierarchy AS (
+		SELECT s.id, s.entity_type, s.entity_id, s.user_id, s.muted,
+			CASE WHEN s.entity_id = ph.original_project_id THEN 1 ELSE ph.level + 1 END AS priority,
+			ph.original_project_id
+		FROM subscriptions s INNER JOIN project_hierarchy ph ON s.entity_id = ph.id
+		WHERE s.entity_type = 1
+	)
+	SELECT p.id AS original_entity_id, sh.id AS subscription_id, sh.user_id
+	FROM projects p
+		LEFT JOIN (
+			SELECT *, ROW_NUMBER() OVER (PARTITION BY original_project_id, user_id ORDER BY priority) AS rn
+			FROM subscription_hierarchy
+		) sh ON p.id = sh.original_project_id AND sh.rn = 1
+	WHERE p.id IN (3)
+	ORDER BY p.id, sh.user_id`)
+
+	if res.RowCount != 2 {
+		t.Fatalf("expected 2 rows, got %d: %v", res.RowCount, res.Rows)
+	}
+	got := map[int64]bool{}
+	for _, row := range res.Rows {
+		got[row[1].(int64)] = true
+	}
+	if !got[1] || !got[2] {
+		t.Fatalf("expected subscriptions 1 and 2, got %v", res.Rows)
+	}
+}

pkg/executor/upsert_test.go

@@ -0,0 +1,31 @@
+package executor
+
+import "testing"
+
+// TestOnConflictUniqueIndexUpsert reproduces Vikunja's task_buckets upsert:
+// INSERT ... ON CONFLICT (a, b) DO UPDATE SET col = excluded.col against a
+// unique index that is not the primary key.
+func TestOnConflictUniqueIndexUpsert(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE task_buckets (id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER, project_view_id INTEGER, bucket_id INTEGER)")
+	execMust(t, e, "CREATE UNIQUE INDEX uq_tb ON task_buckets (task_id, project_view_id)")
+	execMust(t, e, "INSERT INTO task_buckets (task_id, project_view_id, bucket_id) VALUES (1, 12, 9)")
+
+	execMust(t, e, "INSERT INTO task_buckets (task_id, project_view_id, bucket_id) VALUES (1, 12, 7) ON CONFLICT (task_id, project_view_id) DO UPDATE SET bucket_id = excluded.bucket_id")
+
+	res := execMust(t, e, "SELECT bucket_id FROM task_buckets WHERE task_id = 1 AND project_view_id = 12")
+	if res.RowCount != 1 {
+		t.Fatalf("expected 1 row, got %d: %v", res.RowCount, res.Rows)
+	}
+	if res.Rows[0][0] != int64(7) {
+		t.Fatalf("bucket_id = %v, want 7", res.Rows[0][0])
+	}
+
+	// A different (task_id, project_view_id) inserts a new row.
+	execMust(t, e, "INSERT INTO task_buckets (task_id, project_view_id, bucket_id) VALUES (2, 12, 5) ON CONFLICT (task_id, project_view_id) DO UPDATE SET bucket_id = excluded.bucket_id")
+	res = execMust(t, e, "SELECT count(*) FROM task_buckets")
+	if res.Rows[0][0] != int64(2) {
+		t.Fatalf("expected 2 rows, got %v", res.Rows[0][0])
+	}
+}

pkg/executor/window.go

@@ -0,0 +1,103 @@
+package executor
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// computeWindowValues evaluates each window function in a projection over the
+// given rows, returning one value per row for each window expression.
+func (e *Executor) computeWindowValues(stmt *parser.SelectStmt, rows []storage.Row) (map[*parser.WindowExpr][]interface{}, error) {
+	var windows []*parser.WindowExpr
+	for _, col := range stmt.Columns {
+		if we, ok := col.Expr.(*parser.WindowExpr); ok {
+			windows = append(windows, we)
+		}
+	}
+	if len(windows) == 0 {
+		return nil, nil
+	}
+
+	out := make(map[*parser.WindowExpr][]interface{}, len(windows))
+	for _, we := range windows {
+		values, err := e.evalWindowExpr(we, rows)
+		if err != nil {
+			return nil, err
+		}
+		out[we] = values
+	}
+	return out, nil
+}
+
+type windowRow struct {
+	idx int
+	key []interface{}
+}
+
+// evalWindowExpr computes ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...).
+// Other window functions are rejected rather than approximated.
+func (e *Executor) evalWindowExpr(we *parser.WindowExpr, rows []storage.Row) ([]interface{}, error) {
+	name := ""
+	if we.Func != nil {
+		name = strings.ToUpper(we.Func.Name)
+	}
+	if name != "ROW_NUMBER" {
+		return nil, fmt.Errorf("unsupported window function: %s", name)
+	}
+
+	partitions := map[string][]int{}
+	var order []string
+	for i, row := range rows {
+		key, err := e.windowPartitionKey(we, row)
+		if err != nil {
+			return nil, err
+		}
+		if _, ok := partitions[key]; !ok {
+			order = append(order, key)
+		}
+		partitions[key] = append(partitions[key], i)
+	}
+
+	values := make([]interface{}, len(rows))
+	for _, key := range order {
+		partition := make([]windowRow, 0, len(partitions[key]))
+		for _, idx := range partitions[key] {
+			wr := windowRow{idx: idx}
+			if len(we.OrderBy) > 0 {
+				wr.key = make([]interface{}, len(we.OrderBy))
+				for k, item := range we.OrderBy {
+					wr.key[k], _ = e.evalExpr(item.Expr, rows[idx])
+				}
+			}
+			partition = append(partition, wr)
+		}
+		if len(we.OrderBy) > 0 {
+			sort.SliceStable(partition, func(a, b int) bool {
+				return orderByLess(partition[a].key, partition[b].key, we.OrderBy)
+			})
+		}
+		for rank, wr := range partition {
+			values[wr.idx] = int64(rank + 1)
+		}
+	}
+	return values, nil
+}
+
+func (e *Executor) windowPartitionKey(we *parser.WindowExpr, row storage.Row) (string, error) {
+	if len(we.PartitionBy) == 0 {
+		return "", nil
+	}
+	var b strings.Builder
+	for _, p := range we.PartitionBy {
+		v, err := e.evalExpr(p, row)
+		if err != nil {
+			return "", err
+		}
+		fmt.Fprintf(&b, "%v\x00", v)
+	}
+	return b.String(), nil
+}

pkg/parser/ast.go

@@ -55,6 +55,18 @@ type SelectStmt struct {
 	Offset   Expr
 	// Compound chains a set operation onto this SELECT (UNION/INTERSECT/EXCEPT).
 	Compound *CompoundSelect
+	// With holds common table expressions that must be materialized before this
+	// SELECT runs. Non-recursive CTEs are desugared by the parser instead; this
+	// list carries recursive CTEs (and the CTEs that depend on them).
+	With []*CTE
+}
+
+// CTE is a common table expression from a WITH clause.
+type CTE struct {
+	Name      string
+	Columns   []string
+	Recursive bool
+	Query     *SelectStmt
 }
 
 func (s *SelectStmt) node()     {}
@@ -96,10 +108,20 @@ const (
 	JoinCross
 )
 
+// NullsOrder selects where NULLs sort in an ORDER BY item.
+type NullsOrder int
+
+const (
+	NullsDefault NullsOrder = iota // SQLite default: NULLs are smallest
+	NullsFirst
+	NullsLast
+)
+
 // OrderByItem represents an ORDER BY item.
 type OrderByItem struct {
-	Expr Expr
-	Desc bool
+	Expr       Expr
+	Desc       bool
+	NullsOrder NullsOrder
 }
 
 // ConflictAction represents the action to take on conflict.
@@ -441,6 +463,16 @@ type FunctionCall struct {
 func (e *FunctionCall) node()     {}
 func (e *FunctionCall) exprNode() {}
 
+// WindowExpr represents a function call with an OVER clause.
+type WindowExpr struct {
+	Func        *FunctionCall
+	PartitionBy []Expr
+	OrderBy     []OrderByItem
+}
+
+func (e *WindowExpr) node()     {}
+func (e *WindowExpr) exprNode() {}
+
 // SubqueryExpr represents a subquery expression.
 type SubqueryExpr struct {
 	Query *SelectStmt

pkg/parser/cte.go

@@ -0,0 +1,314 @@
+package parser
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+)
+
+// cteDef is a single common table expression parsed from a WITH clause.
+type cteDef struct {
+	name  string
+	cols  []string
+	query *SelectStmt
+}
+
+// isWithStart reports whether the current token begins a WITH clause. WITH is
+// not a lexer keyword (it can be a column or table name), so it is recognized
+// by its literal at statement start.
+func (p *Parser) isWithStart() bool {
+	return p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "WITH")
+}
+
+// parseWithStatement parses a WITH clause followed by a SELECT and desugars each
+// non-recursive CTE into a derived table. Downstream stages therefore only ever
+// see regular SELECTs. Recursive CTEs are rejected explicitly rather than being
+// silently mis-executed.
+func (p *Parser) parseWithStatement() (Statement, error) {
+	p.nextToken() // consume WITH
+
+	recursive := false
+	if p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "RECURSIVE") {
+		recursive = true
+		p.nextToken()
+	}
+
+	var ctes []*cteDef
+	for {
+		if !p.curTokenIs(lexer.TokenIdent) {
+			return nil, p.curError("expected CTE name")
+		}
+		cte := &cteDef{name: p.curToken.Literal}
+		p.nextToken()
+
+		if p.curTokenIs(lexer.TokenLParen) {
+			p.nextToken()
+			for {
+				if !p.curTokenIs(lexer.TokenIdent) {
+					return nil, p.curError("expected column name in CTE column list")
+				}
+				cte.cols = append(cte.cols, p.curToken.Literal)
+				p.nextToken()
+				if p.curTokenIs(lexer.TokenComma) {
+					p.nextToken()
+					continue
+				}
+				break
+			}
+			if !p.curTokenIs(lexer.TokenRParen) {
+				return nil, p.curError("expected ) after CTE column list")
+			}
+			p.nextToken()
+		}
+
+		if !p.curTokenIs(lexer.TokenAS) {
+			return nil, p.curError("expected AS in CTE definition")
+		}
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenLParen) {
+			return nil, p.curError("expected ( before CTE query")
+		}
+		p.nextToken()
+
+		query, err := p.parseSelect()
+		if err != nil {
+			return nil, err
+		}
+		if !p.curTokenIs(lexer.TokenRParen) {
+			return nil, p.curError("expected ) after CTE query")
+		}
+		p.nextToken()
+
+		// A recursive CTE's query is a compound (anchor UNION recursive), so its
+		// column names are applied at materialization time instead.
+		if !recursive {
+			if err := applyCTEColumnNames(cte, query); err != nil {
+				return nil, err
+			}
+		}
+		cte.query = query
+		ctes = append(ctes, cte)
+
+		if p.curTokenIs(lexer.TokenComma) {
+			p.nextToken()
+			continue
+		}
+		break
+	}
+
+	stmt, err := p.parseStatement()
+	if err != nil {
+		return nil, err
+	}
+
+	sel, ok := stmt.(*SelectStmt)
+	if !ok {
+		return nil, p.curError("WITH is only supported before a SELECT statement")
+	}
+
+	if recursive {
+		// Recursive CTEs are materialized by the executor, which needs the
+		// definitions; desugaring cannot express self-reference.
+		sel.With = make([]*CTE, 0, len(ctes))
+		for _, c := range ctes {
+			sel.With = append(sel.With, &CTE{
+				Name:      c.name,
+				Columns:   c.cols,
+				Recursive: true,
+				Query:     c.query,
+			})
+		}
+		return sel, nil
+	}
+
+	// Each CTE may reference the CTEs defined before it.
+	for i := range ctes {
+		if err := substituteSelectCTEs(ctes[i].query, ctes[:i]); err != nil {
+			return nil, err
+		}
+	}
+	if err := substituteSelectCTEs(sel, ctes); err != nil {
+		return nil, err
+	}
+	return sel, nil
+}
+
+// applyCTEColumnNames aliases the CTE query's projection columns with the names
+// given in the CTE column list so a derived-table reference exposes them.
+func applyCTEColumnNames(cte *cteDef, query *SelectStmt) error {
+	if len(cte.cols) == 0 {
+		return nil
+	}
+	if query.Compound != nil {
+		return fmt.Errorf("CTE %q: column list on a compound query is not supported", cte.name)
+	}
+	if len(cte.cols) > len(query.Columns) {
+		return fmt.Errorf("CTE %q: %d column names for %d columns", cte.name, len(cte.cols), len(query.Columns))
+	}
+	for i, name := range cte.cols {
+		query.Columns[i].Alias = name
+	}
+	return nil
+}
+
+// substituteSelectCTEs replaces every reference to a named CTE with a derived
+// table carrying that CTE's query. Substitution recurses through set operations,
+// derived tables, joins, and subquery expressions.
+func substituteSelectCTEs(sel *SelectStmt, ctes []*cteDef) error {
+	if sel == nil {
+		return nil
+	}
+	if sel.Compound != nil {
+		if err := substituteSelectCTEs(sel.Compound.Left, ctes); err != nil {
+			return err
+		}
+		if err := substituteSelectCTEs(sel.Compound.Right, ctes); err != nil {
+			return err
+		}
+	}
+	for i := range sel.From {
+		if err := substituteTableRefCTEs(&sel.From[i], ctes); err != nil {
+			return err
+		}
+	}
+	if err := substituteExprCTEs(sel.Where, ctes); err != nil {
+		return err
+	}
+	for i := range sel.Columns {
+		if err := substituteExprCTEs(sel.Columns[i].Expr, ctes); err != nil {
+			return err
+		}
+	}
+	for i := range sel.GroupBy {
+		if err := substituteExprCTEs(sel.GroupBy[i], ctes); err != nil {
+			return err
+		}
+	}
+	if err := substituteExprCTEs(sel.Having, ctes); err != nil {
+		return err
+	}
+	for i := range sel.OrderBy {
+		if err := substituteExprCTEs(sel.OrderBy[i].Expr, ctes); err != nil {
+			return err
+		}
+	}
+	if err := substituteExprCTEs(sel.Limit, ctes); err != nil {
+		return err
+	}
+	return substituteExprCTEs(sel.Offset, ctes)
+}
+
+func lookupCTE(name string, ctes []*cteDef) *cteDef {
+	for _, cte := range ctes {
+		if strings.EqualFold(cte.name, name) {
+			return cte
+		}
+	}
+	return nil
+}
+
+func substituteTableRefCTEs(ref *TableRef, ctes []*cteDef) error {
+	if ref == nil {
+		return nil
+	}
+	if ref.Subquery != nil {
+		if err := substituteSelectCTEs(ref.Subquery, ctes); err != nil {
+			return err
+		}
+	} else if cte := lookupCTE(ref.Name, ctes); cte != nil {
+		alias := ref.Alias
+		if alias == "" {
+			alias = cte.name
+		}
+		ref.Subquery = cte.query
+		ref.Name = ""
+		ref.Alias = alias
+	}
+	if ref.Join != nil {
+		return substituteJoinCTEs(ref.Join, ctes)
+	}
+	return nil
+}
+
+func substituteJoinCTEs(join *JoinClause, ctes []*cteDef) error {
+	if join == nil {
+		return nil
+	}
+	if err := substituteTableRefCTEs(join.Table, ctes); err != nil {
+		return err
+	}
+	return substituteExprCTEs(join.Condition, ctes)
+}
+
+func substituteExprCTEs(expr Expr, ctes []*cteDef) error {
+	if expr == nil {
+		return nil
+	}
+	switch e := expr.(type) {
+	case *InExpr:
+		if err := substituteExprCTEs(e.Left, ctes); err != nil {
+			return err
+		}
+		for _, v := range e.Values {
+			if err := substituteExprCTEs(v, ctes); err != nil {
+				return err
+			}
+		}
+		return substituteSelectCTEs(e.Subquery, ctes)
+	case *SubqueryExpr:
+		return substituteSelectCTEs(e.Query, ctes)
+	case *ExistsExpr:
+		return substituteSelectCTEs(e.Subquery, ctes)
+	case *BinaryExpr:
+		if err := substituteExprCTEs(e.Left, ctes); err != nil {
+			return err
+		}
+		return substituteExprCTEs(e.Right, ctes)
+	case *UnaryExpr:
+		return substituteExprCTEs(e.Operand, ctes)
+	case *BetweenExpr:
+		if err := substituteExprCTEs(e.Left, ctes); err != nil {
+			return err
+		}
+		if err := substituteExprCTEs(e.Low, ctes); err != nil {
+			return err
+		}
+		return substituteExprCTEs(e.High, ctes)
+	case *LikeExpr:
+		if err := substituteExprCTEs(e.Left, ctes); err != nil {
+			return err
+		}
+		if err := substituteExprCTEs(e.Pattern, ctes); err != nil {
+			return err
+		}
+		return substituteExprCTEs(e.Escape, ctes)
+	case *IsNullExpr:
+		return substituteExprCTEs(e.Left, ctes)
+	case *CaseExpr:
+		if err := substituteExprCTEs(e.Operand, ctes); err != nil {
+			return err
+		}
+		for _, w := range e.Whens {
+			if err := substituteExprCTEs(w.Condition, ctes); err != nil {
+				return err
+			}
+			if err := substituteExprCTEs(w.Result, ctes); err != nil {
+				return err
+			}
+		}
+		return substituteExprCTEs(e.Else, ctes)
+	case *FunctionCall:
+		for _, a := range e.Args {
+			if err := substituteExprCTEs(a, ctes); err != nil {
+				return err
+			}
+		}
+		return nil
+	case *ParenExpr:
+		return substituteExprCTEs(e.Expr, ctes)
+	case *CastExpr:
+		return substituteExprCTEs(e.Expr, ctes)
+	}
+	return nil
+}

pkg/parser/parser.go

@@ -106,6 +106,9 @@ func (p *Parser) curError(msg string) error {
 }
 
 func (p *Parser) parseStatement() (Statement, error) {
+	if p.isWithStart() {
+		return p.parseWithStatement()
+	}
 	switch p.curToken.Type {
 	case lexer.TokenSELECT:
 		return p.parseSelect()
@@ -709,6 +712,23 @@ func (p *Parser) parseOrderBy() ([]OrderByItem, error) {
 			p.nextToken()
 		}
 
+		// Optional NULLS FIRST / NULLS LAST.
+		if p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "NULLS") {
+			p.nextToken()
+			if !p.curTokenIs(lexer.TokenIdent) {
+				return nil, p.curError("expected FIRST or LAST after NULLS")
+			}
+			switch {
+			case strings.EqualFold(p.curToken.Literal, "FIRST"):
+				item.NullsOrder = NullsFirst
+			case strings.EqualFold(p.curToken.Literal, "LAST"):
+				item.NullsOrder = NullsLast
+			default:
+				return nil, p.curError("expected FIRST or LAST after NULLS")
+			}
+			p.nextToken()
+		}
+
 		items = append(items, item)
 
 		if !p.curTokenIs(lexer.TokenComma) {
@@ -2311,9 +2331,58 @@ func (p *Parser) parseFunctionCall(name string) (Expr, error) {
 	}
 	p.nextToken()
 
+	// Window function: func(...) OVER (PARTITION BY ... ORDER BY ...).
+	if p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "OVER") {
+		return p.parseWindowSpec(fn)
+	}
+
 	return fn, nil
 }
 
+// parseWindowSpec parses the OVER (...) clause of a window function. Only
+// PARTITION BY and ORDER BY are supported; frame clauses are not.
+func (p *Parser) parseWindowSpec(fn *FunctionCall) (Expr, error) {
+	p.nextToken() // consume OVER
+	if !p.curTokenIs(lexer.TokenLParen) {
+		return nil, p.curError("expected ( after OVER")
+	}
+	p.nextToken()
+
+	window := &WindowExpr{Func: fn}
+
+	if p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "PARTITION") {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenBY) {
+			return nil, p.curError("expected BY after PARTITION")
+		}
+		p.nextToken()
+		exprs, err := p.parseExprList()
+		if err != nil {
+			return nil, err
+		}
+		window.PartitionBy = exprs
+	}
+
+	if p.curTokenIs(lexer.TokenORDER) {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenBY) {
+			return nil, p.curError("expected BY after ORDER")
+		}
+		p.nextToken()
+		orderBy, err := p.parseOrderBy()
+		if err != nil {
+			return nil, err
+		}
+		window.OrderBy = orderBy
+	}
+
+	if !p.curTokenIs(lexer.TokenRParen) {
+		return nil, p.curError("expected ) after window specification")
+	}
+	p.nextToken()
+	return window, nil
+}
+
 func (p *Parser) parseCaseExpr() (Expr, error) {
 	expr := &CaseExpr{}