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

Code before:
	// in memory.
	if iter.EstimatedIndexSize() > 64*1024*1024 {
		w, err = NewTSMWriterWithDiskBuffer(limitWriter)
	} else {
		w, err = NewTSMWriter(limitWriter)
	}
	if err != nil {
		// Close the file and return if we can't create the TSMWriter
		return false, errors.Join(err, fd.Close())
	}
	defer func() {
		var eInProgress errCompactionInProgress

		errs := make([]error, 0, 3)
		errs = append(errs, err)
		closeErr := w.Close()
		errs = append(errs, closeErr)

		// Check for conditions where we should not remove the file
		inProgress := errors.As(err, &eInProgress)
		inProgress = inProgress && errors.Is(eInProgress.err, os.ErrExist)
		maxBlocks := errors.Is(err, ErrMaxBlocksExceeded)
		maxFileSize := errors.Is(err, errMaxFileExceeded)
		if (closeErr == nil) && (inProgress || maxBlocks || maxFileSize) {
			// do not join errors, there is only the one.
			// rollToNext was set in the main body of the method
			return
		} else if err != nil {
			// Remove the file, we have had a problem
			errs = append(errs, w.Remove())
		}
		err = errors.Join(errs...)
	}()

	lastLogSize := w.Size()
	for iter.Next() {
		c.mu.RLock()
		enabled := c.snapshotsEnabled || c.compactionsEnabled
		c.mu.RUnlock()

		if !enabled {
			return false, errCompactionAborted{}
		}
		// Each call to read returns the next sorted key (or the prior one if there are
		// more values to write).  The size of values will be less than or equal to our
		// chunk size (1000)
		key, minTime, maxTime, block, err := iter.Read()
		if err != nil {
			return false, err
		}


Review comment (suggestion):
Since we don't return `closeErr` in this case, is there a way we can log it?

Also, would it make more sense to check if `rollToNext` is set instead of checking for the conditions where `rollToNext` would be set? This feels like something that could get out of sync with the main function.

Code after:
	if c.RateLimit != nil && throttle {
		limitWriter = limiter.NewWriterWithRate(fd, c.RateLimit)
	}

	// Use a disk based TSM buffer if it looks like we might create a big index
	// in memory.
	if iter.EstimatedIndexSize() > 64*1024*1024 {
		w, err = NewTSMWriterWithDiskBuffer(limitWriter)
	} else {
		w, err = NewTSMWriter(limitWriter)
	}
	if err != nil {
		// Close the file and return if we can't create the TSMWriter
		return false, errors.Join(err, fd.Close())
	}
	defer func() {
		var eInProgress errCompactionInProgress

		errs := make([]error, 0, 3)
		errs = append(errs, err)
		closeErr := w.Close()
		errs = append(errs, closeErr)

		// Check for conditions where we should not remove the file
		inProgress := errors.As(err, &eInProgress) && errors.Is(eInProgress.err, fs.ErrExist)
		if (closeErr == nil) && (inProgress || rollToNext) {
			// do not join errors, there is only the one.
			return
		} else if err != nil || closeErr != nil {
			// Remove the file, we have had a problem
			errs = append(errs, w.Remove())
		}
		err = errors.Join(errs...)
	}()

	lastLogSize := w.Size()
	for iter.Next() {
		c.mu.RLock()
		enabled := c.snapshotsEnabled || c.compactionsEnabled
		c.mu.RUnlock()

		if !enabled {
			return false, errCompactionAborted{}
		}
		// Each call to read returns the next sorted key (or the prior one if there are
		// more values to write).  The size of values will be less than or equal to our
		// chunk size (1000)
		key, minTime, maxTime, block, err := iter.Read()
		if err != nil {
			return false, err
		}