From 68542ddc357f0eeaf27aed4b84aa1cf9954e4efd Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Wed, 26 Aug 2020 09:40:23 -0400 Subject: [PATCH 01/26] force ranked cache recalculation in Top after a skipped invalidation --- cache.go | 25 +++++++++++++++++++++++++ cache_test.go | 29 +++++++++++++++++++++++++++++ metrics.go | 2 ++ 3 files changed, 56 insertions(+) diff --git a/cache.go b/cache.go index 9195f63c9..b1208f7ea 100644 --- a/cache.go +++ b/cache.go @@ -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 } diff --git a/cache_test.go b/cache_test.go index d4c0982e9..fe7a3848a 100644 --- a/cache_test.go +++ b/cache_test.go @@ -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) + } +} diff --git a/metrics.go b/metrics.go index 8d58f8a73..961af53c7 100644 --- a/metrics.go +++ b/metrics.go @@ -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" From 2b1c9950f4fc4dc40a1616278c86c13b85e37a1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Fri, 28 Aug 2020 17:24:49 +0200 Subject: [PATCH 02/26] Add rich error types to gRPC interface --- server/grpc.go | 60 +++++++++++++++++++++++++++++++++++++++++++++----- sql/reduce.go | 8 +++---- 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/server/grpc.go b/server/grpc.go index b93b37fe8..07b41273a 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -66,14 +66,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()) } diff --git a/sql/reduce.go b/sql/reduce.go index be0e98695..b349f8ceb 100644 --- a/sql/reduce.go +++ b/sql/reduce.go @@ -282,20 +282,20 @@ func (v *ValCountFuncReducer) Reduce(c pproto.StreamClient, s pproto.StreamServe if len(cols) == 0 { return s.Send(pproto.ErrorCode( errors.New("empty column set"), - codes.Unknown, + codes.NotFound, )) } if idxVal == -1 { return s.Send(pproto.ErrorCode( errors.New("result set has no column: value"), - codes.Unknown, + codes.NotFound, )) } if idxCnt == -1 { return s.Send(pproto.ErrorCode( errors.New("result set has no column: count"), - codes.Unknown, + codes.NotFound, )) } @@ -415,7 +415,7 @@ func (r *AssignHeadersReducer) Reduce(c pproto.StreamClient, s pproto.StreamServ if len(placement) > len(rr.Columns) { return s.Send(pproto.ErrorCode( errors.New("mismatched header placement and column count"), - codes.Unknown, + codes.InvalidArgument, )) } From f7a0e5f53689cb8f7a30edc739107bc56953ec1f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 31 Aug 2020 12:14:58 +0200 Subject: [PATCH 03/26] Fix error code for DeleteVDS --- server/grpc.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/grpc.go b/server/grpc.go index 07b41273a..5a5b699ef 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -162,7 +162,7 @@ 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 } From bfff643f34da4f8f5da4ae5293d247eb22823edf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 31 Aug 2020 12:16:07 +0200 Subject: [PATCH 04/26] Fix error code for PostVDS --- server/grpc.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/grpc.go b/server/grpc.go index 5a5b699ef..6ed428772 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -153,7 +153,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 } From 88a288e77522fa6f9e7c0eb7c0c082ba30324ea7 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Tue, 25 Aug 2020 19:34:29 -0500 Subject: [PATCH 05/26] Embed lattice via statik --- Makefile | 17 ++++++++++++++--- go.mod | 1 + go.sum | 2 ++ http/handler.go | 34 +++++++++++++++++++++++++++++----- server/server.go | 2 ++ statik/.gitignore | 1 + statik/filesystem.go | 37 +++++++++++++++++++++++++++++++++++++ 7 files changed, 86 insertions(+), 8 deletions(-) create mode 100644 statik/.gitignore create mode 100644 statik/filesystem.go diff --git a/Makefile b/Makefile index bd605f1d3..6b7ccfaf5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.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 require-statik test testv testv-race testvsub testvsub-race CLONE_URL=github.com/pilosa/pilosa VERSION := $(shell git describe --tags 2> /dev/null || echo unknown) @@ -135,10 +135,18 @@ prerelease-upload: install: go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa +build-lattice: require-yarn + git clone git@github.com:molecula/lattice.git + cd lattice && 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 +159,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 +355,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 diff --git a/go.mod b/go.mod index b8d683b60..a5e8fa41a 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 5c5045296..47fd55364 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/http/handler.go b/http/handler.go index ea14e097c..1aff0af04 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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 @@ -107,6 +109,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,6 +152,7 @@ func NewHandler(opts ...handlerOption) (*Handler, error) { } }) handler := &Handler{ + FileSystem: pilosa.NopFileSystem, logger: logger.NopLogger, closeTimeout: time.Second * 30, } @@ -338,7 +348,9 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { // newRouter creates a new mux http router. func newRouter(handler *Handler) *mux.Router { router := mux.NewRouter() - router.HandleFunc("/", handler.handleHome).Methods("GET").Name("Home") + router.HandleFunc("/", handler.handleLattice).Methods("GET") + router.HandleFunc("/{file}", handler.handleLattice).Methods("GET") + router.HandleFunc("/static/{file}", handler.handleLattice).Methods("GET") 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") @@ -422,6 +434,22 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.Handler.ServeHTTP(w, r) } +func (h *Handler) handleLattice(w http.ResponseWriter, r *http.Request) { + // If user is using curl, don't chuck HTML at them + if strings.HasPrefix(r.UserAgent(), "curl") { + http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information or try the Lattice UI by visiting this URL in your browser.", http.StatusNotFound) + return + } + filesystem, err := h.FileSystem.New() + + if err != nil { + _ = h.writeQueryResponse(w, r, &pilosa.QueryResponse{Err: err}) + h.logger.Printf("Lattice UI is not available. Please run `make generate-statik` before building Pilosa with `make install`.") + return + } + http.FileServer(filesystem).ServeHTTP(w, r) +} + // successResponse is a general success/error struct for http responses. type successResponse struct { h *Handler @@ -490,10 +518,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. diff --git a/server/server.go b/server/server.go index 546db5d41..bfd54cead 100644 --- a/server/server.go +++ b/server/server.go @@ -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), ) diff --git a/statik/.gitignore b/statik/.gitignore new file mode 100644 index 000000000..485c0c57d --- /dev/null +++ b/statik/.gitignore @@ -0,0 +1 @@ +/statik.go diff --git a/statik/filesystem.go b/statik/filesystem.go new file mode 100644 index 000000000..a5ec437b5 --- /dev/null +++ b/statik/filesystem.go @@ -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 Web 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() +} From d912403eb2ef3ace35b0ad7f8e8f565b144fbb2c Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Tue, 25 Aug 2020 19:43:14 -0500 Subject: [PATCH 06/26] Add missing file --- ctl/server.go | 2 +- filesystem.go | 42 ++++++++++++++++++++++++++++++++++++++++++ statik/filesystem.go | 2 +- 3 files changed, 44 insertions(+), 2 deletions(-) create mode 100644 filesystem.go diff --git a/ctl/server.go b/ctl/server.go index 23a269612..a64b63749 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -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/Lattice UI).") // Cluster flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)") diff --git a/filesystem.go b/filesystem.go new file mode 100644 index 000000000..62fc88d38 --- /dev/null +++ b/filesystem.go @@ -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") +} diff --git a/statik/filesystem.go b/statik/filesystem.go index a5ec437b5..c7f3141fc 100644 --- a/statik/filesystem.go +++ b/statik/filesystem.go @@ -14,7 +14,7 @@ // //go:generate statik -src=../lattice/build -dest=../ // -// Package statik contains static assets for the Web UI. `go generate` or +// 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 From 19ee27fdb6e79e4519336d9360006f5786da76ae Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 14:41:04 -0500 Subject: [PATCH 07/26] Use SPA handler to serve from filesystem, to test routing behavior --- http/handler.go | 41 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/http/handler.go b/http/handler.go index 1aff0af04..4a0771292 100644 --- a/http/handler.go +++ b/http/handler.go @@ -29,6 +29,7 @@ import ( _ "net/http/pprof" // Imported for its side-effect of registering pprof endpoints with the server. "net/url" "os" + "path/filepath" "reflect" "runtime/debug" "runtime/pprof" @@ -345,12 +346,43 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { }) } +// latticeHandler implements the http.Handler interface, so we can use it +// to respond to HTTP requests. The path to the static directory and +// path to the index file within that static directory are used to +// serve the Lattice UI in the given static directory +type latticeHandler struct { + staticPath string + indexPath string +} + +// ServeHTTP inspects the URL path to locate a file within the static dir +// on latticeHandler. If a file is found, it will be served. If not, the +// file located at the index path on the latticeHandler will be served. This +// is suitable behavior for serving an SPA. +func (h latticeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // get the absolute path to prevent directory traversal + path, err := filepath.Abs(r.URL.Path) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) // TODO + return + } + + path = filepath.Join(h.staticPath, path) + _, err = os.Stat(path) // TODO + if os.IsNotExist(err) { + http.ServeFile(w, r, filepath.Join(h.staticPath, h.indexPath)) + return + } else if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + http.FileServer(http.Dir(h.staticPath)).ServeHTTP(w, r) +} + // newRouter creates a new mux http router. func newRouter(handler *Handler) *mux.Router { router := mux.NewRouter() - router.HandleFunc("/", handler.handleLattice).Methods("GET") - router.HandleFunc("/{file}", handler.handleLattice).Methods("GET") - router.HandleFunc("/static/{file}", handler.handleLattice).Methods("GET") 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") @@ -412,6 +444,9 @@ 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") + lattice := latticeHandler{staticPath: "lattice/build", indexPath: "index.html"} + router.PathPrefix("/").Handler(lattice) + router.Use(handler.queryArgValidator) router.Use(handler.addQueryContext) router.Use(handler.extractTracing) From b039bd50a1ca2eaa95ada6b41bd58a824808477f Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 17:44:26 -0500 Subject: [PATCH 08/26] Switch to mux PathPrefix matcher entirely --- http/handler.go | 105 ++++++++++++++++++++++++++---------------------- 1 file changed, 57 insertions(+), 48 deletions(-) diff --git a/http/handler.go b/http/handler.go index 4a0771292..b1981c901 100644 --- a/http/handler.go +++ b/http/handler.go @@ -29,7 +29,6 @@ import ( _ "net/http/pprof" // Imported for its side-effect of registering pprof endpoints with the server. "net/url" "os" - "path/filepath" "reflect" "runtime/debug" "runtime/pprof" @@ -157,8 +156,6 @@ func NewHandler(opts ...handlerOption) (*Handler, error) { logger: logger.NopLogger, closeTimeout: time.Second * 30, } - handler.Handler = newRouter(handler) - handler.populateValidators() for _, opt := range opts { err := opt(handler) @@ -167,6 +164,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") } @@ -203,7 +204,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() @@ -346,40 +346,6 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { }) } -// latticeHandler implements the http.Handler interface, so we can use it -// to respond to HTTP requests. The path to the static directory and -// path to the index file within that static directory are used to -// serve the Lattice UI in the given static directory -type latticeHandler struct { - staticPath string - indexPath string -} - -// ServeHTTP inspects the URL path to locate a file within the static dir -// on latticeHandler. If a file is found, it will be served. If not, the -// file located at the index path on the latticeHandler will be served. This -// is suitable behavior for serving an SPA. -func (h latticeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - // get the absolute path to prevent directory traversal - path, err := filepath.Abs(r.URL.Path) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) // TODO - return - } - - path = filepath.Join(h.staticPath, path) - _, err = os.Stat(path) // TODO - if os.IsNotExist(err) { - http.ServeFile(w, r, filepath.Join(h.staticPath, h.indexPath)) - return - } else if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - http.FileServer(http.Dir(h.staticPath)).ServeHTTP(w, r) -} - // newRouter creates a new mux http router. func newRouter(handler *Handler) *mux.Router { router := mux.NewRouter() @@ -440,12 +406,19 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.handleDeleteRemoteAvailableShard).Methods("DELETE") router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes") router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client - 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") - lattice := latticeHandler{staticPath: "lattice/build", indexPath: "index.html"} - router.PathPrefix("/").Handler(lattice) + // 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) @@ -469,20 +442,56 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.Handler.ServeHTTP(w, r) } -func (h *Handler) handleLattice(w http.ResponseWriter, r *http.Request) { - // If user is using curl, don't chuck HTML at them +// 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 lattice UI at %s", 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") { http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information or try the Lattice UI by visiting this URL in your browser.", http.StatusNotFound) return } - filesystem, err := h.FileSystem.New() - if err != nil { - _ = h.writeQueryResponse(w, r, &pilosa.QueryResponse{Err: err}) - h.logger.Printf("Lattice UI is not available. Please run `make generate-statik` before building Pilosa with `make install`.") + // /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 + } + + if s.statikFS == nil { + msg := "Lattice 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 } - http.FileServer(filesystem).ServeHTTP(w, r) + http.FileServer(s.statikFS).ServeHTTP(w, r) + /* + filesystem, err := s.handler.FileSystem.New() // TODO + if err != nil { + s.handler.logger.Printf("Lattice UI is not available. Please run `make generate-statik` before building Pilosa with `make install`.") + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + http.FileServer(filesystem).ServeHTTP(w, r) + */ } // successResponse is a general success/error struct for http responses. From 99869ce79256c14374300eaf81f222d6695f6e62 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 18:12:03 -0500 Subject: [PATCH 09/26] Minor fixes --- Makefile | 6 ++++-- README.md | 2 ++ http/handler.go | 12 ++---------- 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/Makefile b/Makefile index 6b7ccfaf5..76b6d5487 100644 --- a/Makefile +++ b/Makefile @@ -135,9 +135,11 @@ prerelease-upload: install: go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa -build-lattice: require-yarn +lattice: git clone git@github.com:molecula/lattice.git - cd lattice && yarn install && yarn build + +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 diff --git a/README.md b/README.md index 8057a469d..4543a3524 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/http/handler.go b/http/handler.go index b1981c901..24e7e8142 100644 --- a/http/handler.go +++ b/http/handler.go @@ -406,6 +406,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.handleDeleteRemoteAvailableShard).Methods("DELETE") router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes") router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client + 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") @@ -454,7 +455,7 @@ type statikHandler struct { func NewStatikHandler(h *Handler) statikHandler { fs, err := h.FileSystem.New() if err == nil { - h.logger.Printf("enabled lattice UI at %s", h.api.Node().URI) + h.logger.Printf("enabled Lattice UI at %s", h.api.Node().URI) } return statikHandler{ @@ -483,15 +484,6 @@ func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } http.FileServer(s.statikFS).ServeHTTP(w, r) - /* - filesystem, err := s.handler.FileSystem.New() // TODO - if err != nil { - s.handler.logger.Printf("Lattice UI is not available. Please run `make generate-statik` before building Pilosa with `make install`.") - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - http.FileServer(filesystem).ServeHTTP(w, r) - */ } // successResponse is a general success/error struct for http responses. From 6e917b6a9ad7564671d76cfbc9823f9043269fe9 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 19:26:14 -0500 Subject: [PATCH 10/26] Log lattice version info --- Makefile | 3 ++- api.go | 5 +++++ http/handler.go | 2 +- version.go | 5 +++++ 4 files changed, 13 insertions(+), 2 deletions(-) diff --git a/Makefile b/Makefile index 76b6d5487..246e59f0c 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,7 @@ 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 || echo none) 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)) diff --git a/api.go b/api.go index 05808f61d..773c73ae4 100644 --- a/api.go +++ b/api.go @@ -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 diff --git a/http/handler.go b/http/handler.go index 24e7e8142..f522baf32 100644 --- a/http/handler.go +++ b/http/handler.go @@ -455,7 +455,7 @@ type statikHandler struct { func NewStatikHandler(h *Handler) statikHandler { fs, err := h.FileSystem.New() if err == nil { - h.logger.Printf("enabled Lattice UI at %s", h.api.Node().URI) + h.logger.Printf("enabled Lattice UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI) } return statikHandler{ diff --git a/version.go b/version.go index 8671d683f..4313c9e33 100644 --- a/version.go +++ b/version.go @@ -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 +} From 659a2bb560e5c66371ee06039a61fc95873ab4af Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 20:12:51 -0500 Subject: [PATCH 11/26] Silence stderr in makefile --- Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Makefile b/Makefile index 246e59f0c..241904367 100644 --- a/Makefile +++ b/Makefile @@ -2,7 +2,7 @@ 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 || echo none) +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))) From 384b6511cd22d494f8aa048aaafa8528ca3bae51 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Thu, 27 Aug 2020 22:38:49 -0500 Subject: [PATCH 12/26] Replace null with [] in /schema field response --- holder.go | 1 + 1 file changed, 1 insertion(+) diff --git a/holder.go b/holder.go index abe21b9ce..c531039a4 100644 --- a/holder.go +++ b/holder.go @@ -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, "_") { From f9d30408275e67b76797258c9d06055c6ac335e8 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Fri, 28 Aug 2020 00:33:54 -0500 Subject: [PATCH 13/26] Update gitignore and makefile --- .gitignore | 1 + Makefile | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 8468c5367..8c58b64ee 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ vendor .DS_Store build *~ +lattice \ No newline at end of file diff --git a/Makefile b/Makefile index 241904367..93e5407f3 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.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 require-statik 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) From a3e122dc5ef16dfdb27404e65cf9535278c5cfca Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Fri, 28 Aug 2020 19:58:25 -0500 Subject: [PATCH 14/26] Fix CORS support by applying middleware to router. --- http/handler.go | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/http/handler.go b/http/handler.go index f522baf32..2da8c4c67 100644 --- a/http/handler.go +++ b/http/handler.go @@ -70,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. @@ -94,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 } } @@ -347,7 +349,7 @@ 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("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST").Name("PostClusterResizeAbort") router.HandleFunc("/cluster/resize/remove-node", handler.handlePostClusterResizeRemoveNode).Methods("POST").Name("PostClusterResizeRemoveNode") @@ -425,7 +427,18 @@ func newRouter(handler *Handler) *mux.Router { 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. From b14ebcadae1dc0148f1334f37c1c33db17cc06c0 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Sun, 30 Aug 2020 08:16:19 -0500 Subject: [PATCH 15/26] Unexport statik filesystem --- .gitignore | 2 +- http/handler.go | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.gitignore b/.gitignore index 8c58b64ee..7a41479b9 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,4 @@ vendor .DS_Store build *~ -lattice \ No newline at end of file +lattice diff --git a/http/handler.go b/http/handler.go index 2da8c4c67..c8953984c 100644 --- a/http/handler.go +++ b/http/handler.go @@ -56,7 +56,7 @@ import ( type Handler struct { Handler http.Handler - FileSystem pilosa.FileSystem + fileSystem pilosa.FileSystem logger logger.Logger @@ -113,7 +113,7 @@ func OptHandlerAPI(api *pilosa.API) handlerOption { func OptHandlerFileSystem(fs pilosa.FileSystem) handlerOption { return func(h *Handler) error { - h.FileSystem = fs + h.fileSystem = fs return nil } } @@ -154,7 +154,7 @@ func NewHandler(opts ...handlerOption) (*Handler, error) { } }) handler := &Handler{ - FileSystem: pilosa.NopFileSystem, + fileSystem: pilosa.NopFileSystem, logger: logger.NopLogger, closeTimeout: time.Second * 30, } @@ -466,7 +466,7 @@ type statikHandler struct { // NewStatikHandler returns a new instance of statikHandler func NewStatikHandler(h *Handler) statikHandler { - fs, err := h.FileSystem.New() + fs, err := h.fileSystem.New() if err == nil { h.logger.Printf("enabled Lattice UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI) } From 627b50d89df662aa4395998a4069dd6142ead7f4 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Mon, 31 Aug 2020 08:46:53 -0600 Subject: [PATCH 16/26] misc import optimizations --- rbf/cursor.go | 10 +++++----- rbf/db.go | 1 - rbf/rbf.go | 14 +++++++------- rbf/tx.go | 24 ++++++++++++++++++------ rbf/wal.go | 8 ++++++-- 5 files changed, 36 insertions(+), 21 deletions(-) diff --git a/rbf/cursor.go b/rbf/cursor.go index 5045565d2..bdd67e0ad 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -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 { diff --git a/rbf/db.go b/rbf/db.go index 8816fdedd..fcb5f86a9 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -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() } diff --git a/rbf/rbf.go b/rbf/rbf.go index 728f530dc..f152e3499 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -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") } } diff --git a/rbf/tx.go b/rbf/tx.go index 9135ef752..b884e70ac 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -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 diff --git a/rbf/wal.go b/rbf/wal.go index aee4b07a3..fe7fcd40f 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -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 } From cfcc1da0bb75a6613c9ac3924b705934cf6d2797 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 24 Aug 2020 12:15:06 -0400 Subject: [PATCH 17/26] add primitive types to pg encoder --- server/pg.go | 45 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/server/pg.go b/server/pg.go index 28bbef7af..1edd94ab0 100644 --- a/server/pg.go +++ b/server/pg.go @@ -381,6 +381,51 @@ func pgWriteResult(w pg.QueryResultWriter, result interface{}) error { 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 default: return errors.Errorf("result type %T not yet supported", result) } From 542eb2dba33c29d2e9a90bf3e827b20d3e187161 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Tue, 25 Aug 2020 10:59:56 -0400 Subject: [PATCH 18/26] pg formatter tests --- executor.go | 2 + pg/pgtest/handler.go | 42 ++++++ server/pg.go | 27 +++- server/pg_test.go | 313 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 377 insertions(+), 7 deletions(-) create mode 100644 server/pg_test.go diff --git a/executor.go b/executor.go index 16c9dec39..036d1572d 100644 --- a/executor.go +++ b/executor.go @@ -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{ diff --git a/pg/pgtest/handler.go b/pg/pgtest/handler.go index e6cb22e12..0187d6c69 100644 --- a/pg/pgtest/handler.go +++ b/pg/pgtest/handler.go @@ -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) diff --git a/server/pg.go b/server/pg.go index 1edd94ab0..964472ece 100644 --- a/server/pg.go +++ b/server/pg.go @@ -28,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" @@ -49,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, @@ -66,6 +62,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) @@ -197,6 +203,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) @@ -316,10 +324,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: diff --git a/server/pg_test.go b/server/pg_test.go new file mode 100644 index 000000000..197632b59 --- /dev/null +++ b/server/pg_test.go @@ -0,0 +1,313 @@ +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) + } + } + }) + } +} From 9f368b06bd1f844094259d56e8ab38b2cba03085 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 31 Aug 2020 12:42:37 -0400 Subject: [PATCH 19/26] add licesnse header to pg formatter test --- server/pg_test.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/server/pg_test.go b/server/pg_test.go index 197632b59..86d743ade 100644 --- a/server/pg_test.go +++ b/server/pg_test.go @@ -1,3 +1,17 @@ +// 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 ( From d421558f5838128df60406c3f9fac74d16a55e64 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 31 Aug 2020 11:49:50 -0400 Subject: [PATCH 20/26] fix SQL memory leak --- proto/interface.go | 111 ++--------- server/grpc.go | 27 +-- server/pg.go | 25 --- server/sql.go | 4 +- sql/ddl.go | 6 +- sql/reduce.go | 452 ++++++++++++++++++--------------------------- sql/select.go | 192 +++++++++++-------- sql/show.go | 64 ++----- 8 files changed, 343 insertions(+), 538 deletions(-) diff --git a/proto/interface.go b/proto/interface.go index 6a6df7213..af2599707 100644 --- a/proto/interface.go +++ b/proto/interface.go @@ -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 +} diff --git a/server/grpc.go b/server/grpc.go index 6ed428772..2c55bc9e0 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -18,7 +18,6 @@ import ( "context" "crypto/tls" "fmt" - "io" "net" "strings" "sync" @@ -167,7 +166,7 @@ func (h *GRPCHandler) DeleteVDS(ctx context.Context, req *pb.DeleteVDSRequest) ( return &pb.DeleteVDSResponse{}, nil } -func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.StreamClient, error) { +func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.ToRowser, error) { return execSQL(ctx, h.api, h.logger, queryStr) } @@ -178,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. @@ -212,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. diff --git a/server/pg.go b/server/pg.go index 964472ece..49f697606 100644 --- a/server/pg.go +++ b/server/pg.go @@ -19,7 +19,6 @@ import ( "crypto/tls" "encoding/json" "fmt" - "io" "net" "strconv" "strings" @@ -358,28 +357,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: @@ -392,8 +369,6 @@ 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", diff --git a/server/sql.go b/server/sql.go index 7647e0377..26686c561 100644 --- a/server/sql.go +++ b/server/sql.go @@ -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) diff --git a/sql/ddl.go b/sql/ddl.go index 3b693850e..fd098f6c5 100644 --- a/sql/ddl.go +++ b/sql/ddl.go @@ -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 } diff --git a/sql/reduce.go b/sql/reduce.go index b349f8ceb..e670d1220 100644 --- a/sql/reduce.go +++ b/sql/reduce.go @@ -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.NotFound, - )) + return errors.New("empty column set") } if idxVal == -1 { - return s.Send(pproto.ErrorCode( - errors.New("result set has no column: value"), - codes.NotFound, - )) + 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.NotFound, - )) + 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.InvalidArgument, - )) - } - - 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 ( diff --git a/sql/select.go b/sql/select.go index a195a7c85..8d125a96c 100644 --- a/sql/select.go +++ b/sql/select.go @@ -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,15 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult }, }, }, - }) - respRows.Send(pproto.EOF) //nolint:errcheck - }() + }, + } default: return nil, fmt.Errorf("unsupported result type %T", res) } - // Apply Reducers - result := respRows - for _, red := range mr.Reducers { - out := pproto.NewRowBuffer(0) - - // Run Reducers asyncronously. - // TODO: stop swallowing this error. - // TODO: does this need an EOF as input? - go red.Reduce(result, out) //nolint:errcheck - - result = out + // Apply reducers. + for _, reducer := range mr.Reducers { + result = reducer(result) } return result, nil @@ -172,10 +148,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 +254,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 +338,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 +363,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 +401,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 +412,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 +485,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 +633,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 +705,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 +808,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 } diff --git a/sql/show.go b/sql/show.go index d145344d2..062a80d4a 100644 --- a/sql/show.go +++ b/sql/show.go @@ -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 } From ceb72fc3a9886281f0f1d311c19cadd92ed9ea5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 31 Aug 2020 23:28:35 +0200 Subject: [PATCH 21/26] support null results --- server/pg.go | 4 ++++ sql/select.go | 3 +++ 2 files changed, 7 insertions(+) diff --git a/server/pg.go b/server/pg.go index 49f697606..d29168ea0 100644 --- a/server/pg.go +++ b/server/pg.go @@ -414,6 +414,10 @@ func pgWriteResult(w pg.QueryResultWriter, result interface{}) error { } return nil + + case nil: + return nil + default: return errors.Errorf("result type %T not yet supported", result) } diff --git a/sql/select.go b/sql/select.go index 8d125a96c..c3b2bf020 100644 --- a/sql/select.go +++ b/sql/select.go @@ -127,6 +127,9 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult }, }, } + case nil: + result = pproto.ConstRowser{} + default: return nil, fmt.Errorf("unsupported result type %T", res) } From a597a79e2df3e51ae628a527a0c33888acc23c4f Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 31 Aug 2020 12:49:47 -0500 Subject: [PATCH 22/26] Add embedded UI to default release process --- Makefile | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/Makefile b/Makefile index 93e5407f3..560c0faa0 100644 --- a/Makefile +++ b/Makefile @@ -101,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 From 4ce3879e2cc5805146b52f0789e0cbee84df5cb9 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 31 Aug 2020 17:41:57 -0500 Subject: [PATCH 23/26] Revert UI->Lattice name change, correct the ordering of error checks in statikHandler --- ctl/server.go | 2 +- http/handler.go | 17 +++++++++-------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/ctl/server.go b/ctl/server.go index a64b63749..968b2bd6d 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -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/Lattice UI).") + 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)") diff --git a/http/handler.go b/http/handler.go index c8953984c..2d481d22f 100644 --- a/http/handler.go +++ b/http/handler.go @@ -468,7 +468,7 @@ type statikHandler struct { func NewStatikHandler(h *Handler) statikHandler { fs, err := h.fileSystem.New() if err == nil { - h.logger.Printf("enabled Lattice UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI) + h.logger.Printf("enabled Web UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI) } return statikHandler{ @@ -478,8 +478,15 @@ func NewStatikHandler(h *Handler) statikHandler { } func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + 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 + } + if strings.HasPrefix(r.UserAgent(), "curl") { - http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information or try the Lattice UI by visiting this URL in your browser.", http.StatusNotFound) + http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information or try the Web UI by visiting this URL in your browser.", http.StatusNotFound) return } @@ -490,12 +497,6 @@ func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { r.URL = url } - if s.statikFS == nil { - msg := "Lattice 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 - } http.FileServer(s.statikFS).ServeHTTP(w, r) } From 007b3ffc3c37dc9e742ed26703eee3074dee7ec8 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Tue, 1 Sep 2020 11:32:58 -0500 Subject: [PATCH 24/26] re-re-arrange error checks --- http/handler.go | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/http/handler.go b/http/handler.go index 2d481d22f..60624e5e8 100644 --- a/http/handler.go +++ b/http/handler.go @@ -478,6 +478,15 @@ func NewStatikHandler(h *Handler) statikHandler { } 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) @@ -485,11 +494,6 @@ func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - if strings.HasPrefix(r.UserAgent(), "curl") { - http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information or try the Web UI by visiting this URL in your browser.", http.StatusNotFound) - 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" { From d02cb10687ef2a9c6b65c64971b432e15514c494 Mon Sep 17 00:00:00 2001 From: Travis Date: Tue, 1 Sep 2020 17:34:29 -0500 Subject: [PATCH 25/26] return *Row instead of Row on empty key result --- executor.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/executor.go b/executor.go index 036d1572d..f5fc4fa2c 100644 --- a/executor.go +++ b/executor.go @@ -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{}} From 32b5826d1afb118c5dcd67943d10ebadc6c0283f Mon Sep 17 00:00:00 2001 From: Travis Date: Fri, 28 Aug 2020 16:53:42 -0500 Subject: [PATCH 26/26] fix bug on left/right join mapping --- server/grpc_test.go | 40 ++++++++++++++++++++++++++++++++++++++++ sql/extract.go | 4 ++-- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/server/grpc_test.go b/server/grpc_test.go index f3521f1e9..1b5cc415e 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -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 { diff --git a/sql/extract.go b/sql/extract.go index 464d69383..419811efa 100644 --- a/sql/extract.go +++ b/sql/extract.go @@ -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