Merge remote-tracking branch 'origin/master' into engine/a-add-column-backfill

# Conflicts:
#	pkg/executor/alter_table_test.go

pkg/analyzer/analyzer.go

@@ -522,6 +522,19 @@ func (a *Analyzer) analyzeInsert(stmt *parser.InsertStmt) error {
 		}
 	}
 
+	// An ON CONFLICT target names columns of the target table. An unknown name
+	// must fail the statement rather than silently degrading to a scan. The
+	// hidden rowid aliases are valid targets unless a real column shadows them.
+	for _, colName := range stmt.ConflictTarget {
+		if _, ok := table.GetColumn(colName); ok || isRowIDAlias(colName) {
+			continue
+		}
+		return &AnalysisError{
+			Type:    ErrColumnNotFound,
+			Message: fmt.Sprintf("column not found: %s", colName),
+		}
+	}
+
 	// Validate VALUES
 	for _, row := range stmt.Values {
 		if len(row) != len(targetCols) {
@@ -994,6 +1007,10 @@ func (a *Analyzer) analyzeBinaryExpr(e *parser.BinaryExpr) (*ExprInfo, error) {
 	case lexer.TokenConcat:
 		// String concatenation
 		info.Type = TypeText
+	case lexer.TokenArrow, lexer.TokenArrowText:
+		// JSON extraction. Like json_extract, the result follows the JSON value
+		// and is not a fixed SQL type (a JSON number extracts as a number).
+		info.Type = TypeAny
 	default:
 		info.Type = TypeUnknown
 	}

pkg/analyzer/analyzer_test.go

@@ -383,6 +383,7 @@ func TestAnalyzeInsertErrors(t *testing.T) {
 	}{
 		{"table not found", "INSERT INTO nonexistent VALUES (1)", ErrTableNotFound},
 		{"column not found", "INSERT INTO users (nonexistent) VALUES (1)", ErrColumnNotFound},
+		{"unknown on conflict target", "INSERT INTO users (id, name) VALUES (1, 'a') ON CONFLICT (nonexistent) DO NOTHING", ErrColumnNotFound},
 		{"wrong column count", "INSERT INTO users (id, name) VALUES (1)", ErrTypeMismatch},
 	}
 

pkg/capabilities/capabilities.go

@@ -179,7 +179,7 @@ func Current() *Document {
 			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, and the column is published together with its UNIQUE index in one atomic schema write"},
 			DropColumn:             Feature{Supported: true, Status: "rewrites_rows"},
 			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows"},
-			RenameColumnIndexes:    Feature{Supported: false, Status: "stale_index_metadata", Notes: "drop and recreate indexes that reference a renamed column"},
+			RenameColumnIndexes:    Feature{Supported: true, Status: "rewrites_index_metadata", Notes: "column references in index definitions are rewritten in place, including composite and unique indexes; index names are unchanged; RENAME COLUMN is rejected when an expression index references the column (drop that index, rename, then recreate it)"},
 			CreateTableIfNotExists: true,
 			AddColumnIfNotExists:   true,
 		},
@@ -217,8 +217,8 @@ func Current() *Document {
 		JSON: JSON{
 			Storage:        "TEXT (JSON and JSONB type names are accepted without native semantics; insert JSON as text)",
 			Functions:      "json, json_extract, json_set, json_insert, json_replace, json_remove, json_valid, json_type, json_array, json_array_length, json_object, json_quote, json_group_array; PostgreSQL text compatibility: concat_ws, string_agg, pg_backend_pid",
-			ArrowOperators: Feature{Supported: false, Status: "rejected", SQLState: "42601", Notes: "-> and ->> are not implemented; use json_extract"},
-			InvalidInput:   "json_extract/json_set throw 'malformed JSON'; json_valid returns 0",
+			ArrowOperators: Feature{Supported: true, Status: "supported", Notes: "-> returns JSON text for a member or element, ->> returns the SQL value (text, number, 1/0 for booleans, NULL for JSON null); right operand is a label, an array index, or a $ path; missing paths and NULL operands return NULL; chaining is left-to-right"},
+			InvalidInput:   "json_extract/json_set and ->/->> throw 'malformed JSON'; json_valid returns 0",
 		},
 
 		ErrorCodes: map[string]string{

pkg/capabilities/capabilities_test.go

@@ -39,6 +39,22 @@ func TestCurrentDocumentIsVersionedAndMarshalable(t *testing.T) {
 	if round.Notifications.PayloadLimit != 8000 || round.Notifications.DurableReplay {
 		t.Fatalf("unexpected notification limits: %+v", round.Notifications)
 	}
+	if !round.DDL.RenameColumnIndexes.Supported {
+		t.Fatal("RENAME COLUMN must advertise index metadata rewriting")
+	}
+}
+
+func TestJSONArrowOperatorsAreAdvertised(t *testing.T) {
+	arrow := Current().JSON.ArrowOperators
+	if !arrow.Supported {
+		t.Fatal("-> and ->> should be advertised as supported")
+	}
+	if arrow.SQLState != "" {
+		t.Fatalf("arrow operator SQLSTATE = %q, want none", arrow.SQLState)
+	}
+	if arrow.Status == "rejected" {
+		t.Fatalf("arrow operator status = %q, want a supported status", arrow.Status)
+	}
 }
 
 func TestAddColumnBackfillIsAdvertised(t *testing.T) {

pkg/executor/alter_table_test.go

@@ -2,6 +2,7 @@ package executor
 
 import (
 	"errors"
+	"strings"
 	"sync"
 	"testing"
 
@@ -378,33 +379,214 @@ func TestAlterTableDropColumnRewritesPopulatedRows(t *testing.T) {
 	}
 }
 
-// RENAME COLUMN rewrites row keys but deliberately leaves index metadata
-// unchanged. Documented guidance: drop and recreate indexes after a rename.
-func TestAlterTableRenameColumnLeavesIndexMetadataStale(t *testing.T) {
+// RENAME COLUMN rewrites the durable index definitions along with the schema
+// and rows, so single- and multi-column indexes name the new column while
+// keeping their original index names.
+func TestAlterTableRenameColumnUpdatesIndexMetadata(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT)")
+	execMust(t, e, "INSERT INTO t (id, a, b) VALUES (1, 'x', 'y')")
+	execMust(t, e, "CREATE INDEX idx_a ON t(a)")
+	execMust(t, e, "CREATE INDEX idx_ab ON t(a, b)")
+
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN a TO aa")
+
+	single, err := schema.GetIndex("idx_a")
+	if err != nil {
+		t.Fatalf("get idx_a: %v", err)
+	}
+	if len(single.Columns) != 1 || single.Columns[0].Name != "aa" {
+		t.Fatalf("idx_a columns = %#v, want [aa]", single.Columns)
+	}
+
+	composite, err := schema.GetIndex("idx_ab")
+	if err != nil {
+		t.Fatalf("get idx_ab: %v", err)
+	}
+	if len(composite.Columns) != 2 || composite.Columns[0].Name != "aa" || composite.Columns[1].Name != "b" {
+		t.Fatalf("idx_ab columns = %#v, want [aa b]", composite.Columns)
+	}
+
+	// index_info reports the renamed column and resolves its column id.
+	res := execMust(t, e, "PRAGMA index_info(idx_a)")
+	if len(res.Rows) != 1 || res.Rows[0][1] != int64(1) || res.Rows[0][2] != "aa" {
+		t.Fatalf("index_info(idx_a) = %v, want [[0 1 aa]]", res.Rows)
+	}
+
+	// sqlite_master renders the recreated index over the new column name.
+	res = execMust(t, e, "SELECT sql FROM sqlite_master WHERE type='index' AND name='idx_ab'")
+	if len(res.Rows) != 1 || !strings.Contains(res.Rows[0][0].(string), "(`aa`, `b`)") {
+		t.Fatalf("recreated idx_ab SQL = %v, want a definition over aa, b", res.Rows)
+	}
+
+	// The definition is durable, not just cached on this SchemaManager.
+	reloaded := storage.NewSchemaManager(pool, "testdb")
+	composite, err = reloaded.GetIndex("idx_ab")
+	if err != nil {
+		t.Fatalf("reload idx_ab: %v", err)
+	}
+	if len(composite.Columns) != 2 || composite.Columns[0].Name != "aa" || composite.Columns[1].Name != "b" {
+		t.Fatalf("reloaded idx_ab columns = %#v, want [aa b]", composite.Columns)
+	}
+
+	// Index-backed equality lookup still returns the row after the rename.
+	res = execMust(t, e, "SELECT id, aa FROM t WHERE aa = 'x'")
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) || res.Rows[0][1] != "x" {
+		t.Fatalf("indexed lookup after rename = %v, want [[1 x]]", res.Rows)
+	}
+}
+
+// A UNIQUE index keeps enforcing uniqueness after its column is renamed.
+func TestAlterTableRenameColumnKeepsUniqueIndexEnforcement(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT, tag TEXT)")
+	execMust(t, e, "CREATE UNIQUE INDEX uq_email ON t(email)")
+	execMust(t, e, "INSERT INTO t (id, email, tag) VALUES (1, 'a', 'first')")
+
+	if _, err := execSQL(e, "INSERT INTO t (id, email, tag) VALUES (2, 'a', 'dup')"); err == nil || !strings.Contains(err.Error(), "UNIQUE constraint failed") {
+		t.Fatalf("pre-rename duplicate error = %v, want UNIQUE constraint failure", err)
+	}
+
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN email TO mail")
+
+	if _, err := execSQL(e, "INSERT INTO t (id, mail, tag) VALUES (3, 'a', 'dup')"); err == nil || !strings.Contains(err.Error(), "UNIQUE constraint failed") {
+		t.Fatalf("post-rename duplicate error = %v, want UNIQUE constraint failure", err)
+	}
+	execMust(t, e, "INSERT INTO t (id, mail, tag) VALUES (4, 'b', 'ok')")
+
+	res := execMust(t, e, "SELECT id, mail, tag FROM t ORDER BY id")
+	if res.RowCount != 2 || res.Rows[0][1] != "a" || res.Rows[1][1] != "b" {
+		t.Fatalf("rows after rename = %v, want two distinct emails", res.Rows)
+	}
+
+	res = execMust(t, e, "PRAGMA index_list(t)")
+	if len(res.Rows) != 1 || res.Rows[0][1] != "uq_email" || res.Rows[0][2] != int64(1) {
+		t.Fatalf("index_list(t) = %v, want the unique uq_email index", res.Rows)
+	}
+}
+
+// ON CONFLICT resolves against the renamed column and rejects the old name.
+func TestAlterTableRenameColumnResolvesOnConflictTarget(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT, tag TEXT)")
+	execMust(t, e, "CREATE UNIQUE INDEX uq_email ON t(email)")
+	execMust(t, e, "INSERT INTO t (id, email, tag) VALUES (1, 'a', 'old')")
+
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN email TO mail")
+
+	execMust(t, e, "INSERT INTO t (id, mail, tag) VALUES (2, 'a', 'new') ON CONFLICT (mail) DO UPDATE SET tag = excluded.tag")
+	res := execMust(t, e, "SELECT id, mail, tag FROM t")
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) || res.Rows[0][2] != "new" {
+		t.Fatalf("upsert on renamed column = %v, want the existing row updated", res.Rows)
+	}
+
+	if _, err := execSQL(e, "INSERT INTO t (id, mail, tag) VALUES (3, 'a', 'x') ON CONFLICT (email) DO UPDATE SET tag = excluded.tag"); err == nil || !strings.Contains(err.Error(), "column not found: email") {
+		t.Fatalf("upsert on old column error = %v, want column not found: email", err)
+	}
+}
+
+// Renaming a column back restores the original index metadata, and renaming to
+// an existing column name still fails without touching that metadata.
+func TestAlterTableRenameColumnCycleRestoresIndexMetadata(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
 	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT)")
+	execMust(t, e, "CREATE UNIQUE INDEX uq_a ON t(a)")
 	execMust(t, e, "INSERT INTO t (id, a) VALUES (1, 'x')")
-	execMust(t, e, "CREATE INDEX idx_a ON t(a)")
 
-	execMust(t, e, "ALTER TABLE t RENAME COLUMN a TO aa")
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN a TO b")
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN b TO a")
 
-	res := execMust(t, e, "SELECT aa FROM t WHERE id = 1")
-	if got := res.Rows[0][0]; got != "x" {
-		t.Fatalf("renamed value = %v, want x", got)
+	idx, err := schema.GetIndex("uq_a")
+	if err != nil {
+		t.Fatalf("get uq_a: %v", err)
+	}
+	if len(idx.Columns) != 1 || idx.Columns[0].Name != "a" {
+		t.Fatalf("uq_a columns after cycle = %#v, want [a]", idx.Columns)
+	}
+	if _, err := execSQL(e, "INSERT INTO t (id, a) VALUES (2, 'x')"); err == nil || !strings.Contains(err.Error(), "UNIQUE constraint failed") {
+		t.Fatalf("duplicate error after rename cycle = %v, want UNIQUE constraint failure", err)
 	}
 
-	names, err := schema.ListIndexes()
+	if _, err := execSQL(e, "ALTER TABLE t RENAME COLUMN a TO id"); err == nil || !strings.Contains(err.Error(), "column already exists: id") {
+		t.Fatalf("rename to existing column error = %v, want column already exists: id", err)
+	}
+	idx, err = schema.GetIndex("uq_a")
 	if err != nil {
-		t.Fatalf("list indexes: %v", err)
+		t.Fatalf("get uq_a after rejected rename: %v", err)
+	}
+	if len(idx.Columns) != 1 || idx.Columns[0].Name != "a" {
+		t.Fatalf("uq_a columns after rejected rename = %#v, want [a]", idx.Columns)
+	}
+}
+
+// A column referenced by an expression index cannot be renamed: the engine
+// stores expression text verbatim, so the rename is rejected instead of leaving
+// a UNIQUE expression index silently unbound from the column it enforces.
+func TestAlterTableRenameColumnRejectsExpressionIndexDependency(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT, tag TEXT)")
+	execMust(t, e, "CREATE UNIQUE INDEX uq_lower_email ON t(lower(email))")
+	execMust(t, e, "INSERT INTO t (id, email, tag) VALUES (1, 'a@x', 'keep')")
+
+	_, err := execSQL(e, "ALTER TABLE t RENAME COLUMN email TO mail")
+	if err == nil {
+		t.Fatal("rename of a column used by an expression index must be rejected")
 	}
-	found := false
-	for _, name := range names {
-		if name == "idx_a" {
-			found = true
+	for _, want := range []string{"uq_lower_email", "lower(email)"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Fatalf("rename error = %v, want it to name %q", err, want)
 		}
 	}
-	if !found {
-		t.Fatalf("indexes = %v, want idx_a to still exist (rename does not rewrite index metadata)", names)
+
+	// Nothing was renamed, and the UNIQUE expression index still enforces.
+	sch, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, ok := sch.GetColumn("email"); !ok {
+		t.Fatal("rejected rename removed the email column")
+	}
+	if _, ok := sch.GetColumn("mail"); ok {
+		t.Fatal("rejected rename added the mail column")
+	}
+	if _, err := execSQL(e, "INSERT INTO t (id, email, tag) VALUES (2, 'a@x', 'dup')"); err == nil || !strings.Contains(err.Error(), "UNIQUE constraint failed") {
+		t.Fatalf("duplicate on expression index after rejected rename = %v, want UNIQUE constraint failure", err)
+	}
+
+	// A column the expression does not reference can still be renamed.
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN tag TO label")
+	idx, err := schema.GetIndex("uq_lower_email")
+	if err != nil {
+		t.Fatalf("get uq_lower_email: %v", err)
+	}
+	if len(idx.Columns) != 1 || idx.Columns[0].Expression != "lower(email)" {
+		t.Fatalf("uq_lower_email columns = %#v, want the untouched lower(email) expression", idx.Columns)
+	}
+	if res := execMust(t, e, "SELECT label FROM t WHERE id = 1"); res.RowCount != 1 || res.Rows[0][0] != "keep" {
+		t.Fatalf("renamed label = %v, want keep", res.Rows)
+	}
+}
+
+// Conflict targets that name a hidden rowid alias are not table columns but are
+// still valid targets, including after a rename.
+func TestAlterTableRenameColumnKeepsRowIDConflictTarget(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT)")
+	execMust(t, e, "INSERT INTO t (id, a) VALUES (1, 'x')")
+
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN a TO b")
+
+	execMust(t, e, "INSERT INTO t (id, b) VALUES (2, 'y') ON CONFLICT (rowid) DO NOTHING")
+	execMust(t, e, "INSERT INTO t (id, b) VALUES (3, 'z') ON CONFLICT (oid) DO NOTHING")
+	execMust(t, e, "INSERT INTO t (id, b) VALUES (4, 'w') ON CONFLICT (_rowid_) DO NOTHING")
+
+	if res := execMust(t, e, "SELECT count(*) FROM t"); res.Rows[0][0] != int64(4) {
+		t.Fatalf("row count = %v, want 4", res.Rows[0][0])
 	}
 }

pkg/executor/dashboard_repro_test.go

@@ -24,7 +24,7 @@ func TestSelectColumnTypesGroupBy(t *testing.T) {
 	execMust(t, e, `INSERT INTO hit_stats VALUES (1, 1, '2024-05-06', '[]')`)
 
 	res := execMust(t, e, `SELECT hour, sum(total) FROM hit_counts GROUP BY hour ORDER BY hour`)
-	want := []string{"TIMESTAMP", "TEXT"}
+	want := []string{"TIMESTAMP", "BIGINT"}
 	if len(res.ColumnTypes) != len(want) {
 		t.Fatalf("group-by column types = %v, want %v", res.ColumnTypes, want)
 	}

pkg/executor/executor.go

@@ -445,6 +445,7 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		} else {
 			result.AddColumn("column1")
 		}
+		result.ColumnTypes = []string{inferExprType(stmt.Columns[0].Expr, nil)}
 		result.AddRow(int64(count))
 		return result, nil
 	}
@@ -481,20 +482,21 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 				return e.executeAggregateSelect(stmt, []storage.Row{}, schema)
 			}
 		}
-		// Non-aggregate query with WHERE false: return empty result
+		// Non-aggregate query with WHERE false: return empty result using the
+		// same wildcard expansion and type metadata the scanned path reports.
+		refs := collectAllTableRefs(stmt.From)
+		tableStars, starErr := e.resolveTableStars(stmt.From, stmt.Columns)
+		if starErr != nil {
+			return nil, starErr
+		}
 		result := NewResult("SELECT")
-		for i, col := range stmt.Columns {
-			if col.Alias != "" {
-				result.AddColumn(col.Alias)
-			} else if ref, ok := col.Expr.(*parser.ColumnRef); ok {
-				result.AddColumn(ref.Column)
-			} else if col.Star {
-				for _, c := range schema.Columns {
-					result.AddColumn(c.Name)
-				}
-			} else {
-				result.AddColumn(fmt.Sprintf("column%d", i+1))
-			}
+		for _, name := range e.selectColumnNames(stmt, schema, refs, isMultiTable, tableStars) {
+			result.AddColumn(name)
+		}
+		if isMultiTable || stmt.From[0].Join != nil {
+			result.ColumnTypes = e.joinedSelectColumnTypes(stmt, refs)
+		} else {
+			result.ColumnTypes = selectColumnTypes(stmt, schema)
 		}
 		return result, nil
 	}
@@ -1223,58 +1225,16 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 	hasJoin := len(stmt.From) > 0 && stmt.From[0].Join != nil
 	allTableRefs := collectAllTableRefs(stmt.From)
 
-	// Resolve qualified wildcard (table.*) projections once, keyed by qualifier.
-	type starProjection struct {
-		cols   []storage.Column
-		prefix string
-	}
-	tableStars := make(map[string]starProjection)
-	for _, col := range stmt.Columns {
-		if col.TableStar == "" {
-			continue
-		}
-		key := strings.ToUpper(col.TableStar)
-		if _, ok := tableStars[key]; ok {
-			continue
-		}
-		cols, prefix, err := e.resolveTableStar(stmt.From, col.TableStar)
-		if err != nil {
-			return nil, err
-		}
-		tableStars[key] = starProjection{cols: cols, prefix: prefix}
+	// Resolve qualified wildcard (table.*) projections once, keyed by qualifier,
+	// so names, types, and row values expand exactly the same columns.
+	tableStars, err := e.resolveTableStars(stmt.From, stmt.Columns)
+	if err != nil {
+		return nil, err
 	}
 
 	// Determine columns
-	for i, col := range stmt.Columns {
-		if col.Alias != "" {
-			result.AddColumn(col.Alias)
-		} else if ref, ok := col.Expr.(*parser.ColumnRef); ok {
-			result.AddColumn(ref.Column)
-		} else if col.Star {
-			if isMultiTable || hasJoin {
-				// Add columns from ALL joined tables in order
-				for _, tref := range allTableRefs {
-					sch, _ := e.schema.GetSchema(tref.Name)
-					if sch != nil {
-						for _, c := range sch.Columns {
-							result.AddColumn(c.Name)
-						}
-					}
-				}
-			} else {
-				// Handle SELECT * - add all columns from schema
-				for _, c := range schema.Columns {
-					result.AddColumn(c.Name)
-				}
-			}
-		} else if col.TableStar != "" {
-			proj := tableStars[strings.ToUpper(col.TableStar)]
-			for _, c := range proj.cols {
-				result.AddColumn(c.Name)
-			}
-		} else {
-			result.AddColumn(fmt.Sprintf("column%d", i+1))
-		}
+	for _, name := range e.selectColumnNames(stmt, schema, allTableRefs, isMultiTable, tableStars) {
+		result.AddColumn(name)
 	}
 
 	// Populate column types from schema metadata so protocol consumers can
@@ -1382,6 +1342,9 @@ func (e *Executor) executeCompound(c *parser.CompoundSelect) (*Result, error) {
 	for _, col := range left.Columns {
 		result.AddColumn(col)
 	}
+	// A compound SELECT reports the left branch's projection metadata, as
+	// PostgreSQL does: the set operation does not retype its columns.
+	result.ColumnTypes = append([]string(nil), left.ColumnTypes...)
 
 	switch c.Op {
 	case parser.SetOpUnion:
@@ -1479,6 +1442,9 @@ func (e *Executor) executeSelectExpr(stmt *parser.SelectStmt) (*Result, error) {
 		} else {
 			result.AddColumn(fmt.Sprintf("column%d", i+1))
 		}
+		// A SELECT without FROM has no schema, but literal and operator
+		// projections still have a known type.
+		result.ColumnTypes = append(result.ColumnTypes, inferExprType(col.Expr, nil))
 	}
 
 	// Evaluate expressions
@@ -1711,6 +1677,7 @@ func schemaFromColumns(columns []string) *storage.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")
+	resolve := schemaTypeResolver(schema)
 
 	// Determine columns and evaluate aggregates
 	for i, col := range stmt.Columns {
@@ -1721,6 +1688,12 @@ func (e *Executor) executeAggregateSelect(stmt *parser.SelectStmt, rows []storag
 		} else {
 			result.AddColumn(fmt.Sprintf("column%d", i+1))
 		}
+		if col.Star || col.TableStar != "" {
+			// The aggregate path emits one synthesized column per wildcard.
+			result.ColumnTypes = append(result.ColumnTypes, "TEXT")
+			continue
+		}
+		result.ColumnTypes = append(result.ColumnTypes, inferExprType(col.Expr, resolve))
 	}
 
 	// Calculate values
@@ -2484,6 +2457,75 @@ func collectAllTableRefs(from []parser.TableRef) []parser.TableRef {
 	return refs
 }
 
+// starProjection is a resolved qualified wildcard: the table's columns and the
+// row-key prefix used to read their values.
+type starProjection struct {
+	cols   []storage.Column
+	prefix string
+}
+
+// resolveTableStars resolves every qualified wildcard in a projection.
+func (e *Executor) resolveTableStars(from []parser.TableRef, columns []parser.SelectColumn) (map[string]starProjection, error) {
+	stars := make(map[string]starProjection)
+	for _, col := range columns {
+		if col.TableStar == "" {
+			continue
+		}
+		key := strings.ToUpper(col.TableStar)
+		if _, ok := stars[key]; ok {
+			continue
+		}
+		cols, prefix, err := e.resolveTableStar(from, col.TableStar)
+		if err != nil {
+			return nil, err
+		}
+		stars[key] = starProjection{cols: cols, prefix: prefix}
+	}
+	return stars, nil
+}
+
+// selectColumnNames expands a projection into result column names using the
+// same wildcard expansion as the row values: every joined table for an
+// unqualified *, the resolved table for table.*, and the schema for a
+// single-table *.
+func (e *Executor) selectColumnNames(stmt *parser.SelectStmt, schema *storage.Schema, refs []parser.TableRef, multiTable bool, stars map[string]starProjection) []string {
+	hasJoin := len(stmt.From) > 0 && stmt.From[0].Join != nil
+	names := make([]string, 0, len(stmt.Columns))
+	for i, col := range stmt.Columns {
+		switch {
+		case col.Alias != "":
+			names = append(names, col.Alias)
+		case col.Star:
+			if multiTable || hasJoin {
+				for _, ref := range refs {
+					if refSchema, err := e.schema.GetSchema(ref.Name); err == nil {
+						for _, c := range refSchema.Columns {
+							names = append(names, c.Name)
+						}
+					}
+				}
+				continue
+			}
+			for _, c := range schema.Columns {
+				names = append(names, c.Name)
+			}
+		case col.TableStar != "":
+			for _, c := range stars[strings.ToUpper(col.TableStar)].cols {
+				names = append(names, c.Name)
+			}
+		case col.Expr != nil:
+			if ref, ok := col.Expr.(*parser.ColumnRef); ok {
+				names = append(names, ref.Column)
+				continue
+			}
+			names = append(names, fmt.Sprintf("column%d", i+1))
+		default:
+			names = append(names, fmt.Sprintf("column%d", i+1))
+		}
+	}
+	return names
+}
+
 // resolveTableStar resolves a qualified wildcard (table.*) qualifier to the
 // matching table ref in a FROM clause. It returns the table's schema columns
 // and the key prefix used to look values up in a joined/aliased row (the
@@ -4216,6 +4258,9 @@ func (e *Executor) executeAlterTableRename(table string, action *parser.RenameTa
 
 // executeAlterTableRenameColumn renames a column.
 func (e *Executor) executeAlterTableRenameColumn(table string, action *parser.RenameColumnAction) (*Result, error) {
+	if err := e.rejectExpressionIndexRename(table, action.OldName); err != nil {
+		return nil, err
+	}
 	if err := e.schema.RenameColumn(table, action.OldName, action.NewName); err != nil {
 		return nil, err
 	}
@@ -4227,6 +4272,36 @@ func (e *Executor) executeAlterTableRenameColumn(table string, action *parser.Re
 	return result, nil
 }
 
+// rejectExpressionIndexRename refuses to rename a column that a stored index
+// expression references. Expression text is persisted verbatim, so the rename
+// cannot rewrite it; leaving the old identifier behind would silently unbind a
+// UNIQUE expression index, because its rows no longer resolve the referenced
+// column.
+func (e *Executor) rejectExpressionIndexRename(table, column string) error {
+	indexes, err := e.schema.ListTableIndexes(table)
+	if err != nil {
+		return err
+	}
+	for _, idx := range indexes {
+		for _, ic := range idx.Columns {
+			if ic.Expression == "" {
+				continue
+			}
+			references, err := storedExpressionReferencesColumn(ic.Expression, column)
+			if err != nil {
+				return fmt.Errorf("index %s: %w", idx.Name, err)
+			}
+			if references {
+				return pgerror.Unsupported(
+					"cannot rename column %s: index %s depends on it through the expression %s; drop the index, rename the column, then recreate the index",
+					column, idx.Name, ic.Expression,
+				)
+			}
+		}
+	}
+	return nil
+}
+
 // Transaction execution methods
 
 // executeBegin starts a new transaction.
@@ -4998,6 +5073,10 @@ func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (int
 		return applyBitwise(expr.Op, left, right)
 	}
 
+	if expr.Op == lexer.TokenArrow || expr.Op == lexer.TokenArrowText {
+		return jsonArrowExtract(left, right, expr.Op == lexer.TokenArrowText)
+	}
+
 	switch expr.Op {
 	case lexer.TokenPlus:
 		if left == nil || right == nil {
@@ -5121,6 +5200,9 @@ func (e *Executor) applyBinaryOp(op lexer.TokenType, left, right interface{}) (i
 		}
 		return applyBitwise(op, left, right)
 	}
+	if op == lexer.TokenArrow || op == lexer.TokenArrowText {
+		return jsonArrowExtract(left, right, op == lexer.TokenArrowText)
+	}
 	switch op {
 	case lexer.TokenPlus:
 		if left == nil || right == nil {
@@ -7183,6 +7265,14 @@ func toInt64(v interface{}) int64 {
 			return 1
 		}
 		return 0
+	case jsonText:
+		// -> returns JSON text, so integer conversion parses its text form.
+		// Keep exact integers exact before falling back to float truncation.
+		if i, err := strconv.ParseInt(string(val), 10, 64); err == nil {
+			return i
+		}
+		f, _ := strconv.ParseFloat(string(val), 64)
+		return int64(f)
 	default:
 		return 0
 	}
@@ -7206,6 +7296,9 @@ func toFloat(v interface{}) float64 {
 	case string:
 		f, _ := strconv.ParseFloat(val, 64)
 		return f
+	case jsonText:
+		f, _ := strconv.ParseFloat(string(val), 64)
+		return f
 	default:
 		return 0
 	}
@@ -7225,6 +7318,9 @@ func toBool(v interface{}) bool {
 		return val != 0
 	case string:
 		return val != "" && val != "0" && strings.ToLower(val) != "false"
+	case jsonText:
+		s := string(val)
+		return s != "" && s != "0" && strings.ToLower(s) != "false"
 	default:
 		return false
 	}
@@ -7304,6 +7400,9 @@ func toNumeric(v interface{}) (float64, bool) {
 	case string:
 		f, err := strconv.ParseFloat(val, 64)
 		return f, err == nil
+	case jsonText:
+		f, err := strconv.ParseFloat(string(val), 64)
+		return f, err == nil
 	default:
 		return 0, false
 	}

pkg/executor/indexexpr.go

@@ -51,6 +51,22 @@ func EvalStoredExpression(text string, row storage.Row) (interface{}, error) {
 	return statelessEval.evalExpr(expr, row)
 }
 
+// storedExpressionReferencesColumn reports whether a persisted index expression
+// names column. The engine stores expression SQL text verbatim, so a column
+// rename cannot rewrite the reference.
+func storedExpressionReferencesColumn(text, column string) (bool, error) {
+	expr, err := parseStoredExpr(text)
+	if err != nil {
+		return false, err
+	}
+	for _, ref := range collectColumnRefs(expr) {
+		if strings.EqualFold(ref, column) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // ValidateDeterministicExpr rejects an expression that SQLite would not allow in
 // an index or generated column: subqueries, window functions, aggregates, and
 // non-deterministic or unknown functions.

pkg/executor/json.go

@@ -329,6 +329,63 @@ func newJSONContainer(seg jsonPathSeg) interface{} {
 	return map[string]interface{}{}
 }
 
+// jsonArrowExtract evaluates the SQLite -> and ->> operators. The key operand is
+// a field label, an array index, or a $ path. A NULL operand or a path that
+// selects nothing yields SQL NULL; malformed JSON in the document is an error,
+// as it is for json_extract(). scalar selects ->> (SQL value of the selected
+// node) instead of -> (JSON text for the selected node).
+func jsonArrowExtract(document, key interface{}, scalar bool) (interface{}, error) {
+	if document == nil || key == nil {
+		return nil, nil
+	}
+	node, err := jsonInput(document)
+	if err != nil {
+		return nil, err
+	}
+	segs, err := jsonArrowPath(key)
+	if err != nil {
+		return nil, err
+	}
+	found, ok := jsonLookup(node, segs)
+	if !ok {
+		return nil, nil
+	}
+	if scalar {
+		return jsonNodeToSQL(found), nil
+	}
+	s, err := jsonEncode(found)
+	if err != nil {
+		return nil, err
+	}
+	return jsonText(s), nil
+}
+
+// jsonArrowPath turns an arrow-operator operand into a path. Text that starts
+// with $ is a JSON path; other text names a single object member, so labels
+// containing dots or brackets are not reinterpreted as paths. Integers select
+// an array position, with negative values counting from the end.
+func jsonArrowPath(key interface{}) ([]jsonPathSeg, error) {
+	switch t := key.(type) {
+	case string:
+		return jsonArrowPathText(t)
+	case jsonText:
+		return jsonArrowPathText(string(t))
+	case int64:
+		return []jsonPathSeg{{isIndex: true, index: int(t), fromEnd: t < 0}}, nil
+	case int:
+		return []jsonPathSeg{{isIndex: true, index: t, fromEnd: t < 0}}, nil
+	default:
+		return nil, fmt.Errorf("JSON path error")
+	}
+}
+
+func jsonArrowPathText(path string) ([]jsonPathSeg, error) {
+	if strings.HasPrefix(path, "$") {
+		return parseJSONPath(path)
+	}
+	return []jsonPathSeg{{key: path}}, nil
+}
+
 // jsonTypeName returns the JSON1 type name for a decoded node.
 func jsonTypeName(v interface{}) string {
 	switch v.(type) {

pkg/executor/json_operator_test.go

@@ -0,0 +1,161 @@
+package executor
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestJSONArrowObjectAndArrayAccess(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+
+	tests := []struct {
+		name string
+		sql  string
+		want interface{}
+	}{
+		{"object member as JSON text", `SELECT '{"a":1,"s":"x"}' -> 'a'`, "1"},
+		{"string member keeps JSON quotes", `SELECT '{"a":1,"s":"x"}' -> 's'`, `"x"`},
+		{"string member as SQL text", `SELECT '{"a":1,"s":"x"}' ->> 's'`, "x"},
+		{"object member as JSON text", `SELECT '{"b":[10,20]}' -> 'b'`, "[10,20]"},
+		{"array element as JSON text", `SELECT '[10,20,30]' -> 1`, "20"},
+		{"array element as SQL number", `SELECT '[10,20,30]' ->> 1`, int64(20)},
+		{"array element as JSON text", `SELECT '["a","b"]' -> 0`, `"a"`},
+		{"negative array index", `SELECT '[10,20,30]' ->> -1`, int64(30)},
+		{"nested chaining", `SELECT '{"a":{"b":[{"c":7}]}}' -> 'a' -> 'b' -> 0 ->> 'c'`, int64(7)},
+		{"numeric key selects an array index", `SELECT '[10,20]' -> 1`, "20"},
+		{"string key names an object member", `SELECT '{"1":"one"}' -> '1'`, `"one"`},
+		{"numeric key on an object is missing", `SELECT '{"1":"one"}' -> 1`, nil},
+		{"json null as JSON text", `SELECT '{"a":null}' -> 'a'`, "null"},
+		{"json null as SQL NULL", `SELECT '{"a":null}' ->> 'a'`, nil},
+		{"json true as JSON text", `SELECT '{"a":true}' -> 'a'`, "true"},
+		{"json true as SQL integer", `SELECT '{"a":true}' ->> 'a'`, int64(1)},
+		{"json false as SQL integer", `SELECT '{"a":false}' ->> 'a'`, int64(0)},
+		{"path-style key", `SELECT '{"a":{"b":2}}' -> '$.a' ->> 'b'`, int64(2)},
+		{"whole document path", `SELECT '{"a":1}' ->> '$'`, `{"a":1}`},
+		{"missing object key", `SELECT '{"a":1}' -> 'b'`, nil},
+		{"missing object key as SQL", `SELECT '{"a":1}' ->> 'b'`, nil},
+		{"array index past the end", `SELECT '[1,2]' -> 5`, nil},
+		{"NULL document", `SELECT NULL -> 'a'`, nil},
+		{"NULL document as SQL", `SELECT NULL ->> 'a'`, nil},
+		{"NULL key", `SELECT '{"a":1}' -> NULL`, nil},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res := execMust(t, e, tt.sql)
+			got := res.Rows[0][0]
+			if got != tt.want {
+				t.Fatalf("%s = %v (%T), want %v (%T)", tt.sql, got, got, tt.want, tt.want)
+			}
+		})
+	}
+}
+
+func TestJSONArrowMalformedInputIsAnError(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+
+	for _, sql := range []string{
+		`SELECT 'not json' -> 'a'`,
+		`SELECT 'not json' ->> 'a'`,
+		`SELECT '{"a":1}' -> '$.['`,
+	} {
+		if _, err := execSQL(e, sql); err == nil {
+			t.Fatalf("%s: expected an error", sql)
+		} else if !strings.Contains(err.Error(), "JSON") {
+			t.Fatalf("%s: unexpected error: %v", sql, err)
+		}
+	}
+}
+
+func TestJSONArrowInWhereOrderByAndUpdate(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE docs (id INTEGER PRIMARY KEY, doc TEXT, v TEXT)")
+	execMust(t, e, `INSERT INTO docs VALUES (1, '{"kind":"a","n":1}', NULL)`)
+	execMust(t, e, `INSERT INTO docs VALUES (2, '{"kind":"b","n":2}', NULL)`)
+	execMust(t, e, `INSERT INTO docs VALUES (3, '{"kind":"c","n":3}', NULL)`)
+
+	res := execMust(t, e, `SELECT id FROM docs WHERE doc ->> 'kind' = 'b'`)
+	if res.RowCount != 1 || res.Rows[0][0] != int64(2) {
+		t.Fatalf("WHERE doc ->> 'kind' = 'b' rows = %v, want id 2", res.Rows)
+	}
+
+	res = execMust(t, e, `SELECT doc ->> 'kind' AS kind FROM docs ORDER BY doc -> 'n' DESC`)
+	if res.RowCount != 3 {
+		t.Fatalf("ORDER BY rows = %d, want 3", res.RowCount)
+	}
+	if res.Rows[0][0] != "c" || res.Rows[1][0] != "b" || res.Rows[2][0] != "a" {
+		t.Fatalf("ORDER BY doc -> 'n' DESC = %v, want c, b, a", res.Rows)
+	}
+
+	execMust(t, e, `UPDATE docs SET v = doc ->> 'kind'`)
+	res = execMust(t, e, `SELECT id, v FROM docs ORDER BY id`)
+	for i, want := range []string{"a", "b", "c"} {
+		if res.Rows[i][1] != want {
+			t.Fatalf("row %d v = %v, want %q", i, res.Rows[i][1], want)
+		}
+	}
+}
+
+// The -> operator returns JSON text (with the engine's internal JSON subtype),
+// not a plain string. Numeric, boolean, and comparison coercions must still
+// read the text form, exactly as they do for SQL TEXT; otherwise arithmetic
+// silently treats every extracted number as 0.
+func TestJSONArrowOperatorValueCoercions(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE metrics (id INTEGER PRIMARY KEY, doc TEXT)")
+	execMust(t, e, `INSERT INTO metrics VALUES (1, '{"n":2,"big":10,"flag":1}')`)
+	execMust(t, e, `INSERT INTO metrics VALUES (2, '{"n":10,"big":20,"flag":0}')`)
+
+	// Arithmetic uses the text numeric value. Like '2' + 1 elsewhere in the
+	// engine, the result is REAL rather than INTEGER for JSON text.
+	res := execMust(t, e, `SELECT (doc -> 'n') + 1 FROM metrics WHERE id = 1`)
+	if got := res.Rows[0][0]; got != float64(3) {
+		t.Fatalf("(doc -> 'n') + 1 = %v (%T), want 3", got, got)
+	}
+
+	// Bitwise operators coerce through toInt64.
+	res = execMust(t, e, `SELECT (doc -> 'n') & 3 FROM metrics WHERE id = 1`)
+	if got := res.Rows[0][0]; got != int64(2) {
+		t.Fatalf("(doc -> 'n') & 3 = %v (%T), want int64 2", got, got)
+	}
+
+	// Numeric comparison must not fall back to string comparison, where
+	// "10" < "9" and multi-digit values order incorrectly.
+	res = execMust(t, e, `SELECT id FROM metrics WHERE doc -> 'big' > 9 ORDER BY id`)
+	if res.RowCount != 2 || res.Rows[0][0] != int64(1) || res.Rows[1][0] != int64(2) {
+		t.Fatalf("WHERE doc -> 'big' > 9 rows = %v, want ids 1, 2", res.Rows)
+	}
+	res = execMust(t, e, `SELECT id FROM metrics WHERE doc -> 'big' = 10`)
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+		t.Fatalf("WHERE doc -> 'big' = 10 rows = %v, want id 1", res.Rows)
+	}
+
+	// CAST goes through toFloat.
+	res = execMust(t, e, `SELECT CAST(doc -> 'big' AS INTEGER) FROM metrics WHERE id = 1`)
+	if got := res.Rows[0][0]; got != int64(10) {
+		t.Fatalf("CAST(doc -> 'big' AS INTEGER) = %v (%T), want int64 10", got, got)
+	}
+
+	// WHERE truthiness goes through toBool.
+	res = execMust(t, e, `SELECT id FROM metrics WHERE doc -> 'flag'`)
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+		t.Fatalf("WHERE doc -> 'flag' rows = %v, want id 1", res.Rows)
+	}
+
+	// ORDER BY compares the extracted values numerically.
+	res = execMust(t, e, `SELECT id FROM metrics ORDER BY doc -> 'n'`)
+	if res.RowCount != 2 || res.Rows[0][0] != int64(1) || res.Rows[1][0] != int64(2) {
+		t.Fatalf("ORDER BY doc -> 'n' rows = %v, want ids 1, 2", res.Rows)
+	}
+
+	// The JSON subtype must survive -> so JSON functions embed the extracted
+	// value as JSON rather than as a quoted string.
+	res = execMust(t, e, `SELECT json_array(doc -> 'n') FROM metrics WHERE id = 1`)
+	if got := res.Rows[0][0]; got != "[2]" {
+		t.Fatalf("json_array(doc -> 'n') = %v, want [2]", got)
+	}
+}

pkg/executor/result_types.go

@@ -1,19 +1,43 @@
 package executor
 
 import (
+	"strconv"
 	"strings"
 
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
 	"github.com/danfragoso/pizzasql-next/pkg/parser"
 	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
 
+// columnTypeResolver resolves a possibly-qualified column reference to the SQL
+// type name declared in the schema.
+type columnTypeResolver func(table, column string) (string, bool)
+
+// schemaTypeResolver builds a resolver over a single table schema. The table
+// qualifier is ignored because the analyzer has already validated it.
+func schemaTypeResolver(schema *storage.Schema) columnTypeResolver {
+	return func(_ string, column string) (string, bool) {
+		if schema == nil {
+			return "", false
+		}
+		for _, c := range schema.Columns {
+			if strings.EqualFold(c.Name, column) {
+				return c.Type, true
+			}
+		}
+		return "", false
+	}
+}
+
 // selectColumnTypes computes the result column types for a single-table SELECT
 // projection, parallel to the column names produced by the normal projection
-// expansion. Direct column references (including aliases) and SELECT * resolve
-// to the declared schema type; any expression whose type the engine does not
-// know is reported as TEXT. Types are derived from schema metadata rather than
-// row values so they remain correct even when the result set is empty.
+// expansion. Direct column references (including aliases) resolve to the
+// declared schema type and common expressions resolve to the type the executor
+// produces for them. Types are derived from schema metadata and the expression
+// shape rather than row values so they remain correct even when the result set
+// is empty.
 func selectColumnTypes(stmt *parser.SelectStmt, schema *storage.Schema) []string {
+	resolve := schemaTypeResolver(schema)
 	types := make([]string, 0, len(stmt.Columns))
 	for _, col := range stmt.Columns {
 		if col.Star {
@@ -31,36 +55,22 @@ func selectColumnTypes(stmt *parser.SelectStmt, schema *storage.Schema) []string
 			}
 			continue
 		}
-		types = append(types, projectionColumnType(col, schema))
+		types = append(types, inferExprType(col.Expr, resolve))
 	}
 	return types
 }
 
-// projectionColumnType resolves the type of a single projected expression. A
-// direct column reference resolves to the declared schema type (matched
-// case-insensitively and ignoring any table qualifier); anything else is
-// unknown and reported as TEXT.
+// projectionColumnType resolves the type of a single projected expression
+// against a single table schema.
 func projectionColumnType(col parser.SelectColumn, schema *storage.Schema) string {
-	ref, ok := col.Expr.(*parser.ColumnRef)
-	if !ok {
-		return "TEXT"
-	}
-	for _, c := range schema.Columns {
-		if strings.EqualFold(c.Name, ref.Column) {
-			return c.Type
-		}
-	}
-	if storage.IsRowIDColumn(ref.Column) {
-		// The hidden rowid alias is a signed 64-bit integer, not an unknown
-		// expression, so report it as BIGINT (int8) on the wire.
-		return "BIGINT"
-	}
-	return "TEXT"
+	return inferExprType(col.Expr, schemaTypeResolver(schema))
 }
 
-// joinedSelectColumnTypes resolves direct projections against every table in a
-// joined or comma-separated FROM clause. Expressions retain TEXT metadata.
+// joinedSelectColumnTypes resolves projections against every table in a joined
+// or comma-separated FROM clause. An unqualified reference resolves to the first
+// table that declares it, matching how row lookups are performed.
 func (e *Executor) joinedSelectColumnTypes(stmt *parser.SelectStmt, refs []parser.TableRef) []string {
+	resolve := e.joinedTypeResolver(refs)
 	types := make([]string, 0, len(stmt.Columns))
 	for _, projection := range stmt.Columns {
 		if projection.Star {
@@ -82,36 +92,280 @@ func (e *Executor) joinedSelectColumnTypes(stmt *parser.SelectStmt, refs []parse
 			continue
 		}
 
-		ref, ok := projection.Expr.(*parser.ColumnRef)
-		if !ok {
-			types = append(types, "TEXT")
-			continue
-		}
-		columnType := "TEXT"
-		found := false
-		for _, table := range refs {
-			if ref.Table != "" && !strings.EqualFold(ref.Table, table.Alias) && !strings.EqualFold(ref.Table, table.Name) {
+		types = append(types, inferExprType(projection.Expr, resolve))
+	}
+	return types
+}
+
+// joinedTypeResolver resolves a column reference against the FROM clause. It
+// mirrors the row lookup rules: a qualified reference is limited to the matching
+// table alias or name, and the first table declaring an unqualified column wins.
+func (e *Executor) joinedTypeResolver(refs []parser.TableRef) columnTypeResolver {
+	return func(table, column string) (string, bool) {
+		for _, ref := range refs {
+			if table != "" && !strings.EqualFold(table, ref.Alias) && !strings.EqualFold(table, ref.Name) {
 				continue
 			}
-			schema, err := e.schema.GetSchema(table.Name)
+			schema, err := e.schema.GetSchema(ref.Name)
 			if err != nil {
 				continue
 			}
-			for _, column := range schema.Columns {
-				if strings.EqualFold(column.Name, ref.Column) {
-					columnType = column.Type
-					found = true
-					break
+			for _, candidate := range schema.Columns {
+				if strings.EqualFold(candidate.Name, column) {
+					return candidate.Type, true
 				}
 			}
-			if found {
+			if table != "" {
+				// A qualifier names exactly one table; do not fall through to
+				// another relation that happens to declare the same column.
 				break
 			}
 		}
-		if !found && storage.IsRowIDColumn(ref.Column) {
-			columnType = "BIGINT"
+		return "", false
+	}
+}
+
+// inferExprType reports the SQL type name of a projection expression, using the
+// same names the schema stores so the wire layer can map them to PostgreSQL
+// type OIDs. It only describes the values the executor produces; nothing is
+// converted. TEXT is the fallback for expressions whose type is not known.
+func inferExprType(expr parser.Expr, resolve columnTypeResolver) string {
+	switch e := expr.(type) {
+	case nil:
+		return "TEXT"
+	case *parser.ColumnRef:
+		if resolve != nil {
+			if typeName, ok := resolve(e.Table, e.Column); ok && !isUnknownTypeName(typeName) {
+				return typeName
+			}
+		}
+		if storage.IsRowIDColumn(e.Column) {
+			// The hidden rowid alias is a signed 64-bit integer.
+			return "BIGINT"
 		}
-		types = append(types, columnType)
+		return "TEXT"
+	case *parser.ParenExpr:
+		return inferExprType(e.Expr, resolve)
+	case *parser.LiteralExpr:
+		return literalType(e)
+	case *parser.CastExpr:
+		if typeName, ok := castResultType(e.Type.Name); ok {
+			return typeName
+		}
+		return "TEXT"
+	case *parser.FunctionCall:
+		return functionResultType(e, resolve)
+	case *parser.BinaryExpr:
+		return binaryResultType(e, resolve)
+	case *parser.UnaryExpr:
+		return unaryResultType(e, resolve)
+	case *parser.CaseExpr:
+		return caseResultType(e, resolve)
+	case *parser.InExpr, *parser.BetweenExpr, *parser.LikeExpr, *parser.IsNullExpr,
+		*parser.IsDistinctExpr, *parser.ExistsExpr:
+		return "BOOLEAN"
+	default:
+		return "TEXT"
+	}
+}
+
+// literalType reports the type of a literal value. Integer literals are INTEGER
+// when they fit int32 and BIGINT otherwise, matching how the value is carried.
+// Non-integral numeric literals are carried as float64, so they report DOUBLE
+// rather than NUMERIC: the engine has no exact decimal carrier.
+func literalType(lit *parser.LiteralExpr) string {
+	switch lit.Type {
+	case lexer.TokenNumber:
+		if strings.ContainsAny(lit.Value, ".eE") {
+			return "DOUBLE"
+		}
+		if n, err := strconv.ParseInt(lit.Value, 10, 64); err == nil && n >= -2147483648 && n <= 2147483647 {
+			return "INTEGER"
+		}
+		return "BIGINT"
+	case lexer.TokenString:
+		return "TEXT"
+	case lexer.TokenBlob:
+		return "BLOB"
+	case lexer.TokenTRUE, lexer.TokenFALSE:
+		return "BOOLEAN"
+	default: // NULL and anything unrecognized stay unknown/TEXT.
+		return "TEXT"
+	}
+}
+
+// castResultType reports the result type of CAST(x AS name). Only targets the
+// executor actually converts are reported; every other target leaves the value
+// untouched, so the value keeps its original encoding and stays TEXT metadata.
+//
+// Casts report the carrier type of the converted value, not the target width:
+// evalCastExpr always produces int64 for integer targets and float64 for
+// real targets, and it does not range-check them. A narrower advertised type
+// (int2/int4/float4) could otherwise fail to decode a value the engine accepted.
+func castResultType(name string) (string, bool) {
+	switch strings.ToUpper(strings.TrimSpace(name)) {
+	case "INTEGER", "INT", "SMALLINT", "TINYINT", "MEDIUMINT", "BIGINT":
+		return "BIGINT", true
+	case "REAL", "FLOAT", "DOUBLE":
+		return "DOUBLE", true
+	case "TEXT", "VARCHAR", "CHAR", "CHARACTER", "CLOB", "NCHAR", "NVARCHAR":
+		return "TEXT", true
+	case "BLOB":
+		return "BLOB", true
+	default:
+		return "", false
+	}
+}
+
+// functionResultType reports the result type of a function call. Aggregate
+// result types mirror the values the executor computes: COUNT is BIGINT, SUM
+// over integers is BIGINT, SUM over float carriers is DOUBLE, and AVG is always
+// DOUBLE because evalAggregateExpr divides in float64.
+func functionResultType(fn *parser.FunctionCall, resolve columnTypeResolver) string {
+	switch strings.ToUpper(fn.Name) {
+	case "COUNT":
+		return "BIGINT"
+	case "SUM":
+		return sumResultType(firstArgType(fn, resolve))
+	case "TOTAL":
+		return "DOUBLE"
+	case "AVG":
+		return "DOUBLE"
+	case "MIN", "MAX":
+		// MIN/MAX return one of the input values unchanged.
+		return firstArgType(fn, resolve)
+	default:
+		return "TEXT"
+	}
+}
+
+// firstArgType reports the inferred type of a function's first argument.
+func firstArgType(fn *parser.FunctionCall, resolve columnTypeResolver) string {
+	if len(fn.Args) == 0 {
+		return "TEXT"
+	}
+	return inferExprType(fn.Args[0], resolve)
+}
+
+// sumResultType reports the SUM result type for an input type. The executor
+// accumulates integers in int64 and everything else in float64.
+func sumResultType(argType string) string {
+	switch {
+	case isFloatCarrierTypeName(argType):
+		return "DOUBLE"
+	case isIntegerTypeName(argType):
+		return "BIGINT"
+	default:
+		return "TEXT"
+	}
+}
+
+// binaryResultType reports the result type of a binary operator. Comparisons and
+// boolean operators produce BOOLEAN; arithmetic reports the widest operand type.
+func binaryResultType(e *parser.BinaryExpr, resolve columnTypeResolver) string {
+	switch e.Op {
+	case lexer.TokenEq, lexer.TokenNeq, lexer.TokenLt, lexer.TokenLte, lexer.TokenGt, lexer.TokenGte,
+		lexer.TokenAND, lexer.TokenOR:
+		return "BOOLEAN"
+	case lexer.TokenConcat:
+		return "TEXT"
+	case lexer.TokenBitAnd, lexer.TokenBitOr, lexer.TokenShiftLeft, lexer.TokenShiftRight:
+		return "BIGINT"
+	case lexer.TokenPlus, lexer.TokenMinus, lexer.TokenStar, lexer.TokenSlash, lexer.TokenPercent:
+		left := inferExprType(e.Left, resolve)
+		right := inferExprType(e.Right, resolve)
+		switch {
+		case isFloatCarrierTypeName(left) || isFloatCarrierTypeName(right):
+			return "DOUBLE"
+		case isIntegerTypeName(left) && isIntegerTypeName(right):
+			// Integer arithmetic is accumulated in int64, so report the wider
+			// integer type even when both operands are INTEGER.
+			return "BIGINT"
+		default:
+			return "TEXT"
+		}
+	default:
+		return "TEXT"
+	}
+}
+
+// unaryResultType reports the result type of a unary operator.
+func unaryResultType(e *parser.UnaryExpr, resolve columnTypeResolver) string {
+	switch e.Op {
+	case lexer.TokenNOT:
+		return "BOOLEAN"
+	case lexer.TokenBitNot:
+		return "BIGINT"
+	case lexer.TokenMinus, lexer.TokenPlus:
+		operand := inferExprType(e.Operand, resolve)
+		switch {
+		case isFloatCarrierTypeName(operand):
+			return "DOUBLE"
+		case isIntegerTypeName(operand):
+			return operand
+		default:
+			return "TEXT"
+		}
+	default:
+		return "TEXT"
+	}
+}
+
+// caseResultType reports the type of a CASE expression when every branch agrees
+// on the same non-TEXT type; otherwise the type is unknown.
+func caseResultType(e *parser.CaseExpr, resolve columnTypeResolver) string {
+	result := ""
+	for _, when := range e.Whens {
+		typeName := inferExprType(when.Result, resolve)
+		if result == "" {
+			result = typeName
+			continue
+		}
+		if !strings.EqualFold(result, typeName) {
+			return "TEXT"
+		}
+	}
+	if e.Else != nil {
+		typeName := inferExprType(e.Else, resolve)
+		if result == "" || !strings.EqualFold(result, typeName) {
+			return "TEXT"
+		}
+	}
+	if result == "" {
+		return "TEXT"
+	}
+	return result
+}
+
+// isIntegerTypeName reports whether a type name is an integer type.
+func isIntegerTypeName(name string) bool {
+	switch strings.ToUpper(strings.TrimSpace(name)) {
+	case "INTEGER", "INT", "TINYINT", "SMALLINT", "MEDIUMINT", "BIGINT":
+		return true
+	default:
+		return false
+	}
+}
+
+// isFloatCarrierTypeName reports whether a type name is carried as float64.
+// NUMERIC and DECIMAL have no exact decimal carrier in the engine: their values
+// round-trip through JSON as float64, so they behave like floating-point types.
+func isFloatCarrierTypeName(name string) bool {
+	switch strings.ToUpper(strings.TrimSpace(name)) {
+	case "REAL", "FLOAT", "DOUBLE", "NUMERIC", "DECIMAL":
+		return true
+	default:
+		return false
+	}
+}
+
+// isUnknownTypeName reports whether a schema type name carries no type
+// information, as used for derived-table and CTE columns.
+func isUnknownTypeName(name string) bool {
+	switch strings.ToUpper(strings.TrimSpace(name)) {
+	case "", "ANY", "UNKNOWN", "NULL":
+		return true
+	default:
+		return false
 	}
-	return types
 }

pkg/executor/result_types_test.go

@@ -1,6 +1,11 @@
 package executor
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
 
 func TestSelectColumnTypesFromSchema(t *testing.T) {
 	_, schema, table := newTestDB(t)
@@ -47,12 +52,14 @@ func TestSelectColumnTypesAlias(t *testing.T) {
 	}
 }
 
-func TestSelectColumnTypesUnknownExpression(t *testing.T) {
+func TestSelectColumnTypesArithmeticExpression(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
 	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+	// Integer arithmetic accumulates in int64, so it reports the wider
+	// integer type rather than the schema type of its operands.
 	res := execMust(t, e, "SELECT id + 1, id * 2 FROM t")
-	if len(res.ColumnTypes) != 2 || res.ColumnTypes[0] != "TEXT" || res.ColumnTypes[1] != "TEXT" {
+	if len(res.ColumnTypes) != 2 || res.ColumnTypes[0] != "BIGINT" || res.ColumnTypes[1] != "BIGINT" {
 		t.Fatalf("unexpected column types %v", res.ColumnTypes)
 	}
 }
@@ -122,3 +129,179 @@ func TestUUIDColumnRoundTrip(t *testing.T) {
 		t.Fatalf("column types = %v, want [UUID TEXT]", res.ColumnTypes)
 	}
 }
+
+// TestInferExprType pins the expression type inference behind result metadata.
+func TestInferExprType(t *testing.T) {
+	schema := &storage.Schema{Name: "t", Columns: []storage.Column{
+		{Name: "id", Type: "INTEGER"},
+		{Name: "big", Type: "BIGINT"},
+		{Name: "small", Type: "SMALLINT"},
+		{Name: "price", Type: "NUMERIC"},
+		{Name: "score", Type: "DOUBLE"},
+		{Name: "flag", Type: "BOOLEAN"},
+		{Name: "name", Type: "TEXT"},
+		{Name: "data", Type: "BLOB"},
+		{Name: "created_at", Type: "TIMESTAMP"},
+		{Name: "day", Type: "DATE"},
+	}}
+	resolve := schemaTypeResolver(schema)
+
+	cases := []struct {
+		expr string
+		want string
+	}{
+		// Direct column references keep the declared schema type.
+		{"id", "INTEGER"},
+		{"big", "BIGINT"},
+		{"small", "SMALLINT"},
+		{"price", "NUMERIC"},
+		{"score", "DOUBLE"},
+		{"flag", "BOOLEAN"},
+		{"name", "TEXT"},
+		{"data", "BLOB"},
+		{"created_at", "TIMESTAMP"},
+		{"day", "DATE"},
+		{"rowid", "BIGINT"},
+		{"missing", "TEXT"},
+
+		// Literals.
+		{"1", "INTEGER"},
+		{"2147483647", "INTEGER"},
+		{"2147483648", "BIGINT"},
+		{"1.5", "DOUBLE"},
+		{"'text'", "TEXT"},
+		{"TRUE", "BOOLEAN"},
+		{"FALSE", "BOOLEAN"},
+		{"X'00'", "BLOB"},
+		{"NULL", "TEXT"},
+
+		// Comparisons and boolean expressions.
+		{"id = 1", "BOOLEAN"},
+		{"id <> 1", "BOOLEAN"},
+		{"id > 0 AND flag", "BOOLEAN"},
+		{"flag OR id < 0", "BOOLEAN"},
+		{"NOT flag", "BOOLEAN"},
+		{"name LIKE 'a%'", "BOOLEAN"},
+		{"id BETWEEN 1 AND 3", "BOOLEAN"},
+		{"id IN (1, 2)", "BOOLEAN"},
+		{"name IS NULL", "BOOLEAN"},
+		{"id IS DISTINCT FROM 1", "BOOLEAN"},
+
+		// Aggregates.
+		{"count(*)", "BIGINT"},
+		{"count(id)", "BIGINT"},
+		{"sum(id)", "BIGINT"},
+		{"sum(price)", "DOUBLE"},
+		{"sum(score)", "DOUBLE"},
+		{"avg(id)", "DOUBLE"},
+		{"avg(score)", "DOUBLE"},
+		{"total(id)", "DOUBLE"},
+		{"min(created_at)", "TIMESTAMP"},
+		{"max(name)", "TEXT"},
+
+		// Arithmetic, concatenation, and bitwise operators.
+		{"id + 1", "BIGINT"},
+		{"score + 1", "DOUBLE"},
+		{"price + 1", "DOUBLE"},
+		{"name || 'x'", "TEXT"},
+		{"id & 3", "BIGINT"},
+		{"-id", "INTEGER"},
+		{"-1.5", "DOUBLE"},
+		{"(id)", "INTEGER"},
+
+		// Casts report the target type the executor converts to.
+		{"CAST(id AS BIGINT)", "BIGINT"},
+		{"CAST(id AS SMALLINT)", "BIGINT"},
+		{"CAST(id AS TEXT)", "TEXT"},
+		{"CAST(name AS BLOB)", "BLOB"},
+		{"CAST(name AS DOUBLE)", "DOUBLE"},
+		{"CAST(id AS REAL)", "DOUBLE"},
+		{"CAST(id AS INTEGER)", "BIGINT"},
+		{"CAST(id AS TINYINT)", "BIGINT"},
+		{"CAST(id AS DATE)", "TEXT"},
+		{"CAST(id AS NUMERIC)", "TEXT"},
+
+		// NUMERIC/DECIMAL have no exact decimal carrier: their values are
+		// float64, so expressions over them report DOUBLE.
+		{"price + price", "DOUBLE"},
+		{"-price", "DOUBLE"},
+
+		// CASE keeps a type only when every branch agrees.
+		{"CASE WHEN flag THEN id ELSE 0 END", "INTEGER"},
+		{"CASE WHEN flag THEN id ELSE name END", "TEXT"},
+
+		// Functions whose result type is not modeled stay TEXT.
+		{"lower(name)", "TEXT"},
+	}
+
+	for _, tc := range cases {
+		stmt := parse(t, "SELECT "+tc.expr).(*parser.SelectStmt)
+		if got := inferExprType(stmt.Columns[0].Expr, resolve); got != tc.want {
+			t.Errorf("inferExprType(%q) = %q, want %q", tc.expr, got, tc.want)
+		}
+	}
+}
+
+// TestSelectColumnTypesExpressions verifies that the executor attaches inferred
+// types to non-column projections, including over an empty result set.
+func TestSelectColumnTypesExpressions(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, created_at TIMESTAMP, price NUMERIC, small SMALLINT)")
+	execMust(t, e, "INSERT INTO t VALUES (1, '2024-05-06 07:08:09', 19.5, 7)")
+
+	cases := []struct {
+		sql  string
+		want []string
+	}{
+		{"SELECT count(*) FROM t", []string{"BIGINT"}},
+		{"SELECT sum(id) FROM t", []string{"BIGINT"}},
+		{"SELECT sum(price) FROM t", []string{"DOUBLE"}},
+		{"SELECT avg(id) FROM t", []string{"DOUBLE"}},
+		{"SELECT avg(price) FROM t", []string{"DOUBLE"}},
+		{"SELECT small + 1 FROM t", []string{"BIGINT"}},
+		{"SELECT price + 1 FROM t", []string{"DOUBLE"}},
+		{"SELECT id = 1 FROM t", []string{"BOOLEAN"}},
+		{"SELECT 'text' FROM t", []string{"TEXT"}},
+		{"SELECT 42 FROM t", []string{"INTEGER"}},
+		{"SELECT 1.5 FROM t", []string{"DOUBLE"}},
+		{"SELECT CAST(id AS BIGINT) FROM t", []string{"BIGINT"}},
+		{"SELECT created_at FROM t GROUP BY created_at", []string{"TIMESTAMP"}},
+		{"SELECT id FROM t UNION SELECT id FROM t", []string{"INTEGER"}},
+		// The empty result still reports projection metadata.
+		{"SELECT id, created_at FROM t WHERE id = 999", []string{"INTEGER", "TIMESTAMP"}},
+		{"SELECT id + 1 FROM t WHERE id = 999", []string{"BIGINT"}},
+		{"SELECT count(*), max(created_at) FROM t WHERE id = 999", []string{"BIGINT", "TIMESTAMP"}},
+	}
+
+	for _, tc := range cases {
+		res := execMust(t, e, tc.sql)
+		if len(res.ColumnTypes) != len(tc.want) {
+			t.Errorf("%s: column types = %v, want %v", tc.sql, res.ColumnTypes, tc.want)
+			continue
+		}
+		for i := range tc.want {
+			if res.ColumnTypes[i] != tc.want[i] {
+				t.Errorf("%s: column types = %v, want %v", tc.sql, res.ColumnTypes, tc.want)
+				break
+			}
+		}
+	}
+}
+
+// TestSelectColumnTypesWithoutFrom covers projections with no schema at all.
+func TestSelectColumnTypesWithoutFrom(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+
+	res := execMust(t, e, "SELECT 1, 1.5, 'x', 1 = 1, CAST(1 AS SMALLINT)")
+	want := []string{"INTEGER", "DOUBLE", "TEXT", "BOOLEAN", "BIGINT"}
+	if len(res.ColumnTypes) != len(want) {
+		t.Fatalf("column types = %v, want %v", res.ColumnTypes, want)
+	}
+	for i := range want {
+		if res.ColumnTypes[i] != want[i] {
+			t.Fatalf("column types = %v, want %v", res.ColumnTypes, want)
+		}
+	}
+}

pkg/executor/returning.go

@@ -45,18 +45,11 @@ func returningProjection(cols []parser.SelectColumn, schema *storage.Schema) ([]
 	return names, types
 }
 
-// projectionType resolves the declared type of a direct column reference.
+// projectionType resolves the declared type of a RETURNING expression. Direct
+// column references use the schema type and expressions use the executor's type
+// inference, so RETURNING metadata matches the equivalent SELECT projection.
 func projectionType(expr parser.Expr, schema *storage.Schema) string {
-	ref, ok := expr.(*parser.ColumnRef)
-	if !ok {
-		return "TEXT"
-	}
-	for _, c := range schema.Columns {
-		if strings.EqualFold(c.Name, ref.Column) {
-			return c.Type
-		}
-	}
-	return "TEXT"
+	return inferExprType(expr, schemaTypeResolver(schema))
 }
 
 // returningResult evaluates a RETURNING projection over the affected rows and

pkg/lexer/lexer.go

@@ -171,6 +171,17 @@ func (l *Lexer) NextToken() Token {
 		if l.peekChar() == '-' {
 			// Line comment
 			tok = l.readLineComment()
+		} else if l.peekChar() == '>' {
+			l.readChar() // consume '-'
+			if l.peekChar() == '>' {
+				l.readChar() // consume the first '>'
+				tok.Type = TokenArrowText
+				tok.Literal = "->>"
+			} else {
+				tok.Type = TokenArrow
+				tok.Literal = "->"
+			}
+			l.readChar()
 		} else {
 			tok.Type = TokenMinus
 			tok.Literal = "-"

pkg/lexer/lexer_test.go

@@ -487,6 +487,60 @@ func TestLexerBlobLiteralWhitespaceAndErrors(t *testing.T) {
 	}
 }
 
+func TestLexerJSONArrowOperators(t *testing.T) {
+	tests := []struct {
+		input   string
+		literal string
+	}{
+		{"->", "->"},
+		{"->>", "->>"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.input, func(t *testing.T) {
+			tok := New(tt.input).NextToken()
+			want := TokenArrow
+			if tt.input == "->>" {
+				want = TokenArrowText
+			}
+			if tok.Type != want {
+				t.Fatalf("type = %v, want %v", tok.Type, want)
+			}
+			if tok.Literal != tt.literal {
+				t.Fatalf("literal = %q, want %q", tok.Literal, tt.literal)
+			}
+		})
+	}
+}
+
+func TestLexerJSONArrowOperatorsDoNotChangeMinusOrComments(t *testing.T) {
+	tests := []struct {
+		input    string
+		expected []TokenType
+	}{
+		{"1-2", []TokenType{TokenNumber, TokenMinus, TokenNumber, TokenEOF}},
+		{"-3", []TokenType{TokenMinus, TokenNumber, TokenEOF}},
+		{"-- comment", []TokenType{TokenComment, TokenEOF}},
+		{"a->'b'", []TokenType{TokenIdent, TokenArrow, TokenString, TokenEOF}},
+		{"a->>'b'", []TokenType{TokenIdent, TokenArrowText, TokenString, TokenEOF}},
+		{"a -> 'b' ->> 0", []TokenType{TokenIdent, TokenArrow, TokenString, TokenArrowText, TokenNumber, TokenEOF}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.input, func(t *testing.T) {
+			tokens := New(tt.input).Tokenize()
+			if len(tokens) != len(tt.expected) {
+				t.Fatalf("expected %d tokens, got %d: %v", len(tt.expected), len(tokens), tokens)
+			}
+			for i, want := range tt.expected {
+				if tokens[i].Type != want {
+					t.Errorf("token[%d] = %v, want %v", i, tokens[i].Type, want)
+				}
+			}
+		})
+	}
+}
+
 func BenchmarkLexer(b *testing.B) {
 	input := `
 		SELECT u.id, u.name, u.email, COUNT(o.id) as order_count

pkg/lexer/token.go

@@ -23,6 +23,8 @@ const (
 	TokenSlash      // /
 	TokenPercent    // %
 	TokenConcat     // ||
+	TokenArrow      // ->
+	TokenArrowText  // ->>
 	TokenEq         // =
 	TokenNeq        // <> or !=
 	TokenLt         // <
@@ -402,6 +404,8 @@ var tokenNames = map[TokenType]string{
 	TokenSlash:      "/",
 	TokenPercent:    "%",
 	TokenConcat:     "||",
+	TokenArrow:      "->",
+	TokenArrowText:  "->>",
 	TokenEq:         "=",
 	TokenNeq:        "<>",
 	TokenLt:         "<",

pkg/parser/format.go

@@ -141,6 +141,10 @@ func operatorString(op lexer.TokenType) string {
 		return "%"
 	case lexer.TokenConcat:
 		return "||"
+	case lexer.TokenArrow:
+		return "->"
+	case lexer.TokenArrowText:
+		return "->>"
 	case lexer.TokenEq:
 		return "="
 	case lexer.TokenNeq:

pkg/parser/json_operator_test.go

@@ -0,0 +1,114 @@
+package parser
+
+import (
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+)
+
+func TestParseJSONArrowChainingIsLeftAssociative(t *testing.T) {
+	expr := parseExpr(t, "doc -> 'a' -> 0")
+
+	outer, ok := expr.(*BinaryExpr)
+	if !ok {
+		t.Fatalf("expected *BinaryExpr, got %T", expr)
+	}
+	if outer.Op != lexer.TokenArrow {
+		t.Fatalf("outer operator = %v, want ->", outer.Op)
+	}
+	lit, ok := outer.Right.(*LiteralExpr)
+	if !ok || lit.Value != "0" {
+		t.Fatalf("outer right = %#v, want literal 0", outer.Right)
+	}
+
+	inner, ok := outer.Left.(*BinaryExpr)
+	if !ok {
+		t.Fatalf("outer left = %T, want *BinaryExpr", outer.Left)
+	}
+	if inner.Op != lexer.TokenArrow {
+		t.Fatalf("inner operator = %v, want ->", inner.Op)
+	}
+	ref, ok := inner.Left.(*ColumnRef)
+	if !ok || ref.Table != "" || ref.Column != "doc" {
+		t.Fatalf("inner left = %#v, want column doc", inner.Left)
+	}
+	key, ok := inner.Right.(*LiteralExpr)
+	if !ok || key.Value != "a" {
+		t.Fatalf("inner right = %#v, want literal 'a'", inner.Right)
+	}
+}
+
+func TestParseJSONArrowPrecedence(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   string
+		root    lexer.TokenType
+		childOp lexer.TokenType
+	}{
+		{"concatenation binds looser", "doc -> 'a' || 'x'", lexer.TokenConcat, lexer.TokenArrow},
+		{"addition binds looser", "doc -> 'a' + 1", lexer.TokenPlus, lexer.TokenArrow},
+		{"multiplication binds looser", "doc ->> 'a' * 2", lexer.TokenStar, lexer.TokenArrowText},
+		{"bitwise binds looser", "doc -> 'a' & 1", lexer.TokenBitAnd, lexer.TokenArrow},
+		{"comparison binds looser", "doc ->> 'a' = 'x'", lexer.TokenEq, lexer.TokenArrowText},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			expr := parseExpr(t, tt.input)
+			root, ok := expr.(*BinaryExpr)
+			if !ok {
+				t.Fatalf("expected *BinaryExpr, got %T", expr)
+			}
+			if root.Op != tt.root {
+				t.Fatalf("root operator = %v, want %v", root.Op, tt.root)
+			}
+			child, ok := root.Left.(*BinaryExpr)
+			if !ok {
+				t.Fatalf("root left = %T, want *BinaryExpr", root.Left)
+			}
+			if child.Op != tt.childOp {
+				t.Fatalf("left child operator = %v, want %v", child.Op, tt.childOp)
+			}
+		})
+	}
+}
+
+func TestParseJSONArrowBindsLooserThanUnary(t *testing.T) {
+	expr := parseExpr(t, "-doc -> 'a'")
+	arrow, ok := expr.(*BinaryExpr)
+	if !ok || arrow.Op != lexer.TokenArrow {
+		t.Fatalf("expected root ->, got %#v", expr)
+	}
+	neg, ok := arrow.Left.(*UnaryExpr)
+	if !ok || neg.Op != lexer.TokenMinus {
+		t.Fatalf("arrow left = %#v, want unary minus", arrow.Left)
+	}
+
+	// The right operand may be a negative array index.
+	expr = parseExpr(t, "doc -> -1")
+	arrow, ok = expr.(*BinaryExpr)
+	if !ok || arrow.Op != lexer.TokenArrow {
+		t.Fatalf("expected root ->, got %#v", expr)
+	}
+	neg, ok = arrow.Right.(*UnaryExpr)
+	if !ok || neg.Op != lexer.TokenMinus {
+		t.Fatalf("arrow right = %#v, want unary minus", arrow.Right)
+	}
+}
+
+func TestParseJSONArrowRespectsParentheses(t *testing.T) {
+	expr := parseExpr(t, "(doc -> 'a') ->> 'b'")
+
+	outer, ok := expr.(*BinaryExpr)
+	if !ok || outer.Op != lexer.TokenArrowText {
+		t.Fatalf("expected root ->>, got %#v", expr)
+	}
+	paren, ok := outer.Left.(*ParenExpr)
+	if !ok {
+		t.Fatalf("outer left = %T, want *ParenExpr", outer.Left)
+	}
+	inner, ok := paren.Expr.(*BinaryExpr)
+	if !ok || inner.Op != lexer.TokenArrow {
+		t.Fatalf("parenthesized expression = %#v, want ->", paren.Expr)
+	}
+}

pkg/parser/parser.go

@@ -2562,12 +2562,38 @@ func (p *Parser) parseAddExpr() (Expr, error) {
 }
 
 func (p *Parser) parseMulExpr() (Expr, error) {
-	left, err := p.parseUnaryExpr()
+	left, err := p.parseJSONExpr()
 	if err != nil {
 		return nil, err
 	}
 
 	for p.curTokenIs(lexer.TokenStar) || p.curTokenIs(lexer.TokenSlash) || p.curTokenIs(lexer.TokenPercent) {
+		op := p.curToken.Type
+		p.nextToken()
+		right, err := p.parseJSONExpr()
+		if err != nil {
+			return nil, err
+		}
+		left = &BinaryExpr{Left: left, Op: op, Right: right}
+	}
+
+	return left, nil
+}
+
+// parseJSONExpr parses the JSON extraction operators -> and ->>.
+//
+// Precedence: they bind tighter than multiplicative, additive/concatenation,
+// bitwise, and comparison operators, and looser than unary -, +, and ~. So
+// "a -> 'x' || 'y'" concatenates the extracted value, "a -> 'x' = 1" compares
+// it, and "-a -> 'x'" negates a before extraction. They are left-associative,
+// so "a -> 'x' -> 0" extracts 'x' and then indexes the result at 0.
+func (p *Parser) parseJSONExpr() (Expr, error) {
+	left, err := p.parseUnaryExpr()
+	if err != nil {
+		return nil, err
+	}
+
+	for p.curTokenIs(lexer.TokenArrow) || p.curTokenIs(lexer.TokenArrowText) {
 		op := p.curToken.Type
 		p.nextToken()
 		right, err := p.parseUnaryExpr()

pkg/pgserver/connection.go

@@ -1366,13 +1366,26 @@ func (c *Connection) writeMessage(msgType byte, data []byte) error {
 	return WriteMessage(c.writer, msgType, data)
 }
 
-// getOIDForType returns PostgreSQL OID for type
+// getOIDForType returns PostgreSQL OID for type.
+//
+// Types narrower than the value the engine carries are reported as the carrier
+// type, because the engine does not range-check stored or cast values: a value
+// that does not fit int2/int4 or float4 would fail client-side decoding rather
+// than having been rejected on write. INTEGER and REAL keep their declared
+// mappings, which predate this rule and are what existing clients expect.
 func (c *Connection) getOIDForType(typeName string) int32 {
 	switch strings.ToUpper(typeName) {
-	case "INTEGER", "INT":
+	case "SMALLINT", "INT2", "TINYINT", "MEDIUMINT":
+		// Carried as int64 and not range-checked, so int8 is the honest type.
+		return 20 // INT8OID
+	case "INTEGER", "INT", "INT4":
 		return 23 // INT4OID
-	case "BIGINT":
+	case "BIGINT", "INT8":
 		return 20 // INT8OID
+	case "NUMERIC", "DECIMAL":
+		// Values round-trip through JSON as float64; there is no exact decimal
+		// carrier, so NUMERIC is not advertised.
+		return 701 // FLOAT8OID
 	case "TEXT", "VARCHAR", "CHAR", "UUID":
 		// UUID is stored as text (SQLite-style), so it reports TEXTOID rather
 		// than the native PG UUID OID.
@@ -1396,13 +1409,17 @@ func (c *Connection) getOIDForType(typeName string) int32 {
 	}
 }
 
-// getTypeSizeForType returns type size
+// getTypeSizeForType returns type size in bytes, or -1 for variable length.
 func (c *Connection) getTypeSizeForType(typeName string) int16 {
 	switch strings.ToUpper(typeName) {
-	case "INTEGER", "INT":
+	case "SMALLINT", "INT2", "TINYINT", "MEDIUMINT":
+		return 8 // Reported as int8; see getOIDForType.
+	case "INTEGER", "INT", "INT4":
 		return 4
-	case "BIGINT":
+	case "BIGINT", "INT8":
 		return 8
+	case "NUMERIC", "DECIMAL":
+		return 8 // Reported as float8; see getOIDForType.
 	case "REAL", "FLOAT":
 		return 4
 	case "DOUBLE":

pkg/pgserver/json_contract_test.go

@@ -32,9 +32,9 @@ func jsonErrorField(t *testing.T, msg *Message, want byte) string {
 	return ""
 }
 
-// The supported JSON contract is TEXT storage plus the JSON1 function surface.
-// JSON/JSONB type names are accepted as text-affinity aliases; there is no
-// native binary jsonb type or ->> operator.
+// The supported JSON contract is TEXT storage plus the JSON1 function surface
+// and the -> / ->> operators. JSON/JSONB type names are accepted as
+// text-affinity aliases; there is no native binary jsonb type.
 func TestJSONTextContractOverSimpleQuery(t *testing.T) {
 	c, client := newDataConnection(t)
 	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE settings (id INTEGER PRIMARY KEY, body TEXT)"), 0)})
@@ -82,10 +82,23 @@ func TestJSONInvalidInputOverSimpleQuery(t *testing.T) {
 	}
 }
 
-func TestJSONArrowOperatorUnsupported(t *testing.T) {
+func TestJSONArrowOperatorsOverSimpleQuery(t *testing.T) {
 	c, client := newDataConnection(t)
-	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT '{"a":1}' -> '$.a'`), 0)})
-	if len(msgs) == 0 || msgs[0].Type != MsgErrorResponse {
-		t.Fatalf("expected a parse error for the -> operator, got %v", msgs)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE settings (id INTEGER PRIMARY KEY, body TEXT)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`INSERT INTO settings (id, body) VALUES (1, '{"theme":"dark","widgets":[1,2]}')`), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT body ->> 'theme', body -> 'widgets' -> 0 FROM settings WHERE id = 1`), 0)})
+	values := dataRowValues(t, msgs[1])
+	if len(values) != 2 || values[0] != "dark" || values[1] != "1" {
+		t.Fatalf("arrow operators = %v, want [dark 1]", values)
+	}
+
+	// Malformed JSON through an arrow operator is the same error json_extract throws.
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT 'not json' ->> 'a'`), 0)})
+	if len(msgs) != 2 || msgs[0].Type != MsgErrorResponse {
+		t.Fatalf("expected ErrorResponse+ReadyForQuery, got %v", msgs)
+	}
+	if msg := jsonErrorField(t, msgs[0], ErrorFieldMessage); !strings.Contains(msg, "malformed JSON") {
+		t.Fatalf("error message = %q, want a malformed-JSON error", msg)
 	}
 }

pkg/pgserver/oid_test.go

@@ -3,6 +3,8 @@ package pgserver
 import (
 	"bytes"
 	"encoding/binary"
+	"net"
+	"strconv"
 	"testing"
 
 	"github.com/danfragoso/pizzasql-next/pkg/executor"
@@ -15,12 +17,30 @@ func TestGetOIDForTypeDatetime(t *testing.T) {
 		"DATETIME":  1184,
 		"TIMESTAMP": 1184,
 		"DATE":      1082,
-		"BIGINT":    20,
+		// Integer types narrower than the int64 carrier report int8 because the
+		// engine does not range-check stored values.
+		"SMALLINT":  20,
+		"INT2":      20,
+		"TINYINT":   20,
+		"MEDIUMINT": 20,
 		"INTEGER":   23,
 		"INT":       23,
-		"TEXT":      25,
-		"UUID":      25,
-		"BOOLEAN":   16,
+		"INT4":      23,
+		"BIGINT":    20,
+		"INT8":      20,
+		// NUMERIC/DECIMAL carry float64 values, so they report float8.
+		"NUMERIC": 701,
+		"DECIMAL": 701,
+		"REAL":    700,
+		"FLOAT":   700,
+		"DOUBLE":  701,
+		"TEXT":    25,
+		"VARCHAR": 25,
+		"CHAR":    25,
+		"UUID":    25,
+		"BOOLEAN": 16,
+		"BOOL":    16,
+		"BLOB":    17,
 	}
 	for name, want := range cases {
 		if got := c.getOIDForType(name); got != want {
@@ -108,3 +128,303 @@ func rowDescriptionOIDs(t *testing.T, data []byte) []int32 {
 	}
 	return oids
 }
+
+func TestGetTypeSizeForType(t *testing.T) {
+	c := &Connection{}
+	cases := map[string]int16{
+		"SMALLINT":  8,
+		"TINYINT":   8,
+		"MEDIUMINT": 8,
+		"INTEGER":   4,
+		"BIGINT":    8,
+		"REAL":      4,
+		"DOUBLE":    8,
+		"NUMERIC":   8,
+		"DECIMAL":   8,
+		"BOOLEAN":   1,
+		"DATE":      4,
+		"TIMESTAMP": 8,
+		"TEXT":      -1,
+		"BLOB":      -1,
+	}
+	for name, want := range cases {
+		if got := c.getTypeSizeForType(name); got != want {
+			t.Errorf("getTypeSizeForType(%q) = %d, want %d", name, got, want)
+		}
+	}
+}
+
+// projectionOIDs runs sql and returns the type OIDs in its RowDescription.
+func projectionOIDs(t *testing.T, c *Connection, client net.Conn, sql string) []int32 {
+	t.Helper()
+	msgs := resultMessages(runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(sql), 0)}))
+	return rowDescriptionOIDs(t, findRowDescription(t, msgs).Data)
+}
+
+// TestRowDescriptionProjectionOIDs asserts the wire OIDs for column
+// projections, expression projections, and empty result sets.
+func TestRowDescriptionProjectionOIDs(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`CREATE TABLE typed (
+			i INTEGER PRIMARY KEY,
+			big BIGINT,
+			small SMALLINT,
+			price NUMERIC,
+			flag BOOLEAN,
+			name TEXT,
+			data BLOB,
+			created_at TIMESTAMP,
+			day DATE
+		)`), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`INSERT INTO typed VALUES (1, 9000000000, 7, 19.5, TRUE, 'x', X'00', '2024-05-06 07:08:09', '2024-05-06')`), 0)})
+
+	cases := []struct {
+		name string
+		sql  string
+		want []int32
+	}{
+		{
+			name: "columns",
+			sql:  "SELECT i, big, small, price, flag, name, data, created_at, day FROM typed",
+			want: []int32{23, 20, 20, 701, 16, 25, 17, 1184, 1082},
+		},
+		{name: "count star", sql: "SELECT count(*) FROM typed", want: []int32{20}},
+		{name: "sum integer", sql: "SELECT sum(i) FROM typed", want: []int32{20}},
+		{name: "sum numeric", sql: "SELECT sum(price) FROM typed", want: []int32{701}},
+		{name: "avg integer", sql: "SELECT avg(i) FROM typed", want: []int32{701}},
+		{name: "comparison", sql: "SELECT i = 1 FROM typed", want: []int32{16}},
+		{name: "string literal", sql: "SELECT 'text'", want: []int32{25}},
+		{name: "integer literal", sql: "SELECT 42", want: []int32{23}},
+		{name: "wide integer literal", sql: "SELECT 4294967296", want: []int32{20}},
+		{name: "numeric literal", sql: "SELECT 1.5", want: []int32{701}},
+		{name: "cast", sql: "SELECT CAST(i AS BIGINT) FROM typed", want: []int32{20}},
+		{name: "narrow integer cast", sql: "SELECT CAST(i AS SMALLINT) FROM typed", want: []int32{20}},
+		{
+			name: "empty result",
+			sql:  "SELECT i, created_at, i + 1 FROM typed WHERE i = 999",
+			want: []int32{23, 1184, 20},
+		},
+		{
+			name: "aggregate empty result",
+			sql:  "SELECT count(*), max(created_at) FROM typed WHERE i = 999",
+			want: []int32{20, 1184},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := projectionOIDs(t, c, client, tc.sql)
+			if len(got) != len(tc.want) {
+				t.Fatalf("%s: OIDs = %v, want %v", tc.sql, got, tc.want)
+			}
+			for i := range tc.want {
+				if got[i] != tc.want[i] {
+					t.Fatalf("%s: OIDs = %v, want %v", tc.sql, got, tc.want)
+				}
+			}
+		})
+	}
+}
+
+// rowDescriptionColumns decodes the column names from a RowDescription message.
+func rowDescriptionColumns(t *testing.T, data []byte) []string {
+	t.Helper()
+	if len(data) < 2 {
+		t.Fatalf("row description too short")
+	}
+	count := int(binary.BigEndian.Uint16(data[:2]))
+	pos := 2
+	names := make([]string, 0, count)
+	for i := 0; i < count; i++ {
+		end := bytes.IndexByte(data[pos:], 0)
+		if end < 0 {
+			t.Fatalf("unterminated column name")
+		}
+		names = append(names, string(data[pos:pos+end]))
+		pos += end + 1 // name
+		pos += 4       // table OID
+		pos += 2       // column attribute number
+		pos += 4       // type OID
+		pos += 2       // type size
+		pos += 4       // type modifier
+		pos += 2       // format code
+	}
+	return names
+}
+
+// countMessageType counts messages of a type in a query response.
+func countMessageType(msgs []*Message, msgType byte) int {
+	count := 0
+	for _, msg := range msgs {
+		if msg.Type == msgType {
+			count++
+		}
+	}
+	return count
+}
+
+// TestFloatCarrierOIDsMatchValues asserts that projections carried as float64
+// are advertised as float8 and that their text values parse as float8. NUMERIC
+// (1700) is never advertised because the engine has no exact decimal carrier.
+func TestFloatCarrierOIDsMatchValues(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`CREATE TABLE floats (i INTEGER PRIMARY KEY, price NUMERIC, amount DECIMAL, score DOUBLE)`), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`INSERT INTO floats VALUES (1, 19.5, 3.25, 0.1)`), 0)})
+
+	cases := []struct {
+		name string
+		sql  string
+	}{
+		{name: "numeric column", sql: "SELECT price FROM floats"},
+		{name: "decimal column", sql: "SELECT amount FROM floats"},
+		{name: "double column", sql: "SELECT score FROM floats"},
+		{name: "avg integer", sql: "SELECT avg(i) FROM floats"},
+		{name: "avg numeric", sql: "SELECT avg(price) FROM floats"},
+		{name: "sum numeric", sql: "SELECT sum(price) FROM floats"},
+		{name: "decimal literal", sql: "SELECT 1.5"},
+		{name: "float arithmetic", sql: "SELECT price + 0.5 FROM floats"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			msgs := resultMessages(runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(tc.sql), 0)}))
+			oids := rowDescriptionOIDs(t, findRowDescription(t, msgs).Data)
+			if len(oids) != 1 {
+				t.Fatalf("%s: OIDs = %v, want one column", tc.sql, oids)
+			}
+			if oids[0] == 1700 {
+				t.Fatalf("%s: OID = 1700 (numeric) for a float64 carrier", tc.sql)
+			}
+			if oids[0] != 701 {
+				t.Fatalf("%s: OID = %d, want 701 (float8)", tc.sql, oids[0])
+			}
+			values := dataRowValues(t, msgs[1])
+			if len(values) != 1 {
+				t.Fatalf("%s: values = %v, want one value", tc.sql, values)
+			}
+			if _, err := strconv.ParseFloat(values[0], 64); err != nil {
+				t.Fatalf("%s: float8 value %q does not parse: %v", tc.sql, values[0], err)
+			}
+		})
+	}
+}
+
+// TestNarrowIntegerOIDsUseCarrierType pins the chosen narrow-integer behavior:
+// the engine carries int64 without range checks, so SMALLINT/TINYINT/MEDIUMINT
+// report int8 (20) rather than a width a stored value may not fit.
+func TestNarrowIntegerOIDsUseCarrierType(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`CREATE TABLE narrow (id INTEGER PRIMARY KEY, small SMALLINT, tiny TINYINT, medium MEDIUMINT)`), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`INSERT INTO narrow VALUES (1, 40000, 200, 9000000)`), 0)})
+
+	cases := []struct {
+		name  string
+		sql   string
+		value string
+	}{
+		{name: "smallint column", sql: "SELECT small FROM narrow", value: "40000"},
+		{name: "tinyint column", sql: "SELECT tiny FROM narrow", value: "200"},
+		{name: "mediumint column", sql: "SELECT medium FROM narrow", value: "9000000"},
+		{name: "cast smallint", sql: "SELECT CAST(small AS SMALLINT) FROM narrow", value: "40000"},
+		{name: "cast integer", sql: "SELECT CAST(small AS INTEGER) FROM narrow", value: "40000"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			msgs := resultMessages(runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(tc.sql), 0)}))
+			oids := rowDescriptionOIDs(t, findRowDescription(t, msgs).Data)
+			if len(oids) != 1 || oids[0] != 20 {
+				t.Fatalf("%s: OIDs = %v, want [20] (int8 carrier)", tc.sql, oids)
+			}
+			values := dataRowValues(t, msgs[1])
+			if len(values) != 1 || values[0] != tc.value {
+				t.Fatalf("%s: values = %v, want [%s]", tc.sql, values, tc.value)
+			}
+			if _, err := strconv.ParseInt(values[0], 10, 64); err != nil {
+				t.Fatalf("%s: int8 value %q does not parse: %v", tc.sql, values[0], err)
+			}
+		})
+	}
+}
+
+// TestEmptyResultWildcardMatchesNames guards the constant-false WHERE path:
+// wildcard projections must expand names and types into the same columns.
+func TestEmptyResultWildcardMatchesNames(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`CREATE TABLE star_t (id INTEGER PRIMARY KEY, created_at TIMESTAMP, name TEXT)`), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(
+		`CREATE TABLE star_b (b_id BIGINT PRIMARY KEY, score DOUBLE)`), 0)})
+
+	cases := []struct {
+		name      string
+		sql       string
+		wantNames []string
+		wantOIDs  []int32
+	}{
+		{
+			name:      "qualified star",
+			sql:       "SELECT star_t.* FROM star_t WHERE 1 = 0",
+			wantNames: []string{"id", "created_at", "name"},
+			wantOIDs:  []int32{23, 1184, 25},
+		},
+		{
+			name:      "aliased qualified star",
+			sql:       "SELECT s.* FROM star_t AS s WHERE 1 = 0",
+			wantNames: []string{"id", "created_at", "name"},
+			wantOIDs:  []int32{23, 1184, 25},
+		},
+		{
+			name:      "star",
+			sql:       "SELECT * FROM star_t WHERE 1 = 0",
+			wantNames: []string{"id", "created_at", "name"},
+			wantOIDs:  []int32{23, 1184, 25},
+		},
+		{
+			name:      "column and expression",
+			sql:       "SELECT star_t.name, star_t.id + 1 FROM star_t WHERE 1 = 0",
+			wantNames: []string{"name", "column2"},
+			wantOIDs:  []int32{25, 20},
+		},
+		{
+			name:      "join star",
+			sql:       "SELECT * FROM star_t JOIN star_b ON star_t.id = star_b.b_id WHERE 1 = 0",
+			wantNames: []string{"id", "created_at", "name", "b_id", "score"},
+			wantOIDs:  []int32{23, 1184, 25, 20, 701},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			msgs := resultMessages(runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(tc.sql), 0)}))
+			rd := findRowDescription(t, msgs)
+			names := rowDescriptionColumns(t, rd.Data)
+			oids := rowDescriptionOIDs(t, rd.Data)
+			if len(names) != len(tc.wantNames) {
+				t.Fatalf("%s: names = %v, want %v", tc.sql, names, tc.wantNames)
+			}
+			for i := range tc.wantNames {
+				if names[i] != tc.wantNames[i] {
+					t.Fatalf("%s: names = %v, want %v", tc.sql, names, tc.wantNames)
+				}
+			}
+			if len(oids) != len(tc.wantOIDs) {
+				t.Fatalf("%s: OIDs = %v, want %v", tc.sql, oids, tc.wantOIDs)
+			}
+			for i := range tc.wantOIDs {
+				if oids[i] != tc.wantOIDs[i] {
+					t.Fatalf("%s: OIDs = %v, want %v", tc.sql, oids, tc.wantOIDs)
+				}
+			}
+			if rows := countMessageType(msgs, MsgDataRow); rows != 0 {
+				t.Fatalf("%s: got %d data rows, want 0", tc.sql, rows)
+			}
+		})
+	}
+}

pkg/storage/schema.go

@@ -1276,10 +1276,14 @@ func (m *SchemaManager) renameDataKeys(oldName, newName string) error {
 	})
 }
 
-// repointIndexesLocked updates indexes whose table was renamed. The caller holds
-// m.mu; it reads the durable index list directly rather than calling the
-// self-locking ListIndexes helper.
-func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error {
+// updateIndexesOfTableLocked applies mutate to the durable definition of every
+// index on table and persists the definitions it changed. mutate reports
+// whether it modified the definition; unmodified indexes are left untouched.
+// The changed definitions are written in one batch, so a table's indexes never
+// split between the old and new shape. The caller holds m.mu; it reads the
+// durable index list directly rather than calling the self-locking ListIndexes
+// helper.
+func (m *SchemaManager) updateIndexesOfTableLocked(table string, mutate func(*Index) bool) error {
 	var names []string
 	if m.indexListCached {
 		names = append([]string(nil), m.indexListCache...)
@@ -1301,6 +1305,12 @@ func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error {
 		}
 	}
 
+	type indexUpdate struct {
+		index Index
+		key   string
+		value []byte
+	}
+	var updates []indexUpdate
 	for _, name := range names {
 		key := m.indexKey(name)
 		var data string
@@ -1319,27 +1329,74 @@ func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error {
 		if err := json.Unmarshal([]byte(data), &idx); err != nil {
 			return err
 		}
-		if !strings.EqualFold(idx.Table, oldName) {
+		if !strings.EqualFold(idx.Table, table) {
+			continue
+		}
+		if !mutate(&idx) {
 			continue
 		}
-		idx.Table = newName
 		updated, err := json.Marshal(&idx)
 		if err != nil {
 			return err
 		}
-		if err := m.pool.WithClient(func(c *KVClient) error {
-			return c.Write(key, string(updated))
-		}); err != nil {
-			return err
-		}
-		if cached, ok := m.indexCache[strings.ToLower(name)]; ok {
-			cached.Table = newName
+		updates = append(updates, indexUpdate{index: idx, key: key, value: updated})
+	}
+	if len(updates) == 0 {
+		return nil
+	}
+
+	ops := make([]BatchOp, 0, len(updates))
+	for _, update := range updates {
+		ops = append(ops, BatchOp{Op: batchPut, Key: []byte(update.key), Value: update.value})
+	}
+	if err := m.pool.WithClient(func(c *KVClient) error {
+		_, err := c.BatchWrite(ops, nil)
+		return err
+	}); err != nil {
+		return fmt.Errorf("update index definitions for table %s: %w", table, err)
+	}
+
+	for _, update := range updates {
+		if cached, ok := m.indexCache[strings.ToLower(update.index.Name)]; ok {
+			*cached = update.index
 		}
 	}
 	return nil
 }
 
-// RenameColumn renames a column in a table.
+// repointIndexesLocked updates indexes whose table was renamed.
+func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error {
+	return m.updateIndexesOfTableLocked(oldName, func(idx *Index) bool {
+		idx.Table = newName
+		return true
+	})
+}
+
+// renameIndexColumnsLocked rewrites the stored definition of every index on
+// table so a reference to oldName becomes newName. Index names stay unchanged:
+// they are independent identifiers that SQLite does not derive from column
+// names, and callers may have chosen them explicitly. A rename to an existing
+// column name is rejected before this runs. Expression columns keep their SQL
+// text; callers must reject a rename that such an expression depends on,
+// because the text is not rewritten.
+func (m *SchemaManager) renameIndexColumnsLocked(table, oldName, newName string) error {
+	return m.updateIndexesOfTableLocked(table, func(idx *Index) bool {
+		changed := false
+		for i := range idx.Columns {
+			// Expression columns persist SQL text, not a column reference; their
+			// text is left for the caller to drop and recreate the index.
+			if idx.Columns[i].Expression == "" && strings.EqualFold(idx.Columns[i].Name, oldName) {
+				idx.Columns[i].Name = newName
+				changed = true
+			}
+		}
+		return changed
+	})
+}
+
+// RenameColumn renames a column in a table. Column references in index
+// definitions are rewritten; expression-index text is not, so a SQL caller must
+// reject a rename that an expression depends on.
 func (m *SchemaManager) RenameColumn(table, oldName, newName string) error {
 	tableLock := m.tableLock(table)
 	tableLock.Lock()
@@ -1392,8 +1449,18 @@ func (m *SchemaManager) RenameColumn(table, oldName, newName string) error {
 		return err
 	}
 
-	// Update schema
-	return m.updateSchemaUnsafe(schema)
+	// Publish the renamed schema before the index definitions: reads of the new
+	// column name resolve against rewritten rows as soon as the schema lands.
+	// Index DDL is not transactional (as elsewhere in this engine), so a failure
+	// here leaves stale index metadata for the caller to drop and recreate.
+	if err := m.updateSchemaUnsafe(schema); err != nil {
+		return err
+	}
+
+	// Index metadata drives secondary-index values, UNIQUE enforcement,
+	// ON CONFLICT target resolution, and catalog introspection, so it must name
+	// the renamed column before any later statement uses the table.
+	return m.renameIndexColumnsLocked(table, oldName, newName)
 }
 
 // rewriteRows applies an idempotent name-keyed row transformation in bounded