scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

implement fb_exec_requests system table (#2327)

Browse source 
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 <travis@molecula.com>
This commit is contained in:
pokeeffe-molecula 2022-12-02 16:42:45 -06:00 • committed by GitHub
parent be619fa60d
commit 47d8be26f5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
23 changed files with 586 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
api.go
View file

@ -21,6 +21,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"
@ -268,7 +269,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")
@ -382,7 +383,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...)
@ -438,7 +439,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 {

18
context.go → context/context.go
View file

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

16
dax/queryer/queryer.go
View file

@ -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 {

37
http_handler.go
View file

@ -34,6 +34,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"
@ -47,6 +48,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"
)
@ -698,7 +700,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()
@ -710,7 +712,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))
@ -732,7 +734,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)
@ -749,7 +751,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 {
@ -778,7 +780,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)
@ -876,7 +878,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
}
@ -1410,9 +1412,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
@ -1428,10 +1436,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)
@ -1482,7 +1495,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())
@ -1528,7 +1541,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)

4
internal_client.go
View file

@ -17,6 +17,8 @@ import (
"strings"
"time"
fbcontext "github.com/molecula/featurebase/v3/context"
"github.com/hashicorp/go-retryablehttp"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
@ -260,7 +262,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)
}

3
internal_client_test.go
View file

@ -16,6 +16,7 @@ import (
"github.com/davecgh/go-spew/spew"
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"
@ -1632,7 +1633,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)
}

2
server.go
View file

@ -57,6 +57,8 @@ type Server struct { // nolint: maligned
executorPoolSize int
serializer Serializer
SystemLayer SystemLayerAPI
// Distributed Consensus
disCo disco.DisCo
noder disco.Noder

5
server/server.go
View file

@ -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),

6
sql3/planner/compilealtertable.go
View file

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

2
sql3/planner/compilecreatetable.go
View file

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

2
sql3/planner/compiledroptable.go
View file

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

2
sql3/planner/compileselect.go
View file

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

4
sql3/planner/compileshow.go
View file

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

36
sql3/planner/executionplanner.go
View file

@ -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(),
}
}

2
sql3/planner/executionplannersystemtables.go
View file

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

64
sql3/planner/opquery.go
View file

@ -2,14 +2,20 @@ package planner
import (
"context"
"encoding/json"
"fmt"
"time"
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
}

150
sql3/planner/opsystemtable.go
View file

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

1
sql3/planner/optop.go
View file

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

49
sql3/sql_complex_test.go
View file

@ -62,7 +62,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)
@ -85,12 +85,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"))
}

8
sql3/test/helpers.go
View file

@ -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 {

99
systemlayer.go Normal file
View file

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

108
systemlayer/executionrequests.go Normal file
View file

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

20
systemlayer/systemlayer.go Normal file
View file

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