mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-09 20:37:52 +00:00
Switch Serverless from using BoltDB to Postgres as metadata store. Previously, the controller stored all metadata to BoltDB. This implements SQLDB (currently Postgres flavored) as the backing store for metadata. This will allow us to have multiple instances of the controller running for HA, and to easily inspect and repair the contents of the metadata store. Unfortunately, it was not straightforward to keep the BoltDB implementation working alongside the SQL one, so it will be removed in a later patch. Once that's done, the SQL implementation should allow for a number of simplifications of the schemar and balancer interfaces. Database migration is built directly into the application by embedding the migration files and logic from the `soda` command line tool. When connecting to the RDBMS, the app will always attempt to create the necessary database and apply any outstanding migrations. Integration tests truncate all tables upon start, but *not* at the end, so the state of the database can be inspected after integration tests. Had to refactor some of the controller's background tasks to make sure they get properly shut down on controller exit.
698 lines
20 KiB
Go
698 lines
20 KiB
Go
// Package boltdb contains the boltdb implementation of the Schemar
|
|
// interfaces.
|
|
package boltdb
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/dax/boltdb"
|
|
"github.com/featurebasedb/featurebase/v3/dax/controller/schemar"
|
|
"github.com/featurebasedb/featurebase/v3/errors"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
)
|
|
|
|
var (
|
|
bucketSchemar = boltdb.Bucket("schemar")
|
|
)
|
|
|
|
// SchemarBuckets defines the buckets used by this package. It can be called
|
|
// during setup to create the buckets ahead of time.
|
|
var SchemarBuckets []boltdb.Bucket = []boltdb.Bucket{
|
|
bucketSchemar,
|
|
}
|
|
|
|
// Ensure type implements interface.
|
|
var _ schemar.Schemar = (*Schemar)(nil)
|
|
|
|
type Schemar struct {
|
|
db *boltdb.DB
|
|
|
|
logger logger.Logger
|
|
}
|
|
|
|
// NewSchemar returns a new instance of Schemar with default values.
|
|
func NewSchemar(db *boltdb.DB, logger logger.Logger) *Schemar {
|
|
return &Schemar{
|
|
db: db,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// CreateDatabase creates the database provided. If a database with the same
|
|
// name already exists then an error is returned. For now, we are not going to
|
|
// store the tables in the schemar Database struct.
|
|
func (s *Schemar) CreateDatabase(tx dax.Transaction, qdb *dax.QualifiedDatabase) error {
|
|
// Ensure the database id is not blank.
|
|
if qdb.ID == "" {
|
|
return schemar.NewErrDatabaseIDInvalid(qdb.ID)
|
|
}
|
|
|
|
// Ensure the database name is not blank.
|
|
if qdb.Name == "" {
|
|
return schemar.NewErrDatabaseNameInvalid(qdb.Name)
|
|
}
|
|
|
|
// Set the CreateAt value for the database.
|
|
// TODO(tlt): We may want to consider erroring here if the value is != 0.
|
|
if qdb.CreatedAt == 0 {
|
|
now := timestamp()
|
|
qdb.CreatedAt = now
|
|
}
|
|
|
|
//////////// end validation
|
|
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Ensure a database with that ID doesn't already exist.
|
|
if db, _ := s.databaseByID(txx, qdb.OrganizationID, qdb.ID); db != nil {
|
|
return dax.NewErrDatabaseIDExists(qdb.QualifiedID())
|
|
}
|
|
|
|
if err := s.putDatabase(txx, qdb); err != nil {
|
|
return errors.Wrap(err, "putting database")
|
|
}
|
|
|
|
// In addition to storing the database in databaseKey, we want to store a
|
|
// reverse-lookup (i.e. index) on database name to the databaseKey.
|
|
if err := s.putDatabaseName(txx, qdb); err != nil {
|
|
return errors.Wrap(err, "putting database name")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Schemar) DatabaseByID(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return nil, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.databaseByID(txx, qdbid.OrganizationID, qdbid.DatabaseID)
|
|
}
|
|
|
|
func (s *Schemar) databaseByID(tx *boltdb.Tx, orgID dax.OrganizationID, id dax.DatabaseID) (*dax.QualifiedDatabase, error) {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return nil, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
b := bkt.Get(databaseKey(orgID, id))
|
|
if b == nil {
|
|
return nil, dax.NewErrDatabaseIDDoesNotExist(dax.QualifiedDatabaseID{OrganizationID: orgID, DatabaseID: id})
|
|
}
|
|
|
|
database := &dax.QualifiedDatabase{}
|
|
if err := json.Unmarshal(b, database); err != nil {
|
|
return nil, errors.Wrap(err, "unmarshalling database json")
|
|
}
|
|
|
|
return database, nil
|
|
}
|
|
|
|
func (s *Schemar) DatabaseByName(tx dax.Transaction, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return nil, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.databaseByName(txx, orgID, dbname)
|
|
}
|
|
|
|
func (s *Schemar) databaseByName(tx *boltdb.Tx, orgID dax.OrganizationID, name dax.DatabaseName) (*dax.QualifiedDatabase, error) {
|
|
qdbid, err := s.databaseIDByName(tx, orgID, name)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting database ID")
|
|
}
|
|
|
|
return s.databaseByID(tx, orgID, qdbid.DatabaseID)
|
|
}
|
|
|
|
func (s *Schemar) databaseIDByName(tx *boltdb.Tx, orgID dax.OrganizationID, name dax.DatabaseName) (dax.QualifiedDatabaseID, error) {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return dax.QualifiedDatabaseID{}, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
b := bkt.Get(databaseNameKey(orgID, name))
|
|
if b == nil {
|
|
return dax.QualifiedDatabaseID{}, dax.NewErrDatabaseNameDoesNotExist(name)
|
|
}
|
|
|
|
return keyQualifiedDatabaseID(b)
|
|
}
|
|
|
|
func (s *Schemar) putDatabase(tx *boltdb.Tx, qdb *dax.QualifiedDatabase) error {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
val, err := json.Marshal(qdb)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling database to json")
|
|
}
|
|
|
|
return bkt.Put(databaseKey(qdb.OrganizationID, qdb.ID), val)
|
|
}
|
|
|
|
func (s *Schemar) putDatabaseName(tx *boltdb.Tx, qdb *dax.QualifiedDatabase) error {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
return bkt.Put(databaseNameKey(qdb.OrganizationID, qdb.Name), databaseKey(qdb.OrganizationID, qdb.ID))
|
|
}
|
|
|
|
// DropDatabase drops the given database. If the named/IDed database does not
|
|
// exist then an error is returned.
|
|
func (s *Schemar) DropDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) error {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Ensure the database exists.
|
|
qdb, err := s.databaseByID(txx, qdbid.OrganizationID, qdbid.DatabaseID)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting database by id")
|
|
}
|
|
|
|
bkt := txx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
// Delete the database by ID.
|
|
if err := bkt.Delete(databaseKey(qdb.OrganizationID, qdb.ID)); err != nil {
|
|
return errors.Wrap(err, "deleting database by id")
|
|
}
|
|
|
|
// Delete the reverse-lookup database by Name.
|
|
if err := bkt.Delete(databaseNameKey(qdb.OrganizationID, qdb.Name)); err != nil {
|
|
return errors.Wrap(err, "deleting database by name")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// 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("*boltdb.Tx")
|
|
}
|
|
|
|
// Get the database.
|
|
qdb, err := s.databaseByID(txx, qdbid.OrganizationID, qdbid.DatabaseID)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "getting database: %s", qdbid)
|
|
}
|
|
|
|
// 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 {
|
|
return errors.Wrap(err, "putting database")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Schemar) Databases(tx dax.Transaction, orgID dax.OrganizationID, ids ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return nil, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.getDatabases(txx, orgID, ids...)
|
|
}
|
|
|
|
func (s *Schemar) getDatabases(tx *boltdb.Tx, orgID dax.OrganizationID, ids ...dax.DatabaseID) (dax.QualifiedDatabases, error) {
|
|
c := tx.Bucket(bucketSchemar).Cursor()
|
|
|
|
// Deserialize rows into Database objects.
|
|
databases := make(dax.QualifiedDatabases, 0)
|
|
|
|
var filterByID bool
|
|
if len(ids) > 0 {
|
|
filterByID = true
|
|
}
|
|
|
|
prefix := []byte(fmt.Sprintf(prefixFmtDatabases, orgID))
|
|
if orgID == "" {
|
|
prefix = []byte(prefixDatabases)
|
|
}
|
|
|
|
for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
|
|
if v == nil {
|
|
s.logger.Printf("nil value for key: %s", k)
|
|
continue
|
|
}
|
|
|
|
dbID, err := keyDatabaseID(k)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting database from key")
|
|
}
|
|
|
|
// Only include databases provided in the ids filter.
|
|
if filterByID && !containsDatabaseID(ids, dbID) {
|
|
continue
|
|
}
|
|
|
|
database := &dax.QualifiedDatabase{}
|
|
if err := json.Unmarshal(v, database); err != nil {
|
|
return nil, errors.Wrap(err, "unmarshalling database json")
|
|
}
|
|
|
|
databases = append(databases, database)
|
|
}
|
|
|
|
return databases, nil
|
|
}
|
|
|
|
func containsDatabaseID(s []dax.DatabaseID, e dax.DatabaseID) bool {
|
|
for _, a := range s {
|
|
if a == e {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// CreateTable creates the table provided. If a table with the same name already
|
|
// exists then an error is returned.
|
|
func (s *Schemar) CreateTable(tx dax.Transaction, qtbl *dax.QualifiedTable) error {
|
|
// Ensure the table id is not blank.
|
|
if qtbl.ID == "" {
|
|
return schemar.NewErrTableIDInvalid(qtbl.ID)
|
|
}
|
|
|
|
// Ensure the table name is not blank.
|
|
if qtbl.Name == "" {
|
|
return schemar.NewErrTableNameInvalid(qtbl.Name)
|
|
}
|
|
|
|
// Ensure that a primary key field is present and valid.
|
|
if !qtbl.HasValidPrimaryKey() {
|
|
return schemar.NewErrInvalidPrimaryKey()
|
|
}
|
|
|
|
// Set the CreateAt value for the table.
|
|
// TODO(tlt): We may want to consider erroring here if the value is != 0.
|
|
if qtbl.CreatedAt == 0 {
|
|
now := timestamp()
|
|
qtbl.CreatedAt = now
|
|
|
|
// Set CreatedAt for all of the fields as well.
|
|
for i := range qtbl.Fields {
|
|
qtbl.Fields[i].CreatedAt = now
|
|
}
|
|
}
|
|
|
|
//////////// end validation
|
|
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Ensure the database, defined in the table's QualifiedDatabaseID, exists.
|
|
if _, err := s.databaseByID(txx, qtbl.OrganizationID, qtbl.DatabaseID); err != nil {
|
|
return errors.Wrap(err, "validating database")
|
|
}
|
|
|
|
// Ensure a table with that ID doesn't already exist.
|
|
if t, _ := s.tableByID(txx, qtbl.QualifiedDatabaseID, qtbl.ID); t != nil {
|
|
return dax.NewErrTableIDExists(qtbl.QualifiedID())
|
|
}
|
|
|
|
if err := s.putTable(txx, qtbl); err != nil {
|
|
return errors.Wrap(err, "putting table")
|
|
}
|
|
|
|
// In addition to storing the table in tableKey, we want to store a reverse-lookup
|
|
// (i.e. index) on table name to the tableKey.
|
|
if err := s.putTableName(txx, qtbl); err != nil {
|
|
return errors.Wrap(err, "putting table name")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// CreateField creates the field provided in the given table. If a field with
|
|
// the same name already exists then an error is returned.
|
|
func (s *Schemar) CreateField(tx dax.Transaction, qtid dax.QualifiedTableID, fld *dax.Field) error {
|
|
// Ensure the field name is not blank.
|
|
if fld.Name == "" {
|
|
return schemar.NewErrFieldNameInvalid(fld.Name)
|
|
}
|
|
|
|
//////////// end validation
|
|
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Get the table.
|
|
qtbl, err := s.tableByQTID(txx, qtid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting table by id")
|
|
}
|
|
|
|
// Ensure a field with that name doesn't already exist.
|
|
if _, ok := qtbl.Field(fld.Name); ok {
|
|
return dax.NewErrFieldExists(fld.Name)
|
|
}
|
|
|
|
qtbl.Fields = append(qtbl.Fields, fld)
|
|
|
|
// Write table back to database.
|
|
if err := s.putTable(txx, qtbl); err != nil {
|
|
return errors.Wrap(err, "putting table")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// DropField removes the field from the table.
|
|
func (s *Schemar) DropField(tx dax.Transaction, qtid dax.QualifiedTableID, fldName dax.FieldName) error {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Get the table.
|
|
qtbl, err := s.tableByQTID(txx, qtid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting table by id")
|
|
}
|
|
|
|
// Ensure a field with that name exists.
|
|
if _, ok := qtbl.Field(fldName); !ok {
|
|
return dax.NewErrFieldDoesNotExist(fldName)
|
|
}
|
|
|
|
_ = qtbl.RemoveField(fldName)
|
|
|
|
// Write table back to database.
|
|
if err := s.putTable(txx, qtbl); err != nil {
|
|
return errors.Wrap(err, "putting table")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Schemar) putTable(tx *boltdb.Tx, qtbl *dax.QualifiedTable) error {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
val, err := json.Marshal(qtbl)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling table to json")
|
|
}
|
|
|
|
return bkt.Put(tableKey(qtbl.OrganizationID, qtbl.DatabaseID, qtbl.Table.ID), val)
|
|
}
|
|
|
|
func (s *Schemar) putTableName(tx *boltdb.Tx, qtbl *dax.QualifiedTable) error {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
return bkt.Put(tableNameKey(qtbl.OrganizationID, qtbl.DatabaseID, qtbl.Name), tableKey(qtbl.OrganizationID, qtbl.DatabaseID, qtbl.ID))
|
|
}
|
|
|
|
// Table returns the TableInfo for the given table. An error is returned if the
|
|
// table does not exist.
|
|
func (s *Schemar) Table(tx dax.Transaction, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return nil, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.tableByQTID(txx, qtid)
|
|
}
|
|
|
|
// tableByQTID gets the full qualified table by the QualifiedTableID whether it
|
|
// has Name or ID set.
|
|
func (s *Schemar) tableByQTID(tx *boltdb.Tx, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) {
|
|
if qtid.ID == "" {
|
|
return s.tableByName(tx, qtid.QualifiedDatabaseID, qtid.Name)
|
|
}
|
|
|
|
return s.tableByID(tx, qtid.QualifiedDatabaseID, qtid.ID)
|
|
}
|
|
|
|
func (s *Schemar) tableByName(tx *boltdb.Tx, qdbid dax.QualifiedDatabaseID, name dax.TableName) (*dax.QualifiedTable, error) {
|
|
qtid, err := s.tableIDByName(tx, qdbid, name)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting table ID")
|
|
}
|
|
|
|
return s.tableByID(tx, qtid.QualifiedDatabaseID, qtid.ID) // TODO remove?
|
|
}
|
|
|
|
func (s *Schemar) tableByID(tx *boltdb.Tx, qdbid dax.QualifiedDatabaseID, id dax.TableID) (*dax.QualifiedTable, error) {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return nil, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
b := bkt.Get(tableKey(qdbid.OrganizationID, qdbid.DatabaseID, id))
|
|
if b == nil {
|
|
return nil, dax.NewErrTableIDDoesNotExist(dax.QualifiedTableID{QualifiedDatabaseID: qdbid, ID: id})
|
|
}
|
|
|
|
table := &dax.QualifiedTable{}
|
|
if err := json.Unmarshal(b, table); err != nil {
|
|
return nil, errors.Wrap(err, "unmarshalling table json")
|
|
}
|
|
|
|
return table, nil
|
|
}
|
|
|
|
func (s *Schemar) tableIDByName(tx *boltdb.Tx, qdbid dax.QualifiedDatabaseID, name dax.TableName) (dax.QualifiedTableID, error) {
|
|
bkt := tx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return dax.QualifiedTableID{}, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
b := bkt.Get(tableNameKey(qdbid.OrganizationID, qdbid.DatabaseID, name))
|
|
if b == nil {
|
|
return dax.QualifiedTableID{}, dax.NewErrTableNameDoesNotExist(name)
|
|
}
|
|
|
|
return keyQualifiedTableID(b)
|
|
}
|
|
|
|
// Tables returns a list of Table for all existing tables. If one or more table
|
|
// IDs is provided, then only those will be included in the output.
|
|
func (s *Schemar) Tables(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, ids ...dax.TableID) ([]*dax.QualifiedTable, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return nil, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.getTables(txx, qdbid, ids...)
|
|
}
|
|
|
|
func (s *Schemar) getTables(tx *boltdb.Tx, qdbid dax.QualifiedDatabaseID, ids ...dax.TableID) (dax.QualifiedTables, error) {
|
|
c := tx.Bucket(bucketSchemar).Cursor()
|
|
|
|
// Deserialize rows into Table objects.
|
|
tables := make(dax.QualifiedTables, 0)
|
|
|
|
var filterByID bool
|
|
if len(ids) > 0 {
|
|
filterByID = true
|
|
}
|
|
|
|
prefix := []byte(fmt.Sprintf(prefixFmtTables, qdbid.OrganizationID, qdbid.DatabaseID))
|
|
if qdbid.OrganizationID == "" && qdbid.DatabaseID == "" {
|
|
prefix = []byte(prefixTables)
|
|
} else if qdbid.DatabaseID == "" {
|
|
prefix = []byte(fmt.Sprintf(prefixFmtTablesOrg, qdbid.OrganizationID))
|
|
}
|
|
|
|
for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
|
|
if v == nil {
|
|
s.logger.Printf("nil value for key: %s", k)
|
|
continue
|
|
}
|
|
|
|
tblID, err := keyTableID(k)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting table from key")
|
|
}
|
|
|
|
// Only include tables provided in the ids filter.
|
|
if filterByID && !containsTableID(ids, tblID) {
|
|
continue
|
|
}
|
|
|
|
table := &dax.QualifiedTable{}
|
|
if err := json.Unmarshal(v, table); err != nil {
|
|
return nil, errors.Wrap(err, "unmarshalling table json")
|
|
}
|
|
|
|
tables = append(tables, table)
|
|
}
|
|
|
|
return tables, nil
|
|
}
|
|
|
|
func containsTableID(s []dax.TableID, e dax.TableID) bool {
|
|
for _, a := range s {
|
|
if a == e {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// DropTable drops the given table. If the named/IDed table does not exist
|
|
// then an error is returned.
|
|
func (s *Schemar) DropTable(tx dax.Transaction, qtid dax.QualifiedTableID) error {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
// Ensure the table exists.
|
|
qtbl, err := s.tableByQTID(txx, qtid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting table by id")
|
|
}
|
|
|
|
bkt := txx.Bucket(bucketSchemar)
|
|
if bkt == nil {
|
|
return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar)
|
|
}
|
|
|
|
// Delete the table by ID.
|
|
if err := bkt.Delete(tableKey(qtbl.OrganizationID, qtbl.DatabaseID, qtbl.ID)); err != nil {
|
|
return errors.Wrap(err, "deleting table by id")
|
|
}
|
|
|
|
// Delete the reverse-lookup table by Name.
|
|
if err := bkt.Delete(tableNameKey(qtbl.OrganizationID, qtbl.DatabaseID, qtbl.Name)); err != nil {
|
|
return errors.Wrap(err, "deleting table by name")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Schemar) TableID(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, name dax.TableName) (dax.QualifiedTableID, error) {
|
|
txx, ok := tx.(*boltdb.Tx)
|
|
if !ok {
|
|
return dax.QualifiedTableID{}, dax.NewErrInvalidTransaction("*boltdb.Tx")
|
|
}
|
|
|
|
return s.tableIDByName(txx, qdbid, name)
|
|
}
|
|
|
|
const (
|
|
prefixTables = "tables/"
|
|
prefixFmtTablesOrg = prefixTables + "%s/" // org-id
|
|
prefixFmtTables = prefixFmtTablesOrg + "%s/" // db-id
|
|
prefixFmtTableNames = "tablenames/%s/%s/" // org-id, db-id
|
|
|
|
prefixDatabases = "databases/"
|
|
prefixFmtDatabases = prefixDatabases + "%s/" // org-id
|
|
prefixFmtDatabase = prefixFmtDatabases + "%s" // db-id
|
|
prefixFmtDatabaseNames = "databasenames/%s/" // org-id
|
|
)
|
|
|
|
// databaseKey returns a key based on a qualified database ID.
|
|
func databaseKey(orgID dax.OrganizationID, dbID dax.DatabaseID) []byte {
|
|
key := fmt.Sprintf(prefixFmtDatabase, orgID, dbID)
|
|
return []byte(key)
|
|
}
|
|
|
|
// databaseNameKey returns a key based on a qualified database name.
|
|
func databaseNameKey(orgID dax.OrganizationID, name dax.DatabaseName) []byte {
|
|
key := fmt.Sprintf(prefixFmtDatabaseNames+"%s", orgID, name)
|
|
return []byte(key)
|
|
}
|
|
|
|
// keyDatabaseID gets the DatabaseID out of the key.
|
|
func keyDatabaseID(key []byte) (dax.DatabaseID, error) {
|
|
parts := strings.Split(string(key), "/")
|
|
if len(parts) != 3 {
|
|
return "", errors.New(errors.ErrUncoded, "database key format expected: `databases/orgID/dbID`")
|
|
}
|
|
|
|
return dax.DatabaseID(parts[2]), nil
|
|
}
|
|
|
|
// tableKey returns a key based on a qualified table ID.
|
|
func tableKey(orgID dax.OrganizationID, dbID dax.DatabaseID, tblID dax.TableID) []byte {
|
|
key := fmt.Sprintf(prefixFmtTables+"%s", orgID, dbID, tblID)
|
|
return []byte(key)
|
|
}
|
|
|
|
// tableNameKey returns a key based on a qualified table name.
|
|
func tableNameKey(orgID dax.OrganizationID, dbID dax.DatabaseID, name dax.TableName) []byte {
|
|
key := fmt.Sprintf(prefixFmtTableNames+"%s", orgID, dbID, name)
|
|
return []byte(key)
|
|
}
|
|
|
|
// keyTableID gets the TableID out of the key.
|
|
func keyTableID(key []byte) (dax.TableID, error) {
|
|
parts := strings.Split(string(key), "/")
|
|
if len(parts) != 4 {
|
|
return "", errors.New(errors.ErrUncoded, "table key format expected: `tables/orgID/dbID/tblID`")
|
|
}
|
|
|
|
return dax.TableID(parts[3]), nil
|
|
}
|
|
|
|
// keyQualifedTableID gets the QualifiedTableID out of the key.
|
|
func keyQualifiedTableID(key []byte) (dax.QualifiedTableID, error) {
|
|
parts := strings.Split(string(key), "/")
|
|
if len(parts) != 4 {
|
|
return dax.QualifiedTableID{}, errors.New(errors.ErrUncoded, "table key format expected: `tables/orgID/dbID/tblID`")
|
|
}
|
|
|
|
return dax.NewQualifiedTableID(
|
|
dax.NewQualifiedDatabaseID(
|
|
dax.OrganizationID(parts[1]),
|
|
dax.DatabaseID(parts[2]),
|
|
),
|
|
dax.TableID(parts[3]),
|
|
), nil
|
|
}
|
|
|
|
// keyQualifedDatabaseID gets the QualifiedDatabaseID out of the key.
|
|
func keyQualifiedDatabaseID(key []byte) (dax.QualifiedDatabaseID, error) {
|
|
parts := strings.Split(string(key), "/")
|
|
if len(parts) != 3 {
|
|
return dax.QualifiedDatabaseID{}, errors.New(errors.ErrUncoded, "table key format expected: `databases/orgID/dbID`")
|
|
}
|
|
|
|
return dax.NewQualifiedDatabaseID(
|
|
dax.OrganizationID(parts[1]),
|
|
dax.DatabaseID(parts[2]),
|
|
), nil
|
|
}
|
|
|
|
func timestamp() int64 {
|
|
return time.Now().Unix()
|
|
}
|