Add support for SHOW queries

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

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 {

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

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

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