diff --git a/field.go b/field.go index d5d5c5948..af18d8666 100644 --- a/field.go +++ b/field.go @@ -47,9 +47,10 @@ const ( // Field types. const ( - FieldTypeSet = "set" - FieldTypeInt = "int" - FieldTypeTime = "time" + FieldTypeSet = "set" + FieldTypeInt = "int" + FieldTypeTime = "time" + FieldTypeMutex = "mutex" ) // Field represents a container for views. @@ -138,6 +139,18 @@ func OptFieldTypeTime(timeQuantum TimeQuantum) FieldOption { } } +func OptFieldTypeMutex(cacheType string, cacheSize uint32) FieldOption { + return func(fo *FieldOptions) error { + if fo.Type != "" { + return errors.Errorf("field type is already set to: %s", fo.Type) + } + fo.Type = FieldTypeMutex + fo.CacheType = cacheType + fo.CacheSize = cacheSize + return nil + } +} + // NewField returns a new instance of field. func NewField(path, index, name string, opts FieldOption) (*Field, error) { err := validateName(name) @@ -400,6 +413,18 @@ func (f *Field) applyOptions(opt FieldOptions) error { f.Close() return errors.Wrap(err, "setting time quantum") } + case FieldTypeMutex: + f.options.Type = FieldTypeMutex + if opt.CacheType != "" { + f.options.CacheType = opt.CacheType + } + if opt.CacheSize != 0 { + f.options.CacheSize = opt.CacheSize + } + f.options.Min = 0 + f.options.Max = 0 + f.options.TimeQuantum = "" + f.options.Keys = opt.Keys default: return errors.New("invalid field type") } @@ -658,8 +683,7 @@ func (f *Field) createViewIfNotExistsBase(name string) (*view, bool, error) { } func (f *Field) newView(path, name string) *view { - view := newView(path, f.index, f.name, name, f.options.CacheSize) - view.cacheType = f.options.CacheType + view := newView(path, f.index, f.name, name, f.options) view.logger = f.logger view.rowAttrStore = f.rowAttrStore view.stats = f.Stats.WithTags(fmt.Sprintf("view:%s", name)) @@ -1191,6 +1215,18 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { o.TimeQuantum, o.Keys, }) + case FieldTypeMutex: + return json.Marshal(struct { + Type string `json:"type"` + CacheType string `json:"cacheType"` + CacheSize uint32 `json:"cacheSize"` + Keys bool `json:"keys"` + }{ + o.Type, + o.CacheType, + o.CacheSize, + o.Keys, + }) } return nil, errors.New("invalid field type") } diff --git a/fragment.go b/fragment.go index 3bc72efdc..063848e5c 100644 --- a/fragment.go +++ b/fragment.go @@ -113,6 +113,10 @@ type fragment struct { // This is set by the parent field unless overridden for testing. RowAttrStore AttrStore + // mutexVector is used for mutex field types. It's checked for an + // existing value (to clear) prior to setting a new value. + mutexVector vector + stats StatsClient } @@ -369,9 +373,29 @@ func (f *fragment) unprotectedRow(rowID uint64, checkRowCache bool, updateRowCac func (f *fragment) setBit(rowID, columnID uint64) (changed bool, err error) { f.mu.Lock() defer f.mu.Unlock() + + // handle mutux field type + if f.mutexVector != nil { + if err := f.handleMutex(rowID, columnID); err != nil { + return changed, errors.Wrap(err, "handling mutex") + } + } + return f.unprotectedSetBit(rowID, columnID) } +// handleMutex will clear an existing row and store the new row +// in the vector. +func (f *fragment) handleMutex(rowID, columnID uint64) error { + if existingRowID, found := f.mutexVector.Get(columnID); found && existingRowID != rowID { + if _, err := f.unprotectedClearBit(existingRowID, columnID); err != nil { + return errors.Wrap(err, "clearing mutex value") + } + } + f.mutexVector.Set(columnID, rowID) + return nil +} + func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err error) { changed = false // Determine the position of the bit in the storage. @@ -1997,3 +2021,40 @@ func byteSlicesEqual(a [][]byte) bool { func pos(rowID, columnID uint64) uint64 { return (rowID * ShardWidth) + (columnID % ShardWidth) } + +// vector stores the mapping of colID to rowID. +// It's used for a mutex field type. +type vector interface { + Get(colID uint64) (uint64, bool) + Set(colID, rowID uint64) +} + +// mapVector implements the vector interface using a map. +type mapVector struct { + mu sync.RWMutex + m map[uint64]uint64 +} + +// newMapVector returns a mapVector. +func newMapVector() *mapVector { + return &mapVector{ + m: make(map[uint64]uint64), + } +} + +// Get returns the rowID associated to the given colID. +// Additionaly, it returns true if a value was found, +// otherwise it returns false. +func (m *mapVector) Get(colID uint64) (uint64, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + rowID, ok := m.m[colID] + return rowID, ok +} + +// Set sets the value for colID to rowID. +func (m *mapVector) Set(colID, rowID uint64) { + m.mu.Lock() + defer m.mu.Unlock() + m.m[colID] = rowID +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a04ea5387..6cd538805 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1143,6 +1143,38 @@ func TestFragment_Snapshot_Run(t *testing.T) { } } +// Ensure a fragment can set mutually exclusive values. +func TestFragment_SetMutex(t *testing.T) { + f := mustOpenMutexFragment("i", "f", viewStandard, 0, "") + defer f.Close() + + var cols []uint64 + + // Set a value on column 100. + if _, err := f.setBit(1, 100); err != nil { + t.Fatal(err) + } + // Verify the value was set. + cols = f.row(1).Columns() + if !reflect.DeepEqual(cols, []uint64{100}) { + t.Fatalf("mutex unexpected columns: %v", cols) + } + + // Set a different value on column 100. + if _, err := f.setBit(2, 100); err != nil { + t.Fatal(err) + } + // Verify that value (row 1) was replaced (by row 2). + cols = f.row(1).Columns() + if !reflect.DeepEqual(cols, []uint64{}) { + t.Fatalf("mutex unexpected columns: %v", cols) + } + cols = f.row(2).Columns() + if !reflect.DeepEqual(cols, []uint64{100}) { + t.Fatalf("mutex unexpected columns: %v", cols) + } +} + func BenchmarkFragment_Snapshot(b *testing.B) { if *FragmentPath == "" { b.Skip("no fragment specified") @@ -1260,6 +1292,13 @@ func mustOpenFragment(index, field, view string, shard uint64, cacheType string) return f } +// mustOpenMutexFragment returns a new instance of Fragment for a mutex field. +func mustOpenMutexFragment(index, field, view string, shard uint64, cacheType string) *fragment { + frag := mustOpenFragment(index, field, view, shard, cacheType) + frag.mutexVector = newMapVector() + return frag +} + // Reopen closes the fragment and reopens it as a new instance. func (f *fragment) reopen() error { if err := f.Close(); err != nil { diff --git a/http/handler.go b/http/handler.go index 3f47297d4..8e0fcb709 100644 --- a/http/handler.go +++ b/http/handler.go @@ -684,6 +684,8 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { fos = append(fos, pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max)) case pilosa.FieldTypeTime: fos = append(fos, pilosa.OptFieldTypeTime(*req.Options.TimeQuantum)) + case pilosa.FieldTypeMutex: + fos = append(fos, pilosa.OptFieldTypeMutex(*req.Options.CacheType, *req.Options.CacheSize)) } if req.Options.Keys != nil { if *req.Options.Keys { @@ -761,6 +763,20 @@ func (o *fieldOptions) validate() error { } else if o.TimeQuantum == nil { return pilosa.NewBadRequestError(errors.New("timeQuantum is required for field type time")) } + case pilosa.FieldTypeMutex: + if o.CacheType == nil { + o.CacheType = &defaultCacheType + } + if o.CacheSize == nil { + o.CacheSize = &defaultCacheSize + } + if o.Min != nil { + return pilosa.NewBadRequestError(errors.New("min does not apply to field type mutex")) + } else if o.Max != nil { + return pilosa.NewBadRequestError(errors.New("max does not apply to field type mutex")) + } else if o.TimeQuantum != nil { + return pilosa.NewBadRequestError(errors.New("timeQuantum does not apply to field type mutex")) + } default: return errors.Errorf("invalid field type: %s", o.Type) } diff --git a/view.go b/view.go index 609664304..046a36879 100644 --- a/view.go +++ b/view.go @@ -41,10 +41,11 @@ type view struct { field string name string + fieldType string + cacheType string cacheSize uint32 // Fragments by shard. - cacheType string // passed in by field fragments map[uint64]*fragment // maxShard maintains this view's max shard in order to @@ -58,15 +59,17 @@ type view struct { } // newView returns a new instance of View. -func newView(path, index, field, name string, cacheSize uint32) *view { +func newView(path, index, field, name string, fieldOptions FieldOptions) *view { return &view{ - path: path, - index: index, - field: field, - name: name, - cacheSize: cacheSize, + path: path, + index: index, + field: field, + name: name, + + fieldType: fieldOptions.Type, + cacheType: fieldOptions.CacheType, + cacheSize: fieldOptions.CacheSize, - cacheType: DefaultCacheType, fragments: make(map[uint64]*fragment), broadcaster: NopBroadcaster, @@ -251,6 +254,9 @@ func (v *view) newFragment(path string, shard uint64) *fragment { frag.CacheSize = v.cacheSize frag.Logger = v.logger frag.stats = v.stats.WithTags(fmt.Sprintf("shard:%d", shard)) + if v.fieldType == FieldTypeMutex { + frag.mutexVector = newMapVector() + } return frag } diff --git a/view_internal_test.go b/view_internal_test.go index 3f00fb38a..b71696128 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -26,7 +26,12 @@ func mustOpenView(index, field, name string) *view { panic(err) } - v := newView(path, index, field, name, DefaultCacheSize) + fo := FieldOptions{ + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, + } + + v := newView(path, index, field, name, fo) if err := v.open(); err != nil { panic(err) }