Stub out the queryer/systemAPI implementation

This is used to fix the fb_exec_requests query; by default, it was doing
"fanout", but that doesn't make sense in serverless. So we needed a way
to prevent fanout when running Serverless. This gets that distinction
from the `SystemAPI.ClusterName()` method.

api.go

@@ -40,6 +40,13 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+// Flavor is used to distinguish between serverless and classic. For the moment,
+// we use this in the "ClusterName" returned by a serverless instance.
+const (
+	FlavorClassic    = "Classic"
+	FlavorServerless = "Serverless"
+)
+
 // API provides the top level programmatic interface to Pilosa. It is usually
 // wrapped by a handler which provides an external interface (e.g. HTTP).
 type API struct {

dax/queryer/queryer.go

@@ -189,9 +189,6 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s
 	// put the requestId in the context
 	ctx = fbcontext.WithRequestID(ctx, requestID.String())
 
-	userID := "travis"
-	ctx = fbcontext.WithUserID(ctx, userID)
-
 	st, err := parser.NewParser(multiReader).ParseStatement()
 	if err != nil {
 		applyError(errors.Wrap(err, "parsing sql"))
@@ -204,12 +201,8 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s
 	// Importer
 	imp := idkserverless.NewImporter(q.controller, qdbid, nil)
 
-	// TODO(tlt): We need a serverless-compatible implementation of the
 	// SystemAPI.
-	sysapi := &featurebase.NopSystemAPI{}
-
-	// systemLayer.ExecutionRequests().AddRequest("reqid", "userid", time.Now(), "select foo")
-	// systemLayer.ExecutionRequests().
+	sysapi := newSystemAPI()
 
 	// We intentionally don't pass the sql argument here because we're working
 	// with an io.Reader rather than a string and it's just not necessary to

dax/queryer/system_api.go

@@ -0,0 +1,18 @@
+package queryer
+
+import (
+	featurebase "github.com/featurebasedb/featurebase/v3"
+)
+
+// systemAPI is a no-op implementation of the systemAPI.
+type systemAPI struct {
+	featurebase.NopSystemAPI
+}
+
+func newSystemAPI() *systemAPI {
+	return &systemAPI{}
+}
+
+func (s *systemAPI) ClusterName() string {
+	return featurebase.FlavorServerless
+}

sql3/planner/compileselect.go

@@ -411,14 +411,21 @@ func (p *ExecutionPlanner) compileSource(scope *PlanOpQuery, source parser.Sourc
 			var op types.PlanOperator
 			op = NewPlanOpSystemTable(p, st)
 			if st.requiresFanout {
-				op = NewPlanOpFanout(p, op)
+				// We don't want to do fanout in serverless. There may be a
+				// better place to signify "serverless" vs "classic", but for
+				// now it's gonna be in the ClusterName since the concept of
+				// "cluster" doesn't really apply to serverless anyway.
+				switch p.systemAPI.ClusterName() {
+				case pilosa.FlavorServerless:
+				default:
+					op = NewPlanOpFanout(p, op)
+				}
 			}
 			if sourceExpr.Alias != nil {
 				aliasName := parser.IdentName(sourceExpr.Alias)
 				return NewPlanOpRelAlias(aliasName, op), nil
 			}
 			return op, nil
-
 		}
 		// get all the columns for this table - we will eliminate unused ones
 		// later on in the optimizer

sql3/planner/opquery.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
 	"time"
 
 	pilosa "github.com/featurebasedb/featurebase/v3"
@@ -123,7 +122,6 @@ func (i *queryIterator) Next(ctx context.Context) (types.Row, error) {
 		userId := ""
 		userId, _ = fbcontext.UserID(ctx)
 
-		log.Printf("DEEBUG: AddRequest: %s, %s, %s", requestId, userId, i.query.sql)
 		i.requests.AddRequest(requestId, userId, time.Now(), i.query.sql)
 		i.hasStarted = &struct{}{}
 	}

sql3/planner/opsystemtable.go

@@ -6,7 +6,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"log"
 	"sort"
 
 	pilosa "github.com/featurebasedb/featurebase/v3"
@@ -31,8 +30,19 @@ const (
 )
 
 type systemTable struct {
-	name           string
-	schema         types.Schema
+	name   string
+	schema types.Schema
+
+	// requiresFanout instructs the parser to "fan-out" the query to all cluster
+	// nodes in order to get results from every node. The reasoning is that the
+	// complete data set is distributed across all nodes rather than being
+	// available on a single nodde.
+	//
+	// TODO(tlt): I don't think this logic should be specified by the table.
+	// Rather the implementation of the ExecutionRequests interface should
+	// decide how it wants to retrieve the list of requests. Basically,
+	// "fan-out" should happen as part of the interface implementation; the
+	// fan-out itself shouldn't be calling the interface method.
 	requiresFanout bool
 }
 
@@ -451,7 +461,6 @@ type fbExecRequestsRowIter struct {
 var _ types.RowIterator = (*fbExecRequestsRowIter)(nil)
 
 func (i *fbExecRequestsRowIter) Next(ctx context.Context) (types.Row, error) {
-	log.Printf("DEEBUG: fbExecRequestsRowIter.Next() ")
 	if i.result == nil {
 		var err error
 		i.result, err = i.planner.systemLayerAPI.ExecutionRequests().ListRequests()

systemlayer/executionrequests.go

@@ -2,7 +2,6 @@ package systemlayer
 
 import (
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
@@ -123,7 +122,6 @@ func (e *ExecutionRequests) UpdateRequest(requestID string,
 
 // ListRequests returns the content of the ExecutionRequests struct as copies
 func (e *ExecutionRequests) ListRequests() ([]pilosa.ExecutionRequest, error) {
-	log.Printf("DEEBUG: ListRequests() called, %p", e)
 	e.RLock()
 	defer e.RUnlock()