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

Merge branch 'master' into pg-local-cancel

Browse source
This commit is contained in:
Nia 2020-09-02 14:02:18 -04:00 • committed by GitHub
commit 219e22cc64
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
35 changed files with 1191 additions and 595 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -5,3 +5,4 @@ vendor
.DS_Store
build
*~
lattice

29
Makefile
View file

@ -1,7 +1,8 @@
.PHONY: build check-clean clean cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race
.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race
CLONE_URL=github.com/pilosa/pilosa
VERSION := $(shell git describe --tags 2> /dev/null || echo unknown)
LATTICE_COMMIT := $(shell git -C lattice rev-parse --short HEAD 2>/dev/null)
VARIANT = Molecula
VERSION_ID = $(VERSION)-$(GOOS)-$(GOARCH)
BRANCH := $(if $(TRAVIS_BRANCH),$(TRAVIS_BRANCH),$(if $(CIRCLE_BRANCH),$(CIRCLE_BRANCH),$(shell git rev-parse --abbrev-ref HEAD)))
@ -9,7 +10,7 @@ BRANCH_ID := $(BRANCH)-$(GOOS)-$(GOARCH)
BUILD_TIME := $(shell date -u +%FT%T%z)
SHARD_WIDTH = 20
COMMIT := $(shell git describe --exact-match >/dev/null 2>&1 || git rev-parse --short HEAD)
LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT)"
LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT) -X github.com/pilosa/pilosa/v2.LatticeCommit=$(LATTICE_COMMIT)"
GO_VERSION=latest
RELEASE ?= 0
RELEASE_ENABLED = $(subst 0,,$(RELEASE))
@ -100,10 +101,15 @@ ifndef SKIP_CHECK_CLEAN
endif
# Create release build tarballs for all supported platforms. Linux compilation happens under Docker.
release: check-clean
release: check-clean generate-statik
$(MAKE) release-build GOOS=darwin GOARCH=amd64
$(MAKE) release-build GOOS=linux GOARCH=amd64
# Create release build tarballs for all supported platforms. Same as `release`, but without embedded Lattice UI.
release-sans-ui: check-clean
rm -f statik/statik.go
$(MAKE) release-build GOOS=darwin GOARCH=amd64
$(MAKE) release-build GOOS=linux GOARCH=amd64
# try (e.g.) internal/clustertests/docker-compose-replication2.yml
DOCKER_COMPOSE=internal/clustertests/docker-compose.yml
@ -135,10 +141,20 @@ prerelease-upload:
install:
go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa
lattice:
git clone git@github.com:molecula/lattice.git
build-lattice: lattice require-yarn
cd lattice && git pull && yarn install && yarn build
# `go generate` protocol buffers
generate-protoc: require-protoc require-protoc-gen-gofast
go generate github.com/pilosa/pilosa/v2/internal
# `go generate` statik assets (lattice UI)
generate-statik: build-lattice require-statik
go generate github.com/pilosa/pilosa/v2/statik
# `go generate` stringers
generate-stringer:
go generate github.com/pilosa/pilosa/v2
@ -151,7 +167,7 @@ generate-proto-grpc: require-protoc require-protoc-gen-gofast
protoc -I proto proto/pilosa.proto --go_out=plugins=grpc:proto
# `go generate` all needed packages
generate: generate-protoc generate-stringer generate-pql
generate: generate-protoc generate-statik generate-stringer generate-pql
# Create Docker image from Dockerfile
docker: vendor
@ -347,7 +363,10 @@ require-%:
$(info Verified build dependency "$*" is installed.),\
$(error Build dependency "$*" not installed. To install, try `make install-$*`))
install-build-deps: install-protoc-gen-gofast install-protoc install-stringer install-peg
install-build-deps: install-protoc-gen-gofast install-protoc install-statik install-stringer install-peg
install-statik:
go get -u github.com/rakyll/statik
install-stringer:
GO111MODULE=off go get -u golang.org/x/tools/cmd/stringer

2
README.md
View file

@ -31,6 +31,8 @@ See our [Documentation](https://www.pilosa.com/docs/) for information about inst
1. [Install Pilosa](https://www.pilosa.com/docs/installation/).
Optionally, to include Lattice, the in-browser UI, follow the "Build from source" instructions, and run `make generate-statik` before `make install`. When you run a local Pilosa server on the default host, for example, you can access Lattice at [localhost:10101](http://localhost:10101).
2. [Start Pilosa](https://www.pilosa.com/docs/getting-started/#starting-pilosa) with the default configuration:
```shell

5
api.go
View file

@ -1626,6 +1626,11 @@ func (api *API) Version() string {
return strings.TrimPrefix(Version, "v")
}
// Version returns the Lattice version.
func (api *API) LatticeVersion() string {
return LatticeVersionInfo()
}
// Info returns information about this server instance.
func (api *API) Info() serverInfo {
si := api.server.systemInfo

25
cache.go
View file

@ -142,6 +142,7 @@ type rankCache struct {
entries map[uint64]uint64
rankings bitmapPairs // cached, ordered list
rankingsRead bool
dirty bool
updateN int
updateTime time.Time
@ -173,6 +174,11 @@ func NewRankCache(maxEntries uint32) *rankCache {
func (c *rankCache) Add(id uint64, n uint64) {
c.mu.Lock()
defer c.mu.Unlock()
// Flag the cache as dirty.
// This forces recalculation if top is called before the cache is recalculated.
c.dirty = true
// Ignore if the column count is below the threshold,
// unless the count is 0, which is effectively used
// to clear the cache value.
@ -190,6 +196,11 @@ func (c *rankCache) Add(id uint64, n uint64) {
func (c *rankCache) BulkAdd(id uint64, n uint64) {
c.mu.Lock()
defer c.mu.Unlock()
// Flag the cache as dirty.
// This forces recalculation if top is called before the cache is recalculated.
c.dirty = true
if n < c.thresholdValue {
delete(c.entries, id)
return
@ -246,6 +257,11 @@ func (c *rankCache) invalidate() {
// Don't invalidate more than once every X seconds.
// TODO: consider making this configurable.
if time.Since(c.updateTime).Seconds() < 10 {
// Skipping recalculation means that the ranked cache's growth is unbounded.
// This is somewhat necessary for now since recalculation is not cheap.
// The cache will remain flagged as dirty and will be recalculated if Top is called.
// This may cause unexpected memory growth, so record it in metrics for debugging purposes.
c.stats.Count(MetricInvalidateCacheSkipped, 1, 1.0)
return
}
c.stats.Count(MetricInvalidateCache, 1, 1.0)
@ -295,6 +311,9 @@ func (c *rankCache) recalculate() {
delete(c.entries, pair.ID)
}
}
// The cache is no longer dirty.
c.dirty = false
}
// SetStats defines the stats client used in the cache.
@ -307,6 +326,12 @@ func (c *rankCache) Top() []bitmapPair {
c.mu.Lock()
defer c.mu.Unlock()
if c.dirty {
// The cache is dirty, so we need to recalculate it to get a consistent view.
c.stats.Count(MetricReadDirtyCache, 1, 1.0)
c.recalculate()
}
c.rankingsRead = true
return c.rankings
}

29
cache_test.go
View file

@ -15,6 +15,7 @@
package pilosa_test
import (
"reflect"
"testing"
"github.com/pilosa/pilosa/v2"
@ -54,3 +55,31 @@ func TestCache_Rank_Threshold(t *testing.T) {
t.Fatalf("unexpected cache value after BulkAdd: %d!=%d expected\n", cache.Get(5), 0)
}
}
// Test that consecutive writes show up in Top.
// On later writes, the cache skips recalculation to save CPU time.
// This used to mean that the later writes would not show up in Top.
// Now, the cache is flagged as dirty and recalculated during the call to Top.
func TestCache_Rank_Dirty(t *testing.T) {
cacheSize := uint32(5)
cache := pilosa.NewRankCache(cacheSize)
type pair struct{ ID, Count uint64 }
expect := []pair{
{5, 2},
{4, 1},
}
for _, v := range expect {
cache.Add(v.ID, v.Count)
}
var got []pair
for _, p := range cache.Top() {
got = append(got, pair(p))
}
if !reflect.DeepEqual(expect, got) {
t.Fatalf("wrote %v but got %v", expect, got)
}
}

2
ctl/server.go
View file

@ -39,7 +39,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
SetTLSConfig(flags, "", &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.CACertPath, &srv.Config.TLS.SkipVerify, &srv.Config.TLS.EnableClientVerification)
// Handler
flags.StringSliceVarP(&srv.Config.Handler.AllowedOrigins, "handler.allowed-origins", "", []string{}, "Comma separated list of allowed origin URIs (for CORS/WebUI).")
flags.StringSliceVarP(&srv.Config.Handler.AllowedOrigins, "handler.allowed-origins", "", []string{}, "Comma separated list of allowed origin URIs (for CORS/Web UI).")
// Cluster
flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)")

4
executor.go
View file

@ -90,7 +90,7 @@ func emptyResult(c *pql.Call) interface{} {
return false
case "Row":
return Row{Keys: []string{}}
return &Row{Keys: []string{}}
case "Rows":
return RowIdentifiers{Keys: []string{}}
@ -2899,6 +2899,8 @@ func (t ExtractedTable) ToRows(callback func(*pb.RowResponse) error) error {
for i, r := range c.Rows {
var col *pb.ColumnResponse
switch r := r.(type) {
case nil:
col = &pb.ColumnResponse{}
case bool:
col = &pb.ColumnResponse{
ColumnVal: &pb.ColumnResponse_BoolVal{

42
filesystem.go Normal file
View file

@ -0,0 +1,42 @@
// Copyright 2017 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 pilosa
import (
"fmt"
"net/http"
)
// Ensure nopFileSystem implements interface.
var _ FileSystem = &nopFileSystem{}
// FileSystem represents an interface for file system for serving the Lattice UI.
type FileSystem interface {
New() (http.FileSystem, error)
}
func init() {
NopFileSystem = &nopFileSystem{}
}
// NopFileSystem represents a FileSystem that returns an error if called.
var NopFileSystem FileSystem
type nopFileSystem struct{}
// New is a no-op implementation of FileSystem New method.
func (n *nopFileSystem) New() (http.FileSystem, error) {
return nil, fmt.Errorf("file system not implemented")
}

1
go.mod
View file

@ -27,6 +27,7 @@ require (
github.com/prometheus/client_golang v1.0.0
github.com/prometheus/client_model v0.1.0
github.com/prometheus/prom2json v1.3.0
github.com/rakyll/statik v0.1.7
github.com/remyoudompheng/bigfft v0.0.0-20190728182440-6a916e37a237 // indirect
github.com/satori/go.uuid v1.2.0
github.com/shirou/gopsutil v2.18.12+incompatible

2
go.sum
View file

@ -167,6 +167,8 @@ github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNG
github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
github.com/prometheus/prom2json v1.3.0 h1:BlqrtbT9lLH3ZsOVhXPsHzFrApCTKRifB7gjJuypu6Y=
github.com/prometheus/prom2json v1.3.0/go.mod h1:rMN7m0ApCowcoDlypBHlkNbp5eJQf/+1isKykIP5ZnM=
github.com/rakyll/statik v0.1.7 h1:OF3QCZUuyPxuGEP7B4ypUa7sB/iHtqOTDYZXGM8KOdQ=
github.com/rakyll/statik v0.1.7/go.mod h1:AlZONWzMtEnMs7W4e/1LURLiI49pIMmp6V9Unghqrcc=
github.com/remyoudompheng/bigfft v0.0.0-20190728182440-6a916e37a237 h1:HQagqIiBmr8YXawX/le3+O26N+vPPC1PtjaF3mwnook=
github.com/remyoudompheng/bigfft v0.0.0-20190728182440-6a916e37a237/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=

1
holder.go
View file

@ -756,6 +756,7 @@ func (h *Holder) limitedSchema() []*IndexInfo {
CreatedAt: index.CreatedAt(),
Options: index.Options(),
ShardWidth: ShardWidth,
Fields: make([]*FieldInfo, 0, len(index.Fields())),
}
for _, field := range index.Fields() {
if strings.HasPrefix(field.name, "_") {

102
http/handler.go
View file

@ -56,6 +56,8 @@ import (
type Handler struct {
Handler http.Handler
fileSystem pilosa.FileSystem
logger logger.Logger
// Keeps the query argument validators for each handler
@ -68,6 +70,8 @@ type Handler struct {
closeTimeout time.Duration
server *http.Server
middleware []func(http.Handler) http.Handler
}
// externalPrefixFlag denotes endpoints that are intended to be exposed to clients.
@ -92,10 +96,10 @@ type handlerOption func(s *Handler) error
func OptHandlerAllowedOrigins(origins []string) handlerOption {
return func(h *Handler) error {
h.Handler = handlers.CORS(
h.middleware = append(h.middleware, handlers.CORS(
handlers.AllowedOrigins(origins),
handlers.AllowedHeaders([]string{"Content-Type"}),
)(h.Handler)
))
return nil
}
}
@ -107,6 +111,13 @@ func OptHandlerAPI(api *pilosa.API) handlerOption {
}
}
func OptHandlerFileSystem(fs pilosa.FileSystem) handlerOption {
return func(h *Handler) error {
h.fileSystem = fs
return nil
}
}
func OptHandlerLogger(logger logger.Logger) handlerOption {
return func(h *Handler) error {
h.logger = logger
@ -143,11 +154,10 @@ func NewHandler(opts ...handlerOption) (*Handler, error) {
}
})
handler := &Handler{
fileSystem: pilosa.NopFileSystem,
logger: logger.NopLogger,
closeTimeout: time.Second * 30,
}
handler.Handler = newRouter(handler)
handler.populateValidators()
for _, opt := range opts {
err := opt(handler)
@ -156,6 +166,10 @@ func NewHandler(opts ...handlerOption) (*Handler, error) {
}
}
// if OptHandlerFileSystem is used, it must be before newRouter is called
handler.Handler = newRouter(handler)
handler.populateValidators()
if handler.api == nil {
return nil, errors.New("must pass OptHandlerAPI")
}
@ -192,7 +206,6 @@ func (h *Handler) Close() error {
func (h *Handler) populateValidators() {
h.validators = map[string]*queryValidationSpec{}
h.validators["Home"] = queryValidationSpecRequired()
h.validators["PostClusterResizeAbort"] = queryValidationSpecRequired()
h.validators["PostClusterResizeRemoveNode"] = queryValidationSpecRequired()
h.validators["PostClusterResizeSetCoordinator"] = queryValidationSpecRequired()
@ -336,9 +349,8 @@ func (h *Handler) collectStats(next http.Handler) http.Handler {
}
// newRouter creates a new mux http router.
func newRouter(handler *Handler) *mux.Router {
func newRouter(handler *Handler) http.Handler {
router := mux.NewRouter()
router.HandleFunc("/", handler.handleHome).Methods("GET").Name("Home")
router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST").Name("PostClusterResizeAbort")
router.HandleFunc("/cluster/resize/remove-node", handler.handlePostClusterResizeRemoveNode).Methods("POST").Name("PostClusterResizeRemoveNode")
router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST").Name("PostClusterResizeSetCoordinator")
@ -400,11 +412,33 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/translate/index/{index}/{partition}", handler.handlePostTranslateIndexDB).Methods("POST").Name("PostTranslateIndexDB")
router.HandleFunc("/internal/translate/field/{index}/{field}", handler.handlePostTranslateFieldDB).Methods("POST").Name("PostTranslateFieldDB")
// Endpoints to support lattice UI embedded via statik.
// The messiness here reflects the fact that assets live in a nontrivial
// directory structure that is controlled externally.
latticeHandler := NewStatikHandler(handler)
router.PathPrefix("/static").Handler(latticeHandler)
router.Path("/").Handler(latticeHandler)
router.Path("/vds").Handler(latticeHandler)
router.Path("/favicon.png").Handler(latticeHandler)
router.Path("/favicon.svg").Handler(latticeHandler)
router.Path("/manifest.json").Handler(latticeHandler)
router.Use(handler.queryArgValidator)
router.Use(handler.addQueryContext)
router.Use(handler.extractTracing)
router.Use(handler.collectStats)
return router
var h http.Handler = router
for _, middleware := range handler.middleware {
// Ideally, we would use `router.Use` to inject middleware,
// instead of wrapping the handler. The reason we can't is
// because the router will only apply middleware to matched
// handlers. In this case, it won't match handlers with the
// OPTIONS method, needed by the CORS middleware. This issue
// is described in detail here:
// https://github.com/gorilla/handlers/issues/142
h = middleware(h)
}
return h
}
// ServeHTTP handles an HTTP request.
@ -422,6 +456,54 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Handler.ServeHTTP(w, r)
}
// statikHandler implements the http.Handler interface, and responds to
// requests for static assets with the appropriate file contents embedded
// in a statik filesystem.
type statikHandler struct {
handler *Handler
statikFS http.FileSystem
}
// NewStatikHandler returns a new instance of statikHandler
func NewStatikHandler(h *Handler) statikHandler {
fs, err := h.fileSystem.New()
if err == nil {
h.logger.Printf("enabled Web UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI)
}
return statikHandler{
handler: h,
statikFS: fs,
}
}
func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.UserAgent(), "curl") {
msg := "Welcome. Pilosa v" + s.handler.api.Version() + " is running. Visit https://www.pilosa.com/docs/ for more information."
if s.statikFS != nil {
msg += " Try the Web UI by visiting this URL in your browser."
}
http.Error(w, msg, http.StatusNotFound)
return
}
if s.statikFS == nil {
msg := "Web UI is not available. Please run `make generate-statik` before building Pilosa with `make install`."
s.handler.logger.Printf(msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
// /vds is a front-end route, not a backend route. Without this check, refreshing at /vds
// will request a nonexistent resource and return 404.
if r.URL.String() == "/vds" {
url, _ := url.Parse("/")
r.URL = url
}
http.FileServer(s.statikFS).ServeHTTP(w, r)
}
// successResponse is a general success/error struct for http responses.
type successResponse struct {
h *Handler
@ -490,10 +572,6 @@ func (r *successResponse) write(w http.ResponseWriter, err error) {
}
}
func (h *Handler) handleHome(w http.ResponseWriter, _ *http.Request) {
http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information.", http.StatusNotFound)
}
// validHeaderAcceptJSON returns false if one or more Accept
// headers are present, but none of them are "application/json"
// (or any matching wildcard). Otherwise returns true.

2
metrics.go
View file

@ -22,6 +22,8 @@ const (
MetricDeleteAvailableShard = "delete_available_shard_total"
MetricRecalculateCache = "recalculate_cache_total"
MetricInvalidateCache = "invalidate_cache_total"
MetricInvalidateCacheSkipped = "invalidate_cache_skipped_total"
MetricReadDirtyCache = "dirty_cache_total"
MetricRankCacheLength = "rank_cache_length"
MetricCacheThresholdReached = "cache_threshold_reached_total"
MetricRow = "query_row_total"

42
pg/pgtest/handler.go
View file

@ -16,6 +16,7 @@ package pgtest
import (
"context"
"errors"
"github.com/pilosa/pilosa/v2/pg"
)
@ -29,3 +30,44 @@ func (h HandlerFunc) HandleQuery(ctx context.Context, w pg.QueryResultWriter, q
}
var _ pg.QueryHandler = HandlerFunc(nil)
// ResultSet is a QueryResultWriter that accumulates results in a slice.
type ResultSet struct {
Columns []pg.ColumnInfo
Data [][]string
ResultTag string
}
// WriteHeader writes headers to the result set.
func (rs *ResultSet) WriteHeader(cols ...pg.ColumnInfo) error {
if rs.Columns != nil {
return errors.New("double-write of headers")
}
colsCopy := make([]pg.ColumnInfo, len(cols))
copy(colsCopy, cols)
rs.Columns = colsCopy
return nil
}
// WriteRowText writes a row to the result set.
func (rs *ResultSet) WriteRowText(vals ...string) error {
if rs.Columns == nil {
return errors.New("wrote a row without headers")
}
row := make([]string, len(vals))
copy(row, vals)
rs.Data = append(rs.Data, row)
return nil
}
// Tag applies a tag to the result set.
func (rs *ResultSet) Tag(tag string) {
rs.ResultTag = tag
}
var _ pg.QueryResultWriter = (*ResultSet)(nil)

111
proto/interface.go
View file

@ -40,36 +40,6 @@ 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
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().
@ -93,24 +63,20 @@ type ToRowser interface {
// RowsToTable is a helper function which takes a ToRowser,
// along with the number of rows, and returns a TableResponse.
// Obviously passing the number of rows seems unnecessary,
// and we could remove that requirement, but for now we
// do it to allow for pre-allocation of the rows slice.
// The number of rows is treated as a hint.
func RowsToTable(tr ToRowser, n int) (*TableResponse, error) {
var headers []*ColumnInfo
rows := make([]*Row, n)
rows := make([]*Row, 0, n)
// This callback gets called for every "row" in r.
// Each row populates its position in the pre-allocated
// `rows`. The headers get set based on those received
// in the first row.
var idx int
cb := func(rr *RowResponse) error {
if idx == 0 {
if len(rows) == 0 {
headers = rr.GetHeaders()
}
rows[idx] = &Row{Columns: rr.GetColumns()}
idx++
rows = append(rows, &Row{Columns: rr.GetColumns()})
return nil
}
@ -124,60 +90,6 @@ 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{
Code: 0,
Message: "EOF",
},
}
// Error is a helper function to create a RowResponse
// based on an error message. If the error is a grpc
// Status, then the status code is passed through.
@ -398,3 +310,18 @@ func (r RowResponseSorter) Less(i, j int) bool {
}
return false
}
// ConstRowser implements ToRowser with a slice of row responses.
type ConstRowser []RowResponse
// ToRows calls a function with a pointer to each element of the slice.
func (c ConstRowser) ToRows(fn func(*RowResponse) error) error {
for i := range c {
err := fn(&c[i])
if err != nil {
return err
}
}
return nil
}

10
rbf/cursor.go
View file

@ -429,10 +429,10 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
// Initialize a new root if we are currently the root page.
if c.stack.index == 0 {
assert(newRoot, "leaf write must be root when stack at root")
assert(newRoot) // leaf write must be root when stack at root
return c.writeRoot(origPgno, parents)
}
assert(!newRoot, "leaf write must NOT be root when stack not at root")
assert(!newRoot) // leaf write must NOT be root when stack not at root
// Otherwise update existing parent.
return c.putBranchCells(c.stack.index-1, parents)
@ -553,10 +553,10 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
// Initialize a new root if we are currently the root page.
if stackIndex == 0 {
assert(newRoot, "branch write must be root when stack at root")
assert(newRoot) // branch write must be root when stack at root
return c.writeRoot(origPgno, parents)
}
assert(!newRoot, "branch write must NOT be root when stack at root")
assert(!newRoot) // branch write must NOT be root when stack at root
// Otherwise update existing parent.
return c.putBranchCells(stackIndex-1, parents)
@ -817,7 +817,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
c.buffered = true
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
assert(elem.pgno != 0, "cursor should never point to page zero (meta)")
assert(elem.pgno != 0) // cursor should never point to page zero (meta)
buf, err := c.tx.readPage(elem.pgno)
if err != nil {

1
rbf/db.go
View file

@ -377,7 +377,6 @@ func (db *DB) WALPageN() int64 {
// SyncWAL flushes the active segment to disk.
func (db *DB) SyncWAL() error {
if s := db.ActiveWALSegment(); s != nil {
return s.Sync()
}

14
rbf/rbf.go
View file

@ -434,12 +434,12 @@ func (c *leafCell) countRange(start, end int32) (n int) {
func readLeafCellKey(page []byte, i int) uint64 {
offset := readCellOffset(page, i)
assert(offset < len(page), "cell read beyond page size: offset %d >= page size %d", offset, len(page))
assert(offset < len(page)) // cell read beyond page size
return *(*uint64)(unsafe.Pointer(&page[offset]))
}
func readLeafCell(page []byte, i int) leafCell {
assert(i < readCellN(page), "cell index %d exceeds cell count %d", i, readCellN(page))
assert(i < readCellN(page)) // cell index exceeds cell count
offset := readCellOffset(page, i)
buf := page[offset:]
@ -487,7 +487,7 @@ func writeLeafCell(page []byte, i, offset int, cell leafCell) {
*(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Type)
*(*uint16)(unsafe.Pointer(&page[offset+12])) = uint16(cell.N)
*(*uint16)(unsafe.Pointer(&page[offset+14])) = uint16(cell.BitN)
assert(offset+16+len(cell.Data) <= PageSize, "leaf cell write extends beyond page: offset %d + len(cell.Data)(%v) + 16 == %v > page size %d", offset, len(cell.Data), offset+16+len(cell.Data), PageSize)
assert(offset+16+len(cell.Data) <= PageSize) // leaf cell write extends beyond page
copy(page[offset+16:], cell.Data)
}
@ -513,8 +513,8 @@ func readBranchCellKey(page []byte, i int) uint64 {
}
func readBranchCell(page []byte, i int) branchCell {
assert(i >= 0, "branch cell index must be zero or greater: index=%d", i)
assert(i < readCellN(page), "branch cell index %d must less than cell count %d", i, readCellN(page))
assert(i >= 0) // branch cell index must be zero or greater
assert(i < readCellN(page)) // branch cell index must less than cell count
offset := readCellOffset(page, i)
var cell branchCell
@ -625,9 +625,9 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
}
}
func assert(condition bool, format string, args ...interface{}) {
func assert(condition bool) {
if !condition {
panic(fmt.Sprintf("assertion failed: "+format, args...))
panic("assertion failed")
}
}

24
rbf/tx.go
View file

@ -604,7 +604,10 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) {
func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) {
tx.mu.RLock()
defer tx.mu.RUnlock()
return tx.container(name, key)
}
func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
if tx.db == nil {
return nil, ErrTxClosed
} else if name == "" {
@ -626,9 +629,12 @@ func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) {
func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error {
tx.mu.Lock()
defer tx.mu.Unlock()
return tx.putContainer(name, key, ct)
}
func (tx *Tx) putContainer(name string, key uint64, ct *roaring.Container) error {
if tx.DeleteEmptyContainer && ct.N() == 0 {
return tx.RemoveContainer(name, key)
return tx.removeContainer(name, key)
}
cell := ConvertToLeafArgs(key, ct)
@ -650,7 +656,10 @@ func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error
func (tx *Tx) RemoveContainer(name string, key uint64) error {
tx.mu.Lock()
defer tx.mu.Unlock()
return tx.removeContainer(name, key)
}
func (tx *Tx) removeContainer(name string, key uint64) error {
c, err := tx.cursor(name)
if err != nil {
return err
@ -1414,6 +1423,9 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
return
}
tx.mu.Lock()
defer tx.mu.Unlock()
if err = tx.createBitmapIfNotExists(name); err != nil {
return
}
@ -1438,7 +1450,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
}
// INVAR: nsynth > 0
oldC, err = tx.Container(name, itrKey)
oldC, err = tx.container(name, itrKey)
panicOn(err)
if err != nil {
return
@ -1454,7 +1466,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
changed += nsynth
rowSet[currRow] += nsynth
err = tx.PutContainer(name, itrKey, synthC)
err = tx.putContainer(name, itrKey, synthC)
if err != nil {
return
}
@ -1476,7 +1488,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
changes := int(existN - newC.N())
changed += changes
rowSet[currRow] -= changes
err = tx.PutContainer(name, itrKey, newC)
err = tx.putContainer(name, itrKey, newC)
if err != nil {
return
}
@ -1494,7 +1506,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
// can nsynth be zero? No, because of the continue/invariant above where nsynth > 0
changed += nsynth
rowSet[currRow] += nsynth
err = tx.PutContainer(name, itrKey, synthC)
err = tx.putContainer(name, itrKey, synthC)
if err != nil {
return
}
@ -1511,7 +1523,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
changed += changes
rowSet[currRow] += changes
err = tx.PutContainer(name, itrKey, newC)
err = tx.putContainer(name, itrKey, newC)
if err != nil {
panicOn(err)
return

8
rbf/wal.go
View file

@ -154,6 +154,7 @@ func (s *WALSegment) closeForWrite() error {
if err := s.sync(); err != nil {
return err
}
s.writeCache = nil
// Close underlying file writer.
if s.w != nil {
@ -190,7 +191,7 @@ func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
// WriteWALPage writes a single page to the WAL segment and returns its WAL identifier.
func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) {
assert(len(page) == PageSize, "invalid page size: %d", len(page))
assert(len(page) == PageSize) // invalid page size
s.mu.Lock()
defer s.mu.Unlock()
@ -212,6 +213,9 @@ func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err er
}
// Append write to write buffer & increment page count.
if s.writeCache == nil {
s.writeCache = make([]byte, 0, MaxWALSegmentFileSize+PageSize)
}
s.writeCache = append(s.writeCache, page...)
s.pageN++
@ -229,7 +233,7 @@ func (s *WALSegment) flush() error {
if _, err := s.w.WriteAt(s.writeCache, int64((s.pageN*PageSize)-len(s.writeCache))); err != nil {
return fmt.Errorf("wal segment write: %w", err)
}
s.writeCache = nil
s.writeCache = s.writeCache[:0]
return nil
}

91
server/grpc.go
View file

@ -18,7 +18,6 @@ import (
"context"
"crypto/tls"
"fmt"
"io"
"net"
"strings"
"sync"
@ -66,14 +65,62 @@ func errToStatusError(err error) error {
// Check error string.
switch errors.Cause(err) {
case pilosa.ErrIndexNotFound, pilosa.ErrFieldNotFound:
return status.Error(codes.NotFound, err.Error())
}
// Check error type.
switch errors.Cause(err).(type) {
case pilosa.NotFoundError:
case pilosa.ErrIndexNotFound,
pilosa.ErrFieldNotFound,
pilosa.ErrForeignIndexNotFound,
pilosa.ErrBSIGroupNotFound:
return status.Error(codes.NotFound, err.Error())
case pilosa.ErrIndexExists,
pilosa.ErrFieldExists,
pilosa.ErrBSIGroupExists:
return status.Error(codes.AlreadyExists, err.Error())
case pilosa.ErrIndexRequired,
pilosa.ErrFieldRequired,
pilosa.ErrColumnRequired,
pilosa.ErrBSIGroupNameRequired,
pilosa.ErrName,
pilosa.ErrQueryRequired,
pilosa.ErrFieldsArgumentRequired,
pilosa.ErrIntFieldWithKeys,
pilosa.ErrDecimalFieldWithKeys:
return status.Error(codes.FailedPrecondition, err.Error())
case pilosa.ErrInvalidView,
pilosa.ErrInvalidBSIGroupType,
pilosa.ErrInvalidBSIGroupValueType,
pilosa.ErrInvalidCacheType:
return status.Error(codes.InvalidArgument, err.Error())
case pilosa.ErrDecimalOutOfRange,
pilosa.ErrBSIGroupValueTooLow,
pilosa.ErrBSIGroupValueTooHigh,
pilosa.ErrInvalidRangeOperation,
pilosa.ErrInvalidBetweenValue:
return status.Error(codes.OutOfRange, err.Error())
case pilosa.ErrQueryTimeout:
return status.Error(codes.DeadlineExceeded, err.Error())
case pilosa.ErrQueryCancelled:
return status.Error(codes.Canceled, err.Error())
case pilosa.ErrNotImplemented:
return status.Error(codes.Unimplemented, err.Error())
case pilosa.ErrAborted:
return status.Error(codes.Aborted, err.Error())
case pilosa.ErrClusterDoesNotOwnShard,
pilosa.ErrResizeNoReplicas,
pilosa.ErrResizeNotRunning,
pilosa.ErrNodeNotCoordinator,
pilosa.ErrTooManyWrites,
pilosa.ErrNodeIDNotExists:
return status.Error(codes.Internal, err.Error())
}
return status.Error(codes.Unknown, err.Error())
}
@ -105,7 +152,7 @@ func (h *GRPCHandler) PostVDS(ctx context.Context, req *pb.PostVDSRequest) (*pb.
opts := pilosa.IndexOptions{Keys: req.Keys, TrackExistence: req.TrackExistence}
_, err := h.api.CreateIndex(ctx, req.Name, opts)
if err != nil {
return nil, err
return nil, errToStatusError(err)
}
return &pb.PostVDSResponse{}, nil
}
@ -114,12 +161,12 @@ func (h *GRPCHandler) PostVDS(ctx context.Context, req *pb.PostVDSRequest) (*pb.
func (h *GRPCHandler) DeleteVDS(ctx context.Context, req *pb.DeleteVDSRequest) (*pb.DeleteVDSResponse, error) {
err := h.api.DeleteIndex(ctx, req.Name)
if err != nil {
return nil, err
return nil, errToStatusError(err)
}
return &pb.DeleteVDSResponse{}, nil
}
func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.StreamClient, error) {
func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.ToRowser, error) {
return execSQL(ctx, h.api, h.logger, queryStr)
}
@ -130,21 +177,12 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ
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")
}
err = results.ToRows(stream.Send)
if err != nil {
return errors.Wrap(err, "streaming result")
}
return nil
}
// QuerySQLUnary is a unary-response (non-streaming) version of QuerySQL, returning a TableResponse.
@ -164,7 +202,10 @@ func (h *GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest
if err != nil {
return nil, err
}
return pb.ReadIntoTable(results)
if results, ok := results.(pb.ToTabler); ok {
return results.ToTable()
}
return pb.RowsToTable(results, 0)
}
// QueryPQL handles the PQL request and sends RowResponses to the stream.

40
server/grpc_test.go
View file

@ -755,6 +755,46 @@ func TestQuerySQLUnary(t *testing.T) {
},
eq: equal,
},
// The following cases test different paths within the `case *sqlparser.AndExpr`
// of extract.go by providing different WHERE conditions.
{
// len(left) == 2 && len(right) == 1
// right[0].table == left[0].table
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and j.jointype = 2 and g.age = 16",
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
eq: equalUnordered,
},
{
// len(left) == 2 && len(right) == 1
// right[0].table == left[1].table {
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where j.jointype = 2 and g.color = 'red' and g.age = 16",
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
eq: equalUnordered,
},
{
// len(left) == 1 && len(right) == 1 && left[0].table != right[0].table
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and g.age = 16 and j.jointype = 2",
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
eq: equalUnordered,
},
}
for i, test := range tests {

101
server/pg.go
View file

@ -20,7 +20,6 @@ import (
"crypto/tls"
"encoding/json"
"fmt"
"io"
"net"
"strconv"
"strings"
@ -29,6 +28,7 @@ import (
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pg"
"github.com/pilosa/pilosa/v2/pql"
pb "github.com/pilosa/pilosa/v2/proto"
"github.com/pkg/errors"
@ -50,12 +50,7 @@ func NewPostgresServer(api *pilosa.API, logger logger.Logger, tls *tls.Config) *
api: api,
logger: logger,
s: pg.Server{
QueryHandler: &queryDecodeHandler{
child: &pilosaQueryHandler{
api: api,
logger: logger,
},
},
QueryHandler: NewPostgresHandler(api, logger),
TypeEngine: pg.PrimitiveTypeEngine{},
StartupTimeout: 5 * time.Second,
ReadTimeout: 10 * time.Second,
@ -70,6 +65,16 @@ func NewPostgresServer(api *pilosa.API, logger logger.Logger, tls *tls.Config) *
}
}
// NewPostgresHandler creates a postgres query handler wrapping the pilosa API.
func NewPostgresHandler(api *pilosa.API, logger logger.Logger) pg.QueryHandler {
return &queryDecodeHandler{
child: &pilosaQueryHandler{
api: api,
logger: logger,
},
}
}
// Start a postgres endpoint at the specified address.
func (s *PostgresServer) Start(addr string) error {
l, err := net.Listen("tcp", addr)
@ -201,6 +206,8 @@ func pgFormatVal(val interface{}) string {
return strconv.FormatUint(val, 10)
case string:
return val
case pql.Decimal:
return val.String()
default:
data, _ := json.Marshal(val)
return string(data)
@ -320,10 +327,15 @@ func pgWriteRowser(w pg.QueryResultWriter, result pb.ToRowser) error {
for i, col := range row.Columns {
var v string
switch col := col.ColumnVal.(type) {
case nil:
v = "null"
case *pb.ColumnResponse_BoolVal:
v = strconv.FormatBool(col.BoolVal)
case *pb.ColumnResponse_DecimalVal:
v = col.DecimalVal.String()
v = pql.Decimal{
Value: col.DecimalVal.Value,
Scale: col.DecimalVal.Scale,
}.String()
case *pb.ColumnResponse_Float64Val:
v = strconv.FormatFloat(col.Float64Val, 'g', -1, 64)
case *pb.ColumnResponse_Int64Val:
@ -349,28 +361,6 @@ 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:
@ -383,8 +373,55 @@ 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})
case uint64:
err := w.WriteHeader(pg.ColumnInfo{
Name: "count",
Type: pg.TypeCharoid,
})
if err != nil {
return errors.Wrap(err, "writing headers")
}
err = w.WriteRowText(strconv.FormatUint(result, 10))
if err != nil {
return errors.Wrap(err, "writing count")
}
return nil
case int64:
err := w.WriteHeader(pg.ColumnInfo{
Name: "value",
Type: pg.TypeCharoid,
})
if err != nil {
return errors.Wrap(err, "writing headers")
}
err = w.WriteRowText(strconv.FormatInt(result, 10))
if err != nil {
return errors.Wrap(err, "writing count")
}
return nil
case bool:
err := w.WriteHeader(pg.ColumnInfo{
Name: "result",
Type: pg.TypeCharoid,
})
if err != nil {
return errors.Wrap(err, "writing headers")
}
err = w.WriteRowText(strconv.FormatBool(result))
if err != nil {
return errors.Wrap(err, "writing count")
}
return nil
case nil:
return nil
default:
return errors.Errorf("result type %T not yet supported", result)
}

327
server/pg_test.go Normal file
View file

@ -0,0 +1,327 @@
// 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_test
import (
"context"
"math"
"reflect"
"testing"
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pg"
"github.com/pilosa/pilosa/v2/pg/pgtest"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
)
func TestPostgresHandler(t *testing.T) {
m := test.RunCommand(t)
defer m.Close()
pgh := server.NewPostgresHandler(m.API, logger.NewLogfLogger(t))
m.MustCreateIndex(t, "i", pilosa.IndexOptions{TrackExistence: true})
m.MustCreateField(t, "i", "set")
m.MustCreateField(t, "i", "keyset", pilosa.OptFieldKeys())
m.MustCreateField(t, "i", "mutex", pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0))
m.MustCreateField(t, "i", "keymutex", pilosa.OptFieldKeys(), pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0))
m.MustCreateField(t, "i", "int", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
m.MustCreateField(t, "i", "decimal", pilosa.OptFieldTypeDecimal(2))
m.MustCreateField(t, "i", "time", pilosa.OptFieldTypeTime("YMDH"))
m.MustCreateField(t, "i", "bool", pilosa.OptFieldTypeBool())
m.MustCreateIndex(t, "j", pilosa.IndexOptions{TrackExistence: true, Keys: true})
m.MustCreateField(t, "j", "set")
storeOK := pgtest.ResultSet{
Columns: []pg.ColumnInfo{
{
Name: "result",
Type: pg.TypeCharoid,
},
},
Data: [][]string{
{
"true",
},
},
}
cases := []struct {
Name string
Queries []string
Results []pgtest.ResultSet
}{
{
Name: "Extract-Nothing",
Queries: []string{
`[i]Extract(All(), Rows(set))`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{
Name: "_id",
Type: pg.TypeCharoid,
},
{
Name: "set",
Type: pg.TypeCharoid,
},
},
},
},
},
{
Name: "Store",
Queries: []string{
`[i]Store(ConstRow(columns=[1, 2, 3]), set=4)`,
`[i]Store(ConstRow(columns=[0, 2, 4]), set=5)`,
`[j]Store(ConstRow(columns=[1, 2, 3]), set=4)`,
`[j]Store(ConstRow(columns=[0, 2, 4]), set=5)`,
},
Results: []pgtest.ResultSet{
storeOK,
storeOK,
storeOK,
storeOK,
},
},
{
Name: "Set",
Queries: []string{
`[i]Set(1, keyset="a")`,
`[i]Set(2, keyset="b")`,
`[i]Set(3, mutex=3)`,
`[i]Set(4, keymutex="d")`,
`[i]Set(1, int=5)`,
`[i]Set(2, decimal=6.01)`,
`[i]Set(3, time=7, 2016-01-01T00:00)`,
`[i]Set(4, bool=false)`,
},
Results: []pgtest.ResultSet{
storeOK,
storeOK,
storeOK,
storeOK,
storeOK,
storeOK,
storeOK,
storeOK,
},
},
{
Name: "Extract",
Queries: []string{
`[i]Extract(
All(),
Rows(set), Rows(keyset),
Rows(mutex), Rows(keymutex),
Rows(int), Rows(decimal),
Rows(time),
Rows(bool)
)`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "_id", Type: pg.TypeCharoid},
{Name: "set", Type: pg.TypeCharoid},
{Name: "keyset", Type: pg.TypeCharoid},
{Name: "mutex", Type: pg.TypeCharoid},
{Name: "keymutex", Type: pg.TypeCharoid},
{Name: "int", Type: pg.TypeCharoid},
{Name: "decimal", Type: pg.TypeCharoid},
{Name: "time", Type: pg.TypeCharoid},
{Name: "bool", Type: pg.TypeCharoid},
},
Data: [][]string{
{`1`, `[4]`, `["a"]`, `null`, `null`, `5`, `null`, `[]`, `null`},
{`2`, `[4,5]`, `["b"]`, `null`, `null`, `null`, `6.01`, `[]`, `null`},
{`3`, `[4]`, `[]`, `3`, `null`, `null`, `null`, `[7]`, `null`},
{`4`, `[5]`, `[]`, `null`, `d`, `null`, `null`, `[]`, `false`},
},
},
},
},
{
Name: "GroupBy",
Queries: []string{
`[i]GroupBy(Rows(set))`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "set", Type: pg.TypeCharoid},
{Name: "count", Type: pg.TypeCharoid},
{Name: "sum", Type: pg.TypeCharoid},
},
Data: [][]string{
{"4", "3", "0"},
{"5", "3", "0"},
},
},
},
},
{
Name: "Count",
Queries: []string{
`[i]Count(Row(set=4))`,
`[i]Count(Row(int > 0))`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{{"3"}},
},
{
Columns: []pg.ColumnInfo{
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{{"1"}},
},
},
},
{
Name: "FieldValue",
Queries: []string{
`[i]FieldValue(field=int, column=1)`,
`[i]FieldValue(field=decimal, column=2)`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "value", Type: pg.TypeCharoid},
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{
{"5", "1"},
},
},
{
Columns: []pg.ColumnInfo{
{Name: "value", Type: pg.TypeCharoid},
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{
{"6.01", "1"},
},
},
},
},
{
Name: "Rows",
Queries: []string{
`[i]Rows(set)`,
`[i]Rows(keyset)`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "set", Type: pg.TypeCharoid},
},
Data: [][]string{
{"4"},
{"5"},
},
},
{
Columns: []pg.ColumnInfo{
{Name: "keyset", Type: pg.TypeCharoid},
},
Data: [][]string{
{"a"},
{"b"},
},
},
},
},
{
Name: "TopN",
Queries: []string{
`[i]TopN(set)`,
`[i]TopN(keyset)`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "set", Type: pg.TypeCharoid},
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{
{"5", "3"},
{"4", "3"},
},
},
{
Columns: []pg.ColumnInfo{
{Name: "keyset", Type: pg.TypeCharoid},
{Name: "count", Type: pg.TypeCharoid},
},
Data: [][]string{
{"b", "1"},
{"a", "1"},
},
},
},
},
{
Name: "SQL",
Queries: []string{
`select _id from i;`,
},
Results: []pgtest.ResultSet{
{
Columns: []pg.ColumnInfo{
{Name: "_id", Type: pg.TypeCharoid},
},
Data: [][]string{
{`1`},
{`2`},
{`3`},
{`4`},
},
},
},
},
}
for _, c := range cases {
c := c
t.Run(c.Name, func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
for i, q := range c.Queries {
var res pgtest.ResultSet
err := pgh.HandleQuery(ctx, &res, pg.SimpleQuery(q))
if err != nil {
t.Errorf("query %q failed: %v", q, err)
continue
}
expected := c.Results[i]
if !reflect.DeepEqual(res, expected) {
t.Errorf("query %q returned incorrect results: expected %v but got %v", q, expected, res)
}
}
})
}
}

2
server/server.go
View file

@ -46,6 +46,7 @@ import (
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/prometheus"
"github.com/pilosa/pilosa/v2/statik"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/statsd"
"github.com/pilosa/pilosa/v2/syswrap"
@ -427,6 +428,7 @@ func (m *Command) SetupServer() error {
http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins),
http.OptHandlerAPI(m.API),
http.OptHandlerLogger(m.logger),
http.OptHandlerFileSystem(&statik.FileSystem{}),
http.OptHandlerListener(m.ln),
http.OptHandlerCloseTimeout(m.closeTimeout),
)

4
server/sql.go
View file

@ -26,14 +26,14 @@ import (
"google.golang.org/grpc/status"
)
func execSQL(ctx context.Context, api *pilosa.API, logger logger.Logger, queryStr string) (pb.StreamClient, error) {
func execSQL(ctx context.Context, api *pilosa.API, logger logger.Logger, queryStr string) (pb.ToRowser, 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
var results pb.ToRowser
switch query.SQLType {
case sql.SQLTypeSelect:
handler := sql.NewSelectHandler(api)

6
sql/ddl.go
View file

@ -37,7 +37,7 @@ func NewDDLHandler(api *pilosa.API) *DDLHandler {
}
// Handle executes mapped SQL
func (h *DDLHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) {
func (h *DDLHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToRowser, error) {
stmt, ok := mapped.Statement.(*sqlparser.DDL)
if !ok {
return nil, fmt.Errorf("statement is not type DDL: %T", mapped.Statement)
@ -52,7 +52,7 @@ func (h *DDLHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.Stre
}
}
func (h *DDLHandler) execDropTable(ctx context.Context, stmt *sqlparser.DDL) (pproto.StreamClient, error) {
func (h *DDLHandler) execDropTable(ctx context.Context, stmt *sqlparser.DDL) (pproto.ToRowser, error) {
if n := len(stmt.FromTables); n != 1 {
return nil, fmt.Errorf("statement can only contain a single drop table, but got: %d", n)
}
@ -61,5 +61,5 @@ func (h *DDLHandler) execDropTable(ctx context.Context, stmt *sqlparser.DDL) (pp
if err := h.api.DeleteIndex(ctx, indexName); err != nil {
return nil, errors.Wrapf(err, "deleting index %s", indexName)
}
return pproto.EmptyStream{}, nil
return pproto.ConstRowser{}, nil
}

4
sql/extract.go
View file

@ -1042,12 +1042,12 @@ func extractWheres(indexes []*pilosa.Index, tbls parseTables, expr sqlparser.Exp
return []*tableWhere{left[0], right[0]}, nil
}
return nil, errors.Errorf("no matching table on right: %s", left[0].table.name)
} else if len(left) == 1 && len(right) == 2 {
} else if len(left) == 2 && len(right) == 1 {
// if left(2) and right(1),
// then intersect the 1's and return final(2)
if right[0].table == left[0].table {
right[0].where = Intersect(right[0].where, left[0].where)
return []*tableWhere{left[0], right[1]}, nil
return []*tableWhere{left[1], right[0]}, nil
} else if right[0].table == left[1].table {
right[0].where = Intersect(right[0].where, left[1].where)
return []*tableWhere{left[0], right[0]}, nil

452
sql/reduce.go
View file

@ -15,110 +15,105 @@
package sql
import (
"io"
"sort"
"github.com/pilosa/pilosa/v2"
"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
type limitRowser struct {
rowser pproto.ToRowser
limit uint
}
// LimitReducer limits the number of messages passed through.
type LimitReducer struct {
limit uint
func (l *limitRowser) ToRows(fn func(*pproto.RowResponse) error) error {
limit := l.limit
return l.rowser.ToRows(func(row *pproto.RowResponse) error {
if limit == 0 {
return nil
}
limit--
return fn(row)
})
}
// LimitRows applies a limit to a ToRowser.
func LimitRows(rowser pproto.ToRowser, limit uint) pproto.ToRowser {
switch rowser := rowser.(type) {
case pilosa.ExtractedTable:
if uint(len(rowser.Columns)) > limit {
rowser.Columns = rowser.Columns[:limit]
}
return rowser
default:
return &limitRowser{rowser, limit}
}
}
type offsetRowser struct {
rowser pproto.ToRowser
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.
func (o *offsetRowser) ToRows(fn func(*pproto.RowResponse) error) error {
offset := o.offset
var headers []*pproto.ColumnInfo
return o.rowser.ToRows(func(row *pproto.RowResponse) error {
if headers == nil {
headers = row.Headers
}
if offset > 0 {
offset--
return nil
}
row.Headers = headers
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)
return fn(row)
})
}
// 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 {
// OffsetRows applies an offset to a ToRowser.
func OffsetRows(rowser pproto.ToRowser, offset uint) pproto.ToRowser {
if offset == 0 {
return rowser
}
switch rowser := rowser.(type) {
case pilosa.ExtractedTable:
if uint(len(rowser.Columns)) > offset {
rowser.Columns = rowser.Columns[:0]
} else {
rowser.Columns = rowser.Columns[offset:]
}
return rowser
default:
return &offsetRowser{rowser, offset}
}
}
type orderByRowser struct {
rowser pproto.ToRowser
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 {
func (o *orderByRowser) ToRows(fn func(*pproto.RowResponse) error) error {
// hold is a slice of row responses, to be sent to the output
// stream sorted by the sort conditions.
var hold []*pproto.RowResponse
err := o.rowser.ToRows(func(row *pproto.RowResponse) error {
hold = append(hold, row)
return nil
})
if err != nil {
return err
}
if len(hold) == 0 {
return nil
}
// sortColNames contains the names of the columns to
// sort on.
@ -138,49 +133,16 @@ func (o *OrderByReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) er
// 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
holdHeaders := hold[0].Headers
for i, hdr := range holdHeaders {
hdrName := hdr.GetName()
hdrType := hdr.GetDatatype()
for j := range sortColNames {
if sortColNames[j] == hdrName {
sortColIdxs[j] = i
sortColTypes[j] = hdrType
}
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.
@ -191,61 +153,59 @@ func (o *OrderByReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) er
hold,
)
if err != nil {
return s.Send(pproto.ErrorWrap(err, "creating row response sorter"))
return errors.Wrap(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
}
for i := range hold {
// Re-apply the headers to the first record.
if !headerApplied {
hold[i].Headers = holdHeaders
headerApplied = true
}
err := s.Send(hold[i])
err := fn(hold[i])
if err != nil {
return s.Send(pproto.ErrorWrap(err, "sending hold row"))
return errors.Wrap(err, "sending hold row")
}
}
return s.Send(pproto.EOF)
return nil
}
// 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,
// OrderBy sorts a rowser.
func OrderBy(rowser pproto.ToRowser, fields, dirs []string) pproto.ToRowser {
descendings := make([]bool, len(fields))
for i := range dirs {
if dirs[i] == "desc" {
descendings[i] = true
}
}
return &orderByRowser{
rowser: rowser,
fields: fields,
isDescending: descendings,
}
}
// Reduce modifies the stream according to the function.
func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error {
r, err := c.Recv()
type valCountRowser struct {
rowser pproto.ToRowser
fn FuncName
}
func (v *valCountRowser) ToRows(fn func(row *pproto.RowResponse) error) error {
var r *pproto.RowResponse
err := v.rowser.ToRows(func(row *pproto.RowResponse) error {
if r != nil {
return errors.New("extra row in valcount")
}
r = row
return nil
})
if err != nil {
if err == io.EOF {
return s.Send(pproto.EOF)
}
return s.Send(pproto.Error(err))
return err
}
// Get the index of the column with header of "value".
@ -268,7 +228,7 @@ func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServe
returnDataType = sourceDataType
switch v.fn {
case FuncAvg:
returnDataType = DataTypeFloat64
returnDataType = "float64"
}
rr := pproto.RowResponse{
@ -280,29 +240,20 @@ func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServe
cols := r.GetColumns()
if len(cols) == 0 {
return s.Send(pproto.ErrorCode(
errors.New("empty column set"),
codes.Unknown,
))
return errors.New("empty column set")
}
if idxVal == -1 {
return s.Send(pproto.ErrorCode(
errors.New("result set has no column: value"),
codes.Unknown,
))
return errors.New("result set has no column: value")
}
if idxCnt == -1 {
return s.Send(pproto.ErrorCode(
errors.New("result set has no column: count"),
codes.Unknown,
))
return errors.New("result set has no column: count")
}
switch v.fn {
case FuncAvg:
var avg float64
if sourceDataType == DataTypeDecimal {
if sourceDataType == "decimal" {
val := cols[idxVal].GetDecimalVal()
dec := pql.NewDecimal(val.Value, val.Scale)
cnt := cols[idxCnt].GetInt64Val()
@ -314,7 +265,7 @@ func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServe
}
rr.Columns[0] = &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_Float64Val{Float64Val: avg}}
default:
if sourceDataType == DataTypeDecimal {
if sourceDataType == "decimal" {
val := cols[idxVal].GetDecimalVal()
rr.Columns[0] = &pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_DecimalVal{DecimalVal: &pproto.Decimal{Value: val.Value, Scale: val.Scale}}}
} else {
@ -323,127 +274,90 @@ func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServe
}
}
if err := s.Send(&rr); err != nil {
return errors.Wrap(err, "sending row response")
}
return s.Send(pproto.EOF)
return fn(&rr)
}
// CountIDReducer returns a stream of _id's as a count.
type CountIDReducer struct{}
// ApplyValCountFunc converts a ValCount result to the proper
// result for Func
func ApplyValCountFunc(rowser pproto.ToRowser, fn FuncName) pproto.ToRowser {
return &valCountRowser{
rowser: rowser,
fn: fn,
}
}
// 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
type countIDRowser struct {
rowser pproto.ToRowser
}
for {
_, err := c.Recv()
if err != nil {
if err == io.EOF {
break
}
return s.Send(pproto.ErrorWrap(err, "receiving on client stream"))
}
cnt++
func (c *countIDRowser) ToRows(fn func(*pproto.RowResponse) error) error {
var count uint64
err := c.rowser.ToRows(func(row *pproto.RowResponse) error {
count++
return nil
})
if err != nil {
return err
}
rr := pproto.RowResponse{
return fn(&pproto.RowResponse{
Headers: []*pproto.ColumnInfo{
{Name: string(FuncCount), Datatype: "uint64"},
},
Columns: []*pproto.ColumnResponse{
&pproto.ColumnResponse{ColumnVal: &pproto.ColumnResponse_Uint64Val{Uint64Val: cnt}},
{
ColumnVal: &pproto.ColumnResponse_Uint64Val{Uint64Val: count},
},
},
}
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
// CountRows counts the rows from the input rowser.
func CountRows(rowser pproto.ToRowser) pproto.ToRowser {
return &countIDRowser{rowser}
}
// NewAssignHeadersReducer returns a new instance of AssignHeadersReducer.
func NewAssignHeadersReducer(cols []Column) *AssignHeadersReducer {
return &AssignHeadersReducer{
cols: cols,
}
type assignHeadersRowser struct {
rowser pproto.ToRowser
cols []Column
}
// Reduce modifies the stream.
func (r *AssignHeadersReducer) Reduce(c pproto.StreamClient, s pproto.StreamServer) error {
func (a *assignHeadersRowser) ToRows(fn func(*pproto.RowResponse) error) 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)
return a.rowser.ToRows(func(row *pproto.RowResponse) error {
var out pproto.RowResponse
if placement == nil {
// Assign headers and generate placement.
var err error
var labels []string
placement, labels, err = headerAssignment(a.cols, row.Headers)
if err != nil {
return s.Send(pproto.ErrorWrap(err, "getting header assignment"))
return errors.Wrap(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")
headers := make([]*pproto.ColumnInfo, len(placement))
for i, v := range placement {
header := row.Headers[v]
header.Name = labels[i]
headers[i] = header
}
out.Headers = headers
}
cnt++
}
return s.Send(pproto.EOF)
// Re-order the columns.
cols := make([]*pproto.ColumnResponse, len(placement))
for i, v := range placement {
cols[i] = row.Columns[v]
}
out.Columns = cols
return fn(&out)
})
}
// AssignHeaders assigns headers to a ToRowser.
func AssignHeaders(rowser pproto.ToRowser, headers ...Column) pproto.ToRowser {
return &assignHeadersRowser{rowser, headers}
}
var (

195
sql/select.go
View file

@ -41,7 +41,7 @@ func NewSelectHandler(api *pilosa.API) *SelectHandler {
}
// Handle executes mapped SQL
func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) {
func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToRowser, error) {
stmt, ok := mapped.Statement.(*sqlparser.Select)
if !ok {
return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement)
@ -74,7 +74,7 @@ func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Sel
return mr, nil
}
func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.StreamClient, error) {
func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.ToRowser, error) {
if mr.Query == "" {
return nil, errors.New("no pql query created")
}
@ -85,29 +85,15 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult
}
res := resp.Results[0]
// TODO: synchronize this properly somehow.
// It would probbably help to get rid of the streaming too.
respRows := pproto.NewRowBuffer(0)
var result pproto.ToRowser
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
}
}()
result = res
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
}
}()
result = pilosa.GroupCounts(res)
case uint64:
go func() {
respRows.Send(&pproto.RowResponse{ //nolint:errcheck
result = pproto.ConstRowser{
{
Headers: []*pproto.ColumnInfo{
{
Name: "count",
@ -121,12 +107,11 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult
},
},
},
})
respRows.Send(pproto.EOF) //nolint:errcheck
}()
},
}
case bool:
go func() {
respRows.Send(&pproto.RowResponse{ //nolint:errcheck
result = pproto.ConstRowser{
{
Headers: []*pproto.ColumnInfo{
{
Name: "result",
@ -140,24 +125,18 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult
},
},
},
})
respRows.Send(pproto.EOF) //nolint:errcheck
}()
},
}
case nil:
result = pproto.ConstRowser{}
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
// Apply reducers.
for _, reducer := range mr.Reducers {
result = reducer(result)
}
return result, nil
@ -172,10 +151,10 @@ type MappingResult struct {
Offset uint64
Query string
Header []Column
Reducers []Reducer
Reducers []func(pproto.ToRowser) pproto.ToRowser
}
func (mr *MappingResult) addReducer(r Reducer) {
func (mr *MappingResult) addReducer(r func(pproto.ToRowser) pproto.ToRowser) {
mr.Reducers = append(mr.Reducers, r)
}
@ -278,21 +257,24 @@ func (h handlerSelectFieldsFromTableWhere) Apply(stmt *sqlparser.Select, qm Quer
Header: selectFields,
}
// TODO: assign headers
mr.addReducer(NewAssignHeadersReducer(selectFields))
// assign headers
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, 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:
if qm.HasOrderBy() {
// Sort the results.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OrderBy(result, orderByFlds, orderByDirs)
})
// Apply the limit and offset after sorting.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
} else {
// Apply the limit and offset inside the query.
whereQuery = Limit(whereQuery, limit, offset)
case offset != 0:
whereQuery = Offset(whereQuery, offset)
}
if len(fields) > 0 && fields[0] == "_id" {
@ -359,13 +341,23 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
Query: qo,
}
mr.addReducer(NewAssignHeadersReducer(selectFields))
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
if qm.HasOrderBy() {
mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset))
} else {
mr.addReducer(NewLimitReducer(limit, offset))
// Sort the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OrderBy(result, orderByFlds, orderByDirs)
})
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
return mr, nil
}
@ -374,7 +366,7 @@ 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
var reducers []func(pproto.ToRowser) pproto.ToRowser
indexName, err := extractIndexName(stmt)
if err != nil {
@ -412,7 +404,9 @@ func (h handlerSelectCountFromTableWhere) Apply(stmt *sqlparser.Select, qm Query
} else {
// TODO: add the Distinct (for Int fields) here (like we do in handlerSelectDistinctFromTable)
qo = Rows(funcs[0].field.Name())
reducers = append(reducers, &CountIDReducer{})
reducers = append(reducers, func(result pproto.ToRowser) pproto.ToRowser {
return CountRows(result)
})
}
mr := &MappingResult{
IndexName: indexName,
@ -421,9 +415,10 @@ func (h handlerSelectCountFromTableWhere) Apply(stmt *sqlparser.Select, qm Query
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.
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
return mr, nil
}
@ -493,14 +488,28 @@ func (h handlerSelectFuncFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryM
Query: qo,
}
mr.addReducer(NewValCountFuncReducer(funcs[0].funcName))
mr.addReducer(NewAssignHeadersReducer(selectFields))
// Apply the ValCount function.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return ApplyValCountFunc(result, funcs[0].funcName)
})
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
if qm.HasOrderBy() {
mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset))
} else {
mr.addReducer(NewLimitReducer(limit, offset))
// Sort the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OrderBy(result, orderByFlds, orderByDirs)
})
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
return mr, nil
}
@ -627,13 +636,23 @@ func (h handlerSelectGroupBy) Apply(stmt *sqlparser.Select, qm QueryMask, indexF
Query: qo,
}
mr.addReducer(NewAssignHeadersReducer(selectFields))
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
if qm.HasOrderBy() {
mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset))
} else {
mr.addReducer(NewLimitReducer(limit, offset))
// Sort the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OrderBy(result, orderByFlds, orderByDirs)
})
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
return mr, nil
}
@ -689,13 +708,21 @@ func (f handlerSelectIDCountFromTable) Apply(stmt *sqlparser.Select, qm QueryMas
Query: qo,
}
mr.addReducer(NewAssignHeadersReducer(selectFields))
mr.addReducer(NewLimitReducer(limit, offset))
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
// 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.
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
return mr, nil
}
@ -784,12 +811,22 @@ func (h handlerSelectJoin) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc
Query: qo,
}
mr.addReducer(NewAssignHeadersReducer(selectFields))
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
})
if qm.HasOrderBy() {
mr.addReducer(NewOrderByReducer(orderByFlds, orderByDirs, limit, offset))
} else {
mr.addReducer(NewLimitReducer(limit, offset))
// Sort the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OrderBy(result, orderByFlds, orderByDirs)
})
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
return mr, nil
}

64
sql/show.go
View file

@ -37,7 +37,7 @@ func NewShowHandler(api *pilosa.API) *ShowHandler {
}
// Handle executes mapped SQL
func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.StreamClient, error) {
func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToRowser, error) {
stmt, ok := mapped.Statement.(*sqlparser.Show)
if !ok {
return nil, fmt.Errorf("statement is not type show: %T", mapped.Statement)
@ -53,20 +53,12 @@ func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.Str
}
}
func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) {
func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.ToRowser, 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{
result := make(pproto.ConstRowser, len(indexInfo))
for i, ii := range indexInfo {
result[i] = pproto.RowResponse{
Headers: []*pproto.ColumnInfo{
{Name: "Table", Datatype: "string"},
},
@ -74,24 +66,13 @@ func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Sh
{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
// Sort the result.
return OrderBy(result, []string{"Table"}, []string{"asc"}), nil
}
func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.StreamClient, error) {
func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.ToRowser, error) {
indexName := showStmt.OnTable.ToViewName().Name.String()
index, err := s.api.Index(ctx, indexName)
if err != nil {
@ -101,16 +82,8 @@ func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Sh
return nil, errors.WithMessage(pilosa.ErrIndexNotFound, indexName)
}
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)
result := make(pproto.ConstRowser, 0, len(fields))
for _, f := range fields {
if f.Name() == "_exists" {
continue
@ -120,7 +93,7 @@ func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Sh
if err != nil {
return nil, errors.Wrapf(err, "field %s", f.Name())
}
rr := &pproto.RowResponse{
result = append(result, pproto.RowResponse{
Headers: []*pproto.ColumnInfo{
{Name: "Field", Datatype: "string"},
{Name: "Type", Datatype: "string"},
@ -129,20 +102,9 @@ func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Sh
{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
// Sort the result.
return OrderBy(result, []string{"Field"}, []string{"asc"}), nil
}

1
statik/.gitignore vendored Normal file
View file

@ -0,0 +1 @@
/statik.go

37
statik/filesystem.go Normal file
View file

@ -0,0 +1,37 @@
// Copyright 2017 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.
//
//go:generate statik -src=../lattice/build -dest=../
//
// Package statik contains static assets for the Lattice UI. `go generate` or
// `make generate-statik` will produce statik.go, which is ignored by git.
package statik
import (
"net/http"
"github.com/pilosa/pilosa/v2"
"github.com/rakyll/statik/fs"
)
// Ensure nopFileSystem implements interface.
var _ pilosa.FileSystem = &FileSystem{}
// FileSystem represents a static FileSystem.
type FileSystem struct{}
// New is a statik implementation of FileSystem New method.
func (s *FileSystem) New() (http.FileSystem, error) {
return fs.New()
}

5
version.go
View file

@ -20,6 +20,7 @@ var Version string
var Commit string
var Variant string
var BuildTime string
var LatticeCommit string
func VersionInfo() string {
var prefix string
@ -50,3 +51,7 @@ func VersionInfo() string {
return prefix + "Pilosa" + suffix
}
func LatticeVersionInfo() string {
return "g" + LatticeCommit
}