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

@@ -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")
 	}

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)
 }

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
-}

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

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 {

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()

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
 		}

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 {

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)