ronantakizawa/codereview-bench huggingface dataset:
Repository: influxdata/influxdb
File: influxdb3_catalog/src/catalog.rs (Rust)

Code before:
        let subscriptions = Default::default();

        let catalog = Self {
            state: parking_lot::Mutex::new(CatalogState::Active),
            subscriptions,
            time_provider,
            store,
            metrics: Arc::new(CatalogMetrics::new(&metric_registry)),
            metric_registry,
            inner: RwLock::new(inner),
            limits: Default::default(),
        };

        create_internal_db(&catalog).await;
        Ok(catalog)
    }
}

impl TokenProvider for Catalog {
    fn get_token(&self, token_hash: Vec<u8>) -> Option<Arc<TokenInfo>> {
        self.inner.read().tokens.hash_to_info(token_hash)
    }
}

impl ProcessingEngineMetrics for Catalog {
    fn num_triggers(&self) -> u64 {
        self.inner.read().num_triggers() as u64
    }
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct Repository<I: CatalogId, R: CatalogResource> {
    pub(crate) repo: SerdeVecMap<I, Arc<R>>,
    pub(crate) id_name_map: BiHashMap<I, Arc<str>>,
    pub(crate) next_id: I,
}

impl<I: CatalogId, R: CatalogResource> Repository<I, R> {
    pub fn new() -> Self {
        Self {
            repo: SerdeVecMap::new(),
            id_name_map: BiHashMap::new(),
            next_id: I::default(),
        }
    }

    pub(crate) fn get_and_increment_next_id(&mut self) -> I {
        let next_id = self.next_id;
        self.next_id = self.next_id.next();
        next_id
    }

Review comment (suggestion):
can this be changed so that you capture the number of each type? (wal, schedule, request)

Code after:
        let inner = store.load_or_create_catalog().await?;
        let subscriptions = Default::default();

        let catalog = Self {
            state: parking_lot::Mutex::new(CatalogState::Active),
            subscriptions,
            time_provider,
            store,
            metrics: Arc::new(CatalogMetrics::new(&metric_registry)),
            metric_registry,
            inner: RwLock::new(inner),
            limits: Default::default(),
        };

        create_internal_db(&catalog).await;
        Ok(catalog)
    }
}

impl TokenProvider for Catalog {
    fn get_token(&self, token_hash: Vec<u8>) -> Option<Arc<TokenInfo>> {
        self.inner.read().tokens.hash_to_info(token_hash)
    }
}

impl ProcessingEngineMetrics for Catalog {
    fn num_triggers(&self) -> (u64, u64, u64, u64) {
        self.inner.read().num_triggers()
    }
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct Repository<I: CatalogId, R: CatalogResource> {
    pub(crate) repo: SerdeVecMap<I, Arc<R>>,
    pub(crate) id_name_map: BiHashMap<I, Arc<str>>,
    pub(crate) next_id: I,
}

impl<I: CatalogId, R: CatalogResource> Repository<I, R> {
    pub fn new() -> Self {
        Self {
            repo: SerdeVecMap::new(),
            id_name_map: BiHashMap::new(),
            next_id: I::default(),
        }
    }

    pub(crate) fn get_and_increment_next_id(&mut self) -> I {
        let next_id = self.next_id;
        self.next_id = self.next_id.next();
        next_id
Repository: influxdata/influxdb
File: influxdb3_write/src/write_buffer/queryable_buffer.rs (Rust)

Code before:
            )
            .await
            .await
            .unwrap();

        // validate we have two persisted files
        let files = queryable_buffer
            .persisted_files
            .get_files(db.id, table.table_id);
        assert_eq!(files.len(), 2);

        // Verify the `iox::series::key` metadata is present in the parquet file
        {
            let path = Path::from(files[0].path.as_str());
            let res = object_store
                .get(&path)
                .await
                .unwrap()
                .bytes()
                .await
                .unwrap();
            let res = arrow_reader::ParquetRecordBatchReaderBuilder::try_new(res)
                .unwrap()
                .build()
                .unwrap();
            let schema: Schema = Schema::try_from(res.schema()).unwrap();
            let primary_key = schema.primary_key();
            assert_eq!(primary_key, &["t2", "t1", "time"]);
        }
    }
}


Review comment (bug):
This is currently failing because the primary key is still using the lexicographical sort. I was able to debug that the parquet file is persisting the metadata, so I am still working through why this is failing.

Code after:
            )
            .await
            .await
            .unwrap();

        // validate we have two persisted files
        let files = queryable_buffer
            .persisted_files
            .get_files(db.id, table.table_id);
        assert_eq!(files.len(), 2);

        // Verify the `iox::series::key` metadata is present in the parquet file
        {
            let path = Path::from(files[0].path.as_str());
            let res = object_store
                .get(&path)
                .await
                .unwrap()
                .bytes()
                .await
                .unwrap();
            let metadata =
                arrow_reader::ArrowReaderMetadata::load(&res, ArrowReaderOptions::new()).unwrap();
            let schema: Schema = Schema::try_from(Arc::clone(metadata.schema())).unwrap();
            let primary_key = schema.primary_key();
            assert_eq!(primary_key, &["t2", "t1", "time"]);
        }
    }
}
Repository: influxdata/influxdb
File: tsdb/index/tsi1/partition.go (Go)

Code before:
}

// Wait will block until all compactions are finished.
// Must only be called while they are disabled.
func (p *Partition) Wait() {
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()

	// Debug level timeout
	timeoutDuration := 24 * time.Hour
	timeout := time.NewTicker(timeoutDuration)
	defer timeout.Stop()

	for {
		if p.CurrentCompactionN() == 0 {
			return
		}
		select {
		case <-timeout.C:
			files := make([]string, 0)
			for _, v := range p.fileSet.Files() {
				files = append(files, v.Path())
			}
			p.logger.Debug("Partition.Wait() timed out waiting for compactions to complete",
				zap.Int32("stuck_compactions", p.CurrentCompactionN()), zap.Duration("timeout", timeoutDuration),
				zap.Strings("files", files))
		case <-ticker.C:
		}
	}
}

// Close closes the index.
func (p *Partition) Close() error {
	// Wait for goroutines to finish outstanding compactions.
	p.once.Do(func() {
		close(p.closing)
		close(p.compactionInterrupt)
	})
	p.Wait()

	// Lock index and close remaining
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.fileSet == nil {
		return nil
	}

	// Close log files.
	var err []error
	for _, f := range p.fileSet.files {

Review comment (suggestion):
Did we decide earlier that we should not exit on timeout?

If we return on timeout, then the timeout should be a `timer` not a `ticker`. 

If we don't return on a timeout, then the timeout code could be in the ticker with a check if the timeoutDuration has passed since the wait was entered. 

Code after:
}

// Wait will block until all compactions are finished.
// Must only be called while they are disabled.
func (p *Partition) Wait() {
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()

	// Debug level timeout
	timeoutDuration := 24 * time.Hour
	startTime := time.Now()

	for {
		if p.CurrentCompactionN() == 0 {
			return
		}
		select {
		case <-ticker.C:
			elapsed := time.Since(startTime)
			if elapsed >= timeoutDuration {
				files := make([]string, 0)
				for _, v := range p.fileSet.Files() {
					files = append(files, v.Path())
				}
				p.logger.Warn("Partition.Wait() timed out waiting for compactions to complete",
					zap.Int32("stuck_compactions", p.CurrentCompactionN()), zap.Duration("timeout", timeoutDuration),
					zap.Strings("files", files))
				startTime = time.Now()
			}
		}
	}
}

// Close closes the index.
func (p *Partition) Close() error {
	// Wait for goroutines to finish outstanding compactions.
	p.once.Do(func() {
		close(p.closing)
		close(p.compactionInterrupt)
	})
	p.Wait()

	// Lock index and close remaining
	p.mu.Lock()
	defer p.mu.Unlock()

	if p.fileSet == nil {
		return nil
	}

	// Close log files.
Repository: influxdata/influxdb
File: tsdb/index/tsi1/partition.go (Go)

Code before:
		}
		p.seriesIDSet.Merge(ss)
	}
	return nil
}

// CurrentCompactionN returns the number of compactions currently running.
func (p *Partition) CurrentCompactionN() int32 {
	return p.currentCompactionN.Load()
}

// Wait will block until all compactions are finished.
// Must only be called while they are disabled.
func (p *Partition) Wait() {
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()

	// Debug level timeout
	timeoutDuration := 24 * time.Hour

	for {
		if p.CurrentCompactionN() == 0 {
			return
		}
		select {
		case <-time.After(timeoutDuration):
			files := make([]string, 0)
			for _, v := range p.fileSet.Files() {
				files = append(files, v.Path())
			}
			p.logger.Warn("Partition.Wait() timed out waiting for compactions to complete",
				zap.Int32("stuck_compactions", p.CurrentCompactionN()), zap.Duration("timeout", timeoutDuration),
				zap.Strings("files", files))
		case <-ticker.C:
		}
	}
}

// Close closes the index.
func (p *Partition) Close() error {
	// Wait for goroutines to finish outstanding compactions.
	p.once.Do(func() {
		close(p.closing)
		close(p.compactionInterrupt)
	})
	p.Wait()

	// Lock index and close remaining
	p.mu.Lock()
	defer p.mu.Unlock()


Review comment (bug):
`time.After` creates a new channel every call. You need to move it outside the loop.  See here: https://stackoverflow.com/a/39212626

This only prints the warning once., though.  If you want to print the warning repeatedly (every `timeoutDuration`) then you should use an `if` in the ticker case and check elapsed time.

Code after:
		}
		p.seriesIDSet.Merge(ss)
	}
	return nil
}

// CurrentCompactionN returns the number of compactions currently running.
func (p *Partition) CurrentCompactionN() int32 {
	return p.currentCompactionN.Load()
}

// Wait will block until all compactions are finished.
// Must only be called while they are disabled.
func (p *Partition) Wait() {
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()

	// Debug level timeout
	timeoutDuration := 24 * time.Hour
	startTime := time.Now()

	for {
		if p.CurrentCompactionN() == 0 {
			return
		}
		select {
		case <-ticker.C:
			elapsed := time.Since(startTime)
			if elapsed >= timeoutDuration {
				files := make([]string, 0)
				for _, v := range p.fileSet.Files() {
					files = append(files, v.Path())
				}
				p.logger.Warn("Partition.Wait() timed out waiting for compactions to complete",
					zap.Int32("stuck_compactions", p.CurrentCompactionN()), zap.Duration("timeout", timeoutDuration),
					zap.Strings("files", files))
				startTime = time.Now()
			}
		}
	}
}

// Close closes the index.
func (p *Partition) Close() error {
	// Wait for goroutines to finish outstanding compactions.
	p.once.Do(func() {
		close(p.closing)
		close(p.compactionInterrupt)
	})
	p.Wait()
Repository: influxdata/influxdb
File: pkg/metrics/timer.go (Go)

Code before:

import (
	"sync/atomic"
	"time"
)

// The timer type is used to store a duration.
type Timer struct {
	val  int64
	desc *desc
}

// Name returns the name of the timer.
func (t *Timer) Name() string { return t.desc.Name }

// Value atomically returns the value of the timer.
func (t *Timer) Value() time.Duration { return time.Duration(atomic.LoadInt64(&t.val)) }

// Update sets the timer value to d.
func (t *Timer) Update(d time.Duration) { atomic.StoreInt64(&t.val, int64(d)) }

// UpdateSince sets the timer value to the difference between since and the current time.
func (t *Timer) UpdateSince(since time.Time) { t.Update(time.Since(since)) }

// UpdateSinceAccumulate adds to the timer's current value the difference between since and the current time.
func (t *Timer) UpdateSinceAccumulate(since time.Time) { t.Update(time.Since(since) + t.Value()) }

// String returns a string representation using the name and value of the timer.
func (t *Timer) String() string { return t.desc.Name + ": " + time.Duration(t.val).String() }

// Time updates the timer to the duration it takes to call f.
func (t *Timer) Time(f func()) {
	s := time.Now()
	f()
	t.UpdateSince(s)
}


Review comment (bug):
This should use the `atomic` functions to avoid a potential race condition.

```go
func (t *Timer) UpdateSinceAccumulate(since time.Time) { atomic.AddInt64(&t.val, time.Since(since).Nanoseconds()) }
```

I'd also be tempted to change `Timer.val` to an `atomic.Int64` while we're touching this file.

Code after:

import (
	"sync/atomic"
	"time"
)

// Timer type is used to store a duration.
type Timer struct {
	val  atomic.Int64
	desc *desc
}

// Name returns the name of the timer.
func (t *Timer) Name() string { return t.desc.Name }

// Value atomically returns the value of the timer.
func (t *Timer) Value() time.Duration { return time.Duration(t.val.Load()) }

// Update sets the timer value to d.
func (t *Timer) Update(d time.Duration) { t.val.Store(int64(d)) }

// UpdateSince sets the timer value to the difference between since and the current time.
func (t *Timer) UpdateSince(since time.Time) { t.Update(time.Since(since)) }

// AddSince adds to the timer's current value the difference between since and the current time.
func (t *Timer) AddSince(since time.Time) { t.val.Add(int64(time.Since(since))) }

// String returns a string representation using the name and value of the timer.
func (t *Timer) String() string { return t.desc.Name + ": " + time.Duration(t.val.Load()).String() }

// Time updates the timer to the duration it takes to call f.
func (t *Timer) Time(f func()) {
	s := time.Now()
	f()
	t.UpdateSince(s)
}
Repository: influxdata/influxdb
File: tsdb/shard.go (Go)

Code before:
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s._engine == nil || s.index == nil { // Shard not open yet.
		return ""
	}
	return s.index.Type()
}

// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
func (s *Shard) LastModified() time.Time {
	t, err := s.LastModifiedOrErr()
	if err != nil {
		return time.Time{}
	}

	return t
}

// LastModifiedOrErr returns the time when this shard was last modified or an error.
// See: https://github.com/influxdata/plutonium/pull/4275#discussion_r2214077374
// Previously we would throw away any error from Shard.LastModified
func (s *Shard) LastModifiedOrErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}

Review comment (suggestion):
If `LastModifiedOrErr` is guaranteed to return `time.Time{}` on error, this code is redundant. All that's necessary is to throw away `err`.

Code after:
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s._engine == nil || s.index == nil { // Shard not open yet.
		return ""
	}
	return s.index.Type()
}

// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
// On error and 0 TSM files this will return time.Time{} (0001-01-01 00:00:00 +0000 UTC)
func (s *Shard) LastModified() time.Time {
	t, _ := s.LastModifiedWithErr()
	return t
}

// LastModifiedOrErr returns the time when this shard was last modified and an error.
func (s *Shard) LastModifiedWithErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}

// SeriesFile returns a reference the underlying series file. If return an error
// if the series file is nil.
func (s *Shard) SeriesFile() (*SeriesFile, error) {
	s.mu.RLock()
Repository: influxdata/influxdb
File: tsdb/shard.go (Go)

Code before:
// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
func (s *Shard) LastModified() time.Time {
	t, err := s.LastModifiedOrErr()
	if err != nil {
		return time.Time{}
	}

	return t
}

// LastModifiedOrErr returns the time when this shard was last modified or an error.
// See: https://github.com/influxdata/plutonium/pull/4275#discussion_r2214077374
// Previously we would throw away any error from Shard.LastModified
func (s *Shard) LastModifiedOrErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}

// SeriesFile returns a reference the underlying series file. If return an error
// if the series file is nil.
func (s *Shard) SeriesFile() (*SeriesFile, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err

Review comment (suggestion):
Pedantry alert! `LastModifiedOrErr` always returns the last modified time _and_ the error. I think it should be `LastModifiedAndErr` or `LastModifiedWithErr`.

Code after:
// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
// On error and 0 TSM files this will return time.Time{} (0001-01-01 00:00:00 +0000 UTC)
func (s *Shard) LastModified() time.Time {
	t, _ := s.LastModifiedWithErr()
	return t
}

// LastModifiedOrErr returns the time when this shard was last modified and an error.
func (s *Shard) LastModifiedWithErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}

// SeriesFile returns a reference the underlying series file. If return an error
// if the series file is nil.
func (s *Shard) SeriesFile() (*SeriesFile, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.sfile, nil
}

// IsIdle return true if the shard is not receiving writes and is fully compacted.
Repository: influxdata/influxdb
File: tsdb/shard.go (Go)

Code before:
// IndexType returns the index version being used for this shard.
//
// IndexType returns the empty string if it is called before the shard is opened,
// since it is only that point that the underlying index type is known.
func (s *Shard) IndexType() string {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s._engine == nil || s.index == nil { // Shard not open yet.
		return ""
	}
	return s.index.Type()
}

// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
func (s *Shard) LastModified() time.Time {
	t, err := s.LastModifiedOrErr()
	if err != nil {
		return time.Time{}
	}

	return t
}

// LastModifiedOrErr returns the time when this shard was last modified or an error.
// See: https://github.com/influxdata/plutonium/pull/4275#discussion_r2214077374
// Previously we would throw away any error from Shard.LastModified
func (s *Shard) LastModifiedOrErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

Review comment (suggestion):
Might be nice to add that `time.Time` is returned both on error and when there are no TSM files in the shard.

Code after:
// IndexType returns the index version being used for this shard.
//
// IndexType returns the empty string if it is called before the shard is opened,
// since it is only that point that the underlying index type is known.
func (s *Shard) IndexType() string {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s._engine == nil || s.index == nil { // Shard not open yet.
		return ""
	}
	return s.index.Type()
}

// ready determines if the Shard is ready for queries or writes.
// It returns nil if ready, otherwise ErrShardClosed or ErrShardDisabled
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}

// LastModified returns the time when this shard was last modified.
// On error and 0 TSM files this will return time.Time{} (0001-01-01 00:00:00 +0000 UTC)
func (s *Shard) LastModified() time.Time {
	t, _ := s.LastModifiedWithErr()
	return t
}

// LastModifiedOrErr returns the time when this shard was last modified and an error.
func (s *Shard) LastModifiedWithErr() (time.Time, error) {
	engine, err := s.Engine()
	if err != nil {
		return time.Time{}, err
	}
	return engine.LastModified(), nil
}

// Index returns a reference to the underlying index. It returns an error if
// the index is nil.
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}