diff --git a/docs/api-reference.md b/docs/api-reference.md index 8975655d9..890b37941 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -103,13 +103,16 @@ Creates a field in the given index with the given name. The request payload is in JSON, and may contain the `options` field. The `options` field is a JSON object which must contain a `type` along with the corresponding configuration options. * `set` - * `cacheType` (string): [ranked](../data-model/#ranked) or [LRU](../data-model/#lru) caching on this field. Default is `lru`. - * `cacheSize` (int): Number of rows to keep in the cache. Default 50,000. + * `cacheType` (string): [ranked](../data-model/#ranked) or [LRU](../data-model/#lru) caching on this field. Default is `ranked`. + * `cacheSize` (int): Number of rows to keep in the cache. Default is 50,000. * `int` * `min` (int): Minimum integer value allowed for the field. * `max` (int): Maximum integer value allowed for the field. * `time` * `timeQuantum` (string): [Time Quantum](../data-model/#time-quantum) for this field. +* `mutex` + * `cacheType` (string): [ranked](../data-model/#ranked) or [LRU](../data-model/#lru) caching on this field. Default is `ranked`. + * `cacheSize` (int): Number of rows to keep in the cache. Default is 50,000. The following example creates an `int` field called "quantity" capable of storing values from -1000 to 2000: diff --git a/docs/data-model.md b/docs/data-model.md index 4609032fc..ba7c12024 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -117,7 +117,7 @@ Query operations run in parallel, and they are evenly distributed across a clust ### Field Type -Upon creation, fields are configured to be of a certain type. Pilosa supports three field types: `set`, `int`, and `time`. +Upon creation, fields are configured to be of a certain type. Pilosa supports the following field types: `set`, `int`, `time`, and `mutex`. #### Set @@ -188,3 +188,7 @@ Set(3, A=8, 2017-05-19T00:00) ![time quantum field diagram](/img/docs/field-time-quantum.svg) *Time quantum fueld diagram* + +#### Mutex + +Mutex fields are similar to `set` fields, with the distinction of requiring the row value for each column to be mutually exclusive. In other words, each column can only have a single value for the field. If the field value for a column is updated on a `mutex` field, then the previous field value for that column will be cleared. This field type is like a field in an RDBMS table where every record contains a single value for a particular field. diff --git a/docs/glossary.md b/docs/glossary.md index 550b9017b..c39a0bf6f 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -22,7 +22,7 @@ nav = [] Fragment: A Fragment is the intersection of a [field](#field) and a [shard](#shard) in an [index](#index). -[Field](../data-model/#field): Fields are used to group [rows](#row) into different categories. Row IDs are namespaced by field such that the same row ID in a different field refers to a different row. For [ranked](#topn) fields, rows are kept in sorted order within the field. Fields are one of three types: set, [int](#bsi), and time. For more information, see [data model](../data-model/) and [Creating fields](../api-reference/#create-field). +[Field](../data-model/#field): Fields are used to group [rows](#row) into different categories. Row IDs are namespaced by field such that the same row ID in a different field refers to a different row. For [ranked](#topn) fields, rows are kept in sorted order within the field. Fields are one of four types: set, [int](#bsi), time, and mutex. For more information, see [data model](../data-model/) and [Creating fields](../api-reference/#create-field). [Frame](../data-model/#field): Prior to Pilosa 1.0, fields were known as frames. 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..36fab5fa3 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,37 @@ 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) +} + +// rowsVector implements the vector interface by looking +// at row data as needed. +type rowsVector struct { + f *fragment +} + +// newRowsVector returns a rowsVector for a given fragment. +func newRowsVector(f *fragment) *rowsVector { + return &rowsVector{ + f: f, + } +} + +// Get returns the rowID associated to the given colID. +// Additionally, it returns true if a value was found, +// otherwise it returns false. +func (v *rowsVector) Get(colID uint64) (uint64, bool) { + rows := v.f.rowsForColumn(colID) + if len(rows) == 1 { + return rows[0], true + } + return 0, false +} + +// Set is not used for rowsVector. +func (v *rowsVector) Set(colID, rowID uint64) {} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a04ea5387..c85b7e88c 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 = newRowsVector(frag) + 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..e53ac8970 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 = newRowsVector(frag) + } 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) }