From 92f64f42e9fb331373dcc18afd6f634016bb5167 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Thu, 9 Mar 2023 18:06:26 -0600 Subject: [PATCH] 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 | 7 +++++++ dax/queryer/queryer.go | 9 +-------- dax/queryer/system_api.go | 18 ++++++++++++++++++ sql3/planner/compileselect.go | 11 +++++++++-- sql3/planner/opquery.go | 2 -- sql3/planner/opsystemtable.go | 17 +++++++++++++---- systemlayer/executionrequests.go | 2 -- 7 files changed, 48 insertions(+), 18 deletions(-) create mode 100644 dax/queryer/system_api.go diff --git a/api.go b/api.go index 74dd32532..46dfb07da 100644 --- a/api.go +++ b/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 { diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 58352b696..3d6325469 100644 --- a/dax/queryer/queryer.go +++ b/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 diff --git a/dax/queryer/system_api.go b/dax/queryer/system_api.go new file mode 100644 index 000000000..c31ccf29f --- /dev/null +++ b/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 +} diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index 168f9401c..d5088e57c 100644 --- a/sql3/planner/compileselect.go +++ b/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 diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index 1ef857a28..a38b54cf2 100644 --- a/sql3/planner/opquery.go +++ b/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{}{} } diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go index 58123d76b..0905a4ae9 100644 --- a/sql3/planner/opsystemtable.go +++ b/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() diff --git a/systemlayer/executionrequests.go b/systemlayer/executionrequests.go index 411e72280..29b5ff7b4 100644 --- a/systemlayer/executionrequests.go +++ b/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()