package executor import ( "fmt" "math/rand" "sort" "strconv" "strings" "github.com/danfragoso/pizzasql-next/pkg/analyzer" "github.com/danfragoso/pizzasql-next/pkg/lexer" "github.com/danfragoso/pizzasql-next/pkg/parser" "github.com/danfragoso/pizzasql-next/pkg/storage" ) // Executor executes SQL statements. type Executor struct { schema *storage.SchemaManager table *storage.TableManager analyzer *analyzer.Analyzer catalog *analyzer.Catalog // Multi-database support attachedDatabases map[string]*DatabaseConnection // alias -> connection currentDatabase string // current database alias (default is "main") // Transaction state inTransaction bool savepoints []string // stack of savepoint names txLog []txLogEntry // transaction log for rollback // Subquery context for correlated subqueries outerRow storage.Row } // DatabaseConnection represents an attached database. type DatabaseConnection struct { Alias string Path string // Database path or identifier Schema *storage.SchemaManager Table *storage.TableManager } // txLogEntry represents a transaction log entry for rollback support. type txLogEntry struct { operation string // "INSERT", "UPDATE", "DELETE" table string key string oldData storage.Row // for UPDATE/DELETE, the original row data } // New creates a new executor. func New(schema *storage.SchemaManager, table *storage.TableManager) *Executor { catalog := analyzer.NewCatalog() executor := &Executor{ schema: schema, table: table, analyzer: analyzer.New(catalog), catalog: catalog, attachedDatabases: make(map[string]*DatabaseConnection), currentDatabase: "main", } // Register the main database executor.attachedDatabases["main"] = &DatabaseConnection{ Alias: "main", Path: schema.GetDatabaseName(), Schema: schema, Table: table, } return executor } // SyncCatalog synchronizes the analyzer catalog with the storage schema. func (e *Executor) SyncCatalog() error { tables, err := e.schema.ListTables() if err != nil { return err } for _, tableName := range tables { schema, err := e.schema.GetSchema(tableName) if err != nil { continue } // Drop table from catalog if it exists, then recreate with updated schema e.catalog.DropTable(tableName) e.catalog.CreateTable(schema.ToAnalyzerTableInfo()) } return nil } // Execute executes a SQL statement. func (e *Executor) Execute(stmt parser.Statement) (*Result, error) { // PRAGMA doesn't need analysis if pragma, ok := stmt.(*parser.PragmaStmt); ok { return e.executePragma(pragma) } // EXPLAIN doesn't need analysis if explain, ok := stmt.(*parser.ExplainStmt); ok { return e.executeExplain(explain) } // Transaction statements don't need analysis switch s := stmt.(type) { case *parser.BeginStmt: return e.executeBegin(s) case *parser.CommitStmt: return e.executeCommit(s) case *parser.RollbackStmt: return e.executeRollback(s) case *parser.SavepointStmt: return e.executeSavepoint(s) case *parser.ReleaseStmt: return e.executeRelease(s) case *parser.CreateIndexStmt: return e.executeCreateIndex(s) case *parser.DropIndexStmt: return e.executeDropIndex(s) case *parser.AttachStmt: return e.executeAttach(s) case *parser.DetachStmt: return e.executeDetach(s) } // Analyze first — create a fresh analyzer per call so concurrent requests // don't share mutable scope state (e.analyzer.scope would race otherwise). a := analyzer.New(e.catalog) if err := a.Analyze(stmt); err != nil { return nil, err } switch s := stmt.(type) { case *parser.SelectStmt: return e.executeSelect(s) case *parser.InsertStmt: return e.executeInsert(s) case *parser.UpdateStmt: return e.executeUpdate(s) case *parser.DeleteStmt: return e.executeDelete(s) case *parser.CreateTableStmt: return e.executeCreateTable(s) case *parser.DropTableStmt: return e.executeDropTable(s) case *parser.CreateIndexStmt: return e.executeCreateIndex(s) case *parser.DropIndexStmt: return e.executeDropIndex(s) case *parser.AlterTableStmt: return e.executeAlterTable(s) default: return nil, fmt.Errorf("unsupported statement type: %T", stmt) } } // executeSelect executes a SELECT statement. func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) { if len(stmt.From) == 0 { // SELECT without FROM (e.g., SELECT 1+1) return e.executeSelectExpr(stmt) } // Check if FROM clause is a subquery (derived table) if stmt.From[0].Subquery != nil { return e.executeSelectFromSubquery(stmt) } tableName := stmt.From[0].Name schema, err := e.schema.GetSchema(tableName) if err != nil { return nil, err } // Try to use index for WHERE clause var rows []storage.Row usedIndex := false if stmt.Where != nil { // Check if we can use an index colName, colValue, isEquality := e.extractIndexableCondition(stmt.Where) if isEquality { // Look for an index on this column indexes, _ := e.schema.ListTableIndexes(tableName) for _, idx := range indexes { if len(idx.Columns) == 1 && strings.EqualFold(idx.Columns[0].Name, colName) { // Use this index rows, err = e.table.SelectByIndex(tableName, idx.Name, colValue) if err == nil { usedIndex = true } break } } } } // Fall back to full table scan if no index used if !usedIndex { // Build filter function from WHERE clause // Only use filter during scan if there's NO alias (otherwise the filter won't have the right column names) var filter func(storage.Row) bool if stmt.Where != nil && stmt.From[0].Alias == "" { filter = func(row storage.Row) bool { val, err := e.evalExpr(stmt.Where, row) if err != nil { return false } return toBool(val) } } rows, err = e.table.Select(tableName, filter) } if err != nil { return nil, err } // Add table alias to rows if there's an explicit alias // This needs to happen BEFORE filtering so that the WHERE clause can reference the alias if stmt.From[0].Alias != "" { for i := range rows { rows[i] = e.addTableAlias(rows[i], stmt.From[0].Alias) } } // Apply WHERE clause filter if we have an alias (we couldn't filter during scan) if stmt.Where != nil && stmt.From[0].Alias != "" { var filtered []storage.Row for _, row := range rows { val, err := e.evalExpr(stmt.Where, row) if err != nil { continue // Skip rows that error } if toBool(val) { filtered = append(filtered, row) } } rows = filtered } // Handle JOINs if len(stmt.From) > 0 && stmt.From[0].Join != nil { rows, err = e.executeJoins(stmt.From[0], rows) if err != nil { return nil, err } } // Handle GROUP BY if len(stmt.GroupBy) > 0 { return e.executeGroupBy(stmt, rows, schema) } // Check for aggregate functions without GROUP BY hasAggregate := e.hasAggregates(stmt.Columns) if hasAggregate { return e.executeAggregateSelect(stmt, rows, schema) } // Apply ORDER BY if len(stmt.OrderBy) > 0 { e.sortRows(rows, stmt.OrderBy) } // Apply LIMIT/OFFSET if stmt.Offset != nil { offset := e.evalIntExpr(stmt.Offset) if offset < len(rows) { rows = rows[offset:] } else { rows = nil } } if stmt.Limit != nil { limit := e.evalIntExpr(stmt.Limit) if limit < len(rows) { rows = rows[:limit] } } // Build result result := NewResult("SELECT") // 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 { // Handle SELECT * - add all columns from schema for _, c := range schema.Columns { result.AddColumn(c.Name) } } else { result.AddColumn(fmt.Sprintf("column%d", i+1)) } } // Add rows - evaluate each select expression for _, row := range rows { values := make([]interface{}, 0) for _, col := range stmt.Columns { if col.Star { // For SELECT *, add all columns in order for _, c := range schema.Columns { if storage.IsRowIDColumn(c.Name) { values = append(values, row["_rowid_"]) } else { values = append(values, row[c.Name]) } } } else { // Evaluate the expression val, err := e.evalExpr(col.Expr, row) if err != nil { return nil, err } values = append(values, val) } } result.AddRow(values...) } // Apply DISTINCT if specified if stmt.Distinct { result.Rows = e.applyDistinct(result.Rows) } return result, nil } // executeSelectExpr executes a SELECT without FROM. func (e *Executor) executeSelectExpr(stmt *parser.SelectStmt) (*Result, error) { result := NewResult("SELECT") // Determine columns for i, col := range stmt.Columns { if col.Alias != "" { result.AddColumn(col.Alias) } else { result.AddColumn(fmt.Sprintf("column%d", i+1)) } } // Evaluate expressions values := make([]interface{}, len(stmt.Columns)) for i, col := range stmt.Columns { val, err := e.evalExpr(col.Expr, nil) if err != nil { return nil, err } values[i] = val } result.AddRow(values...) return result, nil } // executeSelectFromSubquery executes a SELECT with a subquery in FROM clause. func (e *Executor) executeSelectFromSubquery(stmt *parser.SelectStmt) (*Result, error) { // Execute the subquery to get the derived table subqueryResult, err := e.executeSelect(stmt.From[0].Subquery) if err != nil { 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] } derivedRows = append(derivedRows, row) } // Handle JOINs if present if stmt.From[0].Join != nil { derivedRows, err = e.executeJoin(stmt.From[0], derivedRows) if err != nil { return nil, err } } // Apply WHERE clause on derived table if stmt.Where != nil { filteredRows := make([]storage.Row, 0) for _, row := range derivedRows { val, err := e.evalExpr(stmt.Where, row) if err != nil { continue } if toBool(val) { filteredRows = append(filteredRows, row) } } derivedRows = filteredRows } // 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) } // 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) } // Apply ORDER BY if len(stmt.OrderBy) > 0 { e.sortRows(derivedRows, stmt.OrderBy) } // Apply LIMIT/OFFSET if stmt.Offset != nil { offset := e.evalIntExpr(stmt.Offset) if offset < len(derivedRows) { derivedRows = derivedRows[offset:] } else { derivedRows = nil } } if stmt.Limit != nil { limit := e.evalIntExpr(stmt.Limit) if limit < len(derivedRows) { derivedRows = derivedRows[:limit] } } // Build result 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 { result.AddColumn(colRef.Column) } else { 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 { val, err := e.evalExpr(col.Expr, row) if err != nil { return nil, err } values[i] = val } result.AddRow(values...) } } return result, nil } // 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") // Determine columns and evaluate aggregates for i, col := range stmt.Columns { if col.Alias != "" { result.AddColumn(col.Alias) } else if col.Star { result.AddColumn("*") } else { result.AddColumn(fmt.Sprintf("column%d", i+1)) } } // Calculate values values := make([]interface{}, len(stmt.Columns)) for i, col := range stmt.Columns { val, err := e.evalAggregateExpr(col.Expr, rows) if err != nil { return nil, err } values[i] = val } result.AddRow(values...) return result, nil } // executeGroupBy executes a GROUP BY query. func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, schema *storage.Schema) (*Result, error) { // Group rows groups := make(map[string][]storage.Row) for _, row := range rows { key := e.buildGroupKey(stmt.GroupBy, row) groups[key] = append(groups[key], row) } result := NewResult("SELECT") // Determine columns columnNames := make([]string, len(stmt.Columns)) for i, col := range stmt.Columns { if col.Alias != "" { columnNames[i] = col.Alias result.AddColumn(col.Alias) } else if ref, ok := col.Expr.(*parser.ColumnRef); ok { columnNames[i] = ref.Column result.AddColumn(ref.Column) } else { columnNames[i] = fmt.Sprintf("column%d", i+1) result.AddColumn(columnNames[i]) } } // Process each group for _, groupRows := range groups { // Apply HAVING if stmt.Having != nil { val, err := e.evalAggregateExpr(stmt.Having, groupRows) if err != nil { continue } if !toBool(val) { continue } } values := make([]interface{}, len(stmt.Columns)) for i, col := range stmt.Columns { if e.isAggregate(col.Expr) { val, err := e.evalAggregateExpr(col.Expr, groupRows) if err != nil { return nil, err } values[i] = val } else { // Use first row's value for non-aggregate columns val, err := e.evalExpr(col.Expr, groupRows[0]) if err != nil { return nil, err } values[i] = val } } result.AddRow(values...) } // Apply ORDER BY if len(stmt.OrderBy) > 0 { e.sortResultRows(result, stmt.OrderBy, stmt.Columns, columnNames) } // Apply LIMIT/OFFSET if stmt.Offset != nil { offset := e.evalIntExpr(stmt.Offset) if offset < len(result.Rows) { result.Rows = result.Rows[offset:] } else { result.Rows = nil } result.RowCount = len(result.Rows) } if stmt.Limit != nil { limit := e.evalIntExpr(stmt.Limit) if limit < len(result.Rows) { result.Rows = result.Rows[:limit] } result.RowCount = len(result.Rows) } return result, nil } // executeJoins recursively processes all JOIN clauses in a table reference. func (e *Executor) executeJoins(tableRef parser.TableRef, leftRows []storage.Row) ([]storage.Row, error) { if tableRef.Join == nil || tableRef.Join.Table == nil { return leftRows, nil } // Get the right table name and its data rightTableRef := tableRef.Join.Table rightTable := rightTableRef.Name rightRows, err := e.table.Select(rightTable, nil) if err != nil { return nil, err } // Perform the join between left and right var result []storage.Row leftTableName := tableRef.Name leftAlias := tableRef.Alias rightAlias := rightTableRef.Alias // If leftAlias is empty, use the table name if leftAlias == "" { leftAlias = leftTableName } if rightAlias == "" { rightAlias = rightTable } switch tableRef.Join.Type { case parser.JoinInner: for _, left := range leftRows { for _, right := range rightRows { merged := e.mergeRows(left, right, leftAlias, rightAlias) if tableRef.Join.Condition != nil { match, _ := e.evalExpr(tableRef.Join.Condition, merged) if toBool(match) { result = append(result, merged) } } else { result = append(result, merged) } } } case parser.JoinLeft: for _, left := range leftRows { matched := false for _, right := range rightRows { merged := e.mergeRows(left, right, leftAlias, rightAlias) if tableRef.Join.Condition != nil { match, _ := e.evalExpr(tableRef.Join.Condition, merged) if toBool(match) { result = append(result, merged) matched = true } } } if !matched { // Add left row with nulls for right result = append(result, left) } } case parser.JoinCross: for _, left := range leftRows { for _, right := range rightRows { result = append(result, e.mergeRows(left, right, leftAlias, rightAlias)) } } } // Recursively process any additional joins if rightTableRef.Join != nil { return e.executeJoins(*rightTableRef, result) } return result, nil } // executeJoin executes a JOIN operation. func (e *Executor) executeJoin(tableRef parser.TableRef, leftRows []storage.Row) ([]storage.Row, error) { join := tableRef.Join if join == nil || join.Table == nil { return leftRows, nil } rightTable := join.Table.Name rightRows, err := e.table.Select(rightTable, nil) if err != nil { return nil, err } var result []storage.Row switch join.Type { case parser.JoinInner: for _, left := range leftRows { for _, right := range rightRows { merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias) if join.Condition != nil { match, _ := e.evalExpr(join.Condition, merged) if toBool(match) { result = append(result, merged) } } else { result = append(result, merged) } } } case parser.JoinLeft: for _, left := range leftRows { matched := false for _, right := range rightRows { merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias) if join.Condition != nil { match, _ := e.evalExpr(join.Condition, merged) if toBool(match) { result = append(result, merged) matched = true } } } if !matched { // Add left row with nulls for right result = append(result, left) } } case parser.JoinCross: for _, left := range leftRows { for _, right := range rightRows { result = append(result, e.mergeRows(left, right, tableRef.Alias, join.Table.Alias)) } } } return result, nil } // mergeRows merges two rows with optional table aliases. func (e *Executor) mergeRows(left, right storage.Row, leftAlias, rightAlias string) storage.Row { result := make(storage.Row) for k, v := range left { // Copy the key as-is (it might already be qualified) result[k] = v // Only add qualified name if the key is NOT already qualified and we have an alias if leftAlias != "" && !strings.Contains(k, ".") { result[leftAlias+"."+k] = v } } for k, v := range right { // For unqualified names, only add if they don't already exist // This prevents right table columns from overwriting left table columns if !strings.Contains(k, ".") { if _, exists := result[k]; !exists { result[k] = v } // Add qualified name for right table if rightAlias != "" { result[rightAlias+"."+k] = v } } else { // Already qualified, just copy it result[k] = v } } return result } // addTableAlias adds table-qualified names to a row. func (e *Executor) addTableAlias(row storage.Row, alias string) storage.Row { result := make(storage.Row) for k, v := range row { result[k] = v // Don't add alias to already-qualified names if !strings.Contains(k, ".") { result[alias+"."+k] = v } } return result } // executeInsert executes an INSERT statement. func (e *Executor) executeInsert(stmt *parser.InsertStmt) (*Result, error) { tableName := stmt.Table.Name schema, err := e.schema.GetSchema(tableName) if err != nil { return nil, err } count := 0 for _, values := range stmt.Values { row := make(storage.Row) if len(stmt.Columns) > 0 { // Named columns for i, col := range stmt.Columns { if i < len(values) { val, err := e.evalExpr(values[i], nil) if err != nil { return nil, err } row[col] = val } } } else { // All columns in order for i, col := range schema.Columns { if i < len(values) { val, err := e.evalExpr(values[i], nil) if err != nil { return nil, err } row[col.Name] = val } } } err := e.table.Insert(tableName, row) if err != nil { // Handle conflict based on OnConflict action if strings.Contains(err.Error(), "duplicate") { switch stmt.OnConflict { case parser.ConflictIgnore: // Silently ignore the duplicate continue case parser.ConflictReplace: // Delete existing row and insert new one pkValue := row[schema.PrimaryKey] if pkValue != nil { e.table.Delete(tableName, func(r storage.Row) bool { return fmt.Sprintf("%v", r[schema.PrimaryKey]) == fmt.Sprintf("%v", pkValue) }) // Try insert again if err := e.table.Insert(tableName, row); err != nil { return nil, err } } case parser.ConflictAbort, parser.ConflictFail: return nil, err case parser.ConflictRollback: // In a real implementation, this would rollback the transaction return nil, err default: return nil, err } } else { return nil, err } } count++ } result := NewResult("INSERT") result.SetRowCount(count) return result, nil } // executeUpdate executes an UPDATE statement. func (e *Executor) executeUpdate(stmt *parser.UpdateStmt) (*Result, error) { tableName := stmt.Table.Name // Build filter var filter func(storage.Row) bool if stmt.Where != nil { filter = func(row storage.Row) bool { val, err := e.evalExpr(stmt.Where, row) if err != nil { return false } return toBool(val) } } // Use UpdateFunc to evaluate expressions per-row (supports self-referencing like balance = balance + 100) updateFn := func(row storage.Row) (storage.Row, error) { updates := make(storage.Row) for _, assign := range stmt.Set { val, err := e.evalExpr(assign.Value, row) if err != nil { return nil, err } updates[assign.Column] = val } return updates, nil } count, err := e.table.UpdateFunc(tableName, updateFn, filter) if err != nil { return nil, err } result := NewResult("UPDATE") result.SetRowCount(count) return result, nil } // executeDelete executes a DELETE statement. func (e *Executor) executeDelete(stmt *parser.DeleteStmt) (*Result, error) { tableName := stmt.Table.Name // Build filter var filter func(storage.Row) bool if stmt.Where != nil { filter = func(row storage.Row) bool { val, err := e.evalExpr(stmt.Where, row) if err != nil { return false } return toBool(val) } } count, err := e.table.Delete(tableName, filter) if err != nil { return nil, err } result := NewResult("DELETE") result.SetRowCount(count) return result, nil } // executeCreateTable executes a CREATE TABLE statement. func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, error) { // Check if exists if e.schema.TableExists(stmt.Table.Name) { if stmt.IfNotExists { result := NewResult("CREATE TABLE") return result, nil } return nil, fmt.Errorf("table already exists: %s", stmt.Table.Name) } // Build schema schema := &storage.Schema{ Name: stmt.Table.Name, } for _, colDef := range stmt.Columns { col := storage.Column{ Name: colDef.Name, Type: colDef.Type.Name, Nullable: true, } for _, constraint := range colDef.Constraints { switch constraint.Type { case parser.ConstraintPrimaryKey: col.PrimaryKey = true col.Nullable = false schema.PrimaryKey = col.Name case parser.ConstraintNotNull: col.Nullable = false case parser.ConstraintDefault: if constraint.Default != nil { val, _ := e.evalExpr(constraint.Default, nil) col.Default = val } case parser.ConstraintAutoIncrement: schema.AutoIncrement = true } } schema.Columns = append(schema.Columns, col) } // Handle table-level constraints for _, constraint := range stmt.Constraints { if constraint.Type == parser.ConstraintPrimaryKey && len(constraint.Columns) > 0 { schema.PrimaryKey = constraint.Columns[0] for i := range schema.Columns { if strings.EqualFold(schema.Columns[i].Name, schema.PrimaryKey) { schema.Columns[i].PrimaryKey = true schema.Columns[i].Nullable = false } } } } if err := e.schema.CreateTable(schema); err != nil { return nil, err } // Update analyzer catalog e.catalog.CreateTable(schema.ToAnalyzerTableInfo()) result := NewResult("CREATE TABLE") return result, nil } // executeDropTable executes a DROP TABLE statement. func (e *Executor) executeDropTable(stmt *parser.DropTableStmt) (*Result, error) { for _, tableRef := range stmt.Tables { if !e.schema.TableExists(tableRef.Name) { if stmt.IfExists { continue } return nil, fmt.Errorf("table not found: %s", tableRef.Name) } // First, drop all indexes associated with this table indexes, _ := e.schema.ListTableIndexes(tableRef.Name) for _, idx := range indexes { // Clear index entries columns := make([]string, len(idx.Columns)) for i, col := range idx.Columns { columns[i] = col.Name } e.table.ClearIndex(idx.Name, tableRef.Name, columns) // Drop the index schema e.schema.DropIndex(idx.Name) } // Then, truncate all data rows e.table.Truncate(tableRef.Name) // Finally, drop the table schema if err := e.schema.DropTable(tableRef.Name); err != nil { return nil, err } // Update analyzer catalog e.catalog.DropTable(tableRef.Name) } result := NewResult("DROP TABLE") return result, nil } // executeCreateIndex creates a new index. func (e *Executor) executeCreateIndex(stmt *parser.CreateIndexStmt) (*Result, error) { // Check if index already exists if e.schema.IndexExists(stmt.Name) { if stmt.IfNotExists { result := NewResult("CREATE INDEX") return result, nil } return nil, fmt.Errorf("index already exists: %s", stmt.Name) } // Verify table exists if !e.schema.TableExists(stmt.Table) { return nil, fmt.Errorf("table not found: %s", stmt.Table) } // Verify columns exist schema, err := e.schema.GetSchema(stmt.Table) if err != nil { return nil, err } for _, col := range stmt.Columns { if _, found := schema.GetColumn(col.Name); !found { return nil, fmt.Errorf("column not found: %s", col.Name) } } // Create storage index index := &storage.Index{ Name: stmt.Name, Table: stmt.Table, Unique: stmt.Unique, } for _, col := range stmt.Columns { index.Columns = append(index.Columns, storage.IndexColumn{ Name: col.Name, Desc: col.Desc, }) } if err := e.schema.CreateIndex(index); err != nil { return nil, err } // Build index entries for existing rows columns := make([]string, len(stmt.Columns)) for i, col := range stmt.Columns { columns[i] = col.Name } if err := e.table.BuildIndex(stmt.Name, stmt.Table, columns); err != nil { // Rollback index creation on failure e.schema.DropIndex(stmt.Name) return nil, fmt.Errorf("failed to build index: %w", err) } result := NewResult("CREATE INDEX") return result, nil } // executeDropIndex drops an index. func (e *Executor) executeDropIndex(stmt *parser.DropIndexStmt) (*Result, error) { if !e.schema.IndexExists(stmt.Name) { if stmt.IfExists { result := NewResult("DROP INDEX") return result, nil } return nil, fmt.Errorf("index not found: %s", stmt.Name) } // Get index info to clear entries index, err := e.schema.GetIndex(stmt.Name) if err == nil && index != nil { columns := make([]string, len(index.Columns)) for i, col := range index.Columns { columns[i] = col.Name } e.table.ClearIndex(stmt.Name, index.Table, columns) } if err := e.schema.DropIndex(stmt.Name); err != nil { return nil, err } result := NewResult("DROP INDEX") return result, nil } // executeAlterTable executes an ALTER TABLE statement. func (e *Executor) executeAlterTable(stmt *parser.AlterTableStmt) (*Result, error) { switch action := stmt.Action.(type) { case *parser.AddColumnAction: return e.executeAlterTableAddColumn(stmt.Table, action) case *parser.DropColumnAction: return e.executeAlterTableDropColumn(stmt.Table, action) case *parser.RenameTableAction: return e.executeAlterTableRename(stmt.Table, action) case *parser.RenameColumnAction: return e.executeAlterTableRenameColumn(stmt.Table, action) default: return nil, fmt.Errorf("unsupported ALTER TABLE action: %T", action) } } // executeAlterTableAddColumn adds a column to a table. func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddColumnAction) (*Result, error) { col := storage.Column{ Name: action.Column.Name, Type: action.Column.Type.Name, Nullable: true, } // Process column constraints for _, constraint := range action.Column.Constraints { switch constraint.Type { case parser.ConstraintPrimaryKey: col.PrimaryKey = true col.Nullable = false case parser.ConstraintNotNull: col.Nullable = false case parser.ConstraintDefault: if constraint.Default != nil { val, _ := e.evalExpr(constraint.Default, nil) col.Default = val } } } if err := e.schema.AddColumn(table, col); err != nil { return nil, err } // Update catalog e.SyncCatalog() result := NewResult("ALTER TABLE") return result, nil } // executeAlterTableDropColumn drops a column from a table. func (e *Executor) executeAlterTableDropColumn(table string, action *parser.DropColumnAction) (*Result, error) { if err := e.schema.DropColumn(table, action.Column); err != nil { return nil, err } // Update catalog e.SyncCatalog() result := NewResult("ALTER TABLE") return result, nil } // executeAlterTableRename renames a table. func (e *Executor) executeAlterTableRename(table string, action *parser.RenameTableAction) (*Result, error) { if err := e.schema.RenameTable(table, action.NewName); err != nil { return nil, err } // Update catalog e.SyncCatalog() result := NewResult("ALTER TABLE") return result, nil } // executeAlterTableRenameColumn renames a column. func (e *Executor) executeAlterTableRenameColumn(table string, action *parser.RenameColumnAction) (*Result, error) { if err := e.schema.RenameColumn(table, action.OldName, action.NewName); err != nil { return nil, err } // Update catalog e.SyncCatalog() result := NewResult("ALTER TABLE") return result, nil } // Transaction execution methods // executeBegin starts a new transaction. func (e *Executor) executeBegin(stmt *parser.BeginStmt) (*Result, error) { if e.inTransaction { return nil, fmt.Errorf("cannot start a transaction within a transaction") } e.inTransaction = true e.savepoints = nil e.txLog = nil result := NewResult("BEGIN") return result, nil } // executeCommit commits the current transaction. func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) { if !e.inTransaction { return nil, fmt.Errorf("cannot commit: no transaction in progress") } // Clear transaction state e.inTransaction = false e.savepoints = nil e.txLog = nil result := NewResult("COMMIT") return result, nil } // executeRollback rolls back the current transaction or to a savepoint. func (e *Executor) executeRollback(stmt *parser.RollbackStmt) (*Result, error) { if !e.inTransaction { return nil, fmt.Errorf("cannot rollback: no transaction in progress") } if stmt.Savepoint != "" { // Rollback to savepoint return e.rollbackToSavepoint(stmt.Savepoint) } // Full rollback - undo all operations in reverse order for i := len(e.txLog) - 1; i >= 0; i-- { entry := e.txLog[i] if err := e.undoOperation(entry); err != nil { // Log error but continue with rollback continue } } // Clear transaction state e.inTransaction = false e.savepoints = nil e.txLog = nil result := NewResult("ROLLBACK") return result, nil } // executeSavepoint creates a savepoint. func (e *Executor) executeSavepoint(stmt *parser.SavepointStmt) (*Result, error) { if !e.inTransaction { // SQLite allows SAVEPOINT outside transaction (starts implicit transaction) e.inTransaction = true e.txLog = nil } // Add savepoint marker e.savepoints = append(e.savepoints, stmt.Name) result := NewResult("SAVEPOINT") return result, nil } // executeRelease releases a savepoint. func (e *Executor) executeRelease(stmt *parser.ReleaseStmt) (*Result, error) { if !e.inTransaction { return nil, fmt.Errorf("cannot release savepoint: no transaction in progress") } // Find and remove the savepoint found := false for i := len(e.savepoints) - 1; i >= 0; i-- { if e.savepoints[i] == stmt.Name { e.savepoints = e.savepoints[:i] found = true break } } if !found { return nil, fmt.Errorf("no such savepoint: %s", stmt.Name) } result := NewResult("RELEASE") return result, nil } // executeAttach attaches a database. func (e *Executor) executeAttach(stmt *parser.AttachStmt) (*Result, error) { // Check if alias already exists if _, exists := e.attachedDatabases[stmt.Alias]; exists { return nil, fmt.Errorf("database alias already exists: %s", stmt.Alias) } // Reserved alias check if strings.EqualFold(stmt.Alias, "temp") || strings.EqualFold(stmt.Alias, "temporary") { return nil, fmt.Errorf("reserved database alias: %s", stmt.Alias) } // Get the pool from the main schema manager pool := e.schema.GetPool() // Create new schema and table managers for the attached database // In PizzaKV, each database is just a different namespace/prefix schema := storage.NewSchemaManager(pool, stmt.FilePath) table := storage.NewTableManager(pool, schema, stmt.FilePath) // Register the database connection e.attachedDatabases[stmt.Alias] = &DatabaseConnection{ Alias: stmt.Alias, Path: stmt.FilePath, Schema: schema, Table: table, } // Sync the catalog with the attached database's tables tables, _ := schema.ListTables() for _, tableName := range tables { tSchema, err := schema.GetSchema(tableName) if err != nil { continue } // Add with database prefix tableInfo := tSchema.ToAnalyzerTableInfo() tableInfo.Name = stmt.Alias + "." + tableInfo.Name e.catalog.CreateTable(tableInfo) } result := NewResult("ATTACH") return result, nil } // executeDetach detaches a database. func (e *Executor) executeDetach(stmt *parser.DetachStmt) (*Result, error) { // Cannot detach main database if strings.EqualFold(stmt.Alias, "main") { return nil, fmt.Errorf("cannot detach main database") } // Check if database exists if _, exists := e.attachedDatabases[stmt.Alias]; !exists { return nil, fmt.Errorf("no such database: %s", stmt.Alias) } // Remove from attached databases delete(e.attachedDatabases, stmt.Alias) // Note: We don't remove from catalog as that would be more complex // In a production system, we'd need to track which tables belong to which database result := NewResult("DETACH") return result, nil } // rollbackToSavepoint rolls back to a specific savepoint. func (e *Executor) rollbackToSavepoint(name string) (*Result, error) { // Find savepoint index savepointIdx := -1 for i := len(e.savepoints) - 1; i >= 0; i-- { if e.savepoints[i] == name { savepointIdx = i break } } if savepointIdx == -1 { return nil, fmt.Errorf("no such savepoint: %s", name) } // Count operations to undo (operations after the savepoint) // For simplicity, we track savepoint positions by counting log entries // In a real implementation, we'd track log positions per savepoint // Undo operations in reverse order for i := len(e.txLog) - 1; i >= 0; i-- { entry := e.txLog[i] if err := e.undoOperation(entry); err != nil { continue } } // Remove savepoints after the target e.savepoints = e.savepoints[:savepointIdx+1] result := NewResult("ROLLBACK") return result, nil } // undoOperation reverses a single operation. func (e *Executor) undoOperation(entry txLogEntry) error { switch entry.operation { case "INSERT": // Delete the inserted row _, err := e.table.Delete(entry.table, func(r storage.Row) bool { // Match by primary key stored in entry.key pk := e.getPrimaryKey(entry.table) if pk == "" { return false } return fmt.Sprintf("%v", r[pk]) == entry.key }) return err case "DELETE": // Re-insert the deleted row if entry.oldData != nil { return e.table.Insert(entry.table, entry.oldData) } case "UPDATE": // Restore the old data if entry.oldData != nil { pk := e.getPrimaryKey(entry.table) if pk != "" { // Delete current row and insert old data e.table.Delete(entry.table, func(r storage.Row) bool { return fmt.Sprintf("%v", r[pk]) == entry.key }) return e.table.Insert(entry.table, entry.oldData) } } } return nil } // getPrimaryKey returns the primary key column name for a table. func (e *Executor) getPrimaryKey(tableName string) string { schema, err := e.schema.GetSchema(tableName) if err != nil { return "" } return schema.PrimaryKey } // extractIndexableCondition extracts column name and value from a simple equality condition. // Returns (column, value, true) if the expression is column = literal. func (e *Executor) extractIndexableCondition(expr parser.Expr) (string, interface{}, bool) { binExpr, ok := expr.(*parser.BinaryExpr) if !ok { return "", nil, false } // Only handle equality for now if binExpr.Op != lexer.TokenEq { return "", nil, false } // Check for column = literal pattern colRef, leftIsCol := binExpr.Left.(*parser.ColumnRef) litExpr, rightIsLit := binExpr.Right.(*parser.LiteralExpr) if leftIsCol && rightIsLit { val, _ := e.evalLiteral(litExpr) return colRef.Column, val, true } // Check for literal = column pattern litExpr, leftIsLit := binExpr.Left.(*parser.LiteralExpr) colRef, rightIsCol := binExpr.Right.(*parser.ColumnRef) if leftIsLit && rightIsCol { val, _ := e.evalLiteral(litExpr) return colRef.Column, val, true } return "", nil, false } // executePragma executes a PRAGMA statement. func (e *Executor) executePragma(stmt *parser.PragmaStmt) (*Result, error) { switch stmt.Name { case "table_info": return e.pragmaTableInfo(stmt.Arg) case "table_list": return e.pragmaTableList() case "database_list": return e.pragmaDatabaseList() case "version": return e.pragmaVersion() default: return nil, fmt.Errorf("unknown pragma: %s", stmt.Name) } } // pragmaTableInfo returns column information for a table. func (e *Executor) pragmaTableInfo(tableName string) (*Result, error) { if tableName == "" { return nil, fmt.Errorf("table_info requires a table name") } schema, err := e.schema.GetSchema(tableName) if err != nil { return nil, err } result := NewResult("PRAGMA") result.AddColumn("cid") result.AddColumn("name") result.AddColumn("type") result.AddColumn("notnull") result.AddColumn("dflt_value") result.AddColumn("pk") for i, col := range schema.Columns { notnull := 0 if !col.Nullable { notnull = 1 } pk := 0 if col.PrimaryKey { pk = 1 } result.AddRow(int64(i), col.Name, col.Type, int64(notnull), col.Default, int64(pk)) } return result, nil } // pragmaTableList returns a list of all tables. func (e *Executor) pragmaTableList() (*Result, error) { tables, err := e.schema.ListTables() if err != nil { return nil, err } result := NewResult("PRAGMA") result.AddColumn("schema") result.AddColumn("name") result.AddColumn("type") for _, t := range tables { result.AddRow("main", t, "table") } return result, nil } // pragmaDatabaseList returns a list of databases. func (e *Executor) pragmaDatabaseList() (*Result, error) { result := NewResult("PRAGMA") result.AddColumn("seq") result.AddColumn("name") result.AddColumn("file") // We only have one database result.AddRow(int64(0), "main", "") return result, nil } // pragmaVersion returns the PizzaSQL version. func (e *Executor) pragmaVersion() (*Result, error) { result := NewResult("PRAGMA") result.AddColumn("version") result.AddRow("PizzaSQL 1.0.0") return result, nil } // executeExplain executes an EXPLAIN statement. func (e *Executor) executeExplain(stmt *parser.ExplainStmt) (*Result, error) { result := NewResult("EXPLAIN") if stmt.QueryPlan { // EXPLAIN QUERY PLAN format result.AddColumn("id") result.AddColumn("parent") result.AddColumn("notused") result.AddColumn("detail") plan := e.generateQueryPlan(stmt.Statement) for i, step := range plan { result.AddRow(int64(i), int64(0), int64(0), step) } } else { // Simple EXPLAIN format result.AddColumn("addr") result.AddColumn("opcode") result.AddColumn("p1") result.AddColumn("p2") result.AddColumn("p3") result.AddColumn("p4") result.AddColumn("p5") result.AddColumn("comment") ops := e.generateOpcodes(stmt.Statement) for i, op := range ops { result.AddRow(int64(i), op, int64(0), int64(0), int64(0), "", int64(0), "") } } return result, nil } // generateQueryPlan generates a simple query plan description. func (e *Executor) generateQueryPlan(stmt parser.Statement) []string { var plan []string switch s := stmt.(type) { case *parser.SelectStmt: if len(s.From) > 0 { plan = append(plan, fmt.Sprintf("SCAN TABLE %s", s.From[0].Name)) if s.Where != nil { plan = append(plan, "FILTER") } if len(s.OrderBy) > 0 { plan = append(plan, "SORT") } if s.Limit != nil { plan = append(plan, "LIMIT") } } else { plan = append(plan, "SCALAR EXPRESSION") } case *parser.InsertStmt: plan = append(plan, fmt.Sprintf("INSERT INTO %s", s.Table.Name)) case *parser.UpdateStmt: plan = append(plan, fmt.Sprintf("SCAN TABLE %s", s.Table.Name)) plan = append(plan, "UPDATE") case *parser.DeleteStmt: plan = append(plan, fmt.Sprintf("SCAN TABLE %s", s.Table.Name)) plan = append(plan, "DELETE") default: plan = append(plan, "EXECUTE") } return plan } // generateOpcodes generates simplified opcodes for EXPLAIN. func (e *Executor) generateOpcodes(stmt parser.Statement) []string { var ops []string switch s := stmt.(type) { case *parser.SelectStmt: ops = append(ops, "Init") if len(s.From) > 0 { ops = append(ops, "OpenRead") ops = append(ops, "Rewind") ops = append(ops, "Column") ops = append(ops, "ResultRow") ops = append(ops, "Next") ops = append(ops, "Close") } else { ops = append(ops, "Integer") ops = append(ops, "ResultRow") } ops = append(ops, "Halt") case *parser.InsertStmt: ops = append(ops, "Init") ops = append(ops, "OpenWrite") ops = append(ops, "NewRowid") ops = append(ops, "Insert") ops = append(ops, "Close") ops = append(ops, "Halt") case *parser.UpdateStmt: ops = append(ops, "Init") ops = append(ops, "OpenWrite") ops = append(ops, "Rewind") ops = append(ops, "Column") ops = append(ops, "Update") ops = append(ops, "Next") ops = append(ops, "Close") ops = append(ops, "Halt") case *parser.DeleteStmt: ops = append(ops, "Init") ops = append(ops, "OpenWrite") ops = append(ops, "Rewind") ops = append(ops, "Delete") ops = append(ops, "Next") ops = append(ops, "Close") ops = append(ops, "Halt") default: ops = append(ops, "Init") ops = append(ops, "Halt") } return ops } // evalExpr evaluates an expression. func (e *Executor) evalExpr(expr parser.Expr, row storage.Row) (interface{}, error) { switch ex := expr.(type) { case *parser.LiteralExpr: return e.evalLiteral(ex) case *parser.ColumnRef: return e.evalColumnRef(ex, row) case *parser.BinaryExpr: return e.evalBinaryExpr(ex, row) case *parser.UnaryExpr: return e.evalUnaryExpr(ex, row) case *parser.FunctionCall: return e.evalFunctionCall(ex, row) case *parser.ParenExpr: return e.evalExpr(ex.Expr, row) case *parser.CaseExpr: return e.evalCaseExpr(ex, row) case *parser.InExpr: return e.evalInExpr(ex, row) case *parser.BetweenExpr: return e.evalBetweenExpr(ex, row) case *parser.LikeExpr: return e.evalLikeExpr(ex, row) case *parser.IsNullExpr: return e.evalIsNullExpr(ex, row) case *parser.CastExpr: return e.evalCastExpr(ex, row) case *parser.SubqueryExpr: return e.evalSubqueryExpr(ex, row) case *parser.ExistsExpr: return e.evalExistsExpr(ex, row) default: return nil, fmt.Errorf("unsupported expression type: %T", expr) } } func (e *Executor) evalLiteral(lit *parser.LiteralExpr) (interface{}, error) { switch lit.Type { case lexer.TokenNumber: // Check for scientific notation (e.g., 1e+06) or decimal point if strings.Contains(lit.Value, ".") || strings.ContainsAny(lit.Value, "eE") { f, err := strconv.ParseFloat(lit.Value, 64) if err != nil { return nil, err } // If it's a whole number (no fractional part), return as int64 if f == float64(int64(f)) { return int64(f), nil } return f, nil } return strconv.ParseInt(lit.Value, 10, 64) case lexer.TokenString: return lit.Value, nil case lexer.TokenNULL: return nil, nil case lexer.TokenTRUE: return true, nil case lexer.TokenFALSE: return false, nil default: return lit.Value, nil } } func (e *Executor) evalColumnRef(ref *parser.ColumnRef, row storage.Row) (interface{}, error) { if row == nil { return nil, fmt.Errorf("no row context for column: %s", ref.Column) } // Check for ROWID aliases (rowid, oid, _rowid_) if storage.IsRowIDColumn(ref.Column) { if val, ok := row["_rowid_"]; ok { return val, nil } return nil, nil } // For qualified column references (table.column), check outer row first // This handles correlated subqueries where the qualifier refers to an outer table if ref.Table != "" && e.outerRow != nil { // Try qualified name in outer row first if val, ok := e.outerRow[ref.Table+"."+ref.Column]; ok { return val, nil } // Try case-insensitive in outer row for k, v := range e.outerRow { if strings.EqualFold(k, ref.Table+"."+ref.Column) { return v, nil } } } // Try qualified name in current row if ref.Table != "" { if val, ok := row[ref.Table+"."+ref.Column]; ok { return val, nil } } // Try direct column name if val, ok := row[ref.Column]; ok { return val, nil } // Case-insensitive search in current row for k, v := range row { if strings.EqualFold(k, ref.Column) { return v, nil } if ref.Table != "" && strings.EqualFold(k, ref.Table+"."+ref.Column) { return v, nil } } // For unqualified references, also check the outer row context if e.outerRow != nil { // Try direct column name in outer row if val, ok := e.outerRow[ref.Column]; ok { return val, nil } // Case-insensitive search in outer row for k, v := range e.outerRow { if strings.EqualFold(k, ref.Column) { return v, nil } } } return nil, nil // Column not found, return NULL } func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (interface{}, error) { left, err := e.evalExpr(expr.Left, row) if err != nil { return nil, err } right, err := e.evalExpr(expr.Right, row) if err != nil { return nil, err } switch expr.Op { case lexer.TokenPlus: return toFloat(left) + toFloat(right), nil case lexer.TokenMinus: return toFloat(left) - toFloat(right), nil case lexer.TokenStar: return toFloat(left) * toFloat(right), nil case lexer.TokenSlash: r := toFloat(right) if r == 0 { return nil, nil // Division by zero returns NULL } return toFloat(left) / r, nil case lexer.TokenPercent: return int64(toFloat(left)) % int64(toFloat(right)), nil case lexer.TokenEq: return compare(left, right) == 0, nil case lexer.TokenNeq: return compare(left, right) != 0, nil case lexer.TokenLt: return compare(left, right) < 0, nil case lexer.TokenLte: return compare(left, right) <= 0, nil case lexer.TokenGt: return compare(left, right) > 0, nil case lexer.TokenGte: return compare(left, right) >= 0, nil case lexer.TokenAND: return toBool(left) && toBool(right), nil case lexer.TokenOR: return toBool(left) || toBool(right), nil case lexer.TokenConcat: return toString(left) + toString(right), nil default: return nil, fmt.Errorf("unsupported operator: %v", expr.Op) } } func (e *Executor) evalUnaryExpr(expr *parser.UnaryExpr, row storage.Row) (interface{}, error) { val, err := e.evalExpr(expr.Operand, row) if err != nil { return nil, err } switch expr.Op { case lexer.TokenMinus: return -toFloat(val), nil case lexer.TokenPlus: return toFloat(val), nil case lexer.TokenNOT: return !toBool(val), nil default: return val, nil } } func (e *Executor) evalFunctionCall(fn *parser.FunctionCall, row storage.Row) (interface{}, error) { name := strings.ToUpper(fn.Name) // Evaluate arguments args := make([]interface{}, len(fn.Args)) for i, arg := range fn.Args { val, err := e.evalExpr(arg, row) if err != nil { return nil, err } args[i] = val } switch name { case "UPPER": if len(args) > 0 { if args[0] == nil { return nil, nil // NULL propagation } return strings.ToUpper(toString(args[0])), nil } case "LOWER": if len(args) > 0 { if args[0] == nil { return nil, nil // NULL propagation } return strings.ToLower(toString(args[0])), nil } case "LENGTH": if len(args) > 0 { if args[0] == nil { return nil, nil // NULL propagation } return int64(len(toString(args[0]))), nil } case "ABS": if len(args) > 0 { v := toFloat(args[0]) if v < 0 { return -v, nil } return v, nil } case "COALESCE": for _, arg := range args { if arg != nil { return arg, nil } } return nil, nil case "NULLIF": if len(args) >= 2 && compare(args[0], args[1]) == 0 { return nil, nil } if len(args) > 0 { return args[0], nil } case "IFNULL": if len(args) >= 2 { if args[0] == nil { return args[1], nil } return args[0], nil } case "TYPEOF": if len(args) > 0 { switch args[0].(type) { case nil: return "null", nil case int64, int: return "integer", nil case float64: return "real", nil case string: return "text", nil case []byte: return "blob", nil default: return "text", nil } } case "SUBSTR", "SUBSTRING": if len(args) >= 2 { s := toString(args[0]) start := int(toFloat(args[1])) - 1 // SQL is 1-indexed if start < 0 { start = 0 } if start >= len(s) { return "", nil } if len(args) >= 3 { length := int(toFloat(args[2])) if start+length > len(s) { length = len(s) - start } return s[start : start+length], nil } return s[start:], nil } case "TRIM": if len(args) > 0 { return strings.TrimSpace(toString(args[0])), nil } case "REPLACE": if len(args) >= 3 { return strings.ReplaceAll(toString(args[0]), toString(args[1]), toString(args[2])), nil } // Additional SQLite functions case "PRINTF": if len(args) > 0 { format := toString(args[0]) fmtArgs := make([]interface{}, len(args)-1) for i := 1; i < len(args); i++ { fmtArgs[i-1] = args[i] } return fmt.Sprintf(format, fmtArgs...), nil } case "HEX": if len(args) > 0 { s := toString(args[0]) return strings.ToUpper(fmt.Sprintf("%x", []byte(s))), nil } case "UNHEX": if len(args) > 0 { s := toString(args[0]) var result []byte for i := 0; i < len(s)-1; i += 2 { var b byte fmt.Sscanf(s[i:i+2], "%x", &b) result = append(result, b) } return string(result), nil } case "RANDOM": return rand.Int63(), nil case "RANDOMBLOB": if len(args) > 0 { n := int(toFloat(args[0])) if n <= 0 { n = 1 } if n > 1000000 { n = 1000000 } blob := make([]byte, n) rand.Read(blob) return string(blob), nil } case "ZEROBLOB": if len(args) > 0 { n := int(toFloat(args[0])) if n <= 0 { n = 1 } if n > 1000000 { n = 1000000 } return string(make([]byte, n)), nil } case "INSTR": if len(args) >= 2 { s := toString(args[0]) substr := toString(args[1]) idx := strings.Index(s, substr) if idx < 0 { return int64(0), nil } return int64(idx + 1), nil // SQL is 1-indexed } case "GLOB": if len(args) >= 2 { pattern := toString(args[0]) s := toString(args[1]) return matchGlob(pattern, s), nil } case "ROUND": if len(args) > 0 { v := toFloat(args[0]) decimals := 0 if len(args) >= 2 { decimals = int(toFloat(args[1])) } mult := 1.0 for i := 0; i < decimals; i++ { mult *= 10 } return float64(int64(v*mult+0.5)) / mult, nil } case "MAX": if len(args) > 0 { max := args[0] for _, arg := range args[1:] { if compare(arg, max) > 0 { max = arg } } return max, nil } case "MIN": if len(args) > 0 { min := args[0] for _, arg := range args[1:] { if compare(arg, min) < 0 { min = arg } } return min, nil } case "CONCAT": var result strings.Builder for _, arg := range args { result.WriteString(toString(arg)) } return result.String(), nil } return nil, nil } func (e *Executor) evalCaseExpr(expr *parser.CaseExpr, row storage.Row) (interface{}, error) { var operand interface{} if expr.Operand != nil { var err error operand, err = e.evalExpr(expr.Operand, row) if err != nil { return nil, err } } for _, when := range expr.Whens { cond, err := e.evalExpr(when.Condition, row) if err != nil { return nil, err } var match bool if operand != nil { match = compare(operand, cond) == 0 } else { match = toBool(cond) } if match { return e.evalExpr(when.Result, row) } } if expr.Else != nil { return e.evalExpr(expr.Else, row) } return nil, nil } func (e *Executor) evalInExpr(expr *parser.InExpr, row storage.Row) (interface{}, error) { left, err := e.evalExpr(expr.Left, row) if err != nil { return nil, err } // Handle subquery: IN (SELECT ...) if expr.Subquery != nil { result, err := e.executeSelect(expr.Subquery) if err != nil { return nil, fmt.Errorf("IN subquery error: %w", err) } // Check each row's first column value for _, resultRow := range result.Rows { if len(resultRow) > 0 { if compare(left, resultRow[0]) == 0 { return !expr.Not, nil } } } return expr.Not, nil } // Handle value list: IN (1, 2, 3) for _, val := range expr.Values { v, err := e.evalExpr(val, row) if err != nil { return nil, err } if compare(left, v) == 0 { return !expr.Not, nil } } return expr.Not, nil } func (e *Executor) evalBetweenExpr(expr *parser.BetweenExpr, row storage.Row) (interface{}, error) { val, err := e.evalExpr(expr.Left, row) if err != nil { return nil, err } low, err := e.evalExpr(expr.Low, row) if err != nil { return nil, err } high, err := e.evalExpr(expr.High, row) if err != nil { return nil, err } inRange := compare(val, low) >= 0 && compare(val, high) <= 0 if expr.Not { return !inRange, nil } return inRange, nil } func (e *Executor) evalLikeExpr(expr *parser.LikeExpr, row storage.Row) (interface{}, error) { val, err := e.evalExpr(expr.Left, row) if err != nil { return nil, err } pattern, err := e.evalExpr(expr.Pattern, row) if err != nil { return nil, err } s := toString(val) p := toString(pattern) // Convert SQL LIKE pattern to simple matching // % matches any sequence, _ matches single character matched := matchLike(s, p) if expr.Not { return !matched, nil } return matched, nil } func (e *Executor) evalIsNullExpr(expr *parser.IsNullExpr, row storage.Row) (interface{}, error) { val, err := e.evalExpr(expr.Left, row) if err != nil { return nil, err } isNull := val == nil if expr.Not { return !isNull, nil } return isNull, nil } func (e *Executor) evalCastExpr(expr *parser.CastExpr, row storage.Row) (interface{}, error) { val, err := e.evalExpr(expr.Expr, row) if err != nil { return nil, err } typeName := strings.ToUpper(expr.Type.Name) switch { case strings.Contains(typeName, "INT"): return int64(toFloat(val)), nil case strings.Contains(typeName, "REAL"), strings.Contains(typeName, "FLOAT"), strings.Contains(typeName, "DOUBLE"): return toFloat(val), nil case strings.Contains(typeName, "TEXT"), strings.Contains(typeName, "CHAR"): return toString(val), nil default: return val, nil } } // evalSubqueryExpr executes a scalar subquery and returns its value. // A scalar subquery must return exactly one column. It returns: // - The single value if the subquery returns one row // - NULL if the subquery returns no rows // - Error if the subquery returns more than one row (for strict SQL compliance) func (e *Executor) evalSubqueryExpr(expr *parser.SubqueryExpr, row storage.Row) (interface{}, error) { // Save and set outer row context for correlated subqueries savedOuter := e.outerRow e.outerRow = row defer func() { e.outerRow = savedOuter }() // Execute the subquery result, err := e.executeSelect(expr.Query) if err != nil { return nil, fmt.Errorf("subquery error: %w", err) } // Check for empty result if result.RowCount == 0 { return nil, nil // Return NULL for empty subquery } // Check column count if len(result.Columns) == 0 { return nil, fmt.Errorf("subquery must return at least one column") } // For scalar subquery, return first column of first row // Note: Strict SQL would error if more than one row is returned // but we follow SQLite behavior which just returns the first value if len(result.Rows) > 0 && len(result.Rows[0]) > 0 { return result.Rows[0][0], nil } return nil, nil } // evalExistsExpr evaluates an EXISTS expression. // Returns true if the subquery returns at least one row, false otherwise. func (e *Executor) evalExistsExpr(expr *parser.ExistsExpr, row storage.Row) (interface{}, error) { // Save and set outer row context for correlated subqueries savedOuter := e.outerRow e.outerRow = row defer func() { e.outerRow = savedOuter }() // Execute the subquery result, err := e.executeSelect(expr.Subquery) if err != nil { return nil, fmt.Errorf("EXISTS subquery error: %w", err) } // EXISTS returns true if any rows are returned return len(result.Rows) > 0, nil } // evalAggregateExpr evaluates an aggregate expression over multiple rows. func (e *Executor) evalAggregateExpr(expr parser.Expr, rows []storage.Row) (interface{}, error) { fn, ok := expr.(*parser.FunctionCall) if !ok { // Not a function call - could be a binary expression with aggregates inside // Evaluate it with the aggregate evaluation context return e.evalExprWithAggregates(expr, rows) } name := strings.ToUpper(fn.Name) switch name { case "COUNT": if fn.Star { return int64(len(rows)), nil } if fn.Distinct { seen := make(map[interface{}]struct{}) for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil { seen[val] = struct{}{} } } } return int64(len(seen)), nil } count := int64(0) for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil { count++ } } } return count, nil case "SUM": var sum float64 for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil { sum += toFloat(val) } } } return sum, nil case "AVG": var sum float64 count := 0 for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil { sum += toFloat(val) count++ } } } if count == 0 { return nil, nil } return sum / float64(count), nil case "MIN": var min interface{} for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil && (min == nil || compare(val, min) < 0) { min = val } } } return min, nil case "MAX": var max interface{} for _, row := range rows { if len(fn.Args) > 0 { val, _ := e.evalExpr(fn.Args[0], row) if val != nil && (max == nil || compare(val, max) > 0) { max = val } } } return max, nil default: // Try scalar function if len(rows) > 0 { return e.evalFunctionCall(fn, rows[0]) } return nil, nil } } // evalExprWithAggregates evaluates an expression that may contain aggregate functions func (e *Executor) evalExprWithAggregates(expr parser.Expr, rows []storage.Row) (interface{}, error) { switch ex := expr.(type) { case *parser.BinaryExpr: left, err := e.evalExprWithAggregates(ex.Left, rows) if err != nil { return nil, err } right, err := e.evalExprWithAggregates(ex.Right, rows) if err != nil { return nil, err } // Apply the binary operator switch ex.Op { case lexer.TokenPlus: return toFloat(left) + toFloat(right), nil case lexer.TokenMinus: return toFloat(left) - toFloat(right), nil case lexer.TokenStar: return toFloat(left) * toFloat(right), nil case lexer.TokenSlash: r := toFloat(right) if r == 0 { return nil, nil } return toFloat(left) / r, nil case lexer.TokenPercent: return int64(toFloat(left)) % int64(toFloat(right)), nil case lexer.TokenEq: return compare(left, right) == 0, nil case lexer.TokenNeq: return compare(left, right) != 0, nil case lexer.TokenLt: return compare(left, right) < 0, nil case lexer.TokenLte: return compare(left, right) <= 0, nil case lexer.TokenGt: return compare(left, right) > 0, nil case lexer.TokenGte: return compare(left, right) >= 0, nil case lexer.TokenAND: return toBool(left) && toBool(right), nil case lexer.TokenOR: return toBool(left) || toBool(right), nil case lexer.TokenConcat: return toString(left) + toString(right), nil default: return nil, fmt.Errorf("unsupported operator: %v", ex.Op) } case *parser.FunctionCall: return e.evalAggregateExpr(expr, rows) default: // Non-aggregate expression, use first row if len(rows) > 0 { return e.evalExpr(expr, rows[0]) } return nil, nil } } // Helper functions func (e *Executor) getSelectColumns(stmt *parser.SelectStmt, schema *storage.Schema) []string { var columns []string for _, col := range stmt.Columns { if col.Star { for _, c := range schema.Columns { columns = append(columns, c.Name) } } else if col.Alias != "" { columns = append(columns, col.Alias) } else if ref, ok := col.Expr.(*parser.ColumnRef); ok { columns = append(columns, ref.Column) } else { columns = append(columns, fmt.Sprintf("column%d", len(columns)+1)) } } return columns } func (e *Executor) hasAggregates(columns []parser.SelectColumn) bool { for _, col := range columns { if e.isAggregate(col.Expr) { return true } } return false } func (e *Executor) isAggregate(expr parser.Expr) bool { if fn, ok := expr.(*parser.FunctionCall); ok { name := strings.ToUpper(fn.Name) switch name { case "COUNT", "SUM", "AVG", "MIN", "MAX", "TOTAL", "GROUP_CONCAT": return true } } return false } func (e *Executor) buildGroupKey(groupBy []parser.Expr, row storage.Row) string { var parts []string for _, expr := range groupBy { val, _ := e.evalExpr(expr, row) parts = append(parts, fmt.Sprintf("%v", val)) } return strings.Join(parts, "|") } func (e *Executor) sortRows(rows []storage.Row, orderBy []parser.OrderByItem) { sort.Slice(rows, func(i, j int) bool { for _, item := range orderBy { vi, _ := e.evalExpr(item.Expr, rows[i]) vj, _ := e.evalExpr(item.Expr, rows[j]) cmp := compare(vi, vj) if cmp != 0 { if item.Desc { return cmp > 0 } return cmp < 0 } } return false }) } // sortResultRows sorts Result.Rows based on ORDER BY clauses. // It handles column aliases by matching them against the select columns. func (e *Executor) sortResultRows(result *Result, orderBy []parser.OrderByItem, selectColumns []parser.SelectColumn, columnNames []string) { sort.Slice(result.Rows, func(i, j int) bool { for _, item := range orderBy { var vi, vj interface{} var rowI, rowJ storage.Row // Check if ORDER BY references a column alias if ref, ok := item.Expr.(*parser.ColumnRef); ok && ref.Table == "" { // Look for matching alias in select columns for idx, name := range columnNames { if strings.EqualFold(name, ref.Column) { if idx < len(result.Rows[i]) { vi = result.Rows[i][idx] vj = result.Rows[j][idx] goto compare } } } } // If not found as alias, try to evaluate the expression // Create temporary rows from result rows for evaluation rowI = e.resultRowToStorageRow(result, i) rowJ = e.resultRowToStorageRow(result, j) vi, _ = e.evalExpr(item.Expr, rowI) vj, _ = e.evalExpr(item.Expr, rowJ) compare: cmp := compare(vi, vj) if cmp != 0 { if item.Desc { return cmp > 0 } return cmp < 0 } } return false }) } // resultRowToStorageRow converts a Result row back to storage.Row for expression evaluation. func (e *Executor) resultRowToStorageRow(result *Result, rowIdx int) storage.Row { row := make(storage.Row) for colIdx, colName := range result.Columns { if colIdx < len(result.Rows[rowIdx]) { row[colName] = result.Rows[rowIdx][colIdx] } } return row } func (e *Executor) evalIntExpr(expr parser.Expr) int { val, _ := e.evalExpr(expr, nil) return int(toFloat(val)) } // Type conversion helpers func toFloat(v interface{}) float64 { switch val := v.(type) { case nil: return 0 case int64: return float64(val) case int: return float64(val) case float64: return val case bool: if val { return 1 } return 0 case string: f, _ := strconv.ParseFloat(val, 64) return f default: return 0 } } func toBool(v interface{}) bool { switch val := v.(type) { case nil: return false case bool: return val case int64: return val != 0 case int: return val != 0 case float64: return val != 0 case string: return val != "" && val != "0" && strings.ToLower(val) != "false" default: return false } } func toString(v interface{}) string { if v == nil { return "" } return fmt.Sprintf("%v", v) } func compare(a, b interface{}) int { if a == nil && b == nil { return 0 } if a == nil { return -1 } if b == nil { return 1 } // Try numeric comparison fa, oka := toNumeric(a) fb, okb := toNumeric(b) if oka && okb { if fa < fb { return -1 } if fa > fb { return 1 } return 0 } // String comparison sa := toString(a) sb := toString(b) return strings.Compare(sa, sb) } func toNumeric(v interface{}) (float64, bool) { switch val := v.(type) { case int64: return float64(val), true case int: return float64(val), true case float64: return val, true case string: f, err := strconv.ParseFloat(val, 64) return f, err == nil default: return 0, false } } // matchLike matches a string against a SQL LIKE pattern. func matchLike(s, pattern string) bool { // Simple implementation - convert to lowercase for case-insensitive matching s = strings.ToLower(s) pattern = strings.ToLower(pattern) return matchLikeHelper(s, pattern) } func matchLikeHelper(s, p string) bool { if p == "" { return s == "" } if p[0] == '%' { // % matches any sequence for i := 0; i <= len(s); i++ { if matchLikeHelper(s[i:], p[1:]) { return true } } return false } if s == "" { return false } if p[0] == '_' || p[0] == s[0] { return matchLikeHelper(s[1:], p[1:]) } return false } // matchGlob matches a string against a GLOB pattern. // GLOB uses * for any sequence and ? for single character (case-sensitive). func matchGlob(pattern, s string) bool { return matchGlobHelper(pattern, s) } func matchGlobHelper(p, s string) bool { if p == "" { return s == "" } if p[0] == '*' { // * matches any sequence for i := 0; i <= len(s); i++ { if matchGlobHelper(p[1:], s[i:]) { return true } } return false } if s == "" { return false } if p[0] == '?' || p[0] == s[0] { return matchGlobHelper(p[1:], s[1:]) } // Handle character classes [...] if p[0] == '[' { end := strings.Index(p, "]") if end > 0 { class := p[1:end] match := false negate := false if len(class) > 0 && class[0] == '^' { negate = true class = class[1:] } for _, c := range class { if byte(c) == s[0] { match = true break } } if negate { match = !match } if match { return matchGlobHelper(p[end+1:], s[1:]) } } } return false } // applyDistinct removes duplicate rows from the result func (e *Executor) applyDistinct(rows [][]interface{}) [][]interface{} { if len(rows) == 0 { return rows } seen := make(map[string]bool) uniqueRows := make([][]interface{}, 0) for _, row := range rows { // Create a key from all column values key := "" for i, val := range row { if i > 0 { key += "\x00" // Use null byte as separator } key += fmt.Sprintf("%v", val) } if !seen[key] { seen[key] = true uniqueRows = append(uniqueRows, row) } } return uniqueRows }