Repository: influxdata/influxdb
File: tsdb/engine/tsm1/compact.go (Go)

Code before:
	interrupt chan struct{}

	// maxErrors is the maximum number of errors to store before discarding.
	maxErrors int
	// overflowErrors is the number of errors we have ignored.
	overflowErrors int
}

func (t *tsmBatchKeyIterator) AppendError(err error) bool {
	if t.maxErrors > len(t.errs) {
		t.errs = append(t.errs, err)
		// Was the error stored?
		return true
	} else {
		// Was the error dropped
		t.overflowErrors++
		return false
	}
}

// NewTSMBatchKeyIterator returns a new TSM key iterator from readers.
// size indicates the maximum number of values to encode in a single block.
func NewTSMBatchKeyIterator(size int, fast bool, maxErrors int, interrupt chan struct{}, tsmFiles []string, readers ...*TSMReader) (KeyIterator, error) {
	var iter []*BlockIterator
	for _, r := range readers {
		iter = append(iter, r.BlockIterator())
	}

	return &tsmBatchKeyIterator{
		readers:              readers,
		values:               map[string][]Value{},
		pos:                  make([]int, len(readers)),
		size:                 size,
		iterators:            iter,
		fast:                 fast,
		tsmFiles:             tsmFiles,
		buf:                  make([]blocks, len(iter)),
		mergedFloatValues:    &tsdb.FloatArray{},
		mergedIntegerValues:  &tsdb.IntegerArray{},
		mergedUnsignedValues: &tsdb.UnsignedArray{},
		mergedBooleanValues:  &tsdb.BooleanArray{},
		mergedStringValues:   &tsdb.StringArray{},
		interrupt:            interrupt,
		maxErrors:            maxErrors,
	}, nil
}

func (k *tsmBatchKeyIterator) hasMergedValues() bool {
	return k.mergedFloatValues.Len() > 0 ||
		k.mergedIntegerValues.Len() > 0 ||
		k.mergedUnsignedValues.Len() > 0 ||

Review comment (refactor):
Not sure if we should count how many duplicate errors we received. Might be useful so you can tell the difference between a file that had a single error and a file that had 25 of the same error. I'll approve without this. If you want to add it, I can re-review after the change.

Code after:
	buf []blocks

	// mergeValues are decoded blocks that have been combined
	mergedFloatValues    *tsdb.FloatArray
	mergedIntegerValues  *tsdb.IntegerArray
	mergedUnsignedValues *tsdb.UnsignedArray
	mergedBooleanValues  *tsdb.BooleanArray
	mergedStringValues   *tsdb.StringArray

	// merged are encoded blocks that have been combined or used as is
	// without decode
	merged    blocks
	interrupt chan struct{}

	// maxErrors is the maximum number of errors to store before discarding.
	maxErrors int
	// overflowErrors is the number of errors we have ignored.
	overflowErrors int
}

// AppendError - store unique errors in the order of first appearance,
// up to a limit of maxErrors.  If the error is unique and stored, return true.
func (t *tsmBatchKeyIterator) AppendError(err error) bool {
	s := err.Error()
	if _, ok := t.errSet[s]; ok {
		return true
	} else if t.maxErrors > len(t.errs) {
		t.errs = append(t.errs, err)
		t.errSet[s] = struct{}{}
		return true
	} else {
		// Was the error dropped?
		t.overflowErrors++
		return false
	}
}

// NewTSMBatchKeyIterator returns a new TSM key iterator from readers.
// size indicates the maximum number of values to encode in a single block.
func NewTSMBatchKeyIterator(size int, fast bool, maxErrors int, interrupt chan struct{}, tsmFiles []string, readers ...*TSMReader) (KeyIterator, error) {
	var iter []*BlockIterator
	for _, r := range readers {
		iter = append(iter, r.BlockIterator())
	}

	return &tsmBatchKeyIterator{
		readers:              readers,
		values:               map[string][]Value{},
		pos:                  make([]int, len(readers)),
		errSet:               map[string]struct{}{},
		size:                 size,