From a61d1a9571ab2768d5617d36bf75a741ca57a7d5 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Sun, 11 Dec 2022 11:45:10 -0600 Subject: [PATCH] Introduce TableKeyer interface; use in Execute() calls as "index" (#2350) This commit introduces an interface called `TableKeyer` which anything that means to represent a "table" can implement. Examples are `dax.QualifiedTable`, `dax.Table`, and `string` (for legacy pilosa calls where Execute simply took `index string`). In the case of `orchestrator.Execute()` and `qualifiedOrchestrator.Execute()`, we are intentionally strict about which type of `TableKeyer` the respective method accepts. If we find, in the future, this is too restrictive, we can loosen that; but for now it helps us understand what is expected. --- api.go | 2 +- dax/queryer/orchestrator.go | 27 ++++++++++++++------------- dax/queryer/queryer.go | 25 ++++++++----------------- dax/table.go | 23 ++++++++++++++++++++++- dax/test/dax/dax_test.go | 9 ++++----- executor.go | 6 ++++-- sql3/planner/oppqlaggregate.go | 8 +++++++- sql3/planner/oppqlgroupby.go | 14 +++++++++++--- sql3/planner/oppqltablescan.go | 16 ++++++++++++---- 9 files changed, 83 insertions(+), 47 deletions(-) diff --git a/api.go b/api.go index aadd99b55..386b28f30 100644 --- a/api.go +++ b/api.go @@ -249,7 +249,7 @@ func (api *API) query(ctx context.Context, req *QueryRequest) (QueryResponse, er EmbeddedData: req.EmbeddedData, // precomputed values that needed to be passed with the request MaxMemory: req.MaxMemory, } - resp, err := api.server.executor.Execute(ctx, req.Index, q, req.Shards, execOpts) + resp, err := api.server.executor.Execute(ctx, dax.StringTableKeyer(req.Index), q, req.Shards, execOpts) if err != nil { return QueryResponse{}, errors.Wrap(err, "executing") } diff --git a/dax/queryer/orchestrator.go b/dax/queryer/orchestrator.go index 63842cc64..72429174b 100644 --- a/dax/queryer/orchestrator.go +++ b/dax/queryer/orchestrator.go @@ -105,13 +105,20 @@ func emptyResult(c *pql.Call) interface{} { } // Execute executes a PQL query. -func (o *orchestrator) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *featurebase.ExecOptions) (featurebase.QueryResponse, error) { +func (o *orchestrator) Execute(ctx context.Context, tableKeyer dax.TableKeyer, q *pql.Query, shards []uint64, opt *featurebase.ExecOptions) (featurebase.QueryResponse, error) { span, ctx := tracing.StartSpanFromContext(ctx, "orchestrator.Execute") span.LogKV("pql", q.String()) defer span.Finish() resp := featurebase.QueryResponse{} + qtbl, ok := tableKeyer.(*dax.QualifiedTable) + if !ok { + return resp, errors.New(errors.ErrUncoded, "orchestrator.Execute expects a dax.QualifiedTable") + } + + index := string(qtbl.Key()) + // Check for query cancellation. if err := validateQueryContext(ctx); err != nil { return resp, err @@ -3465,21 +3472,15 @@ func newQualifiedOrchestrator(orch *orchestrator, qual dax.TableQualifier, schem } } -func (o *qualifiedOrchestrator) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *featurebase.ExecOptions) (featurebase.QueryResponse, error) { +func (o *qualifiedOrchestrator) Execute(ctx context.Context, tableKeyer dax.TableKeyer, q *pql.Query, shards []uint64, opt *featurebase.ExecOptions) (featurebase.QueryResponse, error) { resp := featurebase.QueryResponse{} - tkey, err := o.indexToQualifiedTableKey(ctx, index) - if err != nil { - return resp, errors.Wrap(err, "converting index to qualified table key") + tbl, ok := tableKeyer.(*dax.Table) + if !ok { + return resp, errors.New(errors.ErrUncoded, "qualifiedOrchestrator.Execute expects a dax.Table") } - return o.orchestrator.Execute(ctx, string(tkey), q, shards, opt) -} + qtbl := dax.NewQualifiedTable(o.qual, tbl) -func (o *qualifiedOrchestrator) indexToQualifiedTableKey(ctx context.Context, index string) (dax.TableKey, error) { - qtid, err := o.schemar.TableID(ctx, o.qual, dax.TableName(index)) - if err != nil { - return "", errors.Wrap(err, "converting index to qualified table id") - } - return qtid.Key(), nil + return o.orchestrator.Execute(ctx, qtbl, q, shards, opt) } diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 667172909..8074b4c2b 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -214,12 +214,17 @@ func (q *Queryer) QueryPQL(ctx context.Context, qual dax.TableQualifier, table d return nil, errors.Errorf("must have exactly 1 query, but got: %+v", qry.Calls) } - tkey, err := q.indexToQualifiedTableKey(ctx, qual, string(table)) + qtid, err := q.mds.TableID(ctx, qual, dax.TableName(table)) if err != nil { - return nil, errors.Wrapf(err, "converting index to qualified table key: %s", table) + return nil, errors.Wrap(err, "converting index to qualified table id") } - results, err := q.orchestrator.Execute(ctx, string(tkey), qry, nil, &featurebase.ExecOptions{}) + qtbl, err := q.mds.Table(ctx, qtid) + if err != nil { + return nil, errors.Wrap(err, "getting table for qtid") + } + + results, err := q.orchestrator.Execute(ctx, qtbl, qry, nil, &featurebase.ExecOptions{}) if err != nil { return nil, errors.Wrap(err, "orchestrator.Execute") } @@ -321,17 +326,3 @@ func rowToSliceInterface(header []*fbproto.ColumnInfo, row *fbproto.Row) []inter } return ret } - -// TODO(tlt): this method was copied from queryer/batchImporter. Can we centralize -// this logic? -func (q *Queryer) indexToQualifiedTableKey(ctx context.Context, qual dax.TableQualifier, index string) (dax.TableKey, error) { - if strings.HasPrefix(index, dax.PrefixTable+dax.TableKeyDelimiter) { - return dax.TableKey(index), nil - } - - qtid, err := q.mds.TableID(ctx, qual, dax.TableName(index)) - if err != nil { - return "", errors.Wrap(err, "converting index to qualified table id") - } - return qtid.Key(), nil -} diff --git a/dax/table.go b/dax/table.go index ab9c566da..ddc1b0719 100644 --- a/dax/table.go +++ b/dax/table.go @@ -103,6 +103,23 @@ type OrganizationID string // value could be any string. type DatabaseID string +// TableKeyer is an interface implemented by any type which can produce, and be +// represented by, a TableKey. In the case of a QualifiedTable, its TableKey +// might be something like `tbl__org__db__tableid`, while a general pilosa +// implemenation might represent a table as a basic table name `foo`. +type TableKeyer interface { + Key() TableKey +} + +// StringTableKeyer is a helper type which can wrap a string, making it a +// TableKeyer. This is useful for certain calls to Execute() which take a string +// index name. +type StringTableKeyer string + +func (s StringTableKeyer) Key() TableKey { + return TableKey(s) +} + // TableKey is a globally unique identifier for a table; it is effectively the // compound key: (org, database, table). This is (hopefully) the value that will // be used when interfacing with services which are unaware of table qualifiers. @@ -167,6 +184,10 @@ type Table struct { CreatedAt int64 `json:"createdAt,omitempty"` } +func (t *Table) Key() TableKey { + return TableKey(t.ID) +} + // CreateID generates a unique identifier for Table. If Table has already been // assigned an ID, then an error is returned. func (t *Table) CreateID() (TableID, error) { @@ -433,7 +454,7 @@ func (qtid QualifiedTableID) Key() TableKey { } // Equals returns true if `other` is the same as qtid. Note: the `Name` value is -// ignored in this comparison; only `TableQaulifer` and `ID` are considered. +// ignored in this comparison; only `TableQualifer` and `ID` are considered. func (qtid QualifiedTableID) Equals(other QualifiedTableID) bool { if qtid.TableQualifier == other.TableQualifier && qtid.ID == other.ID { return true diff --git a/dax/test/dax/dax_test.go b/dax/test/dax/dax_test.go index 189dd9259..708969313 100644 --- a/dax/test/dax/dax_test.go +++ b/dax/test/dax/dax_test.go @@ -107,11 +107,10 @@ func TestDAXIntegration(t *testing.T) { // skips is a list of tests which are currently not passing in dax. We // need to get these passing before alpha. skips := []string{ - "testinsert/test-5", // error messages differ - "percentile_test/test-6", // related to TODO in orchestrator.executePercentile - "innerjointest/innerjoin-aggregate-groupby", // join test which won't work until we support multiple tables - "alterTable/alterTableBadTable", // looks like table does not exist is a different error in DAX - "top-tests/test-1", // don't know why this is failing at all + "testinsert/test-5", // error messages differ + "percentile_test/test-6", // related to TODO in orchestrator.executePercentile + "alterTable/alterTableBadTable", // looks like table does not exist is a different error in DAX + "top-tests/test-1", // don't know why this is failing at all } doSkip := func(name string) bool { diff --git a/executor.go b/executor.go index 03f27518f..1d052b509 100644 --- a/executor.go +++ b/executor.go @@ -46,7 +46,7 @@ const ( ) type Executor interface { - Execute(context.Context, string, *pql.Query, []uint64, *ExecOptions) (QueryResponse, error) + Execute(context.Context, dax.TableKeyer, *pql.Query, []uint64, *ExecOptions) (QueryResponse, error) } // executor recursively executes calls in a PQL query across all shards. @@ -177,7 +177,9 @@ func (e *executor) InitStats() { } // Execute executes a PQL query. -func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *ExecOptions) (QueryResponse, error) { +func (e *executor) Execute(ctx context.Context, tableKeyer dax.TableKeyer, q *pql.Query, shards []uint64, opt *ExecOptions) (QueryResponse, error) { + index := string(tableKeyer.Key()) + span, ctx := tracing.StartSpanFromContext(ctx, "executor.Execute") span.LogKV("pql", q.String()) defer span.Finish() diff --git a/sql3/planner/oppqlaggregate.go b/sql3/planner/oppqlaggregate.go index 27f920561..345141be6 100644 --- a/sql3/planner/oppqlaggregate.go +++ b/sql3/planner/oppqlaggregate.go @@ -7,6 +7,7 @@ import ( "fmt" 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" @@ -251,7 +252,12 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) { return nil, sql3.NewErrInternalf("unhandled aggregate type '%d'", i.aggregate.AggType()) } - queryResponse, err := i.planner.executor.Execute(ctx, i.tableName, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) + tbl, err := i.planner.schemaAPI.TableByName(ctx, dax.TableName(i.tableName)) + if err != nil { + return nil, sql3.NewErrTableNotFound(0, 0, i.tableName) + } + + queryResponse, err := i.planner.executor.Execute(ctx, tbl, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) if err != nil { return nil, err } diff --git a/sql3/planner/oppqlgroupby.go b/sql3/planner/oppqlgroupby.go index 5e3049a8e..8275fad1c 100644 --- a/sql3/planner/oppqlgroupby.go +++ b/sql3/planner/oppqlgroupby.go @@ -7,6 +7,7 @@ import ( "fmt" 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" @@ -205,15 +206,22 @@ func (i *pqlGroupByRowIter) Next(ctx context.Context) (types.Row, error) { call.Args["filter"] = cond } - queryResponse, err := i.planner.executor.Execute(ctx, i.tableName, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) + tbl, err := i.planner.schemaAPI.TableByName(ctx, dax.TableName(i.tableName)) + if err != nil { + return nil, sql3.NewErrTableNotFound(0, 0, i.tableName) + } + + queryResponse, err := i.planner.executor.Execute(ctx, tbl, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) if err != nil { return nil, err } - tbl, ok := queryResponse.Results[0].(*pilosa.GroupCounts) + + gcs, ok := queryResponse.Results[0].(*pilosa.GroupCounts) if !ok { return nil, sql3.NewErrInternalf("unexpected Extract() result type: %T", queryResponse.Results[0]) } - i.result = tbl.Groups() + + i.result = gcs.Groups() } if len(i.result) > 0 { diff --git a/sql3/planner/oppqltablescan.go b/sql3/planner/oppqltablescan.go index d3a0b6a8f..987564d9b 100644 --- a/sql3/planner/oppqltablescan.go +++ b/sql3/planner/oppqltablescan.go @@ -216,18 +216,26 @@ func (i *tableScanRowIter) Next(ctx context.Context) (types.Row, error) { }, ) } - queryResponse, err := i.planner.executor.Execute(ctx, i.tableName, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) + + tbl, err := i.planner.schemaAPI.TableByName(ctx, dax.TableName(i.tableName)) + if err != nil { + return nil, sql3.NewErrTableNotFound(0, 0, i.tableName) + } + + queryResponse, err := i.planner.executor.Execute(ctx, tbl, &pql.Query{Calls: []*pql.Call{call}}, nil, nil) if err != nil { return nil, err } - tbl, ok := queryResponse.Results[0].(pilosa.ExtractedTable) + + extbl, ok := queryResponse.Results[0].(pilosa.ExtractedTable) if !ok { return nil, sql3.NewErrInternalf("unexpected Extract() result type: %T", queryResponse.Results[0]) } - i.result = tbl.Columns + + i.result = extbl.Columns //set the source index - for idx, fld := range tbl.Fields { + for idx, fld := range extbl.Fields { mappedColumn, ok := i.columnMap[fld.Name] if !ok { return nil, sql3.NewErrInternalf("mapped column not found for column named '%s'", fld.Name)