scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Convert SchemaAPI interface to use dax.Table instead of pilosa.IndexInfo (#2336)

Browse source 
* WIP: Convert SchemaAPI to be DAX-centric

* Tables(), CreateField()

* CreateTable(), DeleteTable(), DeleteField()

* More cleanup

* Remove the old SchemaAPI
This commit is contained in:
Travis Turner 2022-12-08 11:35:17 -06:00 • committed by GitHub
parent d33bf4811f
commit a15783cb49
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
30 changed files with 924 additions and 839 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

91
api.go
View file

@ -3373,25 +3373,16 @@ func shardInShards(i dax.ShardNum, s dax.VersionedShards) bool {
return false
}
// SchemaAPI is a subset of the API methods which have to do with schema. This
// interface was introduced in order to remove, from the sql3 package, the
// pointer to API, and instead use this interface. In the current FeatureBase,
// this interface can be implemented directly with API (well, not directly, but
// with FeatureBaseSchemaAPI, which is a wrapper around API). But in an
// implementation for DAX, for example, we might want something else servicing
// the schema-related calls to the SchemaAPI.
type SchemaAPI interface {
CreateIndexAndFields(ctx context.Context, indexName string, options IndexOptions, fields []CreateFieldObj) error
CreateField(ctx context.Context, indexName string, fieldName string, opts ...FieldOption) (*Field, error)
DeleteField(ctx context.Context, indexName string, fieldName string) error
DeleteIndex(ctx context.Context, indexName string) 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)
// Schema returns the list of tables and fields. While it might make sense
// to have this as part of the SchemaInfoAPI interface instead of here, it's
// never used by consumers of that interface.
Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error)
CreateTable(ctx context.Context, tbl *dax.Table) error
CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error
SchemaInfoAPI
DeleteTable(ctx context.Context, tname dax.TableName) error
DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error
}
type SchemaInfoAPI interface {
@ -3432,74 +3423,6 @@ type QueryAPI interface {
Query(ctx context.Context, req *QueryRequest) (QueryResponse, error)
}
// Ensure type implements interface.
var _ SchemaAPI = (*FeatureBaseSchemaAPI)(nil)
// FeatureBaseSchemaAPI is a wrapper around pilosa.API. It implements the
// SchemaAPI interface with methods which are not a part of pilosa.API.
type FeatureBaseSchemaAPI struct {
*API
}
func (fapi *FeatureBaseSchemaAPI) CreateIndexAndFields(ctx context.Context, indexName string, options IndexOptions, fields []CreateFieldObj) error {
// Add the index.
if _, err := fapi.CreateIndex(ctx, indexName, options); err != nil {
return err
}
// Now add fields.
for _, f := range fields {
if _, err := fapi.CreateField(ctx, indexName, f.Name, f.Options...); err != nil {
return err
}
}
return nil
}
// IndexInfo wraps the API.IndexInfo method and prepends an _id field to its
// list of fields.
func (fapi *FeatureBaseSchemaAPI) IndexInfo(ctx context.Context, indexName string) (*IndexInfo, error) {
idx, err := fapi.API.IndexInfo(ctx, indexName)
if err != nil {
return nil, err
}
// sortedFields will contain the sorted list of fields from IndexInfo, along
// with the primary key field (which will always be at the beginning of the
// list).
sortedFields := make([]*FieldInfo, 0, len(idx.Fields)+1)
// Add the primary key field to the beginning of the list.
idKeys := idx.Options.Keys
idType := "id"
if idKeys {
idType = "string"
}
idFld := &FieldInfo{
Name: "_id",
CreatedAt: idx.CreatedAt,
Options: FieldOptions{
Type: idType,
Keys: idKeys,
},
}
sortedFields = append(sortedFields, idFld)
// Sort idx.Fields by CreatedAt before adding them to sortedFields.
sort.Slice(idx.Fields, func(i, j int) bool {
return idx.Fields[i].CreatedAt < idx.Fields[j].CreatedAt
})
// Add the sorted fields to sortedFields.
sortedFields = append(sortedFields, idx.Fields...)
idx.Fields = sortedFields
return idx, nil
}
// FeatureBaseSystemAPI is a wrapper around pilosa.API. It implements the
// SystemAPI interface
type FeatureBaseSystemAPI struct {

53
api_directive.go
View file

@ -924,57 +924,10 @@ func (api *API) createTableAndFields(tbl *dax.QualifiedTable, partitions dax.Ver
// createField creates a FeatureBase Field in the provided FeatureBase Index
// based on the provided field's type.
//
// TODO: `time` fields
func createField(idx *Index, fld *dax.Field) error {
// Set the cache type and size (or use default) for those fields which
// require them.
cacheType := DefaultCacheType
cacheSize := uint32(DefaultCacheSize)
if fld.Options.CacheType != "" {
cacheType = fld.Options.CacheType
cacheSize = fld.Options.CacheSize
}
opts := []FieldOption{}
switch fld.Type {
case dax.BaseTypeBool:
opts = append(opts,
OptFieldTypeBool(),
)
case dax.BaseTypeDecimal:
opts = append(opts,
OptFieldTypeDecimal(fld.Options.Scale),
)
case dax.BaseTypeID:
opts = append(opts,
OptFieldTypeMutex(cacheType, cacheSize),
)
case dax.BaseTypeIDSet:
opts = append(opts,
OptFieldTypeSet(cacheType, cacheSize),
)
case dax.BaseTypeInt:
opts = append(opts,
OptFieldTypeInt(fld.Options.Min.ToInt64(0), fld.Options.Max.ToInt64(0)),
)
case dax.BaseTypeString:
opts = append(opts,
OptFieldTypeMutex(cacheType, cacheSize),
OptFieldKeys(),
)
case dax.BaseTypeStringSet:
opts = append(opts,
OptFieldTypeSet(cacheType, cacheSize),
OptFieldKeys(),
)
case dax.BaseTypeTimestamp:
opts = append(opts,
OptFieldTypeTimestamp(fld.Options.Epoch, fld.Options.TimeUnit),
)
default:
return errors.Errorf("unsupport field type: %s", fld.Type)
opts, err := FieldOptionsFromField(fld)
if err != nil {
return errors.Wrapf(err, "creating field options from field: %s", fld.Name)
}
if _, err := idx.CreateField(string(fld.Name), "", opts...); err != nil {

2
batch/Makefile
View file

@ -33,7 +33,7 @@ save-%-logs:
TCMD ?= ./...
# do "make startup", then e.g. "make test-run-local TCMD='-run=MyFavTest ./kafka'"
test-run-local:
test-run-local: vendor
pwd
$(DOCKER_COMPOSE) build batch-test
$(DOCKER_COMPOSE) run -T batch-test go test -mod=vendor -tags=odbc,dynamic $(TCMD)

2
batch/batch.go
View file

@ -183,8 +183,6 @@ type Batch struct {
clearFrags fragments
useShardTransactionalEndpoint bool
mdsHost string
}
func (b *Batch) Len() int { return len(b.ids) }

132
batch/batch_test.go
View file

@ -70,9 +70,10 @@ func testStringSliceCombos(t *testing.T, importer Importer, sapi featurebase.Sch
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 5, idx, idx.Fields)
@ -251,9 +252,10 @@ func testImportBatchInts(t *testing.T, importer Importer, sapi featurebase.Schem
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields)
@ -334,9 +336,10 @@ func testImportBatchSorting(t *testing.T, importer Importer, sapi featurebase.Sc
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 100, idx, idx.Fields)
@ -399,9 +402,10 @@ func testTrimNull(t *testing.T, importer Importer, sapi featurebase.SchemaAPI, q
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields)
@ -516,9 +520,10 @@ func testStringSliceEmptyAndNil(t *testing.T, importer Importer, sapi featurebas
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
// first create a batch and test adding a single value with empty
@ -641,9 +646,10 @@ func testStringSlice(t *testing.T, importer Importer, sapi featurebase.SchemaAPI
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields)
@ -765,9 +771,10 @@ func testSingleClearBatchRegression(t *testing.T, importer Importer, sapi featur
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
// Set a bit.
@ -856,9 +863,10 @@ func testBatches(t *testing.T, importer Importer, sapi featurebase.SchemaAPI, qa
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 10, idx, idx.Fields)
@ -1295,9 +1303,10 @@ func testBatchesStringIDs(t *testing.T, importer Importer, sapi featurebase.Sche
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields)
@ -1579,9 +1588,10 @@ func testBatchStaleness(t *testing.T, importer Importer, sapi featurebase.Schema
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields, OptMaxStaleness(time.Millisecond))
@ -1625,9 +1635,10 @@ func testImportBatchMultipleInts(t *testing.T, importer Importer, sapi featureba
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 6, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -1685,9 +1696,10 @@ func testImportBatchMultipleTimestamps(t *testing.T, importer Importer, sapi fea
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b1, err := NewBatch(importer, 6, idx, idx.Fields)
@ -1778,9 +1790,10 @@ func testImportBatchSetsAndClears(t *testing.T, importer Importer, sapi featureb
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 6, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -1867,9 +1880,10 @@ func testTopNCacheRegression(t *testing.T, importer Importer, sapi featurebase.S
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -1949,9 +1963,10 @@ func testMultipleIntSameBatch(t *testing.T, importer Importer, sapi featurebase.
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 4, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -2018,9 +2033,10 @@ func mutexClearRegression(t *testing.T, importer Importer, sapi featurebase.Sche
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 11, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -2098,9 +2114,10 @@ func mutexNilClearID(t *testing.T, importer Importer, sapi featurebase.SchemaAPI
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 11, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -2187,9 +2204,10 @@ func mutexNilClearKey(t *testing.T, importer Importer, sapi featurebase.SchemaAP
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields)
@ -2284,9 +2302,10 @@ func testImportBatchBools(t *testing.T, importer Importer, sapi featurebase.Sche
},
}
createIndexAndFields(t, ctx, sapi, idx)
tbl := featurebase.IndexInfoToTable(idx)
assert.NoError(t, sapi.CreateTable(ctx, tbl))
defer func() {
assert.NoError(t, sapi.DeleteIndex(ctx, idx.Name))
assert.NoError(t, sapi.DeleteTable(ctx, tbl.Name))
}()
b, err := NewBatch(importer, 3, idx, idx.Fields, OptUseShardTransactionalEndpoint(true))
@ -2322,40 +2341,3 @@ func testImportBatchBools(t *testing.T, importer Importer, sapi featurebase.Sche
assert.True(t, ok, "wrong return type: %T", resp.Results[0])
assert.Equal(t, uint64(2), count)
}
func createIndexAndFields(t *testing.T, ctx context.Context, sapi featurebase.SchemaAPI, idx *featurebase.IndexInfo) {
fields := make([]featurebase.CreateFieldObj, 0, len(idx.Fields))
for _, fld := range idx.Fields {
opts := []featurebase.FieldOption{}
if fld.Options.Keys {
opts = append(opts, featurebase.OptFieldKeys())
}
switch fld.Options.Type {
case featurebase.FieldTypeMutex:
opts = append(opts, featurebase.OptFieldTypeMutex(fld.Options.CacheType, fld.Options.CacheSize))
case featurebase.FieldTypeSet:
opts = append(opts, featurebase.OptFieldTypeSet(fld.Options.CacheType, fld.Options.CacheSize))
case featurebase.FieldTypeInt:
opts = append(opts, featurebase.OptFieldTypeInt(fld.Options.Min.ToInt64(0), fld.Options.Max.ToInt64(0)))
case featurebase.FieldTypeTime:
opts = append(opts, featurebase.OptFieldTypeTime(fld.Options.TimeQuantum, fld.Options.TTL.String(), fld.Options.NoStandardView))
case featurebase.FieldTypeTimestamp:
opts = append(opts, featurebase.OptFieldTypeTimestamp(time.Unix(0, 0), fld.Options.TimeUnit))
case featurebase.FieldTypeBool:
opts = append(opts, featurebase.OptFieldTypeBool())
default:
t.Fatalf("unsupported field type: %s", fld.Options.Type)
}
field := featurebase.CreateFieldObj{
Name: fld.Name,
Options: opts,
}
fields = append(fields, field)
}
assert.NoError(t, sapi.CreateIndexAndFields(ctx,
idx.Name,
idx.Options,
fields,
))
}

2
batch/docker-compose.yml
View file

@ -23,7 +23,7 @@ services:
wait:
depends_on:
- "featurebase"
- "featurebase"
build:
context: .
dockerfile: Dockerfile-wait

116
client/api.go
View file

@ -5,6 +5,7 @@ import (
featurebase "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/client/types"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/errors"
)
@ -16,74 +17,67 @@ var _ featurebase.SchemaAPI = &schemaAPI{}
// used for those tests, it may not be functionally complete, and should not be
// used otherwise without further testing and review of this code.
type schemaAPI struct {
*Client
client *Client
}
func NewSchemaAPI(c *Client) *schemaAPI {
return &schemaAPI{
Client: c,
client: c,
}
}
func (s *schemaAPI) CreateIndexAndFields(ctx context.Context, indexName string, options featurebase.IndexOptions, fields []featurebase.CreateFieldObj) error {
schema, err := s.Client.Schema()
func (s *schemaAPI) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) {
return nil, nil
}
func (s *schemaAPI) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) {
return nil, nil
}
func (s *schemaAPI) Tables(ctx context.Context) ([]*dax.Table, error) {
return nil, nil
}
func (s *schemaAPI) CreateTable(ctx context.Context, tbl *dax.Table) error {
schema, err := s.client.Schema()
if err != nil {
return errors.Wrap(err, "getting schema")
}
ii := featurebase.TableToIndexInfo(tbl)
// Add the index.
idx := schema.Index(indexName,
OptIndexKeys(options.Keys),
idx := schema.Index(ii.Name,
OptIndexKeys(ii.Options.Keys),
OptIndexTrackExistence(true),
)
if err := s.Client.CreateIndex(idx); err != nil {
if err := s.client.CreateIndex(idx); err != nil {
return errors.Wrap(err, "creating index")
}
// Now add fields.
for _, f := range fields {
fld, err := s.addFieldToIndex(idx, f.Name, f.Options...)
for _, f := range ii.Fields {
fld, err := s.addFieldToIndex(idx, f.Name, f.Options)
if err != nil {
return errors.Wrapf(err, "adding field to index")
}
if err := s.Client.CreateField(fld); err != nil {
if err := s.client.CreateField(fld); err != nil {
return errors.Wrapf(err, "creating field")
}
}
return nil
}
func (s *schemaAPI) CreateField(ctx context.Context, indexName string, fieldName string, opts ...featurebase.FieldOption) (*featurebase.Field, error) {
schema, err := s.Client.Schema()
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
if !schema.HasIndex(indexName) {
return nil, featurebase.ErrIndexNotFound
}
idx := schema.Index(indexName)
fld, err := s.addFieldToIndex(idx, fieldName, opts...)
if err != nil {
return nil, errors.Wrapf(err, "adding field to index")
}
if err := s.Client.CreateField(fld); err != nil {
return nil, errors.Wrapf(err, "creating field")
}
return nil, nil
func (s *schemaAPI) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error {
return nil
}
func (s *schemaAPI) addFieldToIndex(idx *Index, fieldName string, opts ...featurebase.FieldOption) (*Field, error) {
ffos := &featurebase.FieldOptions{}
for _, opt := range opts {
opt(ffos)
}
func (s *schemaAPI) DeleteTable(ctx context.Context, tname dax.TableName) error {
return s.client.DeleteIndexByName(string(tname))
}
func (s *schemaAPI) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error {
return nil
}
func (s *schemaAPI) addFieldToIndex(idx *Index, fieldName string, ffos featurebase.FieldOptions) (*Field, error) {
cfos := []FieldOption{}
switch ffos.Type {
@ -117,52 +111,6 @@ func (s *schemaAPI) addFieldToIndex(idx *Index, fieldName string, opts ...featur
return idx.Field(fieldName, cfos...), nil
}
func (s *schemaAPI) DeleteField(ctx context.Context, indexName string, fieldName string) error {
schema, err := s.Client.Schema()
if err != nil {
return errors.Wrap(err, "getting schema")
}
if !schema.HasIndex(indexName) {
return featurebase.ErrIndexNotFound
}
idx := schema.Index(indexName)
return s.Client.DeleteField(&Field{
name: fieldName,
index: idx,
})
}
func (s *schemaAPI) DeleteIndex(ctx context.Context, indexName string) error {
return s.Client.DeleteIndexByName(indexName)
}
func (s *schemaAPI) IndexInfo(ctx context.Context, indexName string) (*featurebase.IndexInfo, error) {
schema, err := s.Client.Schema()
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
if !schema.HasIndex(indexName) {
return nil, featurebase.ErrIndexNotFound
}
idx := schema.Index(indexName)
return FromClientIndex(idx), nil
}
// FieldInfo returns the same information as Schema(), but only for a single
// index.
func (s *schemaAPI) FieldInfo(ctx context.Context, indexName, fieldName string) (*featurebase.FieldInfo, error) {
return nil, nil
}
func (s *schemaAPI) Schema(ctx context.Context, withViews bool) ([]*featurebase.IndexInfo, error) {
return nil, errors.New("", "schemaAPI.Schema is not implemented")
}
var _ featurebase.QueryAPI = &queryAPI{}
// queryAPI is a featurebase client wrapper which implements the

2
dax/queryer/importer.go
View file

@ -123,7 +123,7 @@ func (b *batchImporter) indexToQualifiedTableKey(ctx context.Context, index stri
qtid, err := b.schemar.TableID(ctx, b.qual, dax.TableName(index))
if err != nil {
return "", errors.Wrap(err, "converting index to qualified table id")
return "", errors.Wrapf(err, "converting index to qualified table id: %s", index)
}
return qtid.Key(), nil
}

452
dax/queryer/schema_api.go
View file

@ -2,269 +2,13 @@ package queryer
import (
"context"
"fmt"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/dax/mds/schemar"
"github.com/molecula/featurebase/v3/errors"
"github.com/molecula/featurebase/v3/pql"
)
// Ensure type implements interface.
var _ pilosa.SchemaInfoAPI = (*schemaInfoAPI)(nil)
type schemaInfoAPI struct {
schemar schemar.Schemar
}
func NewSchemaInfoAPI(schemar schemar.Schemar) *schemaInfoAPI {
return &schemaInfoAPI{
schemar: schemar,
}
}
func (a *schemaInfoAPI) IndexInfo(ctx context.Context, indexName string) (*pilosa.IndexInfo, error) {
qtid := dax.TableKey(indexName).QualifiedTableID()
tbl, err := a.schemar.Table(ctx, qtid)
if err != nil {
return nil, errors.Wrap(err, "getting table for indexinfo")
}
return daxTableToFeaturebaseIndexInfo(tbl, false)
}
func (a *schemaInfoAPI) FieldInfo(ctx context.Context, indexName, fieldName string) (*pilosa.FieldInfo, error) {
qtid := dax.TableKey(indexName).QualifiedTableID()
tbl, err := a.schemar.Table(ctx, qtid)
fldName := dax.FieldName(fieldName)
if err != nil {
return nil, errors.Wrap(err, "getting table for fieldinfo")
}
fld, ok := tbl.Field(dax.FieldName(fieldName))
if !ok {
return nil, dax.NewErrFieldDoesNotExist(fldName)
}
return daxFieldToFeaturebaseFieldInfo(fld)
}
// daxTableToFeaturebaseIndexInfo converts a dax.Table to a
// featurebase.IndexInfo. If useName is true, the IndexInfo.Name value will
// be set to the qualified table name. Otherwise it will be set to the table key.
func daxTableToFeaturebaseIndexInfo(qtbl *dax.QualifiedTable, useName bool) (*pilosa.IndexInfo, error) {
name := string(qtbl.Key())
if useName {
name = string(qtbl.Name)
}
ii := &pilosa.IndexInfo{
Name: name,
CreatedAt: 0,
Options: pilosa.IndexOptions{
Keys: qtbl.StringKeys(),
TrackExistence: true,
},
ShardWidth: pilosa.ShardWidth,
}
// fields
fields := make([]*pilosa.FieldInfo, len(qtbl.Fields))
var err error
for i := range qtbl.Fields {
fields[i], err = daxFieldToFeaturebaseFieldInfo(qtbl.Fields[i])
if err != nil {
return nil, errors.Wrap(err, "converting field to FieldInfo")
}
}
ii.Fields = fields
return ii, nil
}
// daxFieldToFeaturebaseFieldInfo converts a dax.Field to a
// featurebase.FieldInfo.
func daxFieldToFeaturebaseFieldInfo(field *dax.Field) (*pilosa.FieldInfo, error) {
var timeUnit string
var base int64
min := field.Options.Min
max := field.Options.Max
switch field.Type {
case dax.BaseTypeTimestamp:
timestampOptions, err := daxFieldOptionsToFeaturebaseTimestamp(field.Options)
if err != nil {
return nil, errors.Wrap(err, "getting timestamp options")
}
timeUnit = timestampOptions.TimeUnit
base = timestampOptions.Base
min = timestampOptions.Min
max = timestampOptions.Max
}
fi := &pilosa.FieldInfo{
Name: string(field.Name),
CreatedAt: 0, // TODO(tlt): we need to handle this on MDS schemar
Options: pilosa.FieldOptions{
Type: featurebaseFieldType(field),
Base: base,
Min: min,
Max: max,
Scale: field.Options.Scale,
Keys: field.StringKeys(),
NoStandardView: field.Options.NoStandardView,
CacheType: field.Options.CacheType,
CacheSize: field.Options.CacheSize,
TimeUnit: timeUnit,
TimeQuantum: pilosa.TimeQuantum(field.Options.TimeQuantum),
TTL: field.Options.TTL,
ForeignIndex: field.Options.ForeignIndex,
},
Views: nil, // TODO: do we need views populated?
}
return fi, nil
}
// featurebaseFieldType returns the featurebase.FieldType for the given
// dax.Field.
func featurebaseFieldType(f *dax.Field) string {
switch f.Type {
case dax.BaseTypeID, dax.BaseTypeString:
if f.Name == dax.PrimaryKeyFieldName {
return string(f.Type)
}
return "mutex"
case dax.BaseTypeIDSet, dax.BaseTypeStringSet:
if f.Options.TimeQuantum != "" {
return "time"
}
return "set"
default:
return string(f.Type)
}
}
// featurebaseFieldOptionsToEpoch produces an Epoch (time.Time) value based on
// the given featurebase FieldOptions.
func featurebaseFieldOptionsToEpoch(fo *pilosa.FieldOptions) time.Time {
epochNano := fo.Base * pilosa.TimeUnitNanos(fo.TimeUnit)
return time.Unix(0, epochNano)
}
// daxFieldOptionsToFeaturebaseTimestamp produces a featurebase.FieldOptions
// value with the applicable options populated.
func daxFieldOptionsToFeaturebaseTimestamp(fo dax.FieldOptions) (*pilosa.FieldOptions, error) {
out := &pilosa.FieldOptions{}
// Check if the epoch will overflow when converted to nano.
if err := pilosa.CheckEpochOutOfRange(fo.Epoch, pilosa.MinTimestampNano, pilosa.MaxTimestampNano); err != nil {
return nil, errors.Wrap(err, "checking overflow")
}
out.TimeUnit = fo.TimeUnit
out.Base = fo.Epoch.UnixNano() / pilosa.TimeUnitNanos(fo.TimeUnit)
out.Min = pql.NewDecimal(pilosa.MinTimestamp.UnixNano()/pilosa.TimeUnitNanos(fo.TimeUnit), 0)
out.Max = pql.NewDecimal(pilosa.MaxTimestamp.UnixNano()/pilosa.TimeUnitNanos(fo.TimeUnit), 0)
return out, nil
}
func featurebaseFieldOptionSliceToDaxField(name string, opts []pilosa.FieldOption) (*dax.Field, error) {
fo := &pilosa.FieldOptions{}
for _, opt := range opts {
if err := opt(fo); err != nil {
return nil, errors.Wrap(err, "applying field option")
}
}
return featurebaseFieldOptionsToDaxField(name, fo)
}
func featurebaseFieldOptionsToDaxField(name string, fo *pilosa.FieldOptions) (*dax.Field, error) {
// Initialize field options; to be overridden based on field type
// specific options. Unless determined otherwise, the defaults for these
// values are applied in sql3/planner/createtable.go, so we don't
// initialize with defaults here. In other words, we set these value to
// exactly as we receive them from the caller.
var fieldType dax.BaseType
var min pql.Decimal
var max pql.Decimal
var scale int64
var cacheType string
var cacheSize uint32
var timeUnit string
var epoch time.Time
var foreignIndex string
var timeQuantum dax.TimeQuantum
switch fo.Type {
case pilosa.FieldTypeMutex:
if fo.Keys {
fieldType = dax.BaseTypeString
} else {
fieldType = dax.BaseTypeID
}
cacheType = fo.CacheType
cacheSize = fo.CacheSize
case pilosa.FieldTypeSet:
if fo.Keys {
fieldType = dax.BaseTypeStringSet
} else {
fieldType = dax.BaseTypeIDSet
}
cacheType = fo.CacheType
cacheSize = fo.CacheSize
case pilosa.FieldTypeInt:
min = fo.Min
max = fo.Max
fieldType = dax.BaseTypeInt
foreignIndex = fo.ForeignIndex
case pilosa.FieldTypeDecimal:
min = fo.Min
max = fo.Max
scale = fo.Scale
fieldType = dax.BaseTypeDecimal
case pilosa.FieldTypeTimestamp:
epoch = featurebaseFieldOptionsToEpoch(fo)
timeUnit = fo.TimeUnit
fieldType = dax.BaseTypeTimestamp
case pilosa.FieldTypeBool:
fieldType = dax.BaseTypeBool
case pilosa.FieldTypeTime:
if fo.Keys {
fieldType = dax.BaseTypeStringSet
} else {
fieldType = dax.BaseTypeIDSet
}
timeQuantum = dax.TimeQuantum(fo.TimeQuantum)
default:
return nil, errors.New(errors.ErrUncoded, fmt.Sprintf("unhandled featurebase field type: %s", fo.Type))
}
daxField := &dax.Field{
Name: dax.FieldName(name),
Type: fieldType,
Options: dax.FieldOptions{
Min: min,
Max: max,
Scale: scale,
NoStandardView: fo.NoStandardView,
CacheType: cacheType,
CacheSize: cacheSize,
TimeUnit: timeUnit,
Epoch: epoch,
TimeQuantum: timeQuantum,
TTL: fo.TTL,
ForeignIndex: foreignIndex,
},
}
return daxField, nil
}
// Ensure type implements interface.
var _ pilosa.SchemaAPI = (*qualifiedSchemaAPI)(nil)
@ -284,179 +28,73 @@ func NewQualifiedSchemaAPI(qual dax.TableQualifier, schemar schemar.Schemar) *qu
}
}
func (s *qualifiedSchemaAPI) CreateIndexAndFields(ctx context.Context, indexName string, options pilosa.IndexOptions, fields []pilosa.CreateFieldObj) error {
// Make sure the table for this qualifier doesn't already exist.
//_, err := s.schemar.TableID(ctx, s.qual, dax.TableName(indexName))
_, err := s.schemar.TableID(ctx, s.qual, dax.TableName(indexName))
// TODO(tlt): the following doesn't work when the TableID() call is made
// over http because the error that comes back is `status code: 400: table
// name 'tbl' does not exist\n`, which does not match the check. We really
// need to be able to check these error codes both directly on the error AND
// when they come back via http.
// if !errors.Is(err, dax.ErrTableNameDoesNotExist) {
// if err != nil {
// return errors.Wrapf(err, "checking if table name already exists: %s, %s", s.qual, indexName)
// }
// return dax.NewErrTableNameExists(dax.TableName(indexName))
// }
if err == nil {
return dax.NewErrTableNameExists(dax.TableName(indexName))
func (s *qualifiedSchemaAPI) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) {
qtid, err := s.schemar.TableID(ctx, s.qual, tname)
if err != nil {
return nil, errors.Wrapf(err, "getting table id: (%s) %s", s.qual, tname)
}
partitionN := dax.DefaultPartitionN
// TODO(tlt): until we can thread partitionN through featurebase correctly
// (instead of having it use holder.partitionN or DefaultPartitionN), then
// we can't use a custom partitionN.
// if options.PartitionN > 0 {
// partitionN = options.PartitionN
// }
// Initialize the fields slice with one additional slot for the primary key
// field.
daxFields := make([]*dax.Field, 0, len(fields)+1)
// Add the primary key field.
var fieldType dax.BaseType
if options.Keys {
fieldType = dax.BaseTypeString
} else {
fieldType = dax.BaseTypeID
}
daxFields = append(daxFields, &dax.Field{
Name: dax.PrimaryKeyFieldName,
Type: fieldType,
})
// Add the fields provided in the method call.
for _, fldObj := range fields {
daxField, err := featurebaseFieldOptionSliceToDaxField(fldObj.Name, fldObj.Options)
if err != nil {
return errors.Wrap(err, "converting featurebase field options to dax field")
}
daxFields = append(daxFields, daxField)
qtbl, err := s.schemar.Table(ctx, qtid)
if err != nil {
return nil, errors.Wrapf(err, "getting table: %s", qtid)
}
tbl := &dax.Table{
Name: dax.TableName(indexName),
Fields: daxFields,
PartitionN: partitionN,
return &qtbl.Table, nil
}
func (s *qualifiedSchemaAPI) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) {
qtid := dax.NewQualifiedTableID(s.qual, tid)
qtbl, err := s.schemar.Table(ctx, qtid)
if err != nil {
return nil, errors.Wrapf(err, "getting table: %s", qtid)
}
qtbl := dax.NewQualifiedTable(
s.qual,
tbl,
)
return &qtbl.Table, nil
}
func (s *qualifiedSchemaAPI) Tables(ctx context.Context) ([]*dax.Table, error) {
qtbls, err := s.schemar.Tables(ctx, s.qual)
if err != nil {
return nil, errors.Wrap(err, "getting tables")
}
tbls := make([]*dax.Table, 0, len(qtbls))
for _, qtbl := range qtbls {
tbls = append(tbls, &qtbl.Table)
}
return tbls, nil
}
func (s *qualifiedSchemaAPI) CreateTable(ctx context.Context, tbl *dax.Table) error {
qtbl := dax.NewQualifiedTable(s.qual, tbl)
return s.schemar.CreateTable(ctx, qtbl)
}
func (s *qualifiedSchemaAPI) CreateField(ctx context.Context, indexName string, fieldName string, opts ...pilosa.FieldOption) (*pilosa.Field, error) {
tkey, err := s.indexToQualifiedTableKey(ctx, indexName)
func (s *qualifiedSchemaAPI) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error {
qtid, err := s.schemar.TableID(ctx, s.qual, tname)
if err != nil {
return nil, errors.Wrap(err, "converting index to qualified table key")
return errors.Wrapf(err, "getting table id: (%s) %s", s.qual, tname)
}
daxField, err := featurebaseFieldOptionSliceToDaxField(fieldName, opts)
if err != nil {
return nil, errors.Wrap(err, "converting featurebase field options to dax field")
}
qtid := tkey.QualifiedTableID()
if err := s.schemar.CreateField(ctx, qtid, daxField); err != nil {
return nil, errors.New(errors.ErrUncoded, err.Error())
}
return nil, nil
return s.schemar.CreateField(ctx, qtid, fld)
}
func (s *qualifiedSchemaAPI) DeleteField(ctx context.Context, indexName string, fieldName string) error {
tkey, err := s.indexToQualifiedTableKey(ctx, indexName)
func (s *qualifiedSchemaAPI) DeleteTable(ctx context.Context, tname dax.TableName) error {
qtid, err := s.schemar.TableID(ctx, s.qual, tname)
if err != nil {
return errors.Wrap(err, "converting index to qualified table key")
return errors.Wrapf(err, "getting table id: (%s) %s", s.qual, tname)
}
qtid := tkey.QualifiedTableID()
fldName := dax.FieldName(fieldName)
if err := s.schemar.DropField(ctx, qtid, fldName); err != nil {
return errors.New(errors.ErrUncoded, err.Error())
}
return nil
}
func (s *qualifiedSchemaAPI) DeleteIndex(ctx context.Context, indexName string) error {
tkey, err := s.indexToQualifiedTableKey(ctx, indexName)
if err != nil {
return errors.Wrap(err, "converting index to qualified table key")
}
qtid := tkey.QualifiedTableID()
return s.schemar.DropTable(ctx, qtid)
}
func (s *qualifiedSchemaAPI) IndexInfo(ctx context.Context, indexName string) (*pilosa.IndexInfo, error) {
tkey, err := s.indexToQualifiedTableKey(ctx, indexName)
func (s *qualifiedSchemaAPI) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error {
qtid, err := s.schemar.TableID(ctx, s.qual, tname)
if err != nil {
return nil, errors.Wrap(err, "converting index to qualified table key")
return errors.Wrapf(err, "getting table id: (%s) %s", s.qual, tname)
}
qtid := tkey.QualifiedTableID()
tbl, err := s.schemar.Table(ctx, qtid)
if err != nil {
return nil, errors.Wrap(err, "getting table for qualified indexinfo")
}
return daxTableToFeaturebaseIndexInfo(tbl, true)
}
func (s *qualifiedSchemaAPI) FieldInfo(ctx context.Context, indexName, fieldName string) (*pilosa.FieldInfo, error) {
tkey, err := s.indexToQualifiedTableKey(ctx, indexName)
if err != nil {
return nil, errors.Wrap(err, "converting index to qualified table key")
}
qtid := tkey.QualifiedTableID()
tbl, err := s.schemar.Table(ctx, qtid)
fldName := dax.FieldName(fieldName)
if err != nil {
return nil, errors.Wrap(err, "getting table for qualified fieldinfo")
}
fld, ok := tbl.Field(dax.FieldName(fieldName))
if !ok {
return nil, dax.NewErrFieldDoesNotExist(fldName)
}
return daxFieldToFeaturebaseFieldInfo(fld)
}
func (s *qualifiedSchemaAPI) Schema(ctx context.Context, withViews bool) ([]*pilosa.IndexInfo, error) {
tbls, err := s.schemar.Tables(ctx, s.qual)
if err != nil {
return nil, errors.Wrap(err, "getting tables for qualified schema")
}
indexes := make([]*pilosa.IndexInfo, len(tbls))
for i := range tbls {
// This method appears to be used primarily in "SHOW TABLES", and in
// that case we want to return the human friendly table name used when
// creating the table (i.e. not the table key).
indexes[i], err = daxTableToFeaturebaseIndexInfo(tbls[i], true)
if err != nil {
return nil, errors.Wrap(err, "converting table to IndexInfo")
}
}
return indexes, nil
}
func (s *qualifiedSchemaAPI) indexToQualifiedTableKey(ctx context.Context, index string) (dax.TableKey, error) {
qtid, err := s.schemar.TableID(ctx, s.qual, dax.TableName(index))
if err != nil {
return "", errors.Wrap(err, "converting index to qualified table id")
}
return qtid.Key(), nil
return s.schemar.DropField(ctx, qtid, fname)
}

79
dax/queryer/schema_info_api.go Normal file
View file

@ -0,0 +1,79 @@
package queryer
import (
"context"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/dax/mds/schemar"
"github.com/molecula/featurebase/v3/errors"
)
// Ensure type implements interface.
var _ pilosa.SchemaInfoAPI = (*schemaInfoAPI)(nil)
type schemaInfoAPI struct {
schemar schemar.Schemar
}
func NewSchemaInfoAPI(schemar schemar.Schemar) *schemaInfoAPI {
return &schemaInfoAPI{
schemar: schemar,
}
}
func (a *schemaInfoAPI) IndexInfo(ctx context.Context, indexName string) (*pilosa.IndexInfo, error) {
qtid := dax.TableKey(indexName).QualifiedTableID()
tbl, err := a.schemar.Table(ctx, qtid)
if err != nil {
return nil, errors.Wrap(err, "getting table for indexinfo")
}
return daxTableToFeaturebaseIndexInfo(tbl, false)
}
func (a *schemaInfoAPI) FieldInfo(ctx context.Context, indexName, fieldName string) (*pilosa.FieldInfo, error) {
qtid := dax.TableKey(indexName).QualifiedTableID()
tbl, err := a.schemar.Table(ctx, qtid)
fldName := dax.FieldName(fieldName)
if err != nil {
return nil, errors.Wrap(err, "getting table for fieldinfo")
}
fld, ok := tbl.Field(dax.FieldName(fieldName))
if !ok {
return nil, dax.NewErrFieldDoesNotExist(fldName)
}
return pilosa.FieldToFieldInfo(fld), nil
}
// TODO(tlt): try to get rid of this in favor of pilosa.TableToIndexInfo.
// daxTableToFeaturebaseIndexInfo converts a dax.Table to a
// featurebase.IndexInfo. If useName is true, the IndexInfo.Name value will
// be set to the qualified table name. Otherwise it will be set to the table key.
func daxTableToFeaturebaseIndexInfo(qtbl *dax.QualifiedTable, useName bool) (*pilosa.IndexInfo, error) {
name := string(qtbl.Key())
if useName {
name = string(qtbl.Name)
}
ii := &pilosa.IndexInfo{
Name: name,
CreatedAt: 0,
Options: pilosa.IndexOptions{
Keys: qtbl.StringKeys(),
TrackExistence: true,
},
ShardWidth: pilosa.ShardWidth,
}
// fields
fields := make([]*pilosa.FieldInfo, len(qtbl.Fields))
for i := range qtbl.Fields {
fields[i] = pilosa.FieldToFieldInfo(qtbl.Fields[i])
}
ii.Fields = fields
return ii, nil
}

2
idk/docker-compose.yml
View file

@ -125,8 +125,6 @@ services:
- ./docker-sasl/ssl_keys:/ssl_keys
- ./testdata:/testdata
depends_on:
#- kafka
#- postgres
- fakeidp
wait:
build:

5
idk/ingest_test.go
View file

@ -1774,6 +1774,8 @@ func TestBatchTargetMDS(t *testing.T) {
fieldType: dax.BaseTypeDecimal,
fieldOptions: dax.FieldOptions{
Scale: 4,
Min: pql.NewDecimal(-100, 0),
Max: pql.NewDecimal(100, 0),
},
fieldFn: decimalFn,
in: [][]interface{}{
@ -1868,6 +1870,8 @@ func TestBatchTargetMDS(t *testing.T) {
t.Fatalf("creating table: %v", err)
}
// qtblWithID is the same as qtbl above, but now MDS has
// assigned the table a unique ID.
qtblWithID, err := mdsClient.Table(ctx, qtbl.QualifiedID())
assert.NoError(t, err)
@ -1881,7 +1885,6 @@ func TestBatchTargetMDS(t *testing.T) {
ingester.NewSource = func() (Source, error) { return ts, nil }
ingester.BatchSize = 10
//ingester.PrimaryKeyFields = []string{"rcid"}
ingester.IDField = "id"
if err := ingester.Run(); err != nil {

460
schema.go Normal file
View file

@ -0,0 +1,460 @@
package pilosa
import (
"context"
"fmt"
"log"
"sort"
"time"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/errors"
"github.com/molecula/featurebase/v3/pql"
)
// Ensure type implements interface.
var _ SchemaAPI = (*onPremSchema)(nil)
type onPremSchema struct {
api *API
}
func NewOnPremSchema(api *API) *onPremSchema {
return &onPremSchema{
api: api,
}
}
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 {
return nil, errors.Wrapf(err, "getting index info for table name: %s", tname)
}
return IndexInfoToTable(idx), nil
}
func (s *onPremSchema) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) {
idx, err := s.api.IndexInfo(context.Background(), string(tid))
if err != nil {
return nil, errors.Wrapf(err, "getting index info for table id: %s", tid)
}
return IndexInfoToTable(idx), nil
}
func (s *onPremSchema) Tables(ctx context.Context) ([]*dax.Table, error) {
idxs, err := s.api.Schema(ctx, false)
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
return IndexInfosToTables(idxs), nil
}
func (s *onPremSchema) CreateTable(ctx context.Context, tbl *dax.Table) error {
// We make a slice of fields with the _id field removed. Also, while we're
// at it, we can use the type of the _id field to determine if the index
// should be keyed.
var keyed bool
flds := make([]*dax.Field, 0)
for _, fld := range tbl.Fields {
if fld.Name == "_id" {
if fld.Type == dax.BaseTypeString {
keyed = true
}
continue
}
flds = append(flds, fld)
}
iopts := IndexOptions{
Keys: keyed,
TrackExistence: true,
PartitionN: tbl.PartitionN,
}
// Add the index.
if _, err := s.api.CreateIndex(ctx, string(tbl.Name), iopts); err != nil {
return err
}
// Now add fields.
for _, fld := range flds {
if err := s.CreateField(ctx, tbl.Name, fld); err != nil {
return errors.Wrapf(err, "creating field: %s", fld.Name)
}
}
return nil
}
func (s *onPremSchema) CreateField(ctx context.Context, tname dax.TableName, fld *dax.Field) error {
opts, err := FieldOptionsFromField(fld)
if err != nil {
return errors.Wrapf(err, "creating field options from field: %s", fld.Name)
}
_, err = s.api.CreateField(ctx, string(tname), string(fld.Name), opts...)
return err
}
func (s *onPremSchema) DeleteTable(ctx context.Context, tname dax.TableName) error {
return s.api.DeleteIndex(ctx, string(tname))
}
func (s *onPremSchema) DeleteField(ctx context.Context, tname dax.TableName, fname dax.FieldName) error {
return s.api.DeleteField(ctx, string(tname), string(fname))
}
//////////////////////////////////////////////////////////////////////////////
// The following are helper functions which convert between
// featurebase.IndexInfo and dax.Table, and between featurebase.FieldInfo and
// dax.Field.
//////////////////////////////////////////////////////////////////////////////
//
// Functions to convert from featurebase to dax.
//
// IndexInfosToTables converts a slice of featurebase.IndexInfo to a slice of
// dax.Table.
func IndexInfosToTables(iis []*IndexInfo) []*dax.Table {
tbls := make([]*dax.Table, 0, len(iis))
for _, ii := range iis {
tbls = append(tbls, IndexInfoToTable(ii))
}
return tbls
}
// IndexInfoToTable converts a featurebase.IndexInfo to a dax.Table.
func IndexInfoToTable(ii *IndexInfo) *dax.Table {
tbl := &dax.Table{
// TODO(tlt): be careful here. This ID=Name logic only applies to "onPrem".
ID: dax.TableID(ii.Name),
Name: dax.TableName(ii.Name),
Fields: make([]*dax.Field, 0, len(ii.Fields)+1), // +1 to account for the _id field
PartitionN: dax.DefaultPartitionN,
}
// // sortedFields will contain the sorted list of fields from IndexInfo.
// sortedFields := make([]*FieldInfo, 0, len(ii.Fields))
// Sort ii.Fields by CreatedAt before adding them to sortedFields.
sort.Slice(ii.Fields, func(i, j int) bool {
return ii.Fields[i].CreatedAt < ii.Fields[j].CreatedAt
})
// // Add the sorted fields to sortedFields.
// sortedFields = append(sortedFields, ii.Fields...)
// Add the _id Field.
var idType dax.BaseType = dax.BaseTypeID
if ii.Options.Keys {
idType = dax.BaseTypeString
}
tbl.Fields = append(tbl.Fields, &dax.Field{
Name: "_id",
Type: idType,
})
// Populate the rest of the fields.
for _, fld := range ii.Fields {
tbl.Fields = append(tbl.Fields, FieldInfoToField(fld))
}
return tbl
}
// FieldInfoToField converts a featurebase.FieldInfo to a dax.Field.
func FieldInfoToField(fi *FieldInfo) *dax.Field {
// Initialize field options; to be overridden based on field type specific
// options.
var fieldType dax.BaseType
var min pql.Decimal
var max pql.Decimal
var scale int64
var cacheType string
var cacheSize uint32
var timeUnit string
var epoch time.Time
var foreignIndex string
var timeQuantum dax.TimeQuantum
fo := &fi.Options
switch fo.Type {
case FieldTypeMutex:
if fo.Keys {
fieldType = dax.BaseTypeString
} else {
fieldType = dax.BaseTypeID
}
cacheType = fo.CacheType
cacheSize = fo.CacheSize
case FieldTypeSet:
if fo.Keys {
fieldType = dax.BaseTypeStringSet
} else {
fieldType = dax.BaseTypeIDSet
}
cacheType = fo.CacheType
cacheSize = fo.CacheSize
case FieldTypeInt:
min = fo.Min
max = fo.Max
fieldType = dax.BaseTypeInt
foreignIndex = fo.ForeignIndex
case FieldTypeDecimal:
min = fo.Min
max = fo.Max
scale = fo.Scale
fieldType = dax.BaseTypeDecimal
case FieldTypeTimestamp:
epoch = featurebaseFieldOptionsToEpoch(fo)
timeUnit = fo.TimeUnit
fieldType = dax.BaseTypeTimestamp
case FieldTypeBool:
fieldType = dax.BaseTypeBool
case FieldTypeTime:
if fo.Keys {
fieldType = dax.BaseTypeStringSet
} else {
fieldType = dax.BaseTypeIDSet
}
timeQuantum = dax.TimeQuantum(fo.TimeQuantum)
default:
panic(fmt.Sprintf("unhandled featurebase field type: %s", fo.Type))
}
return &dax.Field{
Name: dax.FieldName(fi.Name),
Type: fieldType,
Options: dax.FieldOptions{
Min: min,
Max: max,
Scale: scale,
NoStandardView: fo.NoStandardView,
CacheType: cacheType,
CacheSize: cacheSize,
TimeUnit: timeUnit,
Epoch: epoch,
TimeQuantum: timeQuantum,
TTL: fo.TTL,
ForeignIndex: foreignIndex,
},
}
}
// featurebaseFieldOptionsToEpoch produces an Epoch (time.Time) value based on
// the given featurebase FieldOptions.
func featurebaseFieldOptionsToEpoch(fo *FieldOptions) time.Time {
epochNano := fo.Base * TimeUnitNanos(fo.TimeUnit)
return time.Unix(0, epochNano)
}
//
// Functions to convert from dax to featurebase.
//
// TablesToIndexInfos converts a slice of dax.Table to a slice of
// featurease.IndexInfo.
func TablesToIndexInfos(tbls []*dax.Table) []*IndexInfo {
iis := make([]*IndexInfo, 0, len(tbls))
for _, tbl := range tbls {
iis = append(iis, TableToIndexInfo(tbl))
}
return iis
}
// TableToIndexInfo converts a dax.Table to a featurease.IndexInfo.
func TableToIndexInfo(tbl *dax.Table) *IndexInfo {
ii := &IndexInfo{
Name: string(tbl.Name), // TODO(tlt): this should be TableKey i think
CreatedAt: 0,
Options: IndexOptions{
Keys: tbl.StringKeys(),
TrackExistence: true,
},
ShardWidth: ShardWidth,
}
// fields
fields := make([]*FieldInfo, 0, len(tbl.Fields)-1)
for i := range tbl.Fields {
if tbl.Fields[i].Name == "_id" {
continue
}
fields = append(fields, FieldToFieldInfo(tbl.Fields[i]))
}
ii.Fields = fields
return ii
}
// FieldToFieldInfo converts a dax.Field to a featurebase.FieldInfo. Note: it
// does not return errors; there is one scenario where a timestamp epoch could
// be out of range. In that case, this function will only log the error, and the
// proceed with timestamp option values which are likely incorrect. We are going
// to leave this as is for now because, since this is used for internal
// conversions of types which already exist and have been validated, we assume
// the option values are valid.
// TODO(tlt): add error handling to this function; worst case: panic.
func FieldToFieldInfo(fld *dax.Field) *FieldInfo {
var timeUnit string
var base int64
min := fld.Options.Min
max := fld.Options.Max
switch fld.Type {
case dax.BaseTypeTimestamp:
timestampOptions, err := fieldOptionsForTimestamp(fld.Options)
if err != nil {
log.Printf("ERROR: converting timestamp options: %v", err)
}
timeUnit = timestampOptions.TimeUnit
base = timestampOptions.Base
min = timestampOptions.Min
max = timestampOptions.Max
}
return &FieldInfo{
Name: string(fld.Name),
CreatedAt: 0, // TODO(tlt): we need to handle this on MDS schemar
Options: FieldOptions{
Type: fieldToFieldType(fld),
Base: base,
Min: min,
Max: max,
Scale: fld.Options.Scale,
Keys: fld.StringKeys(),
NoStandardView: fld.Options.NoStandardView,
CacheType: fld.Options.CacheType,
CacheSize: fld.Options.CacheSize,
TimeUnit: timeUnit,
TimeQuantum: TimeQuantum(fld.Options.TimeQuantum),
TTL: fld.Options.TTL,
ForeignIndex: fld.Options.ForeignIndex,
},
Views: nil, // TODO(tlt): do we need views populated?
}
}
// fieldOptionsForTimestamp produces a featurebase.FieldOptions value with the
// timestamp-related options populated.
func fieldOptionsForTimestamp(fo dax.FieldOptions) (*FieldOptions, error) {
out := &FieldOptions{}
// Check if the epoch will overflow when converted to nano.
if err := CheckEpochOutOfRange(fo.Epoch, MinTimestampNano, MaxTimestampNano); err != nil {
return out, errors.Wrap(err, "checking overflow")
}
out.TimeUnit = fo.TimeUnit
out.Base = fo.Epoch.UnixNano() / TimeUnitNanos(fo.TimeUnit)
out.Min = pql.NewDecimal(MinTimestamp.UnixNano()/TimeUnitNanos(fo.TimeUnit), 0)
out.Max = pql.NewDecimal(MaxTimestamp.UnixNano()/TimeUnitNanos(fo.TimeUnit), 0)
return out, nil
}
// fieldToFieldType returns the featurebase.FieldType for the given dax.Field.
func fieldToFieldType(f *dax.Field) string {
switch f.Type {
case dax.BaseTypeID, dax.BaseTypeString:
if f.Name == dax.PrimaryKeyFieldName {
return string(f.Type)
}
return "mutex"
case dax.BaseTypeIDSet, dax.BaseTypeStringSet:
if f.Options.TimeQuantum != "" {
return "time"
}
return "set"
default:
return string(f.Type)
}
}
func FieldFromFieldOptions(fname dax.FieldName, opts ...FieldOption) (*dax.Field, error) {
fo, err := newFieldOptions(opts...)
if err != nil {
return nil, errors.Wrap(err, "creating new field options")
}
fi := &FieldInfo{
Name: string(fname),
Options: *fo,
}
return FieldInfoToField(fi), nil
}
// FieldOptionsFromField returns a slice of featurebase.FieldOption based on the
// given dax.Field.
func FieldOptionsFromField(fld *dax.Field) ([]FieldOption, error) {
// Set the cache type and size (or use default) for those fields which
// require them.
cacheType := DefaultCacheType
cacheSize := uint32(DefaultCacheSize)
if fld.Options.CacheType != "" {
cacheType = fld.Options.CacheType
cacheSize = fld.Options.CacheSize
}
opts := []FieldOption{}
switch fld.Type {
case dax.BaseTypeBool:
opts = append(opts,
OptFieldTypeBool(),
)
case dax.BaseTypeDecimal:
opts = append(opts,
OptFieldTypeDecimal(fld.Options.Scale, fld.Options.Min, fld.Options.Max),
)
case dax.BaseTypeID:
opts = append(opts,
OptFieldTypeMutex(cacheType, cacheSize),
)
case dax.BaseTypeIDSet:
if fld.Options.TimeQuantum != "" {
opts = append(opts,
OptFieldTypeTime(TimeQuantum(fld.Options.TimeQuantum), fld.Options.TTL.String()),
)
} else {
opts = append(opts,
OptFieldTypeSet(cacheType, cacheSize),
)
}
case dax.BaseTypeInt:
opts = append(opts,
OptFieldTypeInt(fld.Options.Min.ToInt64(0), fld.Options.Max.ToInt64(0)),
)
case dax.BaseTypeString:
opts = append(opts,
OptFieldTypeMutex(cacheType, cacheSize),
OptFieldKeys(),
)
case dax.BaseTypeStringSet:
if fld.Options.TimeQuantum != "" {
opts = append(opts,
OptFieldTypeTime(TimeQuantum(fld.Options.TimeQuantum), fld.Options.TTL.String()),
OptFieldKeys(),
)
} else {
opts = append(opts,
OptFieldTypeSet(cacheType, cacheSize),
OptFieldKeys(),
)
}
case dax.BaseTypeTimestamp:
opts = append(opts,
OptFieldTypeTimestamp(fld.Options.Epoch, fld.Options.TimeUnit),
)
default:
return nil, errors.Errorf("unsupport field type: %s", fld.Type)
}
return opts, nil
}

2
server/server.go
View file

@ -567,7 +567,7 @@ func (m *Command) setupServer() error {
}
executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner {
fapi := &pilosa.FeatureBaseSchemaAPI{API: api}
fapi := pilosa.NewOnPremSchema(api)
fsapi := &pilosa.FeatureBaseSystemAPI{API: api}
fimp := &batch.FeaturebaseImporter{API: api}

12
sql3/planner/compilealtertable.go
View file

@ -7,6 +7,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -27,7 +28,8 @@ func (p *ExecutionPlanner) compileAlterTableStatement(stmt *parser.AlterTableSta
tableName := parser.IdentName(stmt.Name)
// does the table exist
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.Name.NamePos.Line, stmt.Name.NamePos.Column, tableName)
@ -40,8 +42,8 @@ func (p *ExecutionPlanner) compileAlterTableStatement(stmt *parser.AlterTableSta
// does this column exist
found := false
for _, f := range table.Fields {
if strings.EqualFold(f.Name, columnName) {
for _, f := range tbl.Fields {
if strings.EqualFold(string(f.Name), columnName) {
found = true
break
}
@ -56,8 +58,8 @@ func (p *ExecutionPlanner) compileAlterTableStatement(stmt *parser.AlterTableSta
columnName := parser.IdentName(col.Name)
// does this column exist
for _, f := range table.Fields {
if strings.EqualFold(f.Name, columnName) {
for _, f := range tbl.Fields {
if strings.EqualFold(string(f.Name), columnName) {
return nil, sql3.NewErrDuplicateColumn(col.Name.NamePos.Line, col.Name.NamePos.Column, columnName)
}
}

23
sql3/planner/compilebulkinsert.go
View file

@ -10,6 +10,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -21,7 +22,8 @@ import (
func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertStatement) (_ types.PlanOperator, err error) {
tableName := parser.IdentName(stmt.Table)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -103,9 +105,9 @@ func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertSta
// build the target columns
options.targetColumns = make([]*qualifiedRefPlanExpression, 0)
for _, m := range stmt.Columns {
for idx, fld := range table.Fields {
if strings.EqualFold(fld.Name, m.Name) {
options.targetColumns = append(options.targetColumns, newQualifiedRefPlanExpression(tableName, m.Name, idx, fieldSQLDataType(fld)))
for idx, fld := range tbl.Fields {
if strings.EqualFold(string(fld.Name), m.Name) {
options.targetColumns = append(options.targetColumns, newQualifiedRefPlanExpression(tableName, m.Name, idx, fieldSQLDataType(pilosa.FieldToFieldInfo(fld))))
break
}
}
@ -151,7 +153,8 @@ func (p *ExecutionPlanner) compileBulkInsertStatement(stmt *parser.BulkInsertSta
func (p *ExecutionPlanner) analyzeBulkInsertStatement(stmt *parser.BulkInsertStatement) error {
//check referred to table exists
tableName := parser.IdentName(stmt.Table)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -289,10 +292,10 @@ func (p *ExecutionPlanner) analyzeBulkInsertStatement(stmt *parser.BulkInsertSta
// the column list of the table referenced
stmt.Columns = []*parser.Ident{}
for _, fld := range table.Fields {
for _, fld := range tbl.Fields {
stmt.Columns = append(stmt.Columns, &parser.Ident{
NamePos: parser.Pos{Line: 0, Column: 0},
Name: fld.Name,
Name: string(fld.Name),
})
}
}
@ -328,10 +331,10 @@ func (p *ExecutionPlanner) analyzeBulkInsertStatement(stmt *parser.BulkInsertSta
foundID := false
for idx, cm := range stmt.Columns {
found := false
for _, fld := range table.Fields {
if strings.EqualFold(cm.Name, fld.Name) {
for _, fld := range tbl.Fields {
if strings.EqualFold(cm.Name, string(fld.Name)) {
found = true
colDataType := fieldSQLDataType(fld)
colDataType := fieldSQLDataType(pilosa.FieldToFieldInfo(fld))
// if we have transforms check that type and target colum ref are assignment compatible
// else check that the map expressions type and target column ref are assignment compatible

6
sql3/planner/compiledroptable.go
View file

@ -6,6 +6,7 @@ import (
"context"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -16,12 +17,13 @@ import (
// PlanOperator.
func (p *ExecutionPlanner) compileDropTableStatement(stmt *parser.DropTableStatement) (_ types.PlanOperator, err error) {
tableName := parser.IdentName(stmt.Name)
index, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.Name.NamePos.Line, stmt.Name.NamePos.Column, tableName)
}
return nil, err
}
return NewPlanOpQuery(p, NewPlanOpDropTable(p, index), p.sql), nil
return NewPlanOpQuery(p, NewPlanOpDropTable(p, pilosa.TableToIndexInfo(tbl)), p.sql), nil
}

33
sql3/planner/compileinsert.go
View file

@ -7,6 +7,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -20,7 +21,8 @@ func (p *ExecutionPlanner) compileInsertStatement(stmt *parser.InsertStatement)
targetColumns := []*qualifiedRefPlanExpression{}
insertValues := [][]types.PlanExpression{}
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -37,19 +39,19 @@ func (p *ExecutionPlanner) compileInsertStatement(stmt *parser.InsertStatement)
continue
}
for idx, field := range table.Fields {
if strings.EqualFold(colName, field.Name) {
targetColumns = append(targetColumns, newQualifiedRefPlanExpression(tableName, colName, idx, fieldSQLDataType(field)))
for idx, field := range tbl.Fields {
if strings.EqualFold(colName, string(field.Name)) {
targetColumns = append(targetColumns, newQualifiedRefPlanExpression(tableName, colName, idx, fieldSQLDataType(pilosa.FieldToFieldInfo(field))))
break
}
}
}
} else {
for idx, field := range table.Fields {
if strings.EqualFold("_exists", field.Name) {
for idx, field := range tbl.Fields {
if strings.EqualFold("_exists", string(field.Name)) {
continue
}
targetColumns = append(targetColumns, newQualifiedRefPlanExpression(tableName, field.Name, idx, fieldSQLDataType(field)))
targetColumns = append(targetColumns, newQualifiedRefPlanExpression(tableName, string(field.Name), idx, fieldSQLDataType(pilosa.FieldToFieldInfo(field))))
}
}
@ -74,7 +76,8 @@ func (p *ExecutionPlanner) compileInsertStatement(stmt *parser.InsertStatement)
func (p *ExecutionPlanner) analyzeInsertStatement(stmt *parser.InsertStatement) error {
// Check that referred table exists.
tableName := parser.IdentName(stmt.Table)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return sql3.NewErrTableNotFound(stmt.Table.NamePos.Line, stmt.Table.NamePos.Column, tableName)
@ -88,11 +91,11 @@ func (p *ExecutionPlanner) analyzeInsertStatement(stmt *parser.InsertStatement)
// fields in the table.
if len(stmt.Columns) == 0 {
// Generate the list of types from the FeatureBase index.
for _, field := range table.Fields {
if strings.EqualFold("_exists", field.Name) {
for _, field := range tbl.Fields {
if strings.EqualFold("_exists", string(field.Name)) {
continue
}
typeNames = append(typeNames, fieldSQLDataType(field))
typeNames = append(typeNames, fieldSQLDataType(pilosa.FieldToFieldInfo(field)))
}
// Make sure (implicit) insert list and expression list have the same
// number of items.
@ -115,7 +118,7 @@ func (p *ExecutionPlanner) analyzeInsertStatement(stmt *parser.InsertStatement)
// Determine, from the existing table, whether the _id is of
// type ID or STRING.
var idType parser.ExprDataType
if table.Options.Keys {
if tbl.StringKeys() {
idType = parser.NewDataTypeString()
} else {
idType = parser.NewDataTypeID()
@ -127,9 +130,9 @@ func (p *ExecutionPlanner) analyzeInsertStatement(stmt *parser.InsertStatement)
// Find the column in the existing table.
columnFound := false
for _, field := range table.Fields {
if strings.EqualFold(colName, field.Name) {
typeName = fieldSQLDataType(field)
for _, field := range tbl.Fields {
if strings.EqualFold(colName, string(field.Name)) {
typeName = fieldSQLDataType(pilosa.FieldToFieldInfo(field))
columnFound = true
break
}

12
sql3/planner/compileselect.go
View file

@ -7,6 +7,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -313,7 +314,8 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source, scope parser.Stat
case *parser.QualifiedTableName:
// check table exists
tableName := parser.IdentName(source.Name)
table, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return sql3.NewErrTableNotFound(source.Name.NamePos.Line, source.Name.NamePos.Column, tableName)
@ -322,12 +324,12 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source, scope parser.Stat
}
// populate the output columns from the source
for idx, fld := range table.Fields {
for i, fld := range tbl.Fields {
soc := &parser.SourceOutputColumn{
TableName: tableName,
ColumnName: fld.Name,
ColumnIndex: idx,
Datatype: fieldSQLDataType(fld),
ColumnName: string(fld.Name),
ColumnIndex: i,
Datatype: fieldSQLDataType(pilosa.FieldToFieldInfo(fld)),
}
source.OutputColumns = append(source.OutputColumns, soc)
}

16
sql3/planner/compileshow.go
View file

@ -7,6 +7,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -14,9 +15,9 @@ import (
)
func (p *ExecutionPlanner) compileShowTablesStatement(stmt parser.Statement) (types.PlanOperator, error) {
indexInfo, err := p.schemaAPI.Schema(context.Background(), false)
tbls, err := p.schemaAPI.Tables(context.Background())
if err != nil {
return nil, errors.Wrap(err, "getting schema")
return nil, errors.Wrap(err, "getting tables")
}
columns := []types.PlanExpression{
@ -75,12 +76,13 @@ func (p *ExecutionPlanner) compileShowTablesStatement(stmt parser.Statement) (ty
dataType: parser.NewDataTypeString(),
}}
return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseTables(indexInfo)), p.sql), nil
return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseTables(pilosa.TablesToIndexInfos(tbls))), p.sql), nil
}
func (p *ExecutionPlanner) compileShowColumnsStatement(stmt *parser.ShowColumnsStatement) (_ types.PlanOperator, err error) {
tableName := parser.IdentName(stmt.TableName)
index, err := p.schemaAPI.IndexInfo(context.Background(), tableName)
tname := dax.TableName(tableName)
tbl, err := p.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.TableName.NamePos.Line, stmt.TableName.NamePos.Column, tableName)
@ -165,13 +167,13 @@ func (p *ExecutionPlanner) compileShowColumnsStatement(stmt *parser.ShowColumnsS
dataType: parser.NewDataTypeString(),
}}
return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseColumns(index)), p.sql), nil
return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseColumns(tbl)), p.sql), nil
}
func (p *ExecutionPlanner) compileShowCreateTableStatement(stmt *parser.ShowCreateTableStatement) (_ types.PlanOperator, err error) {
tableName := parser.IdentName(stmt.TableName)
_, err = p.schemaAPI.IndexInfo(context.Background(), tableName)
if err != nil {
tname := dax.TableName(tableName)
if _, err := p.schemaAPI.TableByName(context.Background(), tname); err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrTableNotFound(stmt.TableName.NamePos.Line, stmt.TableName.NamePos.Column, tableName)
}

103
sql3/planner/executionplannersystemtables.go
View file

@ -6,6 +6,7 @@ import (
"context"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/pkg/errors"
@ -18,66 +19,100 @@ type systemTableDefintionsWrapper struct {
schemaAPI pilosa.SchemaAPI
}
func newSystemTableDefintionsWrapper(schemaAPI pilosa.SchemaAPI) *systemTableDefintionsWrapper {
func newSystemTableDefintionsWrapper(api pilosa.SchemaAPI) *systemTableDefintionsWrapper {
return &systemTableDefintionsWrapper{
schemaAPI: schemaAPI,
schemaAPI: api,
}
}
func (s *systemTableDefintionsWrapper) CreateIndexAndFields(ctx context.Context, indexName string, options pilosa.IndexOptions, fields []pilosa.CreateFieldObj) error {
return s.schemaAPI.CreateIndexAndFields(ctx, indexName, options, fields)
}
func (s *systemTableDefintionsWrapper) CreateField(ctx context.Context, indexName string, fieldName string, opts ...pilosa.FieldOption) (*pilosa.Field, error) {
return s.schemaAPI.CreateField(ctx, indexName, fieldName, opts...)
}
func (s *systemTableDefintionsWrapper) DeleteField(ctx context.Context, indexName string, fieldName string) error {
return s.schemaAPI.DeleteField(ctx, indexName, fieldName)
}
func (s *systemTableDefintionsWrapper) DeleteIndex(ctx context.Context, indexName string) error {
return s.schemaAPI.DeleteIndex(ctx, indexName)
}
func (s *systemTableDefintionsWrapper) IndexInfo(ctx context.Context, indexName string) (*pilosa.IndexInfo, error) {
i, err := s.schemaAPI.IndexInfo(ctx, indexName)
func (s *systemTableDefintionsWrapper) TableByName(ctx context.Context, tname dax.TableName) (*dax.Table, error) {
tbl, err := s.schemaAPI.TableByName(ctx, tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
st, ok := systemTables[indexName]
st, ok := systemTables[string(tname)]
if !ok {
return nil, pilosa.ErrIndexNotFound
}
return indexInfoFromSystemTable(st)
return indexInfoFromSystemTableB(st)
}
return nil, err
}
return i, nil
return tbl, nil
}
func (s *systemTableDefintionsWrapper) FieldInfo(ctx context.Context, indexName, fieldName string) (*pilosa.FieldInfo, error) {
return nil, pilosa.ErrNotImplemented
func (s *systemTableDefintionsWrapper) TableByID(ctx context.Context, tid dax.TableID) (*dax.Table, error) {
return s.schemaAPI.TableByID(ctx, tid)
}
func (s *systemTableDefintionsWrapper) Schema(ctx context.Context, withViews bool) ([]*pilosa.IndexInfo, error) {
schema, err := s.schemaAPI.Schema(ctx, withViews)
func (s *systemTableDefintionsWrapper) Tables(ctx context.Context) ([]*dax.Table, error) {
tbls, err := s.schemaAPI.Tables(ctx)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting tables")
}
for _, st := range systemTables {
i, err := indexInfoFromSystemTable(st)
// Append the system tables.
for tblName, st := range systemTables {
ii, err := indexInfoFromSystemTable(st)
if err != nil {
return nil, err
return nil, errors.Wrapf(err, "converting system table to table: %s", tblName)
}
schema = append(schema, i)
tbls = append(tbls, pilosa.IndexInfoToTable(ii))
}
return schema, err
return tbls, nil
}
func (s *systemTableDefintionsWrapper) 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 {
return s.schemaAPI.CreateField(ctx, tname, fld)
}
func (s *systemTableDefintionsWrapper) 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 {
return s.schemaAPI.DeleteField(ctx, tname, fname)
}
func indexInfoFromSystemTableB(st *systemTable) (*dax.Table, error) {
fields := make([]*dax.Field, 0)
for _, f := range st.schema {
var baseType dax.BaseType
switch f.Type.(type) {
case *parser.DataTypeInt:
baseType = dax.BaseTypeInt
case *parser.DataTypeBool:
baseType = dax.BaseTypeBool
case *parser.DataTypeString:
baseType = dax.BaseTypeString
case *parser.DataTypeTimestamp:
baseType = dax.BaseTypeTimestamp
default:
return nil, sql3.NewErrInternalf("unexpected system table field type '%T'", f.Type)
}
fld := &dax.Field{
Name: dax.FieldName(f.ColumnName),
Type: baseType,
}
fields = append(fields, fld)
}
tbl := &dax.Table{
Name: dax.TableName(st.name),
Fields: fields,
}
return tbl, nil
}
func indexInfoFromSystemTable(st *systemTable) (*pilosa.IndexInfo, error) {
fields := make([]*pilosa.FieldInfo, 0)
for _, f := range st.schema {

14
sql3/planner/opaltertable.go
View file

@ -6,6 +6,8 @@ import (
"context"
"fmt"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
@ -87,13 +89,21 @@ var _ types.RowIterator = (*alterTableRowIter)(nil)
func (i *alterTableRowIter) Next(ctx context.Context) (types.Row, error) {
switch i.operation {
case alterOpAdd:
_, err := i.planner.schemaAPI.CreateField(ctx, i.tableName, i.columnDef.name, i.columnDef.fos...)
tname := dax.TableName(i.tableName)
fname := dax.FieldName(i.columnDef.name)
fos := i.columnDef.fos
fld, err := pilosa.FieldFromFieldOptions(fname, fos...)
if err != nil {
return nil, err
}
if err := i.planner.schemaAPI.CreateField(ctx, tname, fld); err != nil {
return nil, err
}
case alterOpDrop:
err := i.planner.schemaAPI.DeleteField(ctx, i.tableName, i.oldColumnName)
err := i.planner.schemaAPI.DeleteField(ctx, dax.TableName(i.tableName), dax.FieldName(i.oldColumnName))
if err != nil {
return nil, err
}

40
sql3/planner/opcreatetable.go
View file

@ -7,6 +7,7 @@ import (
"fmt"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3/planner/types"
"github.com/pkg/errors"
)
@ -100,23 +101,38 @@ var _ types.RowIterator = (*createTableRowIter)(nil)
func (i *createTableRowIter) Next(ctx context.Context) (types.Row, error) {
//create the table
options := pilosa.IndexOptions{
Keys: i.isKeyed,
TrackExistence: true,
PartitionN: i.keyPartitions,
Description: i.description,
fields := make([]*dax.Field, 0, len(i.columns)+1)
// add the _id column
var idType dax.BaseType = dax.BaseTypeID
if i.isKeyed {
idType = dax.BaseTypeString
}
fields = append(fields, &dax.Field{
Name: "_id",
Type: idType,
})
for _, f := range i.columns {
fld, err := pilosa.FieldFromFieldOptions(dax.FieldName(f.name), f.fos...)
if err != nil {
return nil, errors.Wrapf(err, "creating field from field options: %s", f.name)
}
fields = append(fields, fld)
}
fields := make([]pilosa.CreateFieldObj, len(i.columns))
for i, f := range i.columns {
fields[i] = pilosa.CreateFieldObj{
Name: f.name,
Options: f.fos,
}
tbl := &dax.Table{
Name: dax.TableName(i.tableName),
Fields: fields,
// TODO(tlt): once we can support different partitionN's per table,
// replace dax.DefaultPartitionN with i.keyPartitions.
PartitionN: dax.DefaultPartitionN,
// TODO(tlt): add Description to dax.Table; = i.description
}
// TODO (pok) add ability to add description here
if err := i.planner.schemaAPI.CreateIndexAndFields(ctx, i.tableName, options, fields); err != nil {
if err := i.planner.schemaAPI.CreateTable(ctx, tbl); err != nil {
if _, ok := errors.Cause(err).(pilosa.ConflictError); ok {
if i.failIfExists {
return nil, err

3
sql3/planner/opdroptable.go
View file

@ -7,6 +7,7 @@ import (
"fmt"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
@ -81,7 +82,7 @@ func (i *dropTableRowIter) Next(ctx context.Context) (types.Row, error) {
return nil, err
}
err = i.planner.schemaAPI.DeleteIndex(ctx, i.index.Name)
err = i.planner.schemaAPI.DeleteTable(ctx, dax.TableName(i.index.Name))
if err != nil {
return nil, err
}

27
sql3/planner/opfeaturebasecolumns.go
View file

@ -7,20 +7,20 @@ import (
"fmt"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/sql3/parser"
"github.com/molecula/featurebase/v3/sql3/planner/types"
)
// PlanOpFeatureBaseColumns wraps an Index that is returned from schemaAPI.Schema().
type PlanOpFeatureBaseColumns struct {
index *pilosa.IndexInfo
tbl *dax.Table
warnings []string
}
func NewPlanOpFeatureBaseColumns(index *pilosa.IndexInfo) *PlanOpFeatureBaseColumns {
func NewPlanOpFeatureBaseColumns(tbl *dax.Table) *PlanOpFeatureBaseColumns {
node := &PlanOpFeatureBaseColumns{
index: index,
tbl: tbl,
warnings: make([]string, 0),
}
return node
@ -135,34 +135,35 @@ func (p *PlanOpFeatureBaseColumns) Children() []types.PlanOperator {
func (p *PlanOpFeatureBaseColumns) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
return &showColumnsRowIter{
index: p.index,
tbl: p.tbl,
}, nil
}
func (p *PlanOpFeatureBaseColumns) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
return NewPlanOpFeatureBaseColumns(p.index), nil
return NewPlanOpFeatureBaseColumns(p.tbl), nil
}
type showColumnsRowIter struct {
index *pilosa.IndexInfo
tbl *dax.Table
rowIndex int
}
var _ types.RowIterator = (*showColumnsRowIter)(nil)
func (i *showColumnsRowIter) Next(ctx context.Context) (types.Row, error) {
if i.rowIndex < len(i.index.Fields) {
fields := i.index.Fields
if i.rowIndex < len(i.tbl.Fields) {
fields := i.tbl.Fields
tm := time.Unix(0, fields[i.rowIndex].CreatedAt)
//tm := time.Unix(0, fields[i.rowIndex].CreatedAt)
tm := time.Unix(0, 0)
row := []interface{}{
fields[i.rowIndex].Name,
fields[i.rowIndex].Name,
fieldSQLDataType(fields[i.rowIndex]).TypeDescription(),
fields[i.rowIndex].Options.Type,
fields[i.rowIndex].Type,
fields[i.rowIndex].Type,
tm.Format(time.RFC3339),
fields[i.rowIndex].Options.Keys,
fields[i.rowIndex].StringKeys(),
fields[i.rowIndex].Options.CacheType,
fields[i.rowIndex].Options.CacheSize,
fields[i.rowIndex].Options.Scale,

5
sql3/planner/opinsert.go
View file

@ -10,6 +10,7 @@ import (
pilosa "github.com/molecula/featurebase/v3"
fbbatch "github.com/molecula/featurebase/v3/batch"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/planner/types"
@ -143,10 +144,12 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
// IndexInfo used in the import (and created below) will be based on the
// information from idxInfoBase, but the fields may be a limited subset, and
// may be in a different order.
idxInfoBase, err := i.planner.schemaAPI.IndexInfo(ctx, i.tableName)
tname := dax.TableName(i.tableName)
tbl, err := i.planner.schemaAPI.TableByName(ctx, tname)
if err != nil {
return nil, sql3.NewErrTableNotFound(0, 0, i.tableName)
}
idxInfoBase := pilosa.TableToIndexInfo(tbl)
// idxInfo is a subset of idxInfoBase, containing only those fields included
// in the INSERT INTO statement (i.e. only i.targetcolumns), and in the

21
sql3/planner/oppqltablescan.go
View file

@ -8,6 +8,7 @@ import (
"strings"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/sql3"
"github.com/molecula/featurebase/v3/sql3/parser"
@ -80,18 +81,19 @@ func (p *PlanOpPQLTableScan) UpdateFilters(filterCondition types.PlanExpression)
func (p *PlanOpPQLTableScan) Schema() types.Schema {
result := make(types.Schema, 0)
table, err := p.planner.schemaAPI.IndexInfo(context.Background(), p.tableName)
tname := dax.TableName(p.tableName)
table, err := p.planner.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
return result
}
for _, col := range p.columns {
for _, fld := range table.Fields {
if strings.EqualFold(fld.Name, col) {
if strings.EqualFold(string(fld.Name), col) {
result = append(result, &types.PlannerColumn{
ColumnName: fld.Name,
ColumnName: string(fld.Name),
RelationName: p.tableName,
Type: fieldSQLDataType(fld),
Type: fieldSQLDataType(pilosa.FieldToFieldInfo(fld)),
})
break
}
@ -147,7 +149,8 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) {
}
//go get the schema def and map names to indexes in the resultant row
table, err := i.planner.schemaAPI.IndexInfo(context.Background(), i.tableName)
tname := dax.TableName(i.tableName)
table, err := i.planner.schemaAPI.TableByName(context.Background(), tname)
if err != nil {
if errors.Is(err, pilosa.ErrIndexNotFound) {
return nil, sql3.NewErrInternalf("table not found '%s'", i.tableName)
@ -159,12 +162,12 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) {
i.columnMap = make(map[string]*targetColumn)
for idx, col := range i.columns {
for _, fld := range table.Fields {
if strings.EqualFold(col, fld.Name) {
i.columnMap[fld.Name] = &targetColumn{
if strings.EqualFold(col, string(fld.Name)) {
i.columnMap[string(fld.Name)] = &targetColumn{
columnIdx: idx,
srcColumnIdx: -1,
columnName: fld.Name,
dataType: fieldSQLDataType(fld),
columnName: string(fld.Name),
dataType: fieldSQLDataType(pilosa.FieldToFieldInfo(fld)),
}
break
}

21
sql3/planner/opsystemtable.go
View file

@ -407,32 +407,27 @@ var _ types.RowIterator = (*fbTableDDLRowIter)(nil)
func (i *fbTableDDLRowIter) Next(ctx context.Context) (types.Row, error) {
if i.result == nil {
schema, err := i.planner.schemaAPI.Schema(ctx, false)
tbls, err := i.planner.schemaAPI.Tables(ctx)
if err != nil {
return nil, err
}
i.result = make([]*fbTableDDLRow, len(schema))
for idx, table := range schema {
index, err := i.planner.schemaAPI.IndexInfo(context.Background(), table.Name)
if err != nil {
return nil, err
}
i.result = make([]*fbTableDDLRow, len(tbls))
for idx, tbl := range tbls {
// build the ddl for this table
var buf bytes.Buffer
buf.WriteString("create table ")
fmt.Fprintf(&buf, "%s", index.Name)
fmt.Fprintf(&buf, "%s", tbl.Name)
buf.WriteString(" (")
for idx, col := range index.Fields {
for idx, col := range tbl.Fields {
if idx > 0 {
buf.WriteString(", ")
}
fmt.Fprintf(&buf, "%s", col.Name)
dataType := fieldSQLDataType(col)
dataType := fieldSQLDataType(pilosa.FieldToFieldInfo(col))
fmt.Fprintf(&buf, " %s", dataType.TypeDescription())
switch dt := dataType.(type) {
@ -504,8 +499,8 @@ func (i *fbTableDDLRowIter) Next(ctx context.Context) (types.Row, error) {
ddl := buf.String()
i.result[idx] = &fbTableDDLRow{
id: table.Name,
name: table.Name,
id: string(tbl.Name),
name: string(tbl.Name),
ddl: ddl,
}
}

25
sql3/planner/types/compile.go Normal file
View file

@ -0,0 +1,25 @@
package types
import (
"context"
"github.com/molecula/featurebase/v3/sql3/parser"
)
type CompilePlanner interface {
CompilePlan(context.Context, parser.Statement) (PlanOperator, error)
}
// Ensure type implements interface.
var _ CompilePlanner = (*nopCompilePlanner)(nil)
// nopCompilePlanner is a no-op implementation of the CompilePlanner interface.
type nopCompilePlanner struct{}
func NewNopCompilePlanner() *nopCompilePlanner {
return &nopCompilePlanner{}
}
func (p *nopCompilePlanner) CompilePlan(ctx context.Context, stmt parser.Statement) (PlanOperator, error) {
return nil, nil
}

2
sql3/sql_complex_test.go
View file

@ -524,7 +524,7 @@ func TestPlanner_CoverCreateTable(t *testing.T) {
}
fbField, ok := fbFields[fld.name]
assert.True(t, ok)
assert.True(t, ok, "expected field: %s", fld.name)
assert.Equal(t, fld.expOptions, fbField.Options)
})
}