code: func (e *CleanupIndexExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
  	req.Reset()
  	if e.done {
  		return nil
  	}
  	for {
  		errInTxn := kv.RunInNewTxn(e.ctx.GetStore(), true, func(txn kv.Transaction) error {
  			err := e.fetchIndex(ctx, txn)
  			if err != nil {
  				return err
  			}
  			values, err := e.batchGetRecord(txn)
  			if err != nil {
  				return err
  			}
  			err = e.deleteDanglingIdx(txn, values)
  			if err != nil {
  				return err
  			}
  			return nil
  		})
  		if errInTxn != nil {
  			return errInTxn
  		}
  		if e.scanRowCnt == 0 {
  			break
  		}
  		e.scanRowCnt = 0
  		e.batchKeys = e.batchKeys[:0]
  		for k := range e.idxValues {
  			delete(e.idxValues, k)
  		}
  	}
  	e.done = true
  	req.AppendUint64(0, e.removeCnt)
  	return nil
  }
docs: // Next implements the Executor Next interface.