add mutex field type

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")
 }

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
+}

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 {

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)
 	}

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
 }
 

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)
 	}