Merge branch 'master' into fix-rbf-race

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

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

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

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

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

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.

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)

proto/interface.go

@@ -31,6 +31,15 @@ type StreamClient interface {
 	Recv() (*RowResponse, error)
 }
 
+// EmptyStream implements StreamClient interface.
+// It always returns empty RowResponse
+type EmptyStream struct{}
+
+// Recv returns io.EOF
+func (EmptyStream) Recv() (*RowResponse, error) {
+	return nil, io.EOF
+}
+
 // ReadIntoTable reads from a StreamClient and stores the result into a table response.
 func ReadIntoTable(cli StreamClient) (*TableResponse, error) {
 	var headers []*ColumnInfo

server/grpc.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.

server/grpc_test.go

@@ -686,7 +686,6 @@ func TestQuerySQLUnary(t *testing.T) {
 			},
 			eq: equalUnordered,
 		},
-
 		{
 			// GroupBy(Rows(field='age'),limit=3)
 			sql: "select age, count(*) as cnt from grouper group by age order by cnt desc, age desc limit 3",
@@ -703,6 +702,35 @@ func TestQuerySQLUnary(t *testing.T) {
 			},
 			eq: equal,
 		},
+		{
+			sql: "show tables",
+			exp: tableResponse{
+				headers: []columnInfo{
+					{"Table", "string"},
+				},
+				rows: []row{
+					{[]columnResponse{"grouper"}},
+					{[]columnResponse{"joiner"}},
+				},
+			},
+			eq: equal,
+		},
+		{
+			sql: "show fields from grouper",
+			exp: tableResponse{
+				headers: []columnInfo{
+					{"Field", "string"},
+					{"Type", "string"},
+				},
+				rows: []row{
+					{[]columnResponse{"age", "int64"}},
+					{[]columnResponse{"color", "[]string"}},
+					{[]columnResponse{"height", "int64"}},
+					{[]columnResponse{"score", "int64"}},
+				},
+			},
+			eq: equal,
+		},
 	}
 
 	for i, test := range tests {

server/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"

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

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

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

sql/select.go

@@ -42,7 +42,11 @@ func NewSelectHandler(api *pilosa.API) *SelectHandler {
 
 // Handle executes mapped SQL
 func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) {
-	mr, err := s.mapSelect(ctx, mapped.Statement.(*sqlparser.Select), mapped.Mask)
+	stmt, ok := mapped.Statement.(*sqlparser.Select)
+	if !ok {
+		return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement)
+	}
+	mr, err := s.mapSelect(ctx, stmt, mapped.Mask)
 	if err != nil {
 		return nil, errors.Wrap(err, "mapping select")
 	}
@@ -70,12 +74,11 @@ func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Sel
 	return mr, nil
 }
 
-func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (stream pproto.StreamClient, err error) {
+func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.StreamClient, error) {
 	if mr.Query == "" {
 		return nil, errors.New("no pql query created")
 	}
 
-	fmt.Println("PQL:", mr.Query)
 	resp, err := s.api.Query(ctx, &pilosa.QueryRequest{Index: mr.IndexName, Query: mr.Query})
 	if err != nil {
 		return nil, errors.Wrap(err, "doing pql query")

sql/show.go

@@ -0,0 +1,148 @@
+// Copyright 2020 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sql
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pilosa/pilosa/v2"
+	pproto "github.com/pilosa/pilosa/v2/proto"
+	"github.com/pkg/errors"
+	"vitess.io/vitess/go/vt/sqlparser"
+)
+
+// ShowHandler executes SQL show table/field statements
+type ShowHandler struct {
+	api *pilosa.API
+}
+
+// NewShowHandler constructor
+func NewShowHandler(api *pilosa.API) *ShowHandler {
+	return &ShowHandler{
+		api: api,
+	}
+}
+
+// Handle executes mapped SQL
+func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) {
+	stmt, ok := mapped.Statement.(*sqlparser.Show)
+	if !ok {
+		return nil, fmt.Errorf("statement is not type show: %T", mapped.Statement)
+	}
+
+	switch stmt.Type {
+	case "tables":
+		return s.execShowTables(ctx, stmt)
+	case "fields":
+		return s.execShowFields(ctx, stmt)
+	default:
+		return nil, fmt.Errorf("cannot show: %s", stmt.Type)
+	}
+}
+
+func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) {
+	indexInfo := s.api.Schema(ctx)
+	sz := len(indexInfo)
+	// If there aren't any indexes, don't bother creating
+	// a result row buffer.
+	if sz == 0 {
+		return pproto.EmptyStream{}, nil
+	}
+
+	// Create a buffer large enough to hold the entire result
+	// set. This way we don't have to use a goroutine.
+	result := pproto.NewRowBuffer(sz)
+	for _, ii := range indexInfo {
+		rr := &pproto.RowResponse{
+			Headers: []*pproto.ColumnInfo{
+				{Name: "Table", Datatype: "string"},
+			},
+			Columns: []*pproto.ColumnResponse{
+				{ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: ii.Name}},
+			},
+		}
+		if err := result.Send(rr); err != nil {
+			return nil, errors.Wrap(err, "sending row response")
+		}
+	}
+	if err := result.Send(pproto.EOF); err != nil {
+		return nil, errors.Wrap(err, "sending EOF")
+	}
+
+	// Apply Sort Reducer
+	out := pproto.NewRowBuffer(0)
+	red := NewOrderByReducer([]string{"Table"}, []string{"asc"}, 0, 0)
+	go red.Reduce(result, out) //nolint:errcheck
+
+	result = out
+	return result, nil
+}
+
+func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) {
+	indexName := showStmt.OnTable.ToViewName().Name.String()
+	index, err := s.api.Index(ctx, indexName)
+	if err != nil {
+		return nil, errors.Wrap(err, "getting schema")
+	}
+	if index == nil {
+		return nil, pilosa.ErrIndexNotFound
+	}
+	fields := index.Fields()
+	sz := len(fields)
+	// If there aren't any fields, don't bother creating
+	// a result row buffer.
+	if sz == 0 {
+		return pproto.EmptyStream{}, nil
+	}
+
+	// Create a buffer large enough to hold the entire result
+	// set. This way we don't have to use a goroutine.
+	result := pproto.NewRowBuffer(sz)
+	for _, f := range fields {
+		if f.Name() == "_exists" {
+			continue
+		}
+
+		dt, err := f.Datatype()
+		if err != nil {
+			return nil, errors.Wrapf(err, "field %s", f.Name())
+		}
+		rr := &pproto.RowResponse{
+			Headers: []*pproto.ColumnInfo{
+				{Name: "Field", Datatype: "string"},
+				{Name: "Type", Datatype: "string"},
+			},
+			Columns: []*pproto.ColumnResponse{
+				{ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: f.Name()}},
+				{ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: dt}},
+			},
+		}
+		if err := result.Send(rr); err != nil {
+			return nil, errors.Wrap(err, "sending row response")
+		}
+	}
+	if err := result.Send(pproto.EOF); err != nil {
+		return nil, errors.Wrap(err, "sending EOF")
+	}
+
+	// Apply Sort Reducer
+	out := pproto.NewRowBuffer(0)
+	red := NewOrderByReducer([]string{"Field"}, []string{"asc"}, 0, 0)
+	go red.Reduce(result, out) //nolint:errcheck
+
+	result = out
+	return result, nil
+}

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