diff --git a/cmd/root_test.go b/cmd/root_test.go index caf53d74b..ec998d346 100644 --- a/cmd/root_test.go +++ b/cmd/root_test.go @@ -189,11 +189,7 @@ bind = "127.0.0.1:10101" [cluster] replicas = 2 - partitions = 128 - hosts = [ - "127.0.0.1:10101", - "127.0.0.1:10111", - ]` + partitions = 128` if _, err := file.Write([]byte(config)); err != nil { t.Fatalf("writing config file: %v", err) } diff --git a/disco/disco.go b/disco/disco.go index 0725d2d82..a4029c97d 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -18,16 +18,21 @@ import ( "context" "fmt" "io" + "sync" "github.com/pilosa/pilosa/v2/roaring" ) var ( - ErrTooManyResults error = fmt.Errorf("too many results") - ErrNoResults error = fmt.Errorf("no results") - ErrKeyDeleted error = fmt.Errorf("key deleted") - ErrIndexExists error = fmt.Errorf("index already exists") - ErrFieldExists error = fmt.Errorf("field already exists") + ErrTooManyResults error = fmt.Errorf("too many results") + ErrNoResults error = fmt.Errorf("no results") + ErrKeyDeleted error = fmt.Errorf("key deleted") + ErrIndexExists error = fmt.Errorf("index already exists") + ErrIndexDoesNotExist error = fmt.Errorf("index does not exist") + ErrFieldExists error = fmt.Errorf("field already exists") + ErrFieldDoesNotExist error = fmt.Errorf("field does not exist") + ErrViewExists error = fmt.Errorf("view already exists") + ErrViewDoesNotExist error = fmt.Errorf("view does not exist") ) type Peer struct { @@ -84,6 +89,10 @@ type Stator interface { NodeStates(context.Context) (map[string]NodeState, error) } +// Schema is a map of all indexes, each of those being a map of fields, then +// views. +type Schema map[string]*Index + // Index is a struct which contains the data encoded for the index as well as // for each of its fields. type Index struct { @@ -99,7 +108,7 @@ type Field struct { } type Schemator interface { - Schema(ctx context.Context) (map[string]*Index, error) + Schema(ctx context.Context) (Schema, error) Index(ctx context.Context, name string) ([]byte, error) CreateIndex(ctx context.Context, name string, val []byte) error DeleteIndex(ctx context.Context, name string) error @@ -252,7 +261,7 @@ var NopSchemator Schemator = &nopSchemator{} type nopSchemator struct{} // Schema is a no-op implementation of the Schemator Schema method. -func (*nopSchemator) Schema(ctx context.Context) (map[string]*Index, error) { return nil, nil } +func (*nopSchemator) Schema(ctx context.Context) (Schema, error) { return nil, nil } // Index is a no-op implementation of the Schemator Index method. func (*nopSchemator) Index(ctx context.Context, name string) ([]byte, error) { return nil, nil } @@ -286,3 +295,160 @@ func (*nopSchemator) CreateView(ctx context.Context, index, field, view string, // DeleteView is a no-op implementation of the Schemator DeleteView method. func (*nopSchemator) DeleteView(ctx context.Context, index, field, view string) error { return nil } + +// InMemSchemator represents a Schemator that manages the schema in memory. The +// intention is that this would be used for testing. +var InMemSchemator Schemator = &inMemSchemator{ + schema: make(Schema), +} + +type inMemSchemator struct { + mu sync.RWMutex + schema Schema +} + +// Schema is an in-memory implementation of the Schemator Schema method. +func (s *inMemSchemator) Schema(ctx context.Context) (Schema, error) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.schema, nil +} + +// Index is an in-memory implementation of the Schemator Index method. +func (s *inMemSchemator) Index(ctx context.Context, name string) ([]byte, error) { + s.mu.RLock() + defer s.mu.RUnlock() + idx, ok := s.schema[name] + if !ok { + return nil, ErrIndexDoesNotExist + } + return idx.Data, nil +} + +// CreateIndex is an in-memory implementation of the Schemator CreateIndex method. +func (s *inMemSchemator) CreateIndex(ctx context.Context, name string, val []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + if idx, ok := s.schema[name]; ok { + // The current logic in pilosa doesn't allow us to return ErrIndexExists + // here, so for now we just update the Data value if the index already + // exists. + idx.Data = val + return nil + } + s.schema[name] = &Index{ + Data: val, + Fields: make(map[string]*Field), + } + return nil +} + +// DeleteIndex is an in-memory implementation of the Schemator DeleteIndex method. +func (s *inMemSchemator) DeleteIndex(ctx context.Context, name string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.schema, name) + return nil +} + +// Field is an in-memory implementation of the Schemator Field method. +func (s *inMemSchemator) Field(ctx context.Context, index, field string) ([]byte, error) { + s.mu.RLock() + defer s.mu.RUnlock() + idx, ok := s.schema[index] + if !ok { + return nil, ErrIndexDoesNotExist + } + fld, ok := idx.Fields[field] + if !ok { + return nil, ErrFieldDoesNotExist + } + return fld.Data, nil +} + +// CreateField is an in-memory implementation of the Schemator CreateField method. +func (s *inMemSchemator) CreateField(ctx context.Context, index, field string, val []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + idx, ok := s.schema[index] + if !ok { + return ErrIndexDoesNotExist + } + if fld, ok := idx.Fields[field]; ok { + // The current logic in pilosa doesn't allow us to return ErrFieldExists + // here, so for now we just update the Data value if the field already + // exists. + fld.Data = val + return nil + } + idx.Fields[field] = &Field{ + Data: val, + Views: make(map[string][]byte), + } + return nil +} + +// DeleteField is an in-memory implementation of the Schemator DeleteField method. +func (s *inMemSchemator) DeleteField(ctx context.Context, index, field string) error { + s.mu.Lock() + defer s.mu.Unlock() + idx, ok := s.schema[index] + if !ok { + return ErrIndexDoesNotExist + } + delete(idx.Fields, field) + return nil +} + +// View is an in-memory implementation of the Schemator View method. +func (s *inMemSchemator) View(ctx context.Context, index, field, view string) ([]byte, error) { + s.mu.RLock() + defer s.mu.RUnlock() + idx, ok := s.schema[index] + if !ok { + return nil, ErrIndexDoesNotExist + } + fld, ok := idx.Fields[field] + if !ok { + return nil, ErrFieldDoesNotExist + } + data, ok := fld.Views[view] + if !ok { + return nil, ErrViewDoesNotExist + } + return data, nil +} + +// CreateView is an in-memory implementation of the Schemator CreateView method. +func (s *inMemSchemator) CreateView(ctx context.Context, index, field, view string, val []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + idx, ok := s.schema[index] + if !ok { + return ErrIndexDoesNotExist + } + fld, ok := idx.Fields[field] + if !ok { + return ErrFieldDoesNotExist + } + // The current logic in pilosa doesn't allow us to return ErrViewExists + // here, so for now we just update the value if the view already exists. + fld.Views[view] = val + return nil +} + +// DeleteView is an in-memory implementation of the Schemator DeleteView method. +func (s *inMemSchemator) DeleteView(ctx context.Context, index, field, view string) error { + s.mu.Lock() + defer s.mu.Unlock() + idx, ok := s.schema[index] + if !ok { + return ErrIndexDoesNotExist + } + fld, ok := idx.Fields[field] + if !ok { + return ErrFieldDoesNotExist + } + delete(fld.Views, view) + return nil +} diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 7226dce7f..f091c77ae 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -1046,7 +1046,7 @@ func (s Serializer) decodeFieldOptions(options *internal.FieldOptions, m *pilosa s.decodeDecimal(options.Max, &m.Max) m.Base = options.Base m.Scale = options.Scale - m.BitDepth = uint(options.BitDepth) + m.BitDepth = uint64(options.BitDepth) m.TimeQuantum = pilosa.TimeQuantum(options.TimeQuantum) m.Keys = options.Keys m.ForeignIndex = options.ForeignIndex diff --git a/etcd/embed.go b/etcd/embed.go index b4d08dc04..09bd82608 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -477,7 +477,7 @@ func (e *Etcd) DeleteNode(ctx context.Context, nodeID string) error { return nil } -func (e *Etcd) Schema(ctx context.Context) (map[string]*disco.Index, error) { +func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) { keys, vals, err := e.getKey(ctx, schemaPrefix) if err != nil { return nil, err @@ -494,7 +494,7 @@ func (e *Etcd) Schema(ctx context.Context) (map[string]*disco.Index, error) { // /index2 // /index2/field1 // - m := make(map[string]*disco.Index) + m := make(disco.Schema) for i, k := range keys { tokens := strings.Split(strings.Trim(k, "/"), "/") // token[0] contains the schemaPrefix diff --git a/executor.go b/executor.go index b1f7d2188..93619c432 100644 --- a/executor.go +++ b/executor.go @@ -4086,7 +4086,7 @@ func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index stri mergeBits(sign, 1<<63, data) // Copy in the significand. - for i := uint(0); i < bsig.BitDepth; i++ { + for i := uint64(0); i < bsig.BitDepth; i++ { bits, err := fragment.row(tx, bsiOffsetBit+uint64(i)) if err != nil { return ExtractedIDMatrix{}, errors.Wrap(err, "loading BSI significand bit from fragment") @@ -5067,7 +5067,7 @@ func (e *executor) executeSet(ctx context.Context, qcx *Qcx, index string, c *pq // Set column on existence field. if ef := idx.existenceField(); ef != nil { // we create tx here, rather than just above, to avoid creating an extra empty shard. - tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Field: ef, Shard: shard}) if err != nil { return false, err } diff --git a/field.go b/field.go index e6ca53afb..34a6aa72e 100644 --- a/field.go +++ b/field.go @@ -752,7 +752,6 @@ func (f *Field) ForeignIndex() string { // openViews opens and initializes the views inside the field. func (f *Field) openViews() error { - view2shards := f.idx.fieldView2shard.getViewsForField(f.name) if view2shards == nil { // no data @@ -844,7 +843,7 @@ func (f *Field) loadMeta() error { f.options.Max = max f.options.Base = pb.Base f.options.Scale = pb.Scale - f.options.BitDepth = uint(pb.BitDepth) + f.options.BitDepth = pb.BitDepth f.options.TimeQuantum = TimeQuantum(pb.TimeQuantum) f.options.Keys = pb.Keys f.options.NoStandardView = pb.NoStandardView @@ -1189,8 +1188,11 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, defer f.mu.Unlock() // Create the view in etcd as the system of record. - if err := f.persistView(context.Background(), cvm); err != nil { - return nil, false, errors.Wrap(err, "persisting view") + // Don't persist views related to the existence field. + if f.name != existenceFieldName { + if err := f.persistView(context.Background(), cvm); err != nil { + return nil, false, errors.Wrap(err, "persisting view") + } } if view := f.viewMap[cvm.View]; view != nil { @@ -1871,9 +1873,9 @@ func (f *Field) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, return err } - var bitDepth uint + var bitDepth uint64 if maxRowID+1 > bsiOffsetBit { - bitDepth = uint(maxRowID + 1 - bsiOffsetBit) + bitDepth = uint64(maxRowID + 1 - bsiOffsetBit) } bsig := f.bsiGroup(f.name) @@ -1915,7 +1917,7 @@ func (p fieldInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name } // FieldOptions represents options to set when initializing a field. type FieldOptions struct { Base int64 `json:"base,omitempty"` - BitDepth uint `json:"bitDepth,omitempty"` + BitDepth uint64 `json:"bitDepth,omitempty"` Min pql.Decimal `json:"min,omitempty"` Max pql.Decimal `json:"max,omitempty"` Scale int64 `json:"scale,omitempty"` @@ -2009,7 +2011,7 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { return json.Marshal(struct { Type string `json:"type"` Base int64 `json:"base"` - BitDepth uint `json:"bitDepth"` + BitDepth uint64 `json:"bitDepth"` Min pql.Decimal `json:"min"` Max pql.Decimal `json:"max"` Keys bool `json:"keys"` @@ -2028,7 +2030,7 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { Type string `json:"type"` Base int64 `json:"base"` Scale int64 `json:"scale"` - BitDepth uint `json:"bitDepth"` + BitDepth uint64 `json:"bitDepth"` Min pql.Decimal `json:"min"` Max pql.Decimal `json:"max"` Keys bool `json:"keys"` @@ -2109,7 +2111,7 @@ type bsiGroup struct { Max int64 `json:"max,omitempty"` Base int64 `json:"base,omitempty"` Scale int64 `json:"scale,omitempty"` - BitDepth uint `json:"bitDepth,omitempty"` + BitDepth uint64 `json:"bitDepth,omitempty"` } // baseValue adjusts the value to align with the range for Field for a certain @@ -2209,12 +2211,12 @@ func isValidCacheType(v string) bool { } // bitDepth returns the number of bits required to store a value. -func bitDepth(v uint64) uint { - return uint(bits.Len64(v)) +func bitDepth(v uint64) uint64 { + return uint64(bits.Len64(v)) } // bitDepthInt64 returns the required bit depth for abs(v). -func bitDepthInt64(v int64) uint { +func bitDepthInt64(v int64) uint64 { if v < 0 { return bitDepth(uint64(-v)) } diff --git a/fragment.go b/fragment.go index 2d6f202b9..3fb39c86c 100644 --- a/fragment.go +++ b/fragment.go @@ -965,7 +965,7 @@ func (f *fragment) bit(tx Tx, rowID, columnID uint64) (bool, error) { } // value uses a column of bits to read a multi-bit value. -func (f *fragment) value(tx Tx, columnID uint64, bitDepth uint) (value int64, exists bool, err error) { +func (f *fragment) value(tx Tx, columnID uint64, bitDepth uint64) (value int64, exists bool, err error) { f.mu.Lock() defer f.mu.Unlock() @@ -977,7 +977,7 @@ func (f *fragment) value(tx Tx, columnID uint64, bitDepth uint) (value int64, ex } // Compute other bits into a value. - for i := uint(0); i < bitDepth; i++ { + for i := uint64(0); i < bitDepth; i++ { if v, err := f.bit(tx, uint64(bsiOffsetBit+i), columnID); err != nil { return 0, false, errors.Wrapf(err, "getting value bit %d", i) } else if v { @@ -996,16 +996,16 @@ func (f *fragment) value(tx Tx, columnID uint64, bitDepth uint) (value int64, ex } // clearValue uses a column of bits to clear a multi-bit value. -func (f *fragment) clearValue(tx Tx, columnID uint64, bitDepth uint, value int64) (changed bool, err error) { +func (f *fragment) clearValue(tx Tx, columnID uint64, bitDepth uint64, value int64) (changed bool, err error) { return f.setValueBase(tx, columnID, bitDepth, value, true) } // setValue uses a column of bits to set a multi-bit value. -func (f *fragment) setValue(tx Tx, columnID uint64, bitDepth uint, value int64) (changed bool, err error) { +func (f *fragment) setValue(tx Tx, columnID uint64, bitDepth uint64, value int64) (changed bool, err error) { return f.setValueBase(tx, columnID, bitDepth, value, false) } -func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value int64, clear bool, toSet, toClear []uint64) ([]uint64, []uint64, error) { +func (f *fragment) positionsForValue(columnID uint64, bitDepth uint64, value int64, clear bool, toSet, toClear []uint64) ([]uint64, []uint64, error) { // Convert value to an unsigned representation. uvalue := uint64(value) if value < 0 { @@ -1030,7 +1030,7 @@ func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value int64 toSet = append(toSet, bit) } - for i := uint(0); i < bitDepth; i++ { + for i := uint64(0); i < bitDepth; i++ { bit, err := f.pos(uint64(bsiOffsetBit+i), columnID) if err != nil { return toSet, toClear, errors.Wrap(err, "getting pos") @@ -1046,7 +1046,7 @@ func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value int64 } // TODO get rid of this and use positionsForValue to generate a single write op, and set that with importPositions. -func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint, value int64, clear bool) (changed bool, err error) { +func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, value int64, clear bool) (changed bool, err error) { f.mu.Lock() defer f.mu.Unlock() @@ -1073,7 +1073,7 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint, value uvalue = uint64(-value) } - for i := uint(0); i < bitDepth; i++ { + for i := uint64(0); i < bitDepth; i++ { if uvalue&(1<= 0; i-- { row, err := f.row(tx, uint64(bsiOffsetBit+i)) if err != nil { @@ -1294,7 +1294,7 @@ func (f *fragment) minUnsigned(tx Tx, filter *Row, bitDepth uint) (min int64, co // max returns the max of a given bsiGroup as well as the number of columns involved. // A bitmap can be passed in to optionally filter the computed columns. -func (f *fragment) max(tx Tx, filter *Row, bitDepth uint) (max int64, count uint64, err error) { +func (f *fragment) max(tx Tx, filter *Row, bitDepth uint64) (max int64, count uint64, err error) { consider, err := f.row(tx, bsiExistsBit) if err != nil { return max, count, err @@ -1323,7 +1323,7 @@ func (f *fragment) max(tx Tx, filter *Row, bitDepth uint) (max int64, count uint } // maxUnsigned the highest value without considering the sign bit. Filter is required. -func (f *fragment) maxUnsigned(tx Tx, filter *Row, bitDepth uint) (max int64, count uint64, err error) { +func (f *fragment) maxUnsigned(tx Tx, filter *Row, bitDepth uint64) (max int64, count uint64, err error) { for i := int(bitDepth - 1); i >= 0; i-- { row, err := f.row(tx, uint64(bsiOffsetBit+i)) if err != nil { @@ -1424,7 +1424,7 @@ func (f *fragment) maxRowID(tx Tx) (_ uint64, err error) { } // rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate. -func (f *fragment) rangeOp(tx Tx, op pql.Token, bitDepth uint, predicate int64) (*Row, error) { +func (f *fragment) rangeOp(tx Tx, op pql.Token, bitDepth uint64, predicate int64) (*Row, error) { switch op { case pql.EQ: return f.rangeEQ(tx, bitDepth, predicate) @@ -1450,7 +1450,7 @@ func absInt64(v int64) uint64 { } } -func (f *fragment) rangeEQ(tx Tx, bitDepth uint, predicate int64) (*Row, error) { +func (f *fragment) rangeEQ(tx Tx, bitDepth uint64, predicate int64) (*Row, error) { // Start with set of columns with values set. b, err := f.row(tx, bsiExistsBit) if err != nil { @@ -1458,7 +1458,7 @@ func (f *fragment) rangeEQ(tx Tx, bitDepth uint, predicate int64) (*Row, error) } upredicate := absInt64(predicate) - if uint(bits.Len64(upredicate)) > bitDepth { + if uint64(bits.Len64(upredicate)) > bitDepth { // Predicate is out of range. return NewRow(), nil } @@ -1492,7 +1492,7 @@ func (f *fragment) rangeEQ(tx Tx, bitDepth uint, predicate int64) (*Row, error) return b, nil } -func (f *fragment) rangeNEQ(tx Tx, bitDepth uint, predicate int64) (*Row, error) { +func (f *fragment) rangeNEQ(tx Tx, bitDepth uint64, predicate int64) (*Row, error) { // Start with set of columns with values set. b, err := f.row(tx, bsiExistsBit) if err != nil { @@ -1511,7 +1511,7 @@ func (f *fragment) rangeNEQ(tx Tx, bitDepth uint, predicate int64) (*Row, error) return b, nil } -func (f *fragment) rangeLT(tx Tx, bitDepth uint, predicate int64, allowEquality bool) (*Row, error) { +func (f *fragment) rangeLT(tx Tx, bitDepth uint64, predicate int64, allowEquality bool) (*Row, error) { if predicate == 1 && !allowEquality { predicate, allowEquality = 0, true } @@ -1557,9 +1557,9 @@ func (f *fragment) rangeLT(tx Tx, bitDepth uint, predicate int64, allowEquality } // rangeLTUnsigned returns all bits LT/LTE the predicate without considering the sign bit. -func (f *fragment) rangeLTUnsigned(tx Tx, filter *Row, bitDepth uint, predicate uint64, allowEquality bool) (*Row, error) { +func (f *fragment) rangeLTUnsigned(tx Tx, filter *Row, bitDepth uint64, predicate uint64, allowEquality bool) (*Row, error) { switch { - case uint(bits.Len64(predicate)) > bitDepth: + case uint64(bits.Len64(predicate)) > bitDepth: fallthrough case predicate == (1< bitDepth: + case !allowEquality && uint64(bits.Len64(predicate)) > bitDepth: // The predicate is bigger than the BSI width, so nothing can be bigger. return NewRow(), nil case allowEquality: @@ -1700,7 +1700,7 @@ func (f *fragment) notNull(tx Tx) (*Row, error) { } // rangeBetween returns bitmaps with a bsiGroup value encoding matching any value between predicateMin and predicateMax. -func (f *fragment) rangeBetween(tx Tx, bitDepth uint, predicateMin, predicateMax int64) (*Row, error) { +func (f *fragment) rangeBetween(tx Tx, bitDepth uint64, predicateMin, predicateMax int64) (*Row, error) { b, err := f.row(tx, bsiExistsBit) if err != nil { return nil, err @@ -1749,7 +1749,7 @@ func (f *fragment) rangeBetween(tx Tx, bitDepth uint, predicateMin, predicateMax } // rangeBetweenUnsigned returns BSI columns for a range of values. Disregards the sign bit. -func (f *fragment) rangeBetweenUnsigned(tx Tx, filter *Row, bitDepth uint, predicateMin, predicateMax uint64) (*Row, error) { +func (f *fragment) rangeBetweenUnsigned(tx Tx, filter *Row, bitDepth uint64, predicateMin, predicateMax uint64) (*Row, error) { switch { case predicateMax > (1<