99% sqllogictest

.gitignore

@@ -3,3 +3,4 @@
 testdata
 .db
 *.log
+.pizzakv.json

README.md

@@ -174,8 +174,21 @@ go build -o pizzasql
 
 ## Quick Start
 
-### 1. Start PizzaKV (in a separate terminal)
+### 1. Start PizzaKV
 
+**Option A: Let PizzaSQL launch it automatically (recommended):**
+```bash
+# PizzaSQL will start PizzaKV on a random port
+./pizzasql -kv
+
+# With PizzaKV flags (e.g., disable WAL)
+./pizzasql -kv -kvflags="-iwal"
+
+# HTTP server mode with auto-launched PizzaKV
+./pizzasql -http -kv
+```
+
+**Option B: Start PizzaKV manually (in a separate terminal):**
 ```bash
 pizzakv
 ```
@@ -1473,9 +1486,12 @@ CREATE TABLE users (
 
 ```bash
 # Database options
--kv string      PizzaKV server address (default "localhost:8085")
--db string      Database name (default "pizzasql")
--e string       Execute single statement and exit
+-kvaddr string    PizzaKV server address (default "localhost:8085") (ignored if -kv is set)
+-kv               Launch PizzaKV automatically
+-kvflags string   Flags to pass to PizzaKV (e.g., "-iwal -port=9090")
+-kvinfo string    Path to PizzaKV info file (default ".pizzakv.json")
+-db string        Database name (default "pizzasql")
+-e string         Execute single statement and exit
 
 # HTTP server options
 -http           Start HTTP server

cmd/sqllogictest/main.go

@@ -212,6 +212,7 @@ func (r *runner) runFile(path string, start time.Time) error {
 		r.execQuery("DROP VIEW IF EXISTS " + v) //nolint:errcheck
 	}
 
+	labelCache := make(map[string][]string) // label → first result
 	failsBefore := r.failed
 	for _, rec := range records {
 		if r.stopOnFail && r.failed > 0 {
@@ -221,7 +222,7 @@ func (r *runner) runFile(path string, start time.Time) error {
 			r.skipped++
 			continue
 		}
-		r.runRecord(rec)
+		r.runRecord(rec, labelCache)
 		r.printProgress(path, start)
 	}
 
@@ -360,7 +361,7 @@ func collectCreatedTables(records []*record) []string {
 	return tables
 }
 
-func (r *runner) runRecord(rec *record) {
+func (r *runner) runRecord(rec *record, labelCache map[string][]string) {
 	resp, err := r.execQuery(rec.sql)
 	if err != nil {
 		r.fail(rec, "http error: %v", err)
@@ -390,29 +391,46 @@ func (r *runner) runRecord(rec *record) {
 		return
 	}
 
+	got := r.formatResults(resp, rec.typeStr)
+	ncols := len(rec.typeStr)
+	if ncols == 0 {
+		ncols = 1
+	}
+
+	// Apply sort before any comparison.
+	switch rec.sortMode {
+	case "rowsort":
+		got = sortRows(got, ncols)
+	case "valuesort":
+		g := append([]string(nil), got...)
+		sort.Strings(g)
+		got = g
+	}
+
+	// Label caching: if this query has a label, compare against first occurrence.
+	if rec.label != "" {
+		if cached, seen := labelCache[rec.label]; seen {
+			if !equalSlices(got, cached) {
+				r.fail(rec, "label %q result mismatch\n    want: %v\n    got:  %v", rec.label, cached, got)
+			} else {
+				r.pass(rec)
+			}
+			return
+		}
+		// First occurrence: store and fall through to normal expected-value check.
+		labelCache[rec.label] = got
+	}
+
 	// hash format: "N values hashing to <md5>"
 	if len(rec.expected) == 1 {
 		parts := strings.Fields(rec.expected[0])
 		if len(parts) == 5 && parts[1] == "values" && parts[2] == "hashing" && parts[3] == "to" {
 			wantCount, _ := strconv.Atoi(parts[0])
 			wantHash := parts[4]
-			got := r.formatResults(resp, rec.typeStr)
 			if len(got) != wantCount {
 				r.fail(rec, "hash record: want %d values got %d", wantCount, len(got))
 				return
 			}
-			ncols := len(rec.typeStr)
-			if ncols == 0 {
-				ncols = 1
-			}
-			switch rec.sortMode {
-			case "rowsort":
-				got = sortRows(got, ncols)
-			case "valuesort":
-				g := append([]string(nil), got...)
-				sort.Strings(g)
-				got = g
-			}
 			h := md5.Sum([]byte(strings.Join(got, "\n") + "\n"))
 			gotHash := fmt.Sprintf("%x", h)
 			if gotHash != wantHash {
@@ -424,23 +442,14 @@ func (r *runner) runRecord(rec *record) {
 		}
 	}
 
-	got := r.formatResults(resp, rec.typeStr)
 	exp := rec.expected
-
 	switch rec.sortMode {
 	case "rowsort":
-		ncols := len(rec.typeStr)
-		if ncols == 0 {
-			ncols = 1
-		}
-		got = sortRows(got, ncols)
 		exp = sortRows(exp, ncols)
 	case "valuesort":
-		g := append([]string(nil), got...)
 		e := append([]string(nil), exp...)
-		sort.Strings(g)
 		sort.Strings(e)
-		got, exp = g, e
+		exp = e
 	}
 
 	if !equalSlices(got, exp) {

main.go

@@ -12,19 +12,23 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/danfragoso/pizzasql-next/pkg/csvexport"
+	"github.com/danfragoso/pizzasql-next/pkg/csvimport"
 	"github.com/danfragoso/pizzasql-next/pkg/executor"
 	"github.com/danfragoso/pizzasql-next/pkg/httpserver"
+	"github.com/danfragoso/pizzasql-next/pkg/kvmanager"
 	"github.com/danfragoso/pizzasql-next/pkg/lexer"
 	"github.com/danfragoso/pizzasql-next/pkg/parser"
-	"github.com/danfragoso/pizzasql-next/pkg/csvexport"
-	"github.com/danfragoso/pizzasql-next/pkg/csvimport"
 	"github.com/danfragoso/pizzasql-next/pkg/sqlexport"
 	"github.com/danfragoso/pizzasql-next/pkg/sqlimport"
 	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
 
 var (
-	kvAddr     = flag.String("kv", "localhost:8085", "PizzaKV server address")
+	kvAddr     = flag.String("kvaddr", "localhost:8085", "PizzaKV server address (ignored if -kv is set)")
+	kvLaunch   = flag.Bool("kv", false, "Launch PizzaKV automatically")
+	kvFlags    = flag.String("kvflags", "", "Flags to pass to PizzaKV (e.g., \"-iwal -port=9090\")")
+	kvInfoFile = flag.String("kvinfo", ".pizzakv.json", "Path to PizzaKV info file")
 	database   = flag.String("db", "pizzasql", "Database name")
 	poolSize   = flag.Int("pool", 5, "Connection pool size")
 	timeout    = flag.Duration("timeout", 30*time.Second, "Query timeout")
@@ -46,9 +50,30 @@ var (
 	createTable  = flag.Bool("create-table", false, "Create table if not exists (CSV import)")
 )
 
+var kvManager *kvmanager.Manager
+
 func main() {
 	flag.Parse()
 
+	// If -kv flag is set, launch PizzaKV
+	if *kvLaunch {
+		if err := launchPizzaKV(); err != nil {
+			fmt.Fprintf(os.Stderr, "Failed to launch PizzaKV: %v\n", err)
+			os.Exit(1)
+		}
+		defer stopPizzaKV()
+
+		// Set up signal handling for graceful shutdown
+		sigChan := make(chan os.Signal, 1)
+		signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+		go func() {
+			<-sigChan
+			fmt.Println("\nShutting down...")
+			stopPizzaKV()
+			os.Exit(0)
+		}()
+	}
+
 	// Check if HTTP server mode is enabled
 	if *httpEnable {
 		runHTTPServer()
@@ -785,3 +810,61 @@ func runHTTPServer() {
 
 	fmt.Println("Server stopped")
 }
+
+// launchPizzaKV starts a PizzaKV instance and updates kvAddr
+func launchPizzaKV() error {
+	kvManager = kvmanager.NewManager()
+	kvManager.SetInfoFile(*kvInfoFile)
+
+	// Clean up any stale process info
+	if err := kvmanager.CleanupStaleProcess(*kvInfoFile); err != nil {
+		// Check if it's an "already running" error
+		if strings.Contains(err.Error(), "already running") {
+			fmt.Fprintf(os.Stderr, "Warning: %v\n", err)
+			fmt.Fprintf(os.Stderr, "\nOptions:\n")
+			fmt.Fprintf(os.Stderr, "  1. Use the existing instance: remove -kv flag and use -kvaddr=localhost:<port>\n")
+			fmt.Fprintf(os.Stderr, "  2. Stop it: kill %d\n", getPIDFromFile(*kvInfoFile))
+			fmt.Fprintf(os.Stderr, "  3. Delete the info file: rm %s\n\n", *kvInfoFile)
+			return err
+		}
+		return fmt.Errorf("failed to cleanup stale process: %w", err)
+	}
+
+	fmt.Println("Starting PizzaKV...")
+	info, err := kvManager.Start(*kvFlags)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("\nPizzaKV started on %s (PID: %d)\n", info.Addr, info.PID)
+	fmt.Printf("Info written to: %s\n", *kvInfoFile)
+	fmt.Println("PizzaKV is ready!\n")
+
+	// Update kvAddr to use the launched instance
+	*kvAddr = info.Addr
+
+	return nil
+}
+
+// stopPizzaKV stops the managed PizzaKV instance
+func stopPizzaKV() {
+	if kvManager != nil {
+		fmt.Println("Stopping PizzaKV...")
+		if err := kvManager.Stop(); err != nil {
+			fmt.Fprintf(os.Stderr, "Error stopping PizzaKV: %v\n", err)
+		} else {
+			fmt.Println("PizzaKV stopped")
+		}
+	}
+}
+
+// getPIDFromFile reads the PID from the info file
+func getPIDFromFile(path string) int {
+	mgr := kvmanager.NewManager()
+	mgr.SetInfoFile(path)
+	info, err := mgr.LoadInfo()
+	if err != nil {
+		return 0
+	}
+	return info.PID
+}

pkg/analyzer/analyzer.go

@@ -297,6 +297,7 @@ func (a *Analyzer) resolveFromClause(tables []parser.TableRef) error {
 				Name:    table.Name,
 				Columns: table.Columns,
 				Alias:   ref.Alias,
+				IsView:  table.IsView,
 			}
 			a.scope.DefineTable(tableInfo)
 		}
@@ -329,6 +330,7 @@ func (a *Analyzer) resolveJoin(join *parser.JoinClause) error {
 		Name:    table.Name,
 		Columns: table.Columns,
 		Alias:   join.Table.Alias,
+		IsView:  table.IsView,
 	}
 	a.scope.DefineTable(tableInfo)
 

pkg/executor/executor.go

@@ -2,6 +2,7 @@ package executor
 
 import (
 	"fmt"
+	"math"
 	"math/rand"
 	"sort"
 	"strconv"
@@ -291,8 +292,8 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		rows = filtered
 	}
 
-	// Handle explicit JOINs
-	if len(stmt.From) > 0 && stmt.From[0].Join != nil {
+	// Handle explicit JOINs from the first FROM entry (single-table+JOIN path)
+	if !isMultiTable && len(stmt.From) > 0 && stmt.From[0].Join != nil {
 		rows, err = e.executeJoins(stmt.From[0], rows)
 		if err != nil {
 			return nil, err
@@ -518,10 +519,10 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 			for _, idx := range comp[1:] {
 				degIdx, degSeed := 0, 0
 				for _, ce := range crossEdges {
-					if (ce.a == idx || ce.b == idx) {
+					if ce.a == idx || ce.b == idx {
 						degIdx++
 					}
-					if (ce.a == seed || ce.b == seed) {
+					if ce.a == seed || ce.b == seed {
 						degSeed++
 					}
 				}
@@ -736,6 +737,16 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 			}
 			rows = filtered
 		}
+
+		// Also process any JOIN clauses within FROM entries (mixed comma+JOIN syntax).
+		for _, tref := range stmt.From {
+			if tref.Join != nil {
+				rows, err = e.executeJoins(tref, rows)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
 	}
 
 	// Handle GROUP BY
@@ -773,6 +784,10 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 	// Build result
 	result := NewResult("SELECT")
 
+	// For multi-table or JOIN queries, collect all table refs for SELECT * expansion.
+	hasJoin := len(stmt.From) > 0 && stmt.From[0].Join != nil
+	allTableRefs := collectAllTableRefs(stmt.From)
+
 	// Determine columns
 	for i, col := range stmt.Columns {
 		if col.Alias != "" {
@@ -780,9 +795,21 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		} 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)
+			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 {
 			result.AddColumn(fmt.Sprintf("column%d", i+1))
@@ -794,12 +821,29 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		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])
+				if isMultiTable || hasJoin {
+					// For multi-table SELECT *, extract columns using qualified names
+					for _, tref := range allTableRefs {
+						sch, _ := e.schema.GetSchema(tref.Name)
+						if sch != nil {
+							for _, c := range sch.Columns {
+								qualKey := tref.Alias + "." + c.Name
+								val, ok := row[qualKey]
+								if !ok {
+									val = row[c.Name]
+								}
+								values = append(values, val)
+							}
+						}
+					}
+				} else {
+					// 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 {
@@ -932,6 +976,11 @@ func (e *Executor) executeCompound(c *parser.CompoundSelect) (*Result, error) {
 
 // executeSelectExpr executes a SELECT without FROM.
 func (e *Executor) executeSelectExpr(stmt *parser.SelectStmt) (*Result, error) {
+	// If any column contains an aggregate, treat as single-group aggregate over one implicit row.
+	if e.hasAggregates(stmt.Columns) {
+		return e.executeAggregateSelect(stmt, []storage.Row{{}}, nil)
+	}
+
 	result := NewResult("SELECT")
 
 	// Determine columns
@@ -1159,10 +1208,7 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		// Apply HAVING
 		if stmt.Having != nil {
 			val, err := e.evalAggregateExpr(stmt.Having, groupRows)
-			if err != nil {
-				continue
-			}
-			if !toBool(val) {
+			if err != nil || val == nil || !toBool(val) {
 				continue
 			}
 		}
@@ -1187,6 +1233,11 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		result.AddRow(values...)
 	}
 
+	// Apply DISTINCT
+	if stmt.Distinct {
+		result.Rows = e.applyDistinct(result.Rows)
+	}
+
 	// Apply ORDER BY
 	if len(stmt.OrderBy) > 0 {
 		e.sortResultRows(result, stmt.OrderBy, stmt.Columns, columnNames)
@@ -1383,6 +1434,28 @@ func (e *Executor) mergeRows(left, right storage.Row, leftAlias, rightAlias stri
 	return result
 }
 
+// collectAllTableRefs returns a flat list of (alias, tableName) pairs for all tables
+// referenced in a FROM clause, following both implicit (comma) and explicit JOIN chains.
+func collectAllTableRefs(from []parser.TableRef) []parser.TableRef {
+	var refs []parser.TableRef
+	for _, tref := range from {
+		cur := tref
+		for {
+			// Shallow copy to hold only this table (no join chain)
+			flat := parser.TableRef{Name: cur.Name, Alias: cur.Alias}
+			if flat.Alias == "" {
+				flat.Alias = flat.Name
+			}
+			refs = append(refs, flat)
+			if cur.Join == nil || cur.Join.Table == nil {
+				break
+			}
+			cur = *cur.Join.Table
+		}
+	}
+	return refs
+}
+
 // addTableAlias adds table-qualified names to a row.
 func (e *Executor) addTableAlias(row storage.Row, alias string) storage.Row {
 	result := make(storage.Row)
@@ -1778,10 +1851,49 @@ func (e *Executor) executeCreateView(stmt *parser.CreateViewStmt) (*Result, erro
 
 	e.views[name] = stmt.Select
 
+	// Derive view columns from SELECT list for catalog registration.
+	var viewCols []analyzer.ColumnInfo
+	hasStar := false
+	for _, col := range stmt.Select.Columns {
+		if col.Star {
+			hasStar = true
+			break
+		}
+		colName := col.Alias
+		if colName == "" {
+			if ref, ok := col.Expr.(*parser.ColumnRef); ok {
+				colName = ref.Column
+			} else {
+				colName = fmt.Sprintf("col_%d", len(viewCols))
+			}
+		}
+		viewCols = append(viewCols, analyzer.ColumnInfo{
+			Name:      colName,
+			TableName: stmt.View.Name,
+			Type:      analyzer.TypeAny,
+			Nullable:  true,
+		})
+	}
+	// For SELECT *, pull columns from the underlying table(s).
+	if hasStar && len(stmt.Select.From) > 0 {
+		baseName := stmt.Select.From[0].Name
+		if schema, err := e.schema.GetSchema(baseName); err == nil {
+			for _, c := range schema.Columns {
+				viewCols = append(viewCols, analyzer.ColumnInfo{
+					Name:      c.Name,
+					TableName: stmt.View.Name,
+					Type:      analyzer.TypeAny,
+					Nullable:  true,
+				})
+			}
+		}
+	}
+
 	// Register in catalog so the analyzer accepts SELECT FROM this view.
 	e.catalog.CreateTable(&analyzer.TableInfo{ //nolint:errcheck
-		Name:   stmt.View.Name,
-		IsView: true,
+		Name:    stmt.View.Name,
+		Columns: viewCols,
+		IsView:  true,
 	})
 
 	return NewResult("CREATE VIEW"), nil
@@ -2385,6 +2497,9 @@ func (e *Executor) generateOpcodes(stmt parser.Statement) []string {
 
 // evalExpr evaluates an expression.
 func (e *Executor) evalExpr(expr parser.Expr, row storage.Row) (interface{}, error) {
+	if expr == nil {
+		return nil, nil
+	}
 	switch ex := expr.(type) {
 	case *parser.LiteralExpr:
 		return e.evalLiteral(ex)
@@ -2552,21 +2667,38 @@ func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (int
 		if left == nil || right == nil {
 			return nil, nil
 		}
+		if isIntVal(left) && isIntVal(right) {
+			return toInt64(left) + toInt64(right), nil
+		}
 		return toFloat(left) + toFloat(right), nil
 	case lexer.TokenMinus:
 		if left == nil || right == nil {
 			return nil, nil
 		}
+		if isIntVal(left) && isIntVal(right) {
+			return toInt64(left) - toInt64(right), nil
+		}
 		return toFloat(left) - toFloat(right), nil
 	case lexer.TokenStar:
 		if left == nil || right == nil {
 			return nil, nil
 		}
+		if isIntVal(left) && isIntVal(right) {
+			return toInt64(left) * toInt64(right), nil
+		}
 		return toFloat(left) * toFloat(right), nil
 	case lexer.TokenSlash:
 		if left == nil || right == nil {
 			return nil, nil
 		}
+		// Integer division when both operands are integers (truncates toward zero, matching SQLite)
+		if isIntVal(left) && isIntVal(right) {
+			ri := toInt64(right)
+			if ri == 0 {
+				return nil, nil
+			}
+			return toInt64(left) / ri, nil
+		}
 		r := toFloat(right)
 		if r == 0 {
 			return nil, nil // Division by zero returns NULL
@@ -2576,6 +2708,13 @@ func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (int
 		if left == nil || right == nil {
 			return nil, nil
 		}
+		if isIntVal(left) && isIntVal(right) {
+			ri := toInt64(right)
+			if ri == 0 {
+				return nil, nil
+			}
+			return toInt64(left) % ri, nil
+		}
 		return int64(toFloat(left)) % int64(toFloat(right)), nil
 	case lexer.TokenEq:
 		if left == nil || right == nil {
@@ -2608,8 +2747,11 @@ func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (int
 		}
 		return compare(left, right) >= 0, nil
 	case lexer.TokenAND:
-		lb, rb := toBool(left), toBool(right)
-		if !lb || !rb {
+		// Three-value logic: FALSE AND x = FALSE; NULL AND TRUE = NULL; TRUE AND TRUE = TRUE
+		if left != nil && !toBool(left) {
+			return false, nil
+		}
+		if right != nil && !toBool(right) {
 			return false, nil
 		}
 		if left == nil || right == nil {
@@ -2617,8 +2759,11 @@ func (e *Executor) evalBinaryExpr(expr *parser.BinaryExpr, row storage.Row) (int
 		}
 		return true, nil
 	case lexer.TokenOR:
-		lb, rb := toBool(left), toBool(right)
-		if lb || rb {
+		// Three-value logic: TRUE OR x = TRUE; NULL OR FALSE = NULL; FALSE OR FALSE = FALSE
+		if left != nil && toBool(left) {
+			return true, nil
+		}
+		if right != nil && toBool(right) {
 			return true, nil
 		}
 		if left == nil || right == nil {
@@ -2643,11 +2788,17 @@ func (e *Executor) evalUnaryExpr(expr *parser.UnaryExpr, row storage.Row) (inter
 		if val == nil {
 			return nil, nil
 		}
+		if isIntVal(val) {
+			return -toInt64(val), nil
+		}
 		return -toFloat(val), nil
 	case lexer.TokenPlus:
 		if val == nil {
 			return nil, nil
 		}
+		if isIntVal(val) {
+			return toInt64(val), nil
+		}
 		return toFloat(val), nil
 	case lexer.TokenNOT:
 		if val == nil {
@@ -3033,14 +3184,52 @@ func (e *Executor) evalBetweenExpr(expr *parser.BetweenExpr, row storage.Row) (i
 		return nil, err
 	}
 
-	if val == nil || low == nil || high == nil {
-		return nil, nil
-	}
-	inRange := compare(val, low) >= 0 && compare(val, high) <= 0
 	if expr.Not {
-		return !inRange, nil
+		// NOT BETWEEN is equivalent to: val < low OR val > high
+		// We need to handle NULL using three-valued OR logic:
+		// NULL OR TRUE = TRUE
+		// NULL OR FALSE = NULL
+		// NULL OR NULL = NULL
+		var lessThan, greaterThan interface{}
+
+		if val == nil || low == nil {
+			lessThan = nil // NULL
+		} else {
+			lessThan = compare(val, low) < 0
+		}
+
+		if val == nil || high == nil {
+			greaterThan = nil // NULL
+		} else {
+			greaterThan = compare(val, high) > 0
+		}
+
+		// Implement three-valued OR
+		if toBool(lessThan) || toBool(greaterThan) {
+			return true, nil
+		}
+		if lessThan == nil || greaterThan == nil {
+			return nil, nil // NULL
+		}
+		return false, nil
+	} else {
+		// BETWEEN is equivalent to: val >= low AND val <= high
+		// Three-value logic: if val < low → FALSE (regardless of high); if val >= low and high is NULL → NULL
+		if val == nil {
+			return nil, nil
+		}
+		if low == nil {
+			return nil, nil // val >= NULL = NULL
+		}
+		if compare(val, low) < 0 {
+			return false, nil // val < low → definitely not in range
+		}
+		// val >= low: now check upper bound
+		if high == nil {
+			return nil, nil // val <= NULL = NULL
+		}
+		return compare(val, high) <= 0, nil
 	}
-	return inRange, nil
 }
 
 func (e *Executor) evalLikeExpr(expr *parser.LikeExpr, row storage.Row) (interface{}, error) {
@@ -3197,14 +3386,30 @@ func (e *Executor) evalAggregateExpr(expr parser.Expr, rows []storage.Row) (inte
 
 	case "SUM":
 		var sum float64
+		hasValues := false
+		var seen map[interface{}]struct{}
+		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 {
+					if fn.Distinct {
+						key := fmt.Sprintf("%v", val)
+						if _, exists := seen[key]; exists {
+							continue
+						}
+						seen[key] = struct{}{}
+					}
 					sum += toFloat(val)
+					hasValues = true
 				}
 			}
 		}
+		if !hasValues {
+			return nil, nil
+		}
 		return sum, nil
 
 	case "AVG":
@@ -3273,18 +3478,33 @@ func (e *Executor) evalExprWithAggregates(expr parser.Expr, rows []storage.Row)
 		// Apply the binary operator
 		switch ex.Op {
 		case lexer.TokenPlus:
+			if left == nil || right == nil {
+				return nil, nil
+			}
 			return toFloat(left) + toFloat(right), nil
 		case lexer.TokenMinus:
+			if left == nil || right == nil {
+				return nil, nil
+			}
 			return toFloat(left) - toFloat(right), nil
 		case lexer.TokenStar:
+			if left == nil || right == nil {
+				return nil, nil
+			}
 			return toFloat(left) * toFloat(right), nil
 		case lexer.TokenSlash:
+			if left == nil || right == nil {
+				return nil, nil
+			}
 			r := toFloat(right)
 			if r == 0 {
 				return nil, nil
 			}
 			return toFloat(left) / r, nil
 		case lexer.TokenPercent:
+			if left == nil || right == nil {
+				return nil, nil
+			}
 			return int64(toFloat(left)) % int64(toFloat(right)), nil
 		case lexer.TokenEq:
 			return compare(left, right) == 0, nil
@@ -3309,8 +3529,45 @@ func (e *Executor) evalExprWithAggregates(expr parser.Expr, rows []storage.Row)
 		}
 	case *parser.FunctionCall:
 		return e.evalAggregateExpr(expr, rows)
+	case *parser.ParenExpr:
+		// Evaluate the inner expression
+		return e.evalExprWithAggregates(ex.Expr, rows)
+	case *parser.UnaryExpr:
+		operand, err := e.evalExprWithAggregates(ex.Operand, rows)
+		if err != nil {
+			return nil, err
+		}
+		// Apply the unary operator
+		switch ex.Op {
+		case lexer.TokenPlus:
+			return operand, nil
+		case lexer.TokenMinus:
+			if operand == nil {
+				return nil, nil
+			}
+			return -toFloat(operand), nil
+		case lexer.TokenNOT:
+			return !toBool(operand), nil
+		default:
+			return nil, fmt.Errorf("unsupported unary operator: %v", ex.Op)
+		}
+	case *parser.CastExpr:
+		// Evaluate the inner expression first
+		val, err := e.evalExprWithAggregates(ex.Expr, rows)
+		if err != nil {
+			return nil, err
+		}
+		// For aggregate context, we just return the value
+		// The actual casting will be handled elsewhere if needed
+		return val, nil
 	default:
-		// Non-aggregate expression, use first row
+		// Non-aggregate expression
+		// For literals and constants, we can evaluate without a row
+		if _, ok := expr.(*parser.LiteralExpr); ok {
+			// Create a dummy empty row for literal evaluation
+			return e.evalExpr(expr, storage.Row{})
+		}
+		// For other expressions, use first row if available
 		if len(rows) > 0 {
 			return e.evalExpr(expr, rows[0])
 		}
@@ -3372,6 +3629,8 @@ func (e *Executor) isAggregate(expr parser.Expr) bool {
 		if ex.Else != nil {
 			return e.isAggregate(ex.Else)
 		}
+	case *parser.CastExpr:
+		return e.isAggregate(ex.Expr)
 	}
 	return false
 }
@@ -3482,6 +3741,40 @@ func (e *Executor) evalIntExpr(expr parser.Expr) int {
 
 // Type conversion helpers
 
+func isIntVal(v interface{}) bool {
+	switch val := v.(type) {
+	case int64:
+		return true
+	case int:
+		return true
+	case float64:
+		return val == math.Trunc(val) && !math.IsInf(val, 0) && !math.IsNaN(val)
+	case bool:
+		return true
+	default:
+		_ = val
+		return false
+	}
+}
+
+func toInt64(v interface{}) int64 {
+	switch val := v.(type) {
+	case int64:
+		return val
+	case int:
+		return int64(val)
+	case float64:
+		return int64(val)
+	case bool:
+		if val {
+			return 1
+		}
+		return 0
+	default:
+		return 0
+	}
+}
+
 func toFloat(v interface{}) float64 {
 	switch val := v.(type) {
 	case nil:

pkg/kvmanager/kvmanager.go

@@ -0,0 +1,390 @@
+package kvmanager
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+)
+
+// KVInfo contains information about the running PizzaKV instance
+type KVInfo struct {
+	PID  int    `json:"pid"`
+	Port int    `json:"port"`
+	Addr string `json:"addr"`
+}
+
+// Manager handles the lifecycle of a PizzaKV process
+type Manager struct {
+	cmd      *exec.Cmd
+	infoFile string
+	info     *KVInfo
+}
+
+// NewManager creates a new KVManager
+func NewManager() *Manager {
+	return &Manager{
+		infoFile: ".pizzakv.json",
+	}
+}
+
+// SetInfoFile sets a custom path for the info file
+func (m *Manager) SetInfoFile(path string) {
+	m.infoFile = path
+}
+
+// Start launches pizzakv with the given flags on a random available port
+func (m *Manager) Start(kvFlags string) (*KVInfo, error) {
+	// Find an available port between 1024-9999
+	port, err := findAvailablePortInRange(1024, 9999)
+	if err != nil {
+		return nil, fmt.Errorf("failed to find available port: %w", err)
+	}
+
+	// Build the command arguments
+	// PizzaKV uses -port=XXXX format (single dash)
+	args := []string{fmt.Sprintf("-port=%d", port)}
+
+	// Parse and add custom flags if provided
+	if kvFlags != "" {
+		customArgs := parseFlags(kvFlags)
+		args = append(args, customArgs...)
+	}
+
+	// Create the command
+	cmd := exec.Command("pizzakv", args...)
+
+	// Set up process group to allow clean shutdown
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true,
+	}
+
+	// Redirect output to /dev/null or capture it
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	// Start the process
+	if err := cmd.Start(); err != nil {
+		return nil, fmt.Errorf("failed to start pizzakv: %w", err)
+	}
+
+	m.cmd = cmd
+	m.info = &KVInfo{
+		PID:  cmd.Process.Pid,
+		Port: port,
+		Addr: fmt.Sprintf("localhost:%d", port),
+	}
+
+	// Wait for the process to start and begin listening
+	// We need to wait longer to ensure PizzaKV is actually listening
+	time.Sleep(500 * time.Millisecond)
+
+	// Check if process is still running
+	if !m.IsRunning() {
+		return nil, fmt.Errorf("pizzakv process exited immediately after starting")
+	}
+
+	fmt.Println("Waiting for PizzaKV to be ready...")
+
+	// Wait for PizzaKV to be ready (finish restoring records, etc.)
+	if err := m.waitForReady(port, 30*time.Second); err != nil {
+		m.Stop()
+		return nil, fmt.Errorf("pizzakv did not become ready: %w", err)
+	}
+
+	// Write info to file
+	if err := m.writeInfoFile(); err != nil {
+		m.Stop()
+		return nil, fmt.Errorf("failed to write info file: %w", err)
+	}
+
+	return m.info, nil
+}
+
+// Stop stops the pizzakv process
+func (m *Manager) Stop() error {
+	if m.cmd == nil || m.cmd.Process == nil {
+		return nil
+	}
+
+	// Try graceful shutdown first
+	if err := m.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+		// If SIGTERM fails, try SIGKILL
+		if err := m.cmd.Process.Kill(); err != nil {
+			return fmt.Errorf("failed to kill process: %w", err)
+		}
+	}
+
+	// Wait for process to exit with timeout
+	done := make(chan error, 1)
+	go func() {
+		_, err := m.cmd.Process.Wait()
+		done <- err
+	}()
+
+	select {
+	case <-done:
+		// Process exited
+	case <-time.After(5 * time.Second):
+		// Timeout, force kill
+		m.cmd.Process.Kill()
+	}
+
+	// Clean up info file
+	os.Remove(m.infoFile)
+
+	return nil
+}
+
+// IsRunning checks if the pizzakv process is still running
+func (m *Manager) IsRunning() bool {
+	if m.cmd == nil || m.cmd.Process == nil {
+		return false
+	}
+
+	// Send signal 0 to check if process exists
+	err := m.cmd.Process.Signal(syscall.Signal(0))
+	return err == nil
+}
+
+// waitForReady waits for PizzaKV to be ready to accept connections
+func (m *Manager) waitForReady(port int, timeout time.Duration) error {
+	addr := fmt.Sprintf("127.0.0.1:%d", port)
+	deadline := time.Now().Add(timeout)
+
+	for time.Now().Before(deadline) {
+		// Check if process is still running
+		if !m.IsRunning() {
+			return fmt.Errorf("process died while waiting for ready")
+		}
+
+		// Try to connect
+		conn, err := net.DialTimeout("tcp", addr, 500*time.Millisecond)
+		if err == nil {
+			conn.Close()
+			// Successfully connected, PizzaKV is ready
+			return nil
+		}
+
+		// Wait a bit before retrying
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	return fmt.Errorf("timeout waiting for PizzaKV to become ready on port %d", port)
+}
+
+// GetInfo returns the KVInfo for the running instance
+func (m *Manager) GetInfo() *KVInfo {
+	return m.info
+}
+
+// LoadInfo loads KVInfo from the info file
+func (m *Manager) LoadInfo() (*KVInfo, error) {
+	data, err := os.ReadFile(m.infoFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read info file: %w", err)
+	}
+
+	var info KVInfo
+	if err := json.Unmarshal(data, &info); err != nil {
+		return nil, fmt.Errorf("failed to parse info file: %w", err)
+	}
+
+	return &info, nil
+}
+
+// writeInfoFile writes the KVInfo to a file
+func (m *Manager) writeInfoFile() error {
+	data, err := json.MarshalIndent(m.info, "", "  ")
+	if err != nil {
+		return fmt.Errorf("failed to marshal info: %w", err)
+	}
+
+	// Create directory if it doesn't exist
+	dir := filepath.Dir(m.infoFile)
+	if dir != "." {
+		if err := os.MkdirAll(dir, 0755); err != nil {
+			return fmt.Errorf("failed to create directory: %w", err)
+		}
+	}
+
+	if err := os.WriteFile(m.infoFile, data, 0644); err != nil {
+		return fmt.Errorf("failed to write info file: %w", err)
+	}
+
+	return nil
+}
+
+// findAvailablePort finds a random available port (kept for compatibility)
+func findAvailablePort() (int, error) {
+	return findAvailablePortInRange(1024, 65535)
+}
+
+// findAvailablePortInRange finds a random available port within the specified range
+func findAvailablePortInRange(minPort, maxPort int) (int, error) {
+	// Try up to 100 times to find an available port
+	for i := 0; i < 100; i++ {
+		// Generate random port in range
+		port := minPort + (int(time.Now().UnixNano()) % (maxPort - minPort + 1))
+
+		// Try to listen on this port
+		addr := fmt.Sprintf("127.0.0.1:%d", port)
+		listener, err := net.Listen("tcp", addr)
+		if err != nil {
+			// Port is in use, try another
+			continue
+		}
+		defer listener.Close()
+
+		// Port is available
+		return port, nil
+	}
+
+	return 0, fmt.Errorf("could not find available port in range %d-%d after 100 attempts", minPort, maxPort)
+}
+
+// parseFlags parses a flag string like "-iwal -port=9090" into a slice of strings
+func parseFlags(flags string) []string {
+	// Trim whitespace
+	flags = strings.TrimSpace(flags)
+	if flags == "" {
+		return nil
+	}
+
+	var result []string
+	var current strings.Builder
+	inQuote := false
+
+	for i, r := range flags {
+		switch r {
+		case '"', '\'':
+			inQuote = !inQuote
+		case ' ':
+			if !inQuote {
+				if current.Len() > 0 {
+					result = append(result, current.String())
+					current.Reset()
+				}
+			} else {
+				current.WriteRune(r)
+			}
+		default:
+			current.WriteRune(r)
+		}
+
+		// Handle last character
+		if i == len(flags)-1 && current.Len() > 0 {
+			result = append(result, current.String())
+		}
+	}
+
+	return result
+}
+
+// CleanupStaleProcess checks if there's a stale PID file and cleans it up
+func CleanupStaleProcess(infoFile string) error {
+	data, err := os.ReadFile(infoFile)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil // No file, nothing to clean
+		}
+		return err
+	}
+
+	var info KVInfo
+	if err := json.Unmarshal(data, &info); err != nil {
+		// Invalid file, just remove it
+		return os.Remove(infoFile)
+	}
+
+	// Check if process is still running
+	process, err := os.FindProcess(info.PID)
+	if err != nil {
+		// Process doesn't exist, remove file
+		return os.Remove(infoFile)
+	}
+
+	// Try to signal the process
+	err = process.Signal(syscall.Signal(0))
+	if err != nil {
+		// Process is dead, remove file
+		return os.Remove(infoFile)
+	}
+
+	// Process exists, but is it actually PizzaKV responding on that port?
+	// Try to connect to the port
+	addr := fmt.Sprintf("127.0.0.1:%d", info.Port)
+	conn, err := net.DialTimeout("tcp", addr, 1*time.Second)
+	if err != nil {
+		// Port is not responding, process might be stale or not PizzaKV
+		// Remove the file and let user launch a new instance
+		return os.Remove(infoFile)
+	}
+	conn.Close()
+
+	// Process is still running and responding on the port
+	return fmt.Errorf("pizzakv process (PID %d) is already running on port %d", info.PID, info.Port)
+}
+
+// KillExisting kills an existing pizzakv process based on the info file
+func KillExisting(infoFile string) error {
+	data, err := os.ReadFile(infoFile)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil // No file, nothing to kill
+		}
+		return err
+	}
+
+	var info KVInfo
+	if err := json.Unmarshal(data, &info); err != nil {
+		// Invalid file, just remove it
+		return os.Remove(infoFile)
+	}
+
+	// Try to kill the process
+	process, err := os.FindProcess(info.PID)
+	if err != nil {
+		// Process doesn't exist, remove file
+		return os.Remove(infoFile)
+	}
+
+	// Try SIGTERM first
+	if err := process.Signal(syscall.SIGTERM); err == nil {
+		// Wait a bit for graceful shutdown
+		time.Sleep(1 * time.Second)
+
+		// Check if still running
+		if err := process.Signal(syscall.Signal(0)); err == nil {
+			// Still running, force kill
+			process.Kill()
+		}
+	} else {
+		// SIGTERM failed, try SIGKILL
+		process.Kill()
+	}
+
+	// Remove the info file
+	return os.Remove(infoFile)
+}
+
+// ParsePort parses a port from a string (e.g., "localhost:8085" -> 8085)
+func ParsePort(addr string) (int, error) {
+	parts := strings.Split(addr, ":")
+	if len(parts) != 2 {
+		return 0, fmt.Errorf("invalid address format: %s", addr)
+	}
+
+	port, err := strconv.Atoi(parts[1])
+	if err != nil {
+		return 0, fmt.Errorf("invalid port: %s", parts[1])
+	}
+
+	return port, nil
+}

pkg/kvmanager/kvmanager_test.go

@@ -0,0 +1,165 @@
+package kvmanager
+
+import (
+	"testing"
+)
+
+func TestFindAvailablePort(t *testing.T) {
+	port, err := findAvailablePort()
+	if err != nil {
+		t.Fatalf("Failed to find available port: %v", err)
+	}
+
+	if port < 1024 || port > 65535 {
+		t.Errorf("Port %d is outside valid range 1024-65535", port)
+	}
+}
+
+func TestFindAvailablePortInRange(t *testing.T) {
+	// Test PizzaKV range (1024-9999)
+	port, err := findAvailablePortInRange(1024, 9999)
+	if err != nil {
+		t.Fatalf("Failed to find available port: %v", err)
+	}
+
+	if port < 1024 || port > 9999 {
+		t.Errorf("Port %d is outside requested range 1024-9999", port)
+	}
+
+	// Test custom range
+	port, err = findAvailablePortInRange(5000, 5100)
+	if err != nil {
+		t.Fatalf("Failed to find available port in custom range: %v", err)
+	}
+
+	if port < 5000 || port > 5100 {
+		t.Errorf("Port %d is outside requested range 5000-5100", port)
+	}
+}
+
+func TestParseFlags(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected []string
+	}{
+		{
+			name:     "empty string",
+			input:    "",
+			expected: nil,
+		},
+		{
+			name:     "single flag",
+			input:    "-iwal",
+			expected: []string{"-iwal"},
+		},
+		{
+			name:     "multiple flags",
+			input:    "-iwal -port=9090",
+			expected: []string{"-iwal", "-port=9090"},
+		},
+		{
+			name:     "flags with quotes",
+			input:    "-path=\"/tmp/my path\" -verbose",
+			expected: []string{"-path=/tmp/my path", "-verbose"},
+		},
+		{
+			name:     "flags with extra spaces",
+			input:    "  -iwal   -port=9090  ",
+			expected: []string{"-iwal", "-port=9090"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := parseFlags(tt.input)
+
+			if len(result) != len(tt.expected) {
+				t.Errorf("Expected %d args, got %d", len(tt.expected), len(result))
+				return
+			}
+
+			for i := range result {
+				if result[i] != tt.expected[i] {
+					t.Errorf("Arg %d: expected %q, got %q", i, tt.expected[i], result[i])
+				}
+			}
+		})
+	}
+}
+
+func TestParsePort(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected int
+		wantErr  bool
+	}{
+		{
+			name:     "valid address",
+			input:    "localhost:8085",
+			expected: 8085,
+			wantErr:  false,
+		},
+		{
+			name:     "valid IP address",
+			input:    "127.0.0.1:9090",
+			expected: 9090,
+			wantErr:  false,
+		},
+		{
+			name:     "invalid format",
+			input:    "localhost",
+			expected: 0,
+			wantErr:  true,
+		},
+		{
+			name:     "invalid port",
+			input:    "localhost:abc",
+			expected: 0,
+			wantErr:  true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := ParsePort(tt.input)
+
+			if tt.wantErr {
+				if err == nil {
+					t.Error("Expected error but got none")
+				}
+				return
+			}
+
+			if err != nil {
+				t.Errorf("Unexpected error: %v", err)
+				return
+			}
+
+			if result != tt.expected {
+				t.Errorf("Expected port %d, got %d", tt.expected, result)
+			}
+		})
+	}
+}
+
+func TestKVInfo(t *testing.T) {
+	info := &KVInfo{
+		PID:  12345,
+		Port: 8085,
+		Addr: "localhost:8085",
+	}
+
+	if info.PID != 12345 {
+		t.Errorf("Expected PID 12345, got %d", info.PID)
+	}
+
+	if info.Port != 8085 {
+		t.Errorf("Expected Port 8085, got %d", info.Port)
+	}
+
+	if info.Addr != "localhost:8085" {
+		t.Errorf("Expected Addr 'localhost:8085', got %s", info.Addr)
+	}
+}

pkg/parser/parser.go

@@ -512,7 +512,33 @@ func (p *Parser) parseTableRef() (*TableRef, error) {
 
 			return ref, nil
 		}
-		return nil, p.curError("expected SELECT after ( in FROM clause")
+		// Parenthesized table expression: (table1 JOIN table2) or (table1, table2)
+		innerRefs, err := p.parseTableRefs()
+		if err != nil {
+			return nil, err
+		}
+		if !p.curTokenIs(lexer.TokenRParen) {
+			return nil, p.curError("expected ) after table expression")
+		}
+		p.nextToken()
+		// Flatten: use first ref as base, attach remaining as joins
+		if len(innerRefs) == 0 {
+			return nil, p.curError("empty table expression")
+		}
+		base := &innerRefs[0]
+		for i := 1; i < len(innerRefs); i++ {
+			cur := base
+			for cur.Join != nil && cur.Join.Table != nil {
+				cur = cur.Join.Table
+			}
+			extra := innerRefs[i]
+			if cur.Join == nil {
+				cur.Join = &JoinClause{Type: JoinCross, Table: &extra}
+			} else {
+				cur.Join.Table = &extra
+			}
+		}
+		return base, nil
 	}
 
 	if !p.curTokenIs(lexer.TokenIdent) {
@@ -2145,10 +2171,13 @@ func (p *Parser) parseFunctionCall(name string) (Expr, error) {
 
 	p.nextToken() // consume (
 
-	// Check for DISTINCT
+	// Check for DISTINCT or ALL
 	if p.curTokenIs(lexer.TokenDISTINCT) {
 		fn.Distinct = true
 		p.nextToken()
+	} else if p.curTokenIs(lexer.TokenALL) {
+		// ALL is the default behavior, just skip it
+		p.nextToken()
 	}
 
 	// Check for * (COUNT(*))