From 87011e4294286ae0bb4fde3ccac9cbe7ef85c2eb Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Tue, 14 Feb 2023 09:23:51 -0600 Subject: [PATCH] CLI: make it more like psql (#2235) * Refactor CLI to mimic psql's meta-commands This PR adds support for meta-commands (also known as "backslash commands") like those in psql, Postgres's CLI. Only a few meta-commands are currently implemented, but this was meant to demonstrate how we could use something like `\i file.csv` to insert local files into SQL statements. * Meta-commands: \file and \include The initial implementation used `\i` as a streaming file handle. This commit changes that to `\file`, and then implements `\i` (or `\include`) as handling multiple sql commands. * Add meta-command "help" (\?) This is basically a copy of the psql help output, but includes only those options we currently support. * Add support for \o [file], and \timing The \o meta-command writes query output to a file. The \timing meta-command turns on/off the timing display sent to stdout. * Add meta-commands: \l (show databases) and \dt (show tables) * Add meta-command: \watch [period] * Update meta-command \connect to take database name instead of ID * Add support for \echo, \qecho, and \warn This commit contains an known issue in that the `-n` option will exclude the line feed, but if the output is the terminal, the readline package clobbers any content on the current line (i.e. anything without a line feed). That will need to be addressed at some point. * Add support for \w [FILE] (write query buffer to file) * Add SchemaAPI no-op implementation * Refactor query handler to align with /sql and /databases endpoints We want to standardize on: /sql /databases/{databaseID}/sql * Add CLI support for expanded, border, tuples_only (and pset) * Add help text for \pset and \t --- api.go | 40 + cli/buffer.go | 91 +++ cli/cli.go | 627 ++++++++------- cli/cli_test.go | 34 +- cli/fbcloud/client.go | 28 +- cli/meta.go | 926 +++++++++++++++++++++++ cli/parts.go | 130 ++++ cli/queryer.go | 63 +- cli/splitter.go | 119 +++ cli/splitter_test.go | 146 ++++ cli/workingdir.go | 21 + cli/writer.go | 191 +++++ cli/writer_test.go | 164 ++++ cmd/cli.go | 4 +- dax/controller/schemar/boltdb/schemar.go | 2 +- dax/queryer/client/client.go | 71 +- dax/queryer/http/handler.go | 118 ++- dax/queryer/queryer.go | 41 +- dax/test/dax/dax_test.go | 6 +- sql3/planner/opalterdatabase.go | 2 +- 20 files changed, 2298 insertions(+), 526 deletions(-) create mode 100644 cli/buffer.go create mode 100644 cli/meta.go create mode 100644 cli/parts.go create mode 100644 cli/splitter.go create mode 100644 cli/splitter_test.go create mode 100644 cli/workingdir.go create mode 100644 cli/writer.go create mode 100644 cli/writer_test.go diff --git a/api.go b/api.go index 2b7bb9a74..ccc984558 100644 --- a/api.go +++ b/api.go @@ -3335,6 +3335,46 @@ type SchemaAPI interface { DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error } +// Ensure type implements interface. +var _ SchemaAPI = (*NopSchemaAPI)(nil) + +// NopSchemaAPI is a no-op implementation of the SchemaAPI. +type NopSchemaAPI struct{} + +func (n *NopSchemaAPI) ClusterName() string { + return "" +} +func (n *NopSchemaAPI) CreateDatabase(context.Context, *dax.Database) error { return nil } +func (n *NopSchemaAPI) DropDatabase(context.Context, dax.DatabaseID) error { return nil } +func (n *NopSchemaAPI) DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) { + return nil, nil +} +func (n *NopSchemaAPI) DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) { + return nil, nil +} +func (n *NopSchemaAPI) SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error { + return nil +} +func (n *NopSchemaAPI) Databases(context.Context, ...dax.DatabaseID) ([]*dax.Database, error) { + return nil, nil +} +func (n *NopSchemaAPI) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { + return nil, nil +} +func (n *NopSchemaAPI) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) { + return nil, nil +} +func (n *NopSchemaAPI) Tables(ctx context.Context) ([]*dax.Table, error) { return nil, nil } + +func (n *NopSchemaAPI) CreateTable(ctx context.Context, tbl *dax.Table) error { return nil } +func (n *NopSchemaAPI) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error { + return nil +} +func (n *NopSchemaAPI) DeleteTable(ctx context.Context, tname dax.TableName) error { return nil } +func (n *NopSchemaAPI) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error { + return nil +} + type ClusterNode struct { ID string State string diff --git a/cli/buffer.go b/cli/buffer.go new file mode 100644 index 000000000..e1ae69325 --- /dev/null +++ b/cli/buffer.go @@ -0,0 +1,91 @@ +package cli + +import ( + "io" + "strings" + + "github.com/featurebasedb/featurebase/v3/errors" +) + +// buffer is a query buffer for SQL statements. Note that this is not a query +// buffer as you would find on a database server (buffering query results). +// Rather, this buffers the working SQL statement. The buffer has two +// components: the buffer of query parts making up the working, incomplete SQL +// statement, and the last completed SQL statement submitted to the Queryer. +type buffer struct { + parts []queryPart + lastQuery query + + hasBatchFile bool +} + +func newBuffer() *buffer { + return &buffer{} +} + +// addPart adds the given queryPart to the buffer. If the part is of type +// `partTerminator` (which is generally singified in the CLI by a ";"), the +// buffer will finalize the query and return it. In all other cases, the +// returned query is nil. +func (b *buffer) addPart(part queryPart) (query, error) { + // Check for part type compatibility. For example, multiple batchFile parts + // are not allowed in the same query. + switch part.(type) { + case *partBatchFile: + if b.hasBatchFile { + return nil, errors.Errorf("multiple batch files in one query is not supported") + } + b.hasBatchFile = true + case *partTerminator: + return b.finalize(), nil + } + + b.parts = append(b.parts, part) + return nil, nil +} + +// finalize copies the contents (queryParts) of buffer to lastQuery and then +// resets the buffer. It returns the query that was finalized. +func (b *buffer) finalize() query { + q := make(query, len(b.parts)) + copy(q, b.parts) + b.lastQuery = q + b.reset() + return q + +} + +// print returns the contents of the buffer as a string. This is generally used +// to visually inspect the state of the buffer (for example, when a user issues +// a `\p` meta-command in the CLI). +func (b *buffer) print() string { + if len(b.parts) > 0 { + return query(b.parts).String() + } else if b.lastQuery != nil { + return b.lastQuery.String() + ";" + } + return "Query buffer is empty." +} + +// reset clears the buffer. It returns a message which may optionally be used to +// display to a user. +func (b *buffer) reset() string { + b.parts = b.parts[:0] + b.hasBatchFile = false + return "Query buffer reset (cleared)." +} + +func (b *buffer) Reader() io.Reader { + if len(b.parts) > 0 { + return query(b.parts).Reader() + } else if b.lastQuery != nil { + r := b.lastQuery.Reader() + // TODO(tlt): terminating the query here results in a line feed just + // before the semi-colon (for example, when you print out the query + // buffer using `\w [FILE]`). The removal and re-introduction of line + // feeds is kind of a mess. + term := strings.NewReader(";") + return io.MultiReader(r, term) + } + return strings.NewReader("") +} diff --git a/cli/cli.go b/cli/cli.go index 1f06e1320..946965c71 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -1,3 +1,4 @@ +// Package cli contains a FeatureBase command line interface. package cli import ( @@ -14,8 +15,6 @@ import ( featurebase "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/cli/fbcloud" "github.com/featurebasedb/featurebase/v3/logger" - "github.com/jedib0t/go-pretty/table" - "github.com/jedib0t/go-pretty/text" "github.com/pkg/errors" ) @@ -24,7 +23,6 @@ const ( promptBegin string = "fbsql> " promptMid string = " -> " terminationChar string = ";" - exitCommand string = "exit" nullValue string = "NULL" ) @@ -36,7 +34,7 @@ var ( var ( splash string = fmt.Sprintf(`FeatureBase CLI (%s) -Type "exit" to quit. +Type "\q" to quit. `, featurebase.Version) ) @@ -51,17 +49,28 @@ type CLICommand struct { Email string `json:"email"` Password string `json:"password"` - // commands holds the list of sql commands to be executed. - commands []string + splitter *splitter + buffer *buffer + workingDir *workingDir OrganizationID string `json:"org-id"` - DatabaseID string `json:"db-id"` + Database string `json:"db"` + databaseID string + databaseName string Queryer Queryer `json:"-"` Stdin io.ReadCloser `json:"-"` Stdout io.Writer `json:"-"` Stderr io.Writer `json:"-"` + + // output is where actual results are written. This might point to stdout, + // or to a file, based on the current configuration. + output io.Writer `json:"-"` + writeOptions *writeOptions + + // quit gets closed when Run should stop listening for input. + quit chan struct{} } func NewCLICommand(logdest logger.Logger) *CLICommand { @@ -70,14 +79,202 @@ func NewCLICommand(logdest logger.Logger) *CLICommand { HistoryPath: "", OrganizationID: "", - DatabaseID: "", + Database: "", + + splitter: newSplitter(), + buffer: newBuffer(), + workingDir: newWorkingDir(), Stdin: Stdin, Stdout: Stdout, Stderr: Stderr, + + output: Stdout, + writeOptions: defaultWriteOptions(), + + quit: make(chan struct{}), } } +// Run is the main entry-point to the CLI. Currently it handles the interaction +// with a user, as opposed to calling `featurebase cli` in a script. +func (cmd *CLICommand) Run(ctx context.Context) error { + // Print the splash message. + cmd.Printf(splash) + cmd.setupHistory() + if err := cmd.setupClient(); err != nil { + return errors.Wrap(err, "setting up client") + } + cmd.printConnInfo() + if err := cmd.connectToDatabase(cmd.Database); err != nil { + cmd.Errorf(errors.Wrap(err, "connecting to database").Error() + "\n") + } + + rl, err := readline.NewEx(&readline.Config{ + Prompt: promptBegin, + HistoryFile: cmd.HistoryPath, + HistoryLimit: 100000, + DisableAutoSaveHistory: true, + + Stdin: cmd.Stdin, + Stdout: cmd.Stdout, + Stderr: cmd.Stderr, + }) + if err != nil { + return errors.Wrap(err, "getting readline") + } + defer rl.Close() + + // inMidCommand indicates whether a partial command has been received and + // we're still waiting for a termination character. + var inMidCommand bool + + for { + if inMidCommand { + rl.SetPrompt(promptMid) + } else { + rl.SetPrompt(promptBegin) + } + + // Read user provided input. + line, err := rl.Readline() + if err != nil { + return errors.Wrap(err, "reading line") + } + + // We append a line feed at the end of each line because at this point + // we have effectively stripped any intentional line feeds (since we are + // reading a line at a time), and we don't want to do that. An example + // of an intentional line feed is in a BULK INSERT CSV STREAM like this + // example: + // + // bulk replace + // into foo (_id, age) + // map (0 id, 1 int) + // from + // x'3,33 + // 4,44 + // 5,55' + // with + // format 'CSV' + // input 'STREAM'; + // + // We want to preserve the line feeds that are contained in the x'' + // block; those are intentional as they demarc records within the csv. + qps, mcs, err := cmd.splitter.split(line + "\n") + if err != nil { + cmd.Errorf("error splitting line: %s\n", err) + continue + } + + // Save line in the history. + if err := rl.SaveHistory(line); err != nil { + cmd.Errorf("Couldn't save history: %v\n", err) + } + + // This is wrapped in an anonymous function so we can capture any + // errors, ignore the rest of the line, and return back to a prompt. + if err := func() error { + for i := range qps { + if qry, err := cmd.buffer.addPart(qps[i]); err != nil { + return errors.Wrap(err, "adding part to buffer") + } else if qry != nil { + if err := cmd.executeAndWriteQuery(qry); err != nil { + return errors.Wrap(err, "executing query") + } + // In addition to saving each line in the history, we also + // save each successful query. + if err := rl.SaveHistory(qry.String() + ";"); err != nil { + cmd.Errorf("Couldn't save query in history: %v\n", err) + } + inMidCommand = false + } else { + inMidCommand = true + } + } + return nil + }(); err != nil { + cmd.Errorf(err.Error() + "\n") + inMidCommand = false + continue + } + + // This is wrapped in an anonymous function so we can capture any + // errors, ignore the rest of the line, and return back to a prompt. + if err := func() error { + for i := range mcs { + action, err := mcs[i].execute(cmd) + if err != nil { + return errors.Wrap(err, "executing meta command") + } + switch action { + case actionQuit: + close(cmd.quit) + return nil + case actionReset: + inMidCommand = false + } + } + return nil + }(); err != nil { + cmd.Errorf(err.Error() + "\n") + inMidCommand = false + continue + } + + select { + case <-cmd.quit: + if err := cmd.close(); err != nil { + cmd.Errorf("closing: %s\n", err) + } + return nil + default: + //pass + } + } +} + +// close is called upon quitting. It should close any remaining open file +// handles used by the CLICommand. +func (cmd *CLICommand) close() error { + return cmd.closeOutput() +} + +func (cmd *CLICommand) executeAndWriteQuery(qry query) error { + queryResponse, err := cmd.executeQuery(qry) + if err != nil { + return errors.Wrap(err, "making query") + } + + if err := writeTable(queryResponse, cmd.writeOptions, cmd.output, cmd.Stdout, cmd.Stderr); err != nil { + return errors.Wrap(err, "writing out response") + } + + return nil +} + +func (cmd *CLICommand) executeQuery(qry query) (*featurebase.WireQueryResponse, error) { + return cmd.Queryer.Query(cmd.OrganizationID, cmd.databaseID, qry.Reader()) +} + +// Printf is a helper method which sends the given payload to stdout. +func (cmd *CLICommand) Printf(format string, a ...any) { + out := fmt.Sprintf(format, a...) + cmd.Stdout.Write([]byte(out)) +} + +// Outputf is a helper method which sends the given payload to output. +func (cmd *CLICommand) Outputf(format string, a ...any) { + out := fmt.Sprintf(format, a...) + cmd.output.Write([]byte(out)) +} + +// Errorf is a helper method which sends the given payload to stderr. +func (cmd *CLICommand) Errorf(format string, a ...any) { + out := fmt.Sprintf(format, a...) + cmd.Stderr.Write([]byte(out)) +} + func (cmd *CLICommand) setupHistory() { // If HistoryPath has already been configured (i.e. with a command flag), // don't bother setting up the default in the home directory. @@ -87,12 +284,12 @@ func (cmd *CLICommand) setupHistory() { historyPath := "" if home, err := os.UserHomeDir(); err != nil { - cmd.Printf("Error getting home directory, command history persistence will be disabled: %v\n", err) + cmd.Errorf("Error getting home directory, command history persistence will be disabled: %v\n", err) } else { historyDir := filepath.Join(home, ".featurebase") err := os.MkdirAll(historyDir, 0o750) if err != nil { - cmd.Printf("Creating directory for history: %v\n", err) + cmd.Errorf("Creating directory for history: %v\n", err) } else { historyPath = filepath.Join(historyDir, "cli_history") } @@ -100,13 +297,55 @@ func (cmd *CLICommand) setupHistory() { cmd.HistoryPath = historyPath } -// printQualifiers displays the currently set OrganizationID and DatabaseID. -func (cmd *CLICommand) printQualifiers() { - cmd.Printf(" Host: %s\n Org: %s\n DB: %s\n", - hostPort(cmd.Host, cmd.Port), - cmd.OrganizationID, - cmd.DatabaseID, - ) +// printConnInfo displays the currently set host. +// TODO(tlt): extend this to be the output of the /conninfo meta-command. +func (cmd *CLICommand) printConnInfo() { + cmd.Printf("Host: %s\n", hostPort(cmd.Host, cmd.Port)) +} + +func (cmd *CLICommand) connectToDatabase(dbName string) error { + if dbName == "" { + cmd.databaseID = "" + cmd.databaseName = "" + cmd.Printf(cmd.connectionMessage()) + return nil + } + + // Look up dbID based on dbName. + qry := []queryPart{ + newPartRaw("SHOW DATABASES"), + } + + qr, err := cmd.executeQuery(qry) + if err != nil { + return errors.Wrap(err, "executing query") + } + + for _, db := range qr.Data { + // 0: _id + // 1: name + if db[1] == dbName { + cmd.databaseName = dbName + cmd.databaseID = db[0].(string) + cmd.Printf(cmd.connectionMessage()) + return nil + } + } + return errors.Errorf("invalid database: %s", dbName) +} + +func (cmd *CLICommand) orgMessage() string { + if cmd.OrganizationID == "" { + return "You have not set an organization.\n" + } + return fmt.Sprintf("You have set organization \"%s\".\n", cmd.OrganizationID) +} + +func (cmd *CLICommand) connectionMessage() string { + if cmd.databaseName == "" { + return "You are not connected to a database.\n" + } + return fmt.Sprintf("You are now connected to database \"%s\" (%s) as user \"???\".\n", cmd.databaseName, cmd.databaseID) } func (cmd *CLICommand) setupClient() error { @@ -117,7 +356,7 @@ func (cmd *CLICommand) setupClient() error { } if strings.TrimSpace(cmd.Host) == "" { - return errors.Errorf("no host provided") + return errors.Errorf("no host provided\n") } if !strings.HasPrefix(cmd.Host, "http") { @@ -130,20 +369,33 @@ func (cmd *CLICommand) setupClient() error { } switch typ { - case featurebaseTypeStandard: - cmd.Printf("Detected standard deployment\n") + case featurebaseTypeOnPremClassic: + cmd.Printf("Detected on-prem, classic deployment.\n") cmd.Queryer = &standardQueryer{ Host: cmd.Host, Port: cmd.Port, } - case featurebaseTypeDAX: - cmd.Printf("Detected dax deployment\n") - cmd.Queryer = &daxQueryer{ + case featurebaseTypeOnPremServerless: + cmd.Printf("Detected on-prem, serverless deployment.\n") + cmd.Queryer = &serverlessQueryer{ Host: cmd.Host, Port: cmd.Port, } case featurebaseTypeCloud: - cmd.Printf("Detected cloud deployment\n") + cmd.Printf("Detected cloud deployment.\n") + cmd.Queryer = &fbcloud.Queryer{ + Host: hostPort(cmd.Host, cmd.Port), + + ClientID: cmd.ClientID, + Region: cmd.Region, + Email: cmd.Email, + Password: cmd.Password, + } + case featurebaseTypeUnknown: + cmd.Printf("Could not detect deployment\n") + // cmd.Queryer = &nopQueryer{} + // Instead of using a no-op queryer when the type can't be detected, we + // default to using a cloud queryer. cmd.Queryer = &fbcloud.Queryer{ Host: hostPort(cmd.Host, cmd.Port), @@ -161,9 +413,10 @@ func (cmd *CLICommand) setupClient() error { type featurebaseType string const ( - featurebaseTypeStandard featurebaseType = "standard" - featurebaseTypeDAX featurebaseType = "dax" - featurebaseTypeCloud featurebaseType = "cloud" + featurebaseTypeUnknown featurebaseType = "unknown" // unknown + featurebaseTypeOnPremClassic featurebaseType = "on-prem-standard" // on-prem, classic + featurebaseTypeOnPremServerless featurebaseType = "on-prem-serverless" // on-prem, serverless + featurebaseTypeCloud featurebaseType = "cloud" // cloud, (both classic and serverless)? ) func hostPort(host, port string) string { @@ -186,51 +439,55 @@ func (cmd *CLICommand) detectFBType() (featurebaseType, error) { // process is running there which can support the cli requests. trials := []trial{} + var clientTimeout time.Duration if cmd.Port != "" { + clientTimeout = 100 * time.Millisecond trials = append(trials, - // dax + // on-prem, serverless trial{ port: cmd.Port, health: "/queryer/health", - typ: featurebaseTypeDAX, + typ: featurebaseTypeOnPremServerless, }, - // standard + // on-prem, classic trial{ port: cmd.Port, health: "/status", - typ: featurebaseTypeStandard, + typ: featurebaseTypeOnPremClassic, }, ) } else if strings.HasPrefix(cmd.Host, "https") { // https suggesting we might be connecting to a cloud host + clientTimeout = 1 * time.Second trials = append(trials, // cloud trial{ port: "", - health: "health", + health: "/health", typ: featurebaseTypeCloud, }, ) } else { // Try default ports just in case. + clientTimeout = 100 * time.Millisecond trials = append(trials, - // dax + // on-prem, serverless trial{ port: "8080", health: "/queryer/health", - typ: featurebaseTypeDAX, + typ: featurebaseTypeOnPremServerless, }, - // standard + // on-prem, classic trial{ port: "10101", health: "/status", - typ: featurebaseTypeStandard, + typ: featurebaseTypeOnPremClassic, }, ) } client := http.Client{ - Timeout: 100 * time.Millisecond, + Timeout: clientTimeout, } for _, trial := range trials { url := hostPort(cmd.Host, trial.port) + trial.health @@ -242,301 +499,17 @@ func (cmd *CLICommand) detectFBType() (featurebaseType, error) { } } - return featurebaseTypeCloud, nil + return featurebaseTypeUnknown, nil } -func (cmd *CLICommand) Run(ctx context.Context) error { - // Print the splash message. - cmd.Printf(splash) - cmd.setupHistory() - if err := cmd.setupClient(); err != nil { - return errors.Wrap(err, "setting up client") +func (cmd *CLICommand) closeOutput() error { + if cmd.output == nil { + return nil } - cmd.printQualifiers() - rl, err := readline.NewEx(&readline.Config{ - Prompt: promptBegin, - HistoryFile: cmd.HistoryPath, - HistoryLimit: 100000, - DisableAutoSaveHistory: true, - - Stdin: cmd.Stdin, - Stdout: cmd.Stdout, - Stderr: cmd.Stderr, - }) - if err != nil { - return errors.Wrap(err, "getting readline") - } - defer rl.Close() - - // partialCommand holds all input prior to receiving a termination - // character. - var partialCommand string - - // inMidCommand indicates whether a partial command has been received and - // we're still waiting for a termination character. - var inMidCommand bool - - for { - if inMidCommand { - rl.SetPrompt(promptMid) - } else { - rl.SetPrompt(promptBegin) - // Add some white space before each new prompt. - cmd.Printf("\n") - } - - // Read user provided input. - line, err := rl.Readline() - if err != nil { - return errors.Wrap(err, "reading line") - } - - if !inMidCommand { - // Handle the exit command. - if line == exitCommand || line == exitCommand+terminationChar { - break - } - } - - // We append a line feed at the end of each line because at this point - // we have effectively stripped any intentional line feeds (since we are - // reading a line at a time), and we don't want to do that. An example - // of an intentional line feed is in a BULK INSERT CSV STREAM like this - // example: - // - // bulk replace - // into foo (_id, age) - // map (0 id, 1 int) - // from - // x'3,33 - // 4,44 - // 5,55' - // with - // format 'CSV' - // input 'STREAM'; - // - // We want to preserve the line feeds that are contained in the x'' - // block; those are intentional as they demarc records within the csv. - line += "\n" - - // Look for a termination character; - parts := strings.Split(line, terminationChar) - - // Length of 1 means a termination character was not received. - if len(parts) == 1 { - if parts[0] != "" { - partialCommand = appendCommand(partialCommand, parts[0]) - inMidCommand = true - } - continue - } - - for i, part := range parts { - partIsFinal := i == len(parts)-1 - partIsBlank := strings.TrimSpace(part) == "" - - if partIsBlank && partIsFinal { - continue - } - - if partIsBlank && !partIsFinal { - if inMidCommand { - cmd.commands = append(cmd.commands, strings.TrimSpace(partialCommand)) - partialCommand = "" - inMidCommand = false - } - continue - } - - if !partIsBlank && partIsFinal { - partialCommand = part - inMidCommand = true - continue - } - - if !partIsBlank && !partIsFinal { - partialCommand = appendCommand(partialCommand, part) - cmd.commands = append(cmd.commands, strings.TrimSpace(partialCommand)) - partialCommand = "" - inMidCommand = false - } - } - - err = rl.SaveHistory(strings.Join(cmd.commands, "; ") + ";") - if err != nil { - cmd.Printf("Couldn't save history: %v\n", err) - } - - if err := cmd.executeCommands(ctx); err != nil { - return errors.Wrap(err, "executing commands") - } + if closer, ok := cmd.output.(io.Closer); ok { + return closer.Close() } return nil } - -func appendCommand(orig string, part string) string { - if orig == "" { - return part - } else { - return orig + part - } -} - -func (cmd *CLICommand) executeCommands(ctx context.Context) error { - // Clear out the buffered commands on any exit from this method. - defer func() { - cmd.commands = nil - }() - - for _, sql := range cmd.commands { - // Handle non-sql commands (for example, SET commands). - if handled, err := cmd.handleIfNonSQLCommand(ctx, sql); err != nil { - return errors.Wrapf(err, "handling non-SQL command: %s", sql) - } else if handled { - continue - } - - sqlResponse, err := cmd.Queryer.Query(cmd.OrganizationID, cmd.DatabaseID, sql) - if err != nil { - cmd.Printf("making query: %v\n", err) - continue - } - err = writeOut(sqlResponse, cmd.Stdout, cmd.Stderr) - if err != nil { - return errors.Wrap(err, "writing out response") - } - } - - return nil -} - -// Printf is a helper method which sends the given payload to stdout. -func (cmd *CLICommand) Printf(format string, a ...any) { - out := fmt.Sprintf(format, a...) - cmd.Stdout.Write([]byte(out)) -} - -// handleIfNonSQLCommand will handle special case command like "SET ..." and -// "USE ...". If the sql command matches one of these conditions and is handled, -// the bool returned will be true; -func (cmd *CLICommand) handleIfNonSQLCommand(ctx context.Context, sql string) (bool, error) { - var handled bool - - // Get the first token from the SQL: - parts := strings.Split(sql, " ") - if len(parts) < 1 { - return handled, nil - } - - token := strings.ToUpper(parts[0]) - - // Supported: - // SET ORG acme - // SET DB db1 - // USE db1 - switch token { - case "SET": - handled = true - switch len(parts) { - case 1: - // This will fall through and just print the qualifiers. - case 3: - switch strings.ToUpper(parts[1]) { - case "HOST": - cmd.Host = parts[2] - case "ORG": - cmd.OrganizationID = parts[2] - case "DB": - cmd.DatabaseID = parts[2] - } - default: - return handled, errors.Errorf("SET command takes a name and a value (SET DB db1)") - } - - case "USE": - handled = true - if len(parts) != 2 { - return handled, errors.Errorf("USE command takes a single value (USE db1)") - } - cmd.DatabaseID = parts[1] - default: - return handled, nil - } - - cmd.printQualifiers() - return handled, nil -} - -func writeWarnings(r *featurebase.WireQueryResponse, w io.Writer) error { - if len(r.Warnings) > 0 { - if _, err := w.Write([]byte("\n")); err != nil { - return errors.Wrapf(err, "writing warning: %s", r.Error) - } - for _, warning := range r.Warnings { - if _, err := w.Write([]byte("Warning: " + warning + "\n")); err != nil { - return errors.Wrapf(err, "writing warning: %s", r.Error) - } - } - } - return nil -} - -func writeOut(r *featurebase.WireQueryResponse, wOut io.Writer, wErr io.Writer) error { - if r == nil { - return errors.New("attempt to write out nil response") - } - if r.Error != "" { - if _, err := wErr.Write([]byte("Error: " + r.Error + "\n")); err != nil { - return errors.Wrapf(err, "writing error: %s", r.Error) - } - return writeWarnings(r, wOut) - } - - t := table.NewWriter() - t.SetOutputMirror(wOut) - - // Don't uppercase the header values. - t.Style().Format.Header = text.FormatDefault - - t.AppendHeader(schemaToRow(r.Schema)) - for _, row := range r.Data { - // If the value is nil, replace it with a null string; go-pretty doesn't - // expect nil pointers in the data values. - for i := range row { - if row[i] == nil { - row[i] = nullValue - } - } - t.AppendRow(table.Row(row)) - } - t.Render() - - err := writeWarnings(r, wOut) - if err != nil { - return err - } - lifeAffirmingMessage := "" - if r.ExecutionTime < 1000000 { - lifeAffirmingMessage = " (You're welcome! 🚀)" - } - - if r.ExecutionTime > 5000000 { - lifeAffirmingMessage = " (Sorry! That took longer than expected 😭)" - } - - if _, err := wOut.Write([]byte(fmt.Sprintf("\nExecution time: %dμs%s\n", r.ExecutionTime, lifeAffirmingMessage))); err != nil { - return errors.Wrapf(err, "writing execution time: %s", r.Error) - } - - return nil -} - -func schemaToRow(schema featurebase.WireQuerySchema) []interface{} { - ret := make([]interface{}, len(schema.Fields)) - for i, field := range schema.Fields { - ret[i] = field.Name - } - return ret -} diff --git a/cli/cli_test.go b/cli/cli_test.go index ceac7a059..9a5ee576f 100644 --- a/cli/cli_test.go +++ b/cli/cli_test.go @@ -33,38 +33,32 @@ func TestCLI(t *testing.T) { none := []string{} // One statement, one line. - capture.Assert("one;", []string{`one`}) + capture.Assert("one;", []string{"one\n"}) // One statement, multiple lines. capture.Assert("one", none) capture.Assert(" two ", none) - capture.Assert("three;", []string{`one - two -three`}) + capture.Assert("three;", []string{"one\ntwo\nthree\n"}) // Multiple statements, one line. - capture.Assert("foo; bar;", []string{`foo`, `bar`}) + capture.Assert("foo; bar;", []string{"foo\n", "bar\n"}) // Multiple statements, multiple lines. capture.Assert("a1", none) - capture.Assert("a2; b1", []string{`a1 -a2`}) - capture.Assert("b2;", []string{`b1 -b2`}) + capture.Assert("a2; b1", []string{"a1\na2\n"}) + capture.Assert("b2;", []string{"b1\nb2\n"}) // Blank lines. capture.Assert("one", none) capture.Assert("", none) - capture.Assert("three;", []string{`one - -three`}) + capture.Assert("three;", []string{"one\nthree\n"}) // Just a semi-colon. - capture.Assert(";", none) + capture.Assert(";", []string{""}) // Multi-line with just a semi-colon. capture.Assert("one", none) - capture.Assert(";", []string{`one`}) + capture.Assert(";", []string{"one\n"}) // Ensure a clean exit with no errors. assert.NoError(t, capture.Exit()) @@ -111,7 +105,7 @@ func newCapture(t *testing.T) *capture { } func (c *capture) Exit() error { - c.sendLine("exit") + c.sendLine(`\q`) c.mu.RLock() defer c.mu.RUnlock() return c.err @@ -183,9 +177,15 @@ func (c *capture) Write(b []byte) (n int, err error) { // Query is called by the CLI command once a full SQL statement is received // (signified by the terminator: `;`). -func (c *capture) Query(org, db, sql string) (*featurebase.WireQueryResponse, error) { +func (c *capture) Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) { + tmpBuf := new(strings.Builder) + _, err := io.Copy(tmpBuf, sql) + if err != nil { + return nil, err + } + c.mu.Lock() - c.sqls = append(c.sqls, sql) + c.sqls = append(c.sqls, tmpBuf.String()) c.mu.Unlock() select { diff --git a/cli/fbcloud/client.go b/cli/fbcloud/client.go index 53c2d5035..b65b154ae 100644 --- a/cli/fbcloud/client.go +++ b/cli/fbcloud/client.go @@ -1,7 +1,6 @@ package fbcloud import ( - "bytes" "encoding/json" "fmt" "io" @@ -40,37 +39,26 @@ func (cq *Queryer) tokenRefresh() error { return nil } -type tokenizedSQL struct { - Language string `json:"language"` - Statement string `json:"statement"` -} - // Query issues a SQL query formatted for the FeatureBase cloud query endpoint. -func (cq *Queryer) Query(org, db, sql string) (*featurebase.WireQueryResponse, error) { +func (cq *Queryer) Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) { if time.Since(cq.lastRefresh) > TokenRefreshTimeout { if err := cq.tokenRefresh(); err != nil { return nil, errors.Wrap(err, "refreshing token") } } - url := fmt.Sprintf("%s/v2/databases/%s/query/sql", cq.Host, db) - - sqlReq := &tokenizedSQL{ - Language: "sql", - Statement: sql, - } - var buf bytes.Buffer - if err := json.NewEncoder(&buf).Encode(sqlReq); err != nil { - return nil, errors.Wrapf(err, "encoding sql request: %s", sql) + url := fmt.Sprintf("%s/databases/%s/sql", cq.Host, db) + if db == "" { + url = fmt.Sprintf("%s/sql", cq.Host) } client := &http.Client{ Timeout: time.Second * 30, } - req, err := http.NewRequest(http.MethodPost, url, &buf) + req, err := http.NewRequest(http.MethodPost, url, sql) if err != nil { return nil, errors.Wrap(err, "creating new post request") } - req.Header.Add("Content-Type", "application/json") + req.Header.Add("Content-Type", "text/plain") req.Header.Add("Authorization", cq.token) var resp *http.Response @@ -141,7 +129,3 @@ func (cq *Queryer) HTTPRequest(method, path, body string, v interface{}) ([]byte return bodbytes, nil } - -type cloudResponse struct { - Results featurebase.WireQueryResponse `json:"results"` -} diff --git a/cli/meta.go b/cli/meta.go new file mode 100644 index 000000000..f4ce8f302 --- /dev/null +++ b/cli/meta.go @@ -0,0 +1,926 @@ +package cli + +import ( + "bufio" + "io" + "os" + "os/exec" + "os/signal" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" + "unicode" + + "github.com/featurebasedb/featurebase/v3/errors" +) + +// action is used to indicate how CLICommand should respond after execution a +// given metaCommand. For example, an action of type "reset" tells CLICommand +// that the buffer has been reset and it needs to change its user prompt. +type action string + +const ( + actionNone = "" + actionQuit = "quit" + actionReset = "reset" +) + +// metaCommand is the interface for any type responding to a "\" meta-command. +type metaCommand interface { + execute(cmd *CLICommand) (action, error) +} + +// Ensure type implements interface. +var _ metaCommand = (*metaBang)(nil) +var _ metaCommand = (*metaBorder)(nil) +var _ metaCommand = (*metaChangeDirectory)(nil) +var _ metaCommand = (*metaConnect)(nil) +var _ metaCommand = (*metaEcho)(nil) +var _ metaCommand = (*metaExpanded)(nil) +var _ metaCommand = (*metaFile)(nil) +var _ metaCommand = (*metaHelp)(nil) +var _ metaCommand = (*metaInclude)(nil) +var _ metaCommand = (*metaListDatabases)(nil) +var _ metaCommand = (*metaListTables)(nil) +var _ metaCommand = (*metaOrg)(nil) +var _ metaCommand = (*metaOutput)(nil) +var _ metaCommand = (*metaPrint)(nil) +var _ metaCommand = (*metaPSet)(nil) +var _ metaCommand = (*metaQEcho)(nil) +var _ metaCommand = (*metaQuit)(nil) +var _ metaCommand = (*metaReset)(nil) +var _ metaCommand = (*metaSet)(nil) +var _ metaCommand = (*metaTiming)(nil) +var _ metaCommand = (*metaTuplesOnly)(nil) +var _ metaCommand = (*metaWarn)(nil) +var _ metaCommand = (*metaWatch)(nil) +var _ metaCommand = (*metaWrite)(nil) + +// //////////////////////////////////////////////////////////////////////////// +// bang (!) +// //////////////////////////////////////////////////////////////////////////// +type metaBang struct { + args []string +} + +func newMetaBang(args []string) *metaBang { + return &metaBang{ + args: args, + } +} + +func (m *metaBang) execute(cmd *CLICommand) (action, error) { + if len(m.args) == 0 { + return actionNone, errors.Errorf("meta command '!' requires at least one argument") + } + c := exec.Command(m.args[0]) + c.Args = m.args + c.Stdout = cmd.Stdout + err := c.Run() + return actionNone, errors.Wrap(err, "running bang command") +} + +// //////////////////////////////////////////////////////////////////////////// +// border (sub-command of pset) +// //////////////////////////////////////////////////////////////////////////// +type metaBorder struct { + args []string +} + +func newMetaBorder(args []string) *metaBorder { + return &metaBorder{ + args: args, + } +} + +func (m *metaBorder) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + // pass + case 1: + switch m.args[0] { + case "1": + cmd.writeOptions.border = 1 + case "2": + cmd.writeOptions.border = 2 + default: + cmd.writeOptions.border = 0 + } + default: + return actionNone, errors.Errorf("meta command 'border' takes zero or one argument") + } + + cmd.Printf("Border style is %d.\n", cmd.writeOptions.border) + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// cd +// //////////////////////////////////////////////////////////////////////////// +type metaChangeDirectory struct { + args []string +} + +func newMetaChangeDirectory(args []string) *metaChangeDirectory { + return &metaChangeDirectory{ + args: args, + } +} + +func (m *metaChangeDirectory) execute(cmd *CLICommand) (action, error) { + if len(m.args) != 1 { + return actionNone, errors.Errorf("meta command 'cd' requires exactly one argument") + } + err := cmd.workingDir.cd(m.args[0]) + return actionNone, errors.Wrap(err, "running cd command") +} + +// //////////////////////////////////////////////////////////////////////////// +// connect (or c) +// //////////////////////////////////////////////////////////////////////////// +type metaConnect struct { + args []string +} + +func newMetaConnect(args []string) *metaConnect { + return &metaConnect{ + args: args, + } +} + +func (m *metaConnect) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + cmd.Printf(cmd.connectionMessage()) + return actionNone, nil + case 1: + err := cmd.connectToDatabase(m.args[0]) + return actionNone, err + + default: + return actionNone, errors.Errorf("meta command 'connect' takes zero or one argument") + } +} + +// //////////////////////////////////////////////////////////////////////////// +// echo +// //////////////////////////////////////////////////////////////////////////// +type metaEcho struct { + args []string +} + +func newMetaEcho(args []string) *metaEcho { + return &metaEcho{ + args: args, + } +} + +func (m *metaEcho) execute(cmd *CLICommand) (action, error) { + return echo(m.args, cmd.Stdout) +} + +func echo(args []string, w io.Writer) (action, error) { + switch len(args) { + case 0: + w.Write([]byte("\n")) + return actionNone, nil + default: + var s string + switch args[0] { + // TODO(tlt): currently the "-n" doesn't *appear* to work because + // the readline package, on its next iteration of the read loop, + // clobbers anything written to the current line (i.e. anything + // without a line feed). We need to figure out how to keep the + // contents of the current line, and append the next readline prompt + // to the end of it. + case "-n": + s = strings.Join(args[1:], " ") + default: + s = strings.Join(args, " ") + "\n" + } + w.Write([]byte(s)) + return actionNone, nil + } +} + +// //////////////////////////////////////////////////////////////////////////// +// expanded (x) +// //////////////////////////////////////////////////////////////////////////// +type metaExpanded struct { + args []string +} + +func newMetaExpanded(args []string) *metaExpanded { + return &metaExpanded{ + args: args, + } +} + +func (m *metaExpanded) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + cmd.writeOptions.expanded = !cmd.writeOptions.expanded + case 1: + switch m.args[0] { + case "on": + cmd.writeOptions.expanded = true + case "off": + cmd.writeOptions.expanded = false + default: + return actionNone, errors.Errorf("unrecognized value \"%s\" for \"expanded\": Boolean expected", m.args[0]) + } + default: + return actionNone, errors.Errorf("meta command 'expanded' takes zero or one argument") + } + + sExpanded := "on" + if !cmd.writeOptions.expanded { + sExpanded = "off" + } + + cmd.Printf("Expanded display is %s.\n", sExpanded) + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// file +// //////////////////////////////////////////////////////////////////////////// +type metaFile struct { + args []string +} + +func newMetaFile(args []string) *metaFile { + return &metaFile{ + args: args, + } +} + +func (m *metaFile) execute(cmd *CLICommand) (action, error) { + if len(m.args) != 1 { + return actionNone, errors.Errorf("meta command 'file' requires exactly one argument") + } + + // TODO(tlt): I think instead of opening the file here, we should get the + // absolute path to the file and store that until we actually need to open + // the file (i.e until we actualy execute the query). But to do that we'll + // need to change the Reader() method to return an error. + file, err := os.Open(m.args[0]) + if err != nil { + return actionNone, errors.Wrapf(err, "opening file: %s", m.args[0]) + } + + pf := newPartFile(file) + + // TODO: addPart returns a query. We'll have to revisit this when we update + // the \i command to accept files containing SQL. + _, err = cmd.buffer.addPart(pf) + return actionNone, errors.Wrap(err, "adding part file") +} + +// //////////////////////////////////////////////////////////////////////////// +// help (?) +// //////////////////////////////////////////////////////////////////////////// +type metaHelp struct { + args []string +} + +func newMetaHelp(args []string) *metaHelp { + return &metaHelp{ + args: args, + } +} + +func (m *metaHelp) execute(cmd *CLICommand) (action, error) { + helpText := `General + \q[uit] quit psql + \watch [SEC] execute query every SEC seconds + +Help + \? [commands] show help on backslash commands + +Query Buffer + \p[rint] show the contents of the query buffer + \r[eset] reset (clear) the query buffer + \w FILE write query buffer to file + +Input/Output + \echo [-n] [STRING] write string to standard output (-n for no newline) + \file ... reference a local file to stream to the server + \i[nclude] FILE execute commands from file + \o [FILE] send all query results to file + \qecho [-n] [STRING] write string to \o output stream (-n for no newline) + \warn [-n] [STRING] write string to standard error (-n for no newline) + +Informational + \d list tables and views + \dt list tables + \dv list views + \l list databases + +Formatting + \pset [NAME [VALUE]] set table output option + (border|expanded|tuples_only) + \t [on|off] show only rows + \x [on|off|auto] toggle expanded output + +Connection + \c[onnect] [DBNAME] connect to new database + \org [ORGNAME] set organization id + +Operating System + \cd [DIR] change the current working directory + \timing [on|off] toggle timing of commands + \! [COMMAND] execute command in shell or start interactive shell +` + cmd.Printf("%s\n", helpText) + + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// include (or i) +// //////////////////////////////////////////////////////////////////////////// +type metaInclude struct { + args []string +} + +func newMetaInclude(args []string) *metaInclude { + return &metaInclude{ + args: args, + } +} + +func (m *metaInclude) execute(cmd *CLICommand) (action, error) { + if len(m.args) != 1 { + return actionNone, errors.Errorf("meta command 'include' requires exactly one argument") + } + + file, err := os.Open(m.args[0]) + if err != nil { + return actionNone, errors.Wrapf(err, "opening file: %s", m.args[0]) + } + defer file.Close() + + splitter := newSplitter() + buffer := newBuffer() + + // Read the file by line, pushing the lines into a new line splitter, then + // sending that output to a new buffer. + sc := bufio.NewScanner(file) + for sc.Scan() { + line := sc.Text() // GET the line string + + qps, mcs, err := splitter.split(line) + if err != nil { + return actionNone, errors.Wrapf(err, "splitting lines") + } else if len(mcs) > 0 { + return actionNone, errors.Errorf("include does not support meta-commands") + } + + for i := range qps { + if qry, err := buffer.addPart(qps[i]); err != nil { + return actionNone, errors.Wrap(err, "adding part to buffer") + } else if qry != nil { + if err := cmd.executeAndWriteQuery(qry); err != nil { + return actionNone, errors.Wrap(err, "executing query") + } + } + } + } + if err := sc.Err(); err != nil { + return actionNone, errors.Wrapf(err, "scanning file: %s", m.args[0]) + } + + return actionReset, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// list databases (l) +// //////////////////////////////////////////////////////////////////////////// +type metaListDatabases struct{} + +func newMetaListDatabases() *metaListDatabases { + return &metaListDatabases{} +} + +func (m *metaListDatabases) execute(cmd *CLICommand) (action, error) { + qry := []queryPart{ + newPartRaw("SHOW DATABASES"), + } + + if err := cmd.executeAndWriteQuery(qry); err != nil { + return actionNone, errors.Wrap(err, "executing query") + } + + return actionReset, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// list tables (d or dt) +// //////////////////////////////////////////////////////////////////////////// +type metaListTables struct{} + +func newMetaListTables() *metaListTables { + return &metaListTables{} +} + +func (m *metaListTables) execute(cmd *CLICommand) (action, error) { + qry := []queryPart{ + newPartRaw("SHOW TABLES"), + } + + if err := cmd.executeAndWriteQuery(qry); err != nil { + return actionNone, errors.Wrap(err, "executing query") + } + + return actionReset, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// org +// //////////////////////////////////////////////////////////////////////////// +type metaOrg struct { + args []string +} + +func newMetaOrg(args []string) *metaOrg { + return &metaOrg{ + args: args, + } +} + +func (m *metaOrg) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + + case 0: + cmd.Printf(cmd.orgMessage()) + return actionNone, nil + case 1: + cmd.OrganizationID = m.args[0] + cmd.Printf(cmd.orgMessage()) + return actionNone, nil + + default: + return actionNone, errors.Errorf("meta command 'org' takes zero or one argument") + } +} + +// //////////////////////////////////////////////////////////////////////////// +// output (o) +// //////////////////////////////////////////////////////////////////////////// +type metaOutput struct { + args []string +} + +func newMetaOutput(args []string) *metaOutput { + return &metaOutput{ + args: args, + } +} + +func (m *metaOutput) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + // Close cmd.output (if closable). + if err := cmd.closeOutput(); err != nil { + return actionNone, errors.Wrapf(err, "closing output") + } + + // Set cmd.output to cmd.Stdout. + cmd.output = cmd.Stdout + + return actionNone, nil + + case 1: + // If the argument is a fully-qualifed file path, just use that. + // Otherwise, prepend it with the current working directory. + fpath, err := filepath.Abs(m.args[0]) + if err != nil { + return actionNone, errors.Wrapf(err, "getting absolute file path for file: %s", m.args[0]) + } + + cmd.output, err = os.OpenFile(fpath, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o600) + if err != nil { + return actionNone, errors.Wrapf(err, "opening file: %s", fpath) + } + + return actionNone, nil + + default: + return actionNone, errors.Errorf("meta command 'output' takes zero or one argument") + } +} + +// //////////////////////////////////////////////////////////////////////////// +// print (or p) +// //////////////////////////////////////////////////////////////////////////// +type metaPrint struct{} + +func newMetaPrint() *metaPrint { + return &metaPrint{} +} + +func (m *metaPrint) execute(cmd *CLICommand) (action, error) { + cmd.Printf(cmd.buffer.print() + "\n") + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// pset +// //////////////////////////////////////////////////////////////////////////// +type metaPSet struct { + args []string +} + +func newMetaPSet(args []string) *metaPSet { + return &metaPSet{ + args: args, + } +} + +func (m *metaPSet) print(cmd *CLICommand) { + onOff := func(b bool) string { + if b { + return "on" + } + return "off" + } + + fmt := `border %d +expanded %s +tuples_only %s +` + + cmd.Printf(fmt, + cmd.writeOptions.border, + onOff(cmd.writeOptions.expanded), + onOff(cmd.writeOptions.tuplesOnly), + ) + +} + +func (m *metaPSet) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + m.print(cmd) + return actionNone, nil + case 1, 2: + switch m.args[0] { + case "border": + sub := newMetaBorder(m.args[1:]) + return sub.execute(cmd) + case "expanded": + sub := newMetaExpanded(m.args[1:]) + return sub.execute(cmd) + case "tuples_only": + sub := newMetaTuplesOnly(m.args[1:]) + return sub.execute(cmd) + default: + return actionNone, errors.Errorf("unrecognized value \"%s\" for \"pset\"", m.args[0]) + } + default: + return actionNone, errors.Errorf("meta command 'pset' takes zero, one, or two arguments") + } +} + +// //////////////////////////////////////////////////////////////////////////// +// qecho +// //////////////////////////////////////////////////////////////////////////// +type metaQEcho struct { + args []string +} + +func newMetaQEcho(args []string) *metaQEcho { + return &metaQEcho{ + args: args, + } +} + +func (m *metaQEcho) execute(cmd *CLICommand) (action, error) { + return echo(m.args, cmd.output) +} + +// //////////////////////////////////////////////////////////////////////////// +// quit (or q) +// //////////////////////////////////////////////////////////////////////////// +type metaQuit struct{} + +func newMetaQuit() *metaQuit { + return &metaQuit{} +} + +func (m *metaQuit) execute(cmd *CLICommand) (action, error) { + return actionQuit, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// reset (or r) +// //////////////////////////////////////////////////////////////////////////// +type metaReset struct{} + +func newMetaReset() *metaReset { + return &metaReset{} +} + +func (m *metaReset) execute(cmd *CLICommand) (action, error) { + cmd.Printf(cmd.buffer.reset()) + return actionReset, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// set +// //////////////////////////////////////////////////////////////////////////// +type metaSet struct { + args []string +} + +func newMetaSet(args []string) *metaSet { + return &metaSet{ + args: args, + } +} + +func (m *metaSet) execute(cmd *CLICommand) (action, error) { + // TODO: set the variable (or clear it, etc) + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// timing +// //////////////////////////////////////////////////////////////////////////// +type metaTiming struct { + args []string +} + +func newMetaTiming(args []string) *metaTiming { + return &metaTiming{ + args: args, + } +} + +func (m *metaTiming) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + cmd.writeOptions.timing = !cmd.writeOptions.timing + case 1: + switch m.args[0] { + case "on": + cmd.writeOptions.timing = true + case "off": + cmd.writeOptions.timing = false + default: + return actionNone, errors.Errorf("unrecognized value \"%s\" for \"\timing\": Boolean expected", m.args[0]) + } + default: + return actionNone, errors.Errorf("meta command 'timing' takes zero or one argument") + } + + sTiming := "on" + if !cmd.writeOptions.timing { + sTiming = "off" + } + + cmd.Printf("Timing is %s.\n", sTiming) + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// tuples_only (t) +// //////////////////////////////////////////////////////////////////////////// +type metaTuplesOnly struct { + args []string +} + +func newMetaTuplesOnly(args []string) *metaTuplesOnly { + return &metaTuplesOnly{ + args: args, + } +} + +func (m *metaTuplesOnly) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + cmd.writeOptions.tuplesOnly = !cmd.writeOptions.tuplesOnly + case 1: + switch m.args[0] { + case "on": + cmd.writeOptions.tuplesOnly = true + case "off": + cmd.writeOptions.tuplesOnly = false + default: + return actionNone, errors.Errorf("unrecognized value \"%s\" for \"tuples_only\": Boolean expected", m.args[0]) + } + default: + return actionNone, errors.Errorf("meta command 'tuples_only' takes zero or one argument") + } + + sTuplesOnly := "on" + if !cmd.writeOptions.tuplesOnly { + sTuplesOnly = "off" + } + + cmd.Printf("Tuples only is %s.\n", sTuplesOnly) + return actionNone, nil +} + +// //////////////////////////////////////////////////////////////////////////// +// warn +// //////////////////////////////////////////////////////////////////////////// +type metaWarn struct { + args []string +} + +func newMetaWarn(args []string) *metaWarn { + return &metaWarn{ + args: args, + } +} + +func (m *metaWarn) execute(cmd *CLICommand) (action, error) { + return echo(m.args, cmd.Stderr) +} + +// //////////////////////////////////////////////////////////////////////////// +// watch +// //////////////////////////////////////////////////////////////////////////// +type metaWatch struct { + args []string +} + +func newMetaWatch(args []string) *metaWatch { + return &metaWatch{ + args: args, + } +} + +func (m *metaWatch) execute(cmd *CLICommand) (action, error) { + period := 2 * time.Second + + qry := cmd.buffer.lastQuery + if qry == nil { + cmd.Errorf(`\watch cannot be used with an empty query` + "\n") + return actionNone, nil + } + + switch len(m.args) { + case 1: + val, err := strconv.Atoi(m.args[0]) + if err != nil { + return actionNone, errors.Errorf("invalid watch argument: %s", m.args[0]) + } + period = time.Duration(val) * time.Second + fallthrough + case 0: + // Listen for a SIGTERM to cancel out of the \watch loop. + controlC := make(chan os.Signal, 2) + signal.Notify(controlC, os.Interrupt, syscall.SIGTERM) + + // In the absence of a SIGTERM, use a ticker to execute the query every + // "period" duration. + ticker := time.NewTicker(period) + for { + cmd.Printf("%s (every %s)\n\n", time.Now(), period) + if err := cmd.executeAndWriteQuery(qry); err != nil { + return actionNone, errors.Wrap(err, "executing query") + } + select { + case <-controlC: + return actionNone, nil + case <-ticker.C: + } + } + default: + return actionNone, errors.Errorf("meta command 'watch' takes zero or one argument") + } +} + +// //////////////////////////////////////////////////////////////////////////// +// write (w) +// //////////////////////////////////////////////////////////////////////////// +type metaWrite struct { + args []string +} + +func newMetaWrite(args []string) *metaWrite { + return &metaWrite{ + args: args, + } +} + +func (m *metaWrite) execute(cmd *CLICommand) (action, error) { + switch len(m.args) { + case 0: + cmd.Errorf(`\w: missing required argument` + "\n") + return actionNone, nil + + case 1: + // Open file. + fpath, err := filepath.Abs(m.args[0]) + if err != nil { + return actionNone, errors.Wrapf(err, "getting absolute file path for file: %s", m.args[0]) + } + + file, err := os.OpenFile(fpath, os.O_WRONLY|os.O_CREATE, 0o600) + if err != nil { + return actionNone, errors.Wrapf(err, "opening file: %s", fpath) + } + defer file.Close() + + // Write query buffer to file. + if _, err := io.Copy(file, cmd.buffer.Reader()); err != nil { + return actionNone, errors.Wrapf(err, "writing query buffer to file: %s", fpath) + } + + return actionNone, nil + + default: + return actionNone, errors.Errorf("meta command 'w' exactly one argument") + } +} + +////////////////////////////////////////////////////////////////////////////// + +// splitMetaCommand takes a string which follows a backslash, with a +// formats like: +// +// `cmd` +// `cmd arg1 arg2` +// `cmd 'arg1' arg2 'arg three'` +// +// It returns the metaCommand which maps to `cmd`. +func splitMetaCommand(in string) (metaCommand, error) { + parts := strings.SplitN(in, ` `, 2) + key := strings.TrimRightFunc(parts[0], unicode.IsSpace) + + args := []string{} + if len(parts) > 1 { + sb := &strings.Builder{} + quoted := false + for _, r := range parts[1] { + if r == '\'' { + quoted = !quoted + } else if !quoted && r == ' ' { + args = append(args, sb.String()) + sb.Reset() + } else { + sb.WriteRune(r) + } + } + if sb.Len() > 0 { + args = append(args, sb.String()) + } + } + + switch key { + case "!": + return newMetaBang(args), nil + case "cd": + return newMetaChangeDirectory(args), nil + case "c", "connect": + return newMetaConnect(args), nil + case "d", "dt": + return newMetaListTables(), nil + case "echo": + return newMetaEcho(args), nil + case "file": + return newMetaFile(args), nil + case "?": + return newMetaHelp(args), nil + case "i", "include": + return newMetaInclude(args), nil + case "l": + return newMetaListDatabases(), nil + case "o": + return newMetaOutput(args), nil + case "org": + return newMetaOrg(args), nil + case "p", "print": + return newMetaPrint(), nil + case "pset": + return newMetaPSet(args), nil + case "qecho": + return newMetaQEcho(args), nil + case "q", "quit": + return newMetaQuit(), nil + case "r", "reset": + return newMetaReset(), nil + case "set": + return newMetaSet(args), nil + case "t": + return newMetaTuplesOnly(args), nil + case "timing": + return newMetaTiming(args), nil + case "warn": + return newMetaWarn(args), nil + case "watch": + return newMetaWatch(args), nil + case "w": + return newMetaWrite(args), nil + case "x": + return newMetaExpanded(args), nil + default: + return nil, errors.Errorf("unsupported meta-command: '%s'", key) + } +} diff --git a/cli/parts.go b/cli/parts.go new file mode 100644 index 000000000..a936ebc59 --- /dev/null +++ b/cli/parts.go @@ -0,0 +1,130 @@ +package cli + +import ( + "fmt" + "io" + "os" + "strings" +) + +// query is a collection of queryParts which, when applied together, make up an +// executable SQL query. +type query []queryPart + +func (q query) String() string { + var sb strings.Builder + for i := range q { + sb.WriteString(q[i].String()) + if i < len(q)-1 { + sb.WriteRune('\n') + } + } + return sb.String() +} + +// Reader returns the query as an io.Reader so that it can be passed to, for +// example, http.Post(). +func (q query) Reader() io.Reader { + readers := make([]io.Reader, 0, len(q)) + for i := range q { + readers = append(readers, q[i].Reader()) + } + return io.MultiReader(readers...) +} + +// queryPart is an interface representing anything which can use to build up a +// query. +type queryPart interface { + fmt.Stringer + Reader() io.Reader +} + +// //////////////////////////////////////////////////////////////////////////// +// raw +// //////////////////////////////////////////////////////////////////////////// + +// Ensure type implements interface. +var _ queryPart = (*partRaw)(nil) + +type partRaw struct { + raw string +} + +func newPartRaw(s string) *partRaw { + return &partRaw{ + raw: s, + } +} + +func (p *partRaw) Reader() io.Reader { + return strings.NewReader(p.raw + "\n") +} + +func (p *partRaw) String() string { + return p.raw +} + +// //////////////////////////////////////////////////////////////////////////// +// file +// //////////////////////////////////////////////////////////////////////////// + +// Ensure type implements interface. +var _ queryPart = (*partFile)(nil) + +type partFile struct { + file *os.File +} + +func newPartFile(f *os.File) *partFile { + return &partFile{ + file: f, + } +} + +func (p *partFile) Reader() io.Reader { + return p.file +} + +func (p *partFile) String() string { + return fmt.Sprintf("[file: %s]", p.file.Name()) +} + +// //////////////////////////////////////////////////////////////////////////// +// batch file +// //////////////////////////////////////////////////////////////////////////// + +// Ensure type implements interface. +var _ queryPart = (*partBatchFile)(nil) + +type partBatchFile struct { + file *os.File +} + +func (p *partBatchFile) Reader() io.Reader { + return p.file +} + +func (p *partBatchFile) String() string { + return p.file.Name() +} + +// //////////////////////////////////////////////////////////////////////////// +// terminator (i.e. ";") +// //////////////////////////////////////////////////////////////////////////// + +// Ensure type implements interface. +var _ queryPart = (*partTerminator)(nil) + +type partTerminator struct{} + +func newPartTerminator() *partTerminator { + return &partTerminator{} +} + +func (p *partTerminator) Reader() io.Reader { + return nil +} + +func (p *partTerminator) String() string { + return terminationChar +} diff --git a/cli/queryer.go b/cli/queryer.go index bde2b6cbc..6d7ec7865 100644 --- a/cli/queryer.go +++ b/cli/queryer.go @@ -1,20 +1,26 @@ package cli import ( - "bytes" - "encoding/json" "fmt" "io" "net/http" + "time" featurebase "github.com/featurebasedb/featurebase/v3" - "github.com/featurebasedb/featurebase/v3/dax" - queryerhttp "github.com/featurebasedb/featurebase/v3/dax/queryer/http" "github.com/pkg/errors" ) type Queryer interface { - Query(org, db, sql string) (*featurebase.WireQueryResponse, error) + Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) +} + +// Ensure type implements interface. +var _ Queryer = (*nopQueryer)(nil) + +type nopQueryer struct{} + +func (qryr *nopQueryer) Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) { + return nil, errors.Errorf("no-op queryer") } // Ensure type implements interface. @@ -27,13 +33,10 @@ type standardQueryer struct { Port string } -func (qryr *standardQueryer) Query(org, db, sql string) (*featurebase.WireQueryResponse, error) { - buf := bytes.Buffer{} +func (qryr *standardQueryer) Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) { url := fmt.Sprintf("%s/sql", hostPort(qryr.Host, qryr.Port)) - buf.Write([]byte(sql)) - - resp, err := http.Post(url, "application/json", &buf) + resp, err := http.Post(url, "application/json", sql) if err != nil { return nil, errors.Wrapf(err, "posting query") } @@ -53,32 +56,36 @@ func (qryr *standardQueryer) Query(org, db, sql string) (*featurebase.WireQueryR } // Ensure type implements interface. -var _ Queryer = (*daxQueryer)(nil) +var _ Queryer = (*serverlessQueryer)(nil) -// daxQueryer is similar to the standardQueryer except that it hits a different -// endpoint, and its payload is a json object which includes, in addition to the -// sql statement, things like org and db. -type daxQueryer struct { +// serverlessQueryer is similar to the standardQueryer except that it hits a +// different endpoint, and its payload is database-aware. +type serverlessQueryer struct { Host string Port string } -func (qryr *daxQueryer) Query(org, db, sql string) (*featurebase.WireQueryResponse, error) { - buf := bytes.Buffer{} - url := fmt.Sprintf("%s/queryer/sql", hostPort(qryr.Host, qryr.Port)) - - sqlReq := &queryerhttp.SQLRequest{ - OrganizationID: dax.OrganizationID(org), - DatabaseID: dax.DatabaseID(db), - SQL: sql, - } - if err := json.NewEncoder(&buf).Encode(sqlReq); err != nil { - return nil, errors.Wrapf(err, "encoding sql request: %s", sql) +func (qryr *serverlessQueryer) Query(org string, db string, sql io.Reader) (*featurebase.WireQueryResponse, error) { + // buf := bytes.Buffer{} + url := fmt.Sprintf("%s/queryer/databases/%s/sql", hostPort(qryr.Host, qryr.Port), db) + if db == "" { + url = fmt.Sprintf("%s/queryer/sql", hostPort(qryr.Host, qryr.Port)) } - resp, err := http.Post(url, "application/json", &buf) + client := &http.Client{ + Timeout: time.Second * 30, + } + + req, err := http.NewRequest(http.MethodPost, url, sql) if err != nil { - return nil, errors.Wrapf(err, "posting query") + return nil, errors.Wrap(err, "creating new post request") + } + req.Header.Add("Content-Type", "text/plain") + req.Header.Add("OrganizationID", org) + + var resp *http.Response + if resp, err = client.Do(req); err != nil { + return nil, errors.Wrap(err, "executing post request") } fullbod, err := io.ReadAll(resp.Body) diff --git a/cli/splitter.go b/cli/splitter.go new file mode 100644 index 000000000..ca62eb257 --- /dev/null +++ b/cli/splitter.go @@ -0,0 +1,119 @@ +package cli + +import ( + "strings" + + "github.com/pkg/errors" +) + +// splitter is a line splitter which splits a line into queryParts and +// metaCommands. It may not be necessary to have this be a separate struct since +// it contains no members and just has the one `split()` method, but here we +// are. +type splitter struct{} + +func newSplitter() *splitter { + return &splitter{} +} + +// split splits the given line into queryParts and metaCommands. +// If a metaCommand is found, everything after that is considered either arguments to that +// metaCommand, or additional metaCommands. In other words, queryParts can not follow +// metaCommands in the same line. +// +// A line can contain any of the following patterns: +// 1- [queryParts...]: "select * from tbl; select" +// 2- [metaCommands...]: "\! pwd \q" +// 3- [queryParts...][metaCommands...]: "select * from \i file.sql" +func (s *splitter) split(line string) ([]queryPart, []metaCommand, error) { + // Look for a meta command + parts := strings.SplitN(line, `\`, 2) + + switch len(parts) { + case 1: + // slice of queryParts (pattern 1) + if qps, err := s.splitQueryParts(strings.TrimSpace(parts[0])); err != nil { + return nil, nil, errors.Wrap(err, "splitting query parts") + } else { + return qps, nil, nil + } + case 2: + // slice of parts + slice of meta commands (pattern 3) + // or + // slice of meta commands (pattern 2) + qps, err := s.splitQueryParts(strings.TrimSpace(parts[0])) + if err != nil { + return nil, nil, errors.Wrap(err, "splitting query parts") + } + + mcs, err := s.splitMetaCommands(strings.TrimSpace(parts[1])) + if err != nil { + return nil, nil, errors.Wrap(err, "splitting meta commands") + } + + return qps, mcs, nil + } + + return nil, nil, nil +} + +func (s *splitter) splitQueryParts(line string) ([]queryPart, error) { + if line == "" { + return nil, nil + } + + // Look for a termination character; + parts := strings.Split(line, terminationChar) + + if len(parts) == 1 { + part0 := strings.TrimSpace(parts[0]) + return []queryPart{ + newPartRaw(part0), + }, nil + } + + qps := make([]queryPart, 0) + for i := range parts { + part := strings.TrimSpace(parts[i]) + if part == "" { + // If the line starts with a ";", treat it as a terminator for a + // previous line. + if i == 0 { + qps = append(qps, &partTerminator{}) + } + continue + } + qps = append(qps, newPartRaw(part)) + if i < len(parts)-1 { + qps = append(qps, &partTerminator{}) + } + } + + return qps, nil +} + +func (s *splitter) splitMetaCommands(in string) ([]metaCommand, error) { + parts := strings.Split(in, `\`) + if len(parts) == 1 { + mc, err := splitMetaCommand(parts[0]) + if err != nil { + return nil, errors.Wrapf(err, "splitting meta command: %s", parts[0]) + } + return []metaCommand{mc}, nil + } + + mcs := make([]metaCommand, 0) + for i := range parts { + part := strings.TrimSpace(parts[i]) + if part == "" { + continue + } + mc, err := splitMetaCommand(part) + if err != nil { + return nil, errors.Wrapf(err, "splitting meta command: %s", part) + } + mcs = append(mcs, mc) + } + + return mcs, nil +} diff --git a/cli/splitter_test.go b/cli/splitter_test.go new file mode 100644 index 000000000..54b0fb3eb --- /dev/null +++ b/cli/splitter_test.go @@ -0,0 +1,146 @@ +package cli + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestSplitter(t *testing.T) { + s := newSplitter() + t.Run("Split", func(t *testing.T) { + tests := []struct { + line string + expQueryParts []queryPart + expMetaCommands []metaCommand + expError string + }{ + { + line: `foo`, + expQueryParts: []queryPart{ + newPartRaw("foo"), + }, + }, + { + line: `foo;`, + expQueryParts: []queryPart{ + newPartRaw("foo"), + newPartTerminator(), + }, + }, + { + line: `foo; `, + expQueryParts: []queryPart{ + newPartRaw("foo"), + newPartTerminator(), + }, + }, + { + line: `foo; ; `, + expQueryParts: []queryPart{ + newPartRaw("foo"), + newPartTerminator(), + }, + }, + { + line: `foo; bar`, + expQueryParts: []queryPart{ + newPartRaw("foo"), + newPartTerminator(), + newPartRaw("bar"), + }, + }, + { + line: `foo; bar;`, + expQueryParts: []queryPart{ + newPartRaw("foo"), + newPartTerminator(), + newPartRaw("bar"), + newPartTerminator(), + }, + }, + { + line: `\q`, + expMetaCommands: []metaCommand{ + &metaQuit{}, + }, + }, + { + line: ` \p`, + expMetaCommands: []metaCommand{ + &metaPrint{}, + }, + }, + { + line: `\q \p`, + expMetaCommands: []metaCommand{ + &metaQuit{}, + &metaPrint{}, + }, + }, + { + line: `\q \p arg1 arg2`, + expMetaCommands: []metaCommand{ + &metaQuit{}, + &metaPrint{}, + }, + }, + { + line: `\set`, + expMetaCommands: []metaCommand{ + &metaSet{ + args: []string{}, + }, + }, + }, + { + line: `\set arg1 arg2`, + expMetaCommands: []metaCommand{ + &metaSet{ + args: []string{"arg1", "arg2"}, + }, + }, + }, + { + line: `\set 'arg1' 'arg2'`, + expMetaCommands: []metaCommand{ + &metaSet{ + args: []string{"arg1", "arg2"}, + }, + }, + }, + { + line: `\set 'arg1' '"arg2"'`, + expMetaCommands: []metaCommand{ + &metaSet{ + args: []string{"arg1", "\"arg2\""}, + }, + }, + }, + { + line: `\`, + expError: "unsupported meta-command:", + }, + { + line: `\xyzxyz`, + expError: "unsupported meta-command:", + }, + } + for i, tt := range tests { + t.Run(fmt.Sprintf("test-%d-%s", i, tt.line), func(t *testing.T) { + qps, mcs, err := s.split(tt.line) + if tt.expError != "" { + if assert.Error(t, err) { + assert.Contains(t, err.Error(), tt.expError) + } + return + } + + assert.NoError(t, err) + assert.ElementsMatch(t, tt.expQueryParts, qps) + assert.ElementsMatch(t, tt.expMetaCommands, mcs) + }) + } + }) +} diff --git a/cli/workingdir.go b/cli/workingdir.go new file mode 100644 index 000000000..a5b33964f --- /dev/null +++ b/cli/workingdir.go @@ -0,0 +1,21 @@ +package cli + +import ( + "os" +) + +// workingDir was originally set up with the intention of using it to maintain a +// reference to the current working directory. But it turns out we haven't +// really needed that so far. The `cd()` method is unsed in one of the meta +// commands, but we could probably just call `os.Chdir()` directly there. With +// that said, I'm leaving it here for now until we're abosolutely sure we don't +// need to use this for other directory/file handling functionality. +type workingDir struct{} + +func newWorkingDir() *workingDir { + return &workingDir{} +} + +func (wd *workingDir) cd(dir string) error { + return os.Chdir(dir) +} diff --git a/cli/writer.go b/cli/writer.go new file mode 100644 index 000000000..4111643d9 --- /dev/null +++ b/cli/writer.go @@ -0,0 +1,191 @@ +package cli + +import ( + "fmt" + "io" + + featurebase "github.com/featurebasedb/featurebase/v3" + "github.com/jedib0t/go-pretty/table" + "github.com/jedib0t/go-pretty/text" + "github.com/pkg/errors" +) + +// writeOptions contains user configuration options which describe how to write +// the query output. +type writeOptions struct { + border int + expanded bool + timing bool + tuplesOnly bool +} + +func defaultWriteOptions() *writeOptions { + return &writeOptions{ + border: 1, + expanded: false, + timing: true, + tuplesOnly: false, + } +} + +// writeTable writes the query response, taking the format into consideration. +// It sends query output to qOut, non-error informational output (such as query +// timing) to wOut, and errors to wErr. +func writeTable(r *featurebase.WireQueryResponse, format *writeOptions, qOut io.Writer, wOut io.Writer, wErr io.Writer) error { + if r == nil { + return errors.New("attempt to write out nil response") + } + if r.Error != "" { + if _, err := wErr.Write([]byte("Error: " + r.Error + "\n")); err != nil { + return errors.Wrapf(err, "writing error: %s", r.Error) + } + return writeWarnings(r, wErr) + } + + t := table.NewWriter() + t.SetOutputMirror(qOut) + switch format.border { + case 0: + t.SetStyle(styleBorder0) + case 1: + t.SetStyle(styleBorder1) + default: + t.SetStyle(styleBorder2) + // In expanded mode with a border, we need borders between each record. + if format.expanded { + t.Style().Options.SeparateRows = true + } + } + + // Don't uppercase the header values. + t.Style().Format.Header = text.FormatDefault + + if format.expanded { + // Expanded table + for _, row := range r.Data { + colRow := make([]interface{}, 2) + scolRow := make([]string, 2) + div := "\n" + for i, col := range r.Schema.Fields { + if i == len(r.Schema.Fields)-1 { + div = "" + } + scolRow[0] += fmt.Sprintf("%s%s", col.Name, div) + if row[i] == nil { + scolRow[1] += fmt.Sprintf("%s%s", nullValue, div) + } else { + scolRow[1] += fmt.Sprintf("%v%s", row[i], div) + } + } + colRow[0] = scolRow[0] + colRow[1] = scolRow[1] + t.AppendRow(table.Row(colRow[:])) + } + } else { + // Normal table (i.e. NOT expanded) + if !format.tuplesOnly { + t.AppendHeader(schemaToRow(r.Schema)) + } + for _, row := range r.Data { + // If the value is nil, replace it with a null string; go-pretty doesn't + // expect nil pointers in the data values. + for i := range row { + if row[i] == nil { + row[i] = nullValue + } + } + t.AppendRow(table.Row(row)) + } + } + t.Render() + + if err := writeWarnings(r, wErr); err != nil { + return err + } + + // Add some white space after query results. + qOut.Write([]byte("\n")) + + // Timing. + if format.timing { + if _, err := wOut.Write([]byte(fmt.Sprintf("Execution time: %dμs\n", r.ExecutionTime))); err != nil { + return errors.Wrapf(err, "writing execution time: %s", r.Error) + } + } + + return nil +} + +func schemaToRow(schema featurebase.WireQuerySchema) []interface{} { + ret := make([]interface{}, len(schema.Fields)) + for i, field := range schema.Fields { + ret[i] = field.Name + } + return ret +} + +func writeWarnings(r *featurebase.WireQueryResponse, w io.Writer) error { + if len(r.Warnings) == 0 { + return nil + } + + if _, err := w.Write([]byte("\n")); err != nil { + return errors.Wrapf(err, "writing line feed") + } + for _, warning := range r.Warnings { + if _, err := w.Write([]byte("Warning: " + warning + "\n")); err != nil { + return errors.Wrapf(err, "writing warning: %s", warning) + } + } + return nil +} + +var styleBorder2 table.Style = table.StyleDefault + +var styleBorder1 table.Style = table.Style{ + Name: "StyleBorder1", + Box: table.StyleBoxDefault, + Color: table.ColorOptionsDefault, + Format: table.FormatOptionsDefault, + Options: table.Options{ + DrawBorder: false, + SeparateColumns: true, + SeparateFooter: true, + SeparateHeader: true, + SeparateRows: false, + }, + Title: table.TitleOptionsDefault, +} + +var styleBorder0 table.Style = table.Style{ + Name: "StyleBorder0", + Box: table.BoxStyle{ + BottomLeft: "+", + BottomRight: "+", + BottomSeparator: "+", + Left: "|", + LeftSeparator: "+", + MiddleHorizontal: "-", + MiddleSeparator: " ", + MiddleVertical: " ", + PaddingLeft: "", + PaddingRight: "", + PageSeparator: "\n", + Right: "|", + RightSeparator: "+", + TopLeft: "+", + TopRight: "+", + TopSeparator: "+", + UnfinishedRow: " ~", + }, + Color: table.ColorOptionsDefault, + Format: table.FormatOptionsDefault, + Options: table.Options{ + DrawBorder: false, + SeparateColumns: true, + SeparateFooter: true, + SeparateHeader: true, + SeparateRows: false, + }, + Title: table.TitleOptionsDefault, +} diff --git a/cli/writer_test.go b/cli/writer_test.go new file mode 100644 index 000000000..dcb871278 --- /dev/null +++ b/cli/writer_test.go @@ -0,0 +1,164 @@ +package cli + +import ( + "bytes" + "fmt" + "strings" + "testing" + + featurebase "github.com/featurebasedb/featurebase/v3" + dax "github.com/featurebasedb/featurebase/v3/dax" + "github.com/stretchr/testify/assert" +) + +func TestWriter(t *testing.T) { + t.Run("writeTable", func(t *testing.T) { + wqr := &featurebase.WireQueryResponse{ + Schema: featurebase.WireQuerySchema{ + Fields: []*featurebase.WireQueryField{ + {Name: "_id", Type: dax.BaseTypeID}, + {Name: "name", Type: dax.BaseTypeString}, + {Name: "age", Type: dax.BaseTypeInt}, + }, + }, + Data: [][]interface{}{ + {1, "Amy", 44}, + {2, "Bob", 32}, + {3, "Cindy", 28}, + }, + } + + // TODO(tlt): used for debugging + // format := defaultWriteOptions() + // assert.NoError(t, writeTable(wqr, format, os.Stdout, os.Stdout, os.Stdout)) + // return + + tests := []struct { + format *writeOptions + expQOut string + expOut string + expErr string + }{ + { + // default format + format: defaultWriteOptions(), + expQOut: stringOfLines( + " _id | name | age ", + "-----+-------+-----", + " 1 | Amy | 44 ", + " 2 | Bob | 32 ", + " 3 | Cindy | 28 ", + "", + ), + expOut: "Execution time: 0μs\n", + expErr: "", + }, + { + // format.border = 2 (or higher) + format: &writeOptions{ + border: 2, + expanded: false, + timing: false, + tuplesOnly: false, + }, + expQOut: stringOfLines( + "+-----+-------+-----+", + "| _id | name | age |", + "+-----+-------+-----+", + "| 1 | Amy | 44 |", + "| 2 | Bob | 32 |", + "| 3 | Cindy | 28 |", + "+-----+-------+-----+", + "", + ), + expOut: "", + expErr: "", + }, + { + // format.border = 0 + format: &writeOptions{ + border: 0, + expanded: false, + timing: false, + tuplesOnly: false, + }, + expQOut: stringOfLines( + "_id name age", + "--- ----- ---", + " 1 Amy 44", + " 2 Bob 32", + " 3 Cindy 28", + "", + ), + expOut: "", + expErr: "", + }, + { + // format.tuplesOnly = true + format: &writeOptions{ + border: 1, + expanded: false, + timing: false, + tuplesOnly: true, + }, + expQOut: stringOfLines( + " 1 | Amy | 44 ", + " 2 | Bob | 32 ", + " 3 | Cindy | 28 ", + "", + ), + expOut: "", + expErr: "", + }, + { + // format.border = 2, expanded + format: &writeOptions{ + border: 2, + expanded: true, + timing: false, + tuplesOnly: false, + }, + expQOut: stringOfLines( + "+------+-------+", + "| _id | 1 |", + "| name | Amy |", + "| age | 44 |", + "+------+-------+", + "| _id | 2 |", + "| name | Bob |", + "| age | 32 |", + "+------+-------+", + "| _id | 3 |", + "| name | Cindy |", + "| age | 28 |", + "+------+-------+", + "", + ), + expOut: "", + expErr: "", + }, + } + for i, test := range tests { + t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) { + // Set up buffers to capture the output. + qOut := bytes.NewBuffer(make([]byte, 0, 100000)) + wOut := bytes.NewBuffer(make([]byte, 0, 100000)) + wErr := bytes.NewBuffer(make([]byte, 0, 100000)) + + assert.NoError(t, writeTable(wqr, test.format, qOut, wOut, wErr)) + + assert.Equal(t, test.expQOut, qOut.String()) + assert.Equal(t, test.expOut, wOut.String()) + assert.Equal(t, test.expErr, wErr.String()) + }) + } + }) +} + +func stringOfLines(lines ...string) string { + var sb strings.Builder + for _, line := range lines { + sb.WriteString(line + "\n") + } + return sb.String() +} diff --git a/cmd/cli.go b/cmd/cli.go index 3a97ad47d..993089f16 100644 --- a/cmd/cli.go +++ b/cmd/cli.go @@ -14,7 +14,7 @@ func newCLICommand(logdest logger.Logger) *cobra.Command { cliCmd = cli.NewCLICommand(logdest) cobraCmd := &cobra.Command{ Use: "cli", - Short: "Query FB with SQL3 from the command line", + Short: "Query FeatureBase with SQL from the command line", Long: ``, RunE: usageErrorWrapper(cliCmd), } @@ -24,7 +24,7 @@ func newCLICommand(logdest logger.Logger) *cobra.Command { flags.StringVarP(&cliCmd.Port, "port", "", cliCmd.Port, "port of FeatureBase.") flags.StringVar(&cliCmd.HistoryPath, "history-path", cliCmd.HistoryPath, "path for history files.") flags.StringVar(&cliCmd.OrganizationID, "org-id", cliCmd.OrganizationID, "OrganizationID.") - flags.StringVar(&cliCmd.DatabaseID, "db-id", cliCmd.DatabaseID, "DatabaseID.") + flags.StringVar(&cliCmd.Database, "db", cliCmd.Database, "Name of the database to connect to.") flags.StringVar(&cliCmd.ClientID, "client-id", cliCmd.ClientID, "Cognito Client ID for FeatureBase Cloud access.") flags.StringVar(&cliCmd.Region, "region", cliCmd.Region, "Cloud region for FeatureBase Cloud access (e.g. us-east-2).") diff --git a/dax/controller/schemar/boltdb/schemar.go b/dax/controller/schemar/boltdb/schemar.go index 43e4e8760..997a66cf9 100644 --- a/dax/controller/schemar/boltdb/schemar.go +++ b/dax/controller/schemar/boltdb/schemar.go @@ -694,5 +694,5 @@ func keyQualifiedDatabaseID(key []byte) (dax.QualifiedDatabaseID, error) { } func timestamp() int64 { - return time.Now().UnixNano() + return time.Now().Unix() } diff --git a/dax/queryer/client/client.go b/dax/queryer/client/client.go index 0e9dec772..e8fa8be91 100644 --- a/dax/queryer/client/client.go +++ b/dax/queryer/client/client.go @@ -2,16 +2,15 @@ package client import ( - "bytes" "context" "encoding/json" "fmt" "io" "net/http" + "time" featurebase "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/dax" - queryerhttp "github.com/featurebasedb/featurebase/v3/dax/queryer/http" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" ) @@ -49,67 +48,29 @@ func (c *Client) Health() bool { return true } -func (c *Client) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sql string) (*featurebase.WireQueryResponse, error) { - url := fmt.Sprintf("%s/sql", c.address.WithScheme(defaultScheme)) - - req := &queryerhttp.SQLRequest{ - OrganizationID: qdbid.OrganizationID, - DatabaseID: qdbid.DatabaseID, - SQL: sql, +func (c *Client) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sql io.Reader) (*featurebase.WireQueryResponse, error) { + url := fmt.Sprintf("%s/databases/%s/sql", c.address.WithScheme(defaultScheme), qdbid.DatabaseID) + if qdbid.DatabaseID == "" { + url = fmt.Sprintf("%s/sql", c.address.WithScheme(defaultScheme)) } - // Encode the request. - postBody, err := json.Marshal(req) - if err != nil { - return nil, errors.Wrap(err, "marshalling post request") + client := &http.Client{ + Timeout: time.Second * 30, } - responseBody := bytes.NewBuffer(postBody) // Post the request. c.logger.Debugf("POST query sql request: url: %s", url) - resp, err := http.Post(url, "application/json", responseBody) + req, err := http.NewRequest(http.MethodPost, url, sql) if err != nil { - return nil, errors.Wrap(err, "posting query sql request") + return nil, errors.Wrap(err, "creating new post request") + } + req.Header.Add("Content-Type", "text/plain") + req.Header.Add("OrganizationID", string(qdbid.OrganizationID)) + + var resp *http.Response + if resp, err = client.Do(req); err != nil { + return nil, errors.Wrap(err, "executing post request") } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - b, _ := io.ReadAll(resp.Body) - return nil, errors.Errorf("status code: %d: %s", resp.StatusCode, b) - } - - var wireResp *featurebase.WireQueryResponse - if err := json.NewDecoder(resp.Body).Decode(&wireResp); err != nil { - return nil, errors.Wrap(err, "reading response body") - } - - return wireResp, nil -} - -func (c *Client) QueryPQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, table dax.TableName, pql string) (*featurebase.WireQueryResponse, error) { - url := fmt.Sprintf("%s/query", c.address.WithScheme(defaultScheme)) - - req := &queryerhttp.QueryRequest{ - OrganizationID: qdbid.OrganizationID, - DatabaseID: qdbid.DatabaseID, - Table: table, - PQL: pql, - } - - // Encode the request. - postBody, err := json.Marshal(req) - if err != nil { - return nil, errors.Wrap(err, "marshalling post request") - } - responseBody := bytes.NewBuffer(postBody) - - // Post the request. - c.logger.Debugf("POST query pql request: url: %s", url) - resp, err := http.Post(url, "application/json", responseBody) - if err != nil { - return nil, errors.Wrap(err, "posting query pql request") - } - defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) diff --git a/dax/queryer/http/handler.go b/dax/queryer/http/handler.go index 03f437cc0..fac16a6dd 100644 --- a/dax/queryer/http/handler.go +++ b/dax/queryer/http/handler.go @@ -2,7 +2,9 @@ package http import ( "encoding/json" + "fmt" "net/http" + "strings" "github.com/featurebasedb/featurebase/v3/dax" "github.com/featurebasedb/featurebase/v3/dax/queryer" @@ -16,11 +18,8 @@ func Handler(q *queryer.Queryer) http.Handler { router := mux.NewRouter() router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth") - router.HandleFunc("/query", svr.postQuery).Methods("POST").Name("PostQuery") - - // /sql is a subset of /query, added here to provide an easy integration - // with the FeatureBase cli tool. router.HandleFunc("/sql", svr.postSQL).Methods("POST").Name("PostSQL") + router.HandleFunc("/databases/{databaseID}/sql", svr.postSQL).Methods("POST").Name("PostDatabaseSQL") return router } @@ -34,76 +33,67 @@ func (s *server) getHealth(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// POST /query -func (s *server) postQuery(w http.ResponseWriter, r *http.Request) { - body := r.Body - defer body.Close() - - req := QueryRequest{} - if err := json.NewDecoder(body).Decode(&req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - ctx := r.Context() - - var resp interface{} - var err error - qdbid := dax.NewQualifiedDatabaseID(req.OrganizationID, req.DatabaseID) - if req.SQL != "" { - resp, err = s.queryer.QuerySQL(ctx, qdbid, req.SQL) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - } else { - resp, err = s.queryer.QueryPQL(ctx, qdbid, req.Table, req.PQL) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - } - - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(resp); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } -} - // POST /sql func (s *server) postSQL(w http.ResponseWriter, r *http.Request) { - body := r.Body - defer body.Close() + orgID := getOrganizationID(r) + dbID := dax.DatabaseID(mux.Vars(r)["databaseID"]) - req := SQLRequest{} - if err := json.NewDecoder(body).Decode(&req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } + contentType := r.Header.Get("Content-Type") + switch contentType { + case "text/plain": + qdbid := dax.NewQualifiedDatabaseID(orgID, dbID) + resp, err := s.queryer.QuerySQL(r.Context(), qdbid, r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } - ctx := r.Context() + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } - qdbid := dax.NewQualifiedDatabaseID(req.OrganizationID, req.DatabaseID) - resp, err := s.queryer.QuerySQL(ctx, qdbid, req.SQL) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } + case "application/json": + body := r.Body + defer body.Close() - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(resp); err != nil { + req := SQLRequest{} + if err := json.NewDecoder(body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + ctx := r.Context() + if orgID == "" { + orgID = req.OrganizationID + } + if dbID == "" { + dbID = req.DatabaseID + } + + qdbid := dax.NewQualifiedDatabaseID(orgID, dbID) + resp, err := s.queryer.QuerySQL(ctx, qdbid, strings.NewReader(req.SQL)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + default: + err := fmt.Errorf("unsupported request content-type '%s'", contentType) http.Error(w, err.Error(), http.StatusBadRequest) return } } -type QueryRequest struct { - OrganizationID dax.OrganizationID `json:"org-id"` - DatabaseID dax.DatabaseID `json:"db-id"` - Table dax.TableName `json:"table-name"` - PQL string `json:"pql"` - SQL string `json:"sql"` +func getOrganizationID(r *http.Request) dax.OrganizationID { + return dax.OrganizationID(r.Header.Get("OrganizationID")) } type SQLRequest struct { @@ -111,5 +101,3 @@ type SQLRequest struct { DatabaseID dax.DatabaseID `json:"db-id"` SQL string `json:"sql"` } - -type QueryResponse interface{} diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 64e9e8dc5..205c10e9f 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -2,8 +2,10 @@ package queryer import ( + "bytes" "context" "fmt" + "io" "net/http" "strings" "sync" @@ -122,7 +124,7 @@ func (q *Queryer) Start() error { return nil } -func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sql string) (*featurebase.WireQueryResponse, error) { +func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sql io.Reader) (*featurebase.WireQueryResponse, error) { start := time.Now() ret := &featurebase.WireQueryResponse{} @@ -136,9 +138,31 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s applyExecutionTime() } + // Peek at the first character of sql. If it's "[", then handle this as PQL. + var isPQL bool + peekSize := 1 + peeker := io.LimitReader(sql, int64(peekSize)) + peek, err := io.ReadAll(peeker) + if err != nil { + return nil, errors.Wrap(err, "reading peeker") + } else if len(peek) == 1 && peek[0] == byte('[') { + isPQL = true + } + + // Since we already read peekSize bytes from r, we need to prepend that peek + // data to what's left to read from r. To do that we stitch them back + // together with an io.MultiReader. + peekReader := bytes.NewReader(peek) + multiReader := io.MultiReader(peekReader, sql) + // If PQL, run that instead. - if len(sql) > 0 && sql[0] == '[' { - if pqlResp, err := q.parseAndQueryPQL(ctx, qdbid, sql); err != nil { + if isPQL { + pql, err := io.ReadAll(multiReader) + if err != nil { + applyError(errors.Wrap(err, "reading pql")) + return ret, nil + } + if pqlResp, err := q.parseAndQueryPQL(ctx, qdbid, string(pql)); err != nil { applyError(errors.Wrap(err, "querying pql")) return ret, nil } else { @@ -158,7 +182,7 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s // put the requestId in the context ctx = fbcontext.WithRequestID(ctx, requestID.String()) - st, err := parser.NewParser(strings.NewReader(sql)).ParseStatement() + st, err := parser.NewParser(multiReader).ParseStatement() if err != nil { applyError(errors.Wrap(err, "parsing sql")) return ret, nil @@ -170,12 +194,17 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s // Importer imp := idkserverless.NewImporter(q.controller, qdbid, nil) - // TODO(tlt): We need a dax-compatible implementation of the SystemAPI. + // TODO(tlt): We need a serverless-compatible implementation of the + // SystemAPI. sysapi := &featurebase.NopSystemAPI{} systemLayer := systemlayer.NewSystemLayer() - pl := planner.NewExecutionPlanner(q.Orchestrator(qdbid), sapi, sysapi, systemLayer, imp, q.logger, sql) + // We intentionally don't pass the sql argument here because we're working + // with an io.Reader rather than a string and it's just not necessary to + // send it as a string to this method. Also, what happens if the sql is a + // large BULK INSERT? + pl := planner.NewExecutionPlanner(q.Orchestrator(qdbid), sapi, sysapi, systemLayer, imp, q.logger, "") planOp, err := pl.CompilePlan(ctx, st) if err != nil { diff --git a/dax/test/dax/dax_test.go b/dax/test/dax/dax_test.go index bcc305e71..d3536183f 100644 --- a/dax/test/dax/dax_test.go +++ b/dax/test/dax/dax_test.go @@ -7,6 +7,7 @@ import ( "log" "os" "sort" + "strings" "testing" "time" @@ -956,7 +957,7 @@ func runSQL(tb testing.TB, queryerAddr dax.Address, qdbid dax.QualifiedDatabaseI client := queryerclient.New(queryerAddr, logger.StderrLogger) - resp, err := client.QuerySQL(context.Background(), qdbid, sql) + resp, err := client.QuerySQL(context.Background(), qdbid, strings.NewReader(sql)) assert.NoError(tb, err) return resp @@ -967,7 +968,8 @@ func runPQL(tb testing.TB, queryerAddr dax.Address, qdbid dax.QualifiedDatabaseI client := queryerclient.New(queryerAddr, logger.StderrLogger) - resp, err := client.QueryPQL(context.Background(), qdbid, dax.TableName(table), pql) + sqlPQL := fmt.Sprintf("[%s]%s", table, pql) + resp, err := client.QuerySQL(context.Background(), qdbid, strings.NewReader(sqlPQL)) assert.NoError(tb, err) return resp diff --git a/sql3/planner/opalterdatabase.go b/sql3/planner/opalterdatabase.go index b6e1c9750..aaf92a2c2 100644 --- a/sql3/planner/opalterdatabase.go +++ b/sql3/planner/opalterdatabase.go @@ -91,7 +91,7 @@ func (i *alterDatabaseRowIter) Next(ctx context.Context) (types.Row, error) { switch v := i.option.(type) { case *parser.UnitsOption: e := v.Expr.(*parser.IntegerLit) - optName = "workers-min" + optName = dax.DatabaseOptionWorkersMin optValue = e.Value default: return nil, sql3.NewErrInvalidDatabaseOption(0, 0, i.option.String())