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>

(cherry picked from commit 47d8be26f5)
This commit is contained in:
pokeeffe-molecula 2022-12-02 16:42:45 -06:00 • committed by Fletcher Haynes
parent daf6e29b02
commit dca0dd84e3
23 changed files with 595 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
api.go
View file

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

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

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

9
internal_client.go
View file

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

17
internal_client_test.go
View file

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

2
server.go
View file

@ -58,6 +58,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)
}

68
sql3/planner/opquery.go
View file

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

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

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

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
}