From 126be915a9c72427166d6717b6bc45dfb6bc42c9 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 6 Feb 2023 08:50:28 -0600 Subject: [PATCH] Support for setting individual DatabaseOptions (#2231) * Implement Schemar.SetDatabaseOption(option, value string) This replaces the temporary `SetDatabaseOptions()` method, which replaced the entire DatabaseOptions struct, with `SetDatabaseOption` which takes an option/value pair of strings to set. * Add SetDatabaseOption to controller http handler and client This commit also: - renames some `writeLog` to `writelog` - updates ApplyDirective to call resource.Unlock() on any resources being removed from the local worker * Add Database related methods to SchemaAPI interface Currently all implementations of this interface are implemented with "unimplemented" errors on those methods. Next will be to implement the necessary methods. * SQL: CREATE DATABASE and SHOW DATABASES * SQL: DROP DATABASE * SQL: Add UNITS option to CREATE DATABASE * SQL: ALTER DATABASE * User serverlessStorage.Remove[*]Resource instead of resource.Unlock() * Add WITH keyword to CREATE/ALTER DATABASE * fix some WITH logic * linter fixes * WITH on CREATE DATABASE is not required --- api.go | 8 + api_directive.go | 34 ++++ client/api.go | 20 ++ dax/controller/balancer/balancer_test.go | 16 +- dax/controller/client/client.go | 38 ++++ dax/controller/controller.go | 30 ++- dax/controller/controller_test.go | 12 +- dax/controller/http/handler.go | 27 +++ dax/controller/schemar/boltdb/schemar.go | 12 +- dax/controller/schemar/schemar.go | 4 +- dax/queryer/schema_api.go | 49 +++++ dax/schema.go | 7 +- dax/storage/storage.go | 22 +-- dax/table.go | 59 ++++++ dax/table_test.go | 30 +++ dax/test/dax/dax_test.go | 87 +++++++++ dax/writelogger/writelogger.go | 41 ++++- schema.go | 20 ++ server/server.go | 8 +- sql3/errors.go | 33 ++++ sql3/parser/ast.go | 159 +++++++++++++++- sql3/parser/ast_test.go | 43 +++++ sql3/parser/parser.go | 183 ++++++++++++++++++- sql3/parser/parser_test.go | 83 ++++++++- sql3/parser/token.go | 4 + sql3/planner/compilealterdatabase.go | 40 ++++ sql3/planner/compilealtertable.go | 1 + sql3/planner/compilecreatedatabase.go | 83 +++++++++ sql3/planner/compiledropdatabase.go | 27 +++ sql3/planner/compileshow.go | 59 ++++++ sql3/planner/executionplanner.go | 23 ++- sql3/planner/executionplannersystemtables.go | 42 +++-- sql3/planner/opalterdatabase.go | 107 +++++++++++ sql3/planner/opcreatedatabase.go | 112 ++++++++++++ sql3/planner/opdropdatabase.go | 84 +++++++++ sql3/planner/opfeaturebasedatabases.go | 135 ++++++++++++++ sql3/sql_complex_test.go | 26 +++ 37 files changed, 1690 insertions(+), 78 deletions(-) create mode 100644 sql3/planner/compilealterdatabase.go create mode 100644 sql3/planner/compilecreatedatabase.go create mode 100644 sql3/planner/compiledropdatabase.go create mode 100644 sql3/planner/opalterdatabase.go create mode 100644 sql3/planner/opcreatedatabase.go create mode 100644 sql3/planner/opdropdatabase.go create mode 100644 sql3/planner/opfeaturebasedatabases.go diff --git a/api.go b/api.go index 6fa786aa4..2b7bb9a74 100644 --- a/api.go +++ b/api.go @@ -3316,6 +3316,14 @@ func shardInShards(i dax.ShardNum, s dax.ShardNums) bool { } type SchemaAPI interface { + CreateDatabase(context.Context, *dax.Database) error + DropDatabase(context.Context, dax.DatabaseID) error + + DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) + DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) + SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error + Databases(context.Context, ...dax.DatabaseID) ([]*dax.Database, error) + TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) Tables(ctx context.Context) ([]*dax.Table, error) diff --git a/api_directive.go b/api_directive.go index 1dafbb680..b61df8f6e 100644 --- a/api_directive.go +++ b/api_directive.go @@ -324,6 +324,17 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT // Get the diff between from/to directive.partitions. partComp := newPartitionsComparer(fromD.TranslatePartitionsMap(), toPartitionsMap) + // Remove any partitions which are no longer assigned to this worker. + // TODO(tlt): currently, this is just removing the file lock on the + // resource; it's not actually removing the resource from the local + // computer. We should do that. + for tkey, partitions := range partComp.removed() { + qtid := tkey.QualifiedTableID() + for _, partition := range partitions { + api.serverlessStorage.RemoveTableKeyResource(qtid, partition) + } + } + // Loop over the partition map and load from Writelogger. for tkey, partitions := range partComp.added() { // Get index in order to find the translate stores (by partition) for @@ -411,6 +422,17 @@ func (api *API) pushJobsFieldKeys(ctx context.Context, jobs chan<- directiveJobT // Get the diff between from/to directive.fields. fieldComp := newFieldsComparer(fromD.TranslateFieldsMap(), toD.TranslateFieldsMap()) + // Remove any field keys which are no longer assigned to this worker. + // TODO(tlt): currently, this is just removing the file lock on the + // resource; it's not actually removing the resource from the local + // computer. We should do that. + for tkey, fields := range fieldComp.removed() { + qtid := tkey.QualifiedTableID() + for _, field := range fields { + api.serverlessStorage.RemoveFieldKeyResource(qtid, field) + } + } + // Loop over the field map and load from Writelogger. for tkey, fields := range fieldComp.added() { for _, field := range fields { @@ -495,6 +517,18 @@ func (api *API) pushJobsShards(ctx context.Context, jobs chan<- directiveJobType // Get the diff between from/to directive shards. shardComp := newShardsComparer(fromD.ComputeShardsMap(), shardMap) + // Remove any shards which are no longer assigned to this worker. + // TODO(tlt): currently, this is just removing the file lock on the + // resource; it's not actually removing the resource from the local + // computer. We should do that. + for tkey, shards := range shardComp.removed() { + qtid := tkey.QualifiedTableID() + for _, shard := range shards { + partition := dax.PartitionNum(disco.ShardToShardPartition(string(tkey), uint64(shard), disco.DefaultPartitionN)) + api.serverlessStorage.RemoveShardResource(qtid, partition, shard) + } + } + // Loop over the shard map and load from Writelogger. for tkey, shards := range shardComp.added() { for _, shard := range shards { diff --git a/client/api.go b/client/api.go index 6936df593..574dff77e 100644 --- a/client/api.go +++ b/client/api.go @@ -26,6 +26,26 @@ func NewSchemaAPI(c *Client) *schemaAPI { } } +func (s *schemaAPI) CreateDatabase(context.Context, *dax.Database) error { + return errors.Errorf("unimplemented: schemaAPI.CreateDatabase()") +} +func (s *schemaAPI) DropDatabase(context.Context, dax.DatabaseID) error { + return errors.Errorf("unimplemented: schemaAPI.DropDatabase()") +} + +func (s *schemaAPI) DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) { + return nil, errors.Errorf("unimplemented: schemaAPI.DatabaseByName()") +} +func (s *schemaAPI) DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) { + return nil, errors.Errorf("unimplemented: schemaAPI.DatabaseByID()") +} +func (s *schemaAPI) SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error { + return nil +} +func (s *schemaAPI) Databases(context.Context, ...dax.DatabaseID) ([]*dax.Database, error) { + return nil, errors.Errorf("unimplemented: schemaAPI.Databases()") +} + func (s *schemaAPI) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { return nil, errors.New(errors.ErrUncoded, "schemaAPI.TableByName not implemented") } diff --git a/dax/controller/balancer/balancer_test.go b/dax/controller/balancer/balancer_test.go index 7bfc78650..51078c901 100644 --- a/dax/controller/balancer/balancer_test.go +++ b/dax/controller/balancer/balancer_test.go @@ -398,9 +398,7 @@ func TestBalancer(t *testing.T) { // are updating the database option first, and then adding a worker to // satisfy those options. t.Run(fmt.Sprintf("test-%s", "set database options db min 3"), func(t *testing.T) { - dbOptions.WorkersMin = 3 - dbOptions.WorkersMax = 3 - assert.NoError(t, schemar.SetDatabaseOptions(tx, qdb.QualifiedID(), dbOptions)) + assert.NoError(t, schemar.SetDatabaseOption(tx, qdb.QualifiedID(), dax.DatabaseOptionWorkersMin, "3")) }) runTestPart(testPart{ @@ -675,9 +673,7 @@ func TestBalancer(t *testing.T) { // Set WorkersMin back to 2 so we can test the change to 3 again. t.Run(fmt.Sprintf("test-%s", "set database options db min back to 2"), func(t *testing.T) { - dbOptions.WorkersMin = 2 - dbOptions.WorkersMax = 2 - assert.NoError(t, schemar.SetDatabaseOptions(tx, qdb.QualifiedID(), dbOptions)) + assert.NoError(t, schemar.SetDatabaseOption(tx, qdb.QualifiedID(), dax.DatabaseOptionWorkersMin, "2")) }) runTestPart(testPart{ @@ -705,9 +701,7 @@ func TestBalancer(t *testing.T) { // have added a worker which will satisfy this option, and then updated // the option. t.Run(fmt.Sprintf("test-%s", "set database options db min back to 3"), func(t *testing.T) { - dbOptions.WorkersMin = 3 - dbOptions.WorkersMax = 3 - assert.NoError(t, schemar.SetDatabaseOptions(tx, qdb.QualifiedID(), dbOptions)) + assert.NoError(t, schemar.SetDatabaseOption(tx, qdb.QualifiedID(), dax.DatabaseOptionWorkersMin, "3")) }) // This implies that there is a condition where the database does not @@ -1052,9 +1046,7 @@ func TestBalancer(t *testing.T) { // Update database options on schemar so min worker for db is 3. t.Run(fmt.Sprintf("test-%s", "set database options db min 3"), func(t *testing.T) { - dbOptions.WorkersMin = 3 - dbOptions.WorkersMax = 3 - assert.NoError(t, schemar.SetDatabaseOptions(tx, qdb.QualifiedID(), dbOptions)) + assert.NoError(t, schemar.SetDatabaseOption(tx, qdb.QualifiedID(), dax.DatabaseOptionWorkersMin, "3")) }) // Now, add a worker and confirm that it has received some jobs. diff --git a/dax/controller/client/client.go b/dax/controller/client/client.go index 055007a27..b723b6fbc 100644 --- a/dax/controller/client/client.go +++ b/dax/controller/client/client.go @@ -206,6 +206,44 @@ func (c *Client) Databases(ctx context.Context, orgID dax.OrganizationID, ids .. return qdbs, nil } +func (c *Client) SetDatabaseOption(ctx context.Context, qdbid dax.QualifiedDatabaseID, option string, value string) error { + url := fmt.Sprintf("%s/database/options", c.address.WithScheme(defaultScheme)) + + req := &controllerhttp.DatabaseOptionRequest{ + QualifiedDatabaseID: qdbid, + Option: option, + Value: value, + } + + // Encode the request. + postBody, err := json.Marshal(req) + if err != nil { + return errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + request, err := http.NewRequest(http.MethodPatch, url, responseBody) + if err != nil { + return errors.Wrap(err, "creating http request") + } + request.Header.Set("Content-Type", "application/json") + + // Post the request as PATCH. + c.logger.Debugf("PATCH database/option request: url: %s", url) + resp, err := http.DefaultClient.Do(request) + if err != nil { + return errors.Wrap(err, "posting database/option request") + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + b, _ := io.ReadAll(resp.Body) + return errors.Errorf("status code: %d: %s", resp.StatusCode, b) + } + + return nil +} + // TODO(tlt): collapse Table into this func (c *Client) TableByID(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { return c.Table(ctx, qtid) diff --git a/dax/controller/controller.go b/dax/controller/controller.go index 5be60d9aa..1046b2f93 100644 --- a/dax/controller/controller.go +++ b/dax/controller/controller.go @@ -582,6 +582,11 @@ func (c *Controller) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatab } defer tx.Rollback() + // Create Database ID. + if _, err := qdb.CreateID(); err != nil { + return errors.Wrap(err, "creating database ID") + } + if err := c.Schemar.CreateDatabase(tx, qdb); err != nil { return errors.Wrap(err, "creating database in schemar") } @@ -664,16 +669,33 @@ func (c *Controller) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDataba return qdb, nil } -// SetDatabaseOptions sets the options on the given database. -func (c *Controller) SetDatabaseOptions(ctx context.Context, qdbid dax.QualifiedDatabaseID, opts dax.DatabaseOptions) error { +// SetDatabaseOption sets the option on the given database. +func (c *Controller) SetDatabaseOption(ctx context.Context, qdbid dax.QualifiedDatabaseID, option string, value string) error { tx, err := c.BoltDB.BeginTx(ctx, true) if err != nil { return errors.Wrap(err, "beginning tx") } defer tx.Rollback() - if err := c.Schemar.SetDatabaseOptions(tx, qdbid, opts); err != nil { - return errors.Wrap(err, "setting database options") + if err := c.Schemar.SetDatabaseOption(tx, qdbid, option, value); err != nil { + return errors.Wrapf(err, "setting database option: %s", option) + } + + diffs, err := c.Balancer.BalanceDatabase(tx, qdbid) + if err != nil { + return errors.Wrapf(err, "balancing database: %s", qdbid) + } + + workerSet := NewAddressSet() + for _, diff := range diffs { + workerSet.Add(dax.Address(diff.Address)) + } + + // Convert the slice of addresses into a slice of addressMethod containing + // the appropriate method. + addressMethods := applyAddressMethod(workerSet.SortedSlice(), dax.DirectiveMethodDiff) + if err := c.sendDirectives(tx, addressMethods...); err != nil { + return NewErrDirectiveSendFailure(err.Error()) } return tx.Commit() diff --git a/dax/controller/controller_test.go b/dax/controller/controller_test.go index 3fda6e218..e11af1c0e 100644 --- a/dax/controller/controller_test.go +++ b/dax/controller/controller_test.go @@ -154,9 +154,7 @@ func TestController(t *testing.T) { // Set WorkersMin to 3 so we can used the added nodes that follow. { - dbOptions.WorkersMin = 3 - dbOptions.WorkersMax = 3 - assert.NoError(t, con.SetDatabaseOptions(ctx, qdb1.QualifiedID(), dbOptions)) + assert.NoError(t, con.SetDatabaseOption(ctx, qdb1.QualifiedID(), dax.DatabaseOptionWorkersMin, "3")) } // Register two more nodes. @@ -459,9 +457,7 @@ func TestController(t *testing.T) { // Set WorkersMin to 1 so we can add a single node and have it be used // (currently just adding 1 node won't satisfy the minimum of 3). { - dbOptions.WorkersMin = 1 - dbOptions.WorkersMax = 1 - assert.NoError(t, con.SetDatabaseOptions(ctx, qdb1.QualifiedID(), dbOptions)) + assert.NoError(t, con.SetDatabaseOption(ctx, qdb1.QualifiedID(), dax.DatabaseOptionWorkersMin, "1")) } // Add a new node and ensure that the free shards get assigned to it. @@ -674,9 +670,7 @@ func TestController(t *testing.T) { // Set WorkersMin to 3 so we can used the two added nodes that follow. { - dbOptions.WorkersMin = 3 - dbOptions.WorkersMax = 3 - assert.NoError(t, con.SetDatabaseOptions(ctx, qdb1.QualifiedID(), dbOptions)) + assert.NoError(t, con.SetDatabaseOption(ctx, qdb1.QualifiedID(), dax.DatabaseOptionWorkersMin, "3")) } // Register two more nodes. diff --git a/dax/controller/http/handler.go b/dax/controller/http/handler.go index cb07735bd..edc863a87 100644 --- a/dax/controller/http/handler.go +++ b/dax/controller/http/handler.go @@ -23,6 +23,7 @@ func Handler(c *controller.Controller) http.Handler { router.HandleFunc("/database-by-id", server.postDatabaseByID).Methods("POST").Name("PostDatabaseByID") router.HandleFunc("/database-by-name", server.postDatabaseByName).Methods("POST").Name("PostDatabaseByName") router.HandleFunc("/databases", server.postDatabases).Methods("POST").Name("PostDatabases") + router.HandleFunc("/database/options", server.patchDatabaseOptions).Methods("PATCH").Name("PatchDatabaseOptions") router.HandleFunc("/create-table", server.postCreateTable).Methods("POST").Name("PostCreateTable") router.HandleFunc("/drop-table", server.postDropTable).Methods("POST").Name("PostDropTable") @@ -194,6 +195,32 @@ type DatabasesRequest struct { // DatabaseNames dax.DatabaseNames `json:"database-names"` } +// handlePatchDatabaseOptions handles updates to database options. +func (s *server) patchDatabaseOptions(w http.ResponseWriter, r *http.Request) { + // Decode request. + var req DatabaseOptionRequest + dec := json.NewDecoder(r.Body) + dec.DisallowUnknownFields() + if err := dec.Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := s.controller.SetDatabaseOption(r.Context(), req.QualifiedDatabaseID, req.Option, req.Value); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } +} + +// DatabaseOptionRequest represents a change to a database option. The thinking +// is to only support changing one database option at a time to keep the +// implementation sane. At time of writing, only WorkersMin is supported. +type DatabaseOptionRequest struct { + QualifiedDatabaseID dax.QualifiedDatabaseID `json:"qdbid"` + Option string `json:"option"` + Value string `json:"value"` +} + // POST /create-table func (s *server) postCreateTable(w http.ResponseWriter, r *http.Request) { body := r.Body diff --git a/dax/controller/schemar/boltdb/schemar.go b/dax/controller/schemar/boltdb/schemar.go index 5e2fcc5ab..43e4e8760 100644 --- a/dax/controller/schemar/boltdb/schemar.go +++ b/dax/controller/schemar/boltdb/schemar.go @@ -204,9 +204,9 @@ func (s *Schemar) DropDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabaseID return nil } -// SetDatabaseOptions overwrites the existing database options with those -// provided for the given database. -func (s *Schemar) SetDatabaseOptions(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, opts dax.DatabaseOptions) error { +// SetDatabaseOption overwrites the existing database option with the provided +// value. +func (s *Schemar) SetDatabaseOption(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, option string, value string) error { txx, ok := tx.(*boltdb.Tx) if !ok { return dax.NewErrInvalidTransaction() @@ -218,8 +218,10 @@ func (s *Schemar) SetDatabaseOptions(tx dax.Transaction, qdbid dax.QualifiedData return errors.Wrapf(err, "getting database: %s", qdbid) } - // Set the new options. - qdb.Options = opts + // Set the new option. + if err := qdb.Options.Set(option, value); err != nil { + return errors.Wrapf(err, "setting option on database: %s", qdbid) + } // Put the database. if err := s.putDatabase(txx, qdb); err != nil { diff --git a/dax/controller/schemar/schemar.go b/dax/controller/schemar/schemar.go index 24e0d82fc..fc57d93c2 100644 --- a/dax/controller/schemar/schemar.go +++ b/dax/controller/schemar/schemar.go @@ -11,7 +11,7 @@ type Schemar interface { DatabaseByName(tx dax.Transaction, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) DatabaseByID(dax.Transaction, dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) - SetDatabaseOptions(dax.Transaction, dax.QualifiedDatabaseID, dax.DatabaseOptions) error + SetDatabaseOption(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, option string, value string) error // Databases returns a list of databases. If the list of DatabaseIDs is // empty, all databases will be returned. If greater than zero DatabaseIDs @@ -65,7 +65,7 @@ func (s *NopSchemar) DatabaseByID(dax.Transaction, dax.QualifiedDatabaseID) (*da return nil, nil } -func (s *NopSchemar) SetDatabaseOptions(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, opts dax.DatabaseOptions) error { +func (s *NopSchemar) SetDatabaseOption(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, option string, value string) error { return nil } diff --git a/dax/queryer/schema_api.go b/dax/queryer/schema_api.go index 52c638ec0..132e1a334 100644 --- a/dax/queryer/schema_api.go +++ b/dax/queryer/schema_api.go @@ -27,6 +27,55 @@ func newQualifiedSchemaAPI(qdbid dax.QualifiedDatabaseID, schema dax.Schemar) *q } } +func (s *qualifiedSchemaAPI) CreateDatabase(ctx context.Context, db *dax.Database) error { + qdb := dax.NewQualifiedDatabase(s.qdbid.OrganizationID, db) + return s.schemar.CreateDatabase(ctx, qdb) +} + +func (s *qualifiedSchemaAPI) DropDatabase(ctx context.Context, dbid dax.DatabaseID) error { + qdbid := dax.NewQualifiedDatabaseID(s.qdbid.OrganizationID, dbid) + return s.schemar.DropDatabase(ctx, qdbid) +} + +func (s *qualifiedSchemaAPI) DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) { + orgID := s.qdbid.OrganizationID + qdb, err := s.schemar.DatabaseByName(ctx, orgID, dbname) + if err != nil { + return nil, errors.Wrapf(err, "getting database by name: (%s) %s", orgID, dbname) + } + return &qdb.Database, nil +} + +func (s *qualifiedSchemaAPI) DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) { + qdbid := dax.NewQualifiedDatabaseID(s.qdbid.OrganizationID, dbid) + + qdb, err := s.schemar.DatabaseByID(ctx, qdbid) + if err != nil { + return nil, errors.Wrapf(err, "getting database: %s", qdbid) + } + + return &qdb.Database, nil +} + +func (s *qualifiedSchemaAPI) SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error { + qdbid := dax.NewQualifiedDatabaseID(s.qdbid.OrganizationID, dbid) + return s.schemar.SetDatabaseOption(ctx, qdbid, option, value) +} + +func (s *qualifiedSchemaAPI) Databases(ctx context.Context, dbids ...dax.DatabaseID) ([]*dax.Database, error) { + qdbs, err := s.schemar.Databases(ctx, s.qdbid.OrganizationID, dbids...) + if err != nil { + return nil, errors.Wrap(err, "getting databases") + } + + dbs := make([]*dax.Database, 0, len(qdbs)) + for _, qdbl := range qdbs { + dbs = append(dbs, &qdbl.Database) + } + + return dbs, nil +} + func (s *qualifiedSchemaAPI) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { qtbl, err := s.schemar.TableByName(ctx, s.qdbid, tname) if err != nil { diff --git a/dax/schema.go b/dax/schema.go index 205d325f9..a9d1919f1 100644 --- a/dax/schema.go +++ b/dax/schema.go @@ -19,14 +19,14 @@ type Schemar interface { DatabaseByName(ctx context.Context, orgID OrganizationID, dbname DatabaseName) (*QualifiedDatabase, error) DatabaseByID(ctx context.Context, qdbid QualifiedDatabaseID) (*QualifiedDatabase, error) + SetDatabaseOption(ctx context.Context, qdbid QualifiedDatabaseID, option string, value string) error + // Databases returns a list of databases. If the list of DatabaseIDs is // empty, all databases will be returned. If greater than zero DatabaseIDs // are passed in the second argument, only databases matching those IDs will // be returned. Databases(context.Context, OrganizationID, ...DatabaseID) ([]*QualifiedDatabase, error) - // SetDatabaseOptions(context.Context, QualifiedDatabaseID, DatabaseOptions) error - ////////////////////////////////////////////////////////////////////////// // Table methods ////////////////////////////////////////////////////////////////////////// @@ -76,6 +76,9 @@ func (s *NopSchemar) DatabaseByName(ctx context.Context, orgID OrganizationID, d func (s *NopSchemar) DatabaseByID(ctx context.Context, qdbid QualifiedDatabaseID) (*QualifiedDatabase, error) { return nil, nil } +func (s *NopSchemar) SetDatabaseOption(ctx context.Context, qdbid QualifiedDatabaseID, option string, value string) error { + return nil +} func (s *NopSchemar) Databases(context.Context, OrganizationID, ...DatabaseID) ([]*QualifiedDatabase, error) { return nil, nil } diff --git a/dax/storage/storage.go b/dax/storage/storage.go index 428a3cc82..d64c5e13e 100644 --- a/dax/storage/storage.go +++ b/dax/storage/storage.go @@ -66,7 +66,7 @@ func (mm *ResourceManager) GetShardResource(qtid dax.QualifiedTableID, partition } mm.shardResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.Writelogger, + writelogger: mm.Writelogger, bucket: partitionBucket(qtid.Key(), partition), key: shardKey(shard), log: mm.Logger, @@ -97,7 +97,7 @@ func (mm *ResourceManager) GetTableKeyResource(qtid dax.QualifiedTableID, partit } mm.tableKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.Writelogger, + writelogger: mm.Writelogger, bucket: partitionBucket(qtid.Key(), partition), key: keysFileName, log: mm.Logger, @@ -127,7 +127,7 @@ func (mm *ResourceManager) GetFieldKeyResource(qtid dax.QualifiedTableID, field } mm.fieldKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.Writelogger, + writelogger: mm.Writelogger, bucket: fieldBucket(qtid.Key(), field), key: keysFileName, log: mm.Logger, @@ -227,7 +227,7 @@ func (mm *ResourceManager) RemoveTable(qtid dax.QualifiedTableID) error { // concurrently. type Resource struct { snapshotter computer.SnapshotService - writeLogger computer.WritelogService + writelogger computer.WritelogService bucket string key string @@ -298,7 +298,7 @@ func (m *Resource) LoadWriteLog() (data io.ReadCloser, err error) { if m.loadWLsPastVersion == -2 { return nil, errors.New(errors.ErrUncoded, "LoadWriteLog called in inconsistent state, can't tell what version to load from") } - wLogs, err := m.writeLogger.List(m.bucket, m.key) + wLogs, err := m.writelogger.List(m.bucket, m.key) if err != nil { return nil, errors.Wrap(err, "listing write logs") } @@ -332,7 +332,7 @@ func (m *Resource) LoadWriteLog() (data io.ReadCloser, err error) { m.latestWLVersion = versions[0] m.dirty = true - r, err := m.writeLogger.LogReaderFrom(m.bucket, m.key, versions[0], m.lastWLPos) + r, err := m.writelogger.LogReaderFrom(m.bucket, m.key, versions[0], m.lastWLPos) if err != nil { return nil, errors.Wrap(err, "getting writelog") } @@ -364,7 +364,7 @@ func (m *Resource) LoadWriteLog() (data io.ReadCloser, err error) { func (m *Resource) Lock() error { m.log.Debugf("Lock %s/%s", m.bucket, m.key) // lock is sort of arbitrarily on the write log interface - if err := m.writeLogger.Lock(m.bucket, m.key); err != nil { + if err := m.writelogger.Lock(m.bucket, m.key); err != nil { return errors.Wrap(err, "acquiring lock") } m.locked = true @@ -380,7 +380,7 @@ func (m *Resource) Append(msg []byte) error { return errors.New(errors.ErrUncoded, "can't call append before loading and locking write log") } m.dirty = true - return m.writeLogger.AppendMessage(m.bucket, m.key, m.latestWLVersion, msg) + return m.writelogger.AppendMessage(m.bucket, m.key, m.latestWLVersion, msg) } // IncrementWLVersion should be called during snapshotting with a @@ -416,7 +416,7 @@ func (m *Resource) Snapshot(rc io.ReadCloser) error { if err != nil { return errors.Wrap(err, "writing snapshot") } - err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1) + err = m.writelogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1) return errors.Wrap(err, "deleting old write log") } @@ -429,7 +429,7 @@ func (m *Resource) SnapshotTo(wt io.WriterTo) error { if err != nil { return errors.Wrap(err, "writing snapshot SnapshotTo") } - err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1) + err = m.writelogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1) return errors.Wrap(err, "deleting old write log snapshotTo") } @@ -444,7 +444,7 @@ func (m *Resource) Unlock() error { if !m.locked { return errors.New(errors.ErrUncoded, "resource was not locked") } - if err := m.writeLogger.Unlock(m.bucket, m.key); err != nil { + if err := m.writelogger.Unlock(m.bucket, m.key); err != nil { return errors.Wrap(err, "unlocking") } m.locked = false diff --git a/dax/table.go b/dax/table.go index 13c505273..91f065300 100644 --- a/dax/table.go +++ b/dax/table.go @@ -4,11 +4,13 @@ import ( "crypto/rand" "fmt" "regexp" + "strconv" "strings" "time" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/pql" + uuid "github.com/satori/go.uuid" ) //////////////////////////////////////////////////////////////////////////////// @@ -153,18 +155,75 @@ type Database struct { UpdatedBy string `json:"updatedBy,omitempty"` } +// CreateID generates a unique identifier for Database. If Database has already +// been assigned an ID, then this no-ops. The reason for this is that the cloud +// implementation of FeatureBase may allocate an ID before calling +// CreateDatabase on the controller. +func (d *Database) CreateID() (DatabaseID, error) { + if d.ID != "" { + return d.ID, nil + } + + id, err := uuid.NewV4() + if err != nil { + return "", errors.Wrap(err, "generating uuid") + } + + d.ID = DatabaseID(id.String()) + + return d.ID, nil +} + // DatabaseOptions are used to configure a database. type DatabaseOptions struct { WorkersMin int `json:"workers-min"` WorkersMax int `json:"workers-max"` } +// DatabaseOption is a string key representing a database option. +type DatabaseOption string + +const ( + DatabaseOptionWorkersMin = "workers-min" + DatabaseOptionWorkersMax = "workers-max" +) + +// Set sets the specified option to the provided value. +func (opts *DatabaseOptions) Set(option string, value string) error { + opt := strings.ToLower(option) + switch opt { + case DatabaseOptionWorkersMin: + min, err := strconv.Atoi(value) + if err != nil { + return errors.Wrapf(err, "converting value to int: %s", value) + } + opts.WorkersMin = min + // We don't currently expose WorkersMax because we aren't yet detecting + // how to scale between a range, so for now we just keep it set to the + // same value as WorkersMin. + opts.WorkersMax = min + default: + return errors.Errorf("unsupported database option: %s", option) + } + + return nil +} + // QualifiedDatabase is a Database along with its OrganizationID. type QualifiedDatabase struct { OrganizationID OrganizationID `json:"org-id"` Database } +// NewQualifiedDatabase returns the db as a QualifiedDatabase with the provided +// OrganizationID. +func NewQualifiedDatabase(orgID OrganizationID, db *Database) *QualifiedDatabase { + return &QualifiedDatabase{ + OrganizationID: orgID, + Database: *db, + } +} + type QualifiedDatabases []*QualifiedDatabase // Key returns the string-encoded (delimited by DatabaseKeyDelimiter) globally diff --git a/dax/table_test.go b/dax/table_test.go index 5706ccf9d..7947a8735 100644 --- a/dax/table_test.go +++ b/dax/table_test.go @@ -373,3 +373,33 @@ func TestTable(t *testing.T) { }) }) } + +func TestDatabase(t *testing.T) { + databaseName := dax.DatabaseName("db1") + + t.Run("Options", func(t *testing.T) { + { + db := &dax.Database{ + Name: databaseName, + } + assert.Zero(t, db.Options.WorkersMin) + assert.Zero(t, db.Options.WorkersMax) + + // Set WorkersMin to 5. + assert.NoError(t, db.Options.Set(dax.DatabaseOptionWorkersMin, "5")) + assert.Equal(t, 5, db.Options.WorkersMin) + assert.Equal(t, 5, db.Options.WorkersMax) + + // Set WorkersMin back to 0. + assert.NoError(t, db.Options.Set(dax.DatabaseOptionWorkersMin, "0")) + assert.Zero(t, db.Options.WorkersMin) + assert.Zero(t, db.Options.WorkersMax) + + // Try setting WorkersMin to an invalid value. + assert.Error(t, db.Options.Set(dax.DatabaseOptionWorkersMin, "abc")) + + // Try setting an unsupported option. + assert.Error(t, db.Options.Set("invalid-option", "")) + } + }) +} diff --git a/dax/test/dax/dax_test.go b/dax/test/dax/dax_test.go index 62fa4af13..bcc305e71 100644 --- a/dax/test/dax/dax_test.go +++ b/dax/test/dax/dax_test.go @@ -618,6 +618,93 @@ func TestDAXIntegration(t *testing.T) { assert.False(t, dirIsEmpty(t, rootDir+"/controller")) assert.True(t, dirIsEmpty(t, rootDir+"/wl")) }) + + t.Run("DatabaseOptions", func(t *testing.T) { + cfg := test.DefaultConfig() + cfg.Computer.N = 4 + opt := server.OptCommandConfig(cfg) + mc := test.MustRunManagedCommand(t, opt) + + svcmgr := mc.Manage() + + // Set up Controller client. + controllerClient := controllerclient.New(svcmgr.Controller.Address(), svcmgr.Logger) + + // Create database. + qdb.Options.WorkersMin = 2 + qdb.Options.WorkersMax = 2 + assert.NoError(t, controllerClient.CreateDatabase(context.Background(), qdb)) + + computers := svcmgr.Computers() + computerKey0 := dax.ServiceKey(dax.ServicePrefixComputer + "0") + computerKey1 := dax.ServiceKey(dax.ServicePrefixComputer + "1") + computerKey2 := dax.ServiceKey(dax.ServicePrefixComputer + "2") + // computerKey3 := dax.ServiceKey(dax.ServicePrefixComputer + "3") + + // Ingest and query some data. + runTableTests(t, + svcmgr.Queryer.Address(), + basicTableTestConfig(qdbid, defs.Keyed)..., + ) + + qtid, err := controllerClient.TableID(context.Background(), qdbid, dax.TableName(defs.Keyed.Name(0))) + assert.NoError(t, err) + + // ensure partitions are covered + partitions0 := dax.PartitionNums{0, 2, 4, 6, 8, 10} + partitions1 := dax.PartitionNums{1, 3, 5, 7, 9, 11} + allPartitions := append(partitions0, partitions1...) + sort.Sort(allPartitions) + + nodes, err := controllerClient.TranslateNodes(context.Background(), qtid, allPartitions...) + assert.NoError(t, err) + if assert.Len(t, nodes, 2) { + // computer0 (node0) + assert.Equal(t, computers[computerKey0].Address(), nodes[0].Address) + assert.Equal(t, partitions0, nodes[0].Partitions) + // computer1 (node1) + assert.Equal(t, computers[computerKey1].Address(), nodes[1].Address) + assert.Equal(t, partitions1, nodes[1].Partitions) + } + + // Change DatabaseOptions.WorkersMin to 3. + assert.NoError(t, controllerClient.SetDatabaseOption(context.Background(), qdbid, dax.DatabaseOptionWorkersMin, "3")) + + // Query the same data. + t.Run("query the same data", func(t *testing.T) { + runTableTests(t, + svcmgr.Queryer.Address(), + tableTestConfig{ + qdbid: qdbid, + test: defs.Keyed, + skipCreate: true, + skipInsert: true, + querySet: 0, + }, + ) + }) + + // ensure partitions are still covered + partitions0 = dax.PartitionNums{0, 10} + partitions1 = dax.PartitionNums{1, 11} + partitions2 := dax.PartitionNums{2, 3, 4, 5, 6, 7, 8, 9} + allPartitions = append(append(partitions0, partitions1...), partitions2...) + sort.Sort(allPartitions) + + nodes, err = controllerClient.TranslateNodes(context.Background(), qtid, allPartitions...) + assert.NoError(t, err) + if assert.Len(t, nodes, 3) { + // computer0 (node0) + assert.Equal(t, computers[computerKey0].Address(), nodes[0].Address) + assert.Equal(t, partitions0, nodes[0].Partitions) + // computer1 (node1) + assert.Equal(t, computers[computerKey1].Address(), nodes[1].Address) + assert.Equal(t, partitions1, nodes[1].Partitions) + // computer2 (node2) + assert.Equal(t, computers[computerKey2].Address(), nodes[2].Address) + assert.Equal(t, partitions2, nodes[2].Partitions) + } + }) } func dirIsEmpty(t *testing.T, name string) bool { diff --git a/dax/writelogger/writelogger.go b/dax/writelogger/writelogger.go index b974f8a17..15fca771b 100644 --- a/dax/writelogger/writelogger.go +++ b/dax/writelogger/writelogger.go @@ -11,6 +11,7 @@ import ( "strings" "sync" "syscall" + "time" "github.com/featurebasedb/featurebase/v3/dax" "github.com/featurebasedb/featurebase/v3/dax/computer" @@ -138,10 +139,24 @@ func (w *Writelogger) Lock(bucket, key string) error { return errors.Wrapf(err, "lock dir %s", lockDir) } - f, err := os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL|syscall.O_NONBLOCK, 0644) - if err != nil { + // We introduced this retry logic when we thought that a node which was + // having its partition ownership removed (in place of this node) was still + // holding the lock. While that still may be the case, it's more likely that + // the problem we were seeing was that we weren't actually calling + // `resource.Unlock()` in ApplyDirective for resources being removed. With + // that said, it doesn't hurt to leave this retry logic here. + var f *os.File + var err error + if err := w.retryUntil(10*time.Second, func() error { + f, err = os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL|syscall.O_NONBLOCK, 0644) + if err != nil { + return errors.Wrapf(err, "opening lock file: %s", lockFile) + } + return nil + }); err != nil { return errors.Wrapf(err, "opening lock file: %s", lockFile) } + w.mu.Lock() defer w.mu.Unlock() w.lockFiles[lockFile] = f @@ -154,7 +169,29 @@ func (w *Writelogger) Lock(bucket, key string) error { // Type: syscall.F_WRLCK, // }) return nil +} +// retryUntil repeatedly executes fn until it returns nil or timeout occurs. +func (w *Writelogger) retryUntil(timeout time.Duration, fn func() error) (err error) { + timer := time.NewTimer(timeout) + defer timer.Stop() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + var i int + for { + if err = fn(); err == nil { + return nil + } + i++ + w.logger.Debugf("Writelogger retryUntil try: %d", i) + + select { + case <-timer.C: + return err + case <-ticker.C: + } + } } func (w *Writelogger) Unlock(bucket, key string) error { diff --git a/schema.go b/schema.go index 36fa67613..60d7897ba 100644 --- a/schema.go +++ b/schema.go @@ -25,6 +25,26 @@ func NewOnPremSchema(api *API) *onPremSchema { } } +func (s *onPremSchema) CreateDatabase(context.Context, *dax.Database) error { + return errors.Errorf("unimplemented: onPremSchema.CreateDatabase()") +} +func (s *onPremSchema) DropDatabase(context.Context, dax.DatabaseID) error { + return errors.Errorf("unimplemented: onPremSchema.DropDatabase()") +} + +func (s *onPremSchema) DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) { + return nil, errors.Errorf("unimplemented: onPremSchema.DatabaseByName()") +} +func (s *onPremSchema) DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) { + return nil, errors.Errorf("unimplemented: onPremSchema.DatabaseByID()") +} +func (s *onPremSchema) SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error { + return nil +} +func (s *onPremSchema) Databases(context.Context, ...dax.DatabaseID) ([]*dax.Database, error) { + return []*dax.Database{}, nil +} + func (s *onPremSchema) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { idx, err := s.api.IndexInfo(context.Background(), string(tname)) if err != nil { diff --git a/server/server.go b/server/server.go index d9ecc9ea4..0ebdb513d 100644 --- a/server/server.go +++ b/server/server.go @@ -80,7 +80,7 @@ type Command struct { Registrar computer.Registrar serverlessStorage *storage.ResourceManager - writeLogService computer.WritelogService + writelogService computer.WritelogService snapshotService computer.SnapshotService Handler pilosa.HandlerI @@ -153,7 +153,7 @@ func OptCommandSetConfig(config *Config) CommandOption { func OptCommandInjections(inj Injections) CommandOption { return func(c *Command) error { if inj.Writelogger != nil { - c.writeLogService = inj.Writelogger + c.writelogService = inj.Writelogger } if inj.Snapshotter != nil { c.snapshotService = inj.Snapshotter @@ -556,8 +556,8 @@ func (m *Command) setupServer() error { m.Config.Etcd.Dir = filepath.Join(path, pilosa.DiscoDir) } - if m.writeLogService != nil && m.snapshotService != nil { - m.serverlessStorage = storage.NewResourceManager(m.snapshotService, m.writeLogService, m.logger) + if m.writelogService != nil && m.snapshotService != nil { + m.serverlessStorage = storage.NewResourceManager(m.snapshotService, m.writelogService, m.logger) } executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner { diff --git a/sql3/errors.go b/sql3/errors.go index 895dc0139..ea7e506d3 100644 --- a/sql3/errors.go +++ b/sql3/errors.go @@ -71,6 +71,11 @@ const ( ErrInsertMustHaveIDColumn errors.Code = "ErrInsertMustHaveIDColumn" ErrInsertMustAtLeastOneNonIDColumn errors.Code = "ErrInsertMustAtLeastOneNonIDColumn" + ErrDatabaseNotFound errors.Code = "ErrDatabaseNotFound" + ErrDatabaseExists errors.Code = "ErrDatabaseExists" + ErrInvalidDatabaseOption errors.Code = "ErrInvalidDatabaseOption" + ErrInvalidUnitsValue errors.Code = "ErrInvalidUnitsValue" + ErrTableMustHaveIDColumn errors.Code = "ErrTableMustHaveIDColumn" ErrTableIDColumnType errors.Code = "ErrTableIDColumnType" ErrTableIDColumnConstraints errors.Code = "ErrTableIDColumnConstraints" @@ -498,6 +503,13 @@ func NewErrInsertMustAtLeastOneNonIDColumn(line int, col int) error { ) } +func NewErrDatabaseExists(line, col int, databaseName string) error { + return errors.New( + ErrDatabaseExists, + fmt.Sprintf("[%d:%d] database '%s' already exists", line, col, databaseName), + ) +} + func NewErrTableMustHaveIDColumn(line, col int) error { return errors.New( ErrTableMustHaveIDColumn, @@ -526,6 +538,13 @@ func NewErrTableIDColumnAlter(line, col int) error { ) } +func NewErrDatabaseNotFound(line, col int, databaseName string) error { + return errors.New( + ErrDatabaseNotFound, + fmt.Sprintf("[%d:%d] database '%s' not found", line, col, databaseName), + ) +} + func NewErrTableNotFound(line, col int, tableName string) error { return errors.New( ErrTableNotFound, @@ -561,6 +580,20 @@ func NewErrTableColumnNotFound(line, col int, tableName string, columnName strin ) } +func NewErrInvalidDatabaseOption(line, col int, option string) error { + return errors.New( + ErrInvalidDatabaseOption, + fmt.Sprintf("[%d:%d] invalid database option '%s'", line, col, option), + ) +} + +func NewErrInvalidUnitsValue(line, col int, units int64) error { + return errors.New( + ErrInvalidUnitsValue, + fmt.Sprintf("[%d:%d] invalid value '%d' for units (should be a number between 0-10000)", line, col, units), + ) +} + func NewErrInvalidKeyPartitionsValue(line, col int, keypartitions int64) error { return errors.New( ErrInvalidKeyPartitionsValue, diff --git a/sql3/parser/ast.go b/sql3/parser/ast.go index 80bedc372..149b1b760 100644 --- a/sql3/parser/ast.go +++ b/sql3/parser/ast.go @@ -13,9 +13,12 @@ type Node interface { fmt.Stringer } +func (*AlterDatabaseStatement) node() {} func (*AlterTableStatement) node() {} +func (*AlterViewStatement) node() {} func (*AnalyzeStatement) node() {} func (*Assignment) node() {} +func (*ShowDatabasesStatement) node() {} func (*ShowTablesStatement) node() {} func (*ShowColumnsStatement) node() {} func (*ShowCreateTableStatement) node() {} @@ -32,14 +35,15 @@ func (*CastExpr) node() {} func (*CheckConstraint) node() {} func (*ColumnDefinition) node() {} func (*CommitStatement) node() {} +func (*CreateDatabaseStatement) node() {} func (*CreateIndexStatement) node() {} func (*CreateTableStatement) node() {} func (*CreateFunctionStatement) node() {} func (*CreateViewStatement) node() {} -func (*AlterViewStatement) node() {} func (*DateLit) node() {} func (*DefaultConstraint) node() {} func (*DeleteStatement) node() {} +func (*DropDatabaseStatement) node() {} func (*DropIndexStatement) node() {} func (*DropTableStatement) node() {} func (*DropFunctionStatement) node() {} @@ -87,6 +91,7 @@ func (*TupleLiteralExpr) node() {} func (*Type) node() {} func (*UnaryExpr) node() {} func (*UniqueConstraint) node() {} +func (*UnitsOption) node() {} func (*UpdateStatement) node() {} func (*UpsertClause) node() {} func (*UsingConstraint) node() {} @@ -100,20 +105,24 @@ type Statement interface { stmt() } +func (*AlterDatabaseStatement) stmt() {} func (*AlterTableStatement) stmt() {} +func (*AlterViewStatement) stmt() {} func (*AnalyzeStatement) stmt() {} func (*BeginStatement) stmt() {} func (*BulkInsertStatement) stmt() {} +func (*ShowDatabasesStatement) stmt() {} func (*ShowTablesStatement) stmt() {} func (*ShowColumnsStatement) stmt() {} func (*ShowCreateTableStatement) stmt() {} func (*CommitStatement) stmt() {} +func (*CreateDatabaseStatement) stmt() {} func (*CreateIndexStatement) stmt() {} func (*CreateTableStatement) stmt() {} func (*CreateFunctionStatement) stmt() {} func (*CreateViewStatement) stmt() {} -func (*AlterViewStatement) stmt() {} func (*DeleteStatement) stmt() {} +func (*DropDatabaseStatement) stmt() {} func (*DropIndexStatement) stmt() {} func (*DropTableStatement) stmt() {} func (*DropFunctionStatement) stmt() {} @@ -133,6 +142,8 @@ func CloneStatement(stmt Statement) Statement { } switch stmt := stmt.(type) { + case *AlterDatabaseStatement: + return stmt.Clone() case *AlterTableStatement: return stmt.Clone() case *AnalyzeStatement: @@ -141,6 +152,8 @@ func CloneStatement(stmt Statement) Statement { return stmt.Clone() case *CommitStatement: return stmt.Clone() + case *CreateDatabaseStatement: + return stmt.Clone() case *CreateIndexStatement: return stmt.Clone() case *CreateTableStatement: @@ -151,6 +164,8 @@ func CloneStatement(stmt Statement) Statement { return stmt.Clone() case *DeleteStatement: return stmt.Clone() + case *DropDatabaseStatement: + return stmt.Clone() case *DropIndexStatement: return stmt.Clone() case *DropTableStatement: @@ -470,6 +485,16 @@ func (s *ExplainStatement) String() string { return buf.String() } +type ShowDatabasesStatement struct { + Show Pos // position of SHOW + Databases Pos // position of DATABASES +} + +// String returns the string representation of the statement. +func (s *ShowDatabasesStatement) String() string { + return "SHOW DATABASES" +} + type ShowTablesStatement struct { Show Pos // position of SHOW Tables Pos // position of TABLES @@ -662,9 +687,53 @@ func (s *ReleaseStatement) String() string { return buf.String() } +type CreateDatabaseStatement struct { + Create Pos // position of CREATE keyword + Database Pos // position of DATABASE keyword + If Pos // position of IF keyword (optional) + IfNot Pos // position of NOT keyword (optional) + IfNotExists Pos // position of EXISTS keyword (optional) + Name *Ident // database name + + With Pos // position of WITH keyword + + Options []DatabaseOption // database options +} + +// Clone returns a deep copy of s. +func (s *CreateDatabaseStatement) Clone() *CreateDatabaseStatement { + if s == nil { + return s + } + other := *s + other.Name = s.Name.Clone() + return &other +} + +// String returns the string representation of the statement. +func (s *CreateDatabaseStatement) String() string { + var buf bytes.Buffer + buf.WriteString("CREATE DATABASE") + if s.IfNotExists.IsValid() { + buf.WriteString(" IF NOT EXISTS") + } + buf.WriteString(" ") + buf.WriteString(s.Name.String()) + + if s.With.IsValid() { + buf.WriteString(" WITH") + for _, opt := range s.Options { + buf.WriteString(" ") + buf.WriteString(opt.String()) + } + } + + return buf.String() +} + type CreateTableStatement struct { Create Pos // position of CREATE keyword - Table Pos // position of CREATE keyword + Table Pos // position of TABLE keyword If Pos // position of IF keyword (optional) IfNot Pos // position of NOT keyword (optional) IfNotExists Pos // position of EXISTS keyword (optional) @@ -766,6 +835,26 @@ func (c *ColumnDefinition) String() string { return buf.String() } +type DatabaseOption interface { + Node + dbOption() +} + +func (*UnitsOption) dbOption() {} +func (*CommentOption) dbOption() {} + +type UnitsOption struct { + Units Pos // position of UNITS keyword + Expr Expr // expression +} + +func (o *UnitsOption) String() string { + var buf bytes.Buffer + buf.WriteString("UNITS ") + buf.WriteString(o.Expr.String()) + return buf.String() +} + type TableOption interface { Node option() @@ -1357,6 +1446,39 @@ func (s *AnalyzeStatement) String() string { return fmt.Sprintf("ANALYZE %s", s.Name.String()) } +type AlterDatabaseStatement struct { + Alter Pos // position of ALTER keyword + Database Pos // position of DATABASE keyword + Name *Ident // database name + + With Pos // position of WITH keyword + + Option DatabaseOption +} + +// Clone returns a deep copy of s. +func (s *AlterDatabaseStatement) Clone() *AlterDatabaseStatement { + if s == nil { + return nil + } + other := *s + other.Name = other.Name.Clone() + return &other +} + +// String returns the string representation of the statement. +func (s *AlterDatabaseStatement) String() string { + var buf bytes.Buffer + buf.WriteString("ALTER DATABASE ") + buf.WriteString(s.Name.String()) + + if s.Option != nil { + buf.WriteString(" WITH ") + buf.WriteString(s.Option.String()) + } + return buf.String() +} + type AlterTableStatement struct { Alter Pos // position of ALTER keyword Table Pos // position of TABLE keyword @@ -2459,12 +2581,41 @@ type ColumnArg interface { columnArg() } +type DropDatabaseStatement struct { + Drop Pos // position of DROP keyword + Database Pos // position of DATABASE keyword + If Pos // position of IF keyword + IfExists Pos // position of EXISTS keyword after IF + Name *Ident // database name +} + +// Clone returns a deep copy of s. +func (s *DropDatabaseStatement) Clone() *DropDatabaseStatement { + if s == nil { + return nil + } + other := *s + other.Name = s.Name.Clone() + return &other +} + +// String returns the string representation of the statement. +func (s *DropDatabaseStatement) String() string { + var buf bytes.Buffer + buf.WriteString("DROP DATABASE") + if s.IfExists.IsValid() { + buf.WriteString(" IF EXISTS") + } + fmt.Fprintf(&buf, " %s", s.Name.String()) + return buf.String() +} + type DropTableStatement struct { Drop Pos // position of DROP keyword Table Pos // position of TABLE keyword If Pos // position of IF keyword IfExists Pos // position of EXISTS keyword after IF - Name *Ident // view name + Name *Ident // table name } // Clone returns a deep copy of s. diff --git a/sql3/parser/ast_test.go b/sql3/parser/ast_test.go index b963f2161..ceec49bcc 100644 --- a/sql3/parser/ast_test.go +++ b/sql3/parser/ast_test.go @@ -116,6 +116,38 @@ func TestCreateIndexStatement_String(t *testing.T) { }, `CREATE UNIQUE INDEX IF NOT EXISTS "foo" ON "bar" ("baz", "bat") WHERE TRUE`) } +func TestCreateDatabaseStatement_String(t *testing.T) { + AssertStatementStringer(t, &parser.CreateDatabaseStatement{ + Name: &parser.Ident{Name: "db1"}, + }, `CREATE DATABASE db1`) + + AssertStatementStringer(t, &parser.CreateDatabaseStatement{ + Name: &parser.Ident{Name: "db1"}, + IfNotExists: pos(0), + }, `CREATE DATABASE IF NOT EXISTS db1`) + + AssertStatementStringer(t, &parser.CreateDatabaseStatement{ + Name: &parser.Ident{Name: "db1"}, + With: pos(10), + Options: []parser.DatabaseOption{ + &parser.UnitsOption{ + Units: pos(0), + Expr: &parser.IntegerLit{Value: "4"}, + }, + }, + }, `CREATE DATABASE db1 WITH UNITS 4`) +} + +func TestAlterDatabaseStatement_String(t *testing.T) { + AssertStatementStringer(t, &parser.AlterDatabaseStatement{ + Name: &parser.Ident{Name: "db1"}, + Option: &parser.UnitsOption{ + Units: pos(0), + Expr: &parser.IntegerLit{Value: "4"}, + }, + }, `ALTER DATABASE db1 WITH UNITS 4`) +} + func TestCreateTableStatement_String(t *testing.T) { AssertStatementStringer(t, &parser.CreateTableStatement{ Name: &parser.Ident{Name: "foo"}, @@ -478,6 +510,17 @@ func TestDropIndexStatement_String(t *testing.T) { }, `DROP INDEX IF EXISTS "idx"`) } +func TestDropDatabaseStatement_String(t *testing.T) { + AssertStatementStringer(t, &parser.DropDatabaseStatement{ + Name: &parser.Ident{Name: "db"}, + }, `DROP DATABASE db`) + + AssertStatementStringer(t, &parser.DropDatabaseStatement{ + IfExists: pos(0), + Name: &parser.Ident{Name: "db"}, + }, `DROP DATABASE IF EXISTS db`) +} + func TestDropTableStatement_String(t *testing.T) { AssertStatementStringer(t, &parser.DropTableStatement{ Name: &parser.Ident{Name: "tbl"}, diff --git a/sql3/parser/parser.go b/sql3/parser/parser.go index 8b69d807f..58e0a4234 100644 --- a/sql3/parser/parser.go +++ b/sql3/parser/parser.go @@ -149,6 +149,8 @@ func (p *Parser) parseShowStatement() (Statement, error) { show, _, _ := p.scan() switch p.peek() { + case DATABASES: + return p.parseShowDatabasesStatement(show) case TABLES: return p.parseShowTablesStatement(show) case COLUMNS: @@ -156,7 +158,19 @@ func (p *Parser) parseShowStatement() (Statement, error) { case CREATE: return p.parseShowCreateStatement(show) default: - return nil, p.errorExpected(p.pos, p.tok, "TABLES, COLUMNS or CREATE") + return nil, p.errorExpected(p.pos, p.tok, "DATABASES, TABLES, COLUMNS or CREATE") + } +} + +func (p *Parser) parseShowDatabasesStatement(showPos Pos) (*ShowDatabasesStatement, error) { + switch p.peek() { + case DATABASES: + var stmt ShowDatabasesStatement + stmt.Show = showPos + stmt.Databases, _, _ = p.scan() + return &stmt, nil + default: + return nil, p.errorExpected(p.pos, p.tok, "DATABASES") } } @@ -314,6 +328,8 @@ func (p *Parser) parseCreateStatement() (Statement, error) { pos, tok, _ := p.scan() switch p.peek() { + case DATABASE: + return p.parseCreateDatabaseStatement(pos) case TABLE: return p.parseCreateTableStatement(pos) case VIEW: @@ -323,7 +339,7 @@ func (p *Parser) parseCreateStatement() (Statement, error) { case FUNCTION: return p.parseCreateFunctionStatement(pos) default: - return nil, p.errorExpected(pos, tok, "TABLE, VIEW or FUNCTION") + return nil, p.errorExpected(pos, tok, "DATABASE, TABLE, VIEW or FUNCTION") } } @@ -332,12 +348,14 @@ func (p *Parser) parseAlterStatement() (Statement, error) { pos, tok, _ := p.scan() switch p.peek() { + case DATABASE: + return p.parseAlterDatabaseStatement(pos) case TABLE: return p.parseAlterTableStatement(pos) case VIEW: return p.parseAlterViewStatement(pos) default: - return nil, p.errorExpected(pos, tok, "TABLE or VIEW") + return nil, p.errorExpected(pos, tok, "DATABASE, TABLE or VIEW") } } @@ -346,6 +364,8 @@ func (p *Parser) parseDropStatement() (Statement, error) { pos, tok, _ := p.scan() switch p.peek() { + case DATABASE: + return p.parseDropDatabaseStatement(pos) case TABLE: return p.parseDropTableStatement(pos) case VIEW: @@ -355,10 +375,102 @@ func (p *Parser) parseDropStatement() (Statement, error) { case FUNCTION: return p.parseDropFunctionStatement(pos) default: - return nil, p.errorExpected(pos, tok, "TABLE, VIEW or FUNCTION") + return nil, p.errorExpected(pos, tok, "DATABASE, TABLE, VIEW or FUNCTION") } } +func (p *Parser) parseCreateDatabaseStatement(createPos Pos) (_ *CreateDatabaseStatement, err error) { + assert(p.peek() == DATABASE) + + var stmt CreateDatabaseStatement + stmt.Create = createPos + stmt.Database, _, _ = p.scan() + + // Parse optional "IF NOT EXISTS". + if p.peek() == IF { + stmt.If, _, _ = p.scan() + + pos, tok, _ := p.scan() + if tok != NOT { + return &stmt, p.errorExpected(pos, tok, "NOT") + } + stmt.IfNot = pos + + pos, tok, _ = p.scan() + if tok != EXISTS { + return &stmt, p.errorExpected(pos, tok, "EXISTS") + } + stmt.IfNotExists = pos + } + + if stmt.Name, err = p.parseIdent("database name"); err != nil { + return &stmt, err + } + + switch p.peek() { + case WITH: + stmt.With, _, _ = p.scan() + + // look for database options + if stmt.Options, err = p.parseDatabaseOptions(); err != nil { + return &stmt, err + } + } + + return &stmt, nil +} + +func (p *Parser) parseDatabaseOptions() (_ []DatabaseOption, err error) { + if !isDatabaseOptionStartToken(p.peek()) { + return nil, nil + } + + var a []DatabaseOption + + for { + if !isDatabaseOptionStartToken(p.peek()) { + return a, nil + } + cons, err := p.parseDatabaseOption() + if cons != nil { + a = append(a, cons) + } + if err != nil { + return a, err + } + } +} + +func (p *Parser) parseDatabaseOption() (_ DatabaseOption, err error) { + assert(isDatabaseOptionStartToken(p.peek())) + + var optionPos Pos + + // Parse database options. + switch p.peek() { + case UNITS: + return p.parseUnitsOption(optionPos) + default: + assert(p.peek() == COMMENT) + return p.parseCommentOption(optionPos) + } +} + +func (p *Parser) parseUnitsOption(optionPos Pos) (_ *UnitsOption, err error) { + assert(p.peek() == UNITS) + + var opt UnitsOption + opt.Units, _, _ = p.scan() + + if isLiteralToken(p.peek()) { + opt.Expr = p.mustParseLiteral() + } else { + return &opt, p.errorExpected(p.pos, p.tok, "literal") + } + + return &opt, nil +} + func (p *Parser) parseCreateTableStatement(createPos Pos) (_ *CreateTableStatement, err error) { assert(p.peek() == TABLE) @@ -1009,6 +1121,29 @@ func (p *Parser) parseTimeQuantumConstraint(constraintPos Pos, name *Ident) (_ * return &cons, nil }*/ +func (p *Parser) parseDropDatabaseStatement(dropPos Pos) (_ *DropDatabaseStatement, err error) { + assert(p.peek() == DATABASE) + + var stmt DropDatabaseStatement + stmt.Drop = dropPos + stmt.Database, _, _ = p.scan() + + // Parse optional "IF EXISTS". + if p.peek() == IF { + stmt.If, _, _ = p.scan() + if p.peek() != EXISTS { + return &stmt, p.errorExpected(p.pos, p.tok, "EXISTS") + } + stmt.IfExists, _, _ = p.scan() + } + + if stmt.Name, err = p.parseIdent("database name"); err != nil { + return &stmt, err + } + + return &stmt, nil +} + func (p *Parser) parseDropTableStatement(dropPos Pos) (_ *DropTableStatement, err error) { assert(p.peek() == TABLE) @@ -3283,6 +3418,36 @@ func (p *Parser) parseIntegerLiteral(desc string) (*IntegerLit, error) { } } +func (p *Parser) parseAlterDatabaseStatement(alterPos Pos) (_ *AlterDatabaseStatement, err error) { + var stmt AlterDatabaseStatement + stmt.Alter = alterPos + if p.peek() != DATABASE { + return &stmt, p.errorExpected(p.pos, p.tok, "DATABASE") + } + stmt.Database, _, _ = p.scan() + + if stmt.Name, err = p.parseIdent("database name"); err != nil { + return &stmt, err + } + + switch p.peek() { + case WITH: + stmt.With, _, _ = p.scan() + + // look for database option + if !isDatabaseOptionStartToken(p.peek()) { + return &stmt, p.errorExpected(p.pos, p.tok, "UNITS") + } + if stmt.Option, err = p.parseDatabaseOption(); err != nil { + return &stmt, err + } + default: + return &stmt, p.errorExpected(p.pos, p.tok, "WITH") + } + + return &stmt, nil +} + func (p *Parser) parseAlterTableStatement(alterPos Pos) (_ *AlterTableStatement, err error) { var stmt AlterTableStatement stmt.Alter = alterPos @@ -3454,6 +3619,16 @@ func (e Error) Error() string { return e.Msg } +// isDatabaseOptionStartToken returns true if tok is the initial token of a table option. +func isDatabaseOptionStartToken(tok Token) bool { + switch tok { + case UNITS, COMMENT: + return true + default: + return false + } +} + // isTableOptionStartToken returns true if tok is the initial token of a table option. func isTableOptionStartToken(tok Token) bool { switch tok { diff --git a/sql3/parser/parser_test.go b/sql3/parser/parser_test.go index 67d9316ac..b089db15c 100644 --- a/sql3/parser/parser_test.go +++ b/sql3/parser/parser_test.go @@ -349,6 +349,27 @@ func TestParser_ParseCacheTypeConstraints(t *testing.T) { } func TestParser_ParseAlterStatement(t *testing.T) { + t.Run("AlterDatabase", func(t *testing.T) { + AssertParseStatement(t, `ALTER DATABASE db1 WITH UNITS 4`, &parser.AlterDatabaseStatement{ + Alter: pos(0), + Database: pos(6), + Name: &parser.Ident{NamePos: pos(15), Name: "db1"}, + With: pos(19), + Option: &parser.UnitsOption{ + Units: pos(24), + Expr: &parser.IntegerLit{ + ValuePos: pos(30), + Value: "4", + }, + }, + }) + + AssertParseStatementError(t, `ALTER`, `1:1: expected DATABASE, TABLE or VIEW`) + AssertParseStatementError(t, `ALTER DATABASE`, `1:14: expected database name, found 'EOF'`) + AssertParseStatementError(t, `ALTER DATABASE db1`, `1:18: expected WITH, found 'EOF'`) + AssertParseStatementError(t, `ALTER DATABASE db1 WITH`, `1:23: expected UNITS, found 'EOF'`) + AssertParseStatementError(t, `ALTER DATABASE db1 WITH UNITS`, `1:29: expected literal, found 'EOF'`) + }) t.Run("AlterTable", func(t *testing.T) { /*AssertParseStatement(t, `ALTER TABLE tbl RENAME TO new_tbl`, &parser.AlterTableStatement{ @@ -427,7 +448,7 @@ func TestParser_ParseAlterStatement(t *testing.T) { DropColumnName: &parser.Ident{NamePos: pos(28), Name: "col"}, }) - AssertParseStatementError(t, `ALTER`, `1:1: expected TABLE or VIEW`) + AssertParseStatementError(t, `ALTER`, `1:1: expected DATABASE, TABLE or VIEW`) AssertParseStatementError(t, `ALTER TABLE`, `1:11: expected table name, found 'EOF'`) AssertParseStatementError(t, `ALTER TABLE tbl`, `1:15: expected ADD, DROP or RENAME, found 'EOF'`) AssertParseStatementError(t, `ALTER TABLE tbl RENAME`, `1:22: expected COLUMN keyword or column name, found 'EOF'`) @@ -622,13 +643,17 @@ func TestParser_ParseStatement(t *testing.T) { AssertParseStatementError(t, `123`, `1:1: expected statement, found 123`) }) - t.Run("ShowTables", func(t *testing.T) { + t.Run("ShowDatabasesAndTables", func(t *testing.T) { + AssertParseStatement(t, `SHOW DATABASES`, &parser.ShowDatabasesStatement{ + Show: pos(0), + Databases: pos(5), + }) AssertParseStatement(t, `SHOW TABLES`, &parser.ShowTablesStatement{ Show: pos(0), Tables: pos(5), }) - AssertParseStatementError(t, `SHOW`, `1:4: expected TABLES, COLUMNS or CREATE, found 'EOF'`) - AssertParseStatementError(t, `SHOW BLAH`, `1:6: expected TABLES, COLUMNS or CREATE, found BLAH`) + AssertParseStatementError(t, `SHOW`, `1:4: expected DATABASES, TABLES, COLUMNS or CREATE, found 'EOF'`) + AssertParseStatementError(t, `SHOW BLAH`, `1:6: expected DATABASES, TABLES, COLUMNS or CREATE, found BLAH`) }) t.Run("ShowColumns", func(t *testing.T) { @@ -641,7 +666,7 @@ func TestParser_ParseStatement(t *testing.T) { NamePos: pos(18), }, }) - AssertParseStatementError(t, `SHOW`, `1:4: expected TABLES, COLUMNS or CREATE, found 'EOF'`) + AssertParseStatementError(t, `SHOW`, `1:4: expected DATABASES, TABLES, COLUMNS or CREATE, found 'EOF'`) AssertParseStatementError(t, `SHOW COLUMNS`, `1:12: expected FROM, found 'EOF'`) AssertParseStatementError(t, `SHOW COLUMNS FOO`, `1:14: expected FROM, found FOO`) AssertParseStatementError(t, `SHOW COLUMNS FROM`, `1:17: expected table name, found 'EOF'`) @@ -658,7 +683,7 @@ func TestParser_ParseStatement(t *testing.T) { NamePos: pos(18), }, }) - AssertParseStatementError(t, `SHOW`, `1:4: expected TABLES, COLUMNS or CREATE, found 'EOF'`) + AssertParseStatementError(t, `SHOW`, `1:4: expected DATABASES, TABLES, COLUMNS or CREATE, found 'EOF'`) AssertParseStatementError(t, `SHOW CREATE`, `1:11: expected TABLES, found 'EOF'`) AssertParseStatementError(t, `SHOW CREATE TABLE`, `1:17: expected table name, found 'EOF'`) AssertParseStatementError(t, `SHOW CREATE TABLE 12`, `1:19: expected table name, found 12`) @@ -855,6 +880,32 @@ func TestParser_ParseStatement(t *testing.T) { }) })*/ + t.Run("CreateDatabase", func(t *testing.T) { + AssertParseStatement(t, `CREATE DATABASE db WITH UNITS 4`, &parser.CreateDatabaseStatement{ + Create: pos(0), + Database: pos(7), + Name: &parser.Ident{ + Name: "db", + NamePos: pos(16), + }, + With: pos(19), + Options: []parser.DatabaseOption{ + &parser.UnitsOption{ + Units: pos(24), + Expr: &parser.IntegerLit{ + ValuePos: pos(30), + Value: "4", + }, + }, + }, + }) + + AssertParseStatementError(t, `CREATE DATABASE`, `1:15: expected database name, found 'EOF'`) + AssertParseStatementError(t, `CREATE DATABASE db (`, `1:20: expected semicolon or EOF, found '('`) + AssertParseStatementError(t, `CREATE DATABASE db extra`, `1:20: expected semicolon or EOF, found extra`) + AssertParseStatementError(t, `CREATE DATABASE db WITH UNITS`, `1:29: expected literal, found 'EOF'`) + }) + t.Run("CreateTable", func(t *testing.T) { AssertParseStatement(t, `CREATE TABLE tbl (col1 TEXT, col2 DECIMAL(2))`, &parser.CreateTableStatement{ Create: pos(0), @@ -1547,6 +1598,24 @@ func TestParser_ParseStatement(t *testing.T) { }) }) + t.Run("DropDatabase", func(t *testing.T) { + AssertParseStatement(t, `DROP DATABASE db`, &parser.DropDatabaseStatement{ + Drop: pos(0), + Database: pos(5), + Name: &parser.Ident{NamePos: pos(14), Name: "db"}, + }) + AssertParseStatement(t, `DROP DATABASE IF EXISTS db`, &parser.DropDatabaseStatement{ + Drop: pos(0), + Database: pos(5), + If: pos(14), + IfExists: pos(17), + Name: &parser.Ident{NamePos: pos(24), Name: "db"}, + }) + AssertParseStatementError(t, `DROP DATABASE`, `1:13: expected database name, found 'EOF'`) + AssertParseStatementError(t, `DROP DATABASE IF`, `1:16: expected EXISTS, found 'EOF'`) + AssertParseStatementError(t, `DROP DATABASE IF EXISTS`, `1:23: expected database name, found 'EOF'`) + }) + t.Run("DropTable", func(t *testing.T) { AssertParseStatement(t, `DROP TABLE vw`, &parser.DropTableStatement{ Drop: pos(0), @@ -1636,7 +1705,7 @@ func TestParser_ParseStatement(t *testing.T) { IfExists: pos(13), Name: &parser.Ident{NamePos: pos(20), Name: "vw"}, }) - AssertParseStatementError(t, `DROP`, `1:1: expected TABLE, VIEW or FUNCTION`) + AssertParseStatementError(t, `DROP`, `1:1: expected DATABASE, TABLE, VIEW or FUNCTION`) AssertParseStatementError(t, `DROP VIEW`, `1:9: expected view name, found 'EOF'`) AssertParseStatementError(t, `DROP VIEW IF`, `1:12: expected EXISTS, found 'EOF'`) AssertParseStatementError(t, `DROP VIEW IF EXISTS`, `1:19: expected view name, found 'EOF'`) diff --git a/sql3/parser/token.go b/sql3/parser/token.go index af6721666..ef16f7ef2 100644 --- a/sql3/parser/token.go +++ b/sql3/parser/token.go @@ -113,6 +113,7 @@ const ( CURRENT_DATE CURRENT_TIMESTAMP DATABASE + DATABASES DEFAULT DEFERRABLE DEFERRED @@ -237,6 +238,7 @@ const ( UNBOUNDED UNION UNIQUE + UNITS UPDATE USING VACUUM @@ -337,6 +339,7 @@ var tokens = [...]string{ CURRENT_DATE: "CURRENT_DATE", CURRENT_TIMESTAMP: "CURRENT_TIMESTAMP", DATABASE: "DATABASE", + DATABASES: "DATABASES", DEFAULT: "DEFAULT", DEFERRABLE: "DEFERRABLE", DEFERRED: "DEFERRED", @@ -461,6 +464,7 @@ var tokens = [...]string{ UNBOUNDED: "UNBOUNDED", UNION: "UNION", UNIQUE: "UNIQUE", + UNITS: "UNITS", UPDATE: "UPDATE", USING: "USING", VACUUM: "VACUUM", diff --git a/sql3/planner/compilealterdatabase.go b/sql3/planner/compilealterdatabase.go new file mode 100644 index 000000000..c8c74d54c --- /dev/null +++ b/sql3/planner/compilealterdatabase.go @@ -0,0 +1,40 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3" + "github.com/featurebasedb/featurebase/v3/sql3/parser" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// compileAlterDatabaseStatement compiles an ALTER DATABASE statement into a +// PlanOperator. +func (p *ExecutionPlanner) compileAlterDatabaseStatement(stmt *parser.AlterDatabaseStatement) (_ types.PlanOperator, err error) { + databaseName := parser.IdentName(stmt.Name) + + // does the database exist + dbname := dax.DatabaseName(databaseName) + db, err := p.schemaAPI.DatabaseByName(context.Background(), dbname) + if err != nil { + if isDatabaseNotFoundError(err) { + return nil, sql3.NewErrDatabaseNotFound(stmt.Name.NamePos.Line, stmt.Name.NamePos.Column, databaseName) + } + return nil, err + } + + if stmt.With.IsValid() { + return NewPlanOpQuery(p, NewPlanOpAlterDatabase(p, db, alterOpSet, stmt.Option), p.sql), nil + } else { + return nil, sql3.NewErrInternal("unhandled alter operation") + } +} + +// analyzeAlterDatabaseStatement analyze an ALTER DATABASE statement and returns an +// error if anything is invalid. +func (p *ExecutionPlanner) analyzeAlterDatabaseStatement(stmt *parser.AlterDatabaseStatement) error { + return nil +} diff --git a/sql3/planner/compilealtertable.go b/sql3/planner/compilealtertable.go index cff677f89..db67dda82 100644 --- a/sql3/planner/compilealtertable.go +++ b/sql3/planner/compilealtertable.go @@ -18,6 +18,7 @@ const ( alterOpAdd alterOperation = iota alterOpDrop alterOpRename + alterOpSet ) // compileAlterTableStatement compiles an ALTER TABLE statement into a diff --git a/sql3/planner/compilecreatedatabase.go b/sql3/planner/compilecreatedatabase.go new file mode 100644 index 000000000..f78127127 --- /dev/null +++ b/sql3/planner/compilecreatedatabase.go @@ -0,0 +1,83 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "strconv" + + "github.com/featurebasedb/featurebase/v3/sql3" + "github.com/featurebasedb/featurebase/v3/sql3/parser" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// compileCreateDatabaseStatement compiles a CREATE DATABASE statement into a +// PlanOperator. +func (p *ExecutionPlanner) compileCreateDatabaseStatement(stmt *parser.CreateDatabaseStatement) (_ types.PlanOperator, err error) { + databaseName := parser.IdentName(stmt.Name) + failIfExists := !stmt.IfNotExists.IsValid() + + units := 0 + description := "" + + // apply database options + if stmt.With.IsValid() { + for _, option := range stmt.Options { + switch o := option.(type) { + case *parser.UnitsOption: + e := o.Expr.(*parser.IntegerLit) + i, err := strconv.ParseInt(e.Value, 10, 64) + if err != nil { + return nil, err + } + units = int(i) + case *parser.CommentOption: + e := o.Expr.(*parser.StringLit) + description = e.Value + } + } + } + + cop := NewPlanOpCreateDatabase(p, databaseName, failIfExists, units, description) + return NewPlanOpQuery(p, cop, p.sql), nil +} + +// analyzeCreateDatabaseStatement analyzes a CREATE DATABASE statement and +// returns an error if anything is invalid. +func (p *ExecutionPlanner) analyzeCreateDatabaseStatement(stmt *parser.CreateDatabaseStatement) error { + if stmt.With.IsValid() { + // no checks if WITH is not provided + return nil + } + + //check database options + for _, option := range stmt.Options { + + switch o := option.(type) { + case *parser.UnitsOption: + //check the type of the expression + literal, ok := o.Expr.(*parser.IntegerLit) + if !ok { + return sql3.NewErrIntegerLiteral(o.Expr.Pos().Line, o.Expr.Pos().Column) + } + // units needs to be >=0 and we'll cap conservatively at 10000 + i, err := strconv.ParseInt(literal.Value, 10, 64) + if err != nil { + return err + } + if i < 0 || i > 10000 { + return sql3.NewErrInvalidUnitsValue(o.Expr.Pos().Line, o.Expr.Pos().Column, i) + } + + case *parser.CommentOption: + _, ok := o.Expr.(*parser.StringLit) + if !ok { + return sql3.NewErrStringLiteral(o.Expr.Pos().Line, o.Expr.Pos().Column) + } + + default: + return sql3.NewErrInternalf("unhandled database option type '%T'", option) + } + } + + return nil +} diff --git a/sql3/planner/compiledropdatabase.go b/sql3/planner/compiledropdatabase.go new file mode 100644 index 000000000..3a4708108 --- /dev/null +++ b/sql3/planner/compiledropdatabase.go @@ -0,0 +1,27 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3" + "github.com/featurebasedb/featurebase/v3/sql3/parser" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// compileDropDatabaseStatement compiles a DROP DATABASE statement into a +// PlanOperator. +func (p *ExecutionPlanner) compileDropDatabaseStatement(stmt *parser.DropDatabaseStatement) (_ types.PlanOperator, err error) { + databaseName := parser.IdentName(stmt.Name) + dbname := dax.DatabaseName(databaseName) + db, err := p.schemaAPI.DatabaseByName(context.Background(), dbname) + if err != nil { + if isDatabaseNotFoundError(err) { + return nil, sql3.NewErrDatabaseNotFound(stmt.Name.NamePos.Line, stmt.Name.NamePos.Column, databaseName) + } + return nil, err + } + return NewPlanOpQuery(p, NewPlanOpDropDatabase(p, db), p.sql), nil +} diff --git a/sql3/planner/compileshow.go b/sql3/planner/compileshow.go index d30a302f3..23d45c41a 100644 --- a/sql3/planner/compileshow.go +++ b/sql3/planner/compileshow.go @@ -14,6 +14,65 @@ import ( "github.com/pkg/errors" ) +func (p *ExecutionPlanner) compileShowDatabasesStatement(stmt parser.Statement) (types.PlanOperator, error) { + dbs, err := p.schemaAPI.Databases(context.Background()) + if err != nil { + return nil, errors.Wrap(err, "getting databases") + } + + columns := []types.PlanExpression{ + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "_id", + columnIndex: 0, + dataType: parser.NewDataTypeString(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "name", + columnIndex: 1, + dataType: parser.NewDataTypeString(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "owner", + columnIndex: 2, + dataType: parser.NewDataTypeString(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "updated_by", + columnIndex: 3, + dataType: parser.NewDataTypeString(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "created_at", + columnIndex: 4, + dataType: parser.NewDataTypeTimestamp(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "updated_at", + columnIndex: 5, + dataType: parser.NewDataTypeTimestamp(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "units", + columnIndex: 6, + dataType: parser.NewDataTypeInt(), + }, + &qualifiedRefPlanExpression{ + tableName: "fb_databases", + columnName: "description", + columnIndex: 7, + dataType: parser.NewDataTypeString(), + }} + + return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseDatabases(p, dbs)), p.sql), nil +} + func (p *ExecutionPlanner) compileShowTablesStatement(stmt parser.Statement) (types.PlanOperator, error) { tbls, err := p.schemaAPI.Tables(context.Background()) if err != nil { diff --git a/sql3/planner/executionplanner.go b/sql3/planner/executionplanner.go index 46657ee97..795b4d888 100644 --- a/sql3/planner/executionplanner.go +++ b/sql3/planner/executionplanner.go @@ -21,6 +21,11 @@ import ( "golang.org/x/sync/errgroup" ) +func isDatabaseNotFoundError(err error) bool { + // TODO (pok) take out the second part of this check once we return correct error types across network boundaries + return errors.Is(err, dax.ErrDatabaseNameDoesNotExist) || strings.Contains(err.Error(), "does not exist") +} + func isTableNotFoundError(err error) bool { // TODO (pok) take out the second part of this check once we return correct error types across network boundaries return errors.Is(err, dax.ErrTableNameDoesNotExist) || strings.Contains(err.Error(), "does not exist") @@ -40,7 +45,7 @@ type ExecutionPlanner struct { func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, systemAPI pilosa.SystemAPI, systemLayerAPI pilosa.SystemLayerAPI, importer pilosa.Importer, logger logger.Logger, sql string) *ExecutionPlanner { return &ExecutionPlanner{ executor: executor, - schemaAPI: newSystemTableDefintionsWrapper(schemaAPI), + schemaAPI: newSystemTableDefinitionsWrapper(schemaAPI), systemAPI: systemAPI, systemLayerAPI: systemLayerAPI, importer: importer, @@ -65,20 +70,28 @@ func (p *ExecutionPlanner) CompilePlan(ctx context.Context, stmt parser.Statemen switch stmt := stmt.(type) { case *parser.SelectStatement: rootOperator, err = p.compileSelectStatement(stmt, false) + case *parser.ShowDatabasesStatement: + rootOperator, err = p.compileShowDatabasesStatement(stmt) case *parser.ShowTablesStatement: rootOperator, err = p.compileShowTablesStatement(stmt) case *parser.ShowColumnsStatement: rootOperator, err = p.compileShowColumnsStatement(stmt) case *parser.ShowCreateTableStatement: rootOperator, err = p.compileShowCreateTableStatement(stmt) + case *parser.CreateDatabaseStatement: + rootOperator, err = p.compileCreateDatabaseStatement(stmt) case *parser.CreateTableStatement: rootOperator, err = p.compileCreateTableStatement(stmt) case *parser.CreateViewStatement: rootOperator, err = p.compileCreateViewStatement(stmt) + case *parser.AlterDatabaseStatement: + rootOperator, err = p.compileAlterDatabaseStatement(stmt) case *parser.AlterTableStatement: rootOperator, err = p.compileAlterTableStatement(stmt) case *parser.AlterViewStatement: rootOperator, err = p.compileAlterViewStatement(stmt) + case *parser.DropDatabaseStatement: + rootOperator, err = p.compileDropDatabaseStatement(stmt) case *parser.DropTableStatement: rootOperator, err = p.compileDropTableStatement(stmt) case *parser.DropViewStatement: @@ -118,20 +131,28 @@ func (p *ExecutionPlanner) analyzePlan(stmt parser.Statement) error { case *parser.SelectStatement: _, err := p.analyzeSelectStatement(stmt) return err + case *parser.ShowDatabasesStatement: + return nil case *parser.ShowTablesStatement: return nil case *parser.ShowColumnsStatement: return nil case *parser.ShowCreateTableStatement: return nil + case *parser.CreateDatabaseStatement: + return p.analyzeCreateDatabaseStatement(stmt) case *parser.CreateTableStatement: return p.analyzeCreateTableStatement(stmt) case *parser.CreateViewStatement: return p.analyzeCreateViewStatement(stmt) + case *parser.AlterDatabaseStatement: + return p.analyzeAlterDatabaseStatement(stmt) case *parser.AlterTableStatement: return p.analyzeAlterTableStatement(stmt) case *parser.AlterViewStatement: return p.analyzeAlterViewStatement(stmt) + case *parser.DropDatabaseStatement: + return nil case *parser.DropTableStatement: return nil case *parser.DropViewStatement: diff --git a/sql3/planner/executionplannersystemtables.go b/sql3/planner/executionplannersystemtables.go index 039da5c3a..ac620e3e1 100644 --- a/sql3/planner/executionplannersystemtables.go +++ b/sql3/planner/executionplannersystemtables.go @@ -13,19 +13,39 @@ import ( ) // Ensure type implements interface. -var _ pilosa.SchemaAPI = (*systemTableDefintionsWrapper)(nil) +var _ pilosa.SchemaAPI = (*systemTableDefinitionsWrapper)(nil) -type systemTableDefintionsWrapper struct { +type systemTableDefinitionsWrapper struct { schemaAPI pilosa.SchemaAPI } -func newSystemTableDefintionsWrapper(api pilosa.SchemaAPI) *systemTableDefintionsWrapper { - return &systemTableDefintionsWrapper{ +func newSystemTableDefinitionsWrapper(api pilosa.SchemaAPI) *systemTableDefinitionsWrapper { + return &systemTableDefinitionsWrapper{ schemaAPI: api, } } -func (s *systemTableDefintionsWrapper) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { +func (s *systemTableDefinitionsWrapper) CreateDatabase(ctx context.Context, db *dax.Database) error { + return s.schemaAPI.CreateDatabase(ctx, db) +} +func (s *systemTableDefinitionsWrapper) DropDatabase(ctx context.Context, dbid dax.DatabaseID) error { + return s.schemaAPI.DropDatabase(ctx, dbid) +} + +func (s *systemTableDefinitionsWrapper) DatabaseByName(ctx context.Context, dbname dax.DatabaseName) (*dax.Database, error) { + return s.schemaAPI.DatabaseByName(ctx, dbname) +} +func (s *systemTableDefinitionsWrapper) DatabaseByID(ctx context.Context, dbid dax.DatabaseID) (*dax.Database, error) { + return s.schemaAPI.DatabaseByID(ctx, dbid) +} +func (s *systemTableDefinitionsWrapper) SetDatabaseOption(ctx context.Context, dbid dax.DatabaseID, option string, value string) error { + return s.schemaAPI.SetDatabaseOption(ctx, dbid, option, value) +} +func (s *systemTableDefinitionsWrapper) Databases(ctx context.Context, dbids ...dax.DatabaseID) ([]*dax.Database, error) { + return s.schemaAPI.Databases(ctx, dbids...) +} + +func (s *systemTableDefinitionsWrapper) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) { tbl, err := s.schemaAPI.TableByName(ctx, tname) if err != nil { if isTableNotFoundError(err) { @@ -41,11 +61,11 @@ func (s *systemTableDefintionsWrapper) TableByName(ctx context.Context, tname da return tbl, nil } -func (s *systemTableDefintionsWrapper) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) { +func (s *systemTableDefinitionsWrapper) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) { return s.schemaAPI.TableByID(ctx, tid) } -func (s *systemTableDefintionsWrapper) Tables(ctx context.Context) ([]*dax.Table, error) { +func (s *systemTableDefinitionsWrapper) Tables(ctx context.Context) ([]*dax.Table, error) { tbls, err := s.schemaAPI.Tables(ctx) if err != nil { return nil, errors.Wrap(err, "getting tables") @@ -63,19 +83,19 @@ func (s *systemTableDefintionsWrapper) Tables(ctx context.Context) ([]*dax.Table return tbls, nil } -func (s *systemTableDefintionsWrapper) CreateTable(ctx context.Context, tbl *dax.Table) error { +func (s *systemTableDefinitionsWrapper) CreateTable(ctx context.Context, tbl *dax.Table) error { return s.schemaAPI.CreateTable(ctx, tbl) } -func (s *systemTableDefintionsWrapper) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error { +func (s *systemTableDefinitionsWrapper) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error { return s.schemaAPI.CreateField(ctx, tname, fld) } -func (s *systemTableDefintionsWrapper) DeleteTable(ctx context.Context, tname dax.TableName) error { +func (s *systemTableDefinitionsWrapper) DeleteTable(ctx context.Context, tname dax.TableName) error { return s.schemaAPI.DeleteTable(ctx, tname) } -func (s *systemTableDefintionsWrapper) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error { +func (s *systemTableDefinitionsWrapper) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error { return s.schemaAPI.DeleteField(ctx, tname, fname) } diff --git a/sql3/planner/opalterdatabase.go b/sql3/planner/opalterdatabase.go new file mode 100644 index 000000000..c9f87fd97 --- /dev/null +++ b/sql3/planner/opalterdatabase.go @@ -0,0 +1,107 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + "fmt" + "log" + + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3" + "github.com/featurebasedb/featurebase/v3/sql3/parser" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// PlanOpAlterDatabase plan operator to alter a database. +type PlanOpAlterDatabase struct { + planner *ExecutionPlanner + database *dax.Database + operation alterOperation + option parser.DatabaseOption + warnings []string +} + +func NewPlanOpAlterDatabase(p *ExecutionPlanner, db *dax.Database, operation alterOperation, option parser.DatabaseOption) *PlanOpAlterDatabase { + return &PlanOpAlterDatabase{ + planner: p, + database: db, + operation: operation, + option: option, + warnings: make([]string, 0), + } +} + +func (p *PlanOpAlterDatabase) Plan() map[string]interface{} { + result := make(map[string]interface{}) + result["_op"] = fmt.Sprintf("%T", p) + result["databaseName"] = p.database.Name + result["operation"] = p.operation + result["option"] = p.option + return result +} + +func (p *PlanOpAlterDatabase) AddWarning(warning string) { + p.warnings = append(p.warnings, warning) +} + +func (p *PlanOpAlterDatabase) Warnings() []string { + return p.warnings +} + +func (p *PlanOpAlterDatabase) String() string { + return "" +} + +func (p *PlanOpAlterDatabase) Schema() types.Schema { + return types.Schema{} +} + +func (p *PlanOpAlterDatabase) Children() []types.PlanOperator { + return []types.PlanOperator{} +} + +func (p *PlanOpAlterDatabase) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + return &alterDatabaseRowIter{ + planner: p.planner, + database: p.database, + operation: p.operation, + option: p.option, + }, nil +} + +func (p *PlanOpAlterDatabase) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { + return nil, nil +} + +type alterDatabaseRowIter struct { + planner *ExecutionPlanner + database *dax.Database + operation alterOperation + option parser.DatabaseOption +} + +var _ types.RowIterator = (*alterDatabaseRowIter)(nil) + +func (i *alterDatabaseRowIter) Next(ctx context.Context) (types.Row, error) { + switch i.operation { + case alterOpSet: + var optName string + var optValue string + + switch v := i.option.(type) { + case *parser.UnitsOption: + e := v.Expr.(*parser.IntegerLit) + optName = "workers-min" + optValue = e.Value + default: + return nil, sql3.NewErrInvalidDatabaseOption(0, 0, i.option.String()) + } + + log.Printf("DEEBUG: SetDatabaseOption: %s = %s", optName, optValue) + if err := i.planner.schemaAPI.SetDatabaseOption(ctx, i.database.ID, optName, optValue); err != nil { + return nil, err + } + } + return nil, types.ErrNoMoreRows +} diff --git a/sql3/planner/opcreatedatabase.go b/sql3/planner/opcreatedatabase.go new file mode 100644 index 000000000..2bdb19dcb --- /dev/null +++ b/sql3/planner/opcreatedatabase.go @@ -0,0 +1,112 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + "fmt" + + pilosa "github.com/featurebasedb/featurebase/v3" + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" + "github.com/pkg/errors" +) + +// PlanOpCreateDatabase is a plan operator that creates a database. +type PlanOpCreateDatabase struct { + planner *ExecutionPlanner + databaseName string + failIfExists bool + units int + description string + warnings []string +} + +// NewPlanOpCreateDatabase returns a new PlanOpCreateDatabase planoperator +func NewPlanOpCreateDatabase(p *ExecutionPlanner, databaseName string, failIfExists bool, units int, description string) *PlanOpCreateDatabase { + return &PlanOpCreateDatabase{ + planner: p, + databaseName: databaseName, + failIfExists: failIfExists, + units: units, + description: description, + warnings: make([]string, 0), + } +} + +func (p *PlanOpCreateDatabase) Plan() map[string]interface{} { + result := make(map[string]interface{}) + result["_op"] = fmt.Sprintf("%T", p) + result["name"] = p.databaseName + result["failIfExists"] = p.failIfExists + return result +} + +func (p *PlanOpCreateDatabase) String() string { + return "" +} + +func (p *PlanOpCreateDatabase) AddWarning(warning string) { + p.warnings = append(p.warnings, warning) +} + +func (p *PlanOpCreateDatabase) Warnings() []string { + return p.warnings +} + +func (p *PlanOpCreateDatabase) Schema() types.Schema { + return types.Schema{} +} + +func (p *PlanOpCreateDatabase) Children() []types.PlanOperator { + return []types.PlanOperator{} +} + +func (p *PlanOpCreateDatabase) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + return &createDatabaseRowIter{ + planner: p.planner, + databaseName: p.databaseName, + failIfExists: p.failIfExists, + units: p.units, + description: p.description, + }, nil +} + +func (p *PlanOpCreateDatabase) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { + return nil, nil +} + +type createDatabaseRowIter struct { + planner *ExecutionPlanner + databaseName string + failIfExists bool + units int + description string +} + +var _ types.RowIterator = (*createDatabaseRowIter)(nil) + +func (i *createDatabaseRowIter) Next(ctx context.Context) (types.Row, error) { + //create the database + db := &dax.Database{ + Name: dax.DatabaseName(i.databaseName), + Options: dax.DatabaseOptions{ + WorkersMin: i.units, + WorkersMax: i.units, + }, + + Description: i.description, + } + + if err := i.planner.schemaAPI.CreateDatabase(ctx, db); err != nil { + if _, ok := errors.Cause(err).(pilosa.ConflictError); ok { + if i.failIfExists { + return nil, sql3.NewErrDatabaseExists(0, 0, i.databaseName) + } + } else { + return nil, err + } + } + return nil, types.ErrNoMoreRows +} diff --git a/sql3/planner/opdropdatabase.go b/sql3/planner/opdropdatabase.go new file mode 100644 index 000000000..2420d8d67 --- /dev/null +++ b/sql3/planner/opdropdatabase.go @@ -0,0 +1,84 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + "fmt" + + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// PlanOpDropDatabase plan operator to drop a database. +type PlanOpDropDatabase struct { + planner *ExecutionPlanner + db *dax.Database + warnings []string +} + +func NewPlanOpDropDatabase(p *ExecutionPlanner, db *dax.Database) *PlanOpDropDatabase { + return &PlanOpDropDatabase{ + planner: p, + db: db, + warnings: make([]string, 0), + } +} + +func (p *PlanOpDropDatabase) Plan() map[string]interface{} { + result := make(map[string]interface{}) + result["_op"] = fmt.Sprintf("%T", p) + result["databaseName"] = p.db.Name + return result +} + +func (p *PlanOpDropDatabase) String() string { + return "" +} + +func (p *PlanOpDropDatabase) AddWarning(warning string) { + p.warnings = append(p.warnings, warning) +} + +func (p *PlanOpDropDatabase) Warnings() []string { + return p.warnings +} + +func (p *PlanOpDropDatabase) Schema() types.Schema { + return types.Schema{} +} + +func (p *PlanOpDropDatabase) Children() []types.PlanOperator { + return []types.PlanOperator{} +} + +func (p *PlanOpDropDatabase) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + return &dropDatabaseRowIter{ + planner: p.planner, + db: p.db, + }, nil +} + +func (p *PlanOpDropDatabase) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { + return nil, nil +} + +type dropDatabaseRowIter struct { + planner *ExecutionPlanner + db *dax.Database +} + +var _ types.RowIterator = (*dropDatabaseRowIter)(nil) + +func (i *dropDatabaseRowIter) Next(ctx context.Context) (types.Row, error) { + err := i.planner.checkAccess(ctx, string(i.db.Name), accessTypeDropObject) + if err != nil { + return nil, err + } + + err = i.planner.schemaAPI.DropDatabase(ctx, i.db.ID) + if err != nil { + return nil, err + } + return nil, types.ErrNoMoreRows +} diff --git a/sql3/planner/opfeaturebasedatabases.go b/sql3/planner/opfeaturebasedatabases.go new file mode 100644 index 000000000..838687d73 --- /dev/null +++ b/sql3/planner/opfeaturebasedatabases.go @@ -0,0 +1,135 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + "fmt" + "time" + + "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/sql3/parser" + "github.com/featurebasedb/featurebase/v3/sql3/planner/types" +) + +// PlanOpFeatureBaseDatabases wraps a []*Database that is returned from +// schemaAPI.Schema(). +type PlanOpFeatureBaseDatabases struct { + planner *ExecutionPlanner + dbs []*dax.Database + warnings []string +} + +func NewPlanOpFeatureBaseDatabases(planner *ExecutionPlanner, dbs []*dax.Database) *PlanOpFeatureBaseDatabases { + return &PlanOpFeatureBaseDatabases{ + planner: planner, + dbs: dbs, + warnings: make([]string, 0), + } +} + +func (p *PlanOpFeatureBaseDatabases) Plan() map[string]interface{} { + result := make(map[string]interface{}) + result["_op"] = fmt.Sprintf("%T", p) + result["_schema"] = p.Schema().Plan() + return result +} + +func (p *PlanOpFeatureBaseDatabases) String() string { + return "" +} + +func (p *PlanOpFeatureBaseDatabases) AddWarning(warning string) { + p.warnings = append(p.warnings, warning) +} + +func (p *PlanOpFeatureBaseDatabases) Warnings() []string { + return p.warnings +} + +func (p *PlanOpFeatureBaseDatabases) Schema() types.Schema { + return types.Schema{ + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "_id", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "name", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "owner", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "updated_by", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "created_at", + Type: parser.NewDataTypeTimestamp(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "updated_at", + Type: parser.NewDataTypeTimestamp(), + }, + &types.PlannerColumn{ + RelationName: "fb_databases", + ColumnName: "description", + Type: parser.NewDataTypeString(), + }, + } +} + +func (p *PlanOpFeatureBaseDatabases) Children() []types.PlanOperator { + return []types.PlanOperator{} +} + +func (p *PlanOpFeatureBaseDatabases) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + return &showDatabasesRowIter{ + planner: p.planner, + dbs: p.dbs, + }, nil +} + +func (p *PlanOpFeatureBaseDatabases) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { + return NewPlanOpFeatureBaseDatabases(p.planner, p.dbs), nil +} + +type showDatabasesRowIter struct { + planner *ExecutionPlanner + dbs []*dax.Database + rowIndex int +} + +var _ types.RowIterator = (*showDatabasesRowIter)(nil) + +func (i *showDatabasesRowIter) Next(ctx context.Context) (types.Row, error) { + if i.rowIndex < len(i.dbs) { + + dbID := i.dbs[i.rowIndex].ID + dbName := i.dbs[i.rowIndex].Name + + createdAt := time.Unix(0, i.dbs[i.rowIndex].CreatedAt) + updatedAt := time.Unix(0, i.dbs[i.rowIndex].UpdatedAt) + row := []interface{}{ + dbID, + dbName, + i.dbs[i.rowIndex].Owner, + i.dbs[i.rowIndex].UpdatedBy, + createdAt.Format(time.RFC3339), + updatedAt.Format(time.RFC3339), + i.dbs[i.rowIndex].Options.WorkersMin, + i.dbs[i.rowIndex].Description, + } + i.rowIndex += 1 + return row, nil + } + return nil, types.ErrNoMoreRows +} diff --git a/sql3/sql_complex_test.go b/sql3/sql_complex_test.go index 5e496907f..7a31913c3 100644 --- a/sql3/sql_complex_test.go +++ b/sql3/sql_complex_test.go @@ -194,6 +194,32 @@ func TestPlanner_Show(t *testing.T) { } }) + t.Run("ShowDatabases", func(t *testing.T) { + results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `SHOW DATABASES`) + if err != nil { + t.Fatal(err) + } + + // we don't currently support databases in the on-prem implementation, + // so we expect this to be empty for now. + if len(results) != 0 { + t.Fatal(fmt.Errorf("unexpected result set length")) + } + + if diff := cmp.Diff([]*pilosa.WireQueryField{ + wireQueryFieldString("_id"), + wireQueryFieldString("name"), + wireQueryFieldString("owner"), + wireQueryFieldString("updated_by"), + wireQueryFieldTimestamp("created_at"), + wireQueryFieldTimestamp("updated_at"), + wireQueryFieldInt("units"), + wireQueryFieldString("description"), + }, columns); diff != "" { + t.Fatal(diff) + } + }) + t.Run("ShowTables", func(t *testing.T) { results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `SHOW TABLES`) if err != nil {