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

Code before:
	Validator   func(allFiles []string, groups []tsm1.CompactionGroup) error
}

// AdjacentFileProperty validates that compaction groups don't create gaps
var AdjacentFileProperty = CompactionProperty{
	Name:        "No Gaps Rule",
	Description: "Files should not have gaps within the same compaction level - if files A and C are in the same level group, any file B between them should also be in a group at the same level or higher",
	Validator:   validateNoGaps,
}

// parseFileName uses the existing TSM file parsing functionality
func parseFileName(filename string) (generation int, sequence int, err error) {
	return tsm1.DefaultParseFileName(filename)
}

// fileInfo holds parsed file information for sorting
type fileInfo struct {
	filename   string
	generation int
	sequence   int
}

// validateNoGaps checks that there are no gaps within the same compaction level
// A gap occurs when files A and C are in the same level group, but file B (between A and C)
// is in a different level group, creating a "hole" in the sequence
func validateNoGaps(allFiles []string, groups []tsm1.CompactionGroup) error {
	// Parse all files and sort them by generation, then by sequence
	var fileInfos []fileInfo
	for _, file := range allFiles {
		gen, seq, err := parseFileName(file)
		if err != nil {
			return fmt.Errorf("failed to parse file %s: %v", file, err)
		}
		fileInfos = append(fileInfos, fileInfo{
			filename:   file,
			generation: gen,
			sequence:   seq,
		})
	}

	// Sort by generation first, then by sequence
	sort.Slice(fileInfos, func(i, j int) bool {
		if fileInfos[i].generation != fileInfos[j].generation {
			return fileInfos[i].generation < fileInfos[j].generation
		}
		return fileInfos[i].sequence < fileInfos[j].sequence
	})

	// Create a map of filename to group index
	fileToGroup := make(map[string]int)
	for groupIdx, group := range groups {

Review comment (bug):
the algorithm in here is strange; why did you write it this way? I believe it works but it seems more complicated than it needs to be. 

It also sorts the files that are in the input CompactionGroups... which hides errors of files being out of order in the group (yes this should get caught elsewhere too).

I don't understand why the algorithm isn't:
1. parse the master file list filenames to get gens and seqs
2. sort by generation and then seq within generation into a "sorted master slice"
3. invert slice to a map[filename]->index in sorted master list
4. for each group, for each file in the group check if file[i] has an index in the master map that is -1 from file[i+1] in the master map. 

This will fail if the group's files aren't sorted or if the group's files aren't adjacent. 

e.g. 
sorted master files list -> index in sorted slice
```
0-5 -> 0
1-5 -> 1
2-4 -> 2
3-4 -> 3
4-4 -> 4
5-2 -> 5
6-1 -> 6
7-1 -> 7
```

Any group has to be a strict subset. 

Group0: <-- PASS
1-5 -> 1
2-4 -> 2

Group1:
3-4 -> 3
4-4 -> 4
6-1 -> 6 <--- failure. index 6 is not 4+1. Adjacency failure.

Group2:
6-1 -> 6
5-2 -> 5 <--- failure. index 5 is not 6+1; group file ordering failure. 

For the Group1, it'd be easy to print the files in between because you have the index 4 and index 6, so you print the file at index 5 e.g. 

Code after:
	Validator   func(allFiles []string, groups []tsm1.CompactionGroup) error
}

// AdjacentFileProperty validates that compaction groups don't create gaps
var AdjacentFileProperty = CompactionProperty{
	Name:        "Adjacency Rule",
	Description: "Files should not have non-adjacent files within the same compaction level - if files A and C are in the same level group, any file B between them should also be in a group at the same level or higher",
	Validator:   validateFileAdjacency,
}

type fileInfo struct {
	filename   string
	generation int
	sequence   int
	index      int
}

// validateFileAdjacency checks that there are no adjacency violations between TSM files
// The following example will highlight an adjacency violation:
// Given the following list of files [01-01.tsm, 02-02.tsm, 03-03.tsm] let's say we have the following compaction plans created
// Group 1: [01-01.tsm, 03-03.tsm] & Group 2: [02-02.tsm]
// This violates file adjacency as the first compaction group sees [01-01.tsm, X, 03-03.tsm] and the second group: [X, 02-02.tsm, X]
// these are non-contiguous blocks, when the first group performs compaction we will have two files that are out of order compacted together.
// This rule is important to maintain the ordering of files, with improper ordering we cannot determine which point is the newest point when overwrites occur.
// We always want the newest write to win.
func validateFileAdjacency(allFiles []string, groups []tsm1.CompactionGroup) error {
	var fileInfos []fileInfo
	for _, file := range allFiles {
		gen, seq, err := tsm1.DefaultParseFileName(file)
		if err != nil {
			return fmt.Errorf("failed to parse file %s: %v", file, err)
		}
		fileInfos = append(fileInfos, fileInfo{
			filename:   file,
			generation: gen,
			sequence:   seq,
		})
	}

	slices.SortFunc(fileInfos, func(a, b fileInfo) int {
		if a.generation != b.generation {
			return a.generation - b.generation
		}

		return a.sequence - b.sequence
	})

	var fileMap = make(map[string]fileInfo, len(allFiles))
	for i, file := range allFiles {
		fileMap[file] = fileInfo{
			index: i,