diff --git a/delete_test.go b/delete_test.go new file mode 100644 index 000000000..b228b76c3 --- /dev/null +++ b/delete_test.go @@ -0,0 +1,232 @@ +// 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. +// 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) { + pilosa.NotBlueGreenTest(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 0fdc88d0f..593be48dc 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) @@ -8302,3 +8306,117 @@ 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 + } + 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) + } + + 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 + 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) + + 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) + 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 + } + } + } + return changed, nil +} 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") } diff --git a/pql/ast.go b/pql/ast.go index 8229ae74d..1147cb00a 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/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/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/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 } 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.