diff --git a/proto/interface.go b/proto/interface.go index 660aa6273..6a6df7213 100644 --- a/proto/interface.go +++ b/proto/interface.go @@ -31,6 +31,15 @@ type StreamClient interface { Recv() (*RowResponse, error) } +// EmptyStream implements StreamClient interface. +// It always returns empty RowResponse +type EmptyStream struct{} + +// Recv returns io.EOF +func (EmptyStream) Recv() (*RowResponse, error) { + return nil, io.EOF +} + // ReadIntoTable reads from a StreamClient and stores the result into a table response. func ReadIntoTable(cli StreamClient) (*TableResponse, error) { var headers []*ColumnInfo diff --git a/server/grpc_test.go b/server/grpc_test.go index e143c9168..4529c0df8 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -686,7 +686,6 @@ func TestQuerySQLUnary(t *testing.T) { }, eq: equalUnordered, }, - { // GroupBy(Rows(field='age'),limit=3) sql: "select age, count(*) as cnt from grouper group by age order by cnt desc, age desc limit 3", @@ -703,6 +702,35 @@ func TestQuerySQLUnary(t *testing.T) { }, eq: equal, }, + { + sql: "show tables", + exp: tableResponse{ + headers: []columnInfo{ + {"Table", "string"}, + }, + rows: []row{ + {[]columnResponse{"grouper"}}, + {[]columnResponse{"joiner"}}, + }, + }, + eq: equal, + }, + { + sql: "show fields from grouper", + exp: tableResponse{ + headers: []columnInfo{ + {"Field", "string"}, + {"Type", "string"}, + }, + rows: []row{ + {[]columnResponse{"age", "int64"}}, + {[]columnResponse{"color", "[]string"}}, + {[]columnResponse{"height", "int64"}}, + {[]columnResponse{"score", "int64"}}, + }, + }, + eq: equal, + }, } for i, test := range tests { diff --git a/server/sql.go b/server/sql.go index 1e648b9e3..0f1de916c 100644 --- a/server/sql.go +++ b/server/sql.go @@ -41,6 +41,12 @@ func execSQL(ctx context.Context, api *pilosa.API, logger logger.Logger, querySt if err != nil { return nil, errors.Wrap(err, "failed to start SQL query") } + case sql.SQLTypeShow: + handler := sql.NewShowHandler(api) + results, err = handler.Handle(ctx, query) + if err != nil { + return nil, errors.Wrap(err, "failed to start SQL query") + } default: return nil, status.Errorf(codes.Unimplemented, "query type not supported") } diff --git a/sql/select.go b/sql/select.go index 1c11cdfca..a195a7c85 100644 --- a/sql/select.go +++ b/sql/select.go @@ -42,7 +42,11 @@ func NewSelectHandler(api *pilosa.API) *SelectHandler { // Handle executes mapped SQL func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) { - mr, err := s.mapSelect(ctx, mapped.Statement.(*sqlparser.Select), mapped.Mask) + stmt, ok := mapped.Statement.(*sqlparser.Select) + if !ok { + return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement) + } + mr, err := s.mapSelect(ctx, stmt, mapped.Mask) if err != nil { return nil, errors.Wrap(err, "mapping select") } @@ -70,12 +74,11 @@ func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Sel return mr, nil } -func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (stream pproto.StreamClient, err error) { +func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.StreamClient, error) { if mr.Query == "" { return nil, errors.New("no pql query created") } - fmt.Println("PQL:", mr.Query) resp, err := s.api.Query(ctx, &pilosa.QueryRequest{Index: mr.IndexName, Query: mr.Query}) if err != nil { return nil, errors.Wrap(err, "doing pql query") diff --git a/sql/show.go b/sql/show.go new file mode 100644 index 000000000..4ac3c390d --- /dev/null +++ b/sql/show.go @@ -0,0 +1,148 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sql + +import ( + "context" + "fmt" + + "github.com/pilosa/pilosa/v2" + pproto "github.com/pilosa/pilosa/v2/proto" + "github.com/pkg/errors" + "vitess.io/vitess/go/vt/sqlparser" +) + +// ShowHandler executes SQL show table/field statements +type ShowHandler struct { + api *pilosa.API +} + +// NewShowHandler constructor +func NewShowHandler(api *pilosa.API) *ShowHandler { + return &ShowHandler{ + api: api, + } +} + +// Handle executes mapped SQL +func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) { + stmt, ok := mapped.Statement.(*sqlparser.Show) + if !ok { + return nil, fmt.Errorf("statement is not type show: %T", mapped.Statement) + } + + switch stmt.Type { + case "tables": + return s.execShowTables(ctx, stmt) + case "fields": + return s.execShowFields(ctx, stmt) + default: + return nil, fmt.Errorf("cannot show: %s", stmt.Type) + } +} + +func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) { + indexInfo := s.api.Schema(ctx) + sz := len(indexInfo) + // If there aren't any indexes, don't bother creating + // a result row buffer. + if sz == 0 { + return pproto.EmptyStream{}, nil + } + + // Create a buffer large enough to hold the entire result + // set. This way we don't have to use a goroutine. + result := pproto.NewRowBuffer(sz) + for _, ii := range indexInfo { + rr := &pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: "Table", Datatype: "string"}, + }, + Columns: []*pproto.ColumnResponse{ + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: ii.Name}}, + }, + } + if err := result.Send(rr); err != nil { + return nil, errors.Wrap(err, "sending row response") + } + } + if err := result.Send(pproto.EOF); err != nil { + return nil, errors.Wrap(err, "sending EOF") + } + + // Apply Sort Reducer + out := pproto.NewRowBuffer(0) + red := NewOrderByReducer([]string{"Table"}, []string{"asc"}, 0, 0) + go red.Reduce(result, out) //nolint:errcheck + + result = out + return result, nil +} + +func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) { + indexName := showStmt.OnTable.ToViewName().Name.String() + index, err := s.api.Index(ctx, indexName) + if err != nil { + return nil, errors.Wrap(err, "getting schema") + } + if index == nil { + return nil, pilosa.ErrIndexNotFound + } + fields := index.Fields() + sz := len(fields) + // If there aren't any fields, don't bother creating + // a result row buffer. + if sz == 0 { + return pproto.EmptyStream{}, nil + } + + // Create a buffer large enough to hold the entire result + // set. This way we don't have to use a goroutine. + result := pproto.NewRowBuffer(sz) + for _, f := range fields { + if f.Name() == "_exists" { + continue + } + + dt, err := f.Datatype() + if err != nil { + return nil, errors.Wrapf(err, "field %s", f.Name()) + } + rr := &pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: "Field", Datatype: "string"}, + {Name: "Type", Datatype: "string"}, + }, + Columns: []*pproto.ColumnResponse{ + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: f.Name()}}, + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: dt}}, + }, + } + if err := result.Send(rr); err != nil { + return nil, errors.Wrap(err, "sending row response") + } + } + if err := result.Send(pproto.EOF); err != nil { + return nil, errors.Wrap(err, "sending EOF") + } + + // Apply Sort Reducer + out := pproto.NewRowBuffer(0) + red := NewOrderByReducer([]string{"Field"}, []string{"asc"}, 0, 0) + go red.Reduce(result, out) //nolint:errcheck + + result = out + return result, nil +}