diff --git a/api.go b/api.go index 1d7be7448..a2dc92b4a 100644 --- a/api.go +++ b/api.go @@ -2896,6 +2896,27 @@ type SchemaAPI interface { Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error) } +type ClusterNode struct { + ID string + State string + URI string + GRPCURI string + IsPrimary bool +} + +type SystemAPI interface { + ClusterName() string + Version() string + PlatformDescription() string + PlatformVersion() string + ClusterNodeCount() int + ClusterReplicaCount() int + ShardWidth() int + ClusterState() string + + ClusterNodes() []ClusterNode +} + // CreateFieldObj is used to encapsulate the information required for creating a // field in the SchemaAPI.CreateIndexAndFields interface method. type CreateFieldObj struct { @@ -2980,3 +3001,74 @@ func (fapi *FeatureBaseSchemaAPI) IndexInfo(ctx context.Context, indexName strin return idx, nil } + +// FeatureBaseSystemAPI is a wrapper around pilosa.API. It implements the +// SystemAPI interface +type FeatureBaseSystemAPI struct { + *API +} + +func (fsapi *FeatureBaseSystemAPI) ClusterName() string { + return fsapi.API.ClusterName() +} + +func (fsapi *FeatureBaseSystemAPI) Version() string { + return fsapi.API.Version() +} + +func (fsapi *FeatureBaseSystemAPI) PlatformDescription() string { + si := fsapi.server.systemInfo + platform, err := si.Platform() + if err != nil { + return "unknown" + } + return platform +} + +func (fsapi *FeatureBaseSystemAPI) PlatformVersion() string { + si := fsapi.server.systemInfo + platformVersion, err := si.OSVersion() + if err != nil { + return "unknown" + } + return platformVersion +} + +func (fsapi *FeatureBaseSystemAPI) ClusterNodeCount() int { + return len(fsapi.cluster.noder.Nodes()) +} + +func (fsapi *FeatureBaseSystemAPI) ClusterReplicaCount() int { + return fsapi.cluster.ReplicaN +} + +func (fsapi *FeatureBaseSystemAPI) ShardWidth() int { + return ShardWidth +} + +func (fsapi *FeatureBaseSystemAPI) ClusterState() string { + state, err := fsapi.State() + if err != nil { + return "UNKNOWN" + } + return string(state) +} + +func (fsapi *FeatureBaseSystemAPI) ClusterNodes() []ClusterNode { + result := make([]ClusterNode, 0) + + nodes := fsapi.Hosts(context.Background()) + + for _, n := range nodes { + scn := ClusterNode{ + ID: n.ID, + State: string(n.State), + URI: n.URI.String(), + GRPCURI: n.GRPCURI.String(), + IsPrimary: n.IsPrimary, + } + result = append(result, scn) + } + + return result +} diff --git a/server/server.go b/server/server.go index dfd23e520..613706e96 100644 --- a/server/server.go +++ b/server/server.go @@ -444,8 +444,10 @@ func (m *Command) SetupServer() error { executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner { fapi := &pilosa.FeatureBaseSchemaAPI{API: api} + fsapi := &pilosa.FeatureBaseSystemAPI{API: api} fimp := &batch.FeaturebaseImporter{API: api} - return planner.NewExecutionPlanner(e, fapi, api, fimp, m.logger, sql) + + return planner.NewExecutionPlanner(e, fapi, fsapi, api, fimp, m.logger, sql) } serverOptions := []pilosa.ServerOption{ diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index 6b99cb5cb..36cdf7484 100644 --- a/sql3/planner/compileselect.go +++ b/sql3/planner/compileselect.go @@ -189,6 +189,19 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, source parser return NewPlanOpNestedLoops(topOp, bottomOp, joinCondition), nil case *parser.QualifiedTableName: + + tableName := parser.IdentName(sourceExpr.Name) + + // doing this check here because we don't have a 'system' flag that exists in the FB schema + st, ok := systemTables[strings.ToLower(tableName)] + if ok { + if sourceExpr.Alias != nil { + aliasName := parser.IdentName(sourceExpr.Alias) + return NewPlanOpRelAlias(aliasName, NewPlanOpSystemTable(p, st)), nil + } + return NewPlanOpSystemTable(p, st), nil + + } // get all the qualified refs that refer to this table extractColumns := make([]string, 0) for _, r := range scope.referenceList { @@ -205,14 +218,11 @@ func (p *ExecutionPlanner) compileSelectSource(scope *PlanOpQuery, source parser } } } - - tableName := parser.IdentName(sourceExpr.Name) - if sourceExpr.Alias != nil { aliasName := parser.IdentName(sourceExpr.Alias) + return NewPlanOpRelAlias(aliasName, NewPlanOpPQLTableScan(p, tableName, extractColumns)), nil } - return NewPlanOpPQLTableScan(p, tableName, extractColumns), nil case *parser.TableValuedFunction: @@ -312,7 +322,7 @@ func (p *ExecutionPlanner) analyzeSource(source parser.Source, scope parser.Stat return nil case *parser.TableValuedFunction: - //check it actually is a table valued function - we only support one right now; subtable() + // check it actually is a table valued function - we only support one right now; subtable() switch strings.ToUpper(source.Name.Name) { case "SUBTABLE": _, err := p.analyzeCallExpression(source.Call, scope) diff --git a/sql3/planner/executionplanner.go b/sql3/planner/executionplanner.go index 2a3dc8c35..b9e966d15 100644 --- a/sql3/planner/executionplanner.go +++ b/sql3/planner/executionplanner.go @@ -24,6 +24,7 @@ type PlannerScope struct { type ExecutionPlanner struct { executor pilosa.Executor schemaAPI pilosa.SchemaAPI + systemAPI pilosa.SystemAPI computeAPI pilosa.ComputeAPI importer batch.Importer logger logger.Logger @@ -31,10 +32,11 @@ type ExecutionPlanner struct { scopeStack *scopeStack } -func NewExecutionPlanner(executor pilosa.Executor, schemaAPI pilosa.SchemaAPI, 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, importer batch.Importer, logger logger.Logger, sql string) *ExecutionPlanner { return &ExecutionPlanner{ executor: executor, - schemaAPI: schemaAPI, + schemaAPI: newSystemTableDefintionsWrapper(schemaAPI), + systemAPI: systemAPI, computeAPI: computeAPI, importer: importer, logger: logger, diff --git a/sql3/planner/executionplanner_test.go b/sql3/planner/executionplanner_test.go index 8cc0c30c5..6a05c6079 100644 --- a/sql3/planner/executionplanner_test.go +++ b/sql3/planner/executionplanner_test.go @@ -44,12 +44,35 @@ func TestPlanner_Show(t *testing.T) { t.Fatal(err) } + t.Run("SystemTables", 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) + } + if len(results) != 1 { + t.Fatal(fmt.Errorf("unexpected result set length")) + } + + if diff := cmp.Diff([]*planner_types.PlannerColumn{ + {ColumnName: "name", Type: parser.NewDataTypeString()}, + {ColumnName: "platform", Type: parser.NewDataTypeString()}, + {ColumnName: "platform_version", Type: parser.NewDataTypeString()}, + {ColumnName: "db_version", Type: parser.NewDataTypeString()}, + {ColumnName: "state", Type: parser.NewDataTypeString()}, + {ColumnName: "node_count", Type: parser.NewDataTypeInt()}, + {ColumnName: "shard_width", Type: parser.NewDataTypeInt()}, + {ColumnName: "replica_count", Type: parser.NewDataTypeInt()}, + }, 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) != 2 { + if len(results) != 4 { t.Fatal(fmt.Errorf("unexpected result set length")) } diff --git a/sql3/planner/executionplannersystemtables.go b/sql3/planner/executionplannersystemtables.go new file mode 100644 index 000000000..f079dd402 --- /dev/null +++ b/sql3/planner/executionplannersystemtables.go @@ -0,0 +1,104 @@ +// Copyright 2022 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + + pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/parser" + "github.com/pkg/errors" +) + +type systemTableDefintionsWrapper struct { + schemaAPI pilosa.SchemaAPI +} + +func newSystemTableDefintionsWrapper(schemaAPI pilosa.SchemaAPI) *systemTableDefintionsWrapper { + return &systemTableDefintionsWrapper{ + schemaAPI: schemaAPI, + } +} + +func (s *systemTableDefintionsWrapper) CreateIndexAndFields(ctx context.Context, indexName string, options pilosa.IndexOptions, fields []pilosa.CreateFieldObj) error { + return s.schemaAPI.CreateIndexAndFields(ctx, indexName, options, fields) +} + +func (s *systemTableDefintionsWrapper) CreateField(ctx context.Context, indexName string, fieldName string, opts ...pilosa.FieldOption) (*pilosa.Field, error) { + return s.schemaAPI.CreateField(ctx, indexName, fieldName, opts...) +} + +func (s *systemTableDefintionsWrapper) DeleteField(ctx context.Context, indexName string, fieldName string) error { + return s.schemaAPI.DeleteField(ctx, indexName, fieldName) +} + +func (s *systemTableDefintionsWrapper) DeleteIndex(ctx context.Context, indexName string) error { + return s.schemaAPI.DeleteIndex(ctx, indexName) +} + +func (s *systemTableDefintionsWrapper) IndexInfo(ctx context.Context, indexName string) (*pilosa.IndexInfo, error) { + i, err := s.schemaAPI.IndexInfo(ctx, indexName) + if err != nil { + if errors.Is(err, pilosa.ErrIndexNotFound) { + st, ok := systemTables[indexName] + if !ok { + return nil, pilosa.ErrIndexNotFound + } + + return indexInfoFromSystemTable(st) + } + return nil, err + } + return i, nil +} + +func (s *systemTableDefintionsWrapper) Schema(ctx context.Context, withViews bool) ([]*pilosa.IndexInfo, error) { + schema, err := s.schemaAPI.Schema(ctx, withViews) + if err != nil { + return nil, err + } + for _, st := range systemTables { + + i, err := indexInfoFromSystemTable(st) + if err != nil { + return nil, err + } + schema = append(schema, i) + } + return schema, err +} + +func indexInfoFromSystemTable(st *systemTable) (*pilosa.IndexInfo, error) { + + fields := make([]*pilosa.FieldInfo, 0) + + for _, f := range st.schema { + + var opts pilosa.FieldOptions + switch f.Type.(type) { + case *parser.DataTypeInt: + opts.Type = pilosa.FieldTypeInt + case *parser.DataTypeBool: + opts.Type = pilosa.FieldTypeBool + case *parser.DataTypeString: + opts.Type = pilosa.FieldTypeMutex + opts.Keys = true + default: + return nil, sql3.NewErrInternalf("unexpected system table field type '%T'", f.Type) + } + + fld := &pilosa.FieldInfo{ + Name: f.ColumnName, + Options: opts, + } + fields = append(fields, fld) + } + + i := &pilosa.IndexInfo{ + Name: st.name, + Fields: fields, + } + + return i, nil +} diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go new file mode 100644 index 000000000..a8f348b56 --- /dev/null +++ b/sql3/planner/opsystemtable.go @@ -0,0 +1,241 @@ +// Copyright 2021 Molecula Corp. All rights reserved. + +package planner + +import ( + "context" + "fmt" + + pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/sql3" + "github.com/molecula/featurebase/v3/sql3/parser" + "github.com/molecula/featurebase/v3/sql3/planner/types" +) + +//fb_exec_requests +// session +// user +// start_time +// end_time +// status +// plan +// wait_type +// wait_time +// wait_resource +// cpu_time +// elapsed_time +// reads +// writes +// logical_reads +// row_count + +// exclude this file from SonarCloud dupe eval + +const ( + fbClusterInfo = "fb_cluster_info" + fbClusterNodes = "fb_cluster_nodes" +) + +type systemTable struct { + name string + schema types.Schema +} + +var systemTables = map[string]*systemTable{ + fbClusterInfo: { + name: fbClusterInfo, + schema: types.Schema{ + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "name", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "platform", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "platform_version", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "db_version", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "state", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "node_count", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "shard_width", + Type: parser.NewDataTypeInt(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "replica_count", + Type: parser.NewDataTypeInt(), + }, + }, + }, + fbClusterNodes: { + name: fbClusterNodes, + schema: types.Schema{ + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "id", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "state", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "uri", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "grpc_uri", + Type: parser.NewDataTypeString(), + }, + &types.PlannerColumn{ + RelationName: fbClusterInfo, + ColumnName: "is_primary", + Type: parser.NewDataTypeBool(), + }, + }, + }, +} + +// PlanOpSystemTable handles system tables +type PlanOpSystemTable struct { + planner *ExecutionPlanner + table *systemTable + warnings []string +} + +func NewPlanOpSystemTable(p *ExecutionPlanner, table *systemTable) *PlanOpSystemTable { + return &PlanOpSystemTable{ + planner: p, + table: table, + warnings: make([]string, 0), + } +} + +func (p *PlanOpSystemTable) Plan() map[string]interface{} { + result := make(map[string]interface{}) + result["_op"] = fmt.Sprintf("%T", p) + ps := make([]string, 0) + for _, e := range p.Schema() { + ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeName())) + } + result["_schema"] = ps + return result +} + +func (p *PlanOpSystemTable) String() string { + return "" +} + +func (p *PlanOpSystemTable) AddWarning(warning string) { + p.warnings = append(p.warnings, warning) +} + +func (p *PlanOpSystemTable) Warnings() []string { + return p.warnings +} + +func (p *PlanOpSystemTable) Schema() types.Schema { + return p.table.schema +} + +func (p *PlanOpSystemTable) Children() []types.PlanOperator { + return []types.PlanOperator{} +} + +func (p *PlanOpSystemTable) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) { + switch p.table.name { + case fbClusterInfo: + return &fbClusterInfoRowIter{ + planner: p.planner, + }, nil + case fbClusterNodes: + return &fbClusterNodesRowIter{ + planner: p.planner, + }, nil + default: + return nil, sql3.NewErrInternalf("unable to find system table '%s'", p.table.name) + } +} + +func (p *PlanOpSystemTable) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) { + if len(children) > 0 { + return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children)) + } + return NewPlanOpSystemTable(p.planner, p.table), nil +} + +type fbClusterInfoRowIter struct { + planner *ExecutionPlanner + rowIndex int +} + +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.PlatformDescription(), + i.planner.systemAPI.PlatformVersion(), + i.planner.systemAPI.Version(), + i.planner.systemAPI.ClusterState(), + i.planner.systemAPI.ClusterNodeCount(), + i.planner.systemAPI.ShardWidth(), + i.planner.systemAPI.ClusterReplicaCount(), + } + i.rowIndex += 1 + return row, nil + } + return nil, types.ErrNoMoreRows +} + +type fbClusterNodesRowIter struct { + planner *ExecutionPlanner + result []pilosa.ClusterNode +} + +var _ types.RowIterator = (*fbClusterNodesRowIter)(nil) + +func (i *fbClusterNodesRowIter) Next(ctx context.Context) (types.Row, error) { + if i.result == nil { + i.result = i.planner.systemAPI.ClusterNodes() + } + + if len(i.result) > 0 { + n := i.result[0] + row := []interface{}{ + n.ID, + n.State, + n.URI, + n.GRPCURI, + n.IsPrimary, + } + // Move to next result element. + i.result = i.result[1:] + return row, nil + } + return nil, types.ErrNoMoreRows +}