diff --git a/executor.go b/executor.go index 67932cf8d..f329e336e 100644 --- a/executor.go +++ b/executor.go @@ -2820,6 +2820,119 @@ type ExtractedTable struct { Columns []ExtractedTableColumn `json:"columns"` } +// ToRows implements the ToRowser interface. +func (t ExtractedTable) ToRows(callback func(*pb.RowResponse) error) error { + if len(t.Columns) == 0 { + return nil + } + + headers := make([]*pb.ColumnInfo, len(t.Fields)+1) + colType := "uint64" + if t.Columns[0].Column.Keyed { + colType = "string" + } + headers[0] = &pb.ColumnInfo{ + Name: "_id", + Datatype: colType, + } + dataHeaders := headers[1:] + for i, f := range t.Fields { + dataHeaders[i] = &pb.ColumnInfo{ + Name: f.Name, + Datatype: f.Type, + } + } + + for _, c := range t.Columns { + cols := make([]*pb.ColumnResponse, len(c.Rows)+1) + if c.Column.Keyed { + cols[0] = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_StringVal{ + StringVal: c.Column.Key, + }, + } + } else { + cols[0] = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_Uint64Val{ + Uint64Val: c.Column.ID, + }, + } + } + valCols := cols[1:] + for i, r := range c.Rows { + var col *pb.ColumnResponse + switch r := r.(type) { + case bool: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_BoolVal{ + BoolVal: r, + }, + } + case int64: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_Int64Val{ + Int64Val: r, + }, + } + case uint64: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_Uint64Val{ + Uint64Val: r, + }, + } + case string: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_StringVal{ + StringVal: r, + }, + } + case []uint64: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{ + Uint64ArrayVal: &pb.Uint64Array{ + Vals: r, + }, + }, + } + case []string: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_StringArrayVal{ + StringArrayVal: &pb.StringArray{ + Vals: r, + }, + }, + } + case pql.Decimal: + col = &pb.ColumnResponse{ + ColumnVal: &pb.ColumnResponse_DecimalVal{ + DecimalVal: &pb.Decimal{ + Value: r.Value, + Scale: r.Scale, + }, + }, + } + default: + return errors.Errorf("unsupported field value: %v (type: %T)", r, r) + } + valCols[i] = col + } + err := callback(&pb.RowResponse{ + Headers: headers, + Columns: cols, + }) + if err != nil { + return err + } + } + + return nil +} + +// ToTable converts the table to protobuf format. +func (t ExtractedTable) ToTable() (*pb.TableResponse, error) { + return pb.RowsToTable(t, len(t.Columns)) +} + type ExtractedIDColumn struct { ColumnID uint64 Rows [][]uint64 @@ -5019,15 +5132,17 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index return nil, ErrFieldNotFound } - typ := field.Type() - + datatype, err := field.Datatype() + if err != nil { + return nil, errors.Wrapf(err, "field %s", v) + } fields[i] = ExtractedTableField{ Name: v, - Type: typ, + Type: datatype, } var mapper fieldMapper - switch typ { + switch typ := field.Type(); typ { case FieldTypeBool: mapper = func(ids []uint64) (interface{}, error) { switch len(ids) { diff --git a/executor_test.go b/executor_test.go index 3bd9c1139..e29b8d93b 100644 --- a/executor_test.go +++ b/executor_test.go @@ -4344,7 +4344,11 @@ func TestExecutor_Execute_Extract(t *testing.T) { c := test.MustRunCluster(t, 3) defer c.Close() - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "set") + set := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "set") + dtSet, err := set.Datatype() + if err != nil { + t.Fatal(err) + } c.ImportBits(t, "i", "set", [][2]uint64{ {0, 1}, {0, 2}, @@ -4355,56 +4359,88 @@ func TestExecutor_Execute_Extract(t *testing.T) { }) c.Query(t, "i", fmt.Sprintf("Clear(%d, set=5)", ShardWidth)) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keyset", pilosa.OptFieldKeys()) + keyset := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keyset", pilosa.OptFieldKeys()) + dtKeyset, err := keyset.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, keyset="h") Set(1, keyset="xyzzy") Set(0, keyset="plugh") `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "mutex", pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000)) + mutex := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "mutex", pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000)) + dtMutex, err := mutex.Datatype() + if err != nil { + t.Fatal(err) + } c.ImportBits(t, "i", "mutex", [][2]uint64{ {0, 1}, {0, 2}, {4, 4 * ShardWidth}, }) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keymutex", pilosa.OptFieldKeys(), pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000)) + keymutex := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keymutex", pilosa.OptFieldKeys(), pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000)) + dtKeyMutex, err := keymutex.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, keymutex="h") Set(1, keymutex="xyzzy") Set(3, keymutex="plugh") `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "time", pilosa.OptFieldTypeTime("YMDH")) + tm := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "time", pilosa.OptFieldTypeTime("YMDH")) + dtTm, err := tm.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, time=1, 2016-01-01T00:00) Set(1, time=2, 2017-01-01T00:00) Set(3, time=3, 2018-01-01T00:00) `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keytime", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime("YMDH")) + keytm := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keytime", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime("YMDH")) + dtKeyTm, err := keytm.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, keytime="h", 2016-01-01T00:00) Set(1, keytime="xyzzy", 2017-01-01T00:00) Set(0, keytime="plugh", 2018-01-01T00:00) `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsint", pilosa.OptFieldTypeInt(-100, 100)) + bsiInt := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsint", pilosa.OptFieldTypeInt(-100, 100)) + dtBsiInt, err := bsiInt.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, bsint=1) Set(1, bsint=-1) Set(3, bsint=2) `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsidecimal", pilosa.OptFieldTypeDecimal(2)) + bsidecimal := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsidecimal", pilosa.OptFieldTypeDecimal(2)) + dtBsiDecimal, err := bsidecimal.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, bsidecimal=0.01) Set(1, bsidecimal=1.00) Set(3, bsidecimal=-1.01) `) - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bool", pilosa.OptFieldTypeBool()) + boolean := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bool", pilosa.OptFieldTypeBool()) + dtBoolean, err := boolean.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set(0, bool=true) Set(1, bool=false) @@ -4417,39 +4453,39 @@ func TestExecutor_Execute_Extract(t *testing.T) { Fields: []pilosa.ExtractedTableField{ { Name: "set", - Type: pilosa.FieldTypeSet, + Type: dtSet, }, { Name: "keyset", - Type: pilosa.FieldTypeSet, + Type: dtKeyset, }, { Name: "mutex", - Type: pilosa.FieldTypeMutex, + Type: dtMutex, }, { Name: "keymutex", - Type: pilosa.FieldTypeMutex, + Type: dtKeyMutex, }, { Name: "time", - Type: pilosa.FieldTypeTime, + Type: dtTm, }, { Name: "keytime", - Type: pilosa.FieldTypeTime, + Type: dtKeyTm, }, { Name: "bsint", - Type: pilosa.FieldTypeInt, + Type: dtBsiInt, }, { Name: "bsidecimal", - Type: pilosa.FieldTypeDecimal, + Type: dtBsiDecimal, }, { Name: "bool", - Type: pilosa.FieldTypeBool, + Type: dtBoolean, }, }, Columns: []pilosa.ExtractedTableColumn{ @@ -4574,7 +4610,11 @@ func TestExecutor_Execute_Extract_Keyed(t *testing.T) { c := test.MustRunCluster(t, 3) defer c.Close() - c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true, Keys: true}, "set") + set := c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true, Keys: true}, "set") + dtSet, err := set.Datatype() + if err != nil { + t.Fatal(err) + } c.Query(t, "i", ` Set("h", set=1) Set("h", set=2) @@ -4589,7 +4629,7 @@ func TestExecutor_Execute_Extract_Keyed(t *testing.T) { Fields: []pilosa.ExtractedTableField{ { Name: "set", - Type: "set", + Type: dtSet, }, }, Columns: []pilosa.ExtractedTableColumn{ diff --git a/field.go b/field.go index c557f0e81..704f05ae9 100644 --- a/field.go +++ b/field.go @@ -1321,6 +1321,36 @@ func (f *Field) ClearBit(tx Tx, rowID, colID uint64) (changed bool, err error) { return changed, nil } +// Datatype returns a useful data type (string, +// uint64, bool, etc.) based on the field type. +func (f *Field) Datatype() (string, error) { + switch t := f.Type(); t { + case "set": + if f.Keys() { + return "[]string", nil + } + return "[]uint64", nil + case "mutex": + if f.Keys() { + return "string", nil + } + return "uint64", nil + case "int": + if f.Keys() { + return "string", nil + } + return "int64", nil + case "decimal": + return "decimal", nil + case "bool": + return "bool", nil + case "time": + return "int64", nil // TODO: this is a placeholder + default: + return "", fmt.Errorf("unimplemented field Datatype: %s", t) + } +} + func groupCompare(a, b string, offset int) (lt, eq bool) { if len(a) > offset { a = a[:offset] diff --git a/go.mod b/go.mod index 70221964e..8e545a410 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( google.golang.org/grpc v1.28.0 modernc.org/mathutil v1.0.0 modernc.org/strutil v1.0.0 + vitess.io/vitess v3.0.0-rc.3.0.20190602171040-12bfde34629c+incompatible ) go 1.13 diff --git a/go.sum b/go.sum index f97f5d051..ad4e3d287 100644 --- a/go.sum +++ b/go.sum @@ -288,3 +288,5 @@ modernc.org/mathutil v1.0.0 h1:93vKjrJopTPrtTNpZ8XIovER7iCIH1QU7wNbOQXC60I= modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k= modernc.org/strutil v1.0.0 h1:XVFtQwFVwc02Wk+0L/Z/zDDXO81r5Lhe6iMKmGX3KhE= modernc.org/strutil v1.0.0/go.mod h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs= +vitess.io/vitess v3.0.0-rc.3.0.20190602171040-12bfde34629c+incompatible h1:GWnLrAdetgJM0Co5bwwczO49iFZBSInpyGAT77BP9Y0= +vitess.io/vitess v3.0.0-rc.3.0.20190602171040-12bfde34629c+incompatible/go.mod h1:h4qvkyNYTOC0xI+vcidSWoka0gQAZc9ZPHbkHo48gP0= diff --git a/proto/interface.go b/proto/interface.go index 34bf53829..660aa6273 100644 --- a/proto/interface.go +++ b/proto/interface.go @@ -16,6 +16,7 @@ package pilosa import ( "fmt" + "io" "strings" "github.com/pkg/errors" @@ -30,6 +31,36 @@ type StreamClient interface { Recv() (*RowResponse, error) } +// ReadIntoTable reads from a StreamClient and stores the result into a table response. +func ReadIntoTable(cli StreamClient) (*TableResponse, error) { + var headers []*ColumnInfo + rows := []*Row{} + +rx: + for { + row, err := cli.Recv() + switch err { + case nil: + case io.EOF: + break rx + default: + return nil, err + } + + if headers == nil { + headers = row.Headers + } + rows = append(rows, &Row{ + Columns: row.Columns, + }) + } + + return &TableResponse{ + Headers: headers, + Rows: rows, + }, nil +} + // StreamServer is an interface for a stream // which can accept a RowResponse to be later // returned by the stream via Recv(). @@ -84,6 +115,52 @@ func RowsToTable(tr ToRowser, n int) (*TableResponse, error) { }, nil } +// RowBuffer acts as a Sender/Receiver of RowResponses. +// Note that sending a nil value will cause the Recv +// method to return an io.EOF error. +type RowBuffer struct { + ch chan *RowResponse +} + +// NewRowBuffer returns a new instance of RowBuffer. +// sz is the size of the buffer. +func NewRowBuffer(sz int) *RowBuffer { + var chSz int + if sz > 0 { + // Add one to allow for the EOF record. + chSz = sz + 1 + } + return &RowBuffer{ + ch: make(chan *RowResponse, chSz), + } +} + +// Recv returns a RowResponse. When the buffer is empty, +// calling Recv will return an io.EOF error. +func (rb *RowBuffer) Recv() (*RowResponse, error) { + r := <-rb.ch + + // If the StatusError contains a message then return + // with the approprate error. + se := r.GetStatusError() + code := codes.Code(se.GetCode()) + msg := se.GetMessage() + if code != codes.OK { + return nil, status.Error(code, msg) + } else if msg == "EOF" { + return nil, io.EOF + } else if msg != "" { + return nil, errors.New(msg) + } + + return r, nil +} + +func (rb *RowBuffer) Send(rr *RowResponse) error { + rb.ch <- rr + return nil +} + // EOF acts as an io.EOF encoded into a RowResponse. var EOF *RowResponse = &RowResponse{ StatusError: &StatusError{ diff --git a/server/grpc.go b/server/grpc.go index 5fd16692b..3dcc420a1 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -18,6 +18,7 @@ import ( "context" "crypto/tls" "fmt" + "io" "net" "strings" "time" @@ -25,6 +26,7 @@ import ( "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/logger" pb "github.com/pilosa/pilosa/v2/proto" + "github.com/pilosa/pilosa/v2/sql" "github.com/pilosa/pilosa/v2/stats" "github.com/pkg/errors" "google.golang.org/grpc" @@ -117,9 +119,49 @@ func (h *GRPCHandler) DeleteVDS(ctx context.Context, req *pb.DeleteVDSRequest) ( return &pb.DeleteVDSResponse{}, nil } +func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.StreamClient, error) { + mapper := sql.NewMapper() + mapper.Logger = h.logger + query, err := mapper.MapSQL(queryStr) + if err != nil { + return nil, errors.Wrap(err, "failed to map SQL") + } + var results pb.StreamClient + switch query.SQLType { + case sql.SQLTypeSelect: + handler := sql.NewSelectHandler(h.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") + } + return results, nil +} + // QuerySQL handles the SQL request and sends RowResponses to the stream. -func (*GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQLServer) error { - return status.Errorf(codes.Unimplemented, "method QuerySQL not implemented") +func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQLServer) error { + results, err := h.execSQL(stream.Context(), req.Sql) + if err != nil { + return err + } + + for { + row, err := results.Recv() + switch err { + case nil: + case io.EOF: + return nil + default: + return errors.Wrap(err, "failed to load next row") + } + + err = stream.Send(row) + if err != nil { + return errors.Wrap(err, "failed to send row") + } + } } // QuerySQLUnary is a unary-response (non-streaming) version of QuerySQL, returning a TableResponse. @@ -134,8 +176,12 @@ func (*GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQLS // Futures, which are used by python-molecula to perform multiple queries // concurrently. There is additional discussion and historical context here: // https://github.com/molecula/pilosa/pull/644 -func (*GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest) (*pb.TableResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method QuerySQLUnary not implemented") +func (h *GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest) (*pb.TableResponse, error) { + results, err := h.execSQL(ctx, req.Sql) + if err != nil { + return nil, err + } + return pb.ReadIntoTable(results) } // QueryPQL handles the PQL request and sends RowResponses to the stream. @@ -298,36 +344,6 @@ func ToRowserWrapper(result interface{}) (pb.ToRowser, error) { return toRowser, nil } -// fieldDataType returns a useful data type (string, -// uint64, bool, etc.) based on the Pilosa field type. -func fieldDataType(f *pilosa.Field) string { - switch f.Type() { - case "set": - if f.Keys() { - return "[]string" - } - return "[]uint64" - case "mutex": - if f.Keys() { - return "string" - } - return "uint64" - case "int": - if f.Keys() { - return "string" - } - return "int64" - case "decimal": - return "decimal" - case "bool": - return "bool" - case "time": - return "int64" // TODO: this is a placeholder - default: - panic(fmt.Sprintf("unimplemented fieldDataType: %s", f.Type())) - } -} - // Inspect handles the inspect request and sends an InspectResponse to the stream. func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectServer) error { const defaultLimit = 100000 @@ -422,7 +438,11 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe {Name: "_id", Datatype: "uint64"}, } for _, field := range fields { - ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fieldDataType(field)}) + fdt, err := field.Datatype() + if err != nil { + return errors.Wrapf(err, "field %s", field.Name()) + } + ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fdt}) } // If Columns is empty, then get the _exists list (via All()), @@ -666,7 +686,11 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe {Name: "_id", Datatype: "string"}, } for _, field := range fields { - ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fieldDataType(field)}) + fdt, err := field.Datatype() + if err != nil { + return errors.Wrapf(err, "field %s", field.Name()) + } + ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fdt}) } // If Columns is empty, then get the _exists list (via All()), diff --git a/server/grpc_test.go b/server/grpc_test.go index 2009bbfc6..e143c9168 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -16,9 +16,13 @@ package server_test import ( "context" + "fmt" + "reflect" + "strconv" "testing" "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/pql" pb "github.com/pilosa/pilosa/v2/proto" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" @@ -338,7 +342,6 @@ func TestQueryPQLUnary(t *testing.T) { i := m.MustCreateIndex(t, "i", pilosa.IndexOptions{}) m.MustCreateField(t, i.Name(), "f", pilosa.OptFieldKeys()) - ctx := context.Background() gh := server.NewGRPCHandler(m.API) @@ -361,3 +364,579 @@ func TestQueryPQLUnary(t *testing.T) { t.Fatalf("expected error: InvalidArgument, got: %v", err) } } + +type ( + tableResponse struct { + headers []columnInfo + rows []row + } + columnInfo struct { + name string + datatype string + } + row struct { + columns []columnResponse + } + columnResponse interface{} +) + +func TestQuerySQLUnary(t *testing.T) { + + ctx := context.Background() + gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t) + defer tearDownFunc() + + tests := []struct { + sql string + exp tableResponse + eq func(tableResponse, tableResponse) error + }{ + { + // Extract(Limit(All(), limit=100, offset=0),Rows(age)) + sql: "select age from grouper", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + }, + rows: []row{ + {[]columnResponse{int64(27)}}, + {[]columnResponse{int64(16)}}, + {[]columnResponse{int64(19)}}, + {[]columnResponse{int64(27)}}, + {[]columnResponse{int64(16)}}, + {[]columnResponse{int64(34)}}, + {[]columnResponse{int64(27)}}, + {[]columnResponse{int64(16)}}, + {[]columnResponse{int64(16)}}, + {[]columnResponse{int64(31)}}, + }, + }, + eq: equal, + }, + { + // Extract(Limit(ConstRow(columns=[2]), limit=100, offset=0),Rows(age),Rows(color),Rows(height),Rows(score)) + sql: "select * from grouper where _id=2", + exp: tableResponse{ + headers: []columnInfo{ + {"_id", "uint64"}, + {"age", "int64"}, + {"color", "[]string"}, + {"height", "int64"}, + {"score", "int64"}, + }, + rows: []row{ + {[]columnResponse{uint64(2), int64(16), []string{"blue"}, int64(30), int64(-8)}}, + }, + }, + eq: equal, + }, + // join + { + // Count(Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid'))) + sql: "select count(*) from grouper g INNER JOIN joiner j ON g._id = j.grouperid", + exp: tableResponse{ + headers: []columnInfo{ + {"count(*)", "uint64"}, + }, + rows: []row{ + {[]columnResponse{uint64(8)}}, + }, + }, + eq: equal, + }, + { + // Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid", + exp: tableResponse{ + headers: []columnInfo{{"_id", "uint64"}}, + rows: []row{ + {[]columnResponse{uint64(1)}}, + {[]columnResponse{uint64(2)}}, + {[]columnResponse{uint64(3)}}, + {[]columnResponse{uint64(5)}}, + {[]columnResponse{uint64(6)}}, + {[]columnResponse{uint64(7)}}, + {[]columnResponse{uint64(8)}}, + {[]columnResponse{uint64(9)}}, + }, + }, + eq: equalUnordered, + }, + { + // Intersect(Row(color='red'),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red'", + exp: tableResponse{ + headers: []columnInfo{{"_id", "uint64"}}, + rows: []row{ + {[]columnResponse{uint64(3)}}, + {[]columnResponse{uint64(8)}}, + {[]columnResponse{uint64(9)}}, + }, + }, + eq: equalUnordered, + }, + { + // Intersect(Row(color='red'),Distinct(Row(jointype=2),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and j.jointype = 2", + exp: tableResponse{ + headers: []columnInfo{{"_id", "uint64"}}, + rows: []row{ + {[]columnResponse{uint64(3)}}, + {[]columnResponse{uint64(8)}}, + {[]columnResponse{uint64(9)}}, + }, + }, + eq: equalUnordered, + }, + // order by + { + // Distinct(Row(score!=null),index='grouper',field='score') + sql: "select distinct score from grouper order by score asc", + exp: tableResponse{ + headers: []columnInfo{{"score", "int64"}}, + rows: []row{ + {[]columnResponse{int64(-13)}}, + {[]columnResponse{int64(-10)}}, + {[]columnResponse{int64(-8)}}, + {[]columnResponse{int64(-2)}}, + {[]columnResponse{int64(0)}}, + {[]columnResponse{int64(6)}}, + {[]columnResponse{int64(80)}}, + {[]columnResponse{int64(100)}}, + }, + }, + eq: equal, + }, + { + // Distinct(Row(score!=null),index='grouper',field='score') + sql: "select distinct score from grouper order by score desc", + exp: tableResponse{ + headers: []columnInfo{{"score", "int64"}}, + rows: []row{ + {[]columnResponse{int64(100)}}, + {[]columnResponse{int64(80)}}, + {[]columnResponse{int64(6)}}, + {[]columnResponse{int64(0)}}, + {[]columnResponse{int64(-2)}}, + {[]columnResponse{int64(-8)}}, + {[]columnResponse{int64(-10)}}, + {[]columnResponse{int64(-13)}}, + }, + }, + eq: equal, + }, + { + // Distinct(Row(score!=null),index='grouper',field='score') + sql: "select distinct score from grouper order by score asc limit 5", + exp: tableResponse{ + headers: []columnInfo{{"score", "int64"}}, + rows: []row{ + {[]columnResponse{int64(-13)}}, + {[]columnResponse{int64(-10)}}, + {[]columnResponse{int64(-8)}}, + {[]columnResponse{int64(-2)}}, + {[]columnResponse{int64(0)}}, + }, + }, + eq: equal, + }, + + { + // Distinct(Row(score!=null),index='grouper',field='score') + sql: "select distinct score from grouper order by score desc limit 5", + exp: tableResponse{ + headers: []columnInfo{{"score", "int64"}}, + rows: []row{ + {[]columnResponse{int64(100)}}, + {[]columnResponse{int64(80)}}, + {[]columnResponse{int64(6)}}, + {[]columnResponse{int64(0)}}, + {[]columnResponse{int64(-2)}}, + }, + }, + eq: equal, + }, + + // distinct + { + // Distinct(Row(score!=null),index='grouper',field='score') + sql: "select distinct score from grouper", + exp: tableResponse{ + headers: []columnInfo{{"score", "int64"}}, + rows: []row{ + {[]columnResponse{int64(-13)}}, + {[]columnResponse{int64(-10)}}, + {[]columnResponse{int64(-8)}}, + {[]columnResponse{int64(-2)}}, + {[]columnResponse{int64(0)}}, + {[]columnResponse{int64(6)}}, + {[]columnResponse{int64(80)}}, + {[]columnResponse{int64(100)}}, + }, + }, + eq: equalUnordered, + }, + { + + // Distinct(Row(height!=null),index='grouper',field='height') + sql: "select distinct height from grouper", + exp: tableResponse{ + headers: []columnInfo{{"height", "int64"}}, + rows: []row{ + {[]columnResponse{int64(20)}}, + {[]columnResponse{int64(30)}}, + {[]columnResponse{int64(40)}}, + {[]columnResponse{int64(50)}}, + {[]columnResponse{int64(60)}}, + {[]columnResponse{int64(70)}}, + {[]columnResponse{int64(80)}}, + {[]columnResponse{int64(90)}}, + {[]columnResponse{int64(100)}}, + {[]columnResponse{int64(110)}}, + }, + }, + eq: equalUnordered, + }, + + // groupby + { + // GroupBy(Rows(field='age'),limit=100) + sql: "select age as yrs, count(*) as cnt from grouper group by age", + exp: tableResponse{ + headers: []columnInfo{ + {"yrs", "int64"}, + {"cnt", "uint64"}, + }, + rows: []row{ + {[]columnResponse{int64(16), uint64(4)}}, + {[]columnResponse{int64(19), uint64(1)}}, + {[]columnResponse{int64(27), uint64(3)}}, + {[]columnResponse{int64(31), uint64(1)}}, + {[]columnResponse{int64(34), uint64(1)}}, + }, + }, + eq: equalUnordered, + }, + { + // GroupBy(Rows(field='age'),Rows(field='color'),limit=100) + sql: "select age, color, count(*) from grouper group by age, color", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + {"color", "string"}, + {"count(*)", "uint64"}, + }, + rows: []row{ + {[]columnResponse{int64(16), "blue", uint64(2)}}, + {[]columnResponse{int64(16), "red", uint64(2)}}, + {[]columnResponse{int64(19), "red", uint64(1)}}, + {[]columnResponse{int64(27), "blue", uint64(2)}}, + {[]columnResponse{int64(27), "green", uint64(1)}}, + {[]columnResponse{int64(31), "red", uint64(1)}}, + {[]columnResponse{int64(34), "blue", uint64(1)}}, + }, + }, + eq: equalUnordered, + }, + { + // GroupBy(Rows(field='age'),Rows(field='color'),limit=100,filter=Row(age=27),aggregate=Sum(field='height')) + sql: "select age, color, sum(height) from grouper where age = 27 group by age, color", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + {"color", "string"}, + {"sum(height)", "int64"}, + }, + rows: []row{ + {[]columnResponse{int64(27), "blue", int64(100)}}, + {[]columnResponse{int64(27), "green", int64(50)}}, + }, + }, + eq: equalUnordered, + }, + { + // GroupBy(Rows(field='age'),limit=100,having=Condition(count>1)) + sql: "select age, count(*) from grouper group by age having count > 1", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + {"count(*)", "uint64"}, + }, + rows: []row{ + {[]columnResponse{int64(16), uint64(4)}}, + {[]columnResponse{int64(27), uint64(3)}}, + }, + }, + eq: equalUnordered, + }, + { + // GroupBy(Rows(field='age'),limit=100,having=Condition(1<=count<=3)) + sql: "select age, count(*) from grouper group by age having count between 1 and 3", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + {"count(*)", "uint64"}, + }, + rows: []row{ + {[]columnResponse{int64(19), uint64(1)}}, + {[]columnResponse{int64(27), uint64(3)}}, + {[]columnResponse{int64(31), uint64(1)}}, + {[]columnResponse{int64(34), uint64(1)}}, + }, + }, + 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", + exp: tableResponse{ + headers: []columnInfo{ + {"age", "int64"}, + {"cnt", "uint64"}, + }, + rows: []row{ + {[]columnResponse{int64(16), uint64(4)}}, + {[]columnResponse{int64(27), uint64(3)}}, + {[]columnResponse{int64(19), uint64(1)}}, + }, + }, + eq: equal, + }, + } + + for i, test := range tests { + t.Run("test-"+strconv.Itoa(i), func(t *testing.T) { + resp, err := gh.QuerySQLUnary(ctx, &pb.QuerySQLRequest{Sql: test.sql}) + if err != nil { + t.Fatalf("sql: %s, error: %v", test.sql, err) + } else { + tr := toTableResponse(resp) + if err := test.eq(test.exp, tr); err != nil { + t.Fatalf("sql: %s, error: %+v", test.sql, err) + } + } + }) + } +} + +func setUpTestQuerySQLUnary(ctx context.Context, t *testing.T) (gh *server.GRPCHandler, tearDownFunc func()) { + t.Helper() + + m := test.RunCommand(t) + gh = server.NewGRPCHandler(m.API) + + // grouper + grouper := m.MustCreateIndex(t, "grouper", pilosa.IndexOptions{Keys: false, TrackExistence: true}) + m.MustCreateField(t, grouper.Name(), "color", pilosa.OptFieldKeys()) + for id, color := range map[int]string{ + 1: "blue", + 2: "blue", + 5: "blue", + 6: "blue", + 7: "blue", + 3: "red", + 8: "red", + 9: "red", + 10: "red", + 4: "green", + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: grouper.Name(), + Pql: fmt.Sprintf(`Set(%d, color="%s")`, id, color), + }); err != nil { + t.Fatal(err) + } + } + m.MustCreateField(t, grouper.Name(), "score", pilosa.OptFieldTypeInt(-1000, 1000)) + for id, score := range map[int]int{ + 1: -10, + 2: -8, + 3: 6, + 4: 0, + 5: -2, + 6: 100, + 7: 0, + 8: -13, + 9: 80, + 10: -2, + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: grouper.Name(), + Pql: fmt.Sprintf(`Set(%d, score=%d)`, id, score), + }); err != nil { + t.Fatal(err) + } + } + m.MustCreateField(t, grouper.Name(), "age", pilosa.OptFieldTypeInt(0, 100)) + for id, age := range map[int]int{ + 2: 16, + 5: 16, + 8: 16, + 9: 16, + 3: 19, + 1: 27, + 4: 27, + 7: 27, + 10: 31, + 6: 34, + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: grouper.Name(), + Pql: fmt.Sprintf(`Set(%d, age=%d)`, id, age), + }); err != nil { + t.Fatal(err) + } + } + + m.MustCreateField(t, grouper.Name(), "height", pilosa.OptFieldTypeInt(0, 1000)) + for id, height := range map[int]int{ + 1: 20, + 2: 30, + 3: 40, + 4: 50, + 5: 60, + 6: 70, + 7: 80, + 8: 90, + 9: 100, + 10: 110, + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: grouper.Name(), + Pql: fmt.Sprintf(`Set(%d, height=%d)`, id, height), + }); err != nil { + t.Fatal(err) + } + } + + // joiner + joiner := m.MustCreateIndex(t, "joiner", pilosa.IndexOptions{TrackExistence: true}) + m.MustCreateField(t, joiner.Name(), "grouperid", pilosa.OptFieldTypeInt(0, 1000), pilosa.OptFieldForeignIndex(grouper.Name())) + m.MustCreateField(t, joiner.Name(), "jointype", pilosa.OptFieldTypeInt(-1000, 1000)) + for id, grouperid := range map[int]int{ + 1: 1, + 2: 2, + 3: 5, + 4: 6, + 5: 7, + 6: 3, + 7: 8, + 8: 9, + 9: 1, + 10: 2, + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: joiner.Name(), + Pql: fmt.Sprintf(`Set(%d, grouperid=%d)`, id, grouperid), + }); err != nil { + t.Fatal(err) + } + } + for id, jointype := range map[int]int{ + 1: 1, + 2: 1, + 3: 1, + 4: 1, + 5: 1, + 6: 2, + 7: 2, + 8: 2, + 9: 3, + 10: 3, + } { + if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ + Index: joiner.Name(), + Pql: fmt.Sprintf(`Set(%d, jointype=%d)`, id, jointype), + }); err != nil { + t.Fatal(err) + } + } + + return gh, func() { + if err := m.API.DeleteIndex(ctx, joiner.Name()); err != nil { + panic(err) + } + if err := m.API.DeleteIndex(ctx, grouper.Name()); err != nil { + panic(err) + } + if err := m.Close(); err != nil { + panic(err) + } + } +} + +func toTableResponse(resp *pb.TableResponse) tableResponse { + tr := tableResponse{ + headers: make([]columnInfo, len(resp.Headers)), + rows: make([]row, len(resp.Rows)), + } + + for i, h := range resp.Headers { + tr.headers[i] = columnInfo{ + name: h.Name, + datatype: h.Datatype, + } + } + + for i, r := range resp.Rows { + tr.rows[i].columns = make([]columnResponse, len(r.Columns)) + for j, c := range r.Columns { + + switch v := c.GetColumnVal().(type) { + case *pb.ColumnResponse_StringVal: + tr.rows[i].columns[j] = v.StringVal + case *pb.ColumnResponse_Uint64Val: + tr.rows[i].columns[j] = v.Uint64Val + case *pb.ColumnResponse_Int64Val: + tr.rows[i].columns[j] = v.Int64Val + case *pb.ColumnResponse_BoolVal: + tr.rows[i].columns[j] = v.BoolVal + case *pb.ColumnResponse_BlobVal: + tr.rows[i].columns[j] = v.BlobVal + case *pb.ColumnResponse_Uint64ArrayVal: + tr.rows[i].columns[j] = v.Uint64ArrayVal.Vals + case *pb.ColumnResponse_StringArrayVal: + tr.rows[i].columns[j] = v.StringArrayVal.Vals + case *pb.ColumnResponse_Float64Val: + tr.rows[i].columns[j] = v.Float64Val + case *pb.ColumnResponse_DecimalVal: + tr.rows[i].columns[j] = pql.NewDecimal(v.DecimalVal.Value, v.DecimalVal.Scale) + default: + tr.rows[i].columns[j] = nil + } + } + } + + return tr +} + +func equal(exp tableResponse, got tableResponse) error { + if !reflect.DeepEqual(exp, got) { + return fmt.Errorf("got: %+v %[1]T, but expected: %+v", got, exp) + } + return nil +} + +func equalUnordered(exp tableResponse, got tableResponse) error { + if len(exp.headers) != len(got.headers) || !reflect.DeepEqual(exp.headers, got.headers) { + return fmt.Errorf("header does not match: got %+v, but expected %+v", got.headers, exp.headers) + } + + if len(exp.rows) != len(got.rows) { + return fmt.Errorf("rows count does not match: got %+v, but expected %+v", len(got.rows), len(exp.rows)) + } + for _, er := range exp.rows { + for j, gr := range got.rows { + if reflect.DeepEqual(er.columns, gr.columns) { + got.rows[j] = got.rows[len(got.rows)-1] + got.rows = got.rows[:len(got.rows)-1] + break + } + } + } + if len(got.rows) > 0 { + return fmt.Errorf("got incorrect rows: %+v", got.rows) + } + return nil +} diff --git a/sql/column.go b/sql/column.go new file mode 100644 index 000000000..1bf8f8885 --- /dev/null +++ b/sql/column.go @@ -0,0 +1,112 @@ +// 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 ( + "strings" + "time" +) + +const ColID = "_id" + +type FuncName string + +const ( + FuncCount FuncName = "count" + FuncMin FuncName = "min" + FuncMax FuncName = "max" + FuncSum FuncName = "sum" + FuncAvg FuncName = "avg" +) + +// Column is an interface which supports mapping to +// source headers, and aliasing column names. +// Alias() should always return a value; +// either a unique alias, or the same value +// return by Name(), but never an empty string. +type Column interface { + Source() string + Name() string + Alias() string +} + +type BasicColumn struct { + source string + name string + alias string +} + +func NewBasicColumn(s, n, a string) *BasicColumn { + return &BasicColumn{ + source: s, + name: n, + alias: a, + } +} + +func (b *BasicColumn) Source() string { + return b.source +} +func (b *BasicColumn) Name() string { + return b.name +} +func (b *BasicColumn) Alias() string { + if b.alias != "" { + return b.alias + } + return b.name +} + +type StarColumn struct{} + +func NewStarColumn() *StarColumn { + return &StarColumn{} +} + +func (s *StarColumn) Source() string { + return "" +} +func (s *StarColumn) Name() string { + return "" +} +func (s *StarColumn) Alias() string { + return "" +} + +func ConvertToTime(text string) (time.Time, bool) { + timeFormats := []string{ + "2006-01-02 15:04:05", + "2006-01-02 15:04", + "2006-01-02", + } + for _, timeFormat := range timeFormats { + if t, err := time.Parse(timeFormat, text); err == nil { + return t, true + } + } + return time.Now(), false +} + +func ExtractFieldName(columName string) (fieldName string, isSpecial bool) { + isSpecial = false + fieldName = columName + if strings.HasPrefix(columName, "_") { + isSpecial = true + if strings.HasSuffix(columName, "_time") { + fieldName = columName[1 : len(columName)-5] + } + } + return +} diff --git a/sql/extract.go b/sql/extract.go new file mode 100644 index 000000000..8ea1c00cf --- /dev/null +++ b/sql/extract.go @@ -0,0 +1,1177 @@ +// 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 ( + "fmt" + "reflect" + "strconv" + "strings" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/pql" + "github.com/pkg/errors" + "vitess.io/vitess/go/vt/sqlparser" +) + +// parseColumn is a column parsed from a sql query. Its qualifier +// value should map to either the name or alias of a parseTable. +type parseColumn struct { + name string + qualifier string +} + +// parseTable is a table parsed from a sql query. It includes +// its name and alias, along with a boolean indicating whether +// the table is the primary side of a join statement (i.e. it +// refers to the Pilosa column _id). +type parseTable struct { + name string + alias string + primary bool // indicates the side of the join representing the column _id + column *parseColumn // contains the related column from the ON clause + index *pilosa.Index // the pilosa index related to this table +} + +// parseTables is a slice of parseTable parsed from a single sql query. +type parseTables []*parseTable + +// byName returns the parseTable from the slice which matches on name. +// If there is no match it returns nil. +func (j parseTables) byName(n string) *parseTable { + for i := range j { + if j[i].name == n { + return j[i] + } + } + return nil +} + +// byName returns the parseTable from the slice which matches on alias. +// If there is no match it returns nil. +func (j parseTables) byAlias(a string) *parseTable { + for i := range j { + if j[i].alias == a { + return j[i] + } + } + return nil +} + +// primary returns the primary parseTable from the slice. +// In order for this to be useful, it is assumed that +// joinTables contains exactly two joinTable pointers +// (a primary and a secondary). +func (j parseTables) primary() *parseTable { + for i := range j { + if j[i].primary { + return j[i] + } + } + return nil +} + +// secondary returns the secondary parseTable from the slice. +func (j parseTables) secondary() *parseTable { + for i := range j { + if !j[i].primary { + return j[i] + } + } + return nil +} + +// tableWhere represents a parseTable from a sql query along +// with the portion of the where clause that relates to +// that table. For example, if a sql query had: +// from tbl1, tbl2 +// where tbl1.field1=1 and tbl2.field2=2 +// then each table would have a separate tableWhere object +// with the where made up of only the field with matching qualifier. +type tableWhere struct { + table *parseTable + where string +} + +// tableWheres is a slice of tableWhere. +type tableWheres []*tableWhere + +// extractParseTable returns a parseTable for the sqlparser.TableExpr. +func extractParseTable(tableExpr sqlparser.TableExpr) (*parseTable, error) { + switch tbl := tableExpr.(type) { + case *sqlparser.AliasedTableExpr: + tableName := tbl.Expr.(sqlparser.TableName).ToViewName().Name.String() + alias := tbl.As.String() + if alias == "" { + alias = tableName + } + return &parseTable{ + name: tableName, + alias: alias, + }, nil + } + + return nil, errors.New("unsupported table expression") +} + +func extractSelectFields(index *pilosa.Index, stmt *sqlparser.Select) ([]Column, selectFeatures, error) { + columns := []Column{} + features := selectFeatures{} + for _, item := range stmt.SelectExprs { + switch expr := item.(type) { + case *sqlparser.AliasedExpr: + var column Column + var alias string = expr.As.String() + switch colExpr := expr.Expr.(type) { + case *sqlparser.ColName: + fieldName := colExpr.Name.String() + + if fieldName == ColID { + if index.Options().Keys { + column = NewKeyIndexColumn(index, alias) + } else { + column = NewIDIndexColumn(index, alias) + } + } else { + column = NewFieldColumn(index.Field(fieldName), alias) + } + case *sqlparser.FuncExpr: + funcName := FuncName(strings.ToLower(colExpr.Name.String())) + var field *pilosa.Field + + if len(colExpr.Exprs) != 1 { + return nil, features, errors.New("function should have a single argument") + } + + switch expr := colExpr.Exprs[0].(type) { + case *sqlparser.AliasedExpr: + if colExpr, ok := expr.Expr.(*sqlparser.ColName); ok { + fieldName := colExpr.Name.String() + field = index.Field(fieldName) + } else { + return nil, features, errors.New("table name is required") + } + case *sqlparser.StarExpr: + // We don't currently track this; it either has a field or doesn't. + default: + return nil, features, errors.New("table name is required") + } + + switch funcName { + case FuncCount, FuncMin, FuncMax, FuncSum, FuncAvg: + column = NewFuncColumn(funcName, field, alias) + default: + return nil, features, fmt.Errorf("unknown function: %s", funcName) + } + features.funcs = append(features.funcs, selectFunc{ + funcName: funcName, + field: field, + }) + default: + return nil, features, errors.New("table name is required") + } + columns = append(columns, column) + case *sqlparser.StarExpr: + columns = append(columns, NewStarColumn()) + default: + return nil, features, errors.New("only column names or * are supported in select") + } + } + return columns, features, nil +} + +func extractIndexName(stmt *sqlparser.Select) (string, error) { + if len(stmt.From) != 1 { + return "", errors.New("selecting from multiple tables is not supported") + } + + fromExpr := stmt.From[0] + switch from := fromExpr.(type) { + case *sqlparser.AliasedTableExpr: + indexName := from.Expr.(sqlparser.TableName).ToViewName().Name.String() + return indexName, nil + } + + return "", errors.New("unsupported from clause") +} + +// extractParseColumn returns a parseColumn for the sqlparser.ColName. +func extractParseColumn(col *sqlparser.ColName) (*parseColumn, error) { + colName := col.Name.String() + qualifier := col.Qualifier.ToViewName().Name.String() + + return &parseColumn{ + name: colName, + qualifier: qualifier, + }, nil +} + +func extractWhere(index *pilosa.Index, expr sqlparser.Expr) (string, error) { + switch e := expr.(type) { + case *sqlparser.ComparisonExpr: + parseCol, op, val, err := extractComparison(e) + if err != nil { + return "", err + } + + if parseCol.name == "_id" { + switch op { + case "=": + return ConstRow(val), nil + case "in": + switch valExpr := val.(type) { + case []interface{}: + return ConstRow(valExpr...), nil + } + } + } + + field := index.Field(parseCol.name) + if field == nil { + return "", pilosa.ErrFieldNotFound + } + + switch field.Type() { + case pilosa.FieldTypeInt, pilosa.FieldTypeDecimal: + switch op { + case "=": + return Equals(field.Name(), val), nil + case "<": + return LT(field.Name(), val), nil + case "<=": + return LTE(field.Name(), val), nil + case ">": + return GT(field.Name(), val), nil + case ">=": + return GTE(field.Name(), val), nil + case "<>", "!=": + return NotEquals(field.Name(), val), nil + } + default: + switch op { + case "=": + return Row(field.Name(), val) + case "in": + var qs []string + switch valExpr := val.(type) { + case []interface{}: + for _, v := range valExpr { + q, err := Row(field.Name(), v) + if err != nil { + return "", err + } + qs = append(qs, q) + } + return Union(qs...), nil + default: + return "", fmt.Errorf("in operator expects `[]interface{}` but got: %T", valExpr) + } + case "like": + sval, ok := val.(string) + if !ok { + return "", fmt.Errorf("like operator expects `string` but got: %T", val) + } + return Like(field.Name(), sval), nil + } + } + case *sqlparser.AndExpr: + pql, err := extractWhereDateRange(index, e.Left, e.Right) + if err == nil { + return pql, err + } + left, err := extractWhere(index, e.Left) + if err != nil { + return "", err + } + right, err := extractWhere(index, e.Right) + if err != nil { + return "", err + } + return Intersect(left, right), nil + case *sqlparser.OrExpr: + left, err := extractWhere(index, e.Left) + if err != nil { + return "", err + } + right, err := extractWhere(index, e.Right) + if err != nil { + return "", err + } + return Union(left, right), nil + case *sqlparser.NotExpr: + expr, err := extractWhere(index, e.Expr) + if err != nil { + return "", err + } + return Not(expr), nil + case *sqlparser.ParenExpr: + expr, err := extractWhere(index, e.Expr) + if err != nil { + return "", err + } + return expr, nil + case *sqlparser.RangeCond: + if e.Operator != "between" { + return "", errors.New("only between is supported") + } + left, ok := e.Left.(*sqlparser.ColName) + if !ok { + return "", errors.New("left operand must be a column name") + } + columnName := left.Name.String() + fieldName, isSpecial := ExtractFieldName(columnName) + if isSpecial { + return "", errors.New("special fields are not allowed here") + } + field := index.Field(fieldName) + + switch field.Type() { + case pilosa.FieldTypeInt: + fromNum, err := extractInt(e.From) + if err != nil { + return "", err + } + toNum, err := extractInt(e.To) + if err != nil { + return "", err + } + return Between(field.Name(), fromNum, toNum), nil + case pilosa.FieldTypeDecimal: + fromNum, err := extractFloat(e.From) + if err != nil { + return "", err + } + toNum, err := extractFloat(e.To) + if err != nil { + return "", err + } + return Between(field.Name(), fromNum, toNum), nil + default: + return "", errors.New("only int and float64 fields are supported") + } + case *sqlparser.IsExpr: + left, ok := e.Expr.(*sqlparser.ColName) + if !ok { + return "", errors.New("left operand must be a column name") + } + field := index.Field(left.Name.String()) + if field.Type() == pilosa.FieldTypeInt { + if e.Operator == "is not null" { + return NotNull(field.Name()), nil + } + return "", errors.New("only `is not null` is supported for int fields") + } + return "", errors.New("`is` expression is supported only for int fields") + } + return "", errors.New("cannot extract where") +} + +func extractWhereDateRange(index *pilosa.Index, leftExpr sqlparser.Expr, rightExpr sqlparser.Expr) (string, error) { + var fieldName string + var fromExpr sqlparser.Expr + var toExpr sqlparser.Expr + var isSpecial bool + var idOrKey interface{} + + // Where part is expected to be in one of the following forms: + // where FIELD=VALUE and FIELD between DATETIME_FORMAT and DATETIME_FORMAT + // Or: + // where FIELD between DATETIME_FORMAT and DATETIME_FORMAT and FIELD=VALUE + + extract := func(compExpr *sqlparser.ComparisonExpr, rangeCond *sqlparser.RangeCond) error { + parseCol, op, val, err := extractComparison(compExpr) + if err != nil { + return err + } + idOrKey = val + if op != "=" { + // only = operator can exist here. + return errors.New("only = operator can exist here") + } + fieldName, isSpecial = ExtractFieldName(parseCol.name) + if isSpecial { + // column name cannot be special here. + return errors.New("column name cannot be special here") + } + if rangeCond.Operator != "between" { + // only between is accepted at this point. + return errors.New("only between is accepted at this point") + } + condFieldName, _ := ExtractFieldName(rangeCond.Left.(*sqlparser.ColName).Name.String()) + if condFieldName != fieldName { + // the field names in both sides should be the same, otherwise reject. + return errors.New("the field names in both sides should be the same, otherwise reject") + } + fromExpr = rangeCond.From + toExpr = rangeCond.To + return nil + } + + if left, ok := leftExpr.(*sqlparser.ComparisonExpr); ok { + // if FIELD=VALUE part is on the left, and range part is on the right + if right, ok := rightExpr.(*sqlparser.RangeCond); ok { + if err := extract(left, right); err != nil { + return "", err + } + } else { + return "", errors.New("right is not a range cond") + } + } else if right, ok := rightExpr.(*sqlparser.ComparisonExpr); ok { + // if FIELD=VALUE part is on the right, and left part is on the left + if left, ok := leftExpr.(*sqlparser.RangeCond); ok { + if err := extract(right, left); err != nil { + return "", err + } + } else { + return "", errors.New("left is not a range cond") + } + } + + fromStr, err := extractStr(fromExpr) + if err != nil { + return "", err + } + toStr, err := extractStr(toExpr) + if err != nil { + return "", err + } + // try to convert `from` and `to` to time + fromTime, ok := ConvertToTime(fromStr) + if !ok { + return "", errors.New("from operand must be in the correct time format") + } + toTime, ok := ConvertToTime(toStr) + if !ok { + return "", errors.New("to operand must be in the correct time format") + } + + field := index.Field(fieldName) + return RowRange(field.Name(), idOrKey, fromTime, toTime) +} + +func extractVal(e sqlparser.Expr) (interface{}, error) { + val, ok := e.(*sqlparser.SQLVal) + if !ok { + return nil, errors.New("expression must be a value") + } + switch val.Type { + case sqlparser.StrVal: + return string(val.Val), nil + case sqlparser.IntVal: + value, err := strconv.Atoi(string(val.Val)) + if err != nil { + return nil, err + } + return value, nil + case sqlparser.FloatVal: + value, err := strconv.ParseFloat(string(val.Val), 64) + if err != nil { + return nil, err + } + return value, nil + default: + return nil, fmt.Errorf("unknown type: %d", val.Type) + } +} + +func extractInt(e sqlparser.Expr) (int, error) { + val, err := extractVal(e) + if err != nil { + return 0, err + } + num, ok := val.(int) + if !ok { + return 0, errors.New("value must be an integer") + } + return num, nil +} + +func extractFloat(e sqlparser.Expr) (float64, error) { + val, err := extractVal(e) + if err != nil { + return 0, err + } + + var num float64 + switch v := val.(type) { + case int: + num = float64(v) + case float64: + num = v + default: + return 0, errors.New("value must be convertable to a float64") + } + return num, nil +} + +func extractStr(e sqlparser.Expr) (string, error) { + val, err := extractVal(e) + if err != nil { + return "", err + } + s, ok := val.(string) + if !ok { + return "", errors.New("value must be a string") + } + return s, nil +} + +func extractTuple(tuple sqlparser.ValTuple) ([]interface{}, error) { + result := []interface{}{} + for _, item := range tuple { + if val, ok := item.(*sqlparser.SQLVal); ok { + v, err := extractVal(val) + if err != nil { + return nil, err + } + result = append(result, v) + } else { + return nil, errors.New("tuple should contain integers or strings") + } + } + return result, nil +} + +func extractComparison(expr *sqlparser.ComparisonExpr) (col *parseColumn, op string, value interface{}, err error) { + op = expr.Operator + if colExpr, ok := expr.Left.(*sqlparser.ColName); ok { + col, err = extractParseColumn(colExpr) + if err != nil { + return + } + if op == "in" { + switch valExpr := expr.Right.(type) { + case sqlparser.ValTuple: + value, err = extractTuple(valExpr) + + default: + err = fmt.Errorf("in operator excepts only a tuple or a query, received `%s`", + reflect.TypeOf(valExpr).String()) + return + } + } else { + value, err = extractVal(expr.Right) + } + if err != nil { + return + } + } else { + if colExpr, ok := expr.Right.(*sqlparser.ColName); ok { + col, err = extractParseColumn(colExpr) + if err != nil { + return + } + if op == "in" { + value, err = extractTuple(expr.Left.(sqlparser.ValTuple)) + } else { + value, err = extractVal(expr.Left) + } + if err != nil { + return + } + } else { + err = errors.New("either left or right operand should be a column name") + } + } + return +} + +func extractLimitOffset(stmt *sqlparser.Select) (uint, uint, error) { + if stmt.Limit == nil { + return 100, 0, nil + } + var offset, limit uint + if offsetExpr, ok := stmt.Limit.Offset.(*sqlparser.SQLVal); ok { + val, err := extractVal(offsetExpr) + if err != nil { + return 0, 0, err + } + if offsetVal, ok := val.(int); ok { + offset = uint(offsetVal) + } else { + return 0, 0, errors.New("offset must be an integer") + } + } + if limitExpr, ok := stmt.Limit.Rowcount.(*sqlparser.SQLVal); ok { + val, err := extractVal(limitExpr) + if err != nil { + return 0, 0, err + } + if limitVal, ok := val.(int); ok { + limit = uint(limitVal) + } else { + return 0, 0, errors.New("limit must be an integer") + } + } + return limit, offset, nil +} + +// extractOrderBy returns the order by fields and directions (asc/desc) +// as separate string slices. +func extractOrderBy(stmt *sqlparser.Select) ([]string, []string, error) { + var flds []string + var dirs []string + + for _, item := range stmt.OrderBy { + switch colExpr := item.Expr.(type) { + case *sqlparser.ColName: + colName := colExpr.Name.String() + flds = append(flds, colName) + dirs = append(dirs, item.Direction) + } + } + + return flds, dirs, nil +} + +func extractTableNames(stmt *sqlparser.Select) ([]string, error) { + if len(stmt.From) == 0 { + return []string{}, nil + } + + switch from := stmt.From[0].(type) { + case *sqlparser.AliasedTableExpr: + tableName := from.Expr.(sqlparser.TableName).ToViewName().Name.String() + return []string{tableName}, nil + case *sqlparser.JoinTableExpr: + ret := []string{} + switch left := from.LeftExpr.(type) { + case *sqlparser.AliasedTableExpr: + leftTableName := left.Expr.(sqlparser.TableName).ToViewName().Name.String() + ret = append(ret, leftTableName) + } + switch right := from.RightExpr.(type) { + case *sqlparser.AliasedTableExpr: + rightTableName := right.Expr.(sqlparser.TableName).ToViewName().Name.String() + ret = append(ret, rightTableName) + } + return ret, nil + } + + return []string{}, nil +} + +func extractGroupByFieldNames(stmt sqlparser.GroupBy) ([]string, error) { + fields := make([]string, len(stmt)) + for i, item := range stmt { + col, ok := item.(*sqlparser.ColName) + if !ok { + return nil, errors.New("group by accepts columns") + } + fields[i] = col.Name.String() + } + return fields, nil +} + +func extractHavingClause(stmt *sqlparser.Where) (*HavingClause, error) { + if stmt == nil { + return nil, nil + } + if stmt.Type != "having" { + return nil, fmt.Errorf("invalid having type: %s", stmt.Type) + } + + hc := &HavingClause{} + + switch having := stmt.Expr.(type) { + case *sqlparser.RangeCond: + if having.Operator != "between" { + return nil, errors.New("only between is supported") + } + hc.Subj = having.Left.(*sqlparser.ColName).Name.String() + hc.Cond.Op = pql.BETWEEN + fromPred, err := extractInt(having.From) + if err != nil { + return nil, err + } + toPred, err := extractInt(having.To) + if err != nil { + return nil, err + } + vals := make([]interface{}, 2) + switch hc.Subj { + case "count": + vals[0] = uint64(fromPred) + vals[1] = uint64(toPred) + case "sum": + vals[0] = int64(fromPred) + vals[1] = int64(toPred) + } + hc.Cond.Value = vals + return hc, nil + case *sqlparser.AndExpr: + left := having.Left.(*sqlparser.ComparisonExpr) + right := having.Right.(*sqlparser.ComparisonExpr) + leftName := left.Left.(*sqlparser.ColName).Name.String() + rightName := right.Left.(*sqlparser.ColName).Name.String() + if leftName != rightName { + return nil, fmt.Errorf("having comparitors do not match: %s, %s", leftName, rightName) + } + hc.Subj = leftName + + leftOp := extractComparisonOp(left) + rightOp := extractComparisonOp(right) + + leftPred, err := extractInt(left.Right) + if err != nil { + return nil, err + } + rightPred, err := extractInt(right.Right) + if err != nil { + return nil, err + } + + intVals := make([]int, 2) + if leftOp == pql.GT && rightOp == pql.LT { + intVals[0] = leftPred + intVals[1] = rightPred + hc.Cond.Op = pql.BTWN_LT_LT + } else if leftOp == pql.GT && rightOp == pql.LTE { + intVals[0] = leftPred + intVals[1] = rightPred + hc.Cond.Op = pql.BTWN_LT_LTE + } else if leftOp == pql.GTE && rightOp == pql.LT { + intVals[0] = leftPred + intVals[1] = rightPred + hc.Cond.Op = pql.BTWN_LTE_LT + } else if leftOp == pql.GTE && rightOp == pql.LTE { + intVals[0] = leftPred + intVals[1] = rightPred + hc.Cond.Op = pql.BETWEEN + } else if leftOp == pql.LT && rightOp == pql.GT { + intVals[0] = rightPred + intVals[1] = leftPred + hc.Cond.Op = pql.BTWN_LT_LT + } else if leftOp == pql.LT && rightOp == pql.GTE { + intVals[0] = rightPred + intVals[1] = leftPred + hc.Cond.Op = pql.BTWN_LTE_LT + } else if leftOp == pql.LTE && rightOp == pql.GT { + intVals[0] = rightPred + intVals[1] = leftPred + hc.Cond.Op = pql.BTWN_LT_LTE + } else if leftOp == pql.LTE && rightOp == pql.GTE { + intVals[0] = rightPred + intVals[1] = leftPred + hc.Cond.Op = pql.BETWEEN + } + + vals := make([]interface{}, 2) + switch hc.Subj { + case "count": + vals[0] = uint64(intVals[0]) + vals[1] = uint64(intVals[1]) + case "sum": + vals[0] = int64(intVals[0]) + vals[1] = int64(intVals[1]) + } + hc.Cond.Value = vals + + return hc, nil + case *sqlparser.ComparisonExpr: + hc.Subj = having.Left.(*sqlparser.ColName).Name.String() + hc.Cond.Op = extractComparisonOp(having) + switch hc.Subj { + case "count": + pred, err := extractInt(having.Right) + if err != nil { + return nil, err + } + hc.Cond.Value = uint64(pred) + case "sum": + pred, err := extractInt(having.Right) + if err != nil { + return nil, err + } + hc.Cond.Value = int64(pred) + } + return hc, nil + } + + return nil, errors.New("unsupported having clause") +} + +func extractComparisonOp(expr *sqlparser.ComparisonExpr) pql.Token { + switch expr.Operator { + case "==": + return pql.EQ + case "!=": + return pql.NEQ + case "<": + return pql.LT + case "<=": + return pql.LTE + case ">": + return pql.GT + case ">=": + return pql.GTE + } + return pql.ILLEGAL +} + +// extractJoinTables returns a slice of parseTable containing two +// items, the primary and secondary join tables. This function does +// not extract join tables of the form: +// from tbl1, tbl2 +// The from clause must be of the form: +// from tbl1 INNER JOIN tbl2 ON ... +// +func extractJoinTables(stmt *sqlparser.Select) (parseTables, error) { + if len(stmt.From) != 1 { + return nil, errors.New("selecting from multiple tables is not supported") + } + + tbls := make([]*parseTable, 2) + + from, ok := stmt.From[0].(*sqlparser.JoinTableExpr) + if !ok { + return nil, errors.New("unsupported join clause") + } + + leftTable, err := extractParseTable(from.LeftExpr) + if err != nil { + return nil, errors.Wrap(err, "extracting left join table") + } + rightTable, err := extractParseTable(from.RightExpr) + if err != nil { + return nil, errors.Wrap(err, "extracting right join table") + } + + // It is not important which table goes in which tbls position; + // the primary/secondary table will be determined later. + tbls[0] = leftTable + tbls[1] = rightTable + + // Get the ON condition and determine which table is primary. + switch onCond := from.Condition.On.(type) { + case *sqlparser.ComparisonExpr: + if onCond.Operator != "=" { + return nil, errors.Errorf("unsupported on condition comparison type: %s", onCond.Operator) + } + // get left ColName + left, ok := onCond.Left.(*sqlparser.ColName) + if !ok { + return nil, errors.New("left join operand must be a column name") + } + leftJoinCol, err := extractParseColumn(left) + if err != nil { + return nil, errors.Wrap(err, "extracting left join column") + } + // get right ColName + right, ok := onCond.Right.(*sqlparser.ColName) + if !ok { + return nil, errors.New("right join operand must be a column name") + } + rightJoinCol, err := extractParseColumn(right) + if err != nil { + return nil, errors.Wrap(err, "extracting right join column") + } + + // The primary column is set as the column referencing the "_id" field. + var primaryColumn *parseColumn + if leftJoinCol.name == ColID && rightJoinCol.name != ColID { + primaryColumn = leftJoinCol + } else if leftJoinCol.name != ColID && rightJoinCol.name == ColID { + primaryColumn = rightJoinCol + } else { + return nil, errors.Errorf("exactly one join column must be %s, have: %s, %s", ColID, leftJoinCol.name, rightJoinCol.name) + } + + // populate the joinTable column and primary fields + for _, jc := range []*parseColumn{leftJoinCol, rightJoinCol} { + var found bool + for i := range tbls { + if tbls[i].alias == jc.qualifier { + tbls[i].column = jc + if jc == primaryColumn { + tbls[i].primary = true + } + found = true + } + } + if !found { + return nil, errors.Errorf("no tables match qualifier: %s", jc.qualifier) + } + } + + default: + return nil, errors.Errorf("unsupported on condition type: %T", onCond) + } + + return tbls, nil +} + +// extractWheres returns the slice of tableWhere for the sql query. +func extractWheres(indexes []*pilosa.Index, tbls parseTables, expr sqlparser.Expr) (tableWheres, error) { + wheres := make([]*tableWhere, 0) + + // Set the index associated with each parseTable. + // TODO: may be able to move this to parseTables creation? + for _, idx := range indexes { + tbl := tbls.byName(idx.Name()) + if tbl == nil { + return nil, errors.Errorf("index not in parseTables: %s", idx.Name()) + } + tbl.index = idx + } + + // make a map of tbls alias to slice index. + m := make(map[string]int) + for i, tbl := range tbls { + m[tbl.alias] = i + } + + switch e := expr.(type) { + case *sqlparser.ComparisonExpr: + pCol, op, val, err := extractComparison(e) + if err != nil { + return nil, err + } + + pTable := tbls.byAlias(pCol.qualifier) + if pTable == nil { + return nil, errors.Errorf("no index for qaulifier: %s", pCol.qualifier) + } else if pTable.index == nil { + return nil, errors.Errorf("parse table has no index: %s", pTable.name) + } + + field := pTable.index.Field(pCol.name) + + tw := &tableWhere{ + table: pTable, + } + + if field.Type() == pilosa.FieldTypeInt { + num, ok := val.(int) + if !ok { + return nil, errors.New("right operand must be a number") + } + switch op { + case "=": + tw.where = Equals(field.Name(), num) + case "<": + tw.where = LT(field.Name(), num) + case "<=": + tw.where = LTE(field.Name(), num) + case ">": + tw.where = GT(field.Name(), num) + case ">=": + tw.where = GTE(field.Name(), num) + case "<>": + fallthrough + case "!=": + tw.where = NotEquals(field.Name(), num) + } + return append(wheres, tw), nil + } + if op == "=" { + if tw.where, err = Row(field.Name(), val); err != nil { + return nil, err + } + return append(wheres, tw), nil + } + + if op == "in" { + var qs []string + switch valExpr := val.(type) { + case []interface{}: + for _, v := range valExpr { + r, e := Row(field.Name(), v) + if e != nil { + return nil, errors.Wrap(err, "extracting where statements") + } + qs = append(qs, r) + } + tw.where = Union(qs...) + return append(wheres, tw), nil + + default: + return nil, fmt.Errorf("in operator expects `[]interface{}` but got: %T", valExpr) + } + } + + case *sqlparser.AndExpr: + left, err := extractWheres(indexes, tbls, e.Left) + if err != nil { + return nil, err + } + right, err := extractWheres(indexes, tbls, e.Right) + if err != nil { + return nil, err + } + + // The following logic is used to build the where portion of the query + // related to each table. The goal is to return one or two tableWhere + // objects (either 0 or 1 for each table in the join). + if len(left) == 1 && len(right) == 1 && left[0].table == right[0].table { + // if left(1) and right(1) are from the same alias, + // then intersect them into left and return left(1) + left[0].where = Intersect(left[0].where, right[0].where) + return left, nil + } else if len(left) == 1 && len(right) == 1 && left[0].table != right[0].table { + // if left(1) and right(1) are NOT from the same alias, + // then return final(2) + return []*tableWhere{left[0], right[0]}, nil + } else if len(left) == 1 && len(right) == 2 { + // if left(1) and right(2) + // then intersect the 1's and return final(2) + if left[0].table == right[0].table { + left[0].where = Intersect(left[0].where, right[0].where) + return []*tableWhere{left[0], right[1]}, nil + } else if left[0].table == right[1].table { + left[0].where = Intersect(left[0].where, right[1].where) + return []*tableWhere{left[0], right[0]}, nil + } + return nil, errors.Errorf("no matching table on right: %s", left[0].table.name) + } else if len(left) == 1 && len(right) == 2 { + // if left(2) and right(1), + // then intersect the 1's and return final(2) + if right[0].table == left[0].table { + right[0].where = Intersect(right[0].where, left[0].where) + return []*tableWhere{left[0], right[1]}, nil + } else if right[0].table == left[1].table { + right[0].where = Intersect(right[0].where, left[1].where) + return []*tableWhere{left[0], right[0]}, nil + } + return nil, errors.Errorf("no matching table on left: %s", right[0].table.name) + } else if len(left) == 2 && len(right) == 2 { + // if left(2) and right(2) + // then intsect both and return final(2) + if left[0].table == right[0].table && left[1].table == right[1].table { + left[0].where = Intersect(left[0].where, right[0].where) + left[1].where = Intersect(left[1].where, right[1].where) + return left, nil + } else if left[0].table == right[1].table && left[1].table == right[0].table { + left[0].where = Intersect(left[0].where, right[1].where) + left[1].where = Intersect(left[1].where, right[0].where) + return left, nil + } + return nil, errors.Errorf("non-matching tables: %s/%s, %s/%s", left[0].table.name, left[1].table.name, right[0].table.name, right[1].table.name) + } + return nil, errors.Errorf("invalid table count; expected 1 or 2, but got: %d, %d", len(left), len(right)) + case *sqlparser.OrExpr: + left, err := extractWheres(indexes, tbls, e.Left) + if err != nil { + return nil, err + } + right, err := extractWheres(indexes, tbls, e.Right) + if err != nil { + return nil, err + } + + if len(left) == 1 && len(right) == 1 && left[0].table == right[0].table { + // if left(1) and right(1) are from the same alias, + // then union them and return final(1) + left[0].where = Union(left[0].where, right[0].where) + return left, nil + } + return nil, errors.Errorf("invalid table count; expected 1/1, but got: %d/%d", len(left), len(right)) + case *sqlparser.NotExpr: + expr, err := extractWheres(indexes, tbls, e.Expr) + if err != nil { + return nil, err + } + if len(expr) == 1 { + expr[0].where = Not(expr[0].where) + return expr, nil + } + return nil, errors.Errorf("not support a single expression. got: %d", len(expr)) + case *sqlparser.ParenExpr: + expr, err := extractWheres(indexes, tbls, e.Expr) + if err != nil { + return nil, err + } + if len(expr) == 1 { + return expr, nil + } + return nil, errors.Errorf("not support a single expression. got: %d", len(expr)) + case *sqlparser.RangeCond: + if e.Operator != "between" { + return nil, errors.New("only between is supported") + } + left, ok := e.Left.(*sqlparser.ColName) + if !ok { + return nil, errors.New("left operand must be a column name") + } + + pCol, err := extractParseColumn(left) + if err != nil { + return nil, errors.Wrap(err, "extracting parse column") + } + + pTable := tbls.byAlias(pCol.qualifier) + if pTable == nil { + return nil, errors.Errorf("no index for qaulifier: %s", pCol.qualifier) + } else if pTable.index == nil { + return nil, errors.Errorf("parse table has no index: %s", pTable.name) + } + + fieldName, isSpecial := ExtractFieldName(pCol.name) + if isSpecial { + return nil, errors.New("special fields are not allowed here") + } + + tw := &tableWhere{ + table: pTable, + } + + field := pTable.index.Field(fieldName) + if field.Type() == pilosa.FieldTypeInt { + fromNum, err := extractInt(e.From) + if err != nil { + return nil, err + } + toNum, err := extractInt(e.To) + if err != nil { + return nil, err + } + tw.where = Between(field.Name(), fromNum, toNum) + return append(wheres, tw), nil + } + return nil, errors.New("only int fields are supported") + case *sqlparser.IsExpr: + left, ok := e.Expr.(*sqlparser.ColName) + if !ok { + return nil, errors.New("left operand must be a column name") + } + + pCol, err := extractParseColumn(left) + if err != nil { + return nil, errors.Wrap(err, "extracting parse column") + } + + pTable := tbls.byAlias(pCol.qualifier) + if pTable == nil { + return nil, errors.Errorf("no index for qaulifier: %s", pCol.qualifier) + } else if pTable.index == nil { + return nil, errors.Errorf("parse table has no index: %s", pTable.name) + } + + tw := &tableWhere{ + table: pTable, + } + + field := pTable.index.Field(pCol.name) + if field.Type() == pilosa.FieldTypeInt { + if e.Operator == "is not null" { + tw.where = NotNull(field.Name()) + return append(wheres, tw), nil + } + return nil, errors.New("only `is not null` is supported for int fields") + } + return nil, errors.New("`is` expression is supported only for int fields") + } + return nil, errors.New("cannot extract where") +} diff --git a/sql/mapper.go b/sql/mapper.go new file mode 100644 index 000000000..a43dba06d --- /dev/null +++ b/sql/mapper.go @@ -0,0 +1,113 @@ +// 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 ( + "strings" + + "github.com/pilosa/pilosa/v2/logger" + "github.com/pkg/errors" + "vitess.io/vitess/go/vt/sqlparser" +) + +const ( + SQLTypeSelect = "select" + SQLTypeShow = "show" + SQLTypeEmpty = "" +) + +// System errors. +var ( + ErrMultipleSQLStatements = errors.New("statement contains multiple sql queries") +) + +type Attributes map[string]interface{} + +type MappedSQL struct { + SQLType string + Statement sqlparser.Statement + Mask QueryMask + Tables []string +} + +// Mapper is responsible for mapping a SQL query to structure representation +type Mapper struct { + Logger logger.Logger +} + +func NewMapper() *Mapper { + return &Mapper{ + Logger: logger.NopLogger, + } +} + +// Parse parses SQL query +func (m *Mapper) Parse(sql string) (sqlparser.Statement, QueryMask, error) { + parsed, err := sqlparser.Parse(sql) + if err != nil { + return nil, QueryMask{}, errors.Wrap(err, "parsing sql") + } + + qm := GenerateMask(parsed) + + return parsed, qm, nil +} + +// MapSQL converts a sql string into a MappedSQL object, +// which includes the parsed query and the query mask, +// among other information about the query. +func (m *Mapper) MapSQL(sql string) (*MappedSQL, error) { + // In the case where `sql` contains more than one query—since + // we don't support multiple return sets—we're going to just + // ignore everything and return a specific error type. This + // will allow the caller to handle it as needed (i.e. it can + // return the error, or return an empty result set). + if parts := strings.Split(sql, ";"); len(parts) > 1 { + var partCount int + for _, part := range parts { + if trimmed := strings.TrimSpace(part); trimmed != "" && trimmed != "\x00" { + partCount++ + } + } + if partCount != 1 { + return nil, ErrMultipleSQLStatements + } + } + + stmt, qm, err := m.Parse(sql) + if err != nil { + return nil, errors.Wrap(err, "parsing sql") + } + + var sqlType string + var tableNames []string + switch slct := stmt.(type) { + case *sqlparser.Select: + sqlType = SQLTypeSelect + tableNames, err = extractTableNames(slct) + if err != nil { + return nil, errors.Wrap(err, "extracting table names") + } + case *sqlparser.Show: + sqlType = SQLTypeShow + } + + return &MappedSQL{ + SQLType: sqlType, + Statement: stmt, + Mask: qm, + Tables: tableNames, + }, nil +} diff --git a/sql/mapper_test.go b/sql/mapper_test.go new file mode 100644 index 000000000..216ffda4f --- /dev/null +++ b/sql/mapper_test.go @@ -0,0 +1,394 @@ +// 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 ( + "fmt" + "testing" + + "vitess.io/vitess/go/vt/sqlparser" +) + +func TestParse(t *testing.T) { + tests := []struct { + sql string + expMask QueryMask + }{ + { + sql: "select _id from tbl", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartTable, + }, + }, + { + sql: "select * from tbl", + expMask: QueryMask{ + SelectMask: SelectPartStar, + FromMask: FromPartTable, + }, + }, + { + sql: "select fld from tbl", + expMask: QueryMask{ + SelectMask: SelectPartField, + FromMask: FromPartTable, + }, + }, + { + sql: "select fld1, fld2 from tbl", + expMask: QueryMask{ + SelectMask: SelectPartFields, + FromMask: FromPartTable, + }, + }, + { + sql: "select count(*) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartTable, + }, + }, + { + sql: "select min(fld) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartMinField, + FromMask: FromPartTable, + }, + }, + { + sql: "select max(fld) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartMaxField, + FromMask: FromPartTable, + }, + }, + { + sql: "select sum(fld) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartSumField, + FromMask: FromPartTable, + }, + }, + { + sql: "select avg(fld) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartAvgField, + FromMask: FromPartTable, + }, + }, + { + sql: "select _id, count(*) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartID | SelectPartCountStar, + FromMask: FromPartTable, + }, + }, + { + sql: "select _id from tbl1, tbl2", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartTables, + }, + }, + { + sql: "select _id from tbl1 INNER JOIN tbl2", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartJoin, + }, + }, + { + sql: "select * from tbl where _id = 1", + expMask: QueryMask{ + SelectMask: SelectPartStar, + FromMask: FromPartTable, + WhereMask: WherePartIDCondition, + }, + }, + { + sql: "select * from tbl where fld = 1", + expMask: QueryMask{ + SelectMask: SelectPartStar, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + }, + }, + { + sql: "select * from tbl group by fld", + expMask: QueryMask{ + SelectMask: SelectPartStar, + FromMask: FromPartTable, + GroupByMask: GroupByPartField, + }, + }, + { + sql: "select * from tbl group by fld1, fld2", + expMask: QueryMask{ + SelectMask: SelectPartStar, + FromMask: FromPartTable, + GroupByMask: GroupByPartFields, + }, + }, + { + sql: "select fld, sum(fld) from tbl group by fld", + expMask: QueryMask{ + SelectMask: SelectPartField | SelectPartSumField, + FromMask: FromPartTable, + GroupByMask: GroupByPartField, + }, + }, + { + sql: "select fld, sum(fld) from tbl group by fld having sum > 10", + expMask: QueryMask{ + SelectMask: SelectPartField | SelectPartSumField, + FromMask: FromPartTable, + GroupByMask: GroupByPartField, + HavingMask: HavingPartCondition, + }, + }, + { + sql: "select fld from tbl order by fld", + expMask: QueryMask{ + SelectMask: SelectPartField, + FromMask: FromPartTable, + OrderByMask: OrderByPartField, + }, + }, + { + sql: "select fld from tbl order by fld1, fld2", + expMask: QueryMask{ + SelectMask: SelectPartField, + FromMask: FromPartTable, + OrderByMask: OrderByPartFields, + }, + }, + { + sql: "select fld from tbl limit 10", + expMask: QueryMask{ + SelectMask: SelectPartField, + FromMask: FromPartTable, + LimitMask: LimitPartLimit, + }, + }, + { + sql: "select fld from tbl limit 10, 5", + expMask: QueryMask{ + SelectMask: SelectPartField, + FromMask: FromPartTable, + LimitMask: LimitPartLimit | LimitPartOffset, + }, + }, + { + sql: "select distinct fld from tbl", + expMask: QueryMask{ + SelectMask: SelectPartDistinct | SelectPartField, + FromMask: FromPartTable, + }, + }, + { + sql: "select count(*) from tbl where fld = 1", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + }, + }, + { + sql: "select count(*) from tbl where fld1 = 1 and fld2 = 2", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartTable, + WhereMask: WherePartMultiFieldCondition, + }, + }, + { + sql: "select count(*) from tbl where fld1 = 1 or fld2 = 2", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartTable, + WhereMask: WherePartMultiFieldCondition, + }, + }, + { + sql: "select _id from tbl where not fld = 1 limit 10", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + LimitMask: LimitPartLimit, + }, + }, + { + sql: "select _id from tbl where fld between 1 and 3", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + }, + }, + { + sql: "select _id from tbl where fld1 between 1 and 3 and fld2 = 2", + expMask: QueryMask{ + SelectMask: SelectPartID, + FromMask: FromPartTable, + WhereMask: WherePartMultiFieldCondition, + }, + }, + { + sql: "select count(*) from tbl where fld is not null", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + }, + }, + { + sql: "select fld, count(*) from tbl group by fld", + expMask: QueryMask{ + SelectMask: SelectPartField | SelectPartCountStar, + FromMask: FromPartTable, + GroupByMask: GroupByPartField, + }, + }, + { + sql: "select fld1, fld2, count(*) from grouper group by fld1, fld2", + expMask: QueryMask{ + SelectMask: SelectPartFields | SelectPartCountStar, + FromMask: FromPartTable, + GroupByMask: GroupByPartFields, + }, + }, + { + sql: "select fld1, fld2, count(*) from tbl where fld1 = 1 group by fld1, fld2 limit 1", + expMask: QueryMask{ + SelectMask: SelectPartFields | SelectPartCountStar, + FromMask: FromPartTable, + WhereMask: WherePartFieldCondition, + GroupByMask: GroupByPartFields, + LimitMask: LimitPartLimit, + }, + }, + { + sql: "select count(distinct fld) from tbl", + expMask: QueryMask{ + SelectMask: SelectPartCountDistinctField, + FromMask: FromPartTable, + }, + }, + { + sql: "select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartJoin, + }, + }, + { + sql: "select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where tbl1.fld1 = 1 and tbl2.fld2 = 2", + expMask: QueryMask{ + SelectMask: SelectPartCountStar, + FromMask: FromPartJoin, + WhereMask: WherePartMultiFieldCondition, + }, + }, + } + + mapper := NewMapper() + for i, test := range tests { + t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) { + _, mask, err := mapper.Parse(test.sql) + if err != nil { + t.Fatal(err) + } + + if mask != test.expMask { + t.Fatalf("expected mask: %v, but got: %v", test.expMask, mask) + } + }) + } +} + +// This thing passes even if join is not implemented. +// TODO: make this actually a test +func TestSelectJoin(t *testing.T) { + tests := []struct { + sql string + }{ + { + // Count(Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid'))) + sql: "select count(*) from grouper g INNER JOIN joiner j ON g._id = j.grouperid", + }, + { + // Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid", + }, + { + // Intersect(Row(color='red'),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red'", + }, + { + // Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')) + sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and j.jointype = 2", + }, + } + + mapper := NewMapper() + for i, test := range tests { + t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) { + if m, err := mapper.MapSQL(test.sql); err != nil { + t.Fatal(err) + } else { + if stmt, ok := m.Statement.(*sqlparser.Select); !ok { + t.Fatalf("%s: expected Statement: sqlparser.Select, got %T", test.sql, m.Statement) + } else { + t.Logf("%+v\n", stmt) + } + } + }) + } +} + +func TestOrderBy(t *testing.T) { + tests := []struct { + sql string + }{ + { + sql: "select distinct score from grouper order by score asc", + }, + { + sql: "select distinct score from grouper order by score desc", + }, + { + sql: "select distinct score from grouper order by score asc limit 5", + }, + { + sql: "select distinct score from grouper order by score desc limit 5", + }, + } + mapper := NewMapper() + for i, test := range tests { + t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) { + if m, err := mapper.MapSQL(test.sql); err != nil { + t.Fatal(err) + } else { + if stmt, ok := m.Statement.(*sqlparser.Select); !ok { + t.Fatalf("%s: expected Statement: sqlparser.Select, got %T", test.sql, m.Statement) + } else { + t.Logf("%+v\n", stmt) + } + } + }) + } +} diff --git a/sql/mask.go b/sql/mask.go new file mode 100644 index 000000000..e1f859dff --- /dev/null +++ b/sql/mask.go @@ -0,0 +1,399 @@ +// 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 ( + "strings" + + "vitess.io/vitess/go/vt/sqlparser" +) + +type selectPart int + +const ( + SelectPartDistinct selectPart = 1 << iota + SelectPartID + SelectPartStar + SelectPartField + SelectPartFields + SelectPartCountStar + SelectPartCountField + SelectPartCountDistinctField + SelectPartMinField + SelectPartMaxField + SelectPartSumField + SelectPartAvgField +) + +type fromPart int + +const ( + FromPartTable fromPart = 1 << iota + FromPartTables + FromPartJoin +) + +type wherePart int + +const ( + WherePartIDCondition wherePart = 1 << iota + WherePartFieldCondition + WherePartMultiFieldCondition +) + +type groupByPart int + +const ( + GroupByPartField groupByPart = 1 << iota + GroupByPartFields +) + +type havingPart int + +const ( + HavingPartCondition havingPart = 1 << iota +) + +type orderByPart int + +const ( + OrderByPartField orderByPart = 1 << iota + OrderByPartFields +) + +type limitPart int + +const ( + LimitPartLimit limitPart = 1 << iota + LimitPartOffset +) + +type QueryMask struct { + SelectMask selectPart + FromMask fromPart + WhereMask wherePart + GroupByMask groupByPart + HavingMask havingPart + OrderByMask orderByPart + LimitMask limitPart +} + +func NewQueryMask(sp selectPart, fp fromPart, wp wherePart, gp groupByPart, hp havingPart) QueryMask { + qm := QueryMask{} + qm.orSelect(sp) + qm.orFrom(fp) + qm.orWhere(wp) + qm.orGroupBy(gp) + qm.orHaving(hp) + return qm +} + +// ApplyFilter returns true if m passes the filter f. +// Note: only certain query parts are included; namely, +// the orderBy and limit masks are not applied to the +// filter. +func (m *QueryMask) ApplyFilter(f QueryMask) bool { + if m.SelectMask&f.SelectMask != m.SelectMask { + return false + } + if m.FromMask&f.FromMask != m.FromMask { + return false + } + if m.WhereMask&f.WhereMask != m.WhereMask { + return false + } + if m.GroupByMask&f.GroupByMask != m.GroupByMask { + return false + } + if m.HavingMask&f.HavingMask != m.HavingMask { + return false + } + return true +} + +// orSelect applies the bitwise-OR operation to the selectMask. +func (m *QueryMask) orSelect(o selectPart) { + m.SelectMask |= o +} + +// orFrom applies the bitwise-OR operation to the fromMask. +func (m *QueryMask) orFrom(o fromPart) { + m.FromMask |= o +} + +// orWhere applies the bitwise-OR operation to the whereMask. +func (m *QueryMask) orWhere(parts ...wherePart) { + for _, o := range parts { + m.WhereMask |= o + } +} + +// orGroupBy applies the bitwise-OR operation to the groupByMask. +func (m *QueryMask) orGroupBy(o groupByPart) { + m.GroupByMask |= o +} + +// orHaving applies the bitwise-OR operation to the havingMask. +func (m *QueryMask) orHaving(o havingPart) { + m.HavingMask |= o +} + +// orOrderBy applies the bitwise-OR operation to the orderByMask. +func (m *QueryMask) orOrderBy(o orderByPart) { + m.OrderByMask |= o +} + +// orLimit applies the bitwise-OR operation to the limitMask. +func (m *QueryMask) orLimit(o limitPart) { + m.LimitMask |= o +} + +// HasSelect returns true if the mask contains a supported select clause. +func (m *QueryMask) HasSelect() bool { + return m.SelectMask > 0 +} + +// HasSelectPart returns true if the mask contains the provided select part. +func (m *QueryMask) HasSelectPart(p selectPart) bool { + return (m.SelectMask & p) > 0 +} + +// HasFrom returns true if the mask contains a supported from clause. +func (m *QueryMask) HasFrom() bool { + return m.FromMask > 0 +} + +// HasWhere returns true if the mask contains a supported where clause. +func (m *QueryMask) HasWhere() bool { + return m.WhereMask > 0 +} + +// HasGroupBy returns true if the mask contains a supported group by clause. +func (m *QueryMask) HasGroupBy() bool { + return m.GroupByMask > 0 +} + +// HasHaving returns true if the mask contains a supported having clause. +func (m *QueryMask) HasHaving() bool { + return m.HavingMask > 0 +} + +// HasOrderBy returns true if the mask contains a supported order by clause. +func (m *QueryMask) HasOrderBy() bool { + return m.OrderByMask > 0 +} + +// HasLimit returns true if the mask contains a supported limit clause. +func (m *QueryMask) HasLimit() bool { + return m.LimitMask > 0 +} + +////////////////////////////////////////////////////////////////////////// + +func MustGenerateMask(sql string) QueryMask { + parsed, err := sqlparser.Parse(sql) + if err != nil { + return QueryMask{} + } + return GenerateMask(parsed) +} + +func GenerateMask(parsed sqlparser.Statement) QueryMask { + qm := QueryMask{} + + switch stmt := parsed.(type) { + case *sqlparser.Select: + if strings.ToLower(strings.TrimSpace(stmt.Distinct)) == "distinct" { + qm.orSelect(SelectPartDistinct) + } + // select parts + var fldCount int + for _, item := range stmt.SelectExprs { + switch expr := item.(type) { + case *sqlparser.AliasedExpr: + switch colExpr := expr.Expr.(type) { + case *sqlparser.ColName: + name := colExpr.Name.String() + if name == ColID { + qm.orSelect(SelectPartID) + } else { + fldCount++ + } + case *sqlparser.FuncExpr: + funcName := FuncName(strings.ToLower(colExpr.Name.String())) + switch len(colExpr.Exprs) { + case 1: + var isStar bool + var isField bool + var isDistinctField bool + switch exp := colExpr.Exprs[0].(type) { + case *sqlparser.AliasedExpr: + switch exp.Expr.(type) { + case *sqlparser.ColName: + isField = true + isDistinctField = colExpr.Distinct + } + case *sqlparser.StarExpr: + isStar = true + } + + switch funcName { + case FuncCount: + if isStar { + qm.orSelect(SelectPartCountStar) + } else if isDistinctField { + qm.orSelect(SelectPartCountDistinctField) + } else if isField { + qm.orSelect(SelectPartCountField) + } + case FuncMin: + if isField { + qm.orSelect(SelectPartMinField) + } + case FuncMax: + if isField { + qm.orSelect(SelectPartMaxField) + } + case FuncSum: + if isField { + qm.orSelect(SelectPartSumField) + } + case FuncAvg: + if isField { + qm.orSelect(SelectPartAvgField) + } + } + } + } + case *sqlparser.StarExpr: + qm.orSelect(SelectPartStar) + } + } + if fldCount == 1 { + qm.orSelect(SelectPartField) + } else if fldCount > 1 { + qm.orSelect(SelectPartFields) + } + + // from parts + switch len(stmt.From) { + case 1: + switch stmt.From[0].(type) { + case *sqlparser.AliasedTableExpr: + qm.orFrom(FromPartTable) + case *sqlparser.JoinTableExpr: + qm.orFrom(FromPartJoin) + } + case 2: + switch stmt.From[0].(type) { + case *sqlparser.AliasedTableExpr: + switch stmt.From[1].(type) { + case *sqlparser.AliasedTableExpr: + qm.orFrom(FromPartTables) + } + } + } + + // where parts + where := stmt.Where + if where != nil { + switch where.Type { + case "where": + qm.orWhere(generateWhereMask(where.Expr)...) + } + } + + // group by parts + var groupByFieldCount int + for _, item := range stmt.GroupBy { + switch item.(type) { + case *sqlparser.ColName: + groupByFieldCount++ + } + } + if groupByFieldCount == 1 { + qm.orGroupBy(GroupByPartField) + } else if groupByFieldCount > 1 { + qm.orGroupBy(GroupByPartFields) + } + + // having parts + if stmt.Having != nil { + qm.orHaving(HavingPartCondition) + } + + // order by parts + var orderByFieldCount int + for _, item := range stmt.OrderBy { + switch item.Expr.(type) { + case *sqlparser.ColName: + orderByFieldCount++ + } + } + if orderByFieldCount == 1 { + qm.orOrderBy(OrderByPartField) + } else if orderByFieldCount > 1 { + qm.orOrderBy(OrderByPartFields) + } + + // limit parts + if stmt.Limit != nil { + switch stmt.Limit.Rowcount.(type) { + case *sqlparser.SQLVal: + qm.orLimit(LimitPartLimit) + } + switch stmt.Limit.Offset.(type) { + case *sqlparser.SQLVal: + qm.orLimit(LimitPartOffset) + } + } + } + return qm +} + +// TODO: add more recursion within the comparison operators (left/right parts) +func generateWhereMask(e sqlparser.Expr) []wherePart { + var wp []wherePart + + switch expr := e.(type) { + case *sqlparser.ComparisonExpr: + var leftName string + if colExpr, ok := expr.Left.(*sqlparser.ColName); ok { + leftName = colExpr.Name.String() + } + switch leftName { + case ColID: + wp = append(wp, WherePartIDCondition) + case "": + // + default: + wp = append(wp, WherePartFieldCondition) + } + case *sqlparser.RangeCond: + wp = append(wp, WherePartFieldCondition) + case *sqlparser.AndExpr, *sqlparser.OrExpr: + // TODO: we need to recursively ensure that the left/right + // sides of these and/or expressions are field-op-val, and + // that none of the fields are "_id" + // TODO: could we use extractComparison or something like it? + wp = append(wp, WherePartMultiFieldCondition) + case *sqlparser.NotExpr: + wp = append(wp, generateWhereMask(expr.Expr)...) + case *sqlparser.IsExpr: + wp = append(wp, WherePartFieldCondition) + } + + return wp +} diff --git a/sql/model.go b/sql/model.go new file mode 100644 index 000000000..76ba80498 --- /dev/null +++ b/sql/model.go @@ -0,0 +1,141 @@ +// 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 ( + "fmt" + + "github.com/pilosa/pilosa/v2" + "github.com/pkg/errors" +) + +var ( + ErrUnsupportedQuery = errors.New("unsupported query") +) + +// TODO: what the difference between this and IDIndexColumn? +type KeyIndexColumn struct { + Index *pilosa.Index + text string + alias string +} + +func NewKeyIndexColumn(index *pilosa.Index, alias string) *KeyIndexColumn { + return &KeyIndexColumn{ + Index: index, + text: ColID, + alias: alias, + } +} + +func (i *KeyIndexColumn) Source() string { + return "" // TODO +} +func (i *KeyIndexColumn) Name() string { + return i.text +} +func (i *KeyIndexColumn) Alias() string { + if i.alias != "" { + return i.alias + } + return i.Name() +} + +type IDIndexColumn struct { + Index *pilosa.Index + text string + alias string +} + +func NewIDIndexColumn(index *pilosa.Index, alias string) *IDIndexColumn { + return &IDIndexColumn{ + Index: index, + text: ColID, + alias: alias, + } +} + +func (i *IDIndexColumn) Source() string { + return "" // TODO +} +func (i *IDIndexColumn) Name() string { + return i.text +} +func (i *IDIndexColumn) Alias() string { + if i.alias != "" { + return i.alias + } + return i.Name() +} + +type FieldColumn struct { + Field *pilosa.Field + text string + alias string +} + +func NewFieldColumn(field *pilosa.Field, alias string) *FieldColumn { + return &FieldColumn{ + Field: field, + text: field.Name(), + alias: alias, + } +} + +func (f *FieldColumn) Source() string { + return "" +} +func (f *FieldColumn) Name() string { + return f.text +} +func (f *FieldColumn) Alias() string { + if f.alias != "" { + return f.alias + } + return f.Name() +} + +type FuncColumn struct { + Field *pilosa.Field + FuncName FuncName + alias string +} + +func NewFuncColumn(funcName FuncName, field *pilosa.Field, alias string) *FuncColumn { + return &FuncColumn{ + Field: field, + FuncName: funcName, + alias: alias, + } +} + +func (f *FuncColumn) Source() string { + return string(f.FuncName) +} + +func (f *FuncColumn) Name() string { + fieldName := "*" + if f.Field != nil { + fieldName = f.Field.Name() + } + return fmt.Sprintf("%s(%s)", f.FuncName, fieldName) +} + +func (f *FuncColumn) Alias() string { + if f.alias != "" { + return f.alias + } + return f.Name() +} diff --git a/sql/query.go b/sql/query.go new file mode 100644 index 000000000..9996110a9 --- /dev/null +++ b/sql/query.go @@ -0,0 +1,295 @@ +// 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 ( + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "github.com/pkg/errors" +) + +const timeFormat = "2006-01-02T15:04" + +// LT creates a less than query. +func LT(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s<%s)", fieldName, intOrFloat(value)) +} + +// LTE creates a less than or equal query. +func LTE(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s<=%s)", fieldName, intOrFloat(value)) +} + +// GT creates a greater than query. +func GT(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s>%s)", fieldName, intOrFloat(value)) +} + +// GTE creates a greater than or equal query. +func GTE(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s>=%s)", fieldName, intOrFloat(value)) +} + +// Equals creates an equals query. +func Equals(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s=%s)", fieldName, intOrFloat(value)) +} + +// NotEquals creates a not equals query. +func NotEquals(fieldName string, value interface{}) string { + return fmt.Sprintf("Row(%s!=%s)", fieldName, intOrFloat(value)) +} + +// NotNull creates a not equal to null query. +func NotNull(fieldName string) string { + return fmt.Sprintf("Row(%s!=null)", fieldName) +} + +// Row query +func Row(fieldName string, rowIDOrKey interface{}) (string, error) { + rowStr, err := formatIDKeyBool(rowIDOrKey) + if err != nil { + return "", err + } + text := fmt.Sprintf("Row(%s=%s)", fieldName, rowStr) + return text, nil +} + +// RowRange is a Row query with from,to times +func RowRange(fieldName string, rowIDOrKey interface{}, start time.Time, end time.Time) (string, error) { + rowStr, err := formatIDKeyBool(rowIDOrKey) + if err != nil { + return "", err + } + text := fmt.Sprintf("Row(%s=%s,from='%s',to='%s')", fieldName, rowStr, start.Format(timeFormat), end.Format(timeFormat)) + return text, nil +} + +// Union query - see rowOperation +func Union(rows ...string) string { + return rowOperation("Union", rows...) +} + +// Intersect query - see rowOperation +func Intersect(rows ...string) string { + return rowOperation("Intersect", rows...) +} + +// Not query +func Not(rows ...string) string { + return rowOperation("Not", rows...) +} + +// Like creates a Rows query filtered by a pattern. +// An underscore ('_') can be used as a placeholder for a single UTF-8 codepoint or a percent sign ('%') can be used as a placeholder for 0 or more codepoints. +// All other codepoints in the pattern are matched exactly. +func Like(fieldName string, pattern string) string { + pattern = strings.ReplaceAll(pattern, `\`, `\\`) + pattern = strings.ReplaceAll(pattern, `'`, `\'`) + return fmt.Sprintf("UnionRows(Rows(field='%s',like='%s'))", fieldName, pattern) +} + +// Between creates a between query. +func Between(fieldName string, a interface{}, b interface{}) string { + return fmt.Sprintf("Row(%s >< [%s,%s])", fieldName, intOrFloat(a), intOrFloat(b)) +} + +// Distinct creates a Distinct query. +func Distinct(indexName, fieldName string) string { + return fmt.Sprintf("Distinct(Row(%s!=null),index='%s',field='%s')", fieldName, indexName, fieldName) +} + +// RowDistinct creates a Distinct query with the given row filter. +func RowDistinct(indexName, fieldName string, row string) string { + return fmt.Sprintf("Distinct(%s,index='%s',field='%s')", row, indexName, fieldName) +} + +// Rows creates a Rows query with defaults +func Rows(fieldName string) string { + return fmt.Sprintf("Rows(field='%s')", fieldName) +} + +// RowsLimit creates a Rows query with the given limit +func RowsLimit(fieldName string, limit int64) (string, error) { + if limit < 0 { + return "", errors.New("rows limit must be non-negative") + } + text := fmt.Sprintf("Rows(field='%s',limit=%d)", fieldName, limit) + return text, nil +} + +// All creates an All query. +// Returns the set columns with existence true. +func All() string { + return "All()" +} + +// Count creates a Count query. +// Returns the number of set columns in the ROW_CALL passed in. +func Count(rowCall string) string { + return fmt.Sprintf("Count(%s)", rowCall) +} + +// Sum creates a sum query. +func Sum(fieldName string, row string) string { + return valQuery(fieldName, "Sum", row) +} + +// Min creates a min query. +func Min(fieldName string, row string) string { + return valQuery(fieldName, "Min", row) +} + +// Max creates a max query. +func Max(fieldName string, row string) string { + return valQuery(fieldName, "Max", row) +} + +// TopN creates a TopN query with the given item count. +// Returns the id and count of the top n rows (by count of columns) in the field. +func TopN(fieldName string, n uint64) string { + return fmt.Sprintf("TopN(%s,n=%d)", fieldName, n) +} + +// RowTopN creates a TopN query with the given item count and row. +// This variant supports customizing the row query. +func RowTopN(fieldName string, n uint64, row string) string { + return fmt.Sprintf("TopN(%s,%s,n=%d)", fieldName, row, n) +} + +// GroupByBase creates a GroupBy query with the given functional options. +func GroupByBase(rows []string, limit int64, filter, aggregate, having string) (string, error) { + if len(rows) == 0 { + return "", errors.New("there should be at least one rows query") + } + if limit < 0 { + return "", errors.New("limit must be non-negative") + } + + // rows + text := fmt.Sprintf("GroupBy(%s", strings.Join(rows, ",")) + + // limit + if limit > 0 { + text += fmt.Sprintf(",limit=%d", limit) + } + + // filter + if filter != "" { + text += fmt.Sprintf(",filter=%s", filter) + } + + // aggregate + if aggregate != "" { + text += fmt.Sprintf(",aggregate=%s", aggregate) + } + + // having + if having != "" { + text += fmt.Sprintf(",having=%s", having) + } + + text += ")" + return text, nil +} + +// Limit creates a limit query. +func Limit(row string, limit uint, offset uint) string { + return fmt.Sprintf("Limit(%s, limit=%d, offset=%d)", row, limit, offset) +} + +// Offset creates a limit query but only with an offset. +func Offset(row string, offset uint) string { + return fmt.Sprintf("Limit(%s, offset=%d)", row, offset) +} + +// ConstRow creates a query value that uses a list of columns in place of a Row query. +func ConstRow(ids ...interface{}) string { + if ids == nil { + ids = []interface{}{} + } + data, _ := json.Marshal(ids) + return fmt.Sprintf("ConstRow(columns=%s)", data) +} + +// Extract creates an Extract query. +// It accepts a bitmap query to select columns and a list of fields to select rows. +func Extract(rowCall string, fields ...string) string { + var rowsCall string + for _, r := range fields { + rowsCall += "," + fmt.Sprintf("Rows(%s)", r) + } + + return fmt.Sprintf("Extract(%s%s)", rowCall, rowsCall) +} + +func valQuery(fieldName string, op string, row string) string { + if row != "" { + row += "," + } + return fmt.Sprintf("%s(%sfield='%s')", op, row, fieldName) +} + +func rowOperation(name string, rows ...string) string { + return fmt.Sprintf("%s(%s)", name, strings.Join(rows, ",")) +} + +func formatIDKeyBool(idKeyBool interface{}) (string, error) { + if b, ok := idKeyBool.(bool); ok { + return strconv.FormatBool(b), nil + } + if flt, ok := idKeyBool.(float64); ok { + return fmt.Sprintf("%f", flt), nil + } + return formatIDKey(idKeyBool) +} + +func formatIDKey(idKey interface{}) (string, error) { + switch v := idKey.(type) { + case uint: + return strconv.FormatUint(uint64(v), 10), nil + case uint32: + return strconv.FormatUint(uint64(v), 10), nil + case uint64: + return strconv.FormatUint(v, 10), nil + case int: + return strconv.FormatInt(int64(v), 10), nil + case int32: + return strconv.FormatInt(int64(v), 10), nil + case int64: + return strconv.FormatInt(v, 10), nil + case string: + v = strings.ReplaceAll(v, `\`, `\\`) + return fmt.Sprintf(`'%s'`, strings.ReplaceAll(v, `'`, `\'`)), nil + default: + return "", errors.Errorf("id/key is not a string or integer type: %#v", idKey) + } +} + +func intOrFloat(value interface{}) string { + switch value.(type) { + case float64, float32: + // In order to test expected values, we set the precision + // to 8. TODO: It's likely we'll need to address this + // at some point. + return fmt.Sprintf("%.8f", value) + default: + return fmt.Sprintf("%d", value) + } +} diff --git a/sql/reduce.go b/sql/reduce.go new file mode 100644 index 000000000..be0e98695 --- /dev/null +++ b/sql/reduce.go @@ -0,0 +1,500 @@ +// 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 ( + "io" + "sort" + + "github.com/pilosa/pilosa/v2/pql" + pproto "github.com/pilosa/pilosa/v2/proto" + "github.com/pkg/errors" + "google.golang.org/grpc/codes" +) + +// DataType contants describe the possible values +// for the Datatype value in the RowResponse header. +const ( + DataTypeDecimal = "decimal" + DataTypeFloat64 = "float64" + DataTypeInt64 = "int64" + DataTypeString = "string" + DataTypeUint64Array = "[]uint64" +) + +type Reducer interface { + Reduce(pproto.StreamClient, pproto.StreamServer) error +} + +// LimitReducer limits the number of messages passed through. +type LimitReducer struct { + limit uint + offset uint +} + +// NewLimitReducer returns a new instance of LimitReducer. +func NewLimitReducer(limit, offset uint) *LimitReducer { + return &LimitReducer{ + limit: limit, + offset: offset, + } +} + +// Reduce applies the limit reducer to the client stream and sends the results +// to the server stream. +func (l *LimitReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error { + offsetCountdown := l.offset + + // in the case of an offset, since we'll be skipping the first record + // which contains the headers, we need to pull the headers, save them, + // and apply them to the first record that we actually send through. + var headers []*pproto.ColumnInfo + + for i := uint(0); i < l.limit+l.offset || l.limit == 0; i++ { + r, err := c.Recv() + if err == io.EOF { + break + } else if err != nil { + return s.Send(pproto.ErrorWrap(err, "receiving on client stream")) + } + if offsetCountdown > 0 { + if headers == nil { + headers = r.Headers + } + offsetCountdown-- + continue + } + if headers != nil { + r.Headers = headers + headers = nil + } + if err := s.Send(r); err != nil { + return s.Send(pproto.ErrorWrap(err, "sending on server stream")) + } + } + return s.Send(pproto.EOF) +} + +// OrderByReducer orders the results based on the provide conditions. +// It also takes limit and offset to reduce the amount of items +// needing to be held in memory for sorting. +type OrderByReducer struct { + fields []string + isDescending []bool // direction[asc: false, desc: true] + limit uint + offset uint +} + +// NewOrderByReducer returns a new instance of OrderByReducer. +func NewOrderByReducer(fields, dirs []string, limit, offset uint) *OrderByReducer { + descendings := make([]bool, len(fields)) + for i := range dirs { + if dirs[i] == "desc" { + descendings[i] = true + } + } + return &OrderByReducer{ + fields: fields, + isDescending: descendings, + limit: limit, + offset: offset, + } +} + +// Reduce applies the order by reducer to the client stream and sends the results +// to the server stream. +func (o *OrderByReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error { + // hold is a slice of row responses, to be sent to the output + // stream sorted by the sort conditions. + var hold []*pproto.RowResponse + + // sortColNames contains the names of the columns to + // sort on. + sortColNames := o.fields + + // sortColIdxs contains the positions of the sort columns + // in the result set. + sortColIdxs := make([]int, len(sortColNames)) + + // sortColTypes contains the data types of the columns + // to be sorted. (ex: "uint64", "string", etc.). This + // is used to determine how to convert it to a typed + // field for sorting. + sortColTypes := make([]string, len(sortColNames)) + + // holdHeaders is used to stash the headers (from + // the first row) so they can be applied later + // to what will eventually be the first row after + // sorting has occurred. + var holdHeaders []*pproto.ColumnInfo + + ii := 0 + for { + rr, err := c.Recv() + if err != nil { + if err == io.EOF { + break + } + return s.Send(pproto.ErrorWrap(err, "receiving row response")) + } + + // On the first row, get the sort column information + // from the headers. Also, stash the headers for + // later in the `holdHeaders` var. + if ii == 0 { + holdHeaders = rr.Headers + for i, rrHdr := range rr.Headers { + hdrName := rrHdr.GetName() + hdrType := rrHdr.GetDatatype() + for j := range sortColNames { + if sortColNames[j] == hdrName { + sortColIdxs[j] = i + sortColTypes[j] = hdrType + } + } + } + // Clear the headers in case this record is + // no longer first (we re-apply the headers + // to the first outgoing record later). + rr.Headers = nil + } + + // Put each row in the hold. + hold = append(hold, rr) + + ii++ + + // TODO: in the case where limit is provided and the number of possible + // rows is large, it might be more efficient to periodically sort/trim + // the hold so it doesn't become too large. For example, it could + // be constrained to size (limit + offset + buffer), where buffer is + // an amount that the hold can grow before being trimmed. + } + + // Sort the hold. + sorter, err := pproto.NewRowResponseSorter( + sortColIdxs, + o.isDescending, + sortColTypes, + hold, + ) + if err != nil { + return s.Send(pproto.ErrorWrap(err, "creating row response sorter")) + } + sort.Sort(sorter) + + var rowsToConsider uint = uint(len(hold)) + var offsetCountdown uint + if o.limit > 0 { + offsetCountdown = o.offset + if o.limit+o.offset < rowsToConsider { + rowsToConsider = o.limit + o.offset + } + } + + // Loop over hold and send each row response. + // Apply the header to the first row that is sent. + var headerApplied bool + for i := uint(0); i < rowsToConsider; i++ { + if offsetCountdown > 0 { + offsetCountdown-- + continue + } + // Re-apply the headers to the first record. + if !headerApplied { + hold[i].Headers = holdHeaders + headerApplied = true + } + err := s.Send(hold[i]) + if err != nil { + return s.Send(pproto.ErrorWrap(err, "sending hold row")) + } + } + return s.Send(pproto.EOF) +} + +// ValCountFuncReducer converts a ValCount result to the proper +// result for Func. +type ValCountFuncReducer struct { + fn FuncName +} + +// NewValCountFuncReducer returns a new instance of ValCountFuncReducer. +func NewValCountFuncReducer(fn FuncName) *ValCountFuncReducer { + return &ValCountFuncReducer{ + fn: fn, + } +} + +// Reduce modifies the stream according to the function. +func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error { + r, err := c.Recv() + if err != nil { + if err == io.EOF { + return s.Send(pproto.EOF) + } + return s.Send(pproto.Error(err)) + } + + // Get the index of the column with header of "value". + var idxVal int = -1 + var idxCnt int = -1 + headers := r.GetHeaders() + for i, hdr := range headers { + switch hdr.GetName() { + case "value": + idxVal = i + case "count": + idxCnt = i + } + } + + var sourceDataType string + var returnDataType string + + sourceDataType = headers[idxVal].GetDatatype() + returnDataType = sourceDataType + switch v.fn { + case FuncAvg: + returnDataType = DataTypeFloat64 + } + + rr := pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: string(v.fn), Datatype: returnDataType}, + }, + Columns: make([]*pproto.ColumnResponse, 1), + } + + cols := r.GetColumns() + if len(cols) == 0 { + return s.Send(pproto.ErrorCode( + errors.New("empty column set"), + codes.Unknown, + )) + } + + if idxVal == -1 { + return s.Send(pproto.ErrorCode( + errors.New("result set has no column: value"), + codes.Unknown, + )) + } + if idxCnt == -1 { + return s.Send(pproto.ErrorCode( + errors.New("result set has no column: count"), + codes.Unknown, + )) + } + + switch v.fn { + case FuncAvg: + var avg float64 + if sourceDataType == DataTypeDecimal { + val := cols[idxVal].GetDecimalVal() + dec := pql.NewDecimal(val.Value, val.Scale) + cnt := cols[idxCnt].GetInt64Val() + avg = dec.Float64() / float64(cnt) + } else { + val := cols[idxVal].GetInt64Val() + cnt := cols[idxCnt].GetInt64Val() + avg = float64(val) / float64(cnt) + } + rr.Columns[0] = &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_Float64Val{Float64Val: avg}} + default: + if sourceDataType == DataTypeDecimal { + val := cols[idxVal].GetDecimalVal() + rr.Columns[0] = &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_DecimalVal{DecimalVal: &pproto.Decimal{Value: val.Value, Scale: val.Scale}}} + } else { + val := cols[idxVal].GetInt64Val() + rr.Columns[0] = &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_Int64Val{Int64Val: val}} + } + } + + if err := s.Send(&rr); err != nil { + return errors.Wrap(err, "sending row response") + } + return s.Send(pproto.EOF) +} + +// CountIDReducer returns a stream of _id's as a count. +type CountIDReducer struct{} + +// Reduce counts the stream of IDs and returns a single record. +func (r *CountIDReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error { + var cnt uint64 + + for { + _, err := c.Recv() + if err != nil { + if err == io.EOF { + break + } + return s.Send(pproto.ErrorWrap(err, "receiving on client stream")) + } + cnt++ + } + + rr := pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: string(FuncCount), Datatype: "uint64"}, + }, + Columns: []*pproto.ColumnResponse{ + &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_Uint64Val{Uint64Val: cnt}}, + }, + } + + if err := s.Send(&rr); err != nil { + return errors.Wrap(err, "sending row response") + } + return s.Send(pproto.EOF) +} + +// AssignHeadersReducer overwrites the headers on the first record +// according to field names and aliases from sql. It also reorders +// the columns in the result stream to match the sql select clause. +type AssignHeadersReducer struct { + cols []Column +} + +// NewAssignHeadersReducer returns a new instance of AssignHeadersReducer. +func NewAssignHeadersReducer(cols []Column) *AssignHeadersReducer { + return &AssignHeadersReducer{ + cols: cols, + } +} + +// Reduce modifies the stream. +func (r *AssignHeadersReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error { + var placement []uint + var labels []string + + var cnt int + for { + rr, err := c.Recv() + if err != nil { + if err == io.EOF { + break + } + return s.Send(pproto.ErrorWrap(err, "receiving on client stream")) + } + + // If the placement slice is [0-n] where n == len(Headers) + // then we don't need to alter rr on records after cnt == 0. + // If we don't apply aliases, we don't have to alter Headers + // either, but that may not be worth messing with. + + if cnt == 0 { + placement, labels, err = headerAssignment(r.cols, rr.Headers) + if err != nil { + return s.Send(pproto.ErrorWrap(err, "getting header assignment")) + } + + // mod is the modified RowResponse object that gets populated + // according to placement and labels, then sent. + mod := &pproto.RowResponse{ + Headers: make([]*pproto.ColumnInfo, len(placement)), + Columns: make([]*pproto.ColumnResponse, len(placement)), + } + + // For now, we assume that the column count in each RowResponse + // is consistent (i.e. we can validate one time, here, on the + // first row, and not every time, in the `else` statement below). + if len(placement) > len(rr.Columns) { + return s.Send(pproto.ErrorCode( + errors.New("mismatched header placement and column count"), + codes.Unknown, + )) + } + + for i := 0; i < len(placement); i++ { + mod.Headers[i] = rr.Headers[placement[i]] + mod.Headers[i].Name = labels[i] + mod.Columns[i] = rr.Columns[placement[i]] + } + if err := s.Send(mod); err != nil { + return errors.Wrap(err, "sending mod") + } + } else { + // mod is the modified RowResponse object that gets populated + // according to placement and labels, then sent. + mod := &pproto.RowResponse{ + Columns: make([]*pproto.ColumnResponse, len(placement)), + } + for i := 0; i < len(placement); i++ { + mod.Columns[i] = rr.Columns[placement[i]] + } + if err := s.Send(mod); err != nil { + return errors.Wrap(err, "sending mod") + } + } + cnt++ + } + + return s.Send(pproto.EOF) +} + +var ( + ErrIncompleteHeaders = errors.New("incomplete header assignment") + ErrFieldNotInHeaders = errors.New("field not found in source header") +) + +func headerAssignment(cols []Column, hdrs []*pproto.ColumnInfo) ([]uint, []string, error) { + // If any of the columns are "*" (i.e. type StarColumn), + // then ignore everything else and just use all result + // headers. + var hasStar bool + for _, col := range cols { + if _, ok := col.(*StarColumn); ok { + hasStar = true + break + } + } + if hasStar { + placement := make([]uint, len(hdrs)) + labels := make([]string, len(hdrs)) + for i, hdr := range hdrs { + placement[i] = uint(i) + labels[i] = hdr.Name + } + return placement, labels, nil + } + + if len(cols) > len(hdrs) { + return nil, nil, ErrIncompleteHeaders + } + placement := make([]uint, len(cols)) + labels := make([]string, len(cols)) + + // Make a map of the RowResponse headers. + hdrMap := make(map[string]uint) + for i, hdr := range hdrs { + hdrMap[hdr.Name] = uint(i) + } + + // Lookup each column in the hdrMap and determine the desired placement. + for i, col := range cols { + if srcHdrIdx, ok := hdrMap[col.Source()]; ok { + placement[i] = srcHdrIdx + labels[i] = col.Alias() + } else if nameHdrIdx, ok := hdrMap[col.Name()]; ok { + placement[i] = nameHdrIdx + labels[i] = col.Alias() + } else { + return nil, nil, errors.Wrapf(ErrFieldNotInHeaders, "field: %s", col.Name()) + } + } + return placement, labels, nil +} diff --git a/sql/reduce_test.go b/sql/reduce_test.go new file mode 100644 index 000000000..27f645c99 --- /dev/null +++ b/sql/reduce_test.go @@ -0,0 +1,120 @@ +// 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 ( + "fmt" + "reflect" + "testing" + + pproto "github.com/pilosa/pilosa/v2/proto" + "github.com/pkg/errors" +) + +func TestHeaderAssignment(t *testing.T) { + abcdHdrs := []*pproto.ColumnInfo{ + {Name: "a"}, {Name: "b"}, {Name: "c"}, {Name: "d"}, + } + tests := []struct { + cols []Column + hdrs []*pproto.ColumnInfo + expPlacement []uint + expLabels []string + expErr error + }{ + { + cols: []Column{ + NewBasicColumn("a", "namea", ""), + }, + hdrs: abcdHdrs, + expPlacement: []uint{0}, + expLabels: []string{"namea"}, + }, + { + cols: []Column{ + NewBasicColumn("", "a", ""), + }, + hdrs: abcdHdrs, + expPlacement: []uint{0}, + expLabels: []string{"a"}, + }, + { + cols: []Column{ + NewBasicColumn("", "a", "aliasa"), + }, + hdrs: abcdHdrs, + expPlacement: []uint{0}, + expLabels: []string{"aliasa"}, + }, + { + cols: []Column{ + NewBasicColumn("", "a", "aliasa"), + NewBasicColumn("c", "namec", ""), + }, + hdrs: abcdHdrs, + expPlacement: []uint{0, 2}, + expLabels: []string{"aliasa", "namec"}, + }, + { + cols: []Column{ + NewBasicColumn("d", "c", "aliasd"), + NewBasicColumn("b", "nameb", ""), + }, + hdrs: abcdHdrs, + expPlacement: []uint{3, 1}, + expLabels: []string{"aliasd", "nameb"}, + }, + // Errors + { + cols: []Column{ + NewBasicColumn("", "x", ""), + }, + hdrs: abcdHdrs, + expErr: ErrFieldNotInHeaders, + }, + { + cols: []Column{ + NewBasicColumn("", "a", ""), + NewBasicColumn("", "b", ""), + NewBasicColumn("", "c", ""), + NewBasicColumn("", "d", ""), + NewBasicColumn("", "e", ""), + }, + hdrs: abcdHdrs, + expErr: ErrIncompleteHeaders, + }, + } + for i, test := range tests { + t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) { + placement, labels, err := headerAssignment(test.cols, test.hdrs) + + if test.expErr == nil { + if err != nil { + t.Fatal(err) + } + } else { + if test.expErr != errors.Cause(err) { + t.Fatalf("expected error: %v, but got: %v", test.expErr, err) + } + } + + if !reflect.DeepEqual(placement, test.expPlacement) { + t.Fatalf("expected placement: %v, but got: %v", test.expPlacement, placement) + } else if !reflect.DeepEqual(labels, test.expLabels) { + t.Fatalf("expected labels: %v, but got: %v", test.expLabels, labels) + } + }) + } +} diff --git a/sql/router.go b/sql/router.go new file mode 100644 index 000000000..f64daa338 --- /dev/null +++ b/sql/router.go @@ -0,0 +1,162 @@ +// 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 + +type router struct { + direct map[QueryMask]handler + filters []maskFilter +} + +type maskFilter struct { + optional QueryMask + required []QueryMask + handler handler +} + +func newRouter() *router { + selectRouter := &router{ + direct: make(map[QueryMask]handler), + } + + selectRouter.addFilter( + NewQueryMask( + SelectPartID|SelectPartStar|SelectPartField|SelectPartFields, + FromPartTable, + WherePartFieldCondition|WherePartMultiFieldCondition|WherePartIDCondition, + 0, + 0, + ), + []QueryMask{}, + handlerSelectFieldsFromTableWhere{}, + ) + //// + selectRouter.addRoute("select distinct fld from tbl", handlerSelectDistinctFromTable{}) + //// + selectRouter.addFilter( + NewQueryMask( + SelectPartCountStar|SelectPartCountField|SelectPartCountDistinctField, + FromPartTable, + WherePartFieldCondition|WherePartMultiFieldCondition|WherePartIDCondition, + 0, + 0, + ), + []QueryMask{}, + handlerSelectCountFromTableWhere{}, + ) + //// + selectRouter.addFilter( + NewQueryMask( + SelectPartMinField|SelectPartMaxField|SelectPartSumField|SelectPartAvgField, + FromPartTable, + WherePartFieldCondition|WherePartMultiFieldCondition|WherePartIDCondition, + 0, + 0, + ), + []QueryMask{}, + handlerSelectFuncFromTableWhere{}, + ) + //// + groupByOptional := NewQueryMask( + SelectPartField|SelectPartFields|SelectPartCountStar|SelectPartSumField, + FromPartTable, + WherePartFieldCondition, // TODO: this can probably handle fields as well + GroupByPartField|GroupByPartFields, + HavingPartCondition, + ) + selectRouter.addFilter( + groupByOptional, + []QueryMask{NewQueryMask(0, 0, 0, GroupByPartField, 0)}, + handlerSelectGroupBy{}, + ) + selectRouter.addFilter( + groupByOptional, + []QueryMask{NewQueryMask(0, 0, 0, GroupByPartFields, 0)}, + handlerSelectGroupBy{}, + ) + selectRouter.addRoute("select fld, count(fld) from tbl group by fld", handlerSelectGroupBy{}) + selectRouter.addRoute("select fld1, count(fld1) from tbl where fld2=1 group by fld1", handlerSelectGroupBy{}) + + selectRouter.addRoute("select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi", handlerSelectJoin{}) + selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi", handlerSelectJoin{}) + selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1", handlerSelectJoin{}) + selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1 and fld2=2", handlerSelectJoin{}) + + return selectRouter +} + +func (r *router) addRoute(sql string, handler handler) { + r.direct[MustGenerateMask(sql)] = handler +} + +func (r *router) addFilter(opt QueryMask, req []QueryMask, handler handler) { + mf := maskFilter{ + optional: opt, + required: req, + handler: handler, + } + r.filters = append(r.filters, mf) +} + +func (r *router) handler(qm QueryMask) handler { + // First, check for a direct mapping. + // Zero out the orderBy and limit mask, because + // those are not specific to the query processing. + zm := QueryMask{ + SelectMask: qm.SelectMask, + FromMask: qm.FromMask, + WhereMask: qm.WhereMask, + GroupByMask: qm.GroupByMask, + HavingMask: qm.HavingMask, + } + if h, ok := r.direct[zm]; ok { + return h + } + for _, mf := range r.filters { + if applyMaskFilter(&qm, mf) { + return mf.handler + } + } + return nil +} + +// applyMaskFilter returns true if m passes the filter mf. +// Note: only certain query parts are included; namely, +// the orderBy and limit masks are not applied to the +// filter. A mask can satisfy any part of the optional +// filter to pass through, but it MUST satisfy all parts +// of the required filter. +func applyMaskFilter(m *QueryMask, mf maskFilter) bool { + if !m.ApplyFilter(mf.optional) { + return false + } + for _, req := range mf.required { + if m.SelectMask&req.SelectMask != req.SelectMask { + return false + } + if m.FromMask&req.FromMask != req.FromMask { + return false + } + if m.WhereMask&req.WhereMask != req.WhereMask { + return false + } + if m.GroupByMask&req.GroupByMask != req.GroupByMask { + return false + } + if m.HavingMask&req.HavingMask != req.HavingMask { + return false + } + } + return true +} diff --git a/sql/select.go b/sql/select.go new file mode 100644 index 000000000..1c11cdfca --- /dev/null +++ b/sql/select.go @@ -0,0 +1,792 @@ +// 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" + "strings" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/pql" + pproto "github.com/pilosa/pilosa/v2/proto" + "github.com/pkg/errors" + "vitess.io/vitess/go/vt/sqlparser" +) + +// SelectHandler executes SQL select statements +type SelectHandler struct { + api *pilosa.API + router *router +} + +// NewSelectHandler constructor +func NewSelectHandler(api *pilosa.API) *SelectHandler { + return &SelectHandler{ + api: api, + router: newRouter(), + } +} + +// 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) + if err != nil { + return nil, errors.Wrap(err, "mapping select") + } + return s.execMappingResult(ctx, mr) +} + +func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Select, qm QueryMask) (*MappingResult, error) { + // Get the handler for this query mask. + handler := s.router.handler(qm) + if handler == nil { + return nil, ErrUnsupportedQuery + } + indexFunc := func(indexName string) *pilosa.Index { + idx, err := s.api.Index(ctx, indexName) + if err != nil { + return nil + } + return idx + } + + mr, err := handler.Apply(selectStmt, qm, indexFunc) + if err != nil { + return nil, errors.Wrap(err, "handling") + } + return mr, nil +} + +func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (stream pproto.StreamClient, err 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") + } + res := resp.Results[0] + + // TODO: synchronize this properly somehow. + // It would probbably help to get rid of the streaming too. + respRows := pproto.NewRowBuffer(0) + switch res := res.(type) { + case pproto.ToRowser: + go func() { + if err := res.ToRows(respRows.Send); err != nil { + respRows.Send(pproto.Error(err)) //nolint:errcheck + } else { + _ = respRows.Send(pproto.EOF) //nolint:errcheck + } + }() + case []pilosa.GroupCount: + go func() { + if err := pilosa.GroupCounts(res).ToRows(respRows.Send); err != nil { + respRows.Send(pproto.Error(err)) //nolint:errcheck + } else { + respRows.Send(pproto.EOF) //nolint:errcheck + } + }() + case uint64: + go func() { + respRows.Send(&pproto.RowResponse{ //nolint:errcheck + Headers: []*pproto.ColumnInfo{ + { + Name: "count", + Datatype: "uint64", + }, + }, + Columns: []*pproto.ColumnResponse{ + { + ColumnVal: &pproto.ColumnResponse_Uint64Val{ + Uint64Val: res, + }, + }, + }, + }) + respRows.Send(pproto.EOF) //nolint:errcheck + }() + case bool: + go func() { + respRows.Send(&pproto.RowResponse{ //nolint:errcheck + Headers: []*pproto.ColumnInfo{ + { + Name: "result", + Datatype: "bool", + }, + }, + Columns: []*pproto.ColumnResponse{ + { + ColumnVal: &pproto.ColumnResponse_BoolVal{ + BoolVal: res, + }, + }, + }, + }) + respRows.Send(pproto.EOF) //nolint:errcheck + }() + default: + return nil, fmt.Errorf("unsupported result type %T", res) + } + + // Apply Reducers + result := respRows + for _, red := range mr.Reducers { + out := pproto.NewRowBuffer(0) + + // Run Reducers asyncronously. + // TODO: stop swallowing this error. + // TODO: does this need an EOF as input? + go red.Reduce(result, out) //nolint:errcheck + + result = out + } + + return result, nil +} + +type MappingResult struct { + IndexName string + ColumnIDs []uint64 + ColumnKeys []string + FieldFilters []string + Limit uint64 + Offset uint64 + Query string + Header []Column + Reducers []Reducer +} + +func (mr *MappingResult) addReducer(r Reducer) { + mr.Reducers = append(mr.Reducers, r) +} + +type SelectProperties struct { + Index *pilosa.Index + Fields []Column + Features selectFeatures + WherePQL string + WhereIDs []uint64 + WhereKeys []string + Offset uint + Limit uint + GroupByFieldNames []string + Having *HavingClause +} + +type selectFunc struct { + funcName FuncName + field *pilosa.Field +} + +type selectFeatures struct { + HasRowAttrs bool + HasColAttrs bool + funcs []selectFunc +} + +type HavingClause struct { + Subj string + Cond pql.Condition +} + +type handler interface { + Apply(*sqlparser.Select, QueryMask, func(string) *pilosa.Index) (*MappingResult, error) +} + +// handlerSelectFieldsFromTable: Inspect() +type handlerSelectFieldsFromTableWhere struct{} + +func (h handlerSelectFieldsFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + var whereQuery string + if qm.HasWhere() { + whereQuery, err = extractWhere(index, stmt.Where.Expr) + if err != nil { + return nil, err + } + } else { + whereQuery = "All()" + } + + selectFields, _, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + var fields []string + for _, fld := range selectFields { + if _, ok := fld.(*StarColumn); ok { + pflds := index.Fields() + fields = []string{"_id"} + for _, f := range pflds { + name := f.Name() + if strings.HasPrefix(name, "_") { + continue + } + fields = append(fields, name) + } + break + } + fields = append(fields, fld.Name()) + } + for i, fld := range fields { + if fld == "_id" && i != 0 { + return nil, errors.New("_id can only be the first field in a select") + } + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit") + } + + orderByFlds, orderByDirs, err := extractOrderBy(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting order by") + } + + mr := &MappingResult{ + IndexName: indexName, + //FieldFilters: fields, + Header: selectFields, + } + + // TODO: assign headers + mr.addReducer(NewAssignHeadersReducer(selectFields)) + + // TODO: If both order and limit/offset are required, then + // we can't supply limit/offset to the InspectRequest; we + // have to get all records, which we don't want to do on + // a large data set. We need to come up with a better + // way to handle that situation. + switch { + case qm.HasOrderBy(): + mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset)) + case limit != 0: + whereQuery = Limit(whereQuery, limit, offset) + case offset != 0: + whereQuery = Offset(whereQuery, offset) + } + + if len(fields) > 0 && fields[0] == "_id" { + fields = fields[1:] + } + mr.Query = Extract(whereQuery, fields...) + + return mr, nil +} + +// handlerSelectDistinctFromTable: Rows, Rows(limit): select distinct fld from tbl +type handlerSelectDistinctFromTable struct{} + +func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + selectFields, _, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + fieldCol, ok := selectFields[0].(*FieldColumn) + if !ok { + return nil, errors.New("distinct requires a valid field column") + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit") + } + + orderByFlds, orderByDirs, err := extractOrderBy(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting order by") + } + + // Determine the type of the field needing distinct. + // If the pilosa field is type int, handle it as a Distinct() query. + // Otherwise, use Rows() + // TODO: ensure this works for all field types (bool, time, etc). + var qo string + if fieldCol.Field.Type() == pilosa.FieldTypeInt { + qo = Distinct(fieldCol.Field.Index(), fieldCol.Field.Name()) + } else { + if !qm.HasOrderBy() && limit > 0 { + if qo, err = RowsLimit(fieldCol.Field.Name(), int64(limit)); err != nil { + return nil, errors.Wrap(err, "creating Rows query") + } + } else { + qo = Rows(fieldCol.Field.Name()) + } + } + + mr := &MappingResult{ + IndexName: indexName, + Header: selectFields, + Query: qo, + } + + mr.addReducer(NewAssignHeadersReducer(selectFields)) + if qm.HasOrderBy() { + mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset)) + } else { + mr.addReducer(NewLimitReducer(limit, offset)) + } + + return mr, nil +} + +// handlerSelectCountFromTableWhere: Count() +type handlerSelectCountFromTableWhere struct{} + +func (h handlerSelectCountFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + var qo string + var reducers []Reducer + + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + selectFields, features, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + var wherePQL string + if stmt.Where != nil { + wherePQL, err = extractWhere(index, stmt.Where.Expr) + if err != nil { + return nil, err + } + } else { + wherePQL = All() + } + + funcs := features.funcs + if len(funcs) != 1 { + return nil, errors.New("handler does not support multiple functions") + } else if funcs[0].funcName != FuncCount { + return nil, errors.Errorf("handler expected func: %s", FuncCount) + } + + if funcs[0].field == nil { + qo = Count(wherePQL) + } else { + // TODO: add the Distinct (for Int fields) here (like we do in handlerSelectDistinctFromTable) + qo = Rows(funcs[0].field.Name()) + reducers = append(reducers, &CountIDReducer{}) + } + mr := &MappingResult{ + IndexName: indexName, + Header: selectFields, + Query: qo, + Reducers: reducers, + } + + mr.addReducer(NewAssignHeadersReducer(selectFields)) + // NOTE: limit and order by don't make sense in this handler + // because it just returns a single row. + + return mr, nil +} + +// handlerSelectFuncFromTableWhere: min(), max(), sum(), avg() +type handlerSelectFuncFromTableWhere struct{} + +func (h handlerSelectFuncFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + var qo string + + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + selectFields, features, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + funcs := features.funcs + if len(funcs) != 1 { + return nil, errors.New("handler does not support multiple functions") + } + + funcField := funcs[0].field + if funcField == nil { + return nil, errors.New("function contains no field") + } + + var wherePQL string + if qm.HasWhere() { + wherePQL, err = extractWhere(index, stmt.Where.Expr) + if err != nil { + return nil, err + } + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit offset") + } + + orderByFlds, orderByDirs, err := extractOrderBy(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting order by") + } + + switch funcs[0].funcName { + case FuncMin: + qo = Min(funcField.Name(), wherePQL) + case FuncMax: + qo = Max(funcField.Name(), wherePQL) + case FuncAvg: + fallthrough + case FuncSum: + qo = Sum(funcField.Name(), wherePQL) + } + + mr := &MappingResult{ + IndexName: indexName, + Header: selectFields, + Query: qo, + } + + mr.addReducer(NewValCountFuncReducer(funcs[0].funcName)) + mr.addReducer(NewAssignHeadersReducer(selectFields)) + if qm.HasOrderBy() { + mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset)) + } else { + mr.addReducer(NewLimitReducer(limit, offset)) + } + + return mr, nil +} + +// handlerSelectGroupBy: GroupBy +type handlerSelectGroupBy struct{} + +func (h handlerSelectGroupBy) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + var qo string + + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + selectFields, features, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + orderByFlds, orderByDirs, err := extractOrderBy(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting order by") + } + + // If the query can be supported by TopN, + // i.e. if it's of the form: + // select fld, count(fld) as cnt from tbl group by fld order by cnt desc limit 1 + // select fld, count(fld) as cnt from tbl where fld2=1 group by fld order by cnt desc limit 1 + // then redirect it to handlerSelectIDCountFromTable. + // Otherwise, handle it as a normal GroupBy query. + // TODO: this level of inspection on the query needs to be built into + // an official query planner. The existing solution, which uses very + // broad masks to route the query to specific handlers, doesn't do + // this kind of finer-grain inspection of, for example, the order by + // fields themselves. + if func() bool { + if !qm.HasLimit() { + return false + } + if len(orderByFlds) != 1 { + return false + } + if orderByDirs[0] != "desc" { + return false + } + if qm == MustGenerateMask("select fld, count(fld) from tbl group by fld order by cnt limit 1") || + qm == MustGenerateMask("select fld, count(fld) from tbl where fld=1 group by fld order by cnt limit 1") { + // Check that the order-by field is the count field. + for i := range selectFields { + if s, ok := selectFields[i].(*FuncColumn); !ok { + continue + } else if s.FuncName == FuncCount && s.Alias() == orderByFlds[0] { + return true + } + } + } + return false + }() { + return handlerSelectIDCountFromTable{}.Apply(stmt, qm, indexFunc) + } + + groupByFieldNames, err := extractGroupByFieldNames(stmt.GroupBy) + if err != nil { + return nil, errors.Wrap(err, "extracting group by fields") + } + + having, err := extractHavingClause(stmt.Having) + if err != nil { + return nil, errors.Wrap(err, "extracting having clause") + } + + rowsQueries := []string{} + for _, fieldName := range groupByFieldNames { + field := index.Field(fieldName) + rowsQueries = append(rowsQueries, Rows(field.Name())) + } + + var wherePQL string + if stmt.Where != nil { + wherePQL, err = extractWhere(index, stmt.Where.Expr) + if err != nil { + return nil, errors.Wrap(err, "extracting where") + } + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit offset") + } + + // Group by queries can any combination of count() and sum() + // in the select fields. + var idxSum int = -1 + funcs := features.funcs + for i := range funcs { + switch funcs[i].funcName { + case FuncSum: + idxSum = i + } + } + + var sumQuery string + if idxSum >= 0 { + sumQuery = Sum(funcs[idxSum].field.Name(), "") + } + + var havingQuery string + if having != nil { + havingQuery = fmt.Sprintf("Condition(%s)", having.Cond.StringWithSubj(having.Subj)) + } + + qo, err = GroupByBase(rowsQueries, int64(limit+offset), wherePQL, sumQuery, havingQuery) + if err != nil { + return nil, err + } + + mr := &MappingResult{ + IndexName: indexName, + Header: selectFields, + Query: qo, + } + + mr.addReducer(NewAssignHeadersReducer(selectFields)) + if qm.HasOrderBy() { + mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset)) + } else { + mr.addReducer(NewLimitReducer(limit, offset)) + } + + return mr, nil +} + +// handlerSelectIDCountFromTable: TopN +type handlerSelectIDCountFromTable struct{} + +func (f handlerSelectIDCountFromTable) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + var qo string + + indexName, err := extractIndexName(stmt) + if err != nil { + return nil, errors.Wrapf(err, "extracting index name") + } + index := indexFunc(indexName) + if index == nil { + return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName) + } + + selectFields, features, err := extractSelectFields(index, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + var wherePQL string + if stmt.Where != nil { + wherePQL, err = extractWhere(index, stmt.Where.Expr) + if err != nil { + return nil, errors.Wrap(err, "extracting where") + } + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit offset") + } + + funcs := features.funcs + if len(funcs) != 1 { + return nil, errors.New("handler does not support multiple functions") + } else if funcs[0].funcName != FuncCount { + return nil, errors.Errorf("handler expected func: %s", FuncCount) + } + + if wherePQL == "" { + qo = TopN(funcs[0].field.Name(), uint64(limit+offset)) + } else { + qo = RowTopN(funcs[0].field.Name(), uint64(limit+offset), wherePQL) + } + + mr := &MappingResult{ + IndexName: indexName, + Header: selectFields, + Query: qo, + } + + mr.addReducer(NewAssignHeadersReducer(selectFields)) + mr.addReducer(NewLimitReducer(limit, offset)) + // TODO: order by is not implemented on this method because order desc + // is handled in pilosa TopN. In order to support asc here, we would + // have to return the entire TopN cache. Instead, we should consider + // supported something like this in Pilosa itself. + + return mr, nil +} + +// handlerSelectJoin: Join/Distinct() +type handlerSelectJoin struct{} + +func (h handlerSelectJoin) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc func(string) *pilosa.Index) (*MappingResult, error) { + var qo string + + pts, err := extractJoinTables(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting join tables") + } + + primary := pts.primary() + secondary := pts.secondary() + + primaryIndexName := primary.name + primaryIndex := indexFunc(primaryIndexName) + primaryField := primaryIndex.Field(primary.column.name) + + secondaryIndexName := secondary.name + secondaryIndex := indexFunc(secondaryIndexName) + secondaryField := secondaryIndex.Field(secondary.column.name) + + var wheres tableWheres + if qm.HasWhere() { + indexes := []*pilosa.Index{primaryIndex, secondaryIndex} + wheres, err = extractWheres(indexes, pts, stmt.Where.Expr) + if err != nil { + return nil, err + } + } + + var primaryWhere string + var secondaryWhere string + for i, w := range wheres { + switch w.table.index { + case primaryIndex: + primaryWhere = wheres[i].where + case secondaryIndex: + secondaryWhere = wheres[i].where + } + } + + // Build the Distinct() portion of the query on the secondary. + var distinctQry string + if secondaryWhere == "" { + distinctQry = Distinct(secondaryField.Index(), secondaryField.Name()) + } else { + distinctQry = RowDistinct(secondaryField.Index(), secondaryField.Name(), secondaryWhere) + } + + var rowQry string + if primaryWhere == "" { + rowQry = Intersect(All(), distinctQry) + } else { + _ = primaryField + rowQry = Intersect(primaryWhere, distinctQry) + } + + selectFields, _, err := extractSelectFields(primaryIndex, stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting select fields") + } + + orderByFlds, orderByDirs, err := extractOrderBy(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting order by") + } + + if qm.HasSelectPart(SelectPartCountStar) { + qo = Count(rowQry) + } else { + qo = rowQry + } + + limit, offset, err := extractLimitOffset(stmt) + if err != nil { + return nil, errors.Wrap(err, "extracting limit") + } + + mr := &MappingResult{ + IndexName: primaryIndex.Name(), + Header: selectFields, + Query: qo, + } + + mr.addReducer(NewAssignHeadersReducer(selectFields)) + if qm.HasOrderBy() { + mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset)) + } else { + mr.addReducer(NewLimitReducer(limit, offset)) + } + + return mr, nil +}