scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

porting sqlmapper from vdsm

Browse source
This commit is contained in:
Kuba Podgórski 2020-08-11 17:53:36 +02:00
parent 039780a85e
commit 2b34976c22
19 changed files with 5134 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

123
executor.go
View file

@ -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) {

80
executor_test.go
View file

@ -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{

30
field.go
View file

@ -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]

1
go.mod
View file

@ -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

2
go.sum
View file

@ -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=

77
proto/interface.go
View file

@ -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{

96
server/grpc.go
View file

@ -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()),

581
server/grpc_test.go
View file

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

112
sql/column.go Normal file
View file

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

1177
sql/extract.go Normal file
View file

File diff suppressed because it is too large Load diff

113
sql/mapper.go Normal file
View file

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

394
sql/mapper_test.go Normal file
View file

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

399
sql/mask.go Normal file
View file

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

141
sql/model.go Normal file
View file

@ -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()
}

295
sql/query.go Normal file
View file

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

500
sql/reduce.go Normal file
View file

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

120
sql/reduce_test.go Normal file
View file

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

162
sql/router.go Normal file
View file

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

792
sql/select.go Normal file
View file

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