From fcdc3b742708b66101a0d604a50e9c997f49033b Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Wed, 19 Sep 2018 20:54:06 +0300 Subject: [PATCH 1/4] trivial comment fix --- roaring/roaring.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index c3f2cc3db..c17c297f5 100644 --- a/roaring/roaring.go +++ b/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 } From e7481f4fd2e6446aba020933698a4bc3b4f96e92 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Tue, 18 Sep 2018 16:38:24 -0500 Subject: [PATCH 2/4] treat import timestamps as UTC --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index 7736b12e5..a8783f502 100644 --- a/api.go +++ b/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 } From b86478c613409ab8258b5bd2d364f61d9acd7b09 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Wed, 19 Sep 2018 16:14:34 -0500 Subject: [PATCH 3/4] test to ensure views match UTC time --- server/server_test.go | 54 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/server/server_test.go b/server/server_test.go index ab860dc1b..67440c21e 100644 --- a/server/server_test.go +++ b/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. From 995a24d0afbe556dbfd00a316c695f5f07c88c11 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Wed, 19 Sep 2018 15:05:16 -0500 Subject: [PATCH 4/4] ensure mutex imports unset previous columns --- fragment.go | 121 +++++++++++++++++++++++++++++++++++--- fragment_internal_test.go | 70 ++++++++++++++++++++++ 2 files changed, 182 insertions(+), 9 deletions(-) diff --git a/fragment.go b/fragment.go index 7f651b135..7e3cd26d9 100644 --- a/fragment.go +++ b/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() diff --git a/fragment_internal_test.go b/fragment_internal_test.go index dd2d00b41..e0865c606 100644 --- a/fragment_internal_test.go +++ b/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")