diff --git a/api.go b/api.go index d8cdef042..8d3bfcea5 100644 --- a/api.go +++ b/api.go @@ -1114,7 +1114,28 @@ func (api *API) ImportAtomicRecord(ctx context.Context, req *AtomicRecord, opts return tx.Commit() } +// This is a hide your face ugly hack, forced upon +// us by the horrible invention of function based options +// by the usually brilliant Rob Pike. - JEA +func addClearToImportOptions(opts []ImportOption) []ImportOption { + var opt ImportOptions + for _, o := range opts { + // check for side-effect of setting io.Clear; that is + // how we know it is present. + _ = o(&opt) + if opt.Clear { + // we already have the clear flag set, so nothing more to do. + return opts + } + } + // no clear flag being set, add that option now. + return append(opts, OptImportOptionsClear(true)) +} + func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error { + if req.Clear { + opts = addClearToImportOptions(opts) + } return api.ImportWithTx(ctx, nil, req, opts...) } @@ -1254,6 +1275,9 @@ func (api *API) ImportWithTx(ctx context.Context, tx Tx, req *ImportRequest, opt } func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error { + if req.Clear { + opts = addClearToImportOptions(opts) + } return api.ImportValueWithTx(ctx, nil, req, opts...) } diff --git a/api_test.go b/api_test.go index f592afb6e..88bb65510 100644 --- a/api_test.go +++ b/api_test.go @@ -474,3 +474,127 @@ type offsetModHasher struct{} func (*offsetModHasher) Hash(key uint64, n int) int { return int(key+1) % n } + +func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) { + c := test.MustRunCluster(t, 1, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + // plan: + // 1. set a bit + // 2. clear with Import() using the ImportRequest.Clear flag + // 3. verifiy the clear is done. + // repeat for ImportValueRequest and ImportValues() + + m0 := c[0] + m0api := m0.API + + ctx := context.Background() + index := "i" + fieldAcct0 := "acct0" + + opts := pilosa.OptFieldTypeInt(-1000, 1000) + + _, err := m0api.CreateIndex(ctx, index, pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m0api.CreateField(ctx, index, fieldAcct0, opts) + if err != nil { + t.Fatalf("creating fieldAcct0: %v", err) + } + + iraField := "ira" // set field. + iraRowID := uint64(3) + _, err = m0api.CreateField(ctx, index, iraField) + if err != nil { + t.Fatalf("creating fieldIRA: %v", err) + } + + acctOwnerID := uint64(78) // ColumnID + shard := acctOwnerID / ShardWidth + acct0bal := int64(500) + + ivr0 := &pilosa.ImportValueRequest{ + Index: index, + Field: fieldAcct0, + Shard: shard, + ColumnIDs: []uint64{acctOwnerID}, + Values: []int64{acct0bal}, + } + ir0 := &pilosa.ImportRequest{ + Index: index, + Field: iraField, + Shard: shard, + ColumnIDs: []uint64{acctOwnerID}, + RowIDs: []uint64{iraRowID}, + } + + if err := m0api.Import(ctx, ir0); err != nil { + t.Fatal(err) + } + if err := m0api.ImportValue(ctx, ivr0); err != nil { + t.Fatal(err) + } + + bitIsSet := func() bool { + query := fmt.Sprintf("Row(%v=%v)", iraField, iraRowID) + res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + panicOn(err) + cols := res.Results[0].(*pilosa.Row).Columns() + for i := range cols { + if cols[i] == acctOwnerID { + return true + } + } + return false + } + + if !bitIsSet() { + panic("IRA bit should have been set") + } + + queryAcct := func(m0api *pilosa.API, acctOwnerID uint64, fieldAcct0, index string) (acctBal int64) { + query := fmt.Sprintf("FieldValue(field=%v, column=%v)", fieldAcct0, acctOwnerID) + res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + panicOn(err) + + if len(res.Results) == 0 { + return 0 + } + valCount := res.Results[0].(pilosa.ValCount) + return valCount.Val + } + + bal := queryAcct(m0api, acctOwnerID, fieldAcct0, index) + + if bal != acct0bal { + panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, bal)) + } + + // clear the bit + ir0.Clear = true + if err := m0api.Import(ctx, ir0); err != nil { + t.Fatal(err) + } + + if bitIsSet() { + panic("IRA bit should have been cleared") + } + + // clear the BSI + ivr0.Clear = true + if err := m0api.ImportValue(ctx, ivr0); err != nil { + t.Fatal(err) + } + bal = queryAcct(m0api, acctOwnerID, fieldAcct0, index) + if bal != 0 { + panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, 0)) + } +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 8ef6d5909..d033a24e6 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -29,6 +29,7 @@ import ( "runtime" "runtime/debug" "sort" + "strings" "testing" "testing/quick" @@ -190,6 +191,8 @@ func TestFragment_RowcacheMap(t *testing.T) { // Ensure a fragment can clear a row. func TestFragment_ClearRow(t *testing.T) { + notBlueGreenTest(t) + f, idx := mustOpenFragment("i", "f", viewStandard, 0, "") _ = idx defer f.Clean(t) @@ -225,6 +228,7 @@ func TestFragment_ClearRow(t *testing.T) { // Ensure a fragment can set a row. func TestFragment_SetRow(t *testing.T) { + notBlueGreenTest(t) f, idx := mustOpenFragment("i", "f", viewStandard, 7, "") _ = idx defer f.Clean(t) @@ -5644,3 +5648,12 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) { t.Fatalf("expected nothing got %v", res) } } + +func notBlueGreenTest(t *testing.T) { + src := os.Getenv("PILOSA_TXSRC") + if strings.Contains(src, "_") { + if strings.Contains(src, "roaring") { + t.Skip("skip under blue green with roaring") + } + } +} diff --git a/http/translator.go b/http/translator.go index f95067f70..0722dcbf9 100644 --- a/http/translator.go +++ b/http/translator.go @@ -22,19 +22,26 @@ import ( "io" "io/ioutil" "net/http" + "sync" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/logger" ) func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderFunc { + return GetOpenTranslateReaderWithLockerFunc(client, nopLocker{}) +} + +func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc { return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) { - return openTranslateReader(ctx, nodeURL, offsets, client) + return openTranslateReader(ctx, nodeURL, offsets, client, locker) } } -func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client) (pilosa.TranslateEntryReader, error) { +func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client, locker sync.Locker) (pilosa.TranslateEntryReader, error) { r := NewTranslateEntryReader(ctx, client) + r.locker = locker + r.URL = nodeURL + "/internal/translate/data" r.Offsets = offsets if err := r.Open(); err != nil { @@ -43,9 +50,16 @@ func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.Tra return r, nil } +type nopLocker struct{} + +func (nopLocker) Lock() {} +func (nopLocker) Unlock() {} + // TranslateEntryReader represents an implementation of pilosa.TranslateEntryReader. // It consolidates all index & field translate entries into a single reader. type TranslateEntryReader struct { + locker sync.Locker + ctx context.Context cancel func() @@ -70,7 +84,7 @@ func NewTranslateEntryReader(ctx context.Context, client *http.Client) *Translat if client == nil { client = http.DefaultClient } - r := &TranslateEntryReader{HTTPClient: client, Logger: logger.NopLogger} + r := &TranslateEntryReader{locker: nopLocker{}, HTTPClient: client, Logger: logger.NopLogger} r.ctx, r.cancel = context.WithCancel(ctx) return r } @@ -116,7 +130,10 @@ func (r *TranslateEntryReader) Close() error { r.cancel() } if r.body != nil { - return r.body.Close() + r.locker.Lock() + err := r.body.Close() + r.locker.Unlock() + return err } return nil } @@ -124,5 +141,8 @@ func (r *TranslateEntryReader) Close() error { // ReadEntry reads the next entry from the stream into entry. // Returns io.EOF at the end of the stream. func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error { + r.locker.Lock() + defer r.locker.Unlock() + return r.dec.Decode(&entry) } diff --git a/http/translator_test.go b/http/translator_test.go index be6db423a..8d4a9362d 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -17,6 +17,7 @@ package http_test import ( "context" "fmt" + "sync" "testing" "time" @@ -151,3 +152,84 @@ func TestTranslateStore_EntryReader(t *testing.T) { }) */ } + +func benchmarkSetup(b *testing.B, ctx context.Context, key string, nkeys int) (string, pilosa.TranslateOffsetMap, func()) { + b.Helper() + + cluster := test.MustRunCluster(b, 1) + primary := cluster[0] + + idx := primary.MustCreateIndex(b, "i", pilosa.IndexOptions{}) + fld := primary.MustCreateField(b, idx.Name(), "f", pilosa.OptFieldKeys()) + offset := make(pilosa.TranslateOffsetMap) + offset.SetIndexPartitionOffset(idx.Name(), 0, 1) + offset.SetFieldOffset(idx.Name(), fld.Name(), 1) + + // Set data on the primary node. + for k := 0; k < nkeys; k++ { + if _, err := primary.API.Query(ctx, &pilosa.QueryRequest{ + Index: idx.Name(), + Query: fmt.Sprintf(`Set(%d, %s="%s%[1]d")`, k, fld.Name(), key), + }); err != nil { + b.Fatalf("quering api: %+v", err) + } + } + + return primary.URL(), offset, func() { + b.Helper() + + if err := primary.API.DeleteIndex(ctx, idx.Name()); err != nil { + panic(err) + } + if err := cluster.Close(); err != nil { + panic(err) + } + } +} + +func benchmarkReadEntry(b *testing.B, r pilosa.TranslateEntryReader, key string, nkeys int) { + var entry pilosa.TranslateEntry + for k := 0; k < nkeys; k++ { + if err := r.ReadEntry(&entry); err != nil { + b.Fatalf("reading entry: %+v", err) + } + if entry.Key != fmt.Sprintf("%s%d", key, k) { + b.Fatalf("got: %s, expected: %s%d", entry.Key, key, k) + } + } +} + +const ( + key = "foo" + nkeys = 1000 +) + +func BenchmarkReadEntryNoMutex(b *testing.B) { + ctx := context.Background() + url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys) + defer teardown() + + for n := 0; n < b.N; n++ { + r, err := http.GetOpenTranslateReaderFunc(nil)(ctx, url, offset) + if err != nil { + b.Fatalf("opening translate reader: %+v", err) + } + benchmarkReadEntry(b, r, key, nkeys) + r.Close() + } +} + +func BenchmarkReadEntryWithMutex(b *testing.B) { + ctx := context.Background() + url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys) + defer teardown() + + for n := 0; n < b.N; n++ { + r, err := http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})(ctx, url, offset) + if err != nil { + b.Fatalf("opening translate reader: %+v", err) + } + benchmarkReadEntry(b, r, key, nkeys) + r.Close() + } +} diff --git a/lmdb.go b/lmdb.go index 9e0d30ed9..b67436abf 100644 --- a/lmdb.go +++ b/lmdb.go @@ -147,9 +147,9 @@ func (r *lmdbRegistrar) openLMDBWrapper(path0 string) (*LMDBWrapper, error) { flags = flags | lmdb.WriteMap | // Use a writable memory map. - lmdb.NoMetaSync | // Don't fsync metapage after commit. - lmdb.NoSync | // Don't fsync after commit. - lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag. + //lmdb.NoMetaSync | // Don't fsync metapage after commit. + //lmdb.NoSync | // Don't fsync after commit. + //lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag. lmdb.NoMemInit // Disable LMDB memory initialization err = env.Open(path, flags, 0644) @@ -344,7 +344,7 @@ func (tx *LMDBTx) Type() string { } func (tx *LMDBTx) UseRowCache() bool { - return false + return true } // Pointer gives us a memory address for the underlying transaction for debugging. diff --git a/pg/server_test.go b/pg/server_test.go index 4903fdb60..9eda44bcf 100644 --- a/pg/server_test.go +++ b/pg/server_test.go @@ -59,7 +59,7 @@ func TestStartupInvalidLength(t *testing.T) { res := testing.Benchmark(func(b *testing.B) { connect, shutdown, err := pgtest.ServeMem(&pg.Server{ MaxStartupSize: 1024, - Logger: logger.NewLogfLogger(t), + Logger: logger.NopLogger, }) if err != nil { t.Fatalf("starting in-memory postgres server: %v", err) diff --git a/proto/interface.go b/proto/interface.go index 660aa6273..6a6df7213 100644 --- a/proto/interface.go +++ b/proto/interface.go @@ -31,6 +31,15 @@ type StreamClient interface { Recv() (*RowResponse, error) } +// EmptyStream implements StreamClient interface. +// It always returns empty RowResponse +type EmptyStream struct{} + +// Recv returns io.EOF +func (EmptyStream) Recv() (*RowResponse, error) { + return nil, io.EOF +} + // ReadIntoTable reads from a StreamClient and stores the result into a table response. func ReadIntoTable(cli StreamClient) (*TableResponse, error) { var headers []*ColumnInfo diff --git a/server/grpc.go b/server/grpc.go index 3dcc420a1..af11b6f3c 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -26,7 +26,6 @@ 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" @@ -120,24 +119,7 @@ func (h *GRPCHandler) DeleteVDS(ctx context.Context, req *pb.DeleteVDSRequest) ( } 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 + return execSQL(ctx, h.api, h.logger, queryStr) } // QuerySQL handles the SQL request and sends RowResponses to the stream. diff --git a/server/grpc_test.go b/server/grpc_test.go index e143c9168..4529c0df8 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -686,7 +686,6 @@ func TestQuerySQLUnary(t *testing.T) { }, eq: equalUnordered, }, - { // GroupBy(Rows(field='age'),limit=3) sql: "select age, count(*) as cnt from grouper group by age order by cnt desc, age desc limit 3", @@ -703,6 +702,35 @@ func TestQuerySQLUnary(t *testing.T) { }, eq: equal, }, + { + sql: "show tables", + exp: tableResponse{ + headers: []columnInfo{ + {"Table", "string"}, + }, + rows: []row{ + {[]columnResponse{"grouper"}}, + {[]columnResponse{"joiner"}}, + }, + }, + eq: equal, + }, + { + sql: "show fields from grouper", + exp: tableResponse{ + headers: []columnInfo{ + {"Field", "string"}, + {"Type", "string"}, + }, + rows: []row{ + {[]columnResponse{"age", "int64"}}, + {[]columnResponse{"color", "[]string"}}, + {[]columnResponse{"height", "int64"}}, + {[]columnResponse{"score", "int64"}}, + }, + }, + eq: equal, + }, } for i, test := range tests { diff --git a/server/handler_test.go b/server/handler_test.go index 31954eff8..ccd91bdcc 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -27,6 +27,7 @@ import ( "net/http/httptest" "reflect" "strings" + "sync" "testing" "time" @@ -1300,7 +1301,7 @@ func TestCluster_TranslateStore(t *testing.T) { cluster[0] = test.NewCommandNode(true, server.OptCommandServerOptions( pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})), ), ) cluster[0].Config.Gossip.Port = "0" @@ -1329,7 +1330,7 @@ func TestClusterTranslator(t *testing.T) { cluster[1] = test.NewCommandNode(false, server.OptCommandServerOptions( pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})), ), ) cluster[1].Config.Gossip.Port = "0" diff --git a/server/pg.go b/server/pg.go index 98117c374..28bbef7af 100644 --- a/server/pg.go +++ b/server/pg.go @@ -19,6 +19,7 @@ import ( "crypto/tls" "encoding/json" "fmt" + "io" "net" "strconv" "strings" @@ -50,7 +51,8 @@ func NewPostgresServer(api *pilosa.API, logger logger.Logger, tls *tls.Config) * s: pg.Server{ QueryHandler: &queryDecodeHandler{ child: &pilosaQueryHandler{ - api: api, + api: api, + logger: logger, }, }, TypeEngine: pg.PrimitiveTypeEngine{}, @@ -125,7 +127,8 @@ func pgDecodePQL(str string) (q pg.Query, err error) { } type pilosaQueryHandler struct { - api *pilosa.API + api *pilosa.API + logger logger.Logger } func pgWriteRow(w pg.QueryResultWriter, row *pilosa.Row) error { @@ -342,6 +345,28 @@ func pgWriteRowser(w pg.QueryResultWriter, result pb.ToRowser) error { }) } +type clientRowser struct { + pb.StreamClient +} + +func (cr *clientRowser) ToRows(f func(*pb.RowResponse) error) error { + for { + resp, err := cr.StreamClient.Recv() + if err != nil { + if err == io.EOF { + return nil + } + + return err + } + + err = f(resp) + if err != nil { + return err + } + } +} + func pgWriteResult(w pg.QueryResultWriter, result interface{}) error { switch result := result.(type) { case *pilosa.Row: @@ -354,6 +379,8 @@ func pgWriteResult(w pg.QueryResultWriter, result interface{}) error { return pgWriteGroupCount(w, result) case pb.ToRowser: // we should avoid protobuf where we can... return pgWriteRowser(w, result) + case pb.StreamClient: + return pgWriteRowser(w, &clientRowser{result}) default: return errors.Errorf("result type %T not yet supported", result) } @@ -374,6 +401,13 @@ func (pqh *pilosaQueryHandler) HandleQuery(ctx context.Context, w pg.QueryResult } return errors.Wrap(pgWriteResult(w, resp.Results[0]), "writing query result") + case pg.SimpleQuery: + resp, err := execSQL(ctx, pqh.api, pqh.logger, string(q)) + if err != nil { + return errors.Wrap(err, "executing query") + } + return errors.Wrap(pgWriteResult(w, resp), "writing query result") + default: return errors.Errorf("query type %T not yet supported (query: %s)", q, q) } diff --git a/server/server.go b/server/server.go index 5578cf35f..e74d460e2 100644 --- a/server/server.go +++ b/server/server.go @@ -31,6 +31,7 @@ import ( "os/signal" "runtime" "strconv" + "sync" "syscall" "time" @@ -389,7 +390,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerDiagnosticsInterval(diagnosticsInterval), pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize), pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(c)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})), pilosa.OptServerLogger(m.logger), pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore), pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()), diff --git a/server/sql.go b/server/sql.go new file mode 100644 index 000000000..0f1de916c --- /dev/null +++ b/server/sql.go @@ -0,0 +1,54 @@ +// 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 server + +import ( + "context" + + "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/pkg/errors" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func execSQL(ctx context.Context, api *pilosa.API, logger logger.Logger, queryStr string) (pb.StreamClient, error) { + mapper := sql.NewMapper() + mapper.Logger = 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(api) + results, err = handler.Handle(ctx, query) + if err != nil { + return nil, errors.Wrap(err, "failed to start SQL query") + } + case sql.SQLTypeShow: + handler := sql.NewShowHandler(api) + results, err = handler.Handle(ctx, query) + if err != nil { + return nil, errors.Wrap(err, "failed to start SQL query") + } + default: + return nil, status.Errorf(codes.Unimplemented, "query type not supported") + } + return results, nil +} diff --git a/sql/select.go b/sql/select.go index 1c11cdfca..a195a7c85 100644 --- a/sql/select.go +++ b/sql/select.go @@ -42,7 +42,11 @@ func NewSelectHandler(api *pilosa.API) *SelectHandler { // Handle executes mapped SQL func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) { - mr, err := s.mapSelect(ctx, mapped.Statement.(*sqlparser.Select), mapped.Mask) + stmt, ok := mapped.Statement.(*sqlparser.Select) + if !ok { + return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement) + } + mr, err := s.mapSelect(ctx, stmt, mapped.Mask) if err != nil { return nil, errors.Wrap(err, "mapping select") } @@ -70,12 +74,11 @@ func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Sel return mr, nil } -func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (stream pproto.StreamClient, err error) { +func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.StreamClient, error) { if mr.Query == "" { return nil, errors.New("no pql query created") } - fmt.Println("PQL:", mr.Query) resp, err := s.api.Query(ctx, &pilosa.QueryRequest{Index: mr.IndexName, Query: mr.Query}) if err != nil { return nil, errors.Wrap(err, "doing pql query") diff --git a/sql/show.go b/sql/show.go new file mode 100644 index 000000000..4ac3c390d --- /dev/null +++ b/sql/show.go @@ -0,0 +1,148 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sql + +import ( + "context" + "fmt" + + "github.com/pilosa/pilosa/v2" + pproto "github.com/pilosa/pilosa/v2/proto" + "github.com/pkg/errors" + "vitess.io/vitess/go/vt/sqlparser" +) + +// ShowHandler executes SQL show table/field statements +type ShowHandler struct { + api *pilosa.API +} + +// NewShowHandler constructor +func NewShowHandler(api *pilosa.API) *ShowHandler { + return &ShowHandler{ + api: api, + } +} + +// Handle executes mapped SQL +func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) { + stmt, ok := mapped.Statement.(*sqlparser.Show) + if !ok { + return nil, fmt.Errorf("statement is not type show: %T", mapped.Statement) + } + + switch stmt.Type { + case "tables": + return s.execShowTables(ctx, stmt) + case "fields": + return s.execShowFields(ctx, stmt) + default: + return nil, fmt.Errorf("cannot show: %s", stmt.Type) + } +} + +func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) { + indexInfo := s.api.Schema(ctx) + sz := len(indexInfo) + // If there aren't any indexes, don't bother creating + // a result row buffer. + if sz == 0 { + return pproto.EmptyStream{}, nil + } + + // Create a buffer large enough to hold the entire result + // set. This way we don't have to use a goroutine. + result := pproto.NewRowBuffer(sz) + for _, ii := range indexInfo { + rr := &pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: "Table", Datatype: "string"}, + }, + Columns: []*pproto.ColumnResponse{ + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: ii.Name}}, + }, + } + if err := result.Send(rr); err != nil { + return nil, errors.Wrap(err, "sending row response") + } + } + if err := result.Send(pproto.EOF); err != nil { + return nil, errors.Wrap(err, "sending EOF") + } + + // Apply Sort Reducer + out := pproto.NewRowBuffer(0) + red := NewOrderByReducer([]string{"Table"}, []string{"asc"}, 0, 0) + go red.Reduce(result, out) //nolint:errcheck + + result = out + return result, nil +} + +func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) { + indexName := showStmt.OnTable.ToViewName().Name.String() + index, err := s.api.Index(ctx, indexName) + if err != nil { + return nil, errors.Wrap(err, "getting schema") + } + if index == nil { + return nil, pilosa.ErrIndexNotFound + } + fields := index.Fields() + sz := len(fields) + // If there aren't any fields, don't bother creating + // a result row buffer. + if sz == 0 { + return pproto.EmptyStream{}, nil + } + + // Create a buffer large enough to hold the entire result + // set. This way we don't have to use a goroutine. + result := pproto.NewRowBuffer(sz) + for _, f := range fields { + if f.Name() == "_exists" { + continue + } + + dt, err := f.Datatype() + if err != nil { + return nil, errors.Wrapf(err, "field %s", f.Name()) + } + rr := &pproto.RowResponse{ + Headers: []*pproto.ColumnInfo{ + {Name: "Field", Datatype: "string"}, + {Name: "Type", Datatype: "string"}, + }, + Columns: []*pproto.ColumnResponse{ + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: f.Name()}}, + {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: dt}}, + }, + } + if err := result.Send(rr); err != nil { + return nil, errors.Wrap(err, "sending row response") + } + } + if err := result.Send(pproto.EOF); err != nil { + return nil, errors.Wrap(err, "sending EOF") + } + + // Apply Sort Reducer + out := pproto.NewRowBuffer(0) + red := NewOrderByReducer([]string{"Field"}, []string{"asc"}, 0, 0) + go red.Reduce(result, out) //nolint:errcheck + + result = out + return result, nil +} diff --git a/tx_test.go b/tx_test.go index 300e3444b..ae5e0b984 100644 --- a/tx_test.go +++ b/tx_test.go @@ -59,7 +59,10 @@ func queryBalances(m0api *pilosa.API, acctOwnerID uint64, fldAcct0, fldAcct1, in return } func skipForRoaring(t *testing.T) { - if strings.Contains(os.Getenv("PILOSA_TXSRC"), "roaring") { + src := os.Getenv("PILOSA_TXSRC") + // once txfactory.go DefaultTxsrc != RoaringTxn, this + // will break, of course. Take out the src == "" below. + if src == "" || strings.Contains(src, "roaring") { t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback") } }