diff --git a/catcher.go b/catcher.go index 0a75ac9f7..a1f128d65 100644 --- a/catcher.go +++ b/catcher.go @@ -26,11 +26,6 @@ func init() { var _ Tx = (*catcherTx)(nil) -func (c *catcherTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - c.b.RemoveChannel(index, field, view, shard, a, resChan) - return -} - func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { return c.b.NewTxIterator(index, field, view, shard) } diff --git a/executor.go b/executor.go index 4a3345393..0f52e09d8 100644 --- a/executor.go +++ b/executor.go @@ -8308,24 +8308,30 @@ 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) +func clearFragment(writeTx Tx, columns *roaring.Bitmap, frag *fragment, toClear []uint64) (changed bool, err error) { + rowSet := make(map[uint64]struct{}) + toClear = toClear[:0] + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) 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 + if len(toClear) > 0 { + err = frag.importPositions(writeTx, []uint64{}, toClear, rowSet) + if err != nil { + return false, err + } + return true, nil + } + return false, nil } func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, idx *Index, shard uint64, normalFlow bool) (bool, error) { @@ -8378,8 +8384,8 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id } }() - resChan := make(chan countResults) + toClear := make([]uint64, 0) for _, field := range idx.Fields() { for _, view := range field.views() { @@ -8387,7 +8393,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id if !ok { continue } - c, err := clearFragment(writeTx, columns, frag, resChan) + c, err := clearFragment(writeTx, columns, frag, toClear) if err != nil { return false, err } @@ -8397,7 +8403,6 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id } } - close(resChan) if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created if normalFlow { existenceFragment.clearRow(writeTx, deletedRowID) @@ -8431,7 +8436,7 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx return } }() - resChan := make(chan countResults) + toClear := make([]uint64, 0) for _, field := range idx.Fields() { for _, view := range field.views() { @@ -8439,7 +8444,7 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx if !ok { continue } - c, err := clearFragment(writeTx, columns, frag, resChan) + c, err := clearFragment(writeTx, columns, frag, toClear) if err != nil { return false, err } @@ -8449,7 +8454,6 @@ 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 return changed, nil } diff --git a/rbf.go b/rbf.go index 1cdc59261..a6e81e463 100644 --- a/rbf.go +++ b/rbf.go @@ -228,71 +228,6 @@ type countResults struct { err error } -// RemoveChannel provides a method of streaming in bits or positions and not requiring a large buffer like add and remove -// the bits are input via the posChanel and the results are returned via the retChannel -func (tx *RBFTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - name := rbfName(index, field, view, shard) - var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter. - var rc *roaring.Container - var hi uint64 - var lo uint16 - var err error - changeCount := 0 - i := 0 - for v := range a { - hi, lo = highbits(v), lowbits(v) - if hi != lastHi { - // either first time through, or changed to a different container. - // do we need put the last updated container now? - if i > 0 { - // not first time through, write what we got. - if rc == nil || (rc.N() == 0) { - err = tx.tx.RemoveContainer(name, lastHi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to remove container")} - return - } - } else { - rc = roaring.Optimize(rc) - - err = tx.tx.PutContainer(name, lastHi, rc) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to put container")} - return - } - } - } - // get the next container - rc, err = tx.tx.Container(name, hi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to retrieve container")} - return - } - } // else same container, keep adding bits to rct. - chng := false - rc, chng = rc.Remove(lo) - if chng { - changeCount++ - } - lastHi = hi - i++ - } - // write the last updates. - if rc == nil || rc.N() == 0 { - err = tx.tx.RemoveContainer(name, hi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to remove container")} - return - } - } else { - err = tx.tx.PutContainer(name, hi, rc) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "put to remove container")} - return - } - } - resChan <- countResults{changeCount, nil} -} func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) { if len(a) == 0 { return 0, nil diff --git a/stattx.go b/stattx.go index 5ce265e90..4780f70bc 100644 --- a/stattx.go +++ b/stattx.go @@ -159,7 +159,6 @@ const ( kOffsetRange kLast // mark the end, always keep this last. The following aren't tracked atm: kType - kRemoveChannel ) func (k kall) String() string { @@ -206,8 +205,6 @@ func (k kall) String() string { return "kLast" case kType: return "kType" - case kRemoveChannel: - return "kRemoveChannel" } vprint.PanicOn(fmt.Sprintf("unknown kall '%v'", int(k))) return "" @@ -224,15 +221,6 @@ func (c *statTx) NewTxIterator(index, field, view string, shard uint64) *roaring }() return c.b.NewTxIterator(index, field, view, shard) } -func (c *statTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - me := kRemoveChannel - t0 := time.Now() - defer func() { - c.stats.add(me, time.Since(t0)) - }() - c.b.RemoveChannel(index, field, view, shard, a, resChan) - return -} func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) { me := kImportRoaringBits diff --git a/tx.go b/tx.go index 7be5a54b7..194b3e8e9 100644 --- a/tx.go +++ b/tx.go @@ -155,7 +155,6 @@ type Tx interface { GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) GetFieldSizeBytes(index, field string) (uint64, error) - RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) } // GenericApplyFilter implements ApplyFilter in terms of tx.ContainerIterator,