ronantakizawa/codereview-bench huggingface dataset:
Repository: influxdata/influxdb
File: tsdb/series_file.go (Go)

Code before:
	p := f.SeriesIDPartition(id)
	if p == nil {
		return ErrInvalidSeriesPartitionID
	}
	return p.DeleteSeriesID(id)
}

// DeleteSeriesIDNoFlush flags a series as permanently deleted.
// This method returns a SeriesPartition that can be then used later
// for fsync operations.
func (f *SeriesFile) DeleteSeriesIDNoFlush(id uint64) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesIDNoFlush(id)
}

// FlushSegments flushes a group of partitions by id
func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	for id := range partitionIDs {
		p := f.partitions[id]
		if segment := p.activeSegment(); segment != nil {
			if err := segment.Flush(); err != nil {
				return err
			}
		}
	}
	return nil
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return false
	}
	return p.IsDeleted(id)
}

// SeriesKey returns the series key for a given id.
func (f *SeriesFile) SeriesKey(id uint64) []byte {
	if id == 0 {
		return nil
	}
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil
	}
	return p.SeriesKey(id)
}

Review comment (bug):
I wonder if we should continue flushing the remaining segments even if one fails? Also, is there any way to capture which segment failed for the error message, either it's filename or maybe just the partition index?

Code after:

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (f *SeriesFile) DeleteSeriesID(id uint64, flush bool) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesID(id, flush)
}

func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	var wg sync.WaitGroup
	errCh := make(chan error, SeriesFilePartitionN)

	for id := range partitionIDs {
		wg.Add(1)
		p := f.partitions[id]
		go func() {
			defer wg.Done()
			p.mu.Lock()
			defer p.mu.Unlock()
			if segment := p.activeSegment(); segment != nil {
				if err := segment.Flush(); err != nil {
					errCh <- fmt.Errorf("unable to flush segment %s: %w", segment.file.Name(), err)
				}
			}
		}()
	}

	wg.Wait()
	close(errCh)

	var errs = make([]error, 0, SeriesFilePartitionN)
	for err := range errCh {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return false
	}
	return p.IsDeleted(id)
}

// SeriesKey returns the series key for a given id.
Repository: influxdata/influxdb
File: tsdb/series_partition.go (Go)

Code before:
				log.Error("series partition compaction failed", zap.Error(err))
			}

			logEnd()

			// Clear compaction flag.
			p.mu.Lock()
			p.compacting = false
			p.mu.Unlock()
		}()
	}

	return nil
}

// Compacting returns if the SeriesPartition is currently compacting.
func (p *SeriesPartition) Compacting() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.compacting
}

// DeleteSeriesIDNoFlush flags a series as permanently deleted.
// This operation does not incur an fsync.
// If the series is reintroduced later then it must create a new id.
func (p *SeriesPartition) DeleteSeriesIDNoFlush(id uint64) error {
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.closed {
		return ErrSeriesPartitionClosed
	}

	// Already tombstoned, ignore.
	if p.index.IsDeleted(id) {
		return nil
	}

	// Write tombstone entry.
	_, err := p.writeLogEntry(AppendSeriesEntry(nil, SeriesEntryTombstoneFlag, id, nil))
	if err != nil {
		return err
	}

	// Mark tombstone in memory.
	p.index.Delete(id)

	return nil
}

// FlushSegment fsyncs a segment to disk

Review comment (refactor):
I'm not a huge fan of copying and pasting code like this. Could `DeleteSeriesID` be changed to call `DeleteSeriesIDNoFlush`? Or if that's not possible, maybe extract the actual work into `deleteSeriesID(id uint64, flush bool)`, which is then used by both `DeleteSeriesID` and `DeleteSeriesIDNoFlush`.

Actually, `DeleteSeriesID` is called in so few places, maybe `DeleteSeriesID` should just have a `flush bool` parameter added to it. Actually, it looks like we only call it one place?!

Code after:
				log.Error("series partition compaction failed", zap.Error(err))
			}

			logEnd()

			// Clear compaction flag.
			p.mu.Lock()
			p.compacting = false
			p.mu.Unlock()
		}()
	}

	return nil
}

// Compacting returns if the SeriesPartition is currently compacting.
func (p *SeriesPartition) Compacting() bool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.compacting
}

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (p *SeriesPartition) DeleteSeriesID(id uint64, flush bool) error {
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.closed {
		return ErrSeriesPartitionClosed
	}

	// Already tombstoned, ignore.
	if p.index.IsDeleted(id) {
		return nil
	}

	// Write tombstone entry.
	_, err := p.writeLogEntry(AppendSeriesEntry(nil, SeriesEntryTombstoneFlag, id, nil))
	if err != nil {
		return err
	}

	// Flush active segment write.
	if flush {
		if segment := p.activeSegment(); segment != nil {
			if err := segment.Flush(); err != nil {
				return fmt.Errorf("unable to flush segment %s: %w", segment.file.Name(), err)
			}
		}
Repository: influxdata/influxdb
File: tsdb/store.go (Go)

Code before:
				if len(errs) != 0 {
					return errors.Join(errs...)
				}
			}

			const DeleteLogTrigger = 10_000
			seriesCount := ss.Cardinality()
			deleteStart := time.Now()
			var deletedCount atomic.Uint64
			var partitionIDs = make(map[int]struct{})

			ss.ForEach(func(id uint64) {
				part, err := sfile.DeleteSeriesIDNoFlush(id)
				if err != nil {
					sfile.Logger.Error(
						"cannot delete series in shard",
						zap.Uint64("series_id", id),
						zap.Uint64("shard_id", shardID),
						zap.Error(err))
				} else {
					partitionIDs[part.id] = struct{}{}
					deleted := deletedCount.Add(1)

					if deleted%DeleteLogTrigger == 0 {
						s.Logger.Info(fmt.Sprintf("DeleteShard: %d series deleted", DeleteLogTrigger),
							zap.String("db", db),
							zap.Uint64("shard_id", shardID),
							zap.Uint64("deleted", deleted),
							zap.Uint64("remaining", seriesCount-deleted),
							zap.Uint64("total", seriesCount),
							zap.Duration("elapsed", time.Since(deleteStart)))
					}
				}
			})

			if err := sfile.FlushSegments(partitionIDs); err != nil {
				return err
			}
		}
	}

	// enter the epoch tracker
	guards, gen := epoch.StartWrite()
	defer epoch.EndWrite(gen)

	// wait for any guards before closing the shard
	for _, guard := range guards {
		guard.Wait()
	}

	// Close the shard.

Review comment (bug):
This should probably eventually be `s.Logger.Debug` to avoid noisy logs, but I can see it needing to be at info level for initial C1 testing.

Code after:
				if len(errs) != 0 {
					return errors.Join(errs...)
				}
			}

			const DeleteLogTrigger = 10_000
			seriesCount := ss.Cardinality()
			deleteStart := time.Now()
			var deletedCount atomic.Uint64
			var partitionIDs = make(map[int]struct{}, SeriesFilePartitionN)

			ss.ForEach(func(id uint64) {
				p, err := sfile.DeleteSeriesID(id, NoFlush)
				if err != nil {
					sfile.Logger.Error(
						"cannot delete series in shard",
						zap.Uint64("series_id", id),
						zap.Uint64("shard_id", shardID),
						zap.String("series_file_path", sfile.Path()),
						zap.Error(err))
				} else {
					partitionIDs[p.id] = struct{}{}
					deleted := deletedCount.Add(1)

					if deleted%DeleteLogTrigger == 0 {
						s.Logger.Info(fmt.Sprintf("DeleteShard: %d series deleted", DeleteLogTrigger),
							zap.String("db", db),
							zap.Uint64("shard_id", shardID),
							zap.String("series_file_path", sfile.Path()),
							zap.Uint64("deleted", deleted),
							zap.Uint64("remaining", seriesCount-deleted),
							zap.Uint64("total", seriesCount),
							zap.Duration("elapsed", time.Since(deleteStart)))
					}
				}
			})

			if err := sfile.FlushSegments(partitionIDs); err != nil {
				sfile.Logger.Error(
					"error while flushing a series file segment",
					zap.Uint64("shard_id", shardID),
					zap.String("series_file_path", sfile.Path()),
					zap.Error(err))
			}
		}
	}

	// enter the epoch tracker
	guards, gen := epoch.StartWrite()
	defer epoch.EndWrite(gen)
Repository: influxdata/influxdb
File: tsdb/store.go (Go)

Code before:
				if err != nil {
					sfile.Logger.Error(
						"cannot delete series in shard",
						zap.Uint64("series_id", id),
						zap.Uint64("shard_id", shardID),
						zap.Error(err))
				} else {
					partitionIDs[p.id] = struct{}{}
					deleted := deletedCount.Add(1)

					if deleted%DeleteLogTrigger == 0 {
						s.Logger.Info(fmt.Sprintf("DeleteShard: %d series deleted", DeleteLogTrigger),
							zap.String("db", db),
							zap.Uint64("shard_id", shardID),
							zap.Uint64("deleted", deleted),
							zap.Uint64("remaining", seriesCount-deleted),
							zap.Uint64("total", seriesCount),
							zap.Duration("elapsed", time.Since(deleteStart)))
					}
				}
			})

			if err := sfile.FlushSegments(partitionIDs); err != nil {
				sfile.Logger.Error(
					"error while flushing a series file segment",
					zap.Uint64("shard_id", shardID),
					zap.Error(err))
			}
		}
	}

	// enter the epoch tracker
	guards, gen := epoch.StartWrite()
	defer epoch.EndWrite(gen)

	// wait for any guards before closing the shard
	for _, guard := range guards {
		guard.Wait()
	}

	// Close the shard.
	if err := sh.Close(); err != nil {
		return err
	}

	// Remove the on-disk shard data.
	if err := os.RemoveAll(sh.path); err != nil {
		return err
	} else if err = os.RemoveAll(sh.walPath); err != nil {
		return err
	} else {

Review comment (suggestion):
I figured we should just log out the errors here and not return if we encounter an error while flushing a segment. 

Code after:
				if err != nil {
					sfile.Logger.Error(
						"cannot delete series in shard",
						zap.Uint64("series_id", id),
						zap.Uint64("shard_id", shardID),
						zap.String("series_file_path", sfile.Path()),
						zap.Error(err))
				} else {
					partitionIDs[p.id] = struct{}{}
					deleted := deletedCount.Add(1)

					if deleted%DeleteLogTrigger == 0 {
						s.Logger.Info(fmt.Sprintf("DeleteShard: %d series deleted", DeleteLogTrigger),
							zap.String("db", db),
							zap.Uint64("shard_id", shardID),
							zap.String("series_file_path", sfile.Path()),
							zap.Uint64("deleted", deleted),
							zap.Uint64("remaining", seriesCount-deleted),
							zap.Uint64("total", seriesCount),
							zap.Duration("elapsed", time.Since(deleteStart)))
					}
				}
			})

			if err := sfile.FlushSegments(partitionIDs); err != nil {
				sfile.Logger.Error(
					"error while flushing a series file segment",
					zap.Uint64("shard_id", shardID),
					zap.String("series_file_path", sfile.Path()),
					zap.Error(err))
			}
		}
	}

	// enter the epoch tracker
	guards, gen := epoch.StartWrite()
	defer epoch.EndWrite(gen)

	// wait for any guards before closing the shard
	for _, guard := range guards {
		guard.Wait()
	}

	// Close the shard.
	if err := sh.Close(); err != nil {
		return err
	}

	// Remove the on-disk shard data.
	if err := os.RemoveAll(sh.path); err != nil {
		return err
Repository: influxdata/influxdb
File: tsdb/engine/tsm1/engine.go (Go)

Code before:
					return err
				} else if deleted {
					actuallyDeleted = append(actuallyDeleted, k)
				}
			}
		}
		if len(actuallyDeleted) > 0 {
			if err := e.fieldset.Save(tsdb.MeasurementsToFieldChangeDeletions(actuallyDeleted)); err != nil {
				return err
			}
		}

		// Remove any series IDs for our set that still exist in other shards.
		// We cannot remove these from the series file yet.
		if err := e.seriesIDSets.ForEach(func(s *tsdb.SeriesIDSet) {
			ids = ids.AndNot(s)
		}); err != nil {
			return err
		}

		// Remove the remaining ids from the series file as they no longer exist
		// in any shard.
		var err error
		ids.ForEach(func(id uint64) {
			name, tags := e.sfile.Series(id)
			if _, err1 := e.sfile.DeleteSeriesID(id, tsdb.Flush); err1 != nil {
				err = err1
				return
			}

			// In the case of the inmem index the series can be removed across
			// the global index (all shards).
			if index, ok := e.index.(*inmem.ShardIndex); ok {
				key := models.MakeKey(name, tags)
				if e := index.Index.DropSeriesGlobal(key); e != nil {
					err = e
				}
			}
		})
		if err != nil {
			return err
		}
	}

	return nil
}

func (e *Engine) cleanupMeasurement(name []byte) (deleted bool, err error) {
	// A sentinel error message to cause DeleteWithLock to not delete the measurement
	abortErr := fmt.Errorf("measurements still exist")


Review comment (suggestion):
Seems like we should consolidate `Flush` calls here, too.  There are some hot code paths like `Drop Measurement` that end up here.


Code after:
					return err
				} else if deleted {
					actuallyDeleted = append(actuallyDeleted, k)
				}
			}
		}
		if len(actuallyDeleted) > 0 {
			if err := e.fieldset.Save(tsdb.MeasurementsToFieldChangeDeletions(actuallyDeleted)); err != nil {
				return err
			}
		}

		// Remove any series IDs for our set that still exist in other shards.
		// We cannot remove these from the series file yet.
		if err := e.seriesIDSets.ForEach(func(s *tsdb.SeriesIDSet) {
			ids = ids.AndNot(s)
		}); err != nil {
			return err
		}

		// Remove the remaining ids from the series file as they no longer exist
		// in any shard.
		var err error
		var partitionIDs = make(map[int]struct{}, tsdb.SeriesFilePartitionN)
		ids.ForEach(func(id uint64) {
			name, tags := e.sfile.Series(id)
			part, err1 := e.sfile.DeleteSeriesID(id, tsdb.NoFlush)
			if err1 != nil {
				err = err1
				return
			}
			partitionIDs[part.ID()] = struct{}{}

			// In the case of the inmem index the series can be removed across
			// the global index (all shards).
			if index, ok := e.index.(*inmem.ShardIndex); ok {
				key := models.MakeKey(name, tags)
				if e := index.Index.DropSeriesGlobal(key); e != nil {
					err = e
				}
			}
		})
		if err != nil {
			return err
		}

		if err := e.sfile.FlushSegments(partitionIDs); err != nil {
			e.sfile.Logger.Error(
				"error while flushing a series file segment",
				zap.String("series_file_path", e.sfile.Path()),
				zap.Error(err))
Repository: influxdata/influxdb
File: tsdb/series_file.go (Go)

Code before:
		})
	}
	if err := g.Wait(); err != nil {
		return nil, err
	}
	return ids, nil
}

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (f *SeriesFile) DeleteSeriesID(id uint64, flush bool) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesID(id, flush)
}

// FlushSegments flushes a group of partitions by id
func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	var errs []error
	for id := range partitionIDs {
		p := f.partitions[id]
		if segment := p.activeSegment(); segment != nil {
			if err := segment.Flush(); err != nil {
				errs = append(errs, err)
			}
		}
	}
	return errors.Join(errs...)
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return false
	}
	return p.IsDeleted(id)
}

// SeriesKey returns the series key for a given id.
func (f *SeriesFile) SeriesKey(id uint64) []byte {
	if id == 0 {
		return nil
	}
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil
	}

Review comment (performance):
Another thought here, perhaps for later: these (up to 8) calls could be parallelized with an `errgroup.Group` or a `sync.WaitGroup` for a likely greater speed-up.  Getting the batched `Flush`es used in `DeleteSeriesRangeWithPredicate` is higher priority, and parallelism is not always a speed-up, so perf testing of parallelization would be important. 

Code after:
		})
	}
	if err := g.Wait(); err != nil {
		return nil, err
	}
	return ids, nil
}

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (f *SeriesFile) DeleteSeriesID(id uint64, flush bool) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesID(id, flush)
}

func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	var wg sync.WaitGroup
	errCh := make(chan error, SeriesFilePartitionN)

	for id := range partitionIDs {
		wg.Add(1)
		p := f.partitions[id]
		go func() {
			defer wg.Done()
			p.mu.Lock()
			defer p.mu.Unlock()
			if segment := p.activeSegment(); segment != nil {
				if err := segment.Flush(); err != nil {
					errCh <- fmt.Errorf("unable to flush segment %s: %w", segment.file.Name(), err)
				}
			}
		}()
	}

	wg.Wait()
	close(errCh)

	var errs = make([]error, 0, SeriesFilePartitionN)
	for err := range errCh {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
Repository: influxdata/influxdb
File: tsdb/series_file_test.go (Go)

Code before:
	for i := 0; i < b.N; i++ {
		// Compact all segments in parallel.
		var g errgroup.Group
		for _, p := range cachedCompactionSeriesFile.Partitions() {
			for _, segment := range p.Segments() {
				p, segment := p, segment
				g.Go(func() error {
					return segment.CompactToPath(segment.Path()+".tmp", p.Index())
				})
			}
		}

		if err := g.Wait(); err != nil {
			b.Fatal(err)
		}
	}
}

func TestSeriesFile_FlushSegments(t *testing.T) {
	sfile := MustOpenSeriesFile()
	defer sfile.Close()

	// Create some series to ensure there's data in the segments.
	var names [][]byte
	var tagsSlice []models.Tags
	for i := 0; i < 100; i++ {
		names = append(names, []byte(fmt.Sprintf("measurement%d", i)))
		tagsSlice = append(tagsSlice, models.NewTags(map[string]string{"tag": "value"}))
	}
	ids, err := sfile.CreateSeriesListIfNotExists(names, tagsSlice, tsdb.NoopStatsTracker())
	require.NoError(t, err)

	// Collect all partition IDs that have series.
	partitionIDs := make(map[int]struct{})
	for _, id := range ids {
		partitionIDs[sfile.SeriesIDPartitionID(id)] = struct{}{}
	}

	// Flush the segments.
	err = sfile.FlushSegments(partitionIDs)
	require.NoError(t, err)

	// Verify series still exist after flush.
	require.Equal(t, uint64(100), sfile.SeriesCount())
}

// Series represents name/tagset pairs that are used in testing.
type Series struct {
	Name    []byte
	Tags    models.Tags
	Deleted bool

Review comment (suggestion):
You should make 100 into a constant since it is used in multiple places. Not important in a test, that would also let you pre-allocate `names` and `tagSlice`.

Code after:
	for i := 0; i < b.N; i++ {
		// Compact all segments in parallel.
		var g errgroup.Group
		for _, p := range cachedCompactionSeriesFile.Partitions() {
			for _, segment := range p.Segments() {
				p, segment := p, segment
				g.Go(func() error {
					return segment.CompactToPath(segment.Path()+".tmp", p.Index())
				})
			}
		}

		if err := g.Wait(); err != nil {
			b.Fatal(err)
		}
	}
}

func TestSeriesFile_FlushSegments(t *testing.T) {
	const SeriesN = 100
	sfile := MustOpenSeriesFile()
	defer func() {
		require.NoError(t, sfile.Close(), "series file close")
	}()

	// Create some series to ensure there's data in the segments.
	var names = make([][]byte, 0, SeriesN)
	var tagsSlice = make([]models.Tags, 0, SeriesN)
	for i := 0; i < SeriesN; i++ {
		names = append(names, []byte(fmt.Sprintf("measurement%d", i)))
		tagsSlice = append(tagsSlice, models.NewTags(map[string]string{"tag": "value"}))
	}
	ids, err := sfile.CreateSeriesListIfNotExists(names, tagsSlice, tsdb.NoopStatsTracker())
	require.NoError(t, err)

	// Collect all partition IDs that have series.
	partitionIDs := make(map[int]struct{}, tsdb.SeriesFilePartitionN)
	for _, id := range ids {
		partitionIDs[sfile.SeriesIDPartitionID(id)] = struct{}{}
	}

	// Flush the segments.
	err = sfile.FlushSegments(partitionIDs)
	require.NoError(t, err)

	// Verify series still exist after flush.
	require.Equal(t, uint64(SeriesN), sfile.SeriesCount())
}

// Series represents name/tagset pairs that are used in testing.
type Series struct {
Repository: influxdata/influxdb
File: tsdb/series_file.go (Go)

Code before:
}

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (f *SeriesFile) DeleteSeriesID(id uint64, flush bool) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesID(id, flush)
}

func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	var wg sync.WaitGroup
	errCh := make(chan error, SeriesFilePartitionN)

	for id := range partitionIDs {
		wg.Add(1)
		p := f.partitions[id]
		go func() {
			defer wg.Done()
			p.mu.Lock()
			defer p.mu.Unlock()
			if segment := p.activeSegment(); segment != nil {
				errCh <- segment.Flush()
			}
		}()
	}

	wg.Wait()
	close(errCh)

	var errs []error
	for err := range errCh {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return false
	}
	return p.IsDeleted(id)
}

// SeriesKey returns the series key for a given id.
func (f *SeriesFile) SeriesKey(id uint64) []byte {

Review comment (suggestion):
You could check for non-nil return and wrap the error with the partition file name (not full path) and only then put it on the channel.  Teensy tiny added benefit, no nil errors on channel.

Code after:
}

// DeleteSeriesID flags a series as permanently deleted.
// If the series is reintroduced later than it must create a new id.
// Setting flush will indicate whether this method triggers a fsync.
func (f *SeriesFile) DeleteSeriesID(id uint64, flush bool) (*SeriesPartition, error) {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return nil, ErrInvalidSeriesPartitionID
	}
	return p, p.DeleteSeriesID(id, flush)
}

func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
	var wg sync.WaitGroup
	errCh := make(chan error, SeriesFilePartitionN)

	for id := range partitionIDs {
		wg.Add(1)
		p := f.partitions[id]
		go func() {
			defer wg.Done()
			p.mu.Lock()
			defer p.mu.Unlock()
			if segment := p.activeSegment(); segment != nil {
				if err := segment.Flush(); err != nil {
					errCh <- fmt.Errorf("unable to flush segment %s: %w", segment.file.Name(), err)
				}
			}
		}()
	}

	wg.Wait()
	close(errCh)

	var errs = make([]error, 0, SeriesFilePartitionN)
	for err := range errCh {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

// IsDeleted returns true if the ID has been deleted before.
func (f *SeriesFile) IsDeleted(id uint64) bool {
	p := f.SeriesIDPartition(id)
	if p == nil {
		return false
	}
	return p.IsDeleted(id)
}