| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927 |
- package main
- import (
- "bufio"
- "context"
- "flag"
- "fmt"
- "net/http"
- "os"
- "os/signal"
- "strings"
- "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/pgserver"
- pizzaruntime "github.com/danfragoso/pizzasql-next/pkg/runtime"
- "github.com/danfragoso/pizzasql-next/pkg/sqlexport"
- "github.com/danfragoso/pizzasql-next/pkg/sqlimport"
- "github.com/danfragoso/pizzasql-next/pkg/sqliteimport"
- "github.com/danfragoso/pizzasql-next/pkg/storage"
- )
- var (
- kvAddr = flag.String("kvaddr", "", "PizzaKV server address (default: auto-connect to managed instance)")
- kvLaunch = flag.Bool("kv", false, "Launch PizzaKV automatically")
- kvFlags = flag.String("kvflags", "", "Flags to pass to PizzaKV (e.g., \"-iwal\")")
- database = flag.String("db", "pizzasql", "Database name")
- poolSize = flag.Int("pool", 100, "Connection pool size")
- timeout = flag.Duration("timeout", 120*time.Second, "Query timeout")
- httpEnable = flag.Bool("http", false, "Enable HTTP server")
- httpHost = flag.String("http-host", "localhost", "HTTP server host")
- httpPort = flag.Int("http-port", 8080, "HTTP server port")
- httpCORS = flag.Bool("http-cors", true, "Enable CORS")
- httpAuth = flag.Bool("http-auth", false, "Enable authentication")
- httpCompression = flag.Bool("http-compression", true, "Enable HTTP response compression")
- quiet = flag.Bool("quiet", false, "Disable request/query logging")
- apiKeys = flag.String("api-keys", "", "Comma-separated API keys")
- // PostgreSQL wire protocol server flags
- pgEnable = flag.Bool("pg", false, "Enable PostgreSQL wire protocol server")
- pgHost = flag.String("pg-host", "localhost", "PostgreSQL server host")
- pgPort = flag.Int("pg-port", 5432, "PostgreSQL server port")
- // Export/Import flags
- exportFile = flag.String("o", "", "Output file for export")
- importFile = flag.String("i", "", "Input file for import")
- exportTable = flag.String("table", "", "Specific table to export (empty = all)")
- exportDrop = flag.Bool("drop", false, "Include DROP TABLE statements in export")
- ignoreErrors = flag.Bool("ignore-errors", false, "Continue import on errors")
- exportFormat = flag.String("format", "", "Export/import format: sql, csv (auto-detect from extension)")
- createTable = flag.Bool("create-table", false, "Create table if not exists (CSV import)")
- )
- var kvManager *kvmanager.Manager
- var startPprofServerHook func() *http.Server
- func main() {
- flag.Parse()
- // Warn if other pizzasql instances are running; prompt to continue.
- if err := pizzaruntime.CheckExistingInstances(); err != nil {
- fmt.Fprintf(os.Stderr, "%v\n", err)
- os.Exit(1)
- }
- // Register this process in its own runtime directory.
- pizzaruntime.WritePizzaSQL(os.Getpid(), 0, 0)
- defer pizzaruntime.Cleanup()
- // 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()
- pizzaruntime.Cleanup()
- os.Exit(0)
- }()
- // If -kv flag is set, always launch a dedicated PizzaKV for this instance.
- if *kvLaunch {
- if err := launchPizzaKV(); err != nil {
- fmt.Fprintf(os.Stderr, "Failed to launch PizzaKV: %v\n", err)
- os.Exit(1)
- }
- defer stopPizzaKV()
- } else if *kvAddr == "" {
- *kvAddr = "localhost:8085"
- }
- // Start whichever servers are enabled, then block until signal.
- if *httpEnable || *pgEnable {
- httpRuntimePort := 0
- pgRuntimePort := 0
- if *httpEnable {
- httpRuntimePort = *httpPort
- }
- if *pgEnable {
- pgRuntimePort = *pgPort
- }
- if err := pizzaruntime.WritePizzaSQL(os.Getpid(), httpRuntimePort, pgRuntimePort); err != nil {
- fmt.Fprintf(os.Stderr, "Failed to write runtime info: %v\n", err)
- }
- runServers()
- return
- }
- // Check for export command
- if *exportFile != "" {
- runExport()
- return
- }
- // Check for import command
- if *importFile != "" {
- runImport()
- return
- }
- // Check for command-line SQL
- args := flag.Args()
- if len(args) > 0 {
- // Execute single SQL statement
- sql := strings.Join(args, " ")
- executeSingle(sql)
- return
- }
- // Check for piped input
- stat, _ := os.Stdin.Stat()
- if (stat.Mode() & os.ModeCharDevice) == 0 {
- // Input is from pipe
- executePipe()
- return
- }
- // Interactive REPL mode
- runREPL()
- }
- func executeSingle(sql string) {
- // Try to connect to PizzaKV
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- // Fall back to expression-only mode
- executeExpressionOnly(sql)
- return
- }
- defer pool.Close()
- schema := storage.NewSchemaManager(pool, *database)
- table := storage.NewTableManager(pool, schema, *database)
- exec := executor.New(schema, table)
- exec.SyncCatalog()
- result, err := executeSQL(exec, sql)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- os.Exit(1)
- }
- fmt.Print(result.String())
- }
- func executePipe() {
- // Try to connect to PizzaKV
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- // Fall back to expression-only mode
- scanner := bufio.NewScanner(os.Stdin)
- for scanner.Scan() {
- sql := strings.TrimSpace(scanner.Text())
- if sql == "" || strings.HasPrefix(sql, "--") {
- continue
- }
- executeExpressionOnly(sql)
- }
- return
- }
- defer pool.Close()
- schema := storage.NewSchemaManager(pool, *database)
- table := storage.NewTableManager(pool, schema, *database)
- exec := executor.New(schema, table)
- exec.SyncCatalog()
- scanner := bufio.NewScanner(os.Stdin)
- for scanner.Scan() {
- sql := strings.TrimSpace(scanner.Text())
- if sql == "" || strings.HasPrefix(sql, "--") {
- continue
- }
- result, err := executeSQL(exec, sql)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- continue
- }
- fmt.Print(result.String())
- }
- }
- func runREPL() {
- fmt.Println("PizzaSQL - SQL-92 compatible database")
- fmt.Println("Type 'help' for usage, 'quit' to exit")
- fmt.Println()
- // Try to connect to PizzaKV
- var pool *storage.KVPool
- var schema *storage.SchemaManager
- var table *storage.TableManager
- var exec *executor.Executor
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- fmt.Printf("Warning: Cannot connect to PizzaKV at %s\n", *kvAddr)
- fmt.Println("Running in expression-only mode (no table storage)")
- fmt.Println()
- } else {
- schema = storage.NewSchemaManager(pool, *database)
- table = storage.NewTableManager(pool, schema, *database)
- exec = executor.New(schema, table)
- exec.SyncCatalog()
- fmt.Printf("Connected to PizzaKV at %s (database: %s)\n\n", *kvAddr, *database)
- }
- reader := bufio.NewReader(os.Stdin)
- var sqlBuffer strings.Builder
- for {
- if sqlBuffer.Len() == 0 {
- fmt.Print("pizzasql> ")
- } else {
- fmt.Print(" -> ")
- }
- line, err := reader.ReadString('\n')
- if err != nil {
- fmt.Println()
- break
- }
- line = strings.TrimSpace(line)
- // Handle special commands
- switch strings.ToLower(line) {
- case "quit", "exit", "\\q":
- fmt.Println("Goodbye!")
- if pool != nil {
- pool.Close()
- }
- return
- case "help", "\\h":
- printHelp()
- continue
- case "tables", "\\dt":
- if schema != nil {
- listTables(schema)
- } else {
- fmt.Println("Not connected to database")
- }
- continue
- case "clear", "\\c":
- sqlBuffer.Reset()
- fmt.Println("Buffer cleared")
- continue
- }
- // Skip empty lines and comments
- if line == "" || strings.HasPrefix(line, "--") {
- continue
- }
- // Accumulate SQL
- if sqlBuffer.Len() > 0 {
- sqlBuffer.WriteString(" ")
- }
- sqlBuffer.WriteString(line)
- // Check if statement is complete (ends with semicolon)
- sql := sqlBuffer.String()
- if !strings.HasSuffix(sql, ";") {
- continue
- }
- // Remove semicolon and execute
- sql = strings.TrimSuffix(sql, ";")
- sqlBuffer.Reset()
- if exec != nil {
- result, err := executeSQL(exec, sql)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- continue
- }
- fmt.Print(result.String())
- } else {
- executeExpressionOnly(sql)
- }
- }
- }
- func executeSQL(exec *executor.Executor, sql string) (*executor.Result, error) {
- l := lexer.New(sql)
- p := parser.New(l)
- stmt, err := p.Parse()
- if err != nil {
- return nil, fmt.Errorf("parse error: %w", err)
- }
- return exec.Execute(stmt)
- }
- func executeExpressionOnly(sql string) {
- l := lexer.New(sql)
- p := parser.New(l)
- stmt, err := p.Parse()
- if err != nil {
- fmt.Fprintf(os.Stderr, "Parse error: %v\n", err)
- return
- }
- // For SELECT statements without FROM, we can evaluate expressions
- if sel, ok := stmt.(*parser.SelectStmt); ok && len(sel.From) == 0 {
- exec := &executor.Executor{}
- result, err := executeSelectExpr(exec, sel)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- return
- }
- fmt.Print(result.String())
- return
- }
- // For other statements, just print what was parsed
- switch s := stmt.(type) {
- case *parser.SelectStmt:
- fmt.Printf("SELECT statement with %d columns\n", len(s.Columns))
- if len(s.From) > 0 {
- fmt.Printf(" FROM: %s\n", s.From[0].Name)
- }
- if s.Where != nil {
- fmt.Println(" WHERE: <condition>")
- }
- fmt.Println("(Not connected to database - cannot execute)")
- case *parser.InsertStmt:
- fmt.Printf("INSERT into %s (%d rows)\n", s.Table.Name, len(s.Values))
- fmt.Println("(Not connected to database - cannot execute)")
- case *parser.UpdateStmt:
- fmt.Printf("UPDATE %s (%d assignments)\n", s.Table.Name, len(s.Set))
- fmt.Println("(Not connected to database - cannot execute)")
- case *parser.DeleteStmt:
- fmt.Printf("DELETE from %s\n", s.Table.Name)
- fmt.Println("(Not connected to database - cannot execute)")
- case *parser.CreateTableStmt:
- fmt.Printf("CREATE TABLE %s (%d columns)\n", s.Table.Name, len(s.Columns))
- fmt.Println("(Not connected to database - cannot execute)")
- case *parser.DropTableStmt:
- fmt.Printf("DROP TABLE %s\n", s.Tables[0].Name)
- fmt.Println("(Not connected to database - cannot execute)")
- default:
- fmt.Printf("Parsed: %T\n", stmt)
- }
- }
- // executeSelectExpr handles SELECT without FROM (expression evaluation)
- func executeSelectExpr(exec *executor.Executor, stmt *parser.SelectStmt) (*executor.Result, error) {
- result := executor.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 using reflection to access private method
- // For simplicity, we'll use a minimal evaluator here
- values := make([]interface{}, len(stmt.Columns))
- for i, col := range stmt.Columns {
- val, err := evalExprSimple(col.Expr)
- if err != nil {
- return nil, err
- }
- values[i] = val
- }
- result.AddRow(values...)
- return result, nil
- }
- // evalExprSimple is a simplified expression evaluator for standalone expressions
- func evalExprSimple(expr parser.Expr) (interface{}, error) {
- switch e := expr.(type) {
- case *parser.LiteralExpr:
- switch e.Type {
- case lexer.TokenNumber:
- if strings.Contains(e.Value, ".") {
- var f float64
- fmt.Sscanf(e.Value, "%f", &f)
- return f, nil
- }
- var i int64
- fmt.Sscanf(e.Value, "%d", &i)
- return i, nil
- case lexer.TokenString:
- return e.Value, nil
- case lexer.TokenNULL:
- return nil, nil
- case lexer.TokenTRUE:
- return true, nil
- case lexer.TokenFALSE:
- return false, nil
- }
- case *parser.BinaryExpr:
- left, err := evalExprSimple(e.Left)
- if err != nil {
- return nil, err
- }
- right, err := evalExprSimple(e.Right)
- if err != nil {
- return nil, err
- }
- return evalBinarySimple(e.Op, left, right)
- case *parser.UnaryExpr:
- val, err := evalExprSimple(e.Operand)
- if err != nil {
- return nil, err
- }
- switch e.Op {
- case lexer.TokenMinus:
- return -toFloatSimple(val), nil
- case lexer.TokenNOT:
- return !toBoolSimple(val), nil
- }
- return val, nil
- case *parser.ParenExpr:
- return evalExprSimple(e.Expr)
- }
- return nil, fmt.Errorf("unsupported expression type: %T", expr)
- }
- func evalBinarySimple(op lexer.TokenType, left, right interface{}) (interface{}, error) {
- switch op {
- case lexer.TokenPlus:
- return toFloatSimple(left) + toFloatSimple(right), nil
- case lexer.TokenMinus:
- return toFloatSimple(left) - toFloatSimple(right), nil
- case lexer.TokenStar:
- return toFloatSimple(left) * toFloatSimple(right), nil
- case lexer.TokenSlash:
- r := toFloatSimple(right)
- if r == 0 {
- return nil, nil
- }
- return toFloatSimple(left) / r, nil
- case lexer.TokenEq:
- return compareSimple(left, right) == 0, nil
- case lexer.TokenNeq:
- return compareSimple(left, right) != 0, nil
- case lexer.TokenLt:
- return compareSimple(left, right) < 0, nil
- case lexer.TokenGt:
- return compareSimple(left, right) > 0, nil
- case lexer.TokenLte:
- return compareSimple(left, right) <= 0, nil
- case lexer.TokenGte:
- return compareSimple(left, right) >= 0, nil
- case lexer.TokenAND:
- return toBoolSimple(left) && toBoolSimple(right), nil
- case lexer.TokenOR:
- return toBoolSimple(left) || toBoolSimple(right), nil
- }
- return nil, fmt.Errorf("unsupported operator: %v", op)
- }
- func toFloatSimple(v interface{}) float64 {
- switch val := v.(type) {
- case int64:
- return float64(val)
- case float64:
- return val
- case bool:
- if val {
- return 1
- }
- return 0
- }
- return 0
- }
- func toBoolSimple(v interface{}) bool {
- switch val := v.(type) {
- case bool:
- return val
- case int64:
- return val != 0
- case float64:
- return val != 0
- }
- return false
- }
- func compareSimple(a, b interface{}) int {
- fa := toFloatSimple(a)
- fb := toFloatSimple(b)
- if fa < fb {
- return -1
- }
- if fa > fb {
- return 1
- }
- return 0
- }
- func printHelp() {
- fmt.Println("PizzaSQL Commands:")
- fmt.Println(" help, \\h Show this help")
- fmt.Println(" quit, \\q Exit the program")
- fmt.Println(" tables, \\dt List all tables")
- fmt.Println(" clear, \\c Clear the input buffer")
- fmt.Println()
- fmt.Println("SQL Statements (end with semicolon):")
- fmt.Println(" SELECT ... FROM ... WHERE ...")
- fmt.Println(" INSERT INTO table (cols) VALUES (...)")
- fmt.Println(" UPDATE table SET col = val WHERE ...")
- fmt.Println(" DELETE FROM table WHERE ...")
- fmt.Println(" CREATE TABLE table (col TYPE, ...)")
- fmt.Println(" DROP TABLE table")
- fmt.Println()
- fmt.Println("Expression Mode (SELECT without FROM):")
- fmt.Println(" SELECT 1 + 2 * 3;")
- fmt.Println(" SELECT UPPER('hello');")
- fmt.Println()
- fmt.Println("Export/Import:")
- fmt.Println(" pizzasql -db mydb -o backup.sql Export database to SQL file")
- fmt.Println(" pizzasql -db mydb -table users -o t.sql Export single table")
- fmt.Println(" pizzasql -db mydb -o backup.sql -drop Include DROP TABLE statements")
- fmt.Println(" pizzasql -db mydb -i backup.sql Import SQL file")
- fmt.Println(" pizzasql -db mydb -i source.db Import SQLite .db file (auto-detected)")
- fmt.Println(" pizzasql -db mydb -i source.db -ignore-errors Import, skip errors")
- fmt.Println()
- fmt.Println("CSV Format:")
- fmt.Println(" pizzasql -db mydb -table users -o users.csv Export table to CSV")
- fmt.Println(" pizzasql -db mydb -table users -i users.csv Import CSV to table")
- fmt.Println(" pizzasql -db mydb -table new -i data.csv -create-table Create table from CSV")
- }
- func listTables(schema *storage.SchemaManager) {
- tables, err := schema.ListTables()
- if err != nil {
- fmt.Fprintf(os.Stderr, "Error: %v\n", err)
- return
- }
- if len(tables) == 0 {
- fmt.Println("No tables found")
- return
- }
- fmt.Println("Tables:")
- for _, t := range tables {
- fmt.Printf(" %s\n", t)
- }
- }
- func runExport() {
- // Connect to PizzaKV
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to connect to PizzaKV at %s: %v\n", *kvAddr, err)
- os.Exit(1)
- }
- defer pool.Close()
- schema := storage.NewSchemaManager(pool, *database)
- table := storage.NewTableManager(pool, schema, *database)
- // Determine format from flag or file extension
- format := strings.ToLower(*exportFormat)
- if format == "" {
- format = detectFileFormat(*exportFile)
- }
- switch format {
- case "csv":
- // CSV export requires a table name
- if *exportTable == "" {
- fmt.Fprintf(os.Stderr, "CSV export requires -table flag\n")
- os.Exit(1)
- }
- csvOpts := csvexport.DefaultExportOptions()
- csvOpts.Table = *exportTable
- data, err := csvexport.ExportTableToBytes(schema, table, csvOpts)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Export failed: %v\n", err)
- os.Exit(1)
- }
- err = os.WriteFile(*exportFile, data, 0644)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to write file: %v\n", err)
- os.Exit(1)
- }
- fmt.Printf("Exported table '%s' to %s (CSV)\n", *exportTable, *exportFile)
- default: // sql, sqlite
- // Configure export options
- opts := sqlexport.ExportOptions{
- IncludeData: true,
- DropTables: *exportDrop,
- }
- if *exportTable != "" {
- opts.Tables = []string{*exportTable}
- }
- // Export database
- sql, err := sqlexport.ExportDatabase(schema, table, opts)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Export failed: %v\n", err)
- os.Exit(1)
- }
- // Write to file
- err = os.WriteFile(*exportFile, []byte(sql), 0644)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to write file: %v\n", err)
- os.Exit(1)
- }
- fmt.Printf("Exported database '%s' to %s\n", *database, *exportFile)
- }
- }
- func runImport() {
- // Connect to PizzaKV
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to connect to PizzaKV at %s: %v\n", *kvAddr, err)
- os.Exit(1)
- }
- defer pool.Close()
- schema := storage.NewSchemaManager(pool, *database)
- table := storage.NewTableManager(pool, schema, *database)
- exec := executor.New(schema, table)
- exec.SyncCatalog()
- // Read file
- data, err := os.ReadFile(*importFile)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to read file: %v\n", err)
- os.Exit(1)
- }
- // Determine format from flag or file extension
- format := strings.ToLower(*exportFormat)
- if format == "" {
- format = detectFileFormat(*importFile)
- }
- switch format {
- case "csv":
- // CSV import requires a table name
- if *exportTable == "" {
- fmt.Fprintf(os.Stderr, "CSV import requires -table flag\n")
- os.Exit(1)
- }
- csvOpts := csvimport.DefaultImportOptions()
- csvOpts.TableName = *exportTable
- csvOpts.IgnoreErrors = *ignoreErrors
- csvOpts.CreateTable = *createTable
- result, err := csvimport.ImportCSV(strings.NewReader(string(data)), schema, table, csvOpts)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Import failed: %v\n", err)
- if len(result.Errors) > 0 {
- fmt.Fprintf(os.Stderr, "Errors:\n")
- for _, e := range result.Errors {
- fmt.Fprintf(os.Stderr, " - %s\n", e)
- }
- }
- os.Exit(1)
- }
- fmt.Printf("CSV import completed successfully\n")
- fmt.Printf(" Rows imported: %d\n", result.RowsImported)
- if result.RowsSkipped > 0 {
- fmt.Printf(" Rows skipped: %d\n", result.RowsSkipped)
- }
- if result.TableCreated {
- fmt.Printf(" Table created: %s\n", *exportTable)
- }
- if len(result.Errors) > 0 {
- fmt.Printf(" Warnings/Errors: %d\n", len(result.Errors))
- for _, e := range result.Errors {
- fmt.Printf(" - %s\n", e)
- }
- }
- case "sqlite":
- // Binary SQLite .db import
- opts := sqliteimport.DefaultImportOptions()
- opts.IgnoreErrors = *ignoreErrors
- result, err := sqliteimport.ImportSQLiteFile(*importFile, exec, opts)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Import failed: %v\n", err)
- if len(result.Errors) > 0 {
- fmt.Fprintf(os.Stderr, "Errors:\n")
- for _, e := range result.Errors {
- fmt.Fprintf(os.Stderr, " - %s\n", e)
- }
- }
- os.Exit(1)
- }
- fmt.Printf("SQLite import completed successfully\n")
- if len(result.TablesCreated) > 0 {
- fmt.Printf(" Tables created: %s\n", strings.Join(result.TablesCreated, ", "))
- }
- if len(result.TablesImported) > 0 {
- fmt.Printf(" Tables imported: %s\n", strings.Join(result.TablesImported, ", "))
- }
- fmt.Printf(" Rows inserted: %d\n", result.RowsInserted)
- if result.IndexesCreated > 0 {
- fmt.Printf(" Indexes created: %d\n", result.IndexesCreated)
- }
- if len(result.Errors) > 0 {
- fmt.Printf(" Warnings/Errors: %d\n", len(result.Errors))
- for _, e := range result.Errors {
- fmt.Printf(" - %s\n", e)
- }
- }
- default: // sql
- // Configure import options
- opts := sqlimport.ImportOptions{
- IgnoreErrors: *ignoreErrors,
- }
- // Import SQL
- result, err := sqlimport.ImportSQL(exec, string(data), opts)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Import failed: %v\n", err)
- if len(result.Errors) > 0 {
- fmt.Fprintf(os.Stderr, "Errors:\n")
- for _, e := range result.Errors {
- fmt.Fprintf(os.Stderr, " - %s\n", e)
- }
- }
- os.Exit(1)
- }
- fmt.Printf("Import completed successfully\n")
- fmt.Printf(" Statements executed: %d\n", result.StatementsExecuted)
- if len(result.TablesCreated) > 0 {
- fmt.Printf(" Tables created: %s\n", strings.Join(result.TablesCreated, ", "))
- }
- if len(result.TablesDropped) > 0 {
- fmt.Printf(" Tables dropped: %s\n", strings.Join(result.TablesDropped, ", "))
- }
- fmt.Printf(" Rows inserted: %d\n", result.RowsInserted)
- if len(result.Errors) > 0 {
- fmt.Printf(" Warnings/Errors: %d\n", len(result.Errors))
- for _, e := range result.Errors {
- fmt.Printf(" - %s\n", e)
- }
- }
- }
- }
- func detectFileFormat(filename string) string {
- lower := strings.ToLower(filename)
- if strings.HasSuffix(lower, ".csv") {
- return "csv"
- }
- if strings.HasSuffix(lower, ".db") || strings.HasSuffix(lower, ".sqlite") || strings.HasSuffix(lower, ".sqlite3") {
- return "sqlite"
- }
- return "sql"
- }
- func runServers() {
- pool, err := storage.NewKVPool(*kvAddr, *poolSize, *timeout)
- if err != nil {
- fmt.Fprintf(os.Stderr, "Failed to connect to PizzaKV at %s: %v\n", *kvAddr, err)
- os.Exit(1)
- }
- defer pool.Close()
- dbManagerConfig := &storage.DatabaseManagerConfig{
- DefaultDatabase: *database,
- AutoCreate: true,
- }
- dbManager := storage.NewDatabaseManager(pool, dbManagerConfig)
- stop := make(chan os.Signal, 1)
- signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
- var httpSrv *httpserver.Server
- var pprofSrv *http.Server
- var pgSrv *pgserver.Server
- if *httpEnable {
- config := httpserver.DefaultConfig()
- config.Host = *httpHost
- config.Port = *httpPort
- config.EnableCORS = *httpCORS
- config.EnableAuth = *httpAuth
- config.EnableCompression = *httpCompression
- config.EnableLogging = !*quiet
- if *apiKeys != "" {
- config.APIKeys = strings.Split(*apiKeys, ",")
- }
- httpSrv = httpserver.NewWithDatabaseManager(config, dbManager)
- if startPprofServerHook != nil {
- pprofSrv = startPprofServerHook()
- }
- go func() {
- if err := httpSrv.Start(); err != nil && err != http.ErrServerClosed {
- fmt.Fprintf(os.Stderr, "HTTP server error: %v\n", err)
- os.Exit(1)
- }
- }()
- fmt.Printf("HTTP http://%s:%d\n", *httpHost, *httpPort)
- }
- if *pgEnable {
- config := pgserver.DefaultConfig()
- config.Host = *pgHost
- config.Port = *pgPort
- config.DefaultDatabase = *database
- config.Quiet = *quiet
- pgSrv = pgserver.New(config, dbManager)
- go func() {
- if err := pgSrv.Start(); err != nil {
- fmt.Fprintf(os.Stderr, "PostgreSQL server error: %v\n", err)
- os.Exit(1)
- }
- }()
- fmt.Printf("PG postgresql://%s:%d/%s\n", *pgHost, *pgPort, *database)
- }
- fmt.Printf("KV %s\n", *kvAddr)
- fmt.Printf("DB %s\n", *database)
- fmt.Println("Press Ctrl+C to stop")
- <-stop
- fmt.Println("\nShutting down...")
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer cancel()
- if httpSrv != nil {
- if err := httpSrv.Shutdown(ctx); err != nil {
- fmt.Fprintf(os.Stderr, "HTTP shutdown error: %v\n", err)
- }
- }
- if pprofSrv != nil {
- pprofSrv.Shutdown(ctx)
- }
- if pgSrv != nil {
- if err := pgSrv.Shutdown(ctx); err != nil {
- fmt.Fprintf(os.Stderr, "PG shutdown error: %v\n", err)
- }
- }
- }
- // launchPizzaKV starts a dedicated PizzaKV instance for this pizzasql process.
- func launchPizzaKV() error {
- if _, err := os.Stat(".db"); err == nil {
- if live := pizzaruntime.LiveInstances(); len(live) > 0 {
- inst := live[0]
- kvAddr := "<addr>"
- if inst.PizzaKV != nil {
- kvAddr = inst.PizzaKV.Addr
- }
- return fmt.Errorf(".db file already exists and another pizzasql instance is running (PID %d)\n"+
- " To connect to its pizzakv: pizzasql -kvaddr=%s\n"+
- " To start fresh (removes data): rm .db && pizzasql -kv\n"+
- " To run a separate instance: cd /other/dir && pizzasql -kv",
- inst.PizzaSQL.PID, kvAddr)
- }
- }
- kvManager = kvmanager.NewManager()
- fmt.Println("Starting PizzaKV...")
- info, err := kvManager.Start(*kvFlags)
- if err != nil {
- return err
- }
- fmt.Printf("PizzaKV started on %s (PID: %d)\n", info.Addr, info.PID)
- fmt.Printf("Runtime: %s\n", pizzaruntime.File)
- fmt.Println("PizzaKV is ready!")
- *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")
- }
- }
- }
|