From dca0dd84e3c1734930ca82ee60481e683a5e7049 Mon Sep 17 00:00:00 2001 From: pokeeffe-molecula <85502298+pokeeffe-molecula@users.noreply.github.com> Date: Fri, 2 Dec 2022 16:42:45 -0600 Subject: [PATCH] implement fb_exec_requests system table (#2327) implements an fb_exec_requests system table. The purpose of this table is to allow access to internal state to see what queries are running and have been run. Co-authored-by: Travis Turner (cherry picked from commit 47d8be26f5293a685f9039c0d5b7ac4bb4b8da95) --- api.go | 7 +- context.go => context/context.go | 18 ++- dax/queryer/queryer.go | 16 +- http_handler.go | 37 +++-- internal_client.go | 9 +- internal_client_test.go | 17 ++- server.go | 2 + server/server.go | 5 +- sql3/planner/compilealtertable.go | 6 +- sql3/planner/compilecreatetable.go | 2 +- sql3/planner/compiledroptable.go | 2 +- sql3/planner/compileselect.go | 2 +- sql3/planner/compileshow.go | 4 +- sql3/planner/executionplanner.go | 36 ++--- sql3/planner/executionplannersystemtables.go | 2 + sql3/planner/opquery.go | 68 ++++++++- sql3/planner/opsystemtable.go | 150 ++++++++++++++++++- sql3/planner/optop.go | 1 + sql3/sql_complex_test.go | 49 +++++- sql3/test/helpers.go | 8 +- systemlayer.go | 99 ++++++++++++ systemlayer/executionrequests.go | 108 +++++++++++++ systemlayer/systemlayer.go | 20 +++ 23 files changed, 595 insertions(+), 73 deletions(-) rename context.go => context/context.go (57%) create mode 100644 systemlayer.go create mode 100644 systemlayer/executionrequests.go create mode 100644 systemlayer/systemlayer.go diff --git a/api.go b/api.go index 847914b14..afcedc9e2 100644 --- a/api.go +++ b/api.go @@ -22,6 +22,7 @@ import ( "sync" "time" + fbcontext "github.com/molecula/featurebase/v3/context" "github.com/molecula/featurebase/v3/dax" "github.com/molecula/featurebase/v3/dax/computer" "github.com/molecula/featurebase/v3/disco" @@ -269,7 +270,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index // get the requestUserID from the context -- assumes the http handler has populated this from // authN/Z info - requestUserID, _ := UserIDFromContext(ctx) // requestUserID is "" if not in ctx + requestUserID, _ := fbcontext.UserID(ctx) // requestUserID is "" if not in ctx if err := api.validate(apiCreateIndex); err != nil { return nil, errors.Wrap(err, "validating api method") @@ -383,7 +384,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str // get the requestUserID from the context -- assumes the http handler has populated this from // authN/Z info - requestUserID, _ := UserIDFromContext(ctx) // requestUserID is "" if not in ctx + requestUserID, _ := fbcontext.UserID(ctx) // requestUserID is "" if not in ctx // Apply and validate functional options. fo, err := newFieldOptions(opts...) @@ -439,7 +440,7 @@ func (api *API) UpdateField(ctx context.Context, indexName, fieldName string, up // get the requestUserID from the context -- assumes the http handler has populated this from // authN/Z info - requestUserID, _ := UserIDFromContext(ctx) + requestUserID, _ := fbcontext.UserID(ctx) cfm, err := index.UpdateField(ctx, fieldName, requestUserID, update) if err != nil { diff --git a/context.go b/context/context.go similarity index 57% rename from context.go rename to context/context.go index e016366ba..c6812db1d 100644 --- a/context.go +++ b/context/context.go @@ -1,14 +1,15 @@ // Copyright 2022 Molecula Corp (DBA FeatureBase). All rights reserved. -package pilosa +package context import "context" // Empty struct to avoid allocations type contextKeyOriginalIP struct{} type contextKeyRequestUserID struct{} +type contextKeyRequestRequestID struct{} -// OriginalIPFromContext gets the original IP from the context. -func OriginalIPFromContext(ctx context.Context) (originalIP string, ok bool) { +// OriginalIP gets the original IP from the context. +func OriginalIP(ctx context.Context) (originalIP string, ok bool) { originalIP, ok = ctx.Value(contextKeyOriginalIP{}).(string) return } @@ -18,7 +19,7 @@ func WithOriginalIP(ctx context.Context, originalIP string) context.Context { return context.WithValue(ctx, contextKeyOriginalIP{}, originalIP) } -func UserIDFromContext(ctx context.Context) (userID string, ok bool) { +func UserID(ctx context.Context) (userID string, ok bool) { userID, ok = ctx.Value(contextKeyRequestUserID{}).(string) return } @@ -26,3 +27,12 @@ func UserIDFromContext(ctx context.Context) (userID string, ok bool) { func WithUserID(ctx context.Context, userID string) context.Context { return context.WithValue(ctx, contextKeyRequestUserID{}, userID) } + +func RequestID(ctx context.Context) (userID string, ok bool) { + userID, ok = ctx.Value(contextKeyRequestRequestID{}).(string) + return +} + +func WithRequestID(ctx context.Context, userID string) context.Context { + return context.WithValue(ctx, contextKeyRequestRequestID{}, userID) +} diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 3cd06bfe2..88220f548 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -9,6 +9,7 @@ import ( "time" featurebase "github.com/molecula/featurebase/v3" + fbcontext "github.com/molecula/featurebase/v3/context" "github.com/molecula/featurebase/v3/dax" "github.com/molecula/featurebase/v3/encoding/proto" "github.com/molecula/featurebase/v3/errors" @@ -21,6 +22,8 @@ import ( "github.com/molecula/featurebase/v3/sql3/planner" plannertypes "github.com/molecula/featurebase/v3/sql3/planner/types" "github.com/molecula/featurebase/v3/stats" + "github.com/molecula/featurebase/v3/systemlayer" + uuid "github.com/satori/go.uuid" ) // Queryer represents the query layer in a Molecula implementation. The idea is @@ -93,6 +96,15 @@ func (q *Queryer) QuerySQL(ctx context.Context, qual dax.TableQualifier, sql str applyExecutionTime() } + // Create a requestID and add it to the context. + requestID, err := uuid.NewV4() + if err != nil { + applyError(errors.Wrap(err, "creating requestID")) + return ret, nil + } + // put the requestId in the context + ctx = fbcontext.WithRequestID(ctx, requestID.String()) + st, err := parser.NewParser(strings.NewReader(sql)).ParseStatement() if err != nil { applyError(errors.Wrap(err, "parsing sql")) @@ -116,7 +128,9 @@ func (q *Queryer) QuerySQL(ctx context.Context, qual dax.TableQualifier, sql str // no-op implementation). sysapi := &featurebase.FeatureBaseSystemAPI{API: nil} - pl := planner.NewExecutionPlanner(orch, sapi, sysapi, capi, imp, q.orchestrator.logger, sql) + systemLayer := systemlayer.NewSystemLayer() + + pl := planner.NewExecutionPlanner(orch, sapi, sysapi, capi, systemLayer, imp, q.orchestrator.logger, sql) planOp, err := pl.CompilePlan(ctx, st) if err != nil { diff --git a/http_handler.go b/http_handler.go index eb232d48d..3b58f5fc4 100644 --- a/http_handler.go +++ b/http_handler.go @@ -35,6 +35,7 @@ import ( "github.com/gorilla/mux" "github.com/molecula/featurebase/v3/authn" "github.com/molecula/featurebase/v3/authz" + fbcontext "github.com/molecula/featurebase/v3/context" "github.com/molecula/featurebase/v3/dax" "github.com/molecula/featurebase/v3/disco" "github.com/molecula/featurebase/v3/logger" @@ -48,6 +49,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" dto "github.com/prometheus/client_model/go" "github.com/prometheus/prom2json" + uuid "github.com/satori/go.uuid" "github.com/zeebo/blake3" ) @@ -699,7 +701,7 @@ func (h *Handler) chkAllowedNetworks(r *http.Request) (bool, context.Context) { // if client IP is in allowed networks // add it to the context for key X-Molecula-Original-IP if h.auth.CheckAllowedNetworks(reqIP) { - ctx := WithOriginalIP(r.Context(), reqIP) + ctx := fbcontext.WithOriginalIP(r.Context(), reqIP) return true, ctx } return false, r.Context() @@ -711,7 +713,7 @@ func (h *Handler) chkAuthN(handler http.HandlerFunc) http.HandlerFunc { // if the request is unauthenticated and we have the appropriate header get the userid from the header requestUserID := r.Header.Get(HeaderRequestUserID) - ctx = WithUserID(ctx, requestUserID) + ctx = fbcontext.WithUserID(ctx, requestUserID) if h.auth == nil { handler.ServeHTTP(w, r.WithContext(ctx)) @@ -733,7 +735,7 @@ func (h *Handler) chkAuthN(handler http.HandlerFunc) http.HandlerFunc { } // prefer the user id from an authenticated request over one in a header - ctx = WithUserID(ctx, uinfo.UserID) + ctx = fbcontext.WithUserID(ctx, uinfo.UserID) // just in case it got refreshed ctx = authn.WithAccessToken(ctx, "Bearer"+access) @@ -750,7 +752,7 @@ func (h *Handler) chkAuthZ(handler http.HandlerFunc, perm authz.Permission) http // if the request is unauthenticated and we have the appropriate header get the userid from the header requestUserID := r.Header.Get(HeaderRequestUserID) - ctx = WithUserID(ctx, requestUserID) + ctx = fbcontext.WithUserID(ctx, requestUserID) // handle the case when auth is not turned on if h.auth == nil { @@ -779,7 +781,7 @@ func (h *Handler) chkAuthZ(handler http.HandlerFunc, perm authz.Permission) http } // prefer the user id from an authenticated request over one in a header - ctx = WithUserID(ctx, uinfo.UserID) + ctx = fbcontext.WithUserID(ctx, uinfo.UserID) ctx = authn.WithAccessToken(ctx, "Bearer "+access) ctx = authn.WithRefreshToken(ctx, refresh) @@ -877,7 +879,7 @@ func GetIP(r *http.Request) string { } // check if original IP is in the context - if ogIP, ok := OriginalIPFromContext(r.Context()); ok && ogIP != "" { + if ogIP, ok := fbcontext.OriginalIP(r.Context()); ok && ogIP != "" { return ogIP } @@ -1411,9 +1413,15 @@ func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) { return } - start := time.Now() + requestID, err := uuid.NewV4() + if err != nil { + h.writeBadRequest(w, r, err) + } + // put the requestId in the context + ctx := fbcontext.WithRequestID(r.Context(), requestID.String()) - rootOperator, err := h.api.CompilePlan(r.Context(), string(b)) + sql := string(b) + rootOperator, err := h.api.CompilePlan(ctx, sql) if err != nil { h.writeBadRequest(w, r, err) return @@ -1429,10 +1437,15 @@ func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) { // Write the closing bracket on any exit from this method. defer func() { - duration := time.Since(start) - value, err := json.Marshal(duration.Microseconds()) + var value []byte + request, err := h.api.server.SystemLayer.ExecutionRequests().GetRequest(requestID.String()) if err != nil { value = big.NewInt(-1).Bytes() + } else { + value, err = json.Marshal(request.ElapsedTime.Microseconds()) + if err != nil { + value = big.NewInt(-1).Bytes() + } } w.Write([]byte(`,"exec_time":`)) w.Write(value) @@ -1483,7 +1496,7 @@ func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) { } // Get a query iterator. - iter, err := rootOperator.Iterator(r.Context(), nil) + iter, err := rootOperator.Iterator(ctx, nil) if err != nil { writeError(err) writeWarnings(rootOperator.Warnings()) @@ -1529,7 +1542,7 @@ func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) { var nextErr error rowCounter := 1 - for currentRow, nextErr = iter.Next(r.Context()); nextErr == nil; currentRow, nextErr = iter.Next(r.Context()) { + for currentRow, nextErr = iter.Next(ctx); nextErr == nil; currentRow, nextErr = iter.Next(ctx) { jsonRow, err := json.Marshal(currentRow) if err != nil { h.logger.Errorf("json encoding error: %s", err) diff --git a/internal_client.go b/internal_client.go index f6a7378d4..fec3052cb 100644 --- a/internal_client.go +++ b/internal_client.go @@ -18,11 +18,8 @@ import ( "strings" "time" - "github.com/featurebasedb/featurebase/v3/authn" - "github.com/featurebasedb/featurebase/v3/disco" - "github.com/featurebasedb/featurebase/v3/logger" - pnet "github.com/featurebasedb/featurebase/v3/net" - "github.com/featurebasedb/featurebase/v3/tracing" + fbcontext "github.com/molecula/featurebase/v3/context" + "github.com/hashicorp/go-retryablehttp" "github.com/pkg/errors" "golang.org/x/oauth2" @@ -261,7 +258,7 @@ func AddAuthToken(ctx context.Context, header *http.Header) { } // set ogIP to request for remote calls - if ogIP, ok := OriginalIPFromContext(ctx); ok && ogIP != "" { + if ogIP, ok := fbcontext.OriginalIP(ctx); ok && ogIP != "" { header.Set(OriginalIPHeader, ogIP) } diff --git a/internal_client_test.go b/internal_client_test.go index 45b13cec9..e0f94faee 100644 --- a/internal_client_test.go +++ b/internal_client_test.go @@ -15,13 +15,14 @@ import ( "time" "github.com/davecgh/go-spew/spew" - pilosa "github.com/featurebasedb/featurebase/v3" - "github.com/featurebasedb/featurebase/v3/authn" - "github.com/featurebasedb/featurebase/v3/disco" - "github.com/featurebasedb/featurebase/v3/encoding/proto" - "github.com/featurebasedb/featurebase/v3/pql" - "github.com/featurebasedb/featurebase/v3/test" - "github.com/featurebasedb/featurebase/v3/vprint" + pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/authn" + fbcontext "github.com/molecula/featurebase/v3/context" + "github.com/molecula/featurebase/v3/disco" + "github.com/molecula/featurebase/v3/encoding/proto" + "github.com/molecula/featurebase/v3/pql" + "github.com/molecula/featurebase/v3/test" + "github.com/molecula/featurebase/v3/vprint" "github.com/pkg/errors" "github.com/ricochet2200/go-disk-usage/du" ) @@ -1633,7 +1634,7 @@ func TestAddAuthToken(t *testing.T) { t.Fatalf("unexpected error: %v", err) } ogIP := "10.0.0.1" - pilosa.AddAuthToken(pilosa.WithOriginalIP(context.Background(), ogIP), &req.Header) + pilosa.AddAuthToken(fbcontext.WithOriginalIP(context.Background(), ogIP), &req.Header) if got := req.Header.Get(pilosa.OriginalIPHeader); got != ogIP { t.Fatalf("got '%v', expected '%v'", got, ogIP) } diff --git a/server.go b/server.go index 149e5784b..177e0b764 100644 --- a/server.go +++ b/server.go @@ -58,6 +58,8 @@ type Server struct { // nolint: maligned executorPoolSize int serializer Serializer + SystemLayer SystemLayerAPI + // Distributed Consensus disCo disco.DisCo noder disco.Noder diff --git a/server/server.go b/server/server.go index bc2997419..3146bfc33 100644 --- a/server/server.go +++ b/server/server.go @@ -27,6 +27,7 @@ import ( "time" "github.com/molecula/featurebase/v3/dax" + "github.com/molecula/featurebase/v3/systemlayer" "golang.org/x/sync/errgroup" pilosa "github.com/molecula/featurebase/v3" @@ -576,7 +577,7 @@ func (m *Command) setupServer() error { fsapi := &pilosa.FeatureBaseSystemAPI{API: api} fimp := &batch.FeaturebaseImporter{API: api} - return planner.NewExecutionPlanner(e, fapi, fsapi, api, fimp, m.logger, sql) + return planner.NewExecutionPlanner(e, fapi, fsapi, api, m.Server.SystemLayer, fimp, m.logger, sql) } serverOptions := []pilosa.ServerOption{ @@ -629,6 +630,8 @@ func (m *Command) setupServer() error { return errors.Wrap(err, "new server") } + m.Server.SystemLayer = systemlayer.NewSystemLayer() + m.API, err = pilosa.NewAPI( pilosa.OptAPIServer(m.Server), pilosa.OptAPIImportWorkerPoolSize(m.Config.ImportWorkerPoolSize), diff --git a/sql3/planner/compilealtertable.go b/sql3/planner/compilealtertable.go index a8ce0af7b..31f2f2a58 100644 --- a/sql3/planner/compilealtertable.go +++ b/sql3/planner/compilealtertable.go @@ -50,7 +50,7 @@ func (p *ExecutionPlanner) compileAlterTableStatement(stmt *parser.AlterTableSta return nil, sql3.NewErrColumnNotFound(stmt.DropColumnName.NamePos.Line, stmt.DropColumnName.NamePos.Column, columnName) } - return NewPlanOpQuery(NewPlanOpAlterTable(p, tableName, alterOpDrop, columnName, "", nil), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpAlterTable(p, tableName, alterOpDrop, columnName, "", nil), p.sql), nil } else if stmt.Add.IsValid() { col := stmt.ColumnDef columnName := parser.IdentName(col.Name) @@ -66,12 +66,12 @@ func (p *ExecutionPlanner) compileAlterTableStatement(stmt *parser.AlterTableSta if err != nil { return nil, err } - return NewPlanOpQuery(NewPlanOpAlterTable(p, tableName, alterOpAdd, "", columnName, column), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpAlterTable(p, tableName, alterOpAdd, "", columnName, column), p.sql), nil } else if stmt.Rename.IsValid() { oldColumnName := parser.IdentName(stmt.OldColumnName) newColumnName := parser.IdentName(stmt.NewColumnName) - return NewPlanOpQuery(NewPlanOpAlterTable(p, tableName, alterOpRename, oldColumnName, newColumnName, nil), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpAlterTable(p, tableName, alterOpRename, oldColumnName, newColumnName, nil), p.sql), nil } else { return nil, sql3.NewErrInternal("unhandled alter operation") } diff --git a/sql3/planner/compilecreatetable.go b/sql3/planner/compilecreatetable.go index 75c057d73..1581ac8d3 100644 --- a/sql3/planner/compilecreatetable.go +++ b/sql3/planner/compilecreatetable.go @@ -62,7 +62,7 @@ func (p *ExecutionPlanner) compileCreateTableStatement(stmt *parser.CreateTableS columns = append(columns, column) } - return NewPlanOpQuery(NewPlanOpCreateTable(p, tableName, failIfExists, isKeyed, keyPartitions, columns), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpCreateTable(p, tableName, failIfExists, isKeyed, keyPartitions, columns), p.sql), nil } // compiles a column def diff --git a/sql3/planner/compiledroptable.go b/sql3/planner/compiledroptable.go index 6bf0e81ad..7684edd59 100644 --- a/sql3/planner/compiledroptable.go +++ b/sql3/planner/compiledroptable.go @@ -23,5 +23,5 @@ func (p *ExecutionPlanner) compileDropTableStatement(stmt *parser.DropTableState } return nil, err } - return NewPlanOpQuery(NewPlanOpDropTable(p, index), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpDropTable(p, index), p.sql), nil } diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index bd3f9365f..e6d27f848 100644 --- a/sql3/planner/compileselect.go +++ b/sql3/planner/compileselect.go @@ -15,7 +15,7 @@ import ( // compileSelectStatment compiles a parser.SelectStatment AST into a PlanOperator func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement, isSubquery bool) (types.PlanOperator, error) { - query := NewPlanOpQuery(NewPlanOpNullTable(), p.sql) + query := NewPlanOpQuery(p, NewPlanOpNullTable(), p.sql) p.scopeStack.push(query) // handle projections diff --git a/sql3/planner/compileshow.go b/sql3/planner/compileshow.go index 7d0243355..3d671ca01 100644 --- a/sql3/planner/compileshow.go +++ b/sql3/planner/compileshow.go @@ -45,7 +45,7 @@ func (p *ExecutionPlanner) compileShowTablesStatement(stmt parser.Statement) (ty dataType: parser.NewDataTypeInt(), }} - return NewPlanOpQuery(NewPlanOpProjection(columns, NewPlanOpFeatureBaseTables(indexInfo)), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseTables(indexInfo)), p.sql), nil } func (p *ExecutionPlanner) compileShowColumnsStatement(stmt *parser.ShowColumnsStatement) (_ types.PlanOperator, err error) { @@ -130,5 +130,5 @@ func (p *ExecutionPlanner) compileShowColumnsStatement(stmt *parser.ShowColumnsS dataType: parser.NewDataTypeString(), }} - return NewPlanOpQuery(NewPlanOpProjection(columns, NewPlanOpFeatureBaseColumns(index)), p.sql), nil + return NewPlanOpQuery(p, NewPlanOpProjection(columns, NewPlanOpFeatureBaseColumns(index)), p.sql), nil } diff --git a/sql3/planner/executionplanner.go b/sql3/planner/executionplanner.go index 5b4781602..1e1e74b95 100644 --- a/sql3/planner/executionplanner.go +++ b/sql3/planner/executionplanner.go @@ -23,26 +23,28 @@ type PlannerScope struct { // ExecutionPlanner compiles SQL text into a query plan type ExecutionPlanner struct { - executor pilosa.Executor - schemaAPI pilosa.SchemaAPI - systemAPI pilosa.SystemAPI - computeAPI pilosa.ComputeAPI - importer batch.Importer - logger logger.Logger - sql string - scopeStack *scopeStack + executor pilosa.Executor + schemaAPI pilosa.SchemaAPI + systemAPI pilosa.SystemAPI + computeAPI pilosa.ComputeAPI + systemLayerAPI pilosa.SystemLayerAPI + importer batch.Importer + logger logger.Logger + sql string + scopeStack *scopeStack } -func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, systemAPI pilosa.SystemAPI, computeAPI pilosa.ComputeAPI, importer batch.Importer, logger logger.Logger, sql string) *ExecutionPlanner { +func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, systemAPI pilosa.SystemAPI, computeAPI pilosa.ComputeAPI, systemLayerAPI pilosa.SystemLayerAPI, importer batch.Importer, logger logger.Logger, sql string) *ExecutionPlanner { return &ExecutionPlanner{ - executor: executor, - schemaAPI: newSystemTableDefintionsWrapper(schemaAPI), - systemAPI: systemAPI, - computeAPI: computeAPI, - importer: importer, - logger: logger, - sql: sql, - scopeStack: newScopeStack(), + executor: executor, + schemaAPI: newSystemTableDefintionsWrapper(schemaAPI), + systemAPI: systemAPI, + computeAPI: computeAPI, + systemLayerAPI: systemLayerAPI, + importer: importer, + logger: logger, + sql: sql, + scopeStack: newScopeStack(), } } diff --git a/sql3/planner/executionplannersystemtables.go b/sql3/planner/executionplannersystemtables.go index 85623c91c..d9b7f3e98 100644 --- a/sql3/planner/executionplannersystemtables.go +++ b/sql3/planner/executionplannersystemtables.go @@ -91,6 +91,8 @@ func indexInfoFromSystemTable(st *systemTable) (*pilosa.IndexInfo, error) { case *parser.DataTypeString: opts.Type = pilosa.FieldTypeMutex opts.Keys = true + case *parser.DataTypeTimestamp: + opts.Type = pilosa.FieldTypeTimestamp default: return nil, sql3.NewErrInternalf("unexpected system table field type '%T'", f.Type) } diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index 112f1189b..4a07fdf31 100644 --- a/sql3/planner/opquery.go +++ b/sql3/planner/opquery.go @@ -2,14 +2,20 @@ package planner import ( "context" + "encoding/json" "fmt" + "time" - "github.com/featurebasedb/featurebase/v3/sql3" - "github.com/featurebasedb/featurebase/v3/sql3/planner/types" + pilosa "github.com/molecula/featurebase/v3" + fbcontext "github.com/molecula/featurebase/v3/context" + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/planner/types" ) // PlanOpQuery is a query - this is the root node of an execution plan type PlanOpQuery struct { + planner *ExecutionPlanner + ChildOp types.PlanOperator // the list of aggregate terms @@ -24,8 +30,9 @@ type PlanOpQuery struct { var _ types.PlanOperator = (*PlanOpQuery)(nil) -func NewPlanOpQuery(child types.PlanOperator, sql string) *PlanOpQuery { +func NewPlanOpQuery(p *ExecutionPlanner, child types.PlanOperator, sql string) *PlanOpQuery { return &PlanOpQuery{ + planner: p, ChildOp: child, warnings: make([]string, 0), sql: sql, @@ -41,7 +48,12 @@ func (p *PlanOpQuery) Child() types.PlanOperator { } func (p *PlanOpQuery) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { - return p.Child().Iterator(ctx, row) + iter, err := p.ChildOp.Iterator(ctx, row) + if err != nil { + return nil, err + } + + return newQueryIterator(p.planner.systemLayerAPI.ExecutionRequests(), p, iter), nil } func (p *PlanOpQuery) Children() []types.PlanOperator { @@ -54,7 +66,7 @@ func (p *PlanOpQuery) WithChildren(children ...types.PlanOperator) (types.PlanOp if len(children) != 1 { return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children)) } - op := NewPlanOpQuery(children[0], p.sql) + op := NewPlanOpQuery(p.planner, children[0], p.sql) op.warnings = append(op.warnings, p.warnings...) return op, nil @@ -91,3 +103,49 @@ func (p *PlanOpQuery) Warnings() []string { func (p *PlanOpQuery) String() string { return "" } + +type queryIterator struct { + requests pilosa.ExecutionRequestsAPI + query *PlanOpQuery + + child types.RowIterator + + hasStarted *struct{} +} + +func newQueryIterator(requests pilosa.ExecutionRequestsAPI, query *PlanOpQuery, child types.RowIterator) *queryIterator { + return &queryIterator{ + requests: requests, + query: query, + child: child, + } +} + +func (i *queryIterator) Next(ctx context.Context) (types.Row, error) { + if i.hasStarted == nil { + i.hasStarted = &struct{}{} + + requestId, ok := fbcontext.RequestID(ctx) + if !ok { + return nil, sql3.NewErrInternalf("unable to get request id from context") + } + + userId := "" + userId, _ = fbcontext.UserID(ctx) + + i.requests.AddRequest(requestId, userId, time.Now(), i.query.sql) + } + + row, err := i.child.Next(ctx) + if err != nil { + // either error or no more rows, either way update the request + requestId, ok := fbcontext.RequestID(ctx) + if !ok { + return nil, sql3.NewErrInternalf("unable to get request id from context") + } + + plan, _ := json.MarshalIndent(i.query.Plan(), "", " ") + i.requests.UpdateRequest(requestId, time.Now(), "complete", "", 0, "", 0, 0, 0, 0, 0, string(plan)) + } + return row, err +} diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go index 78daa66a8..12d23d090 100644 --- a/sql3/planner/opsystemtable.go +++ b/sql3/planner/opsystemtable.go @@ -34,6 +34,7 @@ import ( const ( fbClusterInfo = "fb_cluster_info" fbClusterNodes = "fb_cluster_nodes" + fbExecRequests = "fb_exec_requests" ) type systemTable struct { @@ -50,6 +51,11 @@ var systemTables = map[string]*systemTable{ ColumnName: "name", Type: parser.NewDataTypeString(), }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "name", + Type: parser.NewDataTypeString(), + }, &types.PlannerColumn{ RelationName: fbClusterInfo, ColumnName: "platform", @@ -91,32 +97,118 @@ var systemTables = map[string]*systemTable{ name: fbClusterNodes, schema: types.Schema{ &types.PlannerColumn{ - RelationName: fbClusterInfo, + RelationName: fbClusterNodes, ColumnName: "id", Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - RelationName: fbClusterInfo, + RelationName: fbClusterNodes, ColumnName: "state", Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - RelationName: fbClusterInfo, + RelationName: fbClusterNodes, ColumnName: "uri", Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - RelationName: fbClusterInfo, + RelationName: fbClusterNodes, ColumnName: "grpc_uri", Type: parser.NewDataTypeString(), }, &types.PlannerColumn{ - RelationName: fbClusterInfo, + RelationName: fbClusterNodes, ColumnName: "is_primary", Type: parser.NewDataTypeBool(), }, }, }, + + fbExecRequests: { + name: fbExecRequests, + schema: types.Schema{ + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "request_id", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "user", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "start_time", + Type: parser.NewDataTypeTimestamp(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "end_time", + Type: parser.NewDataTypeTimestamp(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "status", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "wait_type", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "wait_time", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "wait_resource", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "cpu_time", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "elapsed_time", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "reads", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "writes", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "logical_reads", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "row_count", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "sql", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbExecRequests, + ColumnName: "plan", + Type: parser.NewDataTypeString(), + }, + }, + }, } // PlanOpSystemTable handles system tables @@ -175,6 +267,10 @@ func (p *PlanOpSystemTable) Iterator(ctx context.Context, row types.Row) (types. return &fbClusterNodesRowIter{ planner: p.planner, }, nil + case fbExecRequests: + return &fbExecRequestsRowIter{ + planner: p.planner, + }, nil default: return nil, sql3.NewErrInternalf("unable to find system table '%s'", p.table.name) } @@ -197,6 +293,7 @@ var _ types.RowIterator = (*fbClusterInfoRowIter)(nil) func (i *fbClusterInfoRowIter) Next(ctx context.Context) (types.Row, error) { if i.rowIndex < 1 { row := []interface{}{ + i.planner.systemAPI.ClusterName(), i.planner.systemAPI.ClusterName(), i.planner.systemAPI.PlatformDescription(), i.planner.systemAPI.PlatformVersion(), @@ -239,3 +336,46 @@ func (i *fbClusterNodesRowIter) Next(ctx context.Context) (types.Row, error) { } return nil, types.ErrNoMoreRows } + +type fbExecRequestsRowIter struct { + planner *ExecutionPlanner + result []pilosa.ExecutionRequest +} + +var _ types.RowIterator = (*fbExecRequestsRowIter)(nil) + +func (i *fbExecRequestsRowIter) Next(ctx context.Context) (types.Row, error) { + if i.result == nil { + var err error + i.result, err = i.planner.systemLayerAPI.ExecutionRequests().ListRequests() + if err != nil { + return nil, err + } + } + + if len(i.result) > 0 { + n := i.result[0] + row := []interface{}{ + n.RequestID, + n.UserID, + n.StartTime, + n.EndTime, + n.Status, + n.WaitType, + n.WaitTime.Microseconds(), + n.WaitResource, + n.CPUTime.Microseconds(), + n.ElapsedTime.Microseconds(), + n.Reads, + n.Writes, + n.LogicalReads, + n.RowCount, + n.SQL, + n.Plan, + } + // Move to next result element. + i.result = i.result[1:] + return row, nil + } + return nil, types.ErrNoMoreRows +} diff --git a/sql3/planner/optop.go b/sql3/planner/optop.go index b2fe539fe..3c5fcbd38 100644 --- a/sql3/planner/optop.go +++ b/sql3/planner/optop.go @@ -30,6 +30,7 @@ func (p *PlanOpTop) Schema() types.Schema { } func (p *PlanOpTop) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + // TODO (pok) actually implement top return p.ChildOp.Iterator(ctx, row) } diff --git a/sql3/sql_complex_test.go b/sql3/sql_complex_test.go index 61ddad3c0..7fea27c57 100644 --- a/sql3/sql_complex_test.go +++ b/sql3/sql_complex_test.go @@ -68,7 +68,7 @@ func TestPlanner_Show(t *testing.T) { t.Fatal(err) } - t.Run("SystemTables", func(t *testing.T) { + t.Run("SystemTablesInfo", func(t *testing.T) { results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `select name, platform, platform_version, db_version, state, node_count, shard_width, replica_count from fb_cluster_info`) if err != nil { t.Fatal(err) @@ -91,12 +91,57 @@ func TestPlanner_Show(t *testing.T) { } }) + t.Run("SystemTablesNode", func(t *testing.T) { + _, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `select * from fb_cluster_nodes`) + if err != nil { + t.Fatal(err) + } + + if diff := cmp.Diff([]*pilosa.WireQueryField{ + wireQueryFieldString("id"), + wireQueryFieldString("state"), + wireQueryFieldString("uri"), + wireQueryFieldString("grpc_uri"), + wireQueryFieldBool("is_primary"), + }, columns); diff != "" { + t.Fatal(diff) + } + }) + + t.Run("SystemTablesExecRequests", func(t *testing.T) { + _, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `select * from fb_exec_requests`) + if err != nil { + t.Fatal(err) + } + + if diff := cmp.Diff([]*pilosa.WireQueryField{ + wireQueryFieldString("request_id"), + wireQueryFieldString("user"), + wireQueryFieldTimestamp("start_time"), + wireQueryFieldTimestamp("end_time"), + wireQueryFieldString("status"), + wireQueryFieldString("wait_type"), + wireQueryFieldInt("wait_time"), + wireQueryFieldString("wait_resource"), + wireQueryFieldInt("cpu_time"), + wireQueryFieldInt("elapsed_time"), + wireQueryFieldInt("reads"), + wireQueryFieldInt("writes"), + wireQueryFieldInt("logical_reads"), + wireQueryFieldInt("row_count"), + wireQueryFieldString("sql"), + wireQueryFieldString("plan"), + }, columns); diff != "" { + t.Fatal(diff) + } + }) + t.Run("ShowTables", func(t *testing.T) { results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `SHOW TABLES`) if err != nil { t.Fatal(err) } - if len(results) != 4 { + if len(results) != 5 { t.Fatal(fmt.Errorf("unexpected result set length")) } diff --git a/sql3/test/helpers.go b/sql3/test/helpers.go index a49a12411..20335ed94 100644 --- a/sql3/test/helpers.go +++ b/sql3/test/helpers.go @@ -7,15 +7,21 @@ import ( "testing" featurebase "github.com/molecula/featurebase/v3" + fbcontext "github.com/molecula/featurebase/v3/context" "github.com/molecula/featurebase/v3/dax" plannertypes "github.com/molecula/featurebase/v3/sql3/planner/types" + uuid "github.com/satori/go.uuid" ) // MustQueryRows returns the row results as a slice of []interface{}, along with the columns. func MustQueryRows(tb testing.TB, svr *featurebase.Server, q string) ([][]interface{}, []*featurebase.WireQueryField, error) { tb.Helper() + requestId, err := uuid.NewV4() + if err != nil { + return nil, nil, err + } - ctx := context.Background() + ctx := fbcontext.WithRequestID(context.Background(), requestId.String()) stmt, err := svr.CompileExecutionPlan(ctx, q) if err != nil { diff --git a/systemlayer.go b/systemlayer.go new file mode 100644 index 000000000..d42729f82 --- /dev/null +++ b/systemlayer.go @@ -0,0 +1,99 @@ +package pilosa + +import ( + "strings" + "time" +) + +// ExecutionRequest holds data about an (sql) execution request +type ExecutionRequest struct { + // the id of the request + RequestID string + // the id of the user + UserID string + // time the request started + StartTime time.Time + // time the request finished - zero iif it has not finished + EndTime time.Time + // status of the request 'running' or 'complete' now, could have other values later + Status string + // future: if the request is waiting, the type of wait that is occuring + WaitType string + // future: the cumulative wait time for this request + WaitTime time.Duration + // futuure: if the request is waiting, the thing it is waiting on + WaitResource string + // future: the cululative cpu time for this request + CPUTime time.Duration + // the elapsed time for this request + ElapsedTime time.Duration + // future: the cumulative number of physical reads for this request + Reads int64 + // future: the cumulative number of physical writes for this request + Writes int64 + // future: the cumulative number of logical reads for this request + LogicalReads int64 + // future: the cumulative number of rows affected for this request + RowCount int64 + // the query plan for this request formatted in json + Plan string + // the sql for this request + SQL string +} + +// Copy returns a copy of the ExecutionRequest passed +func (e *ExecutionRequest) Copy() ExecutionRequest { + var elapsedTime time.Duration + if !strings.EqualFold(e.Status, "complete") { + elapsedTime = time.Since(e.StartTime) + } else { + elapsedTime = e.EndTime.Sub(e.StartTime) + } + + return ExecutionRequest{ + RequestID: e.RequestID, + UserID: e.UserID, + StartTime: e.StartTime, + EndTime: e.EndTime, + Status: e.Status, + WaitType: e.WaitType, + WaitTime: e.WaitTime, + WaitResource: e.WaitResource, + CPUTime: e.CPUTime, + ElapsedTime: elapsedTime, + SQL: e.SQL, + Plan: e.Plan, + } +} + +// ExecutionRequestsAPI defines the API for storing, updating and querying internal state +// around (sql) execution requests +type ExecutionRequestsAPI interface { + // add a request + AddRequest(requestID string, userID string, startTime time.Time, sql string) error + + // update a request + UpdateRequest(requestID string, + endTime time.Time, + status string, + waitType string, + waitTime time.Duration, + waitResource string, + cpuTime time.Duration, + reads int64, + writes int64, + logicalReads int64, + rowCount int64, + plan string) error + + // list all the requests + ListRequests() ([]ExecutionRequest, error) + + // get a specific request + GetRequest(requestID string) (ExecutionRequest, error) +} + +// SystemLayerAPI defines an api to allow access to internal FeatureBase state +type SystemLayerAPI interface { + ExecutionRequests() ExecutionRequestsAPI +} diff --git a/systemlayer/executionrequests.go b/systemlayer/executionrequests.go new file mode 100644 index 000000000..6e2dc37a4 --- /dev/null +++ b/systemlayer/executionrequests.go @@ -0,0 +1,108 @@ +package systemlayer + +import ( + "fmt" + "sync" + "time" + + pilosa "github.com/molecula/featurebase/v3" +) + +// ExecutionRequests is an internal struct that keeps a list of sql execution requests +// this data allows visbility into queries that have been run and are running +type ExecutionRequests struct { + sync.RWMutex + requests map[string]*pilosa.ExecutionRequest +} + +// Ensure type implements interface. +var _ pilosa.ExecutionRequestsAPI = (*ExecutionRequests)(nil) + +func NewExecutionRequestsAPI() *ExecutionRequests { + return &ExecutionRequests{ + requests: make(map[string]*pilosa.ExecutionRequest), + } +} + +// AddRequest adds a new request to the ExecutionRequests struct +// TODO(pok) ensure a cap on these so we don't suck up too much memory +func (e *ExecutionRequests) AddRequest(requestID string, userID string, startTime time.Time, sql string) error { + e.Lock() + defer e.Unlock() + + _, ok := e.requests[requestID] + if ok { + return fmt.Errorf("request %s already exists", requestID) + } + e.requests[requestID] = &pilosa.ExecutionRequest{ + RequestID: requestID, + UserID: userID, + StartTime: startTime, + Status: "running", + SQL: sql, + } + return nil +} + +// UpdateRequest updates the values for a request in the ExecutionRequests struct +func (e *ExecutionRequests) UpdateRequest(requestID string, + endTime time.Time, + status string, + waitType string, + waitTime time.Duration, + waitResource string, + cpuTime time.Duration, + reads int64, + writes int64, + logicalReads int64, + rowCount int64, + plan string) error { + + e.Lock() + defer e.Unlock() + + request, ok := e.requests[requestID] + if !ok { + return fmt.Errorf("request %s not found", requestID) + } + request.EndTime = endTime + request.Status = status + request.WaitType = waitType + request.WaitTime += waitTime + request.WaitResource = waitResource + request.CPUTime += cpuTime + request.Reads += reads + request.Writes += writes + request.LogicalReads += logicalReads + request.RowCount += rowCount + request.Plan = plan + + return nil +} + +// ListRequests returns the content of the ExecutionRequests struct as copies +func (e *ExecutionRequests) ListRequests() ([]pilosa.ExecutionRequest, error) { + e.RLock() + defer e.RUnlock() + + result := make([]pilosa.ExecutionRequest, len(e.requests)) + + idx := 0 + for _, er := range e.requests { + result[idx] = er.Copy() + idx++ + } + return result, nil +} + +func (e *ExecutionRequests) GetRequest(requestID string) (pilosa.ExecutionRequest, error) { + e.RLock() + defer e.RUnlock() + + er, ok := e.requests[requestID] + if !ok { + return pilosa.ExecutionRequest{}, fmt.Errorf("request %s not found", requestID) + } + + return er.Copy(), nil +} diff --git a/systemlayer/systemlayer.go b/systemlayer/systemlayer.go new file mode 100644 index 000000000..078816f6b --- /dev/null +++ b/systemlayer/systemlayer.go @@ -0,0 +1,20 @@ +package systemlayer + +import pilosa "github.com/molecula/featurebase/v3" + +// SystemLayer is a struct to hold internal FeatureBase state +// Initially this is just the execution requests, but later may include other +// internal state (Buffer Pool?) +type SystemLayer struct { + executionRequests pilosa.ExecutionRequestsAPI +} + +func NewSystemLayer() *SystemLayer { + return &SystemLayer{ + executionRequests: NewExecutionRequestsAPI(), + } +} + +func (e *SystemLayer) ExecutionRequests() pilosa.ExecutionRequestsAPI { + return e.executionRequests +}