Merge branch 'master' into skip-prerelease-if-pr

api.go

@@ -732,7 +732,7 @@ func (api *API) Import(_ context.Context, req *ImportRequest) error {
 		if ts == 0 {
 			continue
 		}
-		t := time.Unix(0, ts)
+		t := time.Unix(0, ts).UTC()
 		timestamps[i] = &t
 	}
 

fragment.go

@@ -1330,16 +1330,29 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64) error {
 		return fmt.Errorf("mismatch of row/column len: %d != %d", len(rowIDs), len(columnIDs))
 	}
 
+	if f.mutexVector != nil {
+		return f.bulkImportMutex(rowIDs, columnIDs)
+	}
+	return f.bulkImportStandard(rowIDs, columnIDs)
+}
+
+// bulkImportStandard performs a bulk import on a standard fragment.
+func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64) error {
 	// Create a temporary bitmap which will be populated by rowIDs and columnIDs
 	// and then merged into the existing fragment's bitmap.
 	localBitmap := roaring.NewBitmap()
+
 	// Disconnect op writer so we don't append updates.
 	localBitmap.OpWriter = nil
 
-	// Process every bit.
-	// If an error occurs then reopen the storage.
-	lastID := uint64(0)
+	// rowSet maintains the set of rowIDs present in this import.
+	// It allows the cache to be updated once per row, instead of once
+	// per bit.
 	rowSet := make(map[uint64]struct{})
+	lastRowID := uint64(0)
+
+	// Process every bit by writing to a local bitmap,
+	// to be merged with fragment storage next.
 	for i := range rowIDs {
 		rowID, columnID := rowIDs[i], columnIDs[i]
 
@@ -1349,18 +1362,18 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64) error {
 			return err
 		}
 
-		// Write to storage.
+		// Write to local storage.
 		_, err = localBitmap.Add(pos)
 		if err != nil {
 			return err
 		}
+
 		// Reduce the StatsD rate for high volume stats
 		f.stats.Count("ImportBit", 1, 0.0001)
-		// import optimization to avoid linear foreach calls
-		// slight risk of concurrent cache counter being off but
-		// no real danger
-		if i == 0 || rowID != lastID {
-			lastID = rowID
+
+		// Add row to rowSet.
+		if i == 0 || rowID != lastRowID {
+			lastRowID = rowID
 			rowSet[rowID] = struct{}{}
 		}
 
@@ -1389,6 +1402,96 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64) error {
 	return unprotectedWriteToFragment(f, results)
 }
 
+// bulkImportMutex performs a bulk import on a fragment while ensuring
+// mutex restrictions. Because the mutex requirements must be checked
+// against storage, this method must acquire a write lock on the fragment
+// during the entire process, and it handles every bit independently.
+func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	// Disconnect op writer so we don't append updates.
+	f.storage.OpWriter = nil
+
+	// If an error occurs then reopen the storage.
+	if err := func() error {
+		// rowSet maintains the set of rowIDs present in this import.
+		// It allows the cache to be updated once per row, instead of once
+		// per bit.
+		rowSet := make(map[uint64]struct{})
+		lastRowID := uint64(0)
+
+		// Process every bit.
+		for i := range rowIDs {
+			rowID, columnID := rowIDs[i], columnIDs[i]
+
+			// Handle mutex vector (i.e. clear an existing row).
+			if existingRowID, found := f.mutexVector.Get(columnID); found && existingRowID != rowID {
+				// Determine the position of the bit in the storage.
+				pos, err := f.pos(existingRowID, columnID)
+				if err != nil {
+					return err
+				}
+
+				// Clear storage.
+				_, err = f.storage.Remove(pos)
+				if err != nil {
+					return err
+				}
+
+				rowSet[existingRowID] = struct{}{}
+			}
+
+			// Determine the position of the bit in the storage.
+			pos, err := f.pos(rowID, columnID)
+			if err != nil {
+				return err
+			}
+
+			// Write to storage.
+			_, err = f.storage.Add(pos)
+			if err != nil {
+				return err
+			}
+
+			// Reduce the StatsD rate for high volume stats
+			f.stats.Count("ImportBit", 1, 0.0001)
+
+			// Add row to rowSet.
+			if i == 0 || rowID != lastRowID {
+				lastRowID = rowID
+				rowSet[rowID] = struct{}{}
+			}
+
+			// Invalidate block checksum.
+			delete(f.checksums, int(rowID/HashBlockSize))
+		}
+
+		// Update cache counts for all rows.
+		for rowID := range rowSet {
+			// Import should ALWAYS have row() load a new bm from fragment.storage
+			// because the row that's in rowCache hasn't been updated with
+			// this import's data.
+			f.cache.BulkAdd(rowID, f.unprotectedRow(rowID).Count())
+		}
+
+		f.cache.Invalidate()
+
+		return nil
+	}(); err != nil {
+		_ = f.closeStorage()
+		_ = f.openStorage()
+		return err
+	}
+
+	// Write the storage to disk and reload.
+	if err := f.snapshot(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // importValue bulk imports a set of range-encoded values.
 func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint) error {
 	f.mu.Lock()

fragment_internal_test.go

@@ -1178,6 +1178,76 @@ func TestFragment_SetMutex(t *testing.T) {
 	}
 }
 
+// Ensure a fragment can import mutually exclusive values.
+func TestFragment_ImportMutex(t *testing.T) {
+	tests := []struct {
+		rowIDs []uint64
+		colIDs []uint64
+		exp    map[uint64][]uint64
+	}{
+		{
+			[]uint64{1, 1, 1, 1},
+			[]uint64{0, 1, 2, 3},
+			map[uint64][]uint64{
+				1: {0, 1, 2, 3},
+			},
+		},
+		{
+			[]uint64{1, 1, 1, 1, 2, 2, 2, 2},
+			[]uint64{0, 1, 2, 3, 0, 1, 2, 3},
+			map[uint64][]uint64{
+				1: {},
+				2: {0, 1, 2, 3},
+			},
+		},
+		{
+			[]uint64{1, 1, 1, 1, 2},
+			[]uint64{0, 1, 2, 3, 1},
+			map[uint64][]uint64{
+				1: {0, 2, 3},
+				2: {1},
+			},
+		},
+		{
+			[]uint64{1, 1, 1, 1, 2, 2, 1},
+			[]uint64{0, 1, 2, 3, 1, 8, 1},
+			map[uint64][]uint64{
+				1: {0, 1, 2, 3},
+				2: {8},
+			},
+		},
+		{
+			[]uint64{1, 2, 3},
+			[]uint64{8, 8, 8},
+			map[uint64][]uint64{
+				1: {},
+				2: {},
+				3: {8},
+			},
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("importmutex%d", i), func(t *testing.T) {
+			f := mustOpenMutexFragment("i", "f", viewStandard, 0, "")
+			defer f.Close()
+
+			err := f.bulkImport(test.rowIDs, test.colIDs)
+			if err != nil {
+				t.Fatalf("bulk importing ids: %v", err)
+			}
+
+			// Check for expected results.
+			for k, v := range test.exp {
+				cols := f.row(k).Columns()
+				if !reflect.DeepEqual(cols, v) {
+					t.Fatalf("expected: %v, but got: %v", v, cols)
+				}
+			}
+		})
+	}
+}
+
 func BenchmarkFragment_Snapshot(b *testing.B) {
 	if *FragmentPath == "" {
 		b.Skip("no fragment specified")

roaring/roaring.go

@@ -46,7 +46,7 @@ const (
 	// at the beginning of every serialized run container.
 	runCountHeaderSize = 2
 
-	// interval32Size is the size of a single run in a container.runs.
+	// interval16Size is the size of a single run in a container.runs.
 	interval16Size = 4
 
 	// bitmapN is the number of values in a container.bitmap.
@@ -1698,7 +1698,7 @@ func (c *Container) arrayWriteTo(w io.Writer) (n int64, err error) {
 	//	assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v)
 	//}
 
-	// Write sizeof(uint32) * cardinality bytes.
+	// Write sizeof(uint16) * cardinality bytes.
 	nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.array[0]))[:2*c.n])
 	return int64(nn), err
 }

server/server_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"math/rand"
 	"reflect"
 	"sort"
@@ -564,4 +565,57 @@ func TestRemoveNodeAfterItDies(t *testing.T) {
 	}
 }
 
+// Ensure program imports timestamps as UTC.
+func TestMain_ImportTimestamp(t *testing.T) {
+	m := test.MustRunCommand()
+	defer m.Close()
+
+	indexName := "i"
+	fieldName := "f"
+
+	// Create index.
+	if _, err := m.API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create field.
+	if _, err := m.API.CreateField(context.Background(), indexName, fieldName, pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMD"))); err != nil {
+		t.Fatal(err)
+	}
+
+	data := pilosa.ImportRequest{
+		Index:      indexName,
+		Field:      fieldName,
+		Shard:      0,
+		RowIDs:     []uint64{1, 2},
+		ColumnIDs:  []uint64{1, 2},
+		Timestamps: []int64{1514764800000000000, 1577833200000000000}, // 2018-01-01T00:00, 2019-12-31T23:00
+	}
+
+	// Import data.
+	if err := m.API.Import(context.Background(), &data); err != nil {
+		t.Fatal(err)
+	}
+
+	// Ensure the correct views were created.
+	dir := fmt.Sprintf("%s/%s/%s/views", m.Config.DataDir, indexName, fieldName)
+	files, err := ioutil.ReadDir(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	exp := []string{
+		"standard", "standard_2018", "standard_201801", "standard_20180101",
+		"standard_2019", "standard_201912", "standard_20191231",
+	}
+	got := []string{}
+	for _, f := range files {
+		got = append(got, f.Name())
+	}
+
+	if !reflect.DeepEqual(got, exp) {
+		t.Fatalf("expected %v, but got %v", exp, got)
+	}
+}
+
 // TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address.