fb-1744 implement system tables (#2276)

* implement system tables that contain internal state information from FeatureBase

* review feedback

* review feedback

* removed sys prefixes

(cherry picked from commit 45067b4e32)

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

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{

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)

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,

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

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

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