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

review suggestions

Browse source
This commit is contained in:
Todd Gruben 2022-03-14 14:18:21 -05:00
parent a54fded24a
commit f9ee18b8b0
1 changed files with 47 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

107
executor.go
View file

@ -8294,6 +8294,9 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, index strin
//bmCall is a bitmap
row, err := e.executeBitmapCallShard(ctx, qcx, index, bmCall, shard)
qcx.Abort()
if len(row.segments) == 0 {
return
}
columns := row.segments[0].data
if columns.Count() == 0 {
return
@ -8305,6 +8308,26 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, index strin
func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool, error) {
return DeleteRowsWithFlow(ctx, src, idx, shard, false)
}
func clearFragment(writeTx Tx, columns *roaring.Bitmap, frag *fragment, resChan chan countResults) (changed bool, err error) {
posChan := make(chan uint64, 8192)
findExisting := roaring.NewBitmapBitmapFilter(columns, func(pos uint64) error {
posChan <- pos
return nil
})
go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan)
err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
close(posChan)
if err != nil {
return false, err
}
r := <-resChan
changed = r.changeCount > 0
err = r.err
return
}
func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, idx *Index, shard uint64, normalFlow bool) (bool, error) {
var existenceFragment *fragment
var deletedRowID uint64
@ -8318,6 +8341,9 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
}
src := NewRowFromBitmap(columns)
deletedRowID, err = transactExistRow(ctx, idx, shard, existenceFragment, src)
if err != nil {
return false, err
}
}
commitor, err = deleteKeyTranslation(ctx, idx, shard, columns)
if err != nil {
@ -8352,24 +8378,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
}
}()
findExisting := roaring.NewBitmapBitmapFilter(columns, func(p uint64) error { return nil })
resChan := make(chan countResults)
clearFragment := func(frag *fragment) (bool, error) {
posChan := make(chan uint64, 8192)
findExisting.SetCallback(func(pos uint64) error {
posChan <- pos
return nil
})
go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan)
err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
close(posChan)
if err != nil {
return false, err
}
r := <-resChan
return r.changeCount > 0, r.err
}
for _, field := range idx.Fields() {
for _, view := range field.views() {
@ -8378,7 +8387,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
if !ok {
continue
}
c, err := clearFragment(frag)
c, err := clearFragment(writeTx, columns, frag, resChan)
if err != nil {
return false, err
}
@ -8406,46 +8415,23 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id
return changed, nil
}
func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx *Index, shard uint64, normalFlow bool) (bool, error) {
func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx *Index, shard uint64, normalFlow bool) (changed bool, err error) {
var existenceFragment *fragment
var deletedRowID uint64
var commitor Commitor = &NopCommitor{}
var err error
writeTx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
if err != nil {
return false, err
}
defer writeTx.Rollback()
changed := false
defer func() {
// if there is an error in the key commit, then rollback the delete
// write records before keys to remove possiblity of unmatch keys=records
err = writeTx.Commit()
err := writeTx.Commit()
if err != nil {
changed = false
commitor.Rollback()
return
}
}()
findExisting := roaring.NewBitmapBitmapFilter(columns, func(p uint64) error { return nil })
resChan := make(chan countResults)
clearFragment := func(frag *fragment) (bool, error) {
posChan := make(chan uint64, 8192)
findExisting.SetCallback(func(pos uint64) error {
posChan <- pos
return nil
})
go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan)
err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
close(posChan)
if err != nil {
return false, err
}
r := <-resChan
return r.changeCount > 0, r.err
}
for _, field := range idx.Fields() {
for _, view := range field.views() {
@ -8453,7 +8439,7 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx
if !ok {
continue
}
c, err := clearFragment(frag)
c, err := clearFragment(writeTx, columns, frag, resChan)
if err != nil {
return false, err
}
@ -8464,24 +8450,27 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx
}
}
close(resChan)
if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created
if normalFlow {
existenceFragment.clearRow(writeTx, deletedRowID)
} else {
// this is if we are recovering from failure and cleaning up
rows, err := existenceFragment.rows(ctx, writeTx, 1)
if err != nil {
return false, err
}
for _, rowId := range rows {
existenceFragment.clearRow(writeTx, rowId)
}
}
if existenceFragment == nil { //a string keys have been deleted and the deleteRow was created
return changed, nil
}
if normalFlow {
existenceFragment.clearRow(writeTx, deletedRowID)
return changed, nil
}
// this is if we are recovering from failure and cleaning up
rows, err := existenceFragment.rows(ctx, writeTx, 1)
if err != nil {
return false, err
}
for _, rowId := range rows {
existenceFragment.clearRow(writeTx, rowId)
}
return changed, nil
}
func DeleteRowsWithFlow(ctx context.Context, src *Row, idx *Index, shard uint64, normalFlow bool) (bool, error) {
func DeleteRowsWithFlow(ctx context.Context, src *Row, idx *Index, shard uint64, normalFlow bool) (change bool, err error) {
if len(src.segments) == 0 { //nothing to remove
return false, nil
}
@ -8496,8 +8485,6 @@ func DeleteRowsWithFlow(ctx context.Context, src *Row, idx *Index, shard uint64,
}
return b
}
var change bool
var err error
limit := idx.holder.cfg.RBFConfig.MaxDelete
for i := 0; i < len(bits); i += limit {
batch := roaring.NewBitmap(bits[i:min(i+limit, len(bits))]...)