From 918644820bc6f6ad886111f4650d2a4b27f6eed2 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 7 Apr 2021 13:57:43 -0500 Subject: [PATCH 1/6] added pql delete function --- delete_test.go | 231 ++++++++++++++++++++++++++++++++++++++++++++++++ executor.go | 124 ++++++++++++++++++++++++++ pql/ast.go | 1 + rbf/rbf.go | 3 +- test/cluster.go | 45 +++++++--- 5 files changed, 391 insertions(+), 13 deletions(-) create mode 100644 delete_test.go diff --git a/delete_test.go b/delete_test.go new file mode 100644 index 000000000..04d50bc6e --- /dev/null +++ b/delete_test.go @@ -0,0 +1,231 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa_test + +import ( + "context" + "fmt" + "math" + "testing" + "time" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/test" + "github.com/stretchr/testify/require" +) + +func TestExecutor_DeleteRecords(t *testing.T) { + indexName := "i" + setup := func(t *testing.T, r *require.Assertions, c *test.Cluster) { + t.Helper() + c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "setfield") + c.ImportBits(t, indexName, "setfield", [][2]uint64{ + {0, 0}, + {0, 1}, + {0, ShardWidth + 2}, + {10, 2}, + {10, ShardWidth}, + {10, 2 * ShardWidth}, + {10, ShardWidth + 1}, + {20, ShardWidth}, + }) + c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "bsi", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)) + c.ImportIntID(t, indexName, "bsi", []test.IntID{ + {ID: 0, Val: 4}, + {ID: 2, Val: 8}, + }) + c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "timefield", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"))) + c.ImportBitsWithTimestamp(t, indexName, "timefield", [][2]uint64{ + {0, 0}, + {0, 1}, + {0, 1}, + {0, 1}, + {0, 1}, + }, []int64{ + time.Date(2020, time.January, 2, 15, 45, 0, 0, time.UTC).Unix(), + time.Date(2019, time.January, 2, 16, 45, 0, 0, time.UTC).Unix(), + time.Date(2019, time.January, 2, 16, 45, 0, 0, time.UTC).Unix(), + time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix(), + time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix(), + }) + + } + setupKeys := func(t *testing.T, r *require.Assertions, c *test.Cluster) { + t.Helper() + c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "timefield", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"))) + c.ImportTimeQuantumKey(t, indexName, "timefield", []test.TimeQuantumKey{ + {RowKey: "fish", ColKey: "one", Ts: time.Date(2019, time.January, 2, 17, 45, 0, 0, time.UTC).Unix()}, + {RowKey: "fish", ColKey: "one", Ts: time.Date(2020, time.January, 2, 17, 45, 0, 0, time.UTC).Unix()}, + {RowKey: "fish", ColKey: "two", Ts: time.Date(2019, time.January, 3, 17, 45, 0, 0, time.UTC).Unix()}, + }) + c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "keystuff") + c.ImportIDKey(t, indexName, "keystuff", []test.KeyID{ + {ID: 1, Key: "A"}, + {ID: 2, Key: "B"}, + {ID: 3, Key: "C"}, + {ID: 4, Key: "D"}, + }) + + } + setupOverlap := func(t *testing.T, r *require.Assertions, c *test.Cluster) { + t.Helper() + c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, "setfield") + c.ImportBits(t, indexName, "setfield", [][2]uint64{ + {0, 0}, + {0, 1}, + {1, 1}, + {2, 1}, + {3, 1}, + {0, ShardWidth}, + {2, ShardWidth}, + {4, ShardWidth}, + {6, ShardWidth}, + }) + } + tearDown := func(t *testing.T, require *require.Assertions, c *test.Cluster) { + t.Helper() + api := c.GetPrimary().API + err := api.DeleteIndex(context.Background(), indexName) + require.NoErrorf(err, "DeleteIndex %v", indexName) + } + require := require.New(t) + t.Run("DeleteRecords", func(t *testing.T) { + c := test.MustNewCluster(t, 1) + for _, n := range c.Nodes { + n.Config.Cluster.ReplicaN = 1 + } + err := c.Start() + require.NoError(err, "Start cluster DeleteRecords") + defer c.Close() + + t.Run("Delete", func(t *testing.T) { + setup(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Extract(All())`) + m := resp.Results[0].(pilosa.ExtractedTable) + before := convert(m.Columns) + require.Equal([]uint64{0, 1, 2, ShardWidth, ShardWidth + 1, ShardWidth + 2, 2 * ShardWidth}, before, "these records are expected") + resp = c.Query(t, indexName, fmt.Sprintf(`Delete(ConstRow(columns=[1,2,3,%v]))`, ShardWidth+1)) + require.NotNil(resp, "Response should not be nil") + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Extract(All())`) + + //Note none of the removed records should remain + m = resp.Results[0].(pilosa.ExtractedTable) + after := convert(m.Columns) + require.Equal([]uint64{0, ShardWidth, ShardWidth + 2, 2 * ShardWidth}, after, "these records should be remaining") + }) + t.Run("DeleteKey", func(t *testing.T) { + setupKeys(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Extract(All())`) + m := resp.Results[0].(pilosa.ExtractedTable) + before := convertKey(m.Columns) + require.Equal([]string{"one", "A", "B", "C", "D", "two"}, before, "these keyed records before") + resp = c.Query(t, indexName, `Delete(ConstRow(columns=["A","one"]))`) + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Extract(All())`) + m = resp.Results[0].(pilosa.ExtractedTable) + after := convertKey(m.Columns) + require.Equal([]string{"B", "C", "D", "two"}, after, "these keyed records after delete") + }) + t.Run("Delete Row", func(t *testing.T) { + setup(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Delete(Row(setfield=20))`) + require.NotNil(resp, "Response should not be nil") + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Extract(All())`) + + //Note none of the removed records should remain + m := resp.Results[0].(pilosa.ExtractedTable) + after := convert(m.Columns) + require.Equal([]uint64{0, 1, 2, ShardWidth + 1, ShardWidth + 2, 2 * ShardWidth}, after, "these records are expected") + }) + t.Run("Delete Not Row", func(t *testing.T) { + setup(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Delete(Not(Row(setfield=20)))`) + require.NotNil(resp, "Response should not be nil") + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Extract(All())`) + + //Note none of the removed records should remain + m := resp.Results[0].(pilosa.ExtractedTable) + after := convert(m.Columns) + require.Equal([]uint64{ShardWidth}, after, "these records are expected") + }) + t.Run("Delete All", func(t *testing.T) { + setup(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Count(All())`) + //Note none of the removed records should remain + before := resp.Results[0].(uint64) + require.Equal(uint64(7), before, "these records are expected") + + resp = c.Query(t, indexName, `Delete(All())`) + require.NotNil(resp, "Response should not be nil") + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Count(All())`) + //Note none of the removed records should remain + after := resp.Results[0].(uint64) + require.Equal(uint64(0), after, "these records are expected") + }) + t.Run("DeleteOverlap", func(t *testing.T) { + setupOverlap(t, require, c) + defer tearDown(t, require, c) + resp := c.Query(t, indexName, `Extract(All())`) + m := resp.Results[0].(pilosa.ExtractedTable) + before := convert(m.Columns) + require.Equal([]uint64{0, 1, ShardWidth}, before, "these records are expected") + resp = c.Query(t, indexName, fmt.Sprintf(`Delete(ConstRow(columns=[%v]))`, ShardWidth)) + require.NotNil(resp, "Response should not be nil") + require.NotEmpty(resp.Results) + require.Equal(true, resp.Results[0], "Change should have happened") + + resp = c.Query(t, indexName, `Extract(All())`) + + //Note none of the removed records should remain + m = resp.Results[0].(pilosa.ExtractedTable) + after := convert(m.Columns) + require.Equal([]uint64{0, 1}, after, "these records should be remaining") + }) + }) + +} +func convert(before []pilosa.ExtractedTableColumn) []uint64 { + result := make([]uint64, 0) + for _, i := range before { + result = append(result, i.Column.ID) + } + return result +} +func convertKey(before []pilosa.ExtractedTableColumn) []string { + result := make([]string, 0) + for _, i := range before { + result = append(result, i.Column.Key) + } + return result +} diff --git a/executor.go b/executor.go index 028a3c74b..3685410f3 100644 --- a/executor.go +++ b/executor.go @@ -840,6 +840,10 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p case "Percentile": res, err := e.executePercentile(ctx, qcx, index, c, shards, opt) return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards)) + case "Delete": + statFn() //TODO(twg) need this? + res, err := e.executeDeleteRecords(ctx, qcx, index, c, shards, opt) + return res, errors.Wrapf(err, "executeDelete %v", shardSlice(shards)) default: // e.g. "Row", "Union", "Intersect" or anything that returns a bitmap. statFn() res, err := e.executeBitmapCall(ctx, qcx, index, c, shards, opt) @@ -8305,3 +8309,123 @@ func decimalToInt64(dec pql.Decimal, opt FieldOptions) int64 { return 0 } + +// executeDeleteRecords executes a delete() call. +func (e *executor) executeDeleteRecords(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (bool, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeDelete") + defer span.Finish() + + if len(c.Children) == 0 { + return false, errors.New("Delete() requires an input bitmap") + } else if len(c.Children) > 1 { + return false, errors.New("Delete() only accepts a single bitmap input") + } + + // Execute calls in bulk on each remote node and merge. + mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) { + return e.executeDeleteRecordFromShard(ctx, qcx, index, c, shard) + } + + // Merge returned results at coordinating node. + reduceFn := func(ctx context.Context, prev, v interface{}) interface{} { + other, _ := prev.(bool) + return other || v.(bool) + } + + result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) + if err != nil { + return false, err + } + n, _ := result.(bool) + + return n, nil +} + +func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (bool, error) { + + span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeDeleteRecordFromShard") + defer span.Finish() + //need to build the bitmap in the call + child := c.Children[0] + row, err := e.executeBitmapCallShard(ctx, qcx, index, child, shard) + if err != nil { + return false, err + } + if len(row.segments) == 0 { //nothing to remove + return false, nil + } + + // Fetch index. + idx := e.Holder.Index(index) + if idx == nil { + return false, newNotFoundError(ErrIndexNotFound, index) + } + columns := row.segments[0].data //should only be one segment + columnIDs := make([]uint64, 0) + none := make([]uint64, 0) // no bits will be set + + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + defer finisher(&err) + changed := false + clearFragment := func(frag *fragment) (bool, error) { + toClear := columnIDs[:0] + rowSet := make(map[uint64]struct{}) + + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) + + err = tx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting) + if err != nil { + return false, err + } + // this will be the remove part + if len(toClear) > 0 { + err = frag.importPositions(tx, none, toClear, rowSet) + if err != nil { + return false, err + } + return true, nil + } + return false, nil + } + for _, field := range idx.Fields() { + for _, view := range field.views() { + + frag, ok := view.fragments[shard] + if !ok { + continue + } + c, err := clearFragment(frag) + if err != nil { + return false, err + } + if c { + changed = true + } + } + } + for _, view := range idx.existenceFld.views() { + frag, ok := view.fragments[shard] + if !ok { + continue + } + for _, bit := range columns.Slice() { + c, err := frag.clearBit(tx, 0, bit) + if err != nil { + return false, nil + } + if c { + changed = true + } + } + } + return changed, nil +} diff --git a/pql/ast.go b/pql/ast.go index d042bcc08..649d4aff9 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -377,6 +377,7 @@ var callInfoByFunc = map[string]callInfo{ // taking field=value cases "Bitmap": {allowUnknown: true}, "Count": {allowUnknown: true}, + "Delete": {allowUnknown: true}, "Row": {allowUnknown: true}, "Range": {allowUnknown: true}, diff --git a/rbf/rbf.go b/rbf/rbf.go index 9d2167007..b9714ac1b 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -367,8 +367,7 @@ func (c *leafCell) Values(tx *Tx) []uint16 { case ContainerTypeArray: return toArray16(c.Data) case ContainerTypeRLE: - //a := make([]uint16, c.N) - a := make([]uint16, ArrayMaxSize) + a := make([]uint16, c.BitN) n := int32(0) for _, r := range toInterval16(c.Data) { for v := int(r.Start); v <= int(r.Last); v++ { diff --git a/test/cluster.go b/test/cluster.go index b34d99291..8eaef7b6f 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -180,12 +180,16 @@ func (c *Cluster) Len() int { return len(c.Nodes) } -func (c *Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uint64) { +func (c *Cluster) ImportBitsWithTimestamp(t testing.TB, index, field string, rowcols [][2]uint64, timestamps []int64) { t.Helper() byShard := make(map[uint64][][2]uint64) - for _, rowcol := range rowcols { + byShardTs := make(map[uint64][]int64) + for i, rowcol := range rowcols { shard := rowcol[1] / pilosa.ShardWidth byShard[shard] = append(byShard[shard], rowcol) + if len(timestamps) > 0 { + byShardTs[shard] = append(byShardTs[shard], timestamps[i]) + } } for shard, bits := range byShard { @@ -208,21 +212,40 @@ func (c *Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uin if com.API.Node().ID != node.ID { continue } + if len(timestamps) == 0 { + err := com.API.Import(context.Background(), nil, &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: shard, + RowIDs: rowIDs, + ColumnIDs: colIDs, + }) + if err != nil { + t.Fatalf("importing data: %v", err) + } + } else { + ts := byShardTs[shard] + err := com.API.Import(context.Background(), nil, &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: shard, + RowIDs: rowIDs, + ColumnIDs: colIDs, + Timestamps: ts, + }) + if err != nil { + t.Fatalf("importing data: %v", err) + } - err := com.API.Import(context.Background(), nil, &pilosa.ImportRequest{ - Index: index, - Field: field, - Shard: shard, - RowIDs: rowIDs, - ColumnIDs: colIDs, - }) - if err != nil { - t.Fatalf("importing data: %v", err) } } } } } +func (c *Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uint64) { + var noTime []int64 + c.ImportBitsWithTimestamp(t, index, field, rowcols, noTime) +} // ImportKeyKey imports data into an index where both the index and // the field are using string keys. From dcd6c649e8996ee5c7c41f19a65929cc940a7fa2 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 8 Apr 2021 09:37:34 -0500 Subject: [PATCH 2/6] skip blue-green on delete test --- delete_test.go | 1 + fragment_internal_test.go | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/delete_test.go b/delete_test.go index 04d50bc6e..66571516a 100644 --- a/delete_test.go +++ b/delete_test.go @@ -27,6 +27,7 @@ import ( ) func TestExecutor_DeleteRecords(t *testing.T) { + pilosa.NotBlueGreenTest(t) indexName := "i" setup := func(t *testing.T, r *require.Assertions, c *test.Cluster) { t.Helper() diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 7da683c58..68d5e2c0a 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -183,7 +183,7 @@ func TestFragment_RowcacheMap(t *testing.T) { // Ensure a fragment can clear a row. func TestFragment_ClearRow(t *testing.T) { - notBlueGreenTest(t) + NotBlueGreenTest(t) f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") _ = idx defer f.Clean(t) @@ -215,7 +215,7 @@ func TestFragment_ClearRow(t *testing.T) { // Ensure a fragment can set a row. func TestFragment_SetRow(t *testing.T) { - notBlueGreenTest(t) + NotBlueGreenTest(t) f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 7, "") _ = idx defer f.Clean(t) @@ -5382,7 +5382,7 @@ func TestFragmentConcurrentReadWrite(t *testing.T) { // actual transaction backends, there won't be any // data, and in particular, the blue-green tests will // note this and fire a false-positive. - notBlueGreenTest(t) + NotBlueGreenTest(t) f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, CacheTypeRanked) defer f.Clean(t) @@ -5543,7 +5543,7 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) { } } -func notBlueGreenTest(t *testing.T) { +func NotBlueGreenTest(t *testing.T) { if strings.Contains(CurrentBackend(), "_") { t.Skip("skip under blue green") } From ac7a8c3dd59425d618c76731c04dce287261ef38 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Fri, 9 Apr 2021 07:24:43 -0500 Subject: [PATCH 3/6] validate existence --- executor.go | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/executor.go b/executor.go index 3685410f3..224f20d2a 100644 --- a/executor.go +++ b/executor.go @@ -8412,18 +8412,20 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i } } } - for _, view := range idx.existenceFld.views() { - frag, ok := view.fragments[shard] - if !ok { - continue - } - for _, bit := range columns.Slice() { - c, err := frag.clearBit(tx, 0, bit) - if err != nil { - return false, nil + if idx.trackExistence { + for _, view := range idx.existenceFld.views() { + frag, ok := view.fragments[shard] + if !ok { + continue } - if c { - changed = true + for _, bit := range columns.Slice() { + c, err := frag.clearBit(tx, 0, bit) + if err != nil { + return false, nil + } + if c { + changed = true + } } } } From 84c269bfc08f8e2e05420d59429c79d17304c216 Mon Sep 17 00:00:00 2001 From: tgruben Date: Mon, 12 Apr 2021 13:38:58 -0500 Subject: [PATCH 4/6] Update delete_test.go --- delete_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/delete_test.go b/delete_test.go index 66571516a..b228b76c3 100644 --- a/delete_test.go +++ b/delete_test.go @@ -1,4 +1,4 @@ -// Copyright 2017 Pilosa Corp. +// Copyright 2021 Pilosa Corp. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. From 269837414e289e77231a0de57d0b5e274f6902b1 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 12 Apr 2021 16:13:16 -0500 Subject: [PATCH 5/6] rbf/intoContainer: ensure correct N, avoid recounting The remake container logic (used to avoid allocating extra containers while applying filters) relied on roaring recomputing N, which it did for bitmaps but didn't do for runs. Fix this both ways; it would now do that for runs, but also we add "with explicit N" variants and use those since we have a correct count already, and don't need it. This means fewer popcounts on bitmaps, and working at all on runs. --- rbf/cursorx.go | 4 ++-- roaring/container_stash.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/rbf/cursorx.go b/rbf/cursorx.go index 0ca291ff1..f1c8f338a 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -197,9 +197,9 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by } c = roaring.RemakeContainerBitmap(replacing, cloneMaybe) case ContainerTypeBitmap: - c = roaring.RemakeContainerBitmap(replacing, toArray64(cpMaybe)) + c = roaring.RemakeContainerBitmapN(replacing, toArray64(cpMaybe), int32(l.BitN)) case ContainerTypeRLE: - c = roaring.RemakeContainerRun(replacing, toInterval16(cpMaybe)) + c = roaring.RemakeContainerRunN(replacing, toInterval16(cpMaybe), int32(l.BitN)) } // Note: If the "roaringparanoia" build tag isn't set, this // should be optimized away entirely. Otherwise it's moderately diff --git a/roaring/container_stash.go b/roaring/container_stash.go index 2cde05743..7dd6eae83 100644 --- a/roaring/container_stash.go +++ b/roaring/container_stash.go @@ -125,6 +125,8 @@ func NewContainer() *Container { return NewContainerArray(nil) } +// RemakeContainerBitmap overwrites the contents of c, which must not be +// frozen, with a provided bitmap, and computes a correct N. func RemakeContainerBitmap(c *Container, bitmap []uint64) *Container { *c = Container{typeID: ContainerBitmap} c.setBitmap(bitmap) @@ -132,15 +134,41 @@ func RemakeContainerBitmap(c *Container, bitmap []uint64) *Container { return c } +// RemakeContainerBitmapN uses the provided n instead of counting bits. The +// provided container must not be frozen. +func RemakeContainerBitmapN(c *Container, bitmap []uint64, n int32) *Container { + *c = Container{typeID: ContainerBitmap} + c.setBitmap(bitmap) + c.n = n + return c +} + +// RemakeContainerArray populates c with an array container using the provided +// array. It must not be used on a frozen container. func RemakeContainerArray(c *Container, array []uint16) *Container { *c = Container{typeID: ContainerArray} c.setArray(array) return c } +// RemakeContainerRun repopulates c with the provided intervals. c must not +// be frozen. func RemakeContainerRun(c *Container, intervals []Interval16) *Container { *c = Container{typeID: ContainerRun} c.setRuns(intervals) + c.n = 0 + for _, r := range intervals { + c.n += int32(r.Last - r.Start + 1) + } + return c +} + +// RemakeContainerRunN repopulates c with the provided intervals, but +// assumes the provided n is accurate. c must not be frozen. +func RemakeContainerRunN(c *Container, intervals []Interval16, n int32) *Container { + *c = Container{typeID: ContainerRun} + c.setRuns(intervals) + c.n = n return c } From 2833365aae444b00c98a846806f0f68a3a63f9f3 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 12 Apr 2021 16:14:45 -0500 Subject: [PATCH 6/6] reuse the findExisting filter between fields, drop separate hack for existence The existence field wasn't working because runs were broken for filters in RBF. Fixing that allows us to simplify the logic. Also, we reuse the findExisting filter because the filter's cached collection of containers can be reused between things, allowing us to reduce allocations when there's a lot of views. --- executor.go | 48 ++++++++++++++++++++---------------------------- 1 file changed, 20 insertions(+), 28 deletions(-) diff --git a/executor.go b/executor.go index 58d99f90e..593be48dc 100644 --- a/executor.go +++ b/executor.go @@ -8351,13 +8351,17 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i if len(row.segments) == 0 { //nothing to remove return false, nil } + columns := row.segments[0].data //should only be one segment + if columns.Count() == 0 { + return false, nil + } // Fetch index. idx := e.Holder.Index(index) if idx == nil { return false, newNotFoundError(ErrIndexNotFound, index) } - columns := row.segments[0].data //should only be one segment + columnIDs := make([]uint64, 0) none := make([]uint64, 0) // no bits will be set @@ -8367,22 +8371,27 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i } defer finisher(&err) changed := false - clearFragment := func(frag *fragment) (bool, error) { - toClear := columnIDs[:0] - rowSet := make(map[uint64]struct{}) + colCounts := make([]int, 0) + toClear := columnIDs[:0] + rowSet := make(map[uint64]struct{}) + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) - callback := func(pos uint64) error { - toClear = append(toClear, pos) - rowID := pos / ShardWidth - rowSet[rowID] = struct{}{} - return nil - } - findExisting := roaring.NewBitmapBitmapFilter(columns, callback) + clearFragment := func(frag *fragment) (bool, error) { + // re-zero these + toClear = columnIDs[:0] + rowSet = make(map[uint64]struct{}) err = tx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting) if err != nil { return false, err } + colCounts = append(colCounts, len(toClear)) // this will be the remove part if len(toClear) > 0 { err = frag.importPositions(tx, none, toClear, rowSet) @@ -8409,22 +8418,5 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i } } } - if idx.trackExistence { - for _, view := range idx.existenceFld.views() { - frag, ok := view.fragments[shard] - if !ok { - continue - } - for _, bit := range columns.Slice() { - c, err := frag.clearBit(tx, 0, bit) - if err != nil { - return false, nil - } - if c { - changed = true - } - } - } - } return changed, nil }