From 4ddbadcea731cef082c55eef51f46b4a4c9e69c4 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 16 Dec 2020 12:33:40 -0600 Subject: [PATCH] review issues: fix unclearSets (now sliceDifference) and prune/fullPrune unclearSets was completely broken and I have no idea why the test I thought was testing it didn't actually catch that problem. Added unit tests and fixed the logic. Improved/clarified prune and fullPrune, and unexported their names because why export methods on an unexported type. Also improve some comments and rename a variable or two to improve clarity. --- fragment.go | 63 +++++++----- fragment_internal_test.go | 198 +++++++++++++++++++++++++++++++++++++- roaring/filter.go | 81 +++++++++++++--- 3 files changed, 302 insertions(+), 40 deletions(-) diff --git a/fragment.go b/fragment.go index 26fdd9608..1d222f655 100644 --- a/fragment.go +++ b/fragment.go @@ -2339,11 +2339,11 @@ type parallelSlices struct { cols, rows []uint64 } -// Prune eliminates values which have the same column key and are +// prune eliminates values which have the same column key and are // adjacent in the slice. It doesn't handle non-adjacent keys, but -// does report whether it saw any. See FullPrune for what you probably +// does report whether it saw any. See fullPrune for what you probably // want to be using. -func (p *parallelSlices) Prune() (unsorted bool) { +func (p *parallelSlices) prune() (unsorted bool) { l := len(p.cols) if l == 0 { return @@ -2374,19 +2374,22 @@ func (p *parallelSlices) Prune() (unsorted bool) { return unsorted } -// FullPrune trims any adjacent values with identical column keys (and +// fullPrune trims any adjacent values with identical column keys (and // the corresponding row values), and if it notices that anything was unsorted, // does a stable sort by column key and tries that again, ensuring that // there's no items with the same column key. The last entry with a given // column key wins. -func (p *parallelSlices) FullPrune() { +func (p *parallelSlices) fullPrune() { if len(p.cols) == 0 { return } - unsorted := p.Prune() + if len(p.rows) != len(p.cols) { + panic("parallelSlices must have same length for rows and columns") + } + unsorted := p.prune() if unsorted { sort.Stable(p) - _ = p.Prune() + _ = p.prune() } } @@ -2496,25 +2499,33 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64 return err } -// unclearSets removes anything in toClear that was found in toSet -func unclearSets(toSet, toClear []uint64) []uint64 { - cn := 0 - cv := toClear[cn] +// sliceDifference removes everything from original that's found in remove, +// updating the slice in place, and returns the compacted slice. The input +// sets should be sorted. +func sliceDifference(original, remove []uint64) []uint64 { + if len(remove) == 0 { + return original + } + rn := 0 + rv := remove[rn] + on := 0 + ov := uint64(0) n := 0 - for _, sv := range toSet { - for cv < sv { - toClear[n] = cv - n++ - cn++ - if cn >= len(toClear) { - return toClear[:n] + + for on, ov = range original { + for rv < ov { + rn++ + if rn >= len(remove) { + return append(original[:n], original[on:]...) } - cv = toClear[cn] + rv = remove[rn] + } + if rv != ov { + original[n] = ov + n++ } } - copy(toClear[n:], toClear[cn:]) - n += len(toClear[cn:]) - return toClear[:n] + return append(original[:n], original[on+1:]...) } // bulkImportMutex performs a bulk import on a fragment while ensuring @@ -2526,7 +2537,7 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { defer f.mu.Unlock() p := parallelSlices{cols: columnIDs, rows: rowIDs} - p.FullPrune() + p.fullPrune() columnIDs = p.cols rowIDs = p.rows @@ -2567,15 +2578,15 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { rowSet[rowID] = struct{}{} return nil } - existing := roaring.NewBitmapBitmapFilter(columns, callback) - err = tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, existing) + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) + err = tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, findExisting) if err != nil { return errors.Wrap(err, "finding existing positions") } // if we're clearing things, anything being set that is being cleared // should not be cleared if len(toClear) > 0 { - toClear = unclearSets(toSet, toClear) + toClear = sliceDifference(toClear, toSet) } return errors.Wrap(f.importPositions(tx, toSet, toClear, rowSet), "importing positions") } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 6b79be177..a120d0b5c 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -5792,12 +5792,12 @@ func testOneParallelSlice(t *testing.T, p *parallelSlices) { for i, c := range p.cols { seen[c] = p.rows[i] } - unsorted := p.Prune() + unsorted := p.prune() t.Logf("unsorted %t, cols %d, rows %d", unsorted, p.cols, p.rows) if unsorted { sort.Stable(p) } - unsorted = p.Prune() + unsorted = p.prune() if unsorted { t.Fatalf("slice still unsorted after sort") } @@ -5849,3 +5849,197 @@ func TestParallelSlices(t *testing.T) { testOneParallelSlice(t, ¶llelSlices{cols: cols, rows: rows}) }) } + +func testOneParallelSliceFullPrune(t *testing.T, p *parallelSlices) { + // the easy answer + seen := make(map[uint64]uint64, len(p.cols)) + //t.Logf("cols %d, rows %d", p.cols, p.rows) + for i, c := range p.cols { + seen[c] = p.cols[i] + } + //t.Logf("before fullPrune: cols %d, rows %d", p.cols, p.rows) + p.fullPrune() + + //t.Logf("after fullPrune, pruned/sorted cols %d, rows %d", p.cols, p.rows) + if len(p.cols) != len(seen) { + t.Fatalf("expected %d entries, found %d", len(seen), len(p.cols)) + } + for i := range p.cols { + if i == 0 { + continue + } + if p.cols[i] <= p.cols[i-1] { + t.Fatalf("expected p.cols[i=%v]=%v <= p.cols[i-1=%v]=%v", i, p.cols[i], i-1, p.cols[i-1]) + } + } +} + +func TestParallelSlicesFullPrune(t *testing.T) { + cols := make([]uint64, 0) + rows := make([]uint64, 0) + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 1) + rows = make([]uint64, 1) + cols[0] = 3 + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 2) + rows = make([]uint64, 2) + cols[1] = 1 // sorted + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 2) + rows = make([]uint64, 2) + // one duplicate 0 + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 2) + rows = make([]uint64, 2) + cols[0] = 1 // unsorted + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 3) + rows = make([]uint64, 3) + cols[0] = 2 // unsorted + cols[1] = 1 // unsorted + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 3) + rows = make([]uint64, 3) + // three duplicate 0s + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 3) + rows = make([]uint64, 3) + // three duplicate 1s + cols[0] = 1 + cols[1] = 1 + cols[2] = 1 + + t.Run("no_loss", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = make([]uint64, 256) + rows = make([]uint64, 256) + + // ensure at least some overlap by coercing columns into a range + // smaller than number of entries + for i := range cols { + cols[i] = rand.Uint64() & ((uint64(len(cols)) / 2) - 1) + rows[i] = rand.Uint64() & 0xff + } + t.Run("random", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + + cols = cols[:cap(cols)] + rows = rows[:cap(rows)] + // in-order but no overlap + col := uint64(0) + for i := range cols { + cols[i] = col + col = col + (rand.Uint64() & 3) + 1 + rows[i] = rand.Uint64() & 0xff + } + t.Run("ordered", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) + cols = cols[:cap(cols)] + rows = rows[:cap(rows)] + // in-order with + col = uint64(0) + for i := range cols { + cols[i] = col + col = col + (rand.Uint64() & 3) + rows[i] = rand.Uint64() & 0xff + } + t.Run("orderlapping", func(t *testing.T) { + testOneParallelSliceFullPrune(t, ¶llelSlices{cols: cols, rows: rows}) + }) +} + +func compareSlices(tb testing.TB, name string, s1, s2 []uint64) { + if len(s1) != len(s2) { + tb.Fatalf("slice length mismatch %q: expected %d items %d, got %d items %d", + name, len(s1), s1, len(s2), s2) + } + for i, v := range s1 { + if s2[i] != v { + tb.Fatalf("row mismatch %q: expected item %d to be %d, got %d", + name, i, s1[i], s2[i]) + } + } +} + +type sliceDifferenceTestCase struct { + original, remove, expected []uint64 +} + +func TestSliceDifference(t *testing.T) { + testCases := map[string]sliceDifferenceTestCase{ + "noOverlap": { + original: []uint64{1, 2, 3}, + remove: []uint64{0, 5}, + expected: []uint64{1, 2, 3}, + }, + "before": { + original: []uint64{3, 5, 7}, + remove: []uint64{0, 6}, + expected: []uint64{3, 5, 7}, + }, + "after": { + original: []uint64{3, 5, 7}, + remove: []uint64{8, 10}, + expected: []uint64{3, 5, 7}, + }, + "all": { + original: []uint64{3, 5, 7}, + remove: []uint64{3, 5, 7}, + expected: []uint64{}, + }, + "first": { + original: []uint64{3, 5, 7}, + remove: []uint64{3}, + expected: []uint64{5, 7}, + }, + "last": { + original: []uint64{3, 5, 7}, + remove: []uint64{7}, + expected: []uint64{3, 5}, + }, + "middle": { + original: []uint64{3, 5, 7}, + remove: []uint64{5}, + expected: []uint64{3, 7}, + }, + } + var scratch []uint64 + for name, tc := range testCases { + scratch = append(scratch[:0], tc.original...) + result := sliceDifference(scratch, tc.remove) + compareSlices(t, name, tc.expected, result) + } +} diff --git a/roaring/filter.go b/roaring/filter.go index fc0eeedd4..61f9b3840 100644 --- a/roaring/filter.go +++ b/roaring/filter.go @@ -31,10 +31,10 @@ import ( // means it needs access to the shard width stuff, which roaring otherwise // studiously avoids knowing about. const ( - rowExponent = (shardwidth.Exponent - 16) - rowWidth = 1 << rowExponent // containers per row - keyMask = (rowWidth - 1) // a mask for offset within the row - rowMask = ^FilterKey(keyMask) // a mask for the row bits, without converting them to a row ID + rowExponent = (shardwidth.Exponent - 16) // for instance, 20-16 = 4 + rowWidth = 1 << rowExponent // containers per row, for instance 1<<4 = 16 + keyMask = (rowWidth - 1) // a mask for offset within the row + rowMask = ^FilterKey(keyMask) // a mask for the row bits, without converting them to a row ID ) type FilterKey uint64 @@ -47,8 +47,8 @@ type FilterKey uint64 // easier to write. It can also report an error, which indicates that the // entire operation should be stopped with that error. type FilterResult struct { - YesKey FilterKey // The lowest container key this container is known NOT to match. - NoKey FilterKey // The highest container key after that that this filter is known to not match. + YesKey FilterKey // The lowest container key this filter is known NOT to match. + NoKey FilterKey // The highest container key after YesKey that this filter is known to not match. Err error // An error which should terminate processing. } @@ -73,7 +73,7 @@ func (f FilterKey) MatchReject(y, n FilterKey) FilterResult { } func (f FilterKey) MatchOne() FilterResult { - return FilterResult{YesKey: f + 1} + return FilterResult{YesKey: f + 1, NoKey: f + 1} } // NeedData() is only really meaningful for ConsiderKey, and indicates @@ -227,6 +227,10 @@ func (f *BitmapColumnFilter) ConsiderData(key FilterKey, data *Container) Filter return key.RejectUntilOffset(uint64(f.key)) } +func NewBitmapColumnFilter(col uint64) BitmapFilter { + return &BitmapColumnFilter{key: uint16((col >> 16) & keyMask), offset: uint16(col & 0xFFFF)} +} + // BitmapRowsFilter is a BitmapFilter which checks for containers that are // in any of a provided list of rows. The row list should be sorted. type BitmapRowsFilter struct { @@ -583,16 +587,12 @@ func (b *BitmapRowFilterMultiFilter) ConsiderData(key FilterKey, data *Container return b.SetResult(key, key.MatchReject(lowestYes, lowestYesNo)) } -func NewBitmapColumnFilter(col uint64) BitmapFilter { - return &BitmapColumnFilter{key: uint16((col >> 16) & keyMask), offset: uint16(col & 0xFFFF)} -} - // BitmapBitmap filter builds a list of positions in the bitmap which // match those in a provided bitmap. It is shard-agnostic; no matter what // offsets the input bitmap's containers have, it matches them against // corresponding keys. type BitmapBitmapFilter struct { - filter *Bitmap // We don't use this, but in ludicrous edge cases it might be holding a generation we need. + filter *Bitmap // We don't use this while iterating, but in ludicrous edge cases it might be holding a generation we need. containers []*Container nextOffsets []uint64 callback func(uint64) error @@ -628,6 +628,11 @@ func (b *BitmapBitmapFilter) ConsiderData(key FilterKey, data *Container) Filter // within a bitmap which are set, and which have positions corresponding to // the specified columns. It calls the provided callback function on // each value it finds, terminating early if that returns an error. +// +// The input filter is assumed to represent one "row" of a shard's data, +// which is to say, a range of up to rowWidth consecutive containers starting +// at some multiple of rowWidth. We coerce that to the 0..rowWidth range +// because offset-within-row is what we care about. func NewBitmapBitmapFilter(filter *Bitmap, callback func(uint64) error) *BitmapBitmapFilter { b := &BitmapBitmapFilter{ filter: filter, @@ -640,6 +645,8 @@ func NewBitmapBitmapFilter(filter *Bitmap, callback func(uint64) error) *BitmapB count := 0 for iter.Next() { k, v := iter.Value() + // Coerce container key into the 0-rowWidth range we'll be + // using to compare against containers within each row. k = k & keyMask b.containers[k] = v last = k @@ -691,6 +698,56 @@ func NewBitmapRowFilter(callback func(uint64) error, filters ...BitmapFilter) Bi return NewBitmapRowFilterMultiFilter(callback, filters...) } +// BitmapRangeFilter limits filter operations to a specified range, and +// performs key or data callbacks. +// +// On seeing a key in its range: +// If the key callback is present, and returns true, match the key. +// Otherwise, if a data callback is present, request the data, and in the +// data handler, call the data callback, then match the single key. +// If neither is present, match the entire range at once. +type BitmapRangeFilter struct { + min, max FilterKey + kcb func(FilterKey, int32) (bool, error) + dcb func(FilterKey, *Container) error +} + +var _ BitmapFilter = &BitmapRangeFilter{} + +func (b *BitmapRangeFilter) ConsiderKey(key FilterKey, n int32) FilterResult { + if key >= b.max { + return key.Done() + } + if key >= b.min { + if b.kcb != nil { + match, err := b.kcb(key, n) + if err != nil { + return key.Fail(err) + } + if match { + return key.MatchOne() + } + } + if b.dcb != nil { + return key.NeedData() + } + return key.MatchReject(b.max, ^FilterKey(0)) + } + return key.RejectUntil(b.min) +} + +func (b *BitmapRangeFilter) ConsiderData(key FilterKey, data *Container) FilterResult { + err := b.dcb(key, data) + if err != nil { + return key.Fail(err) + } + return key.MatchOne() +} + +func NewBitmapRangeFilter(min, max FilterKey, keyCallback func(FilterKey, int32) (bool, error), dataCallback func(FilterKey, *Container) error) *BitmapRangeFilter { + return &BitmapRangeFilter{min: min, max: max, kcb: keyCallback, dcb: dataCallback} +} + // ApplyFilterToIterator is a simplistic implementation that applies a bitmap // filter to a ContainerIterator, returning an error if it encounters an error. //