From f9ee18b8b073d00959f3c7dfe5528ab55bc029cd Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 14 Mar 2022 14:18:21 -0500 Subject: [PATCH] review suggestions --- executor.go | 107 +++++++++++++++++++++++----------------------------- 1 file changed, 47 insertions(+), 60 deletions(-) diff --git a/executor.go b/executor.go index 5b8c2c092..9d689c7bd 100644 --- a/executor.go +++ b/executor.go @@ -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))]...)