scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1560 from tgruben/delete

Browse source 
[CORE-245] added pql delete function
This commit is contained in:
tgruben 2021-04-12 16:47:40 -05:00 • committed by GitHub
commit ebbb196a23
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 420 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

232
delete_test.go Normal file
View file

@ -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
}

118
executor.go
View file

@ -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
}

8
fragment_internal_test.go
View file

@ -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")
}

1
pql/ast.go
View file

@ -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},

4
rbf/cursorx.go
View file

@ -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

3
rbf/rbf.go
View file

@ -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++ {

28
roaring/container_stash.go
View file

@ -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
}

45
test/cluster.go
View file

@ -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.