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

Code before:
	if !c.add(tsmFiles) {
		return nil, errCompactionInProgress{}
	}
	defer c.remove(tsmFiles)

	files, err := c.compact(true, tsmFiles, logger)

	// See if we were disabled while writing a snapshot
	c.mu.RLock()
	enabled = c.compactionsEnabled
	c.mu.RUnlock()

	if !enabled {
		if err := c.removeTmpFiles(files); err != nil {
			return nil, err
		}
		return nil, errCompactionsDisabled
	}

	return files, err

}

// removeTmpFiles is responsible for cleaning up a compaction that
// was started, but then abandoned before the temporary files were dealt with.
func (c *Compactor) removeTmpFiles(files []string) error {
	for _, f := range files {
		if err := os.Remove(f); err != nil {
			return fmt.Errorf("error removing temp compaction file: %v", err)
		}
	}
	return nil
}

// writeNewFiles writes from the iterator into new TSM files, rotating
// to a new file once it has reached the max TSM file size.
func (c *Compactor) writeNewFiles(generation, sequence int, src []string, iter KeyIterator, throttle bool, logger *zap.Logger) ([]string, error) {
	// These are the new TSM files written
	var files []string
	var eInProgress errCompactionInProgress

	for {
		sequence++

		// New TSM files are written to a temp file and renamed when fully completed.
		fileName := filepath.Join(c.Dir, c.formatFileName(generation, sequence)+"."+TSMFileExtension+"."+TmpTSMFileExtension)
		logger.Debug("Compacting files", zap.Int("file_count", len(src)), zap.String("output_file", fileName))

		// Write as much as possible to this file
		rollToNext, err := c.write(fileName, iter, throttle, logger)


Review comment (suggestion):
I wonder if it would make sense to add a little warning log message here if `files <= 0` something like "RemoveTmpFiles provided with 0 files". The expectation is that this gets called with a list of files and that it should not ever be empty 🤔 

Code after:
	if !c.add(tsmFiles) {
		return nil, errCompactionInProgress{}
	}
	defer c.remove(tsmFiles)

	files, err := c.compact(true, tsmFiles, logger)

	// See if we were disabled while writing a snapshot
	c.mu.RLock()
	enabled = c.compactionsEnabled
	c.mu.RUnlock()

	if !enabled {
		if err := c.RemoveTmpFiles(files); err != nil {
			return nil, err
		}
		return nil, errCompactionsDisabled
	}

	return files, err

}

// RemoveTmpFiles is responsible for cleaning up a compaction that
// was started, but then abandoned before the temporary files were dealt with.
func (c *Compactor) RemoveTmpFiles(files []string) error {
	var errs []error
	for _, f := range files {
		if err := os.Remove(f); err != nil {
			errs = append(errs, fmt.Errorf("error removing temp compaction file %s: %w", f, err))
		}
	}
	return errors.Join(errs...)
}

// writeNewFiles writes from the iterator into new TSM files, rotating
// to a new file once it has reached the max TSM file size.
func (c *Compactor) writeNewFiles(generation, sequence int, src []string, iter KeyIterator, throttle bool, logger *zap.Logger) ([]string, error) {
	// These are the new TSM files written
	var files []string
	var eInProgress errCompactionInProgress

	for {
		sequence++

		// New TSM files are written to a temp file and renamed when fully completed.
		fileName := filepath.Join(c.Dir, c.formatFileName(generation, sequence)+"."+TSMFileExtension+"."+TmpTSMFileExtension)
		logger.Debug("Compacting files", zap.Int("file_count", len(src)), zap.String("output_file", fileName))

		// Write as much as possible to this file
		rollToNext, err := c.write(fileName, iter, throttle, logger)