func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
	span := c.tracer.StartSpan("Client.ExperimentalReplayImport")
	defer span.Finish()

	// make work channel
	work := make(chan *importLog, concurrency*2)

	// spawn <concurrency> workers to read from channel
	eg := &errgroup.Group{}
	for i := 0; i < concurrency; i++ {
		eg.Go(func() error {
			for log := range work {
				// regular import doesn't forward to replicas, so we have to get all
				// the nodes.
				nodes, err := c.fetchFragmentNodes(log.Index, log.Shard)
				if err != nil {
					return errors.Wrap(err, "fetching fragment nodes")
				}

				if !log.IsRoaring {
					for _, node := range nodes {
						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
						if err = anyError(resp, err); err != nil {
							return errors.Wrap(err, "doing import")
						}
						resp.Body.Close()
					}
				} else {
					// import-roaring forwards on to all replicas, so we only import to
					// one node.
					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
					if err = anyError(resp, err); err != nil {
						return errors.Wrap(err, "doing import")
					}
					resp.Body.Close()
				}
			}
			return nil
		})
	}

	// wait on importers asynchronously in case they return early with errors
	waitErrCh := make(chan error)
	go func() {
		waitErrCh <- eg.Wait()
		close(waitErrCh)
	}()

	// populate work channel
	dec := newImportLogDecoder(r)
	var err error
	for {
		log := importLog{}
		err = dec.Decode(&log)
		if err != nil {
			err = errors.Wrap(err, "decoding")
			break
		}
		select {
		case work <- &log:
			continue
		case err = <-waitErrCh:
			err = errors.Wrap(err, "waiting")
		}
		break
	}

	// close work channel (now workers can exit)
	close(work)
	<-waitErrCh // make sure workers are complete
	if errors.Cause(err) == io.EOF {
		return nil
	}
	return err
}
func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
	span := c.tracer.StartSpan("Client.ExperimentalReplayImport")
	defer span.Finish()

	// make work channel
	work := make(chan *importLog, concurrency*2)

	// spawn <concurrency> workers to read from channel
	eg := &errgroup.Group{}
	for i := 0; i < concurrency; i++ {
		eg.Go(func() error {
			for log := range work {
				// regular import doesn't forward to replicas, so we have to get all
				// the nodes.
				nodes, err := c.fetchFragmentNodes(log.Index, log.Shard)
				if err != nil {
					return errors.Wrap(err, "fetching fragment nodes")
				}

				if !log.IsRoaring {
					for _, node := range nodes {
						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
						if err = anyError(resp, err); err != nil {
							return errors.Wrap(err, "doing import")
						}
						resp.Body.Close()
					}
				} else {
					// import-roaring forwards on to all replicas, so we only import to
					// one node.
					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
					if err = anyError(resp, err); err != nil {
						return errors.Wrap(err, "doing import")
					}
					resp.Body.Close()
				}
			}
			return nil
		})
	}

	// wait on importers asynchronously in case they return early with errors
	waitErrCh := make(chan error)
	go func() {
		waitErrCh <- eg.Wait()
		close(waitErrCh)
	}()

	// populate work channel
	dec := newImportLogDecoder(r)
	var err error
	for {
		log := importLog{}
		err = dec.Decode(&log)
		if err != nil {
			err = errors.Wrap(err, "decoding")
			break
		}
		select {
		case work <- &log:
			continue
		case err = <-waitErrCh:
			err = errors.Wrap(err, "waiting")
		}
		break
	}

	// close work channel (now workers can exit)
	close(work)
	<-waitErrCh // make sure workers are complete
	if errors.Cause(err) == io.EOF {
		return nil
	}
	return err
}
// ExperimentalReplayImport takes a data stream which was previously recorded by the import
// logging functionality and imports it to Pilosa. The target cluster need not
// be of the same size as the original cluster, but it must already have the
// necessary schema in place. It is an experimental method and may be changed or
// removed.