Merge branch 'master' into 1521-metalinter-vet

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:
 

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.

docs/glossary.md

@@ -22,7 +22,7 @@ nav = []
 
 <strong id="fragment">Fragment:</strong> A Fragment is the intersection of a [field](#field) and a [shard](#shard) in an [index](#index).
 
-<strong id="field">[Field](../data-model/#field):</strong> 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).
+<strong id="field">[Field](../data-model/#field):</strong> 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).
 
 <strong id="frame">[Frame](../data-model/#field):</strong> Prior to Pilosa 1.0, fields were known as frames.
 

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

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 {

holder.go

@@ -232,7 +232,7 @@ func (h *Holder) Schema() []*IndexInfo {
 func (h *Holder) limitedSchema() []*IndexInfo {
 	var a []*IndexInfo
 	for _, index := range h.Indexes() {
-		di := &IndexInfo{Name: index.Name()}
+		di := &IndexInfo{Name: index.Name(), Options: index.Options()}
 		for _, field := range index.Fields() {
 			fi := &FieldInfo{Name: field.Name(), Options: field.Options()}
 			di.Fields = append(di.Fields, fi)

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

http/handler_internal_test.go

@@ -31,6 +31,7 @@ func TestPostIndexRequestUnmarshalJSON(t *testing.T) {
 		err      string
 	}{
 		{json: `{"options": {}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{}}},
+		{json: `{"options": {"keys": true}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{Keys: true}}},
 		{json: `{"options": 4}`, err: "options is not map[string]interface{}"},
 		{json: `{"option": {}}`, err: "Unknown key: option:map[]"},
 		{json: `{"options": {"badKey": "test"}}`, err: "Unknown key: badKey:test"},

index.go

@@ -95,7 +95,7 @@ func (i *Index) Options() IndexOptions {
 }
 
 func (i *Index) options() IndexOptions {
-	return IndexOptions{}
+	return IndexOptions{Keys: i.keys}
 }
 
 // Open opens and initializes the index.
@@ -405,8 +405,9 @@ func (p indexSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() }
 
 // IndexInfo represents schema information for an index.
 type IndexInfo struct {
-	Name   string       `json:"name"`
-	Fields []*FieldInfo `json:"fields"`
+	Name    string       `json:"name"`
+	Options IndexOptions `json:"options"`
+	Fields  []*FieldInfo `json:"fields"`
 }
 
 type indexInfoSlice []*IndexInfo

translate_test.go

@@ -4,11 +4,13 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math/rand"
 	"os"
 	"reflect"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -356,6 +358,24 @@ func TestTranslateFile_Reader(t *testing.T) {
 	})
 }
 
+func TestPrintTranslateFile(t *testing.T) {
+	// I think this is related to the mmap in s.Open.
+	t.Skip("causes fatal error: fault")
+	f, err := ioutil.TempFile("", "")
+	if err != nil {
+		panic(err)
+	}
+	f.Close()
+
+	s := pilosa.NewTranslateFile()
+	s.Path = f.Name()
+	err = s.Open()
+	if err != nil {
+		t.Fatalf("opening : %v", err)
+	}
+	fmt.Println("blah ", s)
+}
+
 func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
 	// Create a primary store that accepts writes.
 	primary := MustOpenTranslateFile()
@@ -495,6 +515,7 @@ func BenchmarkTranslateFile_TranslateColumnToString(b *testing.B) {
 }
 
 type TranslateFile struct {
+	lock sync.Mutex
 	*pilosa.TranslateFile
 }
 
@@ -510,6 +531,12 @@ func NewTranslateFile() *TranslateFile {
 	return s
 }
 
+func (t *TranslateFile) Reader(ctx context.Context, offset int64) (io.ReadCloser, error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return t.TranslateFile.Reader(ctx, offset)
+}
+
 func MustOpenTranslateFile() *TranslateFile {
 	s := NewTranslateFile()
 	if err := s.Open(); err != nil {
@@ -536,7 +563,9 @@ func (s *TranslateFile) Reopen() error {
 		return err
 	}
 
+	s.lock.Lock()
 	s.TranslateFile = pilosa.NewTranslateFile()
+	s.lock.Unlock()
 	s.Path = prev.Path
 	s.PrimaryTranslateStore = prev.PrimaryTranslateStore
 	if err := s.Open(); err != nil {

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
 }
 

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