Merge pull request #1997 from molecula/bad-delete-bandaid

[FB-1290] revert async remove channel

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

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
 	}

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

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

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,