diff --git a/.gitignore b/.gitignore index 8468c5367..7a41479b9 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ vendor .DS_Store build *~ +lattice diff --git a/Makefile b/Makefile index bd605f1d3..93e5407f3 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,8 @@ -.PHONY: build check-clean clean cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race +.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race CLONE_URL=github.com/pilosa/pilosa VERSION := $(shell git describe --tags 2> /dev/null || echo unknown) +LATTICE_COMMIT := $(shell git -C lattice rev-parse --short HEAD 2>/dev/null) VARIANT = Molecula VERSION_ID = $(VERSION)-$(GOOS)-$(GOARCH) BRANCH := $(if $(TRAVIS_BRANCH),$(TRAVIS_BRANCH),$(if $(CIRCLE_BRANCH),$(CIRCLE_BRANCH),$(shell git rev-parse --abbrev-ref HEAD))) @@ -9,7 +10,7 @@ BRANCH_ID := $(BRANCH)-$(GOOS)-$(GOARCH) BUILD_TIME := $(shell date -u +%FT%T%z) SHARD_WIDTH = 20 COMMIT := $(shell git describe --exact-match >/dev/null 2>&1 || git rev-parse --short HEAD) -LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT)" +LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT) -X github.com/pilosa/pilosa/v2.LatticeCommit=$(LATTICE_COMMIT)" GO_VERSION=latest RELEASE ?= 0 RELEASE_ENABLED = $(subst 0,,$(RELEASE)) @@ -135,10 +136,20 @@ prerelease-upload: install: go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa +lattice: + git clone git@github.com:molecula/lattice.git + +build-lattice: lattice require-yarn + cd lattice && git pull && yarn install && yarn build + # `go generate` protocol buffers generate-protoc: require-protoc require-protoc-gen-gofast go generate github.com/pilosa/pilosa/v2/internal +# `go generate` statik assets (lattice UI) +generate-statik: build-lattice require-statik + go generate github.com/pilosa/pilosa/v2/statik + # `go generate` stringers generate-stringer: go generate github.com/pilosa/pilosa/v2 @@ -151,7 +162,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 +358,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/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/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/boltdb/translate.go b/boltdb/translate.go index 7a477db0b..4977c7670 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -271,9 +271,11 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) { } defer func() { _ = tx.Rollback() }() + bucket := tx.Bucket(bucketIDs) + keys := make([]string, len(ids)) for i, id := range ids { - keys[i] = findKeyByID(tx.Bucket(bucketIDs), id) + keys[i] = findKeyByID(bucket, id) } return keys, nil } diff --git a/cmd/export.go b/cmd/export.go index 25ca5f5a0..bceedb98a 100644 --- a/cmd/export.go +++ b/cmd/export.go @@ -50,7 +50,7 @@ The file does not contain any headers. flags.StringVarP(&Exporter.Index, "index", "i", "", "Pilosa index to export") flags.StringVarP(&Exporter.Field, "field", "f", "", "Field to export") flags.StringVarP(&Exporter.Path, "output-file", "o", "", "File to write export to - default stdout") - ctl.SetTLSConfig(flags, &Exporter.TLS.CertificatePath, &Exporter.TLS.CertificateKeyPath, &Exporter.TLS.CACertPath, &Exporter.TLS.SkipVerify, &Exporter.TLS.EnableClientVerification) + ctl.SetTLSConfig(flags, "", &Exporter.TLS.CertificatePath, &Exporter.TLS.CertificateKeyPath, &Exporter.TLS.CACertPath, &Exporter.TLS.SkipVerify, &Exporter.TLS.EnableClientVerification) return exportCmd } diff --git a/cmd/import.go b/cmd/import.go index c18db2b0d..cd091dc9a 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -63,7 +63,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.") flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.") flags.BoolVarP(&Importer.Clear, "clear", "", false, "Clear the data provided in the import.") - ctl.SetTLSConfig(flags, &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.CACertPath, &Importer.TLS.SkipVerify, &Importer.TLS.EnableClientVerification) + ctl.SetTLSConfig(flags, "", &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.CACertPath, &Importer.TLS.SkipVerify, &Importer.TLS.EnableClientVerification) return importCmd } diff --git a/ctl/common.go b/ctl/common.go index db788246e..8cd22451c 100644 --- a/ctl/common.go +++ b/ctl/common.go @@ -31,12 +31,12 @@ type CommandWithTLSSupport interface { } // SetTLSConfig creates common TLS flags -func SetTLSConfig(flags *pflag.FlagSet, certificatePath *string, certificateKeyPath *string, caCertPath *string, skipVerify *bool, enableClientVerification *bool) { - flags.StringVarP(certificatePath, "tls.certificate", "", "", "TLS certificate path (usually has the .crt or .pem extension)") - flags.StringVarP(certificateKeyPath, "tls.key", "", "", "TLS certificate key path (usually has the .key extension)") - flags.StringVarP(caCertPath, "tls.ca-certificate", "", "", "TLS CA certificate path (usually has the .pem extension)") - flags.BoolVarP(skipVerify, "tls.skip-verify", "", false, "Skip TLS certificate server verification (not secure)") - flags.BoolVarP(enableClientVerification, "tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections") +func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string, certificateKeyPath *string, caCertPath *string, skipVerify *bool, enableClientVerification *bool) { + flags.StringVarP(certificatePath, prefix+"tls.certificate", "", "", "TLS certificate path (usually has the .crt or .pem extension)") + flags.StringVarP(certificateKeyPath, prefix+"tls.key", "", "", "TLS certificate key path (usually has the .key extension)") + flags.StringVarP(caCertPath, prefix+"tls.ca-certificate", "", "", "TLS CA certificate path (usually has the .pem extension)") + flags.BoolVarP(skipVerify, prefix+"tls.skip-verify", "", false, "Skip TLS certificate server verification (not secure)") + flags.BoolVarP(enableClientVerification, prefix+"tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections") } // commandClient returns a pilosa.InternalHTTPClient for the command diff --git a/ctl/server.go b/ctl/server.go index 378f951a3..a64b63749 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -36,10 +36,10 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.Uint64Var(&srv.Config.MaxFileCount, "max-file-count", srv.Config.MaxFileCount, "Soft limit on the maximum number of fragment files Pilosa keeps open simultaneously.") // TLS - SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.CACertPath, &srv.Config.TLS.SkipVerify, &srv.Config.TLS.EnableClientVerification) + 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)") @@ -90,5 +90,11 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Txsrc, "tx", "", "", "transaction/storage to use: one of roaring, rbf, badger, rbf_roaring, roaring_rbf, badger_roaring, roaring_badger, badger_rbf, or rbf_badger (default roaring)") // Postgres endpoint - flags.StringVar(&srv.Config.Postgres.Addr, "postgres.addr", "", "address to which to bind a postgres endpoint") + flags.StringVar(&srv.Config.Postgres.Bind, "postgres.bind", srv.Config.Postgres.Bind, "Address to which to bind a postgres endpoint (leave blank to disable)") + SetTLSConfig(flags, "postgres.", &srv.Config.Postgres.TLS.CertificatePath, &srv.Config.Postgres.TLS.CertificateKeyPath, &srv.Config.Postgres.TLS.CACertPath, &srv.Config.Postgres.TLS.SkipVerify, &srv.Config.Postgres.TLS.EnableClientVerification) + flags.DurationVar((*time.Duration)(&srv.Config.Postgres.StartupTimeout), "postgres.startup-timeout", time.Duration(srv.Config.Postgres.StartupTimeout), "Timeout for postgres connection startup. (set 0 to disable)") + flags.DurationVar((*time.Duration)(&srv.Config.Postgres.ReadTimeout), "postgres.read-timeout", time.Duration(srv.Config.Postgres.ReadTimeout), "Timeout for reads on a postgres connection. (set 0 to disable; does not include connection idling)") + flags.DurationVar((*time.Duration)(&srv.Config.Postgres.WriteTimeout), "postgres.write-timeout", time.Duration(srv.Config.Postgres.WriteTimeout), "Timeout for writes on a postgres connection. (set 0 to disable)") + flags.Uint32Var(&srv.Config.Postgres.MaxStartupSize, "postgres.max-startup-size", srv.Config.Postgres.MaxStartupSize, "Maximum acceptable size of a postgres startup packet, in bytes. (set 0 to disable)") + flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)") } diff --git a/executor.go b/executor.go index f176b2a42..16c9dec39 100644 --- a/executor.go +++ b/executor.go @@ -5016,6 +5016,38 @@ func (e *executor) collectResultIDs(index string, idx *Index, call *pql.Call, re return nil } +func (e *executor) translateFieldIDs(field *Field, ids map[uint64]struct{}) (map[uint64]string, error) { + idList := make([]uint64, len(ids)) + { + i := 0 + for id := range ids { + idList[i] = id + i++ + } + } + keyList, err := field.TranslateStore().TranslateIDs(idList) + if err != nil { + return nil, err + } + mapped := make(map[uint64]string, len(idList)) + for i, key := range keyList { + mapped[idList[i]] = key + } + return mapped, nil +} + +// preTranslateMatrixSet translates the IDs of a set field in an extracted matrix. +func (e *executor) preTranslateMatrixSet(mat ExtractedIDMatrix, fieldIdx uint, field *Field) (map[uint64]string, error) { + ids := make(map[uint64]struct{}, len(mat.Columns)) + for _, col := range mat.Columns { + for _, v := range col.Rows[fieldIdx] { + ids[v] = struct{}{} + } + } + + return e.translateFieldIDs(field, ids) +} + func (e *executor) translateResult(ctx context.Context, index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]string) (interface{}, error) { switch result := result.(type) { case *Row: @@ -5094,13 +5126,17 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index return nil, fmt.Errorf("field %q not found", fieldName) } if field.Keys() { + ids := make([]uint64, len(result.Pairs)) + for i := range result.Pairs { + ids[i] = result.Pairs[i].ID + } + keys, err := field.TranslateStore().TranslateIDs(ids) + if err != nil { + return nil, err + } other := make([]Pair, len(result.Pairs)) for i := range result.Pairs { - key, err := field.TranslateStore().TranslateID(result.Pairs[i].ID) - if err != nil { - return nil, err - } - other[i] = Pair{Key: key, Count: result.Pairs[i].Count} + other[i] = Pair{Key: keys[i], Count: result.Pairs[i].Count} } return &PairsField{ Pairs: other, @@ -5110,40 +5146,71 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index } case []GroupCount: - other := make([]GroupCount, 0) + fieldIDs := make(map[*Field]map[uint64]struct{}) + foreignIDs := make(map[*Field]map[uint64]struct{}) for _, gl := range result { - - group := make([]FieldRow, len(gl.Group)) - for i, g := range gl.Group { - group[i] = g - - // TODO: It may be useful to cache this field lookup. + for _, g := range gl.Group { field := idx.Field(g.Field) if field == nil { return nil, newNotFoundError(ErrFieldNotFound, g.Field) } if field.Keys() { - var key string - var err error - if fi := field.ForeignIndex(); fi != "" && g.Value != nil { - val := uint64(*g.Value) // not worried about overflow here because it's a foreign key - keys, err := e.Cluster.translateIndexIDs(ctx, fi, []uint64{val}) - if err != nil { - return nil, errors.Wrap(err, "translating foreign index in Group") - } - if len(keys) == 1 { - key = keys[0] - group[i].Value = nil // Remove value now that it has been translated. - } - } else { - key, err = field.TranslateStore().TranslateID(g.RowID) - if err != nil { - return nil, errors.Wrap(err, "translating row ID in Group") + if g.Value != nil { + if fi := field.ForeignIndex(); fi != "" { + m, ok := foreignIDs[field] + if !ok { + m = make(map[uint64]struct{}, len(result)) + foreignIDs[field] = m + } + + m[uint64(*g.Value)] = struct{}{} + continue } } - group[i].RowKey = key + + m, ok := fieldIDs[field] + if !ok { + m = make(map[uint64]struct{}, len(result)) + fieldIDs[field] = m + } + + m[g.RowID] = struct{}{} } } + } + + fieldTranslations := make(map[string]map[uint64]string) + for field, ids := range fieldIDs { + trans, err := e.translateFieldIDs(field, ids) + if err != nil { + return nil, errors.Wrapf(err, "translating IDs in field %q", field.Name()) + } + fieldTranslations[field.Name()] = trans + } + + foreignTranslations := make(map[string]map[uint64]string) + for field, ids := range foreignIDs { + trans, err := e.Cluster.translateIndexIDSet(ctx, field.ForeignIndex(), ids) + if err != nil { + return nil, errors.Wrapf(err, "translating foreign IDs from index %q", field.ForeignIndex()) + } + foreignTranslations[field.Name()] = trans + } + + other := make([]GroupCount, 0) + for _, gl := range result { + + group := make([]FieldRow, len(gl.Group)) + for i, g := range gl.Group { + if ft, ok := fieldTranslations[g.Field]; ok { + g.RowKey = ft[g.RowID] + } else if ft, ok := foreignTranslations[g.Field]; ok && g.Value != nil { + g.RowKey = ft[uint64(*g.Value)] + g.Value = nil + } + + group[i] = g + } other = append(other, GroupCount{ Group: group, @@ -5166,14 +5233,11 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index if field := idx.Field(fieldName); field == nil { return nil, newNotFoundError(ErrFieldNotFound, fieldName) } else if field.Keys() { - other.Keys = make([]string, len(result)) - for i, id := range result { - key, err := field.TranslateStore().TranslateID(id) - if err != nil { - return nil, errors.Wrap(err, "translating row ID") - } - other.Keys[i] = key + keys, err := field.TranslateStore().TranslateIDs(result) + if err != nil { + return nil, errors.Wrap(err, "translating row IDs") } + other.Keys = keys } else { other.Rows = result } @@ -5222,9 +5286,16 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index } case FieldTypeSet, FieldTypeTime: if field.Keys() { - translator := field.TranslateStore() + translations, err := e.preTranslateMatrixSet(result, uint(i), field) + if err != nil { + return nil, errors.Wrapf(err, "translating IDs of field %q", v) + } mapper = func(ids []uint64) (interface{}, error) { - return translator.TranslateIDs(ids) + keys := make([]string, len(ids)) + for i, id := range ids { + keys[i] = translations[id] + } + return keys, nil } } else { mapper = func(ids []uint64) (interface{}, error) { @@ -5236,13 +5307,16 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index } case FieldTypeMutex: if field.Keys() { - translator := field.TranslateStore() + translations, err := e.preTranslateMatrixSet(result, uint(i), field) + if err != nil { + return nil, errors.Wrapf(err, "translating IDs of field %q", v) + } mapper = func(ids []uint64) (interface{}, error) { switch len(ids) { case 0: return nil, nil case 1: - return translator.TranslateID(ids[0]) + return translations[ids[0]], nil default: return nil, errors.Errorf("mutex %q has too many values: %v", field.Name(), ids) } @@ -5260,14 +5334,50 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index } } case FieldTypeInt: - mapper = func(ids []uint64) (interface{}, error) { - switch len(ids) { - case 0: - return nil, nil - case 1: - return int64(ids[0]), nil - default: - return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) + if fi := field.ForeignIndex(); fi != "" { + if field.Keys() { + ids := make(map[uint64]struct{}, len(result.Columns)) + for _, col := range result.Columns { + for _, v := range col.Rows[i] { + ids[v] = struct{}{} + } + } + trans, err := e.Cluster.translateIndexIDSet(ctx, field.ForeignIndex(), ids) + if err != nil { + return nil, errors.Wrapf(err, "translating foreign IDs from index %q", field.ForeignIndex()) + } + mapper = func(ids []uint64) (interface{}, error) { + switch len(ids) { + case 0: + return nil, nil + case 1: + return trans[ids[0]], nil + default: + return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) + } + } + } else { + mapper = func(ids []uint64) (interface{}, error) { + switch len(ids) { + case 0: + return nil, nil + case 1: + return ids[0], nil + default: + return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) + } + } + } + } else { + mapper = func(ids []uint64) (interface{}, error) { + switch len(ids) { + case 0: + return nil, nil + case 1: + return int64(ids[0]), nil + default: + return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) + } } } case FieldTypeDecimal: 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/fragment.go b/fragment.go index 832d02f32..9922510ac 100644 --- a/fragment.go +++ b/fragment.go @@ -1058,7 +1058,7 @@ func (f *fragment) setValueBase(tx Tx, columnID uint64, bitDepth uint, value int } // importSetValue is a more efficient SetValue just for imports. -func (f *fragment) importSetValue(tx Tx, columnID uint64, bitDepth uint, value int64, clear bool) (changed int, err error) { // nolint: unparam +func (f *fragment) importSetValue(txb *TxBitmap, columnID uint64, bitDepth uint, value int64, clear bool) (changed int, err error) { // nolint: unparam // Convert value to an unsigned representation. uvalue := uint64(value) if value < 0 { @@ -1072,17 +1072,17 @@ func (f *fragment) importSetValue(tx Tx, columnID uint64, bitDepth uint, value i } if uvalue&(1< 0 { + } else if c { changed++ } } else { - changeCount, err := tx.Remove(f.index, f.field, f.view, f.shard, bit) + c, err := txb.Remove(bit) if err != nil { return changed, errors.Wrap(err, "removing") - } else if changeCount > 0 { + } else if c { changed++ } } @@ -1092,15 +1092,15 @@ func (f *fragment) importSetValue(tx Tx, columnID uint64, bitDepth uint, value i if p, err := f.pos(uint64(bsiExistsBit), columnID); err != nil { return changed, errors.Wrap(err, "getting not-null pos") } else if clear { - if c, err := tx.Remove(f.index, f.field, f.view, f.shard, p); err != nil { + if c, err := txb.Remove(p); err != nil { return changed, errors.Wrap(err, "removing not-null from storage") - } else if c > 0 { + } else if c { changed++ } } else { - if c, err := tx.Add(f.index, f.field, f.view, f.shard, !doBatched, p); err != nil { + if c, err := txb.Add(p); err != nil { return changed, errors.Wrap(err, "adding not-null to storage") - } else if c > 0 { + } else if c { changed++ } } @@ -1109,15 +1109,15 @@ func (f *fragment) importSetValue(tx Tx, columnID uint64, bitDepth uint, value i if p, err := f.pos(uint64(bsiSignBit), columnID); err != nil { return changed, errors.Wrap(err, "getting sign pos") } else if value >= 0 || clear { - if c, err := tx.Remove(f.index, f.field, f.view, f.shard, p); err != nil { + if c, err := txb.Remove(p); err != nil { return changed, errors.Wrap(err, "removing sign from storage") - } else if c > 0 { + } else if c { changed++ } } else { - if c, err := tx.Add(f.index, f.field, f.view, f.shard, !doBatched, p); err != nil { + if c, err := txb.Add(p); err != nil { return changed, errors.Wrap(err, "adding sign to storage") - } else if c > 0 { + } else if c { changed++ } } @@ -2460,17 +2460,20 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep if f.storage != nil { f.storage.OpWriter = nil } - totalChanges := 0 + + var totalChanges int if err := func() (err error) { + // Build changes into temporary bitmap. + txb := NewTxBitmap(tx, f.index, f.field, f.view, f.shard) for i := range columnIDs { columnID, value := columnIDs[i], values[i] - changed, err := f.importSetValue(tx, columnID, bitDepth, value, clear) - if err != nil { + if _, err := f.importSetValue(txb, columnID, bitDepth, value, clear); err != nil { return errors.Wrapf(err, "importSetValue") } - totalChanges += changed } - return nil + + // Flush changes in bulk back to the transaction. + return txb.Flush() }(); err != nil { _ = f.openStorage(true) return err 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/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, "_") { diff --git a/http/handler.go b/http/handler.go index ea14e097c..c8953984c 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 @@ -68,6 +70,8 @@ type Handler struct { closeTimeout time.Duration server *http.Server + + middleware []func(http.Handler) http.Handler } // externalPrefixFlag denotes endpoints that are intended to be exposed to clients. @@ -92,10 +96,10 @@ type handlerOption func(s *Handler) error func OptHandlerAllowedOrigins(origins []string) handlerOption { return func(h *Handler) error { - h.Handler = handlers.CORS( + h.middleware = append(h.middleware, handlers.CORS( handlers.AllowedOrigins(origins), handlers.AllowedHeaders([]string{"Content-Type"}), - )(h.Handler) + )) return nil } } @@ -107,6 +111,13 @@ func OptHandlerAPI(api *pilosa.API) handlerOption { } } +func OptHandlerFileSystem(fs pilosa.FileSystem) handlerOption { + return func(h *Handler) error { + h.fileSystem = fs + return nil + } +} + func OptHandlerLogger(logger logger.Logger) handlerOption { return func(h *Handler) error { h.logger = logger @@ -143,11 +154,10 @@ func NewHandler(opts ...handlerOption) (*Handler, error) { } }) handler := &Handler{ + fileSystem: pilosa.NopFileSystem, logger: logger.NopLogger, closeTimeout: time.Second * 30, } - handler.Handler = newRouter(handler) - handler.populateValidators() for _, opt := range opts { err := opt(handler) @@ -156,6 +166,10 @@ func NewHandler(opts ...handlerOption) (*Handler, error) { } } + // if OptHandlerFileSystem is used, it must be before newRouter is called + handler.Handler = newRouter(handler) + handler.populateValidators() + if handler.api == nil { return nil, errors.New("must pass OptHandlerAPI") } @@ -192,7 +206,6 @@ func (h *Handler) Close() error { func (h *Handler) populateValidators() { h.validators = map[string]*queryValidationSpec{} - h.validators["Home"] = queryValidationSpecRequired() h.validators["PostClusterResizeAbort"] = queryValidationSpecRequired() h.validators["PostClusterResizeRemoveNode"] = queryValidationSpecRequired() h.validators["PostClusterResizeSetCoordinator"] = queryValidationSpecRequired() @@ -336,9 +349,8 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { } // newRouter creates a new mux http router. -func newRouter(handler *Handler) *mux.Router { +func newRouter(handler *Handler) http.Handler { router := mux.NewRouter() - router.HandleFunc("/", handler.handleHome).Methods("GET").Name("Home") router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST").Name("PostClusterResizeAbort") router.HandleFunc("/cluster/resize/remove-node", handler.handlePostClusterResizeRemoveNode).Methods("POST").Name("PostClusterResizeRemoveNode") router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST").Name("PostClusterResizeSetCoordinator") @@ -400,11 +412,33 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/translate/index/{index}/{partition}", handler.handlePostTranslateIndexDB).Methods("POST").Name("PostTranslateIndexDB") router.HandleFunc("/internal/translate/field/{index}/{field}", handler.handlePostTranslateFieldDB).Methods("POST").Name("PostTranslateFieldDB") + // Endpoints to support lattice UI embedded via statik. + // The messiness here reflects the fact that assets live in a nontrivial + // directory structure that is controlled externally. + latticeHandler := NewStatikHandler(handler) + router.PathPrefix("/static").Handler(latticeHandler) + router.Path("/").Handler(latticeHandler) + router.Path("/vds").Handler(latticeHandler) + router.Path("/favicon.png").Handler(latticeHandler) + router.Path("/favicon.svg").Handler(latticeHandler) + router.Path("/manifest.json").Handler(latticeHandler) + router.Use(handler.queryArgValidator) router.Use(handler.addQueryContext) router.Use(handler.extractTracing) router.Use(handler.collectStats) - return router + var h http.Handler = router + for _, middleware := range handler.middleware { + // Ideally, we would use `router.Use` to inject middleware, + // instead of wrapping the handler. The reason we can't is + // because the router will only apply middleware to matched + // handlers. In this case, it won't match handlers with the + // OPTIONS method, needed by the CORS middleware. This issue + // is described in detail here: + // https://github.com/gorilla/handlers/issues/142 + h = middleware(h) + } + return h } // ServeHTTP handles an HTTP request. @@ -422,6 +456,49 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.Handler.ServeHTTP(w, r) } +// statikHandler implements the http.Handler interface, and responds to +// requests for static assets with the appropriate file contents embedded +// in a statik filesystem. +type statikHandler struct { + handler *Handler + statikFS http.FileSystem +} + +// NewStatikHandler returns a new instance of statikHandler +func NewStatikHandler(h *Handler) statikHandler { + fs, err := h.fileSystem.New() + if err == nil { + h.logger.Printf("enabled Lattice UI (%s) at %s", h.api.LatticeVersion(), h.api.Node().URI) + } + + return statikHandler{ + handler: h, + statikFS: fs, + } +} + +func (s statikHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.UserAgent(), "curl") { + 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 + } + + // /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(s.statikFS).ServeHTTP(w, r) +} + // successResponse is a general success/error struct for http responses. type successResponse struct { h *Handler @@ -490,10 +567,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/pg/protocol.go b/pg/protocol.go index c55decbd7..8a50baf05 100644 --- a/pg/protocol.go +++ b/pg/protocol.go @@ -73,9 +73,17 @@ func (p Protocol) String() string { // handle reads the startup packet and dispatches an appropriate protocol handler for the connection. func (s *Server) handle(ctx context.Context, conn net.Conn) (err error) { + var hasTLS bool + defer func() { cerr := conn.Close() if cerr != nil && err == nil { + if hasTLS { + if nerr, ok := cerr.(net.Error); ok && nerr.Timeout() { + // TLS does this sometimes. + return + } + } err = errors.Wrap(cerr, "closing connection") } }() @@ -151,6 +159,7 @@ startup: return errors.Wrap(err, "transferring startup deadline to TLS connection") } } + hasTLS = true goto startup } @@ -163,6 +172,11 @@ startup: goto startup } + if s.TLSConfig != nil && !hasTLS { + // Reject the unsecured connection. + return errors.Errorf("client at %s attempted to initiate an unsecured postgres conenction", conn.RemoteAddr()) + } + // Handle regular postgres. return s.handleStandard(ctx, proto, conn, data) } @@ -380,6 +394,14 @@ func (s *Server) handleStandard(ctx context.Context, proto Protocol, conn net.Co return errors.Wrap(err, "failed to send query error to client") } } else { + if !qwriter.wroteHeaders { + // The handler did not write headers. + // Write back an empty set of headers. + err = qwriter.WriteHeader() + if err != nil { + return errors.Wrap(err, "sending empty column headers") + } + } // The query completed normally. // Notify the client of completion. msg, err = encoder.CommandComplete(qwriter.tag) diff --git a/server/config.go b/server/config.go index 7437d65ed..c51a7f95f 100644 --- a/server/config.go +++ b/server/config.go @@ -168,9 +168,9 @@ type Config struct { } `toml:"profile"` Postgres struct { - // Addr is the address to which to bind a postgres endpoint. + // Bind is the address to which to bind a postgres endpoint. // If this is empty, no endpoint will be created. - Addr string `toml:"addr"` + Bind string `toml:"bind"` // TLS configuration for postgres connections. TLS TLSConfig `toml:"tls"` diff --git a/server/grpc.go b/server/grpc.go index b93b37fe8..6ed428772 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()) } @@ -105,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 } @@ -114,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 } diff --git a/server/server.go b/server/server.go index 90061a1b4..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" @@ -176,12 +177,12 @@ func (m *Command) Start() (err error) { // Initialize postgres. m.pgserver = nil - if m.Config.Postgres.Addr != "" { + if m.Config.Postgres.Bind != "" { var tlsConf *tls.Config if m.Config.Postgres.TLS.CertificatePath != "" { conf, err := GetTLSConfig(&m.Config.Postgres.TLS, m.logger.Logger()) if err != nil { - return errors.Wrap(err, "settuing up postgres TLS") + return errors.Wrap(err, "setting up postgres TLS") } tlsConf = conf } @@ -191,7 +192,7 @@ func (m *Command) Start() (err error) { m.pgserver.s.WriteTimeout = time.Duration(m.Config.Postgres.WriteTimeout) m.pgserver.s.MaxStartupSize = m.Config.Postgres.MaxStartupSize m.pgserver.s.ConnectionLimit = m.Config.Postgres.ConnectionLimit - err := m.pgserver.Start(m.Config.Postgres.Addr) + err := m.pgserver.Start(m.Config.Postgres.Bind) if err != nil { return errors.Wrap(err, "starting postgres") } @@ -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/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, )) } diff --git a/sql/router.go b/sql/router.go index f64daa338..4c15cb975 100644 --- a/sql/router.go +++ b/sql/router.go @@ -89,6 +89,8 @@ func newRouter() *router { selectRouter.addRoute("select fld1, count(fld1) from tbl where fld2=1 group by fld1", handlerSelectGroupBy{}) selectRouter.addRoute("select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi", handlerSelectJoin{}) + selectRouter.addRoute("select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1", handlerSelectJoin{}) + selectRouter.addRoute("select count(*) from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1 and fld2=2", handlerSelectJoin{}) selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi", handlerSelectJoin{}) selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1", handlerSelectJoin{}) selectRouter.addRoute("select _id from tbl1 INNER JOIN tbl2 ON tbl1._id = tbl2.bsi where fld1=1 and fld2=2", handlerSelectJoin{}) 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..c7f3141fc --- /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 Lattice UI. `go generate` or +// `make generate-statik` will produce statik.go, which is ignored by git. +package statik + +import ( + "net/http" + + "github.com/pilosa/pilosa/v2" + "github.com/rakyll/statik/fs" +) + +// Ensure nopFileSystem implements interface. +var _ pilosa.FileSystem = &FileSystem{} + +// FileSystem represents a static FileSystem. +type FileSystem struct{} + +// New is a statik implementation of FileSystem New method. +func (s *FileSystem) New() (http.FileSystem, error) { + return fs.New() +} diff --git a/tx.go b/tx.go index a5c154654..2b74b8fa6 100644 --- a/tx.go +++ b/tx.go @@ -231,3 +231,68 @@ type RawRoaringData struct { func (rr *RawRoaringData) Iterator() (roaring.RoaringIterator, error) { return roaring.NewRoaringIterator(rr.data) } + +// TxBitmap represents a bitmap that acts as a cache in front of a transaction. +// Updates to the bitmap first pull in containers as needed and update them +// in memory. The changes can be flushed in bulk using Flush(). +type TxBitmap struct { + b *roaring.Bitmap + tx Tx + index string + field string + view string + shard uint64 +} + +func NewTxBitmap(tx Tx, index, field, view string, shard uint64) *TxBitmap { + return &TxBitmap{ + b: roaring.NewBitmap(), + tx: tx, + index: index, + field: field, + view: view, + shard: shard, + } +} + +func (b *TxBitmap) Add(a ...uint64) (changed bool, err error) { + if err := b.ensureContainers(a...); err != nil { + return false, err + } + return b.b.Add(a...) +} + +func (b *TxBitmap) Remove(a ...uint64) (changed bool, err error) { + if err := b.ensureContainers(a...); err != nil { + return false, err + } + return b.b.Remove(a...) +} + +// ensureContainers pulls containers in from the transaction, if needed. +func (b *TxBitmap) ensureContainers(a ...uint64) error { + for _, v := range a { + key := highbits(v) + if b.b.Containers.Get(key) != nil { + continue + } + + c, err := b.tx.Container(b.index, b.field, b.view, b.shard, key) + if err != nil { + return err + } + b.b.Containers.Put(key, c) + } + return nil +} + +// Flush writes all containers in the bitmap back to the transaction. +func (b *TxBitmap) Flush() error { + for it, _ := b.b.Containers.Iterator(0); it.Next(); { + key, c := it.Value() + if err := b.tx.PutContainer(b.index, b.field, b.view, b.shard, key, c); err != nil { + return err + } + } + return nil +} 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 +}