From a15783cb497e80e708080f4af97c75086cde6e54 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Thu, 8 Dec 2022 11:35:17 -0600 Subject: [PATCH] Convert SchemaAPI interface to use dax.Table instead of pilosa.IndexInfo (#2336) * WIP: Convert SchemaAPI to be DAX-centric * Tables(), CreateField() * CreateTable(), DeleteTable(), DeleteField() * More cleanup * Remove the old SchemaAPI --- api.go | 91 +--- api_directive.go | 53 +-- batch/Makefile | 2 +- batch/batch.go | 2 - batch/batch_test.go | 132 +++--- batch/docker-compose.yml | 2 +- client/api.go | 116 ++--- dax/queryer/importer.go | 2 +- dax/queryer/schema_api.go | 452 ++---------------- dax/queryer/schema_info_api.go | 79 ++++ idk/docker-compose.yml | 2 - idk/ingest_test.go | 5 +- schema.go | 460 +++++++++++++++++++ server/server.go | 2 +- sql3/planner/compilealtertable.go | 12 +- sql3/planner/compilebulkinsert.go | 23 +- sql3/planner/compiledroptable.go | 6 +- sql3/planner/compileinsert.go | 33 +- sql3/planner/compileselect.go | 12 +- sql3/planner/compileshow.go | 16 +- sql3/planner/executionplannersystemtables.go | 103 +++-- sql3/planner/opaltertable.go | 14 +- sql3/planner/opcreatetable.go | 40 +- sql3/planner/opdroptable.go | 3 +- sql3/planner/opfeaturebasecolumns.go | 27 +- sql3/planner/opinsert.go | 5 +- sql3/planner/oppqltablescan.go | 21 +- sql3/planner/opsystemtable.go | 21 +- sql3/planner/types/compile.go | 25 + sql3/sql_complex_test.go | 2 +- 30 files changed, 924 insertions(+), 839 deletions(-) create mode 100644 dax/queryer/schema_info_api.go create mode 100644 schema.go create mode 100644 sql3/planner/types/compile.go diff --git a/api.go b/api.go index 0cec23368..aadd99b55 100644 --- a/api.go +++ b/api.go @@ -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 { diff --git a/api_directive.go b/api_directive.go index ebbee56a8..ce2cf6e1f 100644 --- a/api_directive.go +++ b/api_directive.go @@ -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 { diff --git a/batch/Makefile b/batch/Makefile index 6599c4cf8..f53e06d31 100644 --- a/batch/Makefile +++ b/batch/Makefile @@ -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) diff --git a/batch/batch.go b/batch/batch.go index 35956f37b..270b21e63 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -183,8 +183,6 @@ type Batch struct { clearFrags fragments useShardTransactionalEndpoint bool - - mdsHost string } func (b *Batch) Len() int { return len(b.ids) } diff --git a/batch/batch_test.go b/batch/batch_test.go index c410db53b..40625dddf 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -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, - )) -} diff --git a/batch/docker-compose.yml b/batch/docker-compose.yml index b3a897c04..765018f2b 100644 --- a/batch/docker-compose.yml +++ b/batch/docker-compose.yml @@ -23,7 +23,7 @@ services: wait: depends_on: - - "featurebase" + - "featurebase" build: context: . dockerfile: Dockerfile-wait diff --git a/client/api.go b/client/api.go index 6f25e0bcd..e46bf2635 100644 --- a/client/api.go +++ b/client/api.go @@ -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 diff --git a/dax/queryer/importer.go b/dax/queryer/importer.go index 05b7facbe..cb684fb5b 100644 --- a/dax/queryer/importer.go +++ b/dax/queryer/importer.go @@ -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 } diff --git a/dax/queryer/schema_api.go b/dax/queryer/schema_api.go index 15ad613fe..a0a0ac53d 100644 --- a/dax/queryer/schema_api.go +++ b/dax/queryer/schema_api.go @@ -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) } diff --git a/dax/queryer/schema_info_api.go b/dax/queryer/schema_info_api.go new file mode 100644 index 000000000..3c3984c69 --- /dev/null +++ b/dax/queryer/schema_info_api.go @@ -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 +} diff --git a/idk/docker-compose.yml b/idk/docker-compose.yml index a78f10a55..9f0e3097b 100644 --- a/idk/docker-compose.yml +++ b/idk/docker-compose.yml @@ -125,8 +125,6 @@ services: - ./docker-sasl/ssl_keys:/ssl_keys - ./testdata:/testdata depends_on: - #- kafka - #- postgres - fakeidp wait: build: diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 33cfa9023..c9fe5403a 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -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 { diff --git a/schema.go b/schema.go new file mode 100644 index 000000000..35b10d363 --- /dev/null +++ b/schema.go @@ -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 +} diff --git a/server/server.go b/server/server.go index ff896ef10..6864b66ac 100644 --- a/server/server.go +++ b/server/server.go @@ -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} diff --git a/sql3/planner/compilealtertable.go b/sql3/planner/compilealtertable.go index 31f2f2a58..53bc9abe1 100644 --- a/sql3/planner/compilealtertable.go +++ b/sql3/planner/compilealtertable.go @@ -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) } } diff --git a/sql3/planner/compilebulkinsert.go b/sql3/planner/compilebulkinsert.go index 5b18324df..9ec23a215 100644 --- a/sql3/planner/compilebulkinsert.go +++ b/sql3/planner/compilebulkinsert.go @@ -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 diff --git a/sql3/planner/compiledroptable.go b/sql3/planner/compiledroptable.go index 7837e6b29..8909ca624 100644 --- a/sql3/planner/compiledroptable.go +++ b/sql3/planner/compiledroptable.go @@ -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 } diff --git a/sql3/planner/compileinsert.go b/sql3/planner/compileinsert.go index 0d7bc8326..e553d17d1 100644 --- a/sql3/planner/compileinsert.go +++ b/sql3/planner/compileinsert.go @@ -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 } diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index e7af5c07d..0e5f2d777 100644 --- a/sql3/planner/compileselect.go +++ b/sql3/planner/compileselect.go @@ -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) } diff --git a/sql3/planner/compileshow.go b/sql3/planner/compileshow.go index 192bc9389..087ca70c1 100644 --- a/sql3/planner/compileshow.go +++ b/sql3/planner/compileshow.go @@ -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) } diff --git a/sql3/planner/executionplannersystemtables.go b/sql3/planner/executionplannersystemtables.go index 892cc1a3b..42537a1da 100644 --- a/sql3/planner/executionplannersystemtables.go +++ b/sql3/planner/executionplannersystemtables.go @@ -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 { diff --git a/sql3/planner/opaltertable.go b/sql3/planner/opaltertable.go index a230fe503..3d8ce31bd 100644 --- a/sql3/planner/opaltertable.go +++ b/sql3/planner/opaltertable.go @@ -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 } diff --git a/sql3/planner/opcreatetable.go b/sql3/planner/opcreatetable.go index d7625c29f..7939c2aff 100644 --- a/sql3/planner/opcreatetable.go +++ b/sql3/planner/opcreatetable.go @@ -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 diff --git a/sql3/planner/opdroptable.go b/sql3/planner/opdroptable.go index 942233327..2d67abfae 100644 --- a/sql3/planner/opdroptable.go +++ b/sql3/planner/opdroptable.go @@ -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 } diff --git a/sql3/planner/opfeaturebasecolumns.go b/sql3/planner/opfeaturebasecolumns.go index 4d2995ff7..076029bfc 100644 --- a/sql3/planner/opfeaturebasecolumns.go +++ b/sql3/planner/opfeaturebasecolumns.go @@ -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, diff --git a/sql3/planner/opinsert.go b/sql3/planner/opinsert.go index b12b1cd2c..42bf0c748 100644 --- a/sql3/planner/opinsert.go +++ b/sql3/planner/opinsert.go @@ -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 diff --git a/sql3/planner/oppqltablescan.go b/sql3/planner/oppqltablescan.go index 4bade625d..d3a0b6a8f 100644 --- a/sql3/planner/oppqltablescan.go +++ b/sql3/planner/oppqltablescan.go @@ -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 } diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go index 182f015b6..33c6137be 100644 --- a/sql3/planner/opsystemtable.go +++ b/sql3/planner/opsystemtable.go @@ -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, } } diff --git a/sql3/planner/types/compile.go b/sql3/planner/types/compile.go new file mode 100644 index 000000000..317b93816 --- /dev/null +++ b/sql3/planner/types/compile.go @@ -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 +} diff --git a/sql3/sql_complex_test.go b/sql3/sql_complex_test.go index ee02a4733..82a02c5b0 100644 --- a/sql3/sql_complex_test.go +++ b/sql3/sql_complex_test.go @@ -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) }) }